From cc6406463f0a72dc2ddfcef08293bba259016ea2 Mon Sep 17 00:00:00 2001 From: fbauer Date: Mon, 6 Jul 2026 15:32:59 -0700 Subject: [PATCH] mcp resource read: separate origin routing from tool refresh --- codex-rs/codex-mcp/src/connection_manager.rs | 40 +++++-------------- .../codex-mcp/src/connection_manager_tests.rs | 17 +------- codex-rs/codex-mcp/src/rmcp_client.rs | 35 +++++----------- 3 files changed, 22 insertions(+), 70 deletions(-) diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index 2f896b63b6..33c9219f20 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -538,7 +538,7 @@ impl McpConnectionManager { let managed_client = self.client_by_name(server).await?; Ok(McpServerConnection { server_name: server.to_string(), - tools: managed_client.current_tools(), + tools: managed_client.tools.clone(), managed_client, }) } @@ -598,19 +598,13 @@ impl McpConnectionManager { .client() .await .context("failed to get client")?; - let _tool_refresh_permit = managed_client - .tool_refresh_semaphore - .acquire() - .await - .context("tool refresh semaphore closed")?; - let list_start = Instant::now(); let fetch_start = Instant::now(); let fetch_ticket = managed_client .codex_apps_tools_cache_context .as_ref() .map(|cache_context| cache_context.begin_fetch(CodexAppsToolsFetchSource::HardRefresh)); - let fetched_tools = list_tools_for_client_uncached( + let tools = list_tools_for_client_uncached( CODEX_APPS_MCP_SERVER_NAME, /*is_codex_apps_mcp_server*/ true, &managed_client.client, @@ -627,23 +621,16 @@ impl McpConnectionManager { &[], ); - managed_client.replace_tools(filter_tools( - fetched_tools.clone(), - &managed_client.tool_filter, - )); - - let tools = match ( - managed_client.codex_apps_tools_cache_context.as_ref(), - fetch_ticket, - ) { - (Some(cache_context), Some(fetch_ticket)) => cache_context.publish_if_newest_accepted( + let tools = + match ( + managed_client.codex_apps_tools_cache_context.as_ref(), fetch_ticket, - &managed_client.server_info, - fetched_tools, - ), - (None, None) => fetched_tools, - _ => unreachable!("Codex Apps fetch ticket requires cache context"), - }; + ) { + (Some(cache_context), Some(fetch_ticket)) => cache_context + .publish_if_newest_accepted(fetch_ticket, &managed_client.server_info, tools), + (None, None) => tools, + _ => unreachable!("Codex Apps fetch ticket requires cache context"), + }; emit_duration( MCP_TOOLS_LIST_DURATION_METRIC, list_start.elapsed(), @@ -896,11 +883,6 @@ 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/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index e9c257cd93..25980b31e9 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -115,8 +115,7 @@ async fn create_test_managed_client(tools: Vec) -> ManagedClient { .expect("create in-process RMCP client"), ), server_info: create_test_server_info("Ready"), - tools: Arc::new(arc_swap::ArcSwap::from_pointee(tools)), - tool_refresh_semaphore: Arc::new(tokio::sync::Semaphore::new(1)), + tools, tool_filter: ToolFilter::default(), tool_timeout: None, server_instructions: None, @@ -829,7 +828,6 @@ async fn list_all_tools_uses_shared_codex_apps_cache_while_client_is_pending() { )]) .await; live_client.server_info = create_test_server_info("Live"); - let live_client_for_refresh = live_client.clone(); assert!(live_tx.send(live_client).is_ok()); let connection = manager @@ -839,19 +837,6 @@ async fn list_all_tools_uses_shared_codex_apps_cache_while_client_is_pending() { assert_eq!(connection.server_info().title.as_deref(), Some("Live")); assert!(connection.tool_info("live_tool").is_some()); assert!(connection.tool_info("cached_tool").is_none()); - - live_client_for_refresh.replace_tools(vec![create_test_tool( - CODEX_APPS_MCP_SERVER_NAME, - "refreshed_tool", - )]); - let refreshed = manager - .server_connection(CODEX_APPS_MCP_SERVER_NAME) - .await - .expect("refreshed connection"); - assert!(connection.tool_info("live_tool").is_some()); - assert!(connection.tool_info("refreshed_tool").is_none()); - assert!(refreshed.tool_info("live_tool").is_none()); - assert!(refreshed.tool_info("refreshed_tool").is_some()); } #[tokio::test] diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index ff693c2611..37819203f6 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -37,7 +37,6 @@ use crate::tools::filter_tools; use crate::tools::tool_with_model_visible_input_schema; use anyhow::Result; use anyhow::anyhow; -use arc_swap::ArcSwap; use async_channel::Sender; use codex_api::SharedAuthProvider; use codex_async_utils::CancelErr; @@ -101,8 +100,7 @@ const UNTRUSTED_CONNECTOR_META_KEYS: &[&str] = &[ pub(crate) struct ManagedClient { pub(crate) client: Arc, pub(crate) server_info: McpServerInfo, - pub(crate) tools: Arc>>, - pub(crate) tool_refresh_semaphore: Arc, + pub(crate) tools: Vec, pub(crate) tool_filter: ToolFilter, pub(crate) tool_timeout: Option, pub(crate) server_instructions: Option, @@ -111,15 +109,7 @@ pub(crate) struct ManagedClient { } impl ManagedClient { - pub(crate) fn current_tools(&self) -> Vec { - self.tools.load_full().as_ref().clone() - } - - pub(crate) fn replace_tools(&self, tools: Vec) { - self.tools.store(Arc::new(tools)); - } - - pub(crate) fn listed_tools(&self) -> Vec { + fn listed_tools(&self) -> Vec { let total_start = Instant::now(); if let Some(tools) = self .codex_apps_tools_cache_context @@ -142,7 +132,7 @@ impl ManagedClient { ); } - self.current_tools() + self.tools.clone() } } @@ -834,7 +824,7 @@ async fn start_server_task( let fetch_ticket = codex_apps_tools_cache_context .as_ref() .map(|cache_context| cache_context.begin_fetch(CodexAppsToolsFetchSource::Startup)); - let fetched_tools = list_tools_for_client_uncached( + let tools = list_tools_for_client_uncached( &server_name, is_codex_apps_mcp_server, &client, @@ -849,17 +839,13 @@ async fn start_server_task( &[], ); let server_info = mcp_server_info_from_implementation(initialize_result.server_info); - match (codex_apps_tools_cache_context.as_ref(), fetch_ticket) { + let tools = match (codex_apps_tools_cache_context.as_ref(), fetch_ticket) { (Some(cache_context), Some(fetch_ticket)) => { - cache_context.publish_if_newest_accepted( - fetch_ticket, - &server_info, - fetched_tools.clone(), - ); + cache_context.publish_if_newest_accepted(fetch_ticket, &server_info, tools) } - (None, None) => {} + (None, None) => tools, _ => unreachable!("Codex Apps fetch ticket requires cache context"), - } + }; if is_codex_apps_mcp_server { emit_duration( MCP_TOOLS_LIST_DURATION_METRIC, @@ -867,13 +853,12 @@ async fn start_server_task( &[("cache", "miss")], ); } - let tools = filter_tools(fetched_tools, &tool_filter); + let tools = filter_tools(tools, &tool_filter); let managed = ManagedClient { client: Arc::clone(&client), server_info, - tools: Arc::new(ArcSwap::from_pointee(tools)), - tool_refresh_semaphore: Arc::new(tokio::sync::Semaphore::new(1)), + tools, tool_timeout: Some(tool_timeout), tool_filter, server_instructions: initialize_result.instructions,