everruns_mcp/
executor.rs

1//! MCP tool execution.
2//!
3//! [`McpExecutor`] resolves a tool call's server prefix to a [`McpConnection`]
4//! and executes it via [`McpClient`]. It implements [`McpToolInvoker`] so hosts
5//! register MCP tools as first-class [`everruns_core::tools::Tool`] entries in
6//! the regular `ToolRegistry` (via `everruns_core::build_mcp_proxy_tools`),
7//! instead of routing `mcp_*` calls through a separate executor
8//! (specs/runtime-mcp.md D5).
9
10use 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/// Resolves a sanitized server prefix to a connection. Implementations differ
20/// per host (runtime: effective scoped servers; worker: gRPC lookup).
21#[async_trait]
22pub trait McpConnectionResolver: Send + Sync {
23    async fn resolve(&self, server_prefix: &str) -> Result<Option<McpConnection>>;
24}
25
26/// In-memory resolver over a fixed set of connections, keyed by sanitized
27/// server name. Used by runtime/CLI hosts that hold the effective scoped
28/// servers up front.
29#[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
69/// Executes `mcp_*` tool calls against remote/local MCP servers.
70pub 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        // The server requires an OAuth connection the user has not made yet.
91        // Return a connection_required result (the host renders an inline
92        // connect prompt) instead of letting the call fail with a 401.
93        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}
everruns_mcp/executor.rs

everruns_mcp/
executor.rs
