feat: add remote MCP gateway support with read_resource as first built-in MCP tool

Cargo.toml

@@ -27,6 +27,7 @@ futures = "0.3"
 indexmap = "2"
 http = "1"
 reqwest = { version = "0.12", default-features = false }
+rmcp = { version = "1.8", default-features = false }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 thiserror = "2"

crates/agentic-server-core/Cargo.toml

@@ -22,6 +22,12 @@ futures.workspace = true
 indexmap.workspace = true
 http.workspace = true
 reqwest = { workspace = true, features = ["default-tls", "stream"] }
+rmcp = { workspace = true, features = [
+    "client",
+    "reqwest",
+    "transport-child-process",
+    "transport-streamable-http-client-reqwest",
+] }
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true

crates/agentic-server-core/src/events/normalize.rs

@@ -60,6 +60,8 @@ fn classify_event_type(type_str: &str) -> SSEEventType {
         "response.web_search_call.in_progress" => SSEEventType::WebSearchCallInProgress,
         "response.web_search_call.searching" => SSEEventType::WebSearchCallSearching,
         "response.web_search_call.completed" => SSEEventType::WebSearchCallCompleted,
+        "response.mcp_tool_call.in_progress" => SSEEventType::McpToolCallInProgress,
+        "response.mcp_tool_call.completed" => SSEEventType::McpToolCallCompleted,
         _ => SSEEventType::Other,
     }
 }
@@ -94,6 +96,8 @@ fn extract_payload(event_type: SSEEventType, json: &Value) -> EventPayload {
         | SSEEventType::WebSearchCallInProgress
         | SSEEventType::WebSearchCallSearching
         | SSEEventType::WebSearchCallCompleted
+        | SSEEventType::McpToolCallInProgress
+        | SSEEventType::McpToolCallCompleted
         | SSEEventType::Other => EventPayload::Raw(json.clone()),
     }
 }

crates/agentic-server-core/src/events/types.rs

@@ -7,6 +7,8 @@ use crate::types::io::ResponseUsage;
 pub enum SSEItemType {
     Reasoning,
     FunctionCall,
+    WebSearchCall,
+    McpToolCall,
     Message,
 }
 
@@ -16,6 +18,8 @@ impl SSEItemType {
         match self {
             Self::Reasoning => "reasoning",
             Self::FunctionCall => "function_call",
+            Self::WebSearchCall => "web_search_call",
+            Self::McpToolCall => "mcp_tool_call",
             Self::Message => "message",
         }
     }
@@ -26,6 +30,8 @@ impl From<&str> for SSEItemType {
         match s {
             "reasoning" => Self::Reasoning,
             "function_call" => Self::FunctionCall,
+            "web_search_call" => Self::WebSearchCall,
+            "mcp_tool_call" => Self::McpToolCall,
             _ => Self::Message,
         }
     }
@@ -91,6 +97,8 @@ pub enum SSEEventType {
     WebSearchCallInProgress,
     WebSearchCallSearching,
     WebSearchCallCompleted,
+    McpToolCallInProgress,
+    McpToolCallCompleted,
 
     // Catch-all for unrecognized events
     Other,

crates/agentic-server-core/src/executor/accumulator.rs

@@ -249,11 +249,19 @@ impl ResponseAccumulator {
                         item,
                         text: String::with_capacity(256),
                     }),
+                    SSEItemType::WebSearchCall | SSEItemType::McpToolCall => None,
                 };
                 if let Some(inflight) = entry {
                     self.in_flight.insert(item_id.clone(), inflight);
                 }
             }
