diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index c4b94d8b3f..710c854609 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -1938,6 +1938,7 @@ mod tests { thread_id: Some("thread-1".to_string()), server: "server-a".to_string(), uri: "file:///tmp/resource".to_string(), + origin_call_id: None, }, }; assert_eq!( @@ -2116,6 +2117,7 @@ mod tests { thread_id: None, server: "server-a".to_string(), uri: "file:///tmp/resource".to_string(), + origin_call_id: None, }, }; assert_eq!(mcp_resource_read.serialization_scope(), None); diff --git a/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs b/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs index d24da0c879..0efd1d43c1 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs @@ -85,6 +85,9 @@ pub struct McpResourceReadParams { pub thread_id: Option, pub server: String, pub uri: String, + /// The MCP tool call whose app context identifies the resource provider. Requires `threadId`. + #[ts(optional = nullable)] + pub origin_call_id: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] diff --git a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs index 77d8a1439b..787e40db7a 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/tests.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/tests.rs @@ -72,6 +72,19 @@ fn test_absolute_path() -> AbsolutePathBuf { absolute_path("readable") } +#[test] +fn mcp_resource_read_params_deserialize_origin_call_id() { + let params: McpResourceReadParams = serde_json::from_value(json!({ + "threadId": "thread-123", + "server": "codex_apps", + "uri": "test://codex/resource", + "originCallId": "call-123", + })) + .expect("deserialize resource read params"); + + assert_eq!(params.origin_call_id.as_deref(), Some("call-123")); +} + #[test] fn thread_sources_round_trip_as_scalar_labels() { for (source, label) in [ diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index a7986753ee..183f189679 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -234,7 +234,7 @@ Example with notification opt-out: - `tool/requestUserInput` — prompt the user with 1–3 short questions for a tool call and return their answers (experimental). - `config/mcpServer/reload` — reload MCP server config from disk and queue a refresh for loaded threads (applied on each thread's next active turn); returns `{}`. Use this after editing `config.toml` without restarting the server. - `mcpServerStatus/list` — enumerate configured MCP servers with their tools, auth status, server info, plus resources/resource templates for `full` detail; supports optional `threadId` and cursor+limit pagination. If `threadId` is omitted, the server reads from the latest global config directly. If `detail` is omitted, the server defaults to `full`. -- `mcpServer/resource/read` — read a resource from a configured MCP server by optional `threadId`, `server`, and `uri`, returning text/blob resource `contents`. If `threadId` is omitted, the server reads from the latest MCP config directly. +- `mcpServer/resource/read` — read a resource from a configured MCP server by optional `threadId`, `server`, `uri`, and `originCallId`, returning text/blob resource `contents`. When `originCallId` identifies a Codex Apps tool call backed by plugin-runtime, app-server validates the persisted app context and routes the read through that app's `fetch_resource` tool. Other reads continue to use MCP `resources/read`. `originCallId` requires `threadId`; if `threadId` is omitted, the server reads from the latest MCP config directly. - `mcpServer/tool/call` — call a tool on a thread's configured MCP server by `threadId`, `server`, `tool`, optional `arguments`, and optional `_meta`, returning the MCP tool result. - `windowsSandbox/setupStart` — start Windows sandbox setup for the selected mode (`elevated` or `unelevated`); accepts an optional absolute `cwd` to target setup for a specific workspace, returns `{ started: true }` immediately, and later emits `windowsSandbox/setupCompleted`. - `feedback/upload` — submit a feedback report (classification + optional reason/logs, conversation_id, and optional `extraLogFiles` attachments array); returns the tracking thread id. diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 6ae73784ff..20edb4184e 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -458,6 +458,7 @@ impl MessageProcessor { let mcp_processor = McpRequestProcessor::new( auth_manager.clone(), Arc::clone(&thread_manager), + thread_state_manager.clone(), outgoing.clone(), config_manager.clone(), ); diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index f03351468f..52e3f02e3e 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -508,6 +508,7 @@ mod git_processor; mod initialize_processor; mod marketplace_processor; mod mcp_processor; +mod mcp_resource_read; mod plugins; mod process_exec_processor; mod remote_control_processor; diff --git a/codex-rs/app-server/src/request_processors/mcp_processor.rs b/codex-rs/app-server/src/request_processors/mcp_processor.rs index ac2125c2cf..900337053a 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/codex-rs/app-server/src/request_processors/mcp_processor.rs @@ -6,6 +6,7 @@ const MCP_TOOL_THREAD_ID_META_KEY: &str = "threadId"; pub(crate) struct McpRequestProcessor { auth_manager: Arc, thread_manager: Arc, + thread_state_manager: ThreadStateManager, outgoing: Arc, config_manager: ConfigManager, } @@ -14,12 +15,14 @@ impl McpRequestProcessor { pub(crate) fn new( auth_manager: Arc, thread_manager: Arc, + thread_state_manager: ThreadStateManager, outgoing: Arc, config_manager: ConfigManager, ) -> Self { Self { auth_manager, thread_manager, + thread_state_manager, outgoing, config_manager, } @@ -393,19 +396,34 @@ impl McpRequestProcessor { thread_id, server, uri, + origin_call_id, } = params; - if let Some(thread_id) = thread_id { - let (_, thread) = self.load_thread(&thread_id).await?; + if let Some(thread_id_string) = thread_id { + let (thread_id, thread) = self.load_thread(&thread_id_string).await?; + let thread_state_manager = self.thread_state_manager.clone(); let request_id = request_id.clone(); tokio::spawn(async move { - let result = thread.read_mcp_resource(&server, &uri).await; + let result = mcp_resource_read::read_thread_mcp_resource( + &thread_state_manager, + thread_id, + &thread_id_string, + &thread, + &server, + &uri, + origin_call_id.as_deref(), + ) + .await; Self::send_mcp_resource_read_response(outgoing, request_id, result).await; }); return Ok(()); } + if origin_call_id.is_some() { + return Err(invalid_request("originCallId requires threadId")); + } + let config = self.load_latest_config(/*fallback_cwd*/ None).await?; let mcp_manager = self.thread_manager.mcp_manager(); let mcp_config = mcp_manager.runtime_config(&config).await; @@ -475,7 +493,7 @@ impl McpRequestProcessor { } } -fn with_mcp_tool_call_thread_id_meta( +pub(super) fn with_mcp_tool_call_thread_id_meta( meta: Option, thread_id: &str, ) -> Option { diff --git a/codex-rs/app-server/src/request_processors/mcp_resource_read.rs b/codex-rs/app-server/src/request_processors/mcp_resource_read.rs new file mode 100644 index 0000000000..7f6c407629 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/mcp_resource_read.rs @@ -0,0 +1,300 @@ +use super::mcp_processor::with_mcp_tool_call_thread_id_meta; +use super::*; +use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; +use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY; +use codex_mcp::ToolInfo; + +const LEGACY_CODEX_APPS_MCP_SERVER_NAME: &str = "codex-connectors-mcp"; +const PLUGIN_RUNTIME_MCP_SERVER_NAME: &str = "plugin-runtime"; +const FETCH_RESOURCE_TOOL_NAME: &str = "fetch_resource"; + +pub(super) async fn read_thread_mcp_resource( + thread_state_manager: &ThreadStateManager, + thread_id: ThreadId, + thread_id_string: &str, + thread: &Arc, + server: &str, + uri: &str, + origin_call_id: Option<&str>, +) -> anyhow::Result { + let Some(origin_call_id) = origin_call_id else { + return thread.read_mcp_resource(server, uri).await; + }; + if server != CODEX_APPS_MCP_SERVER_NAME { + return thread.read_mcp_resource(server, uri).await; + } + + let server_info = thread + .mcp_server_info(server) + .await + .ok_or_else(|| anyhow::anyhow!("unable to identify codex_apps MCP server"))?; + match server_info.name.as_str() { + LEGACY_CODEX_APPS_MCP_SERVER_NAME => return thread.read_mcp_resource(server, uri).await, + PLUGIN_RUNTIME_MCP_SERVER_NAME => {} + implementation => { + anyhow::bail!("unsupported codex_apps MCP server implementation `{implementation}`") + } + } + + let origin = + resolve_mcp_resource_origin(thread_state_manager, thread_id, thread, origin_call_id) + .await?; + if origin.server != server { + anyhow::bail!( + "originating MCP tool call server `{}` does not match resource server `{server}`", + origin.server + ); + } + if origin.resource_uri.as_deref() != Some(uri) { + anyhow::bail!( + "originating MCP tool call resource URI {:?} does not match requested URI `{uri}`", + origin.resource_uri + ); + } + + let tool_info = thread + .mcp_tool_info(server, &origin.tool) + .await + .ok_or_else(|| { + anyhow::anyhow!( + "originating MCP tool `{}` is not available on server `{server}`", + origin.tool + ) + })?; + let meta = build_plugin_runtime_fetch_resource_meta(&origin, uri, &tool_info)?; + let meta = with_mcp_tool_call_thread_id_meta(Some(meta), thread_id_string); + let result = thread + .call_mcp_tool( + server, + FETCH_RESOURCE_TOOL_NAME, + Some(serde_json::json!({ "uri": uri })), + meta, + ) + .await?; + plugin_runtime_fetch_resource_response(result) +} + +async fn resolve_mcp_resource_origin( + thread_state_manager: &ThreadStateManager, + thread_id: ThreadId, + thread: &Arc, + origin_call_id: &str, +) -> anyhow::Result { + let active_turn = { + let thread_state = thread_state_manager.thread_state(thread_id).await; + thread_state.lock().await.active_turn_snapshot() + }; + if let Some(origin) = active_turn + .as_ref() + .and_then(|turn| find_mcp_resource_origin(&turn.items, origin_call_id)) + { + return origin; + } + + let history = thread.load_history(/*include_archived*/ true).await?; + let turns = build_api_turns_from_rollout_items(&history.items); + turns + .iter() + .rev() + .find_map(|turn| find_mcp_resource_origin(&turn.items, origin_call_id)) + .unwrap_or_else(|| { + Err(anyhow::anyhow!( + "originating MCP tool call `{origin_call_id}` was not found" + )) + }) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct McpResourceOrigin { + server: String, + tool: String, + connector_id: String, + link_id: Option, + resource_uri: Option, +} + +fn find_mcp_resource_origin( + items: &[ThreadItem], + origin_call_id: &str, +) -> Option> { + items.iter().find_map(|item| { + let ThreadItem::McpToolCall { + id, + server, + tool, + app_context, + .. + } = item + else { + return None; + }; + if id != origin_call_id { + return None; + } + Some(match app_context { + Some(app_context) => Ok(McpResourceOrigin { + server: server.clone(), + tool: tool.clone(), + connector_id: app_context.connector_id.clone(), + link_id: app_context.link_id.clone(), + resource_uri: app_context.resource_uri.clone(), + }), + None => Err(anyhow::anyhow!( + "originating MCP tool call `{origin_call_id}` has no app context" + )), + }) + }) +} + +fn build_plugin_runtime_fetch_resource_meta( + origin: &McpResourceOrigin, + uri: &str, + tool_info: &ToolInfo, +) -> anyhow::Result { + let tool_meta = tool_info + .tool + .meta + .as_ref() + .ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing metadata"))?; + let connector_id = tool_info + .connector_id + .as_deref() + .ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing connector_id"))?; + let link_id = tool_meta + .0 + .get("link_id") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing link_id"))?; + let codex_apps_meta = tool_meta + .0 + .get(MCP_TOOL_CODEX_APPS_META_KEY) + .and_then(serde_json::Value::as_object) + .ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing _codex_apps metadata"))?; + let resource_uri = mcp_app_resource_uri(&tool_meta.0) + .ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing its app resource URI"))?; + if resource_uri != uri { + anyhow::bail!( + "originating MCP tool resource URI `{resource_uri}` does not match requested URI `{uri}`" + ); + } + if connector_id != origin.connector_id { + anyhow::bail!( + "originating MCP tool connector `{connector_id}` does not match app context connector `{}`", + origin.connector_id + ); + } + + let synthetic_link = codex_apps_meta + .get("synthetic_link") + .and_then(serde_json::Value::as_bool) + == Some(true); + match origin.link_id.as_deref() { + Some(origin_link_id) if origin_link_id == link_id => {} + None if synthetic_link => {} + _ => anyhow::bail!("originating MCP tool link does not match app context link"), + } + + let tool_resource_route = codex_apps_meta + .get("resource_uri") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| anyhow::anyhow!("originating MCP tool is missing its resource route"))?; + let route_segments = tool_resource_route + .strip_prefix('/') + .map(|route| route.split('/').collect::>()) + .filter(|segments| segments.len() == 3 && segments.iter().all(|part| !part.is_empty())) + .ok_or_else(|| anyhow::anyhow!("originating MCP tool has an invalid resource route"))?; + if route_segments[0] != connector_id || route_segments[1] != link_id { + anyhow::bail!("originating MCP tool resource route does not match connector and link"); + } + + let mut fetch_resource_meta = serde_json::Map::from_iter([ + ( + "resource_uri".to_string(), + serde_json::Value::String(format!( + "/{connector_id}/{link_id}/{FETCH_RESOURCE_TOOL_NAME}" + )), + ), + ( + "contains_mcp_source".to_string(), + serde_json::Value::Bool( + codex_apps_meta + .get("contains_mcp_source") + .and_then(serde_json::Value::as_bool) + == Some(true), + ), + ), + ]); + if synthetic_link { + fetch_resource_meta.insert("synthetic_link".to_string(), serde_json::Value::Bool(true)); + } + + let mut meta = serde_json::Map::from_iter([ + ( + MCP_TOOL_CODEX_APPS_META_KEY.to_string(), + serde_json::Value::Object(fetch_resource_meta), + ), + ( + "x-codex-turn-metadata".to_string(), + serde_json::json!({ + "mcp_request_meta": { + "selected_connector_ids": [connector_id], + } + }), + ), + ]); + if let Some(connector_name) = tool_info.connector_name.as_ref() { + meta.insert( + "connector_name".to_string(), + serde_json::Value::String(connector_name.clone()), + ); + } + if let Some(connector_description) = tool_info.namespace_description.as_ref() { + meta.insert( + "connector_description".to_string(), + serde_json::Value::String(connector_description.clone()), + ); + } + Ok(serde_json::Value::Object(meta)) +} + +fn mcp_app_resource_uri(meta: &serde_json::Map) -> Option<&str> { + meta.get("ui") + .and_then(serde_json::Value::as_object) + .and_then(|ui| ui.get("resourceUri")) + .and_then(serde_json::Value::as_str) + .or_else(|| { + meta.get("ui/resourceUri") + .and_then(serde_json::Value::as_str) + }) + .or_else(|| { + meta.get("openai/outputTemplate") + .and_then(serde_json::Value::as_str) + }) +} + +fn plugin_runtime_fetch_resource_response( + result: codex_protocol::mcp::CallToolResult, +) -> anyhow::Result { + if result.is_error == Some(true) { + anyhow::bail!("plugin-runtime fetch_resource returned an error"); + } + let mut structured_content = result + .structured_content + .ok_or_else(|| anyhow::anyhow!("plugin-runtime fetch_resource returned no contents"))?; + let contents = structured_content + .get_mut("contents") + .and_then(serde_json::Value::as_array_mut) + .ok_or_else(|| { + anyhow::anyhow!("plugin-runtime fetch_resource returned invalid contents") + })?; + for content in contents { + if let Some(content) = content.as_object_mut() + && !content.contains_key("_meta") + && let Some(meta) = content.remove("meta") + { + content.insert("_meta".to_string(), meta); + } + } + let response = serde_json::from_value::(structured_content)?; + Ok(serde_json::to_value(response)?) +} diff --git a/codex-rs/app-server/tests/suite/v2/mcp_resource.rs b/codex-rs/app-server/tests/suite/v2/mcp_resource.rs index 1bdf06161f..62a2b53f18 100644 --- a/codex-rs/app-server/tests/suite/v2/mcp_resource.rs +++ b/codex-rs/app-server/tests/suite/v2/mcp_resource.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::sync::Mutex; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::time::Duration; @@ -84,9 +85,9 @@ const SKILLS_READ_CALL_ID: &str = "skills-read"; const SKILLS_READ_AGAIN_CALL_ID: &str = "skills-read-again"; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn mcp_resource_read_returns_resource_contents() -> Result<()> { +async fn mcp_resource_read_without_origin_uses_resources_read() -> Result<()> { let responses_server = responses::start_mock_server().await; - let (apps_server_url, _apps_server_calls, apps_server_handle) = + let (apps_server_url, apps_server_calls, apps_server_handle) = start_resource_apps_mcp_server().await?; let responses_server_uri = responses_server.uri(); let (_codex_home, mut mcp) = @@ -110,6 +111,7 @@ async fn mcp_resource_read_returns_resource_contents() -> Result<()> { thread_id: Some(thread.id), server: "codex_apps".to_string(), uri: TEST_RESOURCE_URI.to_string(), + origin_call_id: None, }) .await?; let read_response: JSONRPCResponse = timeout( @@ -121,6 +123,7 @@ async fn mcp_resource_read_returns_resource_contents() -> Result<()> { to_response::(read_response)?, expected_resource_read_response() ); + assert_eq!(apps_server_calls.resource_read_metas(), vec![Meta::new()]); apps_server_handle.abort(); let _ = apps_server_handle.await; @@ -526,8 +529,8 @@ enabled = false } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn mcp_resource_read_returns_resource_contents_without_thread() -> Result<()> { - let (apps_server_url, _apps_server_calls, apps_server_handle) = +async fn mcp_resource_read_without_thread_uses_resources_read() -> Result<()> { + let (apps_server_url, apps_server_calls, apps_server_handle) = start_resource_apps_mcp_server().await?; let codex_home = TempDir::new()?; @@ -560,6 +563,7 @@ apps = true thread_id: None, server: "codex_apps".to_string(), uri: TEST_RESOURCE_URI.to_string(), + origin_call_id: None, }) .await?; let read_response: JSONRPCResponse = timeout( @@ -572,6 +576,7 @@ apps = true to_response::(read_response)?, expected_resource_read_response() ); + assert_eq!(apps_server_calls.resource_read_metas(), vec![Meta::new()]); apps_server_handle.abort(); let _ = apps_server_handle.await; @@ -624,6 +629,7 @@ async fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> { thread_id: Some("00000000-0000-4000-8000-000000000000".to_string()), server: "codex_apps".to_string(), uri: TEST_RESOURCE_URI.to_string(), + origin_call_id: None, }, }) .await; @@ -746,6 +752,7 @@ struct ResourceAppsMcpCalls { list_resources: AtomicUsize, main_prompt_reads: AtomicUsize, reference_reads: AtomicUsize, + resource_read_metas: Mutex>, } impl ResourceAppsMcpCalls { @@ -756,6 +763,13 @@ impl ResourceAppsMcpCalls { reference_reads: self.reference_reads.load(Ordering::Relaxed), } } + + fn resource_read_metas(&self) -> Vec { + self.resource_read_metas + .lock() + .expect("resource read metadata lock poisoned") + .clone() + } } #[derive(Debug, PartialEq, Eq)] @@ -827,8 +841,17 @@ impl ServerHandler for ResourceAppsMcpServer { async fn read_resource( &self, request: ReadResourceRequestParams, - _context: RequestContext, + context: RequestContext, ) -> Result { + let mut meta = context.meta; + // rmcp adds a transport progress token to outgoing requests. Capture + // only the client-supplied metadata that this test is exercising. + meta.0.remove("progressToken"); + self.calls + .resource_read_metas + .lock() + .expect("resource read metadata lock poisoned") + .push(meta); let uri = request.uri; if uri == SKILL_MAIN_PROMPT_URI { self.calls.main_prompt_reads.fetch_add(1, Ordering::Relaxed); diff --git a/codex-rs/app-server/tests/suite/v2/mcp_resource_origin.rs b/codex-rs/app-server/tests/suite/v2/mcp_resource_origin.rs new file mode 100644 index 0000000000..0c117a86ce --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/mcp_resource_origin.rs @@ -0,0 +1,377 @@ +use std::borrow::Cow; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use anyhow::Result; +use app_test_support::ChatGptAuthFixture; +use app_test_support::TestAppServer; +use app_test_support::to_response; +use app_test_support::write_chatgpt_auth; +use axum::Router; +use codex_app_server_protocol::ItemCompletedNotification; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::McpResourceContent; +use codex_app_server_protocol::McpResourceReadParams; +use codex_app_server_protocol::McpResourceReadResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::UserInput; +use codex_config::types::AuthCredentialsStoreMode; +use core_test_support::responses; +use pretty_assertions::assert_eq; +use rmcp::handler::server::ServerHandler; +use rmcp::model::CallToolRequestParams; +use rmcp::model::CallToolResult; +use rmcp::model::Content; +use rmcp::model::Implementation; +use rmcp::model::JsonObject; +use rmcp::model::ListToolsResult; +use rmcp::model::Meta; +use rmcp::model::ProtocolVersion; +use rmcp::model::ServerCapabilities; +use rmcp::model::ServerInfo; +use rmcp::model::Tool; +use rmcp::model::ToolAnnotations; +use rmcp::service::RequestContext; +use rmcp::service::RoleServer; +use rmcp::transport::StreamableHttpServerConfig; +use rmcp::transport::StreamableHttpService; +use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; +use serde_json::Value; +use serde_json::json; +use tempfile::TempDir; +use tokio::net::TcpListener; +use tokio::task::JoinHandle; +use tokio::time::timeout; + +const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10); +const CONNECTOR_ID: &str = "walmart"; +const CONNECTOR_NAME: &str = "Walmart"; +const LINK_ID: &str = "link_walmart"; +const ORIGIN_CALL_ID: &str = "walmart-product-search-call"; +const PRODUCT_SEARCH_TOOL: &str = "walmart_product_search"; +const PRODUCT_SEARCH_MODEL_NAMESPACE: &str = "mcp__codex_apps__walmart"; +const PRODUCT_SEARCH_MODEL_TOOL: &str = "_product_search"; +const RESOURCE_URI: &str = "ui://widget/product-search.html"; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn mcp_resource_read_with_origin_calls_plugin_runtime_fetch_resource() -> Result<()> { + let responses_server = responses::start_mock_server().await; + let (apps_server_url, recorded_calls, apps_server_handle) = + start_plugin_runtime_mcp_server().await?; + let (_codex_home, mut mcp) = + start_test_app_server(&apps_server_url, &responses_server.uri()).await?; + + let response_mock = responses::mount_sse_sequence( + &responses_server, + vec![ + responses::sse(vec![ + responses::ev_response_created("resp-product-search"), + responses::ev_function_call_with_namespace( + ORIGIN_CALL_ID, + PRODUCT_SEARCH_MODEL_NAMESPACE, + PRODUCT_SEARCH_MODEL_TOOL, + &json!({ "query": "bed lamps" }).to_string(), + ), + responses::ev_completed("resp-product-search"), + ]), + responses::sse(vec![ + responses::ev_response_created("resp-done"), + responses::ev_assistant_message("msg-done", "Done"), + responses::ev_completed("resp-done"), + ]), + ], + ) + .await; + + let thread_start_id = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_start_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(thread_start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?; + + let turn_start_id = mcp + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "Find bed lamps".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(turn_start_id)), + ) + .await??; + + let completed = wait_for_mcp_tool_call_completed(&mut mcp, ORIGIN_CALL_ID).await?; + let ThreadItem::McpToolCall { + app_context: Some(app_context), + .. + } = completed.item + else { + anyhow::bail!("originating MCP tool call should include app context"); + }; + assert_eq!(app_context.connector_id, CONNECTOR_ID); + assert_eq!(app_context.link_id.as_deref(), Some(LINK_ID)); + assert_eq!(app_context.resource_uri.as_deref(), Some(RESOURCE_URI)); + + timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let read_request_id = mcp + .send_mcp_resource_read_request(McpResourceReadParams { + thread_id: Some(thread.id.clone()), + server: "codex_apps".to_string(), + uri: RESOURCE_URI.to_string(), + origin_call_id: Some(ORIGIN_CALL_ID.to_string()), + }) + .await?; + let read_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(read_request_id)), + ) + .await??; + + assert_eq!( + to_response::(read_response)?, + McpResourceReadResponse { + contents: vec![McpResourceContent::Text { + uri: RESOURCE_URI.to_string(), + mime_type: Some("text/html".to_string()), + text: "Walmart".to_string(), + meta: Some(json!({ "app": CONNECTOR_ID })), + }], + } + ); + + { + let calls = recorded_calls + .lock() + .expect("recorded tool calls lock poisoned"); + let fetch_resource_call = calls + .iter() + .find(|call| call["name"] == "fetch_resource") + .expect("resource read should call fetch_resource"); + assert_eq!( + fetch_resource_call["arguments"], + json!({ "uri": RESOURCE_URI }) + ); + assert_eq!( + fetch_resource_call.pointer("/meta/_codex_apps"), + Some(&json!({ + "resource_uri": format!("/{CONNECTOR_ID}/{LINK_ID}/fetch_resource"), + "contains_mcp_source": true, + })) + ); + assert_eq!( + fetch_resource_call + .pointer("/meta/x-codex-turn-metadata/mcp_request_meta/selected_connector_ids"), + Some(&json!([CONNECTOR_ID])) + ); + assert_eq!(fetch_resource_call["meta"]["threadId"], thread.id); + } + + assert_eq!(response_mock.requests().len(), 2); + apps_server_handle.abort(); + let _ = apps_server_handle.await; + Ok(()) +} + +async fn start_test_app_server( + apps_server_url: &str, + responses_server_uri: &str, +) -> Result<(TempDir, TestAppServer)> { + let codex_home = TempDir::new()?; + std::fs::write( + codex_home.path().join("config.toml"), + format!( + r#" +model = "mock-model" +approval_policy = "untrusted" +sandbox_mode = "read-only" + +model_provider = "mock_provider" +chatgpt_base_url = "{apps_server_url}" +mcp_oauth_credentials_store = "file" + +[features] +apps = true + +[model_providers.mock_provider] +name = "Mock provider for test" +base_url = "{responses_server_uri}/v1" +wire_api = "responses" +request_max_retries = 0 +stream_max_retries = 0 +"# + ), + )?; + write_chatgpt_auth( + codex_home.path(), + ChatGptAuthFixture::new("chatgpt-token") + .account_id("account-123") + .chatgpt_user_id("user-123") + .chatgpt_account_id("account-123"), + AuthCredentialsStoreMode::File, + )?; + + let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + Ok((codex_home, mcp)) +} + +async fn start_plugin_runtime_mcp_server() +-> Result<(String, Arc>>, JoinHandle<()>)> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + let apps_server_url = format!("http://{addr}"); + let calls = Arc::new(Mutex::new(Vec::new())); + let server_calls = Arc::clone(&calls); + let mcp_service = StreamableHttpService::new( + move || { + Ok(PluginRuntimeMcpServer { + calls: Arc::clone(&server_calls), + }) + }, + Arc::new(LocalSessionManager::default()), + StreamableHttpServerConfig::default(), + ); + let router = Router::new().nest_service("/api/codex/ps/mcp", mcp_service); + let handle = tokio::spawn(async move { + let _ = axum::serve(listener, router).await; + }); + + Ok((apps_server_url, calls, handle)) +} + +#[derive(Clone)] +struct PluginRuntimeMcpServer { + calls: Arc>>, +} + +impl ServerHandler for PluginRuntimeMcpServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) + .with_protocol_version(ProtocolVersion::V_2025_06_18) + .with_server_info(Implementation::new("plugin-runtime", "1.0.0")) + } + + async fn list_tools( + &self, + _request: Option, + _context: RequestContext, + ) -> Result { + let input_schema: JsonObject = serde_json::from_value(json!({ + "type": "object", + "properties": { "query": { "type": "string" } }, + "required": ["query"], + "additionalProperties": false + })) + .map_err(|error| rmcp::ErrorData::internal_error(error.to_string(), None))?; + let mut tool = Tool::new( + Cow::Borrowed(PRODUCT_SEARCH_TOOL), + Cow::Borrowed("Search Walmart products."), + Arc::new(input_schema), + ); + tool.annotations = Some(ToolAnnotations::new().read_only(true)); + tool.meta = Some(Meta(serde_json::Map::from_iter([ + ("connector_id".to_string(), json!(CONNECTOR_ID)), + ("connector_name".to_string(), json!(CONNECTOR_NAME)), + ( + "connector_description".to_string(), + json!("Search Walmart products."), + ), + ("link_id".to_string(), json!(LINK_ID)), + ("openai/outputTemplate".to_string(), json!(RESOURCE_URI)), + ( + "_codex_apps".to_string(), + json!({ + "resource_uri": format!( + "/{CONNECTOR_ID}/{LINK_ID}/{PRODUCT_SEARCH_TOOL}" + ), + "contains_mcp_source": true, + }), + ), + ]))); + + Ok(ListToolsResult { + tools: vec![tool], + next_cursor: None, + meta: None, + }) + } + + async fn call_tool( + &self, + request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + let mut meta = context.meta; + meta.0.remove("progressToken"); + self.calls + .lock() + .expect("recorded tool calls lock poisoned") + .push(json!({ + "name": request.name, + "arguments": request.arguments, + "meta": meta, + })); + + match request.name.as_ref() { + PRODUCT_SEARCH_TOOL => Ok(CallToolResult::structured(json!({ "products": [] }))), + "fetch_resource" => { + let mut result = CallToolResult::structured(json!({ + "contents": [{ + "uri": RESOURCE_URI, + "mimeType": "text/html", + "text": "Walmart", + "meta": { "app": CONNECTOR_ID }, + }] + })); + result.content = vec![Content::text("Fetched Walmart resource")]; + Ok(result) + } + name => Err(rmcp::ErrorData::invalid_params( + format!("unknown tool: {name}"), + None, + )), + } + } +} + +async fn wait_for_mcp_tool_call_completed( + mcp: &mut TestAppServer, + call_id: &str, +) -> Result { + loop { + let notification = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_notification_message("item/completed"), + ) + .await??; + let Some(params) = notification.params else { + continue; + }; + let completed: ItemCompletedNotification = serde_json::from_value(params)?; + if matches!(&completed.item, ThreadItem::McpToolCall { id, .. } if id == call_id) { + return Ok(completed); + } + } +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index bdb03271d0..6ec12c8cab 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -31,6 +31,7 @@ mod marketplace_add; mod marketplace_remove; mod marketplace_upgrade; mod mcp_resource; +mod mcp_resource_origin; mod mcp_server_elicitation; mod mcp_server_status; mod mcp_tool; diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index 2143e3b577..f437c17b2d 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -856,6 +856,11 @@ impl McpConnectionManager { server_infos } + /// Returns available presentation metadata for one server. + pub async fn server_info(&self, server: &str) -> Option { + self.list_available_server_infos().await.remove(server) + } + fn with_server_metadata(&self, mut tool: ToolInfo) -> ToolInfo { let Some(metadata) = self.server_metadata.get(&tool.server_name) else { tool.supports_parallel_tool_calls = false; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index c80268e2f9..68a8cf2fc1 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -644,6 +644,29 @@ impl CodexThread { Ok(serde_json::to_value(result)?) } + pub async fn mcp_server_info( + &self, + server: &str, + ) -> Option { + self.current_mcp_runtime() + .await + .manager() + .server_info(server) + .await + } + + pub async fn mcp_tool_info(&self, server: &str, tool: &str) -> Option { + self.current_mcp_runtime() + .await + .manager() + .list_all_tools() + .await + .into_iter() + .find(|tool_info| { + tool_info.server_name == server && tool_info.tool.name.as_ref() == tool + }) + } + pub async fn call_mcp_tool( &self, server: &str,