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`
2026-07-01 00:31:56 +08:00 · 2026-04-20 19:59:36 -07:00
parent 58e7605efc
commit 1132ef887c
11 changed files with 249 additions and 67 deletions
@@ -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"
@@ -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",
@@ -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",
@@ -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",
@@ -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, };
@@ -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<String>,
    pub server: String,
    pub uri: String,
 }
@@ -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.
@@ -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(&params.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(&params.server, &params.uri).await;
-            match result {
-                Ok(result) => match serde_json::from_value::<McpResourceReadResponse>(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<OutgoingMessageSender>,
+        request_id: ConnectionRequestId,
+        result: anyhow::Result<serde_json::Value>,
+    ) {
+        match result {
+            Ok(result) => match serde_json::from_value::<McpResourceReadResponse>(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(
@@ -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::<McpResourceReadResponse>(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::<McpResourceReadResponse>(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;

@@ -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;
@@ -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<ReadResourceResult> {
+    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>,