+            (SSEEventType::OutputItemDone, EventPayload::OutputItemDone { item, .. }) => {
+                if let Some(output_item @ (OutputItem::WebSearchCall(_) | OutputItem::McpToolCall(_))) =
+                    deserialize_from_value_opt::<OutputItem>(item.clone())
+                {
+                    self.output.push(output_item);
+                }
+            }
             (SSEEventType::ReasoningTextDelta, EventPayload::ReasoningDelta { delta, item_id }) => {
                 if let Some(InFlight::Reasoning { text, .. }) = self.in_flight.get_mut(item_id) {
                     text.push_str(delta);
@@ -494,6 +502,37 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_process_event_mcp_tool_call_done_accumulates_output() {
+        let lines = vec![
+            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"mcp_tool_call","id":"mcp_1","server":"repo","tool":"read_mcp_resource","arguments":{"server":"repo"},"status":"in_progress"}}"#.to_string(),
+            r#"data: {"type":"response.mcp_tool_call.in_progress","item_id":"mcp_1","output_index":0}"#.to_string(),
+            r#"data: {"type":"response.mcp_tool_call.completed","item_id":"mcp_1","output_index":0,"item":{"type":"mcp_tool_call","id":"mcp_1","server":"repo","tool":"read_mcp_resource","arguments":{"server":"repo"},"status":"completed","result":{"contents":[]}}}"#.to_string(),
+            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"mcp_tool_call","id":"mcp_1","server":"repo","tool":"read_mcp_resource","arguments":{"server":"repo"},"status":"completed","result":{"contents":[]}}}"#.to_string(),
+            r#"data: {"type":"response.done","response":{"id":"resp_1","status":"completed","usage":{"input_tokens":5,"output_tokens":2,"total_tokens":7}}}"#.to_string(),
+        ];
+
+        let acc = ResponseAccumulator::from_sse_lines(lines, None);
+        assert_eq!(acc.status, ResponseStatus::Completed);
+        assert_eq!(acc.output.len(), 1);
+        assert!(matches!(acc.output[0], OutputItem::McpToolCall(_)));
+    }
+
+    #[test]
+    fn test_process_event_web_search_done_accumulates_output() {
+        let lines = vec![
+            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"web_search_call","id":"ws_1","status":"in_progress","action":{"type":"search","query":"rust","sources":[]}}}"#.to_string(),
+            r#"data: {"type":"response.web_search_call.in_progress","item_id":"ws_1","output_index":0}"#.to_string(),
+            r#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"web_search_call","id":"ws_1","status":"completed","action":{"type":"search","query":"rust","sources":[]}}}"#.to_string(),
+            r#"data: {"type":"response.done","response":{"id":"resp_1","status":"completed","usage":{"input_tokens":5,"output_tokens":2,"total_tokens":7}}}"#.to_string(),
+        ];
+
+        let acc = ResponseAccumulator::from_sse_lines(lines, None);
+        assert_eq!(acc.status, ResponseStatus::Completed);
+        assert_eq!(acc.output.len(), 1);
+        assert!(matches!(acc.output[0], OutputItem::WebSearchCall(_)));
+    }
+
     #[test]
     fn test_process_event_completed_with_usage() {
         let mut acc = ResponseAccumulator::new("resp_1".into(), None);

crates/agentic-server-core/src/executor/engine.rs

@@ -108,7 +108,7 @@ async fn run_until_gateway_tools_complete(
     stream_events: Option<&mpsc::UnboundedSender<String>>,
 ) -> ExecutorResult<(ResponsePayload, RequestContext)> {
     let registry: ToolRegistry = match ctx.enriched_request.tools.as_ref() {
-        Some(tools) => ToolRegistry::build_with_handlers(tools, |tool_type| exec_ctx.gateway_executors.get(tool_type))?,
+        Some(tools) => ToolRegistry::build_with_handlers(tools, &exec_ctx.gateway_executors).await?,
         None => ToolRegistry::default(),
     };
     let mut combined_output: Vec<crate::OutputItem> = Vec::new();

crates/agentic-server-core/src/executor/gateway.rs

@@ -7,7 +7,7 @@ use tokio::sync::mpsc;
 use crate::executor::error::{ExecutorError, ExecutorResult};
 use crate::executor::request::RequestContext;
 use crate::tool::{GatewayDispatchResult, ToolError, ToolOutput, ToolRegistry, ToolType};
-use crate::types::io::output::{FunctionToolCall, WebSearchCallStatus};
+use crate::types::io::output::{FunctionToolCall, GatewayCallStatus};
 use crate::types::io::{InputItem, OutputItem, ResponsesInput};
 use crate::utils::common::serialize_to_string;
 
@@ -167,9 +167,9 @@ async fn execute_gateway_call_with_timeout(
         ))),
     });
     let (output, status) = match dispatch.output {
-        Ok(output) => (output, WebSearchCallStatus::Completed),
+        Ok(output) => (output, GatewayCallStatus::Completed),
         Err(ToolError::Execution(message) | ToolError::Config(message)) => {
-            (execution_error_output(&call, &message)?, WebSearchCallStatus::Failed)
+            (execution_error_output(&call, &message)?, GatewayCallStatus::Failed)
         }
     };
     let public_output = gateway_public_output(dispatch.tool_type, &call, &output, status);
