From 68a359353a77e45fab923c3146bb2f56550e4eac Mon Sep 17 00:00:00 2001 From: Brian Brunner Date: Tue, 23 Jun 2026 19:03:25 +0000 Subject: [PATCH] [codex] align MCP status with callable thread manager --- .../src/request_processors/mcp_processor.rs | 119 ++++++------ .../tests/suite/v2/mcp_server_status.rs | 181 ++++++++++++++++-- codex-rs/codex-mcp/src/connection_manager.rs | 4 + codex-rs/codex-mcp/src/lib.rs | 1 + codex-rs/codex-mcp/src/mcp/mod.rs | 33 ++++ codex-rs/core/src/codex_thread.rs | 23 +++ 6 files changed, 294 insertions(+), 67 deletions(-) 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 08d464598b..b9c3f91198 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/codex-rs/app-server/src/request_processors/mcp_processor.rs @@ -2,6 +2,18 @@ use super::*; const MCP_TOOL_THREAD_ID_META_KEY: &str = "threadId"; +enum McpServerStatusSource { + Installed { + thread: Arc, + config: Box, + }, + Configured { + mcp_config: Box, + auth: Option, + runtime_context: McpRuntimeContext, + }, +} + #[derive(Clone)] pub(crate) struct McpRequestProcessor { auth_manager: Arc, @@ -204,7 +216,7 @@ impl McpRequestProcessor { let request = request_id.clone(); let outgoing = Arc::clone(&self.outgoing); - let (config, thread) = match params.thread_id.as_deref() { + let source = match params.thread_id.as_deref() { Some(thread_id) => { let (_, thread) = self.load_thread(thread_id).await?; let thread_config = thread.config().await; @@ -213,37 +225,37 @@ impl McpRequestProcessor { .load_latest_config_for_thread(thread_config.as_ref()) .await .map_err(|err| internal_error(format!("failed to reload config: {err}")))?; - (config, Some(thread)) + McpServerStatusSource::Installed { + thread, + config: Box::new(config), + } } - None => (self.load_latest_config(/*fallback_cwd*/ None).await?, None), - }; - let mcp_config = match thread { - Some(thread) => thread.runtime_mcp_config(&config).await, None => { - self.thread_manager + let config = self.load_latest_config(/*fallback_cwd*/ None).await?; + let mcp_config = self + .thread_manager .mcp_manager() .runtime_config(&config) - .await + .await; + let auth = self.auth_manager.auth().await; + let environment_manager = self.thread_manager.environment_manager(); + // This status path has no turn-selected environment. Use config cwd + // as the local stdio fallback; named environment stdio MCPs must + // declare their own absolute cwd. + let runtime_context = McpRuntimeContext::new( + Arc::clone(&environment_manager), + config.cwd.to_path_buf(), + ); + McpServerStatusSource::Configured { + mcp_config: Box::new(mcp_config), + auth, + runtime_context, + } } }; - let auth = self.auth_manager.auth().await; - let environment_manager = self.thread_manager.environment_manager(); - // This status path has no turn-selected environment. Use config cwd - // as the local stdio fallback; named environment stdio MCPs must - // declare their own absolute cwd. - let runtime_context = - McpRuntimeContext::new(Arc::clone(&environment_manager), config.cwd.to_path_buf()); tokio::spawn(async move { - Self::list_mcp_server_status_task( - outgoing, - request, - params, - mcp_config, - auth, - runtime_context, - ) - .await; + Self::list_mcp_server_status_task(outgoing, request, params, source).await; }); Ok(()) } @@ -252,42 +264,41 @@ impl McpRequestProcessor { outgoing: Arc, request_id: ConnectionRequestId, params: ListMcpServerStatusParams, - mcp_config: codex_mcp::McpConfig, - auth: Option, - runtime_context: McpRuntimeContext, + source: McpServerStatusSource, ) { - let result = Self::list_mcp_server_status_response( - request_id.request_id.to_string(), - params, - mcp_config, - auth, - runtime_context, - ) - .await; - outgoing.send_result(request_id, result).await; - } - - async fn list_mcp_server_status_response( - request_id: String, - params: ListMcpServerStatusParams, - mcp_config: codex_mcp::McpConfig, - auth: Option, - runtime_context: McpRuntimeContext, - ) -> Result { let detail = match params.detail.unwrap_or(McpServerStatusDetail::Full) { McpServerStatusDetail::Full => McpSnapshotDetail::Full, McpServerStatusDetail::ToolsAndAuthOnly => McpSnapshotDetail::ToolsAndAuthOnly, }; + let snapshot = match source { + McpServerStatusSource::Installed { thread, config } => { + thread + .mcp_server_status_snapshot(config.as_ref(), detail) + .await + } + McpServerStatusSource::Configured { + mcp_config, + auth, + runtime_context, + } => { + collect_mcp_server_status_snapshot_with_detail( + mcp_config.as_ref(), + auth.as_ref(), + request_id.request_id.to_string(), + runtime_context, + detail, + ) + .await + } + }; + let result = Self::list_mcp_server_status_response(params, snapshot); + outgoing.send_result(request_id, result).await; + } - let snapshot = collect_mcp_server_status_snapshot_with_detail( - &mcp_config, - auth.as_ref(), - request_id, - runtime_context, - detail, - ) - .await; - + fn list_mcp_server_status_response( + params: ListMcpServerStatusParams, + snapshot: McpServerStatusSnapshot, + ) -> Result { let McpServerStatusSnapshot { server_infos, tools_by_server, diff --git a/codex-rs/app-server/tests/suite/v2/mcp_server_status.rs b/codex-rs/app-server/tests/suite/v2/mcp_server_status.rs index 2c34684ada..8d813a82b0 100644 --- a/codex-rs/app-server/tests/suite/v2/mcp_server_status.rs +++ b/codex-rs/app-server/tests/suite/v2/mcp_server_status.rs @@ -5,21 +5,29 @@ use std::sync::Arc; use std::time::Duration; use anyhow::Result; +use app_test_support::ChatGptAuthFixture; use app_test_support::TestAppServer; use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::to_response; +use app_test_support::write_chatgpt_auth; use app_test_support::write_mock_responses_config_toml; +use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url; use axum::Router; +use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::ListMcpServerStatusParams; use codex_app_server_protocol::ListMcpServerStatusResponse; use codex_app_server_protocol::McpServerStatusDetail; +use codex_app_server_protocol::McpServerToolCallParams; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; +use codex_config::types::AuthCredentialsStoreMode; use codex_core::config::set_project_trust_level; use codex_protocol::config_types::TrustLevel; use pretty_assertions::assert_eq; use rmcp::handler::server::ServerHandler; +use rmcp::model::CallToolRequestParams; +use rmcp::model::CallToolResult; use rmcp::model::Implementation; use rmcp::model::JsonObject; use rmcp::model::ListResourceTemplatesResult; @@ -131,6 +139,17 @@ async fn mcp_server_status_list_uses_thread_project_local_config() -> Result<()> )?; std::fs::create_dir_all(workspace.path().join(".git"))?; set_project_trust_level(codex_home.path(), workspace.path(), TrustLevel::Trusted)?; + let project_config_dir = workspace.path().join(".codex"); + std::fs::create_dir_all(&project_config_dir)?; + std::fs::write( + project_config_dir.join("config.toml"), + format!( + r#" +[mcp_servers.project-server] +url = "{mcp_server_url}/mcp" +"# + ), + )?; let mut mcp = TestAppServer::new(codex_home.path()).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; @@ -148,18 +167,6 @@ async fn mcp_server_status_list_uses_thread_project_local_config() -> Result<()> .await??; let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?; - let project_config_dir = workspace.path().join(".codex"); - std::fs::create_dir_all(&project_config_dir)?; - std::fs::write( - project_config_dir.join("config.toml"), - format!( - r#" -[mcp_servers.project-server] -url = "{mcp_server_url}/mcp" -"# - ), - )?; - let threadless_request_id = mcp .send_list_mcp_server_status_request(ListMcpServerStatusParams { cursor: None, @@ -206,6 +213,135 @@ url = "{mcp_server_url}/mcp" Ok(()) } +#[tokio::test] +async fn mcp_server_status_list_never_advertises_uninstalled_codex_apps() -> Result<()> { + let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await; + let (apps_server_url, apps_server_handle) = start_apps_mcp_server("calendar_lookup").await?; + let codex_home = TempDir::new()?; + write_mock_responses_config_toml_with_chatgpt_base_url( + codex_home.path(), + &server.uri(), + &apps_server_url, + )?; + let config_path = codex_home.path().join("config.toml"); + let mut config_toml = std::fs::read_to_string(&config_path)?; + config_toml.push_str( + r#" +[features] +apps = false +"#, + ); + std::fs::write(&config_path, &config_toml)?; + 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(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let stale_thread_start_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let stale_thread_start_response = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(stale_thread_start_id)), + ) + .await??; + let ThreadStartResponse { + thread: stale_thread, + .. + } = to_response(stale_thread_start_response)?; + + std::fs::write( + &config_path, + config_toml.replace("apps = false", "apps = true"), + )?; + + let stale_status_request_id = mcp + .send_list_mcp_server_status_request(ListMcpServerStatusParams { + cursor: None, + limit: None, + detail: Some(McpServerStatusDetail::ToolsAndAuthOnly), + thread_id: Some(stale_thread.id), + }) + .await?; + let stale_status_response = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(stale_status_request_id)), + ) + .await??; + let stale_status_response: ListMcpServerStatusResponse = to_response(stale_status_response)?; + assert_eq!(stale_status_response.data, Vec::new()); + + let current_thread_start_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let current_thread_start_response = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(current_thread_start_id)), + ) + .await??; + let ThreadStartResponse { + thread: current_thread, + .. + } = to_response(current_thread_start_response)?; + + let current_status_request_id = mcp + .send_list_mcp_server_status_request(ListMcpServerStatusParams { + cursor: None, + limit: None, + detail: Some(McpServerStatusDetail::ToolsAndAuthOnly), + thread_id: Some(current_thread.id.clone()), + }) + .await?; + let current_status_response = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(current_status_request_id)), + ) + .await??; + let current_status_response: ListMcpServerStatusResponse = + to_response(current_status_response)?; + + assert_eq!(current_status_response.next_cursor, None); + assert_eq!(current_status_response.data.len(), 1); + let status = ¤t_status_response.data[0]; + assert_eq!(status.name, "codex_apps"); + assert_eq!( + status.tools.keys().cloned().collect::>(), + BTreeSet::from(["calendar_lookup".to_string()]) + ); + + let tool_call_id = mcp + .send_mcp_server_tool_call_request(McpServerToolCallParams { + thread_id: current_thread.id, + server: "codex_apps".to_string(), + tool: "missing_tool".to_string(), + arguments: Some(json!({})), + meta: None, + }) + .await?; + let error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(tool_call_id)), + ) + .await??; + assert!( + error.error.message.contains("unknown tool"), + "a listed server must receive the call; got: {error:?}" + ); + assert!(!error.error.message.contains("unknown MCP server")); + + apps_server_handle.abort(); + let _ = apps_server_handle.await; + + Ok(()) +} + #[derive(Clone)] struct McpStatusServer { tool_name: Arc, @@ -242,6 +378,17 @@ impl ServerHandler for McpStatusServer { meta: None, }) } + + async fn call_tool( + &self, + request: CallToolRequestParams, + _context: RequestContext, + ) -> Result { + Err(rmcp::ErrorData::invalid_params( + format!("unknown tool: {}", request.name), + None, + )) + } } #[derive(Clone)] @@ -451,6 +598,14 @@ url = "{underscore_server_url}/mcp" } async fn start_mcp_server(tool_name: &str) -> Result<(String, JoinHandle<()>)> { + start_mcp_server_at(tool_name, "/mcp").await +} + +async fn start_apps_mcp_server(tool_name: &str) -> Result<(String, JoinHandle<()>)> { + start_mcp_server_at(tool_name, "/api/codex/ps/mcp").await +} + +async fn start_mcp_server_at(tool_name: &str, route: &str) -> Result<(String, JoinHandle<()>)> { let listener = TcpListener::bind("127.0.0.1:0").await?; let addr = listener.local_addr()?; let tool_name = Arc::new(tool_name.to_string()); @@ -463,7 +618,7 @@ async fn start_mcp_server(tool_name: &str) -> Result<(String, JoinHandle<()>)> { Arc::new(LocalSessionManager::default()), StreamableHttpServerConfig::default(), ); - let router = Router::new().nest_service("/mcp", mcp_service); + let router = Router::new().nest_service(route, mcp_service); let handle = tokio::spawn(async move { let _ = axum::serve(listener, router).await; diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index 33631393cc..61d265a5c1 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -338,6 +338,10 @@ impl McpConnectionManager { !self.clients.is_empty() } + pub(crate) fn server_names(&self) -> Vec { + self.clients.keys().cloned().collect() + } + pub(crate) fn contains_server(&self, server_name: &str) -> bool { self.clients.contains_key(server_name) } diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index b4d397eac9..04a5a6811f 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -51,6 +51,7 @@ pub use plugin_config::parse_plugin_mcp_config; pub use mcp::McpServerStatusSnapshot; pub use mcp::McpSnapshotDetail; +pub use mcp::collect_mcp_server_status_snapshot_from_manager_with_detail; pub use mcp::collect_mcp_server_status_snapshot_with_detail; pub use mcp::read_mcp_resource; diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index 4a765e6a62..2e06329958 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -396,6 +396,39 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( snapshot } +/// Collects status from one installed manager generation. +/// +/// Latest config is used only to refresh authentication status for servers +/// that are actually present in the manager's callable client map. +pub async fn collect_mcp_server_status_snapshot_from_manager_with_detail( + mcp_connection_manager: &McpConnectionManager, + config: &McpConfig, + auth: Option<&CodexAuth>, + detail: McpSnapshotDetail, +) -> McpServerStatusSnapshot { + let server_names = mcp_connection_manager.server_names(); + let installed_servers = server_names.iter().cloned().collect::>(); + let mcp_servers = effective_mcp_servers(config, auth) + .into_iter() + .filter(|(name, _)| installed_servers.contains(name)) + .collect::>(); + let auth_status_entries = compute_auth_statuses( + mcp_servers.iter(), + config.mcp_oauth_credentials_store_mode, + config.auth_keyring_backend_kind, + auth, + ) + .await; + + collect_mcp_server_status_snapshot_from_manager( + mcp_connection_manager, + auth_status_entries, + server_names, + detail, + ) + .await +} + /// The Responses API requires tool names to match `^[a-zA-Z0-9_-]+$`. /// MCP server/tool names are user-controlled, so sanitize the fully-qualified /// name we expose to the model by replacing any disallowed character with `_`. diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index dccfd4a763..026fd030b1 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -593,6 +593,29 @@ impl CodexThread { self.codex.session.runtime_mcp_config(config).await } + /// Returns the MCP inventory installed for this thread. + pub async fn mcp_server_status_snapshot( + &self, + config: &crate::config::Config, + detail: codex_mcp::McpSnapshotDetail, + ) -> codex_mcp::McpServerStatusSnapshot { + let mcp_config = self.runtime_mcp_config(config).await; + let auth = self.codex.session.services.auth_manager.auth().await; + let manager = self + .codex + .session + .services + .mcp_connection_manager + .load_full(); + codex_mcp::collect_mcp_server_status_snapshot_from_manager_with_detail( + manager.as_ref(), + &mcp_config, + auth.as_ref(), + detail, + ) + .await + } + pub fn multi_agent_version(&self) -> Option { self.codex.session.multi_agent_version() }