From 1132ef887c37a5780d6d541e2913c6c126eeb026 Mon Sep 17 00:00:00 2001 From: Matthew Zeng Date: Mon, 20 Apr 2026 19:59:36 -0700 Subject: [PATCH] Make MCP resource read threadless (#18292) ## Summary Making thread id optional so that we can better cache resources for MCPs for connectors since their resource templates is universal and not particular to projects. - Make `mcpServer/resource/read` accept an optional `threadId` - Read resources from the current MCP config when no thread is supplied - Keep the existing thread-scoped path when `threadId` is present - Update the generated schemas, README, and integration coverage ## Testing - `just write-app-server-schema` - `just fmt` - `cargo test -p codex-app-server-protocol` - `cargo test -p codex-mcp` - `cargo test -p codex-app-server --test all mcp_resource` - `just fix -p codex-mcp` - `just fix -p codex-app-server-protocol` - `just fix -p codex-app-server` --- .../schema/json/ClientRequest.json | 6 +- .../codex_app_server_protocol.schemas.json | 6 +- .../codex_app_server_protocol.v2.schemas.json | 6 +- .../schema/json/v2/McpResourceReadParams.json | 6 +- .../typescript/v2/McpResourceReadParams.ts | 2 +- .../app-server-protocol/src/protocol/v2.rs | 3 +- codex-rs/app-server/README.md | 2 +- .../app-server/src/codex_message_processor.rs | 116 ++++++++++++---- .../app-server/tests/suite/v2/mcp_resource.rs | 125 +++++++++++++----- codex-rs/codex-mcp/src/lib.rs | 1 + codex-rs/codex-mcp/src/mcp/mod.rs | 43 ++++++ 11 files changed, 249 insertions(+), 67 deletions(-) diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index f9cdca83e..93d21d7e4 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -1304,7 +1304,10 @@ "type": "string" }, "threadId": { - "type": "string" + "type": [ + "string", + "null" + ] }, "uri": { "type": "string" @@ -1312,7 +1315,6 @@ }, "required": [ "server", - "threadId", "uri" ], "type": "object" diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 42ca07196..3f0293828 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -9685,7 +9685,10 @@ "type": "string" }, "threadId": { - "type": "string" + "type": [ + "string", + "null" + ] }, "uri": { "type": "string" @@ -9693,7 +9696,6 @@ }, "required": [ "server", - "threadId", "uri" ], "title": "McpResourceReadParams", diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index d00378de5..db253edfc 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -6169,7 +6169,10 @@ "type": "string" }, "threadId": { - "type": "string" + "type": [ + "string", + "null" + ] }, "uri": { "type": "string" @@ -6177,7 +6180,6 @@ }, "required": [ "server", - "threadId", "uri" ], "title": "McpResourceReadParams", diff --git a/codex-rs/app-server-protocol/schema/json/v2/McpResourceReadParams.json b/codex-rs/app-server-protocol/schema/json/v2/McpResourceReadParams.json index 0242d4148..2fe58155d 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/McpResourceReadParams.json +++ b/codex-rs/app-server-protocol/schema/json/v2/McpResourceReadParams.json @@ -5,7 +5,10 @@ "type": "string" }, "threadId": { - "type": "string" + "type": [ + "string", + "null" + ] }, "uri": { "type": "string" @@ -13,7 +16,6 @@ }, "required": [ "server", - "threadId", "uri" ], "title": "McpResourceReadParams", diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/McpResourceReadParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/McpResourceReadParams.ts index 51d650d9b..c48795f27 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/McpResourceReadParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/McpResourceReadParams.ts @@ -2,4 +2,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type McpResourceReadParams = { threadId: string, server: string, uri: string, }; +export type McpResourceReadParams = { threadId?: string | null, server: string, uri: string, }; diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index a029860f2..0bdaaa367 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -2241,7 +2241,8 @@ pub struct ListMcpServerStatusResponse { #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct McpResourceReadParams { - pub thread_id: String, + #[ts(optional = nullable)] + pub thread_id: Option, pub server: String, pub uri: String, } diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 5b596dea4..57cb6c173 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -197,7 +197,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 and auth status, plus resources/resource templates for `full` detail; supports cursor+limit pagination. If `detail` is omitted, the server defaults to `full`. -- `mcpServer/resource/read` — read a resource from a thread's configured MCP server by `threadId`, `server`, and `uri`, returning text/blob resource `contents`. +- `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/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/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 52a3d2fcb..34c71ac3e 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -290,6 +290,7 @@ use codex_mcp::McpSnapshotDetail; use codex_mcp::collect_mcp_server_status_snapshot_with_detail_and_authorization_header; use codex_mcp::discover_supported_scopes; use codex_mcp::effective_mcp_servers_with_authorization_header; +use codex_mcp::read_mcp_resource as read_mcp_resource_without_thread; use codex_mcp::resolve_oauth_scopes; use codex_models_manager::collaboration_mode_presets::CollaborationModesConfig; use codex_protocol::ThreadId; @@ -5922,50 +5923,115 @@ impl CodexMessageProcessor { params: McpResourceReadParams, ) { let outgoing = Arc::clone(&self.outgoing); - let (_, thread) = match self.load_thread(¶ms.thread_id).await { - Ok(thread) => thread, + let McpResourceReadParams { + thread_id, + server, + uri, + } = params; + + if let Some(thread_id) = thread_id { + let (_, thread) = match self.load_thread(&thread_id).await { + Ok(thread) => thread, + Err(error) => { + self.outgoing.send_error(request_id, error).await; + return; + } + }; + + tokio::spawn(async move { + let result = thread.read_mcp_resource(&server, &uri).await; + Self::send_mcp_resource_read_response(outgoing, request_id, result).await; + }); + return; + } + + let config = match self.load_latest_config(/*fallback_cwd*/ None).await { + Ok(config) => config, Err(error) => { self.outgoing.send_error(request_id, error).await; return; } }; + let mcp_config = config + .to_mcp_config(self.thread_manager.plugins_manager().as_ref()) + .await; + let auth = self.auth_manager.auth().await; + let runtime_environment = match self.thread_manager.environment_manager().current().await { + Ok(Some(environment)) => { + // Resource reads without a thread have no turn cwd. This fallback + // is used only by executor-backed stdio MCPs whose config omits `cwd`. + McpRuntimeEnvironment::new(environment, config.cwd.to_path_buf()) + } + Ok(None) => McpRuntimeEnvironment::new( + Arc::new(codex_exec_server::Environment::default()), + config.cwd.to_path_buf(), + ), + Err(err) => { + let error = JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("failed to create environment: {err}"), + data: None, + }; + self.outgoing.send_error(request_id, error).await; + return; + } + }; tokio::spawn(async move { - let result = thread.read_mcp_resource(¶ms.server, ¶ms.uri).await; - match result { - Ok(result) => match serde_json::from_value::(result) { - Ok(response) => { - outgoing.send_response(request_id, response).await; - } - Err(error) => { - outgoing - .send_error( - request_id, - JSONRPCErrorError { - code: INTERNAL_ERROR_CODE, - message: format!( - "failed to deserialize MCP resource read response: {error}" - ), - data: None, - }, - ) - .await; - } - }, + let result = match read_mcp_resource_without_thread( + &mcp_config, + auth.as_ref(), + runtime_environment, + &server, + &uri, + ) + .await + { + Ok(result) => serde_json::to_value(result).map_err(anyhow::Error::from), + Err(error) => Err(error), + }; + Self::send_mcp_resource_read_response(outgoing, request_id, result).await; + }); + } + + async fn send_mcp_resource_read_response( + outgoing: Arc, + request_id: ConnectionRequestId, + result: anyhow::Result, + ) { + match result { + Ok(result) => match serde_json::from_value::(result) { + Ok(response) => { + outgoing.send_response(request_id, response).await; + } Err(error) => { outgoing .send_error( request_id, JSONRPCErrorError { code: INTERNAL_ERROR_CODE, - message: format!("{error:#}"), + message: format!( + "failed to deserialize MCP resource read response: {error}" + ), data: None, }, ) .await; } + }, + Err(error) => { + outgoing + .send_error( + request_id, + JSONRPCErrorError { + code: INTERNAL_ERROR_CODE, + message: format!("{error:#}"), + data: None, + }, + ) + .await; } - }); + } } async fn call_mcp_server_tool( 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 e09f5beca..c26b456fa 100644 --- a/codex-rs/app-server/tests/suite/v2/mcp_resource.rs +++ b/codex-rs/app-server/tests/suite/v2/mcp_resource.rs @@ -43,6 +43,7 @@ use rmcp::transport::StreamableHttpService; use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; use tempfile::TempDir; use tokio::net::TcpListener; +use tokio::task::JoinHandle; use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10); @@ -54,19 +55,7 @@ const TEST_RESOURCE_TEXT: &str = "Resource body from the MCP server."; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn mcp_resource_read_returns_resource_contents() -> Result<()> { let responses_server = responses::start_mock_server().await; - let listener = TcpListener::bind("127.0.0.1:0").await?; - let addr = listener.local_addr()?; - let apps_server_url = format!("http://{addr}"); - - let mcp_service = StreamableHttpService::new( - move || Ok(ResourceAppsMcpServer), - Arc::new(LocalSessionManager::default()), - StreamableHttpServerConfig::default(), - ); - let router = Router::new().nest_service("/api/codex/apps", mcp_service); - let apps_server_handle = tokio::spawn(async move { - let _ = axum::serve(listener, router).await; - }); + let (apps_server_url, apps_server_handle) = start_resource_apps_mcp_server().await?; let codex_home = TempDir::new()?; let responses_server_uri = responses_server.uri(); @@ -121,7 +110,7 @@ stream_max_retries = 0 let read_request_id = mcp .send_mcp_resource_read_request(McpResourceReadParams { - thread_id: thread.id, + thread_id: Some(thread.id), server: "codex_apps".to_string(), uri: TEST_RESOURCE_URI.to_string(), }) @@ -134,22 +123,59 @@ stream_max_retries = 0 assert_eq!( to_response::(read_response)?, - McpResourceReadResponse { - contents: vec![ - McpResourceContent::Text { - uri: TEST_RESOURCE_URI.to_string(), - mime_type: Some("text/markdown".to_string()), - text: TEST_RESOURCE_TEXT.to_string(), - meta: None, - }, - McpResourceContent::Blob { - uri: TEST_BLOB_RESOURCE_URI.to_string(), - mime_type: Some("application/octet-stream".to_string()), - blob: TEST_RESOURCE_BLOB.to_string(), - meta: None, - }, - ], - } + expected_resource_read_response() + ); + + apps_server_handle.abort(); + let _ = apps_server_handle.await; + Ok(()) +} + +#[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_handle) = start_resource_apps_mcp_server().await?; + + let codex_home = TempDir::new()?; + std::fs::write( + codex_home.path().join("config.toml"), + format!( + r#" +chatgpt_base_url = "{apps_server_url}" +mcp_oauth_credentials_store = "file" + +[features] +apps = true +"# + ), + )?; + 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 = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let read_request_id = mcp + .send_mcp_resource_read_request(McpResourceReadParams { + thread_id: None, + server: "codex_apps".to_string(), + uri: TEST_RESOURCE_URI.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)?, + expected_resource_read_response() ); apps_server_handle.abort(); @@ -198,7 +224,7 @@ async fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> { .request(ClientRequest::McpResourceRead { request_id: RequestId::Integer(1), params: McpResourceReadParams { - thread_id: "00000000-0000-4000-8000-000000000000".to_string(), + thread_id: Some("00000000-0000-4000-8000-000000000000".to_string()), server: "codex_apps".to_string(), uri: TEST_RESOURCE_URI.to_string(), }, @@ -218,6 +244,43 @@ async fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> { Ok(()) } +async fn start_resource_apps_mcp_server() -> Result<(String, JoinHandle<()>)> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let addr = listener.local_addr()?; + let apps_server_url = format!("http://{addr}"); + + let mcp_service = StreamableHttpService::new( + move || Ok(ResourceAppsMcpServer), + Arc::new(LocalSessionManager::default()), + StreamableHttpServerConfig::default(), + ); + let router = Router::new().nest_service("/api/codex/apps", mcp_service); + let apps_server_handle = tokio::spawn(async move { + let _ = axum::serve(listener, router).await; + }); + + Ok((apps_server_url, apps_server_handle)) +} + +fn expected_resource_read_response() -> McpResourceReadResponse { + McpResourceReadResponse { + contents: vec![ + McpResourceContent::Text { + uri: TEST_RESOURCE_URI.to_string(), + mime_type: Some("text/markdown".to_string()), + text: TEST_RESOURCE_TEXT.to_string(), + meta: None, + }, + McpResourceContent::Blob { + uri: TEST_BLOB_RESOURCE_URI.to_string(), + mime_type: Some("application/octet-stream".to_string()), + blob: TEST_RESOURCE_BLOB.to_string(), + meta: None, + }, + ], + } +} + #[derive(Clone, Default)] struct ResourceAppsMcpServer; diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index 5fb634d90..760fc64cf 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -32,6 +32,7 @@ pub use mcp::group_tools_by_server; pub use mcp::mcp_permission_prompt_is_auto_approved; pub use mcp::oauth_login_support; pub use mcp::qualified_mcp_tool_name_prefix; +pub use mcp::read_mcp_resource; pub use mcp::resolve_oauth_scopes; pub use mcp::should_retry_without_scopes; pub use mcp::split_qualified_tool_name; diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index 6a8d9c0a9..3118c7c17 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -32,6 +32,8 @@ use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::McpAuthStatus; use codex_protocol::protocol::McpListToolsResponseEvent; use codex_protocol::protocol::SandboxPolicy; +use rmcp::model::ReadResourceRequestParams; +use rmcp::model::ReadResourceResult; use serde_json::Value; use crate::mcp_connection_manager::McpConnectionManager; @@ -354,6 +356,47 @@ pub fn tool_plugin_provenance(config: &McpConfig) -> ToolPluginProvenance { ToolPluginProvenance::from_capability_summaries(&config.plugin_capability_summaries) } +pub async fn read_mcp_resource( + config: &McpConfig, + auth: Option<&CodexAuth>, + runtime_environment: McpRuntimeEnvironment, + server: &str, + uri: &str, +) -> anyhow::Result { + let mut mcp_servers = effective_mcp_servers(config, auth); + mcp_servers.retain(|name, _| name == server); + let auth_statuses = + compute_auth_statuses(mcp_servers.iter(), config.mcp_oauth_credentials_store_mode).await; + let (tx_event, rx_event) = unbounded(); + drop(rx_event); + let (manager, cancel_token) = McpConnectionManager::new( + &mcp_servers, + config.mcp_oauth_credentials_store_mode, + auth_statuses, + &config.approval_policy, + String::new(), + tx_event, + SandboxPolicy::new_read_only_policy(), + runtime_environment, + config.codex_home.clone(), + codex_apps_tools_cache_key(auth), + tool_plugin_provenance(config), + ) + .await; + + let result = manager + .read_resource( + server, + ReadResourceRequestParams { + meta: None, + uri: uri.to_string(), + }, + ) + .await; + cancel_token.cancel(); + result +} + pub async fn collect_mcp_snapshot( config: &McpConfig, auth: Option<&CodexAuth>,