@@ -184,15 +184,12 @@ fn gateway_public_output(
     tool_type: ToolType,
     call: &FunctionToolCall,
     output: &ToolOutput,
-    status: WebSearchCallStatus,
+    status: GatewayCallStatus,
 ) -> Option<OutputItem> {
     match tool_type {
         ToolType::WebSearch => Some(crate::tool::web_search::output_item(call, output, status)),
-        ToolType::Function
-        | ToolType::CodexNamespace
-        | ToolType::Mcp
-        | ToolType::FileSearch
-        | ToolType::CodeInterpreter => None,
+        ToolType::Mcp => Some(crate::tool::mcp::handler::output_item(call, output, status)),
+        ToolType::Function | ToolType::CodexNamespace | ToolType::FileSearch | ToolType::CodeInterpreter => None,
     }
 }
 
@@ -255,9 +252,9 @@ fn gateway_event_plans(
                 output_index: u32::try_from(output_index).unwrap_or(u32::MAX),
                 started_output: match entry.tool_type {
                     ToolType::WebSearch => Some(crate::tool::web_search::started_output_item(call)),
+                    ToolType::Mcp => Some(crate::tool::mcp::handler::started_output_item(call)),
                     ToolType::Function
                     | ToolType::CodexNamespace
-                    | ToolType::Mcp
                     | ToolType::FileSearch
                     | ToolType::CodeInterpreter => None,
                 },
@@ -290,28 +287,38 @@ fn emit_gateway_start_events(
         let Some(output_item) = &plan.started_output else {
             continue;
         };
-        let OutputItem::WebSearchCall(web_search_call) = output_item else {
-            continue;
-        };
         let item = output_item_value(output_item)?;
         let added_event = serde_json::json!({
                 "type": "response.output_item.added",
                 "output_index": plan.output_index,
                 "item": item
         });
         emit_sse_json(sender, &added_event)?;
-        let in_progress_event = serde_json::json!({
-                "type": "response.web_search_call.in_progress",
-                "item_id": web_search_call.id,
-                "output_index": plan.output_index
-        });
-        emit_sse_json(sender, &in_progress_event)?;
-        let searching_event = serde_json::json!({
-                "type": "response.web_search_call.searching",
-                "item_id": web_search_call.id,
-                "output_index": plan.output_index
-        });
-        emit_sse_json(sender, &searching_event)?;
+        match output_item {
+            OutputItem::WebSearchCall(web_search_call) => {
+                let in_progress_event = serde_json::json!({
+                        "type": "response.web_search_call.in_progress",
+                        "item_id": web_search_call.id,
+                        "output_index": plan.output_index
+                });
+                emit_sse_json(sender, &in_progress_event)?;
+                let searching_event = serde_json::json!({
+                        "type": "response.web_search_call.searching",
+                        "item_id": web_search_call.id,
+                        "output_index": plan.output_index
+                });
+                emit_sse_json(sender, &searching_event)?;
+            }
+            OutputItem::McpToolCall(mcp_tool_call) => {
+                let in_progress_event = serde_json::json!({
+                        "type": "response.mcp_tool_call.in_progress",
+                        "item_id": mcp_tool_call.id,
+                        "output_index": plan.output_index
+                });
+                emit_sse_json(sender, &in_progress_event)?;
+            }
+            OutputItem::Message(_) | OutputItem::FunctionCall(_) | OutputItem::Reasoning(_) | OutputItem::Unknown => {}
+        }
     }
     Ok(())
 }
@@ -325,18 +332,26 @@ fn emit_gateway_completed_events(
         return Ok(());
     };
     for result in results {
-        let Some(OutputItem::WebSearchCall(web_search_call)) = &result.public_output else {
+        let Some(public_output) = &result.public_output else {
             continue;
         };
         let output_index = plans
             .iter()
             .find(|plan| plan.call_id == result.call.call_id)
             .map_or(0, |plan| plan.output_index);
-        let output_item = OutputItem::WebSearchCall(web_search_call.clone());
-        let item = output_item_value(&output_item)?;
+        let (event_type, item_id) = match public_output {
+            OutputItem::WebSearchCall(web_search_call) => {
+                ("response.web_search_call.completed", web_search_call.id.as_str())
+            }
+            OutputItem::McpToolCall(mcp_tool_call) => ("response.mcp_tool_call.completed", mcp_tool_call.id.as_str()),
+            OutputItem::Message(_) | OutputItem::FunctionCall(_) | OutputItem::Reasoning(_) | OutputItem::Unknown => {
+                continue;
+            }
+        };
+        let item = output_item_value(public_output)?;
         let completed_event = serde_json::json!({
-                "type": "response.web_search_call.completed",
-                "item_id": web_search_call.id,
+                "type": event_type,
+                "item_id": item_id,
                 "output_index": output_index,
                 "item": item.clone()
         });
@@ -470,15 +485,13 @@ mod tests {
         assert!(matches!(decision, LoopDecision::Done));
     }
 
-    // --- Per-call timeout ---------------------------------------------------
-
     use std::pin::Pin;
     use std::sync::Arc;
 
     use serde_json::Value;
 
     use super::execute_gateway_call_with_timeout;
-    use crate::tool::{GatewayExecutor, ToolError, ToolHandler, ToolOutput, ToolRegistry, ToolType};
+    use crate::tool::{GatewayExecutor, GatewayExecutors, ToolError, ToolHandler, ToolOutput, ToolRegistry, ToolType};
     use crate::types::io::OutputItem;
     use crate::types::io::tools::FunctionTool;
     use crate::types::tools::ResponsesTool;
@@ -533,11 +546,11 @@ mod tests {
     async fn hung_gateway_call_times_out_into_error_output() {
         let web_search: ResponsesTool =
             serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).expect("web_search tool param");
-        let registry = ToolRegistry::build_with_handlers(&[web_search], |tool_type| match tool_type {
-            ToolType::WebSearch => Some(Arc::new(SlowExecutor) as Arc<dyn GatewayExecutor>),
-            _ => None,
-        })
-        .expect("registry builds");
+        let mut executors = GatewayExecutors::default();
+        executors.insert(Arc::new(SlowExecutor));
+        let registry = ToolRegistry::build_with_handlers(&[web_search], &executors)
+            .await
+            .expect("registry builds");
 
         // 1ms budget vs a 50ms tool → the timeout fires. Must return (not hang):
         // the stuck call becomes an error output the loop can feed back.
@@ -571,7 +584,9 @@ mod tests {
         // not fail the whole request.
         let web_search: ResponsesTool =
             serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).expect("web_search tool param");
-        let registry = ToolRegistry::build_with_handlers(&[web_search], |_tool_type| None).expect("registry builds");
+        let registry = ToolRegistry::build_with_handlers(&[web_search], &GatewayExecutors::default())
+            .await
+            .expect("registry builds");
 
         let result =
             execute_gateway_call_with_timeout(web_search_call("call_no_handler"), &registry, std::time::Duration::ZERO)

crates/agentic-server-core/src/executor/request.rs

@@ -5,51 +5,10 @@ use crate::config::Config;
 use crate::error::Error;
 use crate::executor::modes::{ConversationHandler, ResponseHandler};
 use crate::storage::{ConversationStore, ResponseStore, create_pool_with_schema_and_sqlite_config};
-use crate::tool::{GatewayExecutor, ToolType, WebSearchHandler};
+use crate::tool::{GatewayExecutor, GatewayExecutors};
 use crate::types::io::InputItem;
 use crate::types::request_response::{RequestPayload, ResponsePayload};
 
-#[derive(Clone, Default)]
-pub struct GatewayExecutors {
-    web_search: Option<Arc<dyn GatewayExecutor>>,
-}
-
-impl GatewayExecutors {
-    #[must_use]
-    pub fn from_env(client: Arc<reqwest::Client>) -> Self {
-        Self {
-            web_search: Some(Arc::new(WebSearchHandler::from_env(client))),
-        }
-    }
-
-    pub fn insert(&mut self, executor: Arc<dyn GatewayExecutor>) {
-        match executor.tool_type() {
-            ToolType::WebSearch => self.web_search = Some(executor),
-            other => tracing::debug!(tool_type = ?other, "gateway executor type not wired yet"),
-        }
-    }
-
-    #[must_use]
-    pub fn get(&self, tool_type: ToolType) -> Option<Arc<dyn GatewayExecutor>> {
-        match tool_type {
-            ToolType::WebSearch => self.web_search.clone(),
-            ToolType::Function
-            | ToolType::CodexNamespace
-            | ToolType::Mcp
-            | ToolType::FileSearch
-            | ToolType::CodeInterpreter => None,
-        }
-    }
-}
-
-impl std::fmt::Debug for GatewayExecutors {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("GatewayExecutors")
-            .field("web_search", &self.web_search.is_some())
-            .finish()
-    }
-}
-
 /// Context built by `rehydrate_conversation`, threaded through the execute pipeline.
 #[derive(Debug)]
 pub struct RequestContext {