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-core/Cargo.toml

@@ -14,6 +14,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-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()),
     }
 }
@@ -192,3 +196,33 @@ fn extract_reasoning_done(json: &Value) -> EventPayload {
         item_id: json_str(json, "item_id"),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn mcp_tool_call_events_are_classified() {
+        let added = normalize_sse_line(
+            r#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"mcp_tool_call","id":"mcp_1"}}"#,
+        )
+        .unwrap();
+        assert_eq!(added.event_type, SSEEventType::OutputItemAdded);
+        let EventPayload::OutputItemAdded { item_type, .. } = added.payload else {
+            panic!("expected output item added payload");
+        };
+        assert_eq!(item_type, SSEItemType::McpToolCall);
+
+        let in_progress = normalize_sse_line(
+            r#"data: {"type":"response.mcp_tool_call.in_progress","item_id":"mcp_1","output_index":0}"#,
+        )
+        .unwrap();
+        assert_eq!(in_progress.event_type, SSEEventType::McpToolCallInProgress);
+
+        let completed = normalize_sse_line(
+            r#"data: {"type":"response.mcp_tool_call.completed","item_id":"mcp_1","output_index":0,"item":{"type":"mcp_tool_call","id":"mcp_1"}}"#,
+        )
+        .unwrap();
+        assert_eq!(completed.event_type, SSEEventType::McpToolCallCompleted);
+    }
+}

crates/agentic-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-core/src/executor/accumulator.rs

@@ -245,11 +245,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);
@@ -479,6 +487,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-core/src/executor/engine.rs

@@ -22,9 +22,10 @@ use crate::executor::persist::persist_if_needed;
 use crate::executor::rehydrate::rehydrate_conversation;
 use crate::executor::request::{ExecutionContext, RequestContext};
 use crate::executor::upstream::{fetch_blocking_payload, fetch_stream_payload};
-use crate::tool::ToolRegistry;
+use crate::tool::{GatewayExecutor, McpClientPool, McpHandler, ToolRegistry, ToolType};
 use crate::types::io::{ResponseUsage, ToolChoice};
 use crate::types::request_response::{RequestPayload, ResponsePayload};
+use crate::types::tools::ResponsesTool;
 use crate::utils::common::serialize_to_string;
 
 pub use crate::executor::inference::BoxStream;
@@ -63,6 +64,30 @@ fn error_sse_chunk(message: &str) -> String {
     format!("data: {event_json}\n\n")
 }
 
+async fn build_gateway_registry(tools: &[ResponsesTool], exec_ctx: &ExecutionContext) -> ToolRegistry {
+    let mcp_params: Vec<_> = tools
+        .iter()
+        .filter_map(|tool| match tool {
+            ResponsesTool::Mcp(param) => Some(param.clone()),
+            _ => None,
+        })
+        .collect();
+
+    let request_mcp_executor = if mcp_params.is_empty() {
+        None
+    } else {
+        let pool = Arc::new(McpClientPool::from_params(&mcp_params).await);
+        Some(Arc::new(McpHandler::read_resource(pool)) as Arc<dyn GatewayExecutor>)
+    };
+
+    ToolRegistry::build_with_handlers(tools, |tool_type| match tool_type {
+        ToolType::Mcp => request_mcp_executor
+            .clone()
+            .or_else(|| exec_ctx.gateway_executors.get(tool_type)),
+        _ => exec_ctx.gateway_executors.get(tool_type),
+    })
+}
+
 struct AbortOnDrop<T> {
     handle: tokio::task::JoinHandle<T>,
 }
@@ -102,13 +127,10 @@ async fn run_until_gateway_tools_complete(
     stream_upstream: bool,
     stream_events: Option<&mpsc::UnboundedSender<String>>,
 ) -> ExecutorResult<(ResponsePayload, RequestContext)> {
-    let registry = ctx
-        .enriched_request
-        .tools
-        .as_ref()
-        .map_or_else(ToolRegistry::default, |tools| {
-            ToolRegistry::build_with_handlers(tools, |tool_type| exec_ctx.gateway_executors.get(tool_type))
-        });
+    let registry = match ctx.enriched_request.tools.as_ref() {
+        Some(tools) => build_gateway_registry(tools, exec_ctx).await,
+        None => ToolRegistry::default(),
+    };
     let mut combined_output = Vec::new();
     let mut combined_usage = None;
 

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

@@ -5,7 +5,7 @@ use tokio::sync::mpsc;
 use crate::executor::error::{ExecutorError, ExecutorResult};
 use crate::executor::request::RequestContext;
 use crate::tool::{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;
 
@@ -62,9 +62,9 @@ async fn execute_gateway_call(call: FunctionToolCall, registry: &ToolRegistry) -
         )));
     };
     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);
@@ -79,11 +79,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::Mcp | ToolType::FileSearch | ToolType::CodeInterpreter => None,
+        ToolType::Mcp => Some(crate::tool::mcp::handler::output_item(call, output, status)),
+        ToolType::Function | ToolType::FileSearch | ToolType::CodeInterpreter => None,
     }
 }
 
@@ -148,7 +149,8 @@ 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::Function | ToolType::Mcp | ToolType::FileSearch | ToolType::CodeInterpreter => None,
+                    ToolType::Mcp => Some(crate::tool::mcp::handler::started_output_item(call)),
+                    ToolType::Function | ToolType::FileSearch | ToolType::CodeInterpreter => None,
                 },
             });
         }
@@ -179,28 +181,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(())
 }
@@ -214,18 +226,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()
         });