diff --git a/codex-rs/codex-mcp/src/binding.rs b/codex-rs/codex-mcp/src/binding.rs index 3135437b68..48361c1a25 100644 --- a/codex-rs/codex-mcp/src/binding.rs +++ b/codex-rs/codex-mcp/src/binding.rs @@ -31,6 +31,7 @@ use serde_json::Value as JsonValue; use crate::McpConfig; use crate::McpServerPermissions; use crate::binding_clients::McpBindingClients; +use crate::binding_clients::McpResourceListing; use crate::client_tool_catalog::ToolCatalogSnapshot; use crate::connection_manager::McpConnectionSet; use crate::rmcp_client::ManagedClient; @@ -52,7 +53,7 @@ impl McpBinding { pub fn empty(config: Arc) -> Self { Self::new( Arc::new(McpConnectionSet::empty(config.prefix_mcp_tool_names)), - Arc::new(McpBindingClients::new(HashMap::new())), + Arc::new(McpBindingClients::default()), config, /*plugins_available*/ false, Vec::new(), @@ -130,7 +131,7 @@ impl McpBinding { pub async fn list_all_resources( &self, include_server: impl Fn(&str) -> bool, - ) -> HashMap> { + ) -> McpResourceListing { self.clients.list_all_resources(include_server).await } @@ -154,7 +155,7 @@ impl McpBinding { pub async fn list_all_resource_templates( &self, include_server: impl Fn(&str) -> bool, - ) -> HashMap> { + ) -> McpResourceListing { self.clients .list_all_resource_templates(|server| { include_server(server) && !self.connections.skips_resource_template_listing(server) diff --git a/codex-rs/codex-mcp/src/binding_clients.rs b/codex-rs/codex-mcp/src/binding_clients.rs index 5ef9c4baf9..d08c4297c7 100644 --- a/codex-rs/codex-mcp/src/binding_clients.rs +++ b/codex-rs/codex-mcp/src/binding_clients.rs @@ -17,14 +17,37 @@ use tracing::warn; use crate::pagination::collect_paginated; use crate::rmcp_client::ManagedClient; -/// The ready clients captured for one model step. +/// Items listed by ready servers and names of deferred servers available to query explicitly. +pub struct McpResourceListing { + pub by_server: HashMap>, + pub deferred_servers: Vec, +} + +/// The ready clients and dormant server names captured for one model step. +#[derive(Default)] pub(crate) struct McpBindingClients { clients: HashMap>, + deferred_servers: Vec, } impl McpBindingClients { - pub(crate) fn new(clients: HashMap>) -> Self { - Self { clients } + pub(crate) fn new( + clients: HashMap>, + mut deferred_servers: Vec, + ) -> Self { + deferred_servers.sort_unstable(); + Self { + clients, + deferred_servers, + } + } + + fn deferred_servers(&self, include_server: impl Fn(&str) -> bool) -> Vec { + self.deferred_servers + .iter() + .filter(|name| include_server(name)) + .cloned() + .collect() } pub(crate) fn client(&self, server: &str) -> Option> { @@ -77,7 +100,7 @@ impl McpBindingClients { pub(crate) async fn list_all_resources( &self, include_server: impl Fn(&str) -> bool, - ) -> HashMap> { + ) -> McpResourceListing { let mut join_set = JoinSet::new(); for (server_name, managed) in self .clients @@ -99,13 +122,16 @@ impl McpBindingClients { (server_name, resources) }); } - collect_resource_results(&mut join_set, "resources").await + McpResourceListing { + by_server: collect_resource_results(&mut join_set, "resources").await, + deferred_servers: self.deferred_servers(include_server), + } } pub(crate) async fn list_all_resource_templates( &self, include_server: impl Fn(&str) -> bool, - ) -> HashMap> { + ) -> McpResourceListing { let mut join_set = JoinSet::new(); for (server_name, managed) in self .clients @@ -127,7 +153,10 @@ impl McpBindingClients { (server_name, templates) }); } - collect_resource_results(&mut join_set, "resource templates").await + McpResourceListing { + by_server: collect_resource_results(&mut join_set, "resource templates").await, + deferred_servers: self.deferred_servers(include_server), + } } } diff --git a/codex-rs/codex-mcp/src/binding_tests.rs b/codex-rs/codex-mcp/src/binding_tests.rs index 5d784dd924..58ec1aeabc 100644 --- a/codex-rs/codex-mcp/src/binding_tests.rs +++ b/codex-rs/codex-mcp/src/binding_tests.rs @@ -107,10 +107,10 @@ async fn test_step( client_supports_network_policy_meta_capability: false, codex_apps_tools_cache_context: None, }); - let clients = Arc::new(McpBindingClients::new(HashMap::from([( - SERVER_NAME.to_string(), - Arc::clone(&managed_client), - )]))); + let clients = Arc::new(McpBindingClients::new( + HashMap::from([(SERVER_NAME.to_string(), Arc::clone(&managed_client))]), + Vec::new(), + )); let connections = Arc::new(McpConnectionSet::empty(/*prefix_mcp_tool_names*/ true)); let mut config = crate::mcp::tests::test_mcp_config(std::env::temp_dir()); config.host_cwd = config.host_cwd.join(label); diff --git a/codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs b/codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs index 05b64fcc10..58d8f0dc88 100644 --- a/codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs +++ b/codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs @@ -547,11 +547,21 @@ impl McpConnectionSet { call, ); } + let deferred_servers = self + .servers + .iter() + .filter(|(_, view)| { + view.connection.startup_is_dormant() + && !view.connection.client.cancel_token.is_cancelled() + }) + .map(|(name, _)| name.clone()) + .collect(); let clients = Arc::new(McpBindingClients::new( clients .into_iter() .map(|(server_name, (client, _))| (server_name, client)) .collect(), + deferred_servers, )); McpBinding::new( Arc::clone(self), diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index 0ab671d89c..f324ebd293 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -2341,7 +2341,10 @@ async fn resource_template_listings_skip_only_hosted_apps() -> anyhow::Result<() expected ); let all = binding.list_all_resource_templates(|_| true).await; - assert_eq!(all.get(name), (!expected.is_empty()).then_some(&expected)); + assert_eq!( + all.by_server.get(name), + (!expected.is_empty()).then_some(&expected) + ); } Ok(()) } @@ -6139,11 +6142,11 @@ async fn reconciliation_reuses_connection_without_relisting_regular_tools() -> a .expect("refresh must use the reconciled timeout"); assert_eq!(failed.tools(), old_step.tools()); assert_eq!( - failed.list_all_resources(|_| true).await, + failed.list_all_resources(|_| true).await.by_server, HashMap::from([("docs".to_string(), vec![])]) ); assert_eq!( - failed.list_all_resource_templates(|_| true).await, + failed.list_all_resource_templates(|_| true).await.by_server, HashMap::from([("docs".to_string(), vec![])]) ); let retry_started = tokio::time::Instant::now(); diff --git a/codex-rs/codex-mcp/src/connection_pool_tests.rs b/codex-rs/codex-mcp/src/connection_pool_tests.rs index 2ed0cd858d..932f39d61a 100644 --- a/codex-rs/codex-mcp/src/connection_pool_tests.rs +++ b/codex-rs/codex-mcp/src/connection_pool_tests.rs @@ -514,10 +514,10 @@ async fn shared_startup_survives_first_chat_cancellation_and_rejects_released_bi /*selected_plugin_server*/ false, ) .unwrap(); - let bindings = McpBindingClients::new(HashMap::from([( - "test_server".to_string(), - Arc::new(client.clone()), - )])); + let bindings = McpBindingClients::new( + HashMap::from([("test_server".to_string(), Arc::new(client.clone()))]), + Vec::new(), + ); server.block_listing.store(true, Ordering::Release); let waiting = async { tokio::join!( diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index a6a3596a53..e9bf3852dd 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -4,6 +4,7 @@ pub use binding::PreparedMcpCall; pub use binding::SEARCH_CONNECTOR; pub use binding::SearchDiscovery; pub use binding::is_model_search_tool; +pub use binding_clients::McpResourceListing; pub use client_capabilities::client_mcp_extensions; pub use client_tool_catalog::CodexAppsToolSnapshot; pub use codex_rmcp_client::McpProtocolMode; diff --git a/codex-rs/core/src/tools/handlers/mcp_resource.rs b/codex-rs/core/src/tools/handlers/mcp_resource.rs index 7a3d9223bc..c3fb9d416b 100644 --- a/codex-rs/core/src/tools/handlers/mcp_resource.rs +++ b/codex-rs/core/src/tools/handlers/mcp_resource.rs @@ -5,6 +5,7 @@ use std::time::Duration; use std::time::Instant; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; +use codex_mcp::McpResourceListing; use codex_protocol::items::McpToolCallError; use codex_protocol::items::McpToolCallItem; use codex_protocol::items::McpToolCallStatus; @@ -197,6 +198,8 @@ struct ListResourcesPayload { resources: Vec>, #[serde(skip_serializing_if = "Option::is_none")] next_cursor: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + deferred_servers: Vec, } impl ListResourcesPayload { @@ -205,14 +208,16 @@ impl ListResourcesPayload { resources: ResourceWithServer::from_server(&server, result.resources), server: Some(server), next_cursor: result.next_cursor, + deferred_servers: Vec::new(), } } - fn from_all_servers(resources_by_server: HashMap>) -> Self { + fn from_all_servers(listing: McpResourceListing) -> Self { Self { server: None, - resources: ResourceWithServer::from_all_servers(resources_by_server), + resources: ResourceWithServer::from_all_servers(listing.by_server), next_cursor: None, + deferred_servers: listing.deferred_servers, } } } @@ -225,6 +230,8 @@ struct ListResourceTemplatesPayload { resource_templates: Vec>, #[serde(skip_serializing_if = "Option::is_none")] next_cursor: Option, + #[serde(skip_serializing_if = "Vec::is_empty")] + deferred_servers: Vec, } impl ListResourceTemplatesPayload { @@ -233,14 +240,16 @@ impl ListResourceTemplatesPayload { resource_templates: ResourceWithServer::from_server(&server, result.resource_templates), server: Some(server), next_cursor: result.next_cursor, + deferred_servers: Vec::new(), } } - fn from_all_servers(templates_by_server: HashMap>) -> Self { + fn from_all_servers(listing: McpResourceListing) -> Self { Self { server: None, - resource_templates: ResourceWithServer::from_all_servers(templates_by_server), + resource_templates: ResourceWithServer::from_all_servers(listing.by_server), next_cursor: None, + deferred_servers: listing.deferred_servers, } } } diff --git a/codex-rs/core/src/tools/handlers/mcp_resource_tests.rs b/codex-rs/core/src/tools/handlers/mcp_resource_tests.rs index 7f58175206..b28e7417c0 100644 --- a/codex-rs/core/src/tools/handlers/mcp_resource_tests.rs +++ b/codex-rs/core/src/tools/handlers/mcp_resource_tests.rs @@ -44,7 +44,10 @@ fn list_resources_payload_from_all_servers_is_sorted() { vec![resource("memo://a-1", "a-1"), resource("memo://a-2", "a-2")], ); - let payload = ListResourcesPayload::from_all_servers(map); + let payload = ListResourcesPayload::from_all_servers(McpResourceListing { + by_server: map, + deferred_servers: Vec::new(), + }); let value = serde_json::to_value(&payload).expect("serialize payload"); let uris: Vec = value["resources"] .as_array() @@ -132,7 +135,10 @@ fn list_resource_templates_payload_from_all_servers_is_sorted() { vec![template("memo://alpha/{id}", "alpha")], ); - let payload = ListResourceTemplatesPayload::from_all_servers(templates_by_server); + let payload = ListResourceTemplatesPayload::from_all_servers(McpResourceListing { + by_server: templates_by_server, + deferred_servers: Vec::new(), + }); assert_eq!( serde_json::to_value(payload).expect("serialize resource templates"), diff --git a/codex-rs/core/tests/suite/scenarios_mcp_resource_messages.rs b/codex-rs/core/tests/suite/scenarios_mcp_resource_messages.rs index e1697011c6..2e1fe2fab7 100644 --- a/codex-rs/core/tests/suite/scenarios_mcp_resource_messages.rs +++ b/codex-rs/core/tests/suite/scenarios_mcp_resource_messages.rs @@ -5,9 +5,12 @@ use super::super::rmcp_client::remote_aware_stdio_server_bin; use anyhow::Result; use codex_core::ConfigRefreshOutcome; use codex_features::Feature; +use codex_protocol::mcp::McpServerConnectionStatus; use codex_protocol::openai_models::ToolMode; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::McpStartupStatus; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; use core_test_support::apps_test_server::AppsTestServer; use core_test_support::context_snapshot; use core_test_support::context_snapshot::ContextSnapshotOptions; @@ -63,7 +66,7 @@ async fn mcp_resource_messages(tool_mode: ToolMode) -> Result<()> { ); } let catalog = serde_json::from_value(json!(messages))?; - let test = test_codex() + let mut test = test_codex() .with_config(move |config| { super::configure_scenario_catalog(config); config.code_mode.disable_in_process_fallback = true; @@ -96,34 +99,70 @@ async fn mcp_resource_messages(tool_mode: ToolMode) -> Result<()> { .build_with_auto_env(&server) .await?; wait_for_mcp_server(&test.codex, "resources").await?; - let mut events = vec![responses::ev_response_created("resources-response")]; - events.extend(if tool_mode == ToolMode::CodeModeOnly { - let script = calls - .iter() - .map(|(name, arguments)| format!("text(await tools.{name}({arguments}));")) - .collect::>() - .join("\n"); - vec![responses::ev_custom_tool_call( - "resources-call", - "exec", - &script, - )] - } else { - calls - .iter() - .map(|(name, arguments)| responses::ev_function_call(name, name, arguments)) - .collect() - }); - events.push(responses::ev_completed("resources-response")); + let mut options = test.start_thread_options().await; + options.session_source = Some(SessionSource::SubAgent(SubAgentSource::Other( + "resource-discovery".to_string(), + ))); + test.codex = test.thread_manager.start_thread(options).await?.thread; + let resource_calls = |id: &str, calls: &[(&str, &str)]| { + let mut events = vec![responses::ev_response_created(id)]; + if tool_mode == ToolMode::CodeModeOnly { + let script = calls + .iter() + .map(|(name, args)| format!("text(await tools.{name}({args}));")) + .collect::>() + .join("\n"); + events.push(responses::ev_custom_tool_call(id, "exec", &script)); + } else { + events.extend(calls.iter().map(|(name, args)| { + responses::ev_function_call(&format!("{id}-{name}"), name, args) + })); + } + events.push(responses::ev_completed(id)); + responses::sse(events) + }; + let broad = [ + ("list_mcp_resources", "{}"), + ("list_mcp_resource_templates", "{}"), + ]; let mock = responses::mount_sse_sequence( &server, - vec![responses::sse(events), responses::sse_completed("done")], + vec![ + resource_calls("broad", &broad), + responses::sse_completed("broad-done"), + resource_calls("resources-call", &calls), + responses::sse_completed("done"), + ], ) .await; - test.submit_turn("List the available resources and templates, then read the example note.") + test.submit_turn("List available MCP resources and templates.") + .await?; + let (config, _) = test.codex.current_mcp_config_and_runtime_context().await; + assert_eq!( + test.codex + .mcp_connection_statuses(&config) + .await + .get("resources"), + Some(&McpServerConnectionStatus::NotStarted) + ); + test.submit_turn("Query the deferred server by name and read its example note.") .await?; let requests = mock.requests(); - assert_eq!(requests.len(), 2); + assert_eq!(requests.len(), 4); + if tool_mode == ToolMode::Direct { + for (tool, key) in [ + ("list_mcp_resources", "resources"), + ("list_mcp_resource_templates", "resourceTemplates"), + ] { + let output = requests[1] + .function_call_output_text(&format!("broad-{tool}")) + .expect("aggregate result"); + assert_eq!( + serde_json::from_str::(&output)?, + json!({key: [], "deferredServers": ["resources"]}) + ); + } + } let body = requests[0].body_json(); for (name, message) in &messages { if tool_mode == ToolMode::CodeModeOnly { @@ -149,17 +188,17 @@ async fn mcp_resource_messages(tool_mode: ToolMode) -> Result<()> { ); } } - assert!(requests[1].body_contains_text("Example Note")); - assert!(requests[1].body_contains_text("memo://codex/{slug}")); + assert!(requests[3].body_contains_text("Example Note")); + assert!(requests[3].body_contains_text("memo://codex/{slug}")); assert!( - requests[1] + requests[3] .body_contains_text("This is a sample MCP resource served by the rmcp test server.") ); if tool_mode == ToolMode::CodeModeOnly { insta::assert_snapshot!( "mcp_resource_messages", context_snapshot::format_request_history_snapshot( - "Catalog resource descriptions and parameter guidance accompany discovery and reading through Code Mode.", + "Broad discovery names the dormant server. Calling it by name returns its resources and templates through Code Mode.", &requests, &ContextSnapshotOptions::default().include_request_settings(), ) diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__mcp_resource_messages__mcp_resource_messages.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__mcp_resource_messages__mcp_resource_messages.snap index d1a7f3a7cb..726f629e5e 100644 --- a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__mcp_resource_messages__mcp_resource_messages.snap +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__mcp_resource_messages__mcp_resource_messages.snap @@ -1,8 +1,8 @@ --- source: core/tests/suite/scenarios_mcp_resource_messages.rs -expression: "context_snapshot::format_request_history_snapshot(\"Catalog resource descriptions and parameter guidance accompany discovery and reading through Code Mode.\",\n&requests, &ContextSnapshotOptions::default().include_request_settings(),)" +expression: "context_snapshot::format_request_history_snapshot(\"Broad discovery names the dormant server. Calling it by name returns its resources and templates through Code Mode.\",\n&requests, &ContextSnapshotOptions::default().include_request_settings(),)" --- -Scenario: Catalog resource descriptions and parameter guidance accompany discovery and reading through Code Mode. +Scenario: Broad discovery names the dormant server. Calling it by name returns its resources and templates through Code Mode. ## Window 1 Settings: @@ -15,11 +15,10 @@ Settings: stream: true text: {"verbosity":"low"} tool_choice: "auto" - tools (7; hash=83df59b286cda93b): + tools (6; hash=2fa2390674c481b0): - custom/exec: Run JavaScript code to orchestrate/compose tool calls - Evaluates the provided JavaScript code in...; hash=2051847a9a5d026d - function/wait: Waits on a yielded `exec` cell and returns new output or completion. - Use `wait` only after `exe...; args=[cell_id, max_tokens, terminate, yield_time_ms]; hash=6692d8f6c55be489 - function/request_user_input: Request user input for one to three short questions and wait for the response. This tool is only ...; args=[questions]; hash=64a729b25c424fc6 - - function/request_user_input_async: Ask the user one or more questions during ongoing work. Use this tool only to request missing inf...; args=[questions]; hash=d6d74f33db330162 - namespace/clock: Tools for reading and waiting on time.; hash=f402e8c5e9b5e317 - function/sleep - namespace/collaboration: Tools for spawning and managing sub-agents.; hash=2c8d6295978dbe63 @@ -74,27 +73,7 @@ Settings: Deferred tool namespaces: - mcp__resources: Use these tools to exercise the rmcp test server. -02:message/developer: - You are `/root`, the primary agent in a team of agents collaborating to fulfill the user's goals. - - At the start of your turn, you are the active agent. - You can spawn sub-agents to handle subtasks, and those sub-agents can spawn their own sub-agents. - All agents in the team, including the agents that you can assign tasks to, are equally intelligent and capable, and have access to the same set of tools. - - You can use `spawn_agent` to create a new agent, `followup_task` to give an existing agent a new task an...message to a running agent without triggering a turn. [hash=7d40ff8b58977202] - `send_message` calls may be read by a human, so ensure they are legible. Always put proper spaces between words and/or numbers. - - - All agents use the same current working directory. - - As a result, edits made by one agent are immediately visible to all other agents. - - When calling `wait_agent`, prefer longer waits (minutes) to avoid busy polling. - - There are 4 available concurrency slots, meaning that up to 4 agents can be active at once, including you. - - Full-history forks (`fork_turns` omitted or `"all"`) inherit the parent model and reasoning effort and d... so, set `fork_turns` to `"none"`. [hash=d5928243dc643bcc] -03:message/developer: - Any earlier instruction enabling proactive multi-agent delegation no longer applies. D...elegation, or parallel agent work. [hash=2b6e037067b4390c] -04:message/user: +02:message/user: @@ -102,14 +81,24 @@ Settings: [hash=0592734c170299c1] -05:message/user: - List the available resources and templates, then read the example note. -06:configuration_update:reasoning={"effort":"low"} +03:message/user: + List available MCP resources and templates. +04:configuration_update:reasoning={"effort":"low"} -- request 2 (turn) -- -07:custom_tool_call/exec:text(await tools.list_mcp_resources({"server":"resources"})); +05:custom_tool_call/exec:text(await tools.list_mcp_resources({})); + text(await tools.list_mcp_resource_templates({})); +06:custom_tool_call_output:Script completed + Wall time seconds + Output: + | {"resources":[],"deferredServers":["resources"]} | {"resourceTemplates":[],"deferredServers":["resources"]} +-- request 3 (turn) -- +07:message/user: + Query the deferred server by name and read its example note. +-- request 4 (turn) -- +08:custom_tool_call/exec:text(await tools.list_mcp_resources({"server":"resources"})); text(await tools.list_mcp_resource_templates({"server":"resources"})); text(await tools.read_mcp_resource({"server":"resources","uri":"memo://codex/example-note"})); -08:custom_tool_call_output:Script completed +09:custom_tool_call_output:Script completed Wall time seconds Output: | {"server":"resources","resources":[{"server":"resources","uri":"memo://codex/example-note","name":"examp...ed for integration tests.","mimeType":"text/plain"}]} [hash=75d187638a7af2d4] | {"server":"resources","resourceTemplates":[{"server":"resources","uriTemplate":"memo://codex/{slug}","na... resources used in tests.","mimeType":"text/plain"}]} [hash=b1c7082eb114fc9b] | {"server":"resources","uri":"memo://codex/example-note","contents":[{"uri":"memo://codex/example-note","...mple MCP resource served by the rmcp test server."}]} [hash=45385fc57640b555]