1use crate::client::McpClient;
11use crate::transport::McpConnection;
12use anyhow::{Result, anyhow};
13use async_trait::async_trait;
14use everruns_core::mcp_server::sanitize_mcp_server_name;
15use everruns_core::{AgentLoopError, McpToolInvoker, ToolCall, ToolResult, parse_mcp_tool_name};
16use std::collections::HashMap;
17use std::sync::Arc;
18
19#[async_trait]
22pub trait McpConnectionResolver: Send + Sync {
23 async fn resolve(&self, server_prefix: &str) -> Result<Option<McpConnection>>;
24}
25
26#[derive(Default)]
30pub struct StaticConnectionResolver {
31 connections: HashMap<String, McpConnection>,
32}
33
34impl StaticConnectionResolver {
35 pub fn new() -> Self {
36 Self::default()
37 }
38
39 pub fn insert(&mut self, connection: McpConnection) {
40 let key = sanitize_mcp_server_name(&connection.name);
41 self.connections.insert(key, connection);
42 }
43
44 pub fn with(mut self, connection: McpConnection) -> Self {
45 self.insert(connection);
46 self
47 }
48
49 pub fn from_connections(connections: impl IntoIterator<Item = McpConnection>) -> Self {
50 let mut resolver = Self::new();
51 for connection in connections {
52 resolver.insert(connection);
53 }
54 resolver
55 }
56
57 pub fn is_empty(&self) -> bool {
58 self.connections.is_empty()
59 }
60}
61
62#[async_trait]
63impl McpConnectionResolver for StaticConnectionResolver {
64 async fn resolve(&self, server_prefix: &str) -> Result<Option<McpConnection>> {
65 Ok(self.connections.get(server_prefix).cloned())
66 }
67}
68
69pub struct McpExecutor {
71 client: Arc<McpClient>,
72 resolver: Arc<dyn McpConnectionResolver>,
73}
74
75impl McpExecutor {
76 pub fn new(client: Arc<McpClient>, resolver: Arc<dyn McpConnectionResolver>) -> Self {
77 Self { client, resolver }
78 }
79
80 pub async fn execute_mcp_tool(&self, tool_call: &ToolCall) -> Result<ToolResult> {
81 let (server_prefix, original_tool_name) = parse_mcp_tool_name(&tool_call.name)
82 .ok_or_else(|| anyhow!("Invalid MCP tool name: {}", tool_call.name))?;
83
84 let connection = self
85 .resolver
86 .resolve(&server_prefix)
87 .await?
88 .ok_or_else(|| anyhow!("MCP server not found for prefix: {server_prefix}"))?;
89
90 if let Some(provider) = &connection.pending_oauth_provider {
94 return Ok(ToolResult {
95 tool_call_id: tool_call.id.clone(),
96 result: None,
97 images: None,
98 error: Some(format!(
99 "MCP server '{}' requires an OAuth connection. \
100 Ask the user to connect provider '{provider}'.",
101 connection.name
102 )),
103 connection_required: Some(provider.clone()),
104 raw_output: None,
105 });
106 }
107
108 self.client
109 .call_as_tool_result(
110 &connection,
111 tool_call.id.clone(),
112 &original_tool_name,
113 tool_call.arguments.clone(),
114 )
115 .await
116 }
117}
118
119#[async_trait]
120impl McpToolInvoker for McpExecutor {
121 async fn invoke(&self, tool_call: &ToolCall) -> everruns_core::Result<ToolResult> {
122 self.execute_mcp_tool(tool_call).await.map_err(|e| {
123 tracing::error!(error = %e, "MCP tool execution failed");
124 AgentLoopError::tool(e.to_string())
125 })
126 }
127}