From be2e4afcd7392339d6adbaf0d31b26316bcaa2ab Mon Sep 17 00:00:00 2001 From: thomas Date: Tue, 28 Jul 2026 05:46:41 +0000 Subject: [PATCH] Add MCP 2026-07-28 discovery support (#35724) ## What changed - Add an opt-in `mcp_2026_07_28` protocol mode while preserving the legacy lifecycle by default. - Negotiate the new protocol over streamable HTTP with `server/discover`, including bounded responses, redirect protection, and fallback only when a response establishes that the endpoint is legacy-only. - Require stdio servers to opt in with `CODEX_MCP_PROTOCOL_VERSION=2026-07-28`, and add a bounded local stdio transport for the modern lifecycle. - Consume paginated tool, resource, and resource-template catalogs in modern mode, reject repeated cursors, and retain discovered server identity. - Reconnect reusable MCP clients when their selected protocol mode changes. ## Testing - Cover HTTP JSON and SSE discovery, legacy fallback and rejection cases, redirects, retries, response limits, and pagination. - Cover local and executor stdio discovery, protocol markers, message limits, and legacy compatibility. GitOrigin-RevId: f6a78816e127d2a482292d63b91c8384f1595903 --- codex-rs/codex-mcp/src/binding_clients.rs | 66 +- codex-rs/codex-mcp/src/connection_manager.rs | 27 +- .../codex-mcp/src/connection_manager_tests.rs | 188 ++- codex-rs/codex-mcp/src/lib.rs | 2 + codex-rs/codex-mcp/src/mcp/mod.rs | 3 + codex-rs/codex-mcp/src/mcp/mod_tests.rs | 1 + codex-rs/codex-mcp/src/pagination.rs | 50 + codex-rs/codex-mcp/src/pagination_tests.rs | 107 ++ codex-rs/codex-mcp/src/rmcp_client.rs | 71 +- codex-rs/core/src/config/config_tests.rs | 24 + codex-rs/core/src/config/mod.rs | 10 + .../test_mcp_2026_discovery_stdio_server.rs | 66 + .../rmcp-client/src/bin/test_stdio_server.rs | 5 + .../src/executor_process_transport.rs | 5 +- .../rmcp-client/src/http_client_adapter.rs | 262 +++- codex-rs/rmcp-client/src/incoming_jsonrpc.rs | 50 + .../rmcp-client/src/incoming_jsonrpc_tests.rs | 65 + codex-rs/rmcp-client/src/lib.rs | 4 + .../rmcp-client/src/local_stdio_transport.rs | 172 +++ codex-rs/rmcp-client/src/protocol_mode.rs | 128 ++ codex-rs/rmcp-client/src/rmcp_client.rs | 99 +- .../rmcp-client/src/stdio_server_launcher.rs | 49 +- .../rmcp-client/src/streamable_http_retry.rs | 14 +- .../src/streamable_http_retry_tests.rs | 5 +- .../rmcp-client/tests/mcp_2026_discovery.rs | 1073 +++++++++++++++++ .../tests/mcp_2026_sse_discovery.rs | 203 ++++ .../tests/mcp_2026_stdio_discovery.rs | 159 +++ .../rmcp-client/tests/stdio_message_limits.rs | 88 ++ 28 files changed, 2891 insertions(+), 105 deletions(-) create mode 100644 codex-rs/codex-mcp/src/pagination.rs create mode 100644 codex-rs/codex-mcp/src/pagination_tests.rs create mode 100644 codex-rs/rmcp-client/src/bin/test_mcp_2026_discovery_stdio_server.rs create mode 100644 codex-rs/rmcp-client/src/incoming_jsonrpc.rs create mode 100644 codex-rs/rmcp-client/src/incoming_jsonrpc_tests.rs create mode 100644 codex-rs/rmcp-client/src/local_stdio_transport.rs create mode 100644 codex-rs/rmcp-client/src/protocol_mode.rs create mode 100644 codex-rs/rmcp-client/tests/mcp_2026_discovery.rs create mode 100644 codex-rs/rmcp-client/tests/mcp_2026_sse_discovery.rs create mode 100644 codex-rs/rmcp-client/tests/mcp_2026_stdio_discovery.rs create mode 100644 codex-rs/rmcp-client/tests/stdio_message_limits.rs diff --git a/codex-rs/codex-mcp/src/binding_clients.rs b/codex-rs/codex-mcp/src/binding_clients.rs index c9eb2087b1..fe0c92e4ff 100644 --- a/codex-rs/codex-mcp/src/binding_clients.rs +++ b/codex-rs/codex-mcp/src/binding_clients.rs @@ -14,6 +14,7 @@ use rmcp::model::ResourceTemplate; use tokio::task::JoinSet; use tracing::warn; +use crate::pagination::collect_paginated; use crate::rmcp_client::ManagedClient; /// The ready clients captured for one model step. @@ -90,28 +91,16 @@ impl McpBindingClients { let client = Arc::clone(&managed.client); let timeout = managed.tool_timeout; join_set.spawn(async move { - let mut collected = Vec::new(); - let mut cursor: Option = None; - loop { - let params = cursor.as_ref().map(|next| { - PaginatedRequestParams::default().with_cursor(Some(next.clone())) - }); - let response = match client.list_resources(params, timeout).await { - Ok(result) => result, - Err(error) => return (server_name, Err(error)), - }; - collected.extend(response.resources); - match response.next_cursor { - Some(next) if cursor.as_ref() == Some(&next) => { - return ( - server_name, - Err(anyhow!("resources/list returned duplicate cursor")), - ); + let resources = + collect_paginated("resources/list", /*overall_timeout*/ None, |params| { + let client = Arc::clone(&client); + async move { + let response = client.list_resources(params, timeout).await?; + Ok((response.resources, response.next_cursor)) } - Some(next) => cursor = Some(next), - None => return (server_name, Ok(collected)), - } - } + }) + .await; + (server_name, resources) }); } collect_resource_results(&mut join_set, "resources").await @@ -131,30 +120,19 @@ impl McpBindingClients { let client = Arc::clone(&managed.client); let timeout = managed.tool_timeout; join_set.spawn(async move { - let mut collected = Vec::new(); - let mut cursor: Option = None; - loop { - let params = cursor.as_ref().map(|next| { - PaginatedRequestParams::default().with_cursor(Some(next.clone())) - }); - let response = match client.list_resource_templates(params, timeout).await { - Ok(result) => result, - Err(error) => return (server_name, Err(error)), - }; - collected.extend(response.resource_templates); - match response.next_cursor { - Some(next) if cursor.as_ref() == Some(&next) => { - return ( - server_name, - Err(anyhow!( - "resources/templates/list returned duplicate cursor" - )), - ); + let templates = collect_paginated( + "resources/templates/list", + /*overall_timeout*/ None, + |params| { + let client = Arc::clone(&client); + async move { + let response = client.list_resource_templates(params, timeout).await?; + Ok((response.resource_templates, response.next_cursor)) } - Some(next) => cursor = Some(next), - None => return (server_name, Ok(collected)), - } - } + }, + ) + .await; + (server_name, templates) }); } collect_resource_results(&mut join_set, "resource templates").await diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index 15e79550ee..752b1a3ac8 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -187,6 +187,7 @@ impl McpConnectionSet { let codex_home = config.codex_home.clone(); let prefix_mcp_tool_names = config.prefix_mcp_tool_names; let non_prefixed_mcp_tool_servers = config.non_prefixed_mcp_tool_servers.clone(); + let protocol_mode = config.protocol_mode; let client_elicitation_capability = config.client_elicitation_capability.clone(); let tool_plugin_provenance = crate::mcp::tool_plugin_provenance(&config); let auth = auth.as_ref(); @@ -292,6 +293,26 @@ impl McpConnectionSet { client_elicitation_capability.clone(), supports_openai_form_elicitation, ); + let expected_protocol_mode = match &configured_config.transport { + McpServerTransportConfig::StreamableHttp { .. } => Some(protocol_mode), + McpServerTransportConfig::Stdio { .. } + if protocol_mode == crate::McpProtocolMode::Legacy => + { + Some(crate::McpProtocolMode::Legacy) + } + McpServerTransportConfig::Stdio { env, .. } => match env + .as_ref() + .and_then(|variables| variables.get("CODEX_MCP_PROTOCOL_VERSION")) + { + None => Some(crate::McpProtocolMode::Legacy), + Some(version) + if version == rmcp::model::ProtocolVersion::V_2026_07_28.as_str() => + { + Some(protocol_mode) + } + Some(_) => None, + }, + }; if let Some(previous_view) = reusable_previous.and_then(|previous| previous.servers.get(&server_name)) { @@ -299,7 +320,10 @@ impl McpConnectionSet { if connection .reusable_client(&connection_identity) .await - .is_some() + .is_some_and(|client| { + expected_protocol_mode + .is_some_and(|expected| client.client.protocol_mode() == expected) + }) { servers.insert( server_name.clone(), @@ -346,6 +370,7 @@ impl McpConnectionSet { runtime_auth_provider, client_elicitation_capability.clone(), supports_openai_form_elicitation, + protocol_mode, ); servers.insert( server_name.clone(), diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index e0742e91f2..d78896d63b 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -226,6 +226,8 @@ struct RefreshTestTransportFactory { tool: Tool, list_started: Option>, release_list: Option>, + next_cursor: Option, + list_requests: Arc, } impl ServerHandler for RefreshTestTransportFactory { @@ -238,13 +240,17 @@ impl ServerHandler for RefreshTestTransportFactory { _request: Option, _context: rmcp::service::RequestContext, ) -> Result { + self.list_requests + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); if let Some(list_started) = &self.list_started { list_started.notify_one(); } if let Some(release_list) = &self.release_list { release_list.notified().await; } - Ok(ListToolsResult::with_all_items(vec![self.tool.clone()])) + let mut result = ListToolsResult::with_all_items(vec![self.tool.clone()]); + result.next_cursor = self.next_cursor.clone(); + Ok(result) } } @@ -346,6 +352,48 @@ impl InProcessTransportFactory for DisconnectingToolsTransportFactory { } } +#[tokio::test] +async fn legacy_tool_catalog_does_not_follow_pagination_cursor() -> anyhow::Result<()> { + let requests = Arc::new(AtomicUsize::new(0)); + let client = Arc::new( + RmcpClient::new_in_process_client(Arc::new(RefreshTestTransportFactory { + tool: create_test_tool("legacy", "first-page").tool, + list_started: None, + release_list: None, + next_cursor: Some("next-page".to_string()), + list_requests: Arc::clone(&requests), + })) + .await?, + ); + client + .initialize( + InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("codex-test", "0.0.0-test"), + ) + .with_protocol_version(ProtocolVersion::V_2025_06_18), + Some(Duration::from_secs(5)), + Box::new(|_, _| async { Err(anyhow!("unexpected elicitation")) }.boxed()), + ) + .await?; + + let tools = list_tools_for_client_uncached( + "legacy", + /*is_codex_apps_mcp_server*/ false, + "test", + &client, + Some(Duration::from_secs(5)), + /*server_instructions*/ None, + ) + .await?; + + assert_eq!(tools.len(), 1); + assert_eq!(tools[0].tool.name.as_ref(), "first-page"); + assert_eq!(requests.load(std::sync::atomic::Ordering::SeqCst), 1); + client.shutdown().await; + Ok(()) +} + async fn create_test_managed_client(tools: Vec) -> ManagedClient { ManagedClient { client: Arc::new( @@ -391,6 +439,8 @@ async fn create_test_manager_with_ready_apps_client( tool: tool.tool.clone(), list_started, release_list, + next_cursor: None, + list_requests: Arc::new(AtomicUsize::new(0)), })) .await?, ); @@ -2568,12 +2618,13 @@ fn reusable_server_identity( runtime_context: &McpRuntimeContext, ) -> McpServerConnectionIdentity { let server = EffectiveMcpServer::configured(config.clone()); + let resolved_environment = runtime_context.resolve_server_environment("docs", config); McpServerConnectionIdentity::new( "docs", &server, OAuthCredentialsStoreMode::default(), AuthKeyringBackendKind::default(), - &Ok(None), + &resolved_environment, runtime_context, /*runtime_auth_provider*/ None, /*auth*/ None, @@ -2796,6 +2847,139 @@ async fn reconciliation_reuses_an_unchanged_ready_server() { ); } +#[tokio::test] +async fn reconciliation_reuses_legacy_stdio_server_with_existing_protocol_marker() { + let runtime_context = McpRuntimeContext::new( + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + PathBuf::from("/tmp"), + ); + let mut config = reusable_server_config("http://127.0.0.1:1"); + config.transport = McpServerTransportConfig::Stdio { + command: "legacy-server".to_string(), + args: Vec::new(), + env: Some(HashMap::from([( + "CODEX_MCP_PROTOCOL_VERSION".to_string(), + "1999-01-01".to_string(), + )])), + env_vars: Vec::new(), + cwd: None, + }; + let previous = manager_with_reusable_ready_server( + &config, + &runtime_context, + vec![create_test_tool("docs", "search")], + ) + .await; + + let reconciled = reconcile_reusable_server(&previous, config, runtime_context).await; + + assert!(previous.shares_test_connection_with(&reconciled, "docs")); +} + +#[tokio::test] +async fn reconciliation_replaces_connection_when_protocol_mode_changes() { + let runtime_context = reusable_server_runtime_context(); + let config = reusable_server_config("http://127.0.0.1:1"); + let previous = manager_with_reusable_ready_server( + &config, + &runtime_context, + vec![create_test_tool("docs", "search")], + ) + .await; + let codex_home = tempdir().expect("tempdir"); + let mut mcp_config = crate::mcp::tests::test_mcp_config(codex_home.path().to_path_buf()); + mcp_config.protocol_mode = codex_rmcp_client::McpProtocolMode::V20260728; + + let reconciled = McpConnectionSet::new( + Some(&previous), + McpPublicationGate::already_published(), + McpRuntimeInput { + config: Arc::new(mcp_config), + plugins_available: false, + ready_selected_capability_roots: Vec::new(), + mcp_servers: HashMap::from([( + "docs".to_string(), + EffectiveMcpServer::configured(config), + )]), + submit_id: "refresh".to_string(), + tx_event: None, + startup_cancellation_token: CancellationToken::new(), + runtime_context, + codex_apps_tools_cache: ConnectorRuntimeManager::default(), + tool_catalog_cache: McpToolCatalogCache::default(), + codex_apps_tools_cache_key: ConnectorRuntimeContextKey::personal( + /*account_id*/ None, /*chatgpt_user_id*/ None, + ), + supports_openai_form_elicitation: false, + auth: None, + codex_apps_auth_manager: None, + elicitation_reviewer: None, + elicitation_lifecycle: None, + }, + ElicitationRequestRouter::default(), + ) + .await; + + assert!(!previous.shares_test_connection_with(&reconciled, "docs")); +} + +#[tokio::test] +async fn reconciliation_reuses_legacy_stdio_server_when_modern_protocol_is_enabled() { + let runtime_context = McpRuntimeContext::new( + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + PathBuf::from("/tmp"), + ); + let mut config = reusable_server_config("http://127.0.0.1:1"); + config.transport = McpServerTransportConfig::Stdio { + command: "legacy-server".to_string(), + args: Vec::new(), + env: None, + env_vars: Vec::new(), + cwd: None, + }; + let previous = manager_with_reusable_ready_server( + &config, + &runtime_context, + vec![create_test_tool("docs", "search")], + ) + .await; + let codex_home = tempdir().expect("tempdir"); + let mut mcp_config = crate::mcp::tests::test_mcp_config(codex_home.path().to_path_buf()); + mcp_config.protocol_mode = codex_rmcp_client::McpProtocolMode::V20260728; + + let reconciled = McpConnectionSet::new( + Some(&previous), + McpPublicationGate::already_published(), + McpRuntimeInput { + config: Arc::new(mcp_config), + plugins_available: false, + ready_selected_capability_roots: Vec::new(), + mcp_servers: HashMap::from([( + "docs".to_string(), + EffectiveMcpServer::configured(config), + )]), + submit_id: "refresh".to_string(), + tx_event: None, + startup_cancellation_token: CancellationToken::new(), + runtime_context, + codex_apps_tools_cache: ConnectorRuntimeManager::default(), + tool_catalog_cache: McpToolCatalogCache::default(), + codex_apps_tools_cache_key: ConnectorRuntimeContextKey::personal( + /*account_id*/ None, /*chatgpt_user_id*/ None, + ), + supports_openai_form_elicitation: false, + auth: None, + codex_apps_auth_manager: None, + elicitation_reviewer: None, + elicitation_lifecycle: None, + }, + ElicitationRequestRouter::default(), + ) + .await; + + assert!(previous.shares_test_connection_with(&reconciled, "docs")); +} + #[tokio::test] async fn reconciliation_updates_elicitation_policy_without_restarting_ready_server() { let runtime_context = reusable_server_runtime_context(); diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index 4cc075d35d..5138018873 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -1,5 +1,6 @@ pub use binding::McpBinding; pub use binding::PreparedMcpCall; +pub use codex_rmcp_client::McpProtocolMode; pub use connection_manager::tool_is_model_visible; pub use elicitation::ElicitationLifecycle; pub use elicitation::ElicitationReviewRequest; @@ -91,6 +92,7 @@ pub(crate) mod connection_manager; pub(crate) mod elicitation; pub(crate) mod mcp; mod openai_docs_source_attribution; +mod pagination; mod plugin_config; mod resource_client; pub(crate) mod rmcp_client; diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index 658d75ec0b..ad269671a1 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -48,6 +48,7 @@ use rmcp::model::ReadResourceResult; use serde_json::Value; use tokio_util::sync::CancellationToken; +use crate::McpProtocolMode; use crate::ResolvedMcpCatalog; use crate::connection_manager::McpConnectionSet; use crate::runtime::McpPublicationGate; @@ -157,6 +158,8 @@ pub struct McpConfig { pub prefix_mcp_tool_names: bool, /// MCP servers whose model-visible tool namespaces omit the `mcp__` prefix. pub non_prefixed_mcp_tool_servers: Vec, + /// Protocol compatibility policy captured when this MCP configuration is created. + pub protocol_mode: McpProtocolMode, /// Client-side elicitation capabilities advertised during MCP initialization. pub client_elicitation_capability: ElicitationCapability, /// Resolved MCP registrations keyed by logical server name. diff --git a/codex-rs/codex-mcp/src/mcp/mod_tests.rs b/codex-rs/codex-mcp/src/mcp/mod_tests.rs index cbd5c7854f..1aa31c934c 100644 --- a/codex-rs/codex-mcp/src/mcp/mod_tests.rs +++ b/codex-rs/codex-mcp/src/mcp/mod_tests.rs @@ -37,6 +37,7 @@ pub(crate) fn test_mcp_config(codex_home: PathBuf) -> McpConfig { apps_enabled: false, prefix_mcp_tool_names: true, non_prefixed_mcp_tool_servers: Vec::new(), + protocol_mode: McpProtocolMode::Legacy, client_elicitation_capability: ElicitationCapability::default(), mcp_server_catalog: ResolvedMcpCatalog::default(), connector_snapshot: codex_connectors::ConnectorSnapshot::default(), diff --git a/codex-rs/codex-mcp/src/pagination.rs b/codex-rs/codex-mcp/src/pagination.rs new file mode 100644 index 0000000000..3b699471be --- /dev/null +++ b/codex-rs/codex-mcp/src/pagination.rs @@ -0,0 +1,50 @@ +use std::collections::HashSet; +use std::future::Future; +use std::time::Duration; + +use anyhow::Result; +use anyhow::anyhow; +use rmcp::model::PaginatedRequestParams; + +pub(crate) async fn collect_paginated( + method: &str, + overall_timeout: Option, + mut fetch: F, +) -> Result> +where + F: FnMut(Option) -> Fut, + Fut: Future, Option)>>, +{ + let collect = async { + let mut collected = Vec::new(); + let mut cursor = None; + let mut seen_cursors = HashSet::new(); + + loop { + let params = cursor.as_ref().map(|next: &String| { + PaginatedRequestParams::default().with_cursor(Some(next.clone())) + }); + let (items, next_cursor) = fetch(params).await?; + collected.extend(items); + + let Some(next_cursor) = next_cursor else { + return Ok(collected); + }; + if !seen_cursors.insert(next_cursor.clone()) { + return Err(anyhow!("{method} returned a repeated pagination cursor")); + } + cursor = Some(next_cursor); + } + }; + + match overall_timeout { + Some(timeout) => tokio::time::timeout(timeout, collect) + .await + .map_err(|_| anyhow!("{method} pagination timed out after {timeout:?}"))?, + None => collect.await, + } +} + +#[cfg(test)] +#[path = "pagination_tests.rs"] +mod tests; diff --git a/codex-rs/codex-mcp/src/pagination_tests.rs b/codex-rs/codex-mcp/src/pagination_tests.rs new file mode 100644 index 0000000000..8e58712d9e --- /dev/null +++ b/codex-rs/codex-mcp/src/pagination_tests.rs @@ -0,0 +1,107 @@ +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use anyhow::anyhow; +use pretty_assertions::assert_eq; + +use super::collect_paginated; + +#[tokio::test] +async fn collects_all_pages_including_an_empty_cursor() { + let requests = Arc::new(Mutex::new(Vec::new())); + let observed = Arc::clone(&requests); + + let pages = collect_paginated("tools/list", /*overall_timeout*/ None, move |params| { + let observed = Arc::clone(&observed); + async move { + let cursor = params.and_then(|params| params.cursor); + observed.lock().expect("request lock").push(cursor.clone()); + match cursor.as_deref() { + None => Ok((vec!["first"], Some(String::new()))), + Some("") => Ok((vec!["second"], Some("last".to_string()))), + Some("last") => Ok((vec!["third"], None)), + Some(cursor) => Err(anyhow!("unexpected cursor: {cursor}")), + } + } + }) + .await + .expect("paginated request succeeds"); + + assert_eq!(pages, vec!["first", "second", "third"]); + assert_eq!( + *requests.lock().expect("request lock"), + vec![None, Some(String::new()), Some("last".to_string())] + ); +} + +#[tokio::test] +async fn rejects_nonconsecutive_repeated_cursors() { + let error = collect_paginated( + "resources/list", + /*overall_timeout*/ None, + |params| async move { + let cursor = params.and_then(|params| params.cursor); + let next = match cursor.as_deref() { + None => "first", + Some("first") => "second", + Some("second") => "first", + Some(cursor) => return Err(anyhow!("unexpected cursor: {cursor}")), + }; + Ok((Vec::<()>::new(), Some(next.to_string()))) + }, + ) + .await + .expect_err("a repeated cursor must fail"); + + assert_eq!( + error.to_string(), + "resources/list returned a repeated pagination cursor" + ); +} + +#[tokio::test] +async fn forwards_page_failures() { + let error = collect_paginated( + "resources/templates/list", + /*overall_timeout*/ None, + |_params| async { Err::<(Vec<()>, Option), _>(anyhow!("page failed")) }, + ) + .await + .expect_err("a page error must fail"); + + assert_eq!(error.to_string(), "page failed"); +} + +#[tokio::test(start_paused = true)] +async fn applies_one_timeout_across_individually_timely_pages() { + let requests = Arc::new(Mutex::new(Vec::new())); + let observed = Arc::clone(&requests); + + let error = collect_paginated("tools/list", Some(Duration::from_secs(5)), move |params| { + let observed = Arc::clone(&observed); + async move { + let cursor = params.and_then(|params| params.cursor); + observed.lock().expect("request lock").push(cursor.clone()); + tokio::time::sleep(Duration::from_secs(2)).await; + + match cursor.as_deref() { + None => Ok((vec!["first"], Some("second".to_string()))), + Some("second") => Ok((vec!["second"], Some("third".to_string()))), + Some("third") => Ok((vec!["third"], None)), + Some(cursor) => Err(anyhow!("unexpected cursor: {cursor}")), + } + } + }) + .await + .expect_err("the combined page duration must exceed the shared timeout"); + + assert_eq!( + error.to_string(), + "tools/list pagination timed out after 5s" + ); + assert_eq!( + *requests.lock().expect("request lock"), + vec![None, Some("second".to_string()), Some("third".to_string())] + ); +} diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 8901c36723..3f4ca26efa 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -26,6 +26,7 @@ use crate::elicitation::ElicitationRequestManager; use crate::mcp::CODEX_APPS_MCP_SERVER_NAME; use crate::mcp::ToolPluginProvenance; use crate::openai_docs_source_attribution::maybe_with_openai_docs_source_attribution; +use crate::pagination::collect_paginated; use crate::runtime::McpRuntimeContext; use crate::runtime::emit_duration; use crate::server::EffectiveMcpServer; @@ -52,6 +53,7 @@ use codex_protocol::protocol::McpStartupStatus; use codex_protocol::protocol::McpStartupUpdateEvent; use codex_rmcp_client::ExecutorStdioServerLauncher; use codex_rmcp_client::LocalStdioServerLauncher; +use codex_rmcp_client::McpProtocolMode; use codex_rmcp_client::RmcpClient; use codex_rmcp_client::StdioServerLauncher; use codex_rmcp_client::ToolWithConnectorId; @@ -280,6 +282,7 @@ struct ManagedClientStartup { runtime_auth_provider: Option, client_elicitation_capability: ElicitationCapability, supports_openai_form_elicitation: bool, + protocol_mode: McpProtocolMode, cancel_token: CancellationToken, startup_complete: Arc, } @@ -300,6 +303,7 @@ impl ManagedClientStartup { runtime_auth_provider, client_elicitation_capability, supports_openai_form_elicitation, + protocol_mode, cancel_token, startup_complete, } = self.clone(); @@ -329,6 +333,7 @@ impl ManagedClientStartup { runtime_context, resolved_environment, runtime_auth_provider, + protocol_mode, ), ) .await @@ -415,6 +420,7 @@ impl AsyncManagedClient { runtime_auth_provider: Option, client_elicitation_capability: ElicitationCapability, supports_openai_form_elicitation: bool, + protocol_mode: McpProtocolMode, ) -> Self { let is_codex_apps_mcp_server = server_name == CODEX_APPS_MCP_SERVER_NAME; let reconnect_server_name = server_name.clone(); @@ -441,6 +447,7 @@ impl AsyncManagedClient { runtime_auth_provider, client_elicitation_capability, supports_openai_form_elicitation, + protocol_mode, cancel_token: cancel_token.clone(), startup_complete: Arc::clone(&startup_complete), }); @@ -584,21 +591,36 @@ pub(crate) async fn list_tools_for_client_uncached( server_instructions: Option<&str>, ) -> Result> { let fetch_start = Instant::now(); - let resp = client - .list_tools_with_connector_ids(/*params*/ None, timeout) - .await?; - let tools = resp - .tools - .into_iter() - .map(|tool| { - tool_info_from_listed_tool( - server_name, - is_codex_apps_mcp_server, - server_instructions, - tool, - ) - }) - .collect(); + let tools = match client.protocol_mode() { + McpProtocolMode::Legacy => { + client + .list_tools_with_connector_ids(/*params*/ None, timeout) + .await? + .tools + } + McpProtocolMode::V20260728 => { + collect_paginated("tools/list", timeout, |params| { + let client = Arc::clone(client); + async move { + let response = client + .list_tools_with_connector_ids(params, timeout) + .await?; + Ok((response.tools, response.next_cursor)) + } + }) + .await? + } + } + .into_iter() + .map(|tool| { + tool_info_from_listed_tool( + server_name, + is_codex_apps_mcp_server, + server_instructions, + tool, + ) + }) + .collect(); if is_codex_apps_mcp_server { emit_duration( MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC, @@ -960,6 +982,7 @@ struct StartServerTaskParams { supports_openai_form_elicitation: bool, } +#[allow(clippy::too_many_arguments)] #[instrument(level = "trace", skip_all, fields(server_name = %server_name))] async fn make_rmcp_client( server_name: &str, @@ -969,6 +992,7 @@ async fn make_rmcp_client( runtime_context: McpRuntimeContext, resolved_environment: std::result::Result>, String>, runtime_auth_provider: Option, + protocol_mode: McpProtocolMode, ) -> Result { let config = server.config().clone(); let resolved_environment = @@ -1010,9 +1034,17 @@ async fn make_rmcp_client( }; let cwd = cwd.map(codex_utils_path_uri::LegacyAppPathString::into_string); - RmcpClient::new_stdio_client(command_os, args_os, env_os, &env_vars, cwd, launcher) - .await - .map_err(|err| StartupOutcomeError::from(anyhow!(err))) + RmcpClient::new_stdio_client_with_protocol_mode( + command_os, + args_os, + env_os, + &env_vars, + cwd, + launcher, + protocol_mode, + ) + .await + .map_err(|err| StartupOutcomeError::from(anyhow!(err))) } McpServerTransportConfig::StreamableHttp { url, @@ -1030,7 +1062,7 @@ async fn make_rmcp_client( Ok(token) => token, Err(error) => return Err(error.into()), }; - RmcpClient::new_streamable_http_client( + RmcpClient::new_streamable_http_client_with_protocol_mode( server_name, &url, resolved_bearer_token, @@ -1040,6 +1072,7 @@ async fn make_rmcp_client( keyring_backend_kind, http_client, runtime_auth_provider, + protocol_mode, ) .await .map_err(StartupOutcomeError::from) diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index cf0690ac01..3fefc95b35 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -6396,6 +6396,30 @@ async fn to_mcp_config_flows_mcp_tool_prefix_from_feature() -> std::io::Result<( Ok(()) } +#[tokio::test] +async fn to_mcp_config_flows_mcp_2026_feature_from_config() -> std::io::Result<()> { + let codex_home = TempDir::new()?; + let mut config = Config::load_from_base_config_with_overrides( + ConfigToml::default(), + ConfigOverrides::default(), + codex_home.abs(), + ) + .await?; + let plugins_manager = PluginsManager::new(codex_home.path().to_path_buf()); + + let mcp_config = config.to_mcp_config(&plugins_manager).await; + assert_eq!(mcp_config.protocol_mode, codex_mcp::McpProtocolMode::Legacy); + + let _ = config.features.enable(Feature::Mcp20260728); + let mcp_config = config.to_mcp_config(&plugins_manager).await; + assert_eq!( + mcp_config.protocol_mode, + codex_mcp::McpProtocolMode::V20260728 + ); + + Ok(()) +} + #[tokio::test] async fn to_mcp_config_preserves_auth_elicitation_feature_from_config() -> std::io::Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 3bff0d0fb9..6e9358806f 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -78,6 +78,7 @@ use codex_login::AuthManagerConfig; use codex_login::AuthRouteConfig; use codex_mcp::McpConfig; use codex_mcp::McpPluginAttribution; +use codex_mcp::McpProtocolMode; use codex_mcp::McpServerRegistration; use codex_mcp::ResolvedMcpCatalog; use codex_memories_read::memory_root; @@ -1734,6 +1735,7 @@ impl Config { } else { Vec::new() }, + protocol_mode: self.mcp_protocol_mode(), client_elicitation_capability: if self.features.enabled(Feature::AuthElicitation) { ElicitationCapability::new() .with_form(FormElicitationCapability::new()) @@ -1756,6 +1758,14 @@ impl Config { || self.non_prefixed_mcp_tool_servers.is_some() } + pub fn mcp_protocol_mode(&self) -> McpProtocolMode { + if self.features.enabled(Feature::Mcp20260728) { + McpProtocolMode::V20260728 + } else { + McpProtocolMode::Legacy + } + } + pub async fn rebuild_preserving_session_layers( &self, refreshed_config: &Config, diff --git a/codex-rs/rmcp-client/src/bin/test_mcp_2026_discovery_stdio_server.rs b/codex-rs/rmcp-client/src/bin/test_mcp_2026_discovery_stdio_server.rs new file mode 100644 index 0000000000..fc3af65944 --- /dev/null +++ b/codex-rs/rmcp-client/src/bin/test_mcp_2026_discovery_stdio_server.rs @@ -0,0 +1,66 @@ +use std::io; +use std::io::BufRead; +use std::io::Write; + +use anyhow::Result; +use serde_json::Value; +use serde_json::json; + +fn main() -> Result<()> { + assert!(std::env::var_os("CODEX_MCP_PROTOCOL_VERSION").is_none()); + let mut stdout = io::stdout().lock(); + + for line in io::stdin().lock().lines() { + let request: Value = serde_json::from_str(&line?)?; + let result = match request["method"].as_str() { + Some("server/discover") => { + assert_eq!( + request["params"]["_meta"]["io.modelcontextprotocol/protocolVersion"], + "2026-07-28" + ); + json!({ + "resultType": "complete", + "supportedVersions": ["2026-07-28"], + "capabilities": {"tools": {}, "resources": {}}, + "_meta": { + "io.modelcontextprotocol/serverInfo": { + "name": "strict-stdio-discovery", + "version": "1.0.0", + }, + }, + "ttlMs": 0, + "cacheScope": "private", + }) + } + Some("tools/list") => json!({ + "resultType": "complete", + "tools": [{ + "name": "stdio_echo", + "inputSchema": {"type": "object"}, + }], + }), + Some("resources/list") => json!({ + "resultType": "complete", + "resources": [{ + "uri": "test://stdio/resource", + "name": "stdio resource", + }], + }), + Some("resources/templates/list") => json!({ + "resultType": "complete", + "resourceTemplates": [{ + "uriTemplate": "test://stdio/{name}", + "name": "stdio resource template", + }], + }), + method => anyhow::bail!("unexpected modern stdio discovery method: {method:?}"), + }; + serde_json::to_writer( + &mut stdout, + &json!({"jsonrpc": "2.0", "id": request["id"], "result": result}), + )?; + stdout.write_all(b"\n")?; + stdout.flush()?; + } + Ok(()) +} diff --git a/codex-rs/rmcp-client/src/bin/test_stdio_server.rs b/codex-rs/rmcp-client/src/bin/test_stdio_server.rs index 0978fd073a..19f95e1534 100644 --- a/codex-rs/rmcp-client/src/bin/test_stdio_server.rs +++ b/codex-rs/rmcp-client/src/bin/test_stdio_server.rs @@ -128,6 +128,11 @@ impl TestToolServer { { echo.description = Some(Cow::Owned(format!("Echo from {process_label}."))); } + if std::env::var_os("MCP_TEST_OVERSIZED_TOOL_DESCRIPTION").is_some() + && let Some(echo) = tools.iter_mut().find(|tool| tool.name == "echo") + { + echo.description = Some(Cow::Owned("x".repeat(8 * 1024 * 1024 + 1))); + } let resources = vec![Self::memo_resource()]; let resource_templates = vec![Self::memo_template()]; Self { diff --git a/codex-rs/rmcp-client/src/executor_process_transport.rs b/codex-rs/rmcp-client/src/executor_process_transport.rs index 3ac78a4c45..4f3271f6e1 100644 --- a/codex-rs/rmcp-client/src/executor_process_transport.rs +++ b/codex-rs/rmcp-client/src/executor_process_transport.rs @@ -37,7 +37,6 @@ use rmcp::service::RoleClient; use rmcp::service::RxJsonRpcMessage; use rmcp::service::TxJsonRpcMessage; use rmcp::transport::Transport; -use serde_json::from_slice; use serde_json::to_vec; use tokio::runtime::Handle; use tokio::sync::Semaphore; @@ -46,6 +45,8 @@ use tracing::debug; use tracing::info; use tracing::warn; +use crate::incoming_jsonrpc::deserialize_incoming_jsonrpc_message; + static PROCESS_COUNTER: AtomicUsize = AtomicUsize::new(1); // Tool results can make valid MCP responses large, so keep the protocol // ceiling well above ordinary messages while still bounding hostile input. @@ -426,7 +427,7 @@ impl ExecutorProcessTransport { None => return None, }; let line = Self::trim_trailing_carriage_return(line); - match from_slice::>(&line) { + match deserialize_incoming_jsonrpc_message(&line) { Ok(message) => return Some(message), Err(error) => { debug!( diff --git a/codex-rs/rmcp-client/src/http_client_adapter.rs b/codex-rs/rmcp-client/src/http_client_adapter.rs index fee3b3fc23..87995b5bea 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter.rs @@ -31,8 +31,14 @@ use reqwest::header::HeaderName; use rmcp::model::ClientJsonRpcMessage; use rmcp::model::ClientNotification; use rmcp::model::ConstString; +use rmcp::model::DiscoverRequestMethod; +use rmcp::model::ErrorCode; +use rmcp::model::ErrorData; use rmcp::model::JsonRpcMessage; +use rmcp::model::ProtocolVersion; use rmcp::model::ServerJsonRpcMessage; +use rmcp::model::ServerResult; +use rmcp::transport::common::http_header::HEADER_MCP_PROTOCOL_VERSION; use rmcp::transport::streamable_http_client::AuthRequiredError; use rmcp::transport::streamable_http_client::InsufficientScopeError; use rmcp::transport::streamable_http_client::StreamableHttpClient; @@ -41,6 +47,9 @@ use rmcp::transport::streamable_http_client::StreamableHttpPostResponse; use sse_stream::Sse; use sse_stream::SseStream; +use crate::incoming_jsonrpc::deserialize_incoming_jsonrpc_message; +use crate::local_stdio_transport::MAX_MCP_STDIO_LINE_BYTES; + mod www_authenticate; use self::www_authenticate::insufficient_scope_challenge; @@ -65,6 +74,8 @@ pub(crate) enum StreamableHttpClientAdapterError { HttpRequest(#[from] ExecServerError), #[error("invalid HTTP header: {0}")] Header(String), + #[error("MCP response body exceeds {maximum_bytes} bytes")] + ResponseTooLarge { maximum_bytes: usize }, } impl StreamableHttpClientAdapter { @@ -126,6 +137,18 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { )?; } + let maximum_response_bytes = (mcp_method.as_deref() == Some(DiscoverRequestMethod::VALUE) + || headers + .get(HEADER_MCP_PROTOCOL_VERSION) + .and_then(|value| value.to_str().ok()) + == Some(ProtocolVersion::V_2026_07_28.as_str())) + .then_some(MAX_MCP_STDIO_LINE_BYTES); + let redirect_policy = if mcp_method.as_deref() == Some(DiscoverRequestMethod::VALUE) { + HttpRedirectPolicy::Stop + } else { + mcp_redirect_policy(&headers) + }; + let body = serde_json::to_vec(&message).map_err(StreamableHttpError::Deserialize)?; let has_authorization_header = headers.contains_key(AUTHORIZATION); let response = self @@ -136,7 +159,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { headers: protocol_headers(&headers), body: Some(body.into()), timeout_ms: None, - redirect_policy: HttpRedirectPolicy::Follow, + redirect_policy, request_id: "buffered-request".to_string(), stream_response: true, }) @@ -190,14 +213,42 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { let content_type = response_header(&response.headers, CONTENT_TYPE); let session_id = response_header(&response.headers, HEADER_SESSION_ID); if !status_is_success(response.status) { - let body = collect_body(&mut body_stream).await?; + let body = collect_body(&mut body_stream, maximum_response_bytes).await?; if !retryable_post_response_status(mcp_method.as_deref(), response.status) && content_type .as_deref() .is_some_and(|content_type| content_type.starts_with(JSON_MIME_TYPE)) - && let Some(message) = parse_json_rpc_error(&body) + && let Some(response_message) = parse_json_rpc_error(&body) { - return Ok(StreamableHttpPostResponse::Json(message, session_id)); + return Ok(StreamableHttpPostResponse::Json( + legacy_discovery_fallback_response( + &message, + response_message, + response.status == StatusCode::BAD_REQUEST.as_u16() && !has_session_id, + ), + session_id, + )); + } + if mcp_method.as_deref() == Some(DiscoverRequestMethod::VALUE) + && !has_session_id + && matches!( + StatusCode::from_u16(response.status).ok(), + Some(StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED) + ) + && content_type + .as_deref() + .is_none_or(|content_type| !content_type.starts_with(JSON_MIME_TYPE)) + && let JsonRpcMessage::Request(request) = &message + { + let legacy_error = ServerJsonRpcMessage::error( + ErrorData::new( + ErrorCode::METHOD_NOT_FOUND, + "legacy MCP endpoint does not support server/discover", + None, + ), + Some(request.id.clone()), + ); + return Ok(StreamableHttpPostResponse::Json(legacy_error, session_id)); } return Err(StreamableHttpError::UnexpectedServerResponse( format!( @@ -210,17 +261,55 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { } match content_type.as_deref() { Some(content_type) if content_type.starts_with(EVENT_STREAM_MIME_TYPE) => { - let event_stream = sse_stream_from_body(body_stream); + let mut event_stream = sse_stream_from_body(body_stream); + if mcp_method.as_deref() == Some(DiscoverRequestMethod::VALUE) { + while let Some(event) = event_stream.next().await { + let event = event.map_err(StreamableHttpError::Sse)?; + if !matches!(event.event.as_deref(), None | Some("") | Some("message")) { + continue; + } + let Some(data) = event.data.as_deref() else { + continue; + }; + if data.trim().is_empty() { + continue; + } + + let response = deserialize_incoming_jsonrpc_message(data.as_bytes()) + .map_err(StreamableHttpError::Deserialize)?; + let response = legacy_discovery_fallback_response( + &message, response, /*allow_uncorrelated_http_rejection*/ false, + ); + if matches!( + &response, + JsonRpcMessage::Response(_) | JsonRpcMessage::Error(_) + ) { + return Ok(StreamableHttpPostResponse::Json(response, session_id)); + } + } + + return Err(StreamableHttpError::UnexpectedServerResponse( + "empty sse stream".into(), + )); + } + Ok(StreamableHttpPostResponse::Sse(event_stream, session_id)) } Some(content_type) if content_type.starts_with(JSON_MIME_TYPE) => { - let body = collect_body(&mut body_stream).await?; - let message: ServerJsonRpcMessage = - serde_json::from_slice(&body).map_err(StreamableHttpError::Deserialize)?; - Ok(StreamableHttpPostResponse::Json(message, session_id)) + let body = collect_body(&mut body_stream, maximum_response_bytes).await?; + let response_message = deserialize_incoming_jsonrpc_message(&body) + .map_err(StreamableHttpError::Deserialize)?; + Ok(StreamableHttpPostResponse::Json( + legacy_discovery_fallback_response( + &message, + response_message, + /*allow_uncorrelated_http_rejection*/ false, + ), + session_id, + )) } _ => { - let body = collect_body(&mut body_stream).await?; + let body = collect_body(&mut body_stream, maximum_response_bytes).await?; let content_type = content_type.unwrap_or_else(|| "missing-content-type".into()); Err(StreamableHttpError::UnexpectedContentType(Some(format!( "{content_type}; body: {}", @@ -254,6 +343,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { session.to_string(), StreamableHttpClientAdapterError::Header, )?; + let redirect_policy = mcp_redirect_policy(&headers); let response = self .http_client @@ -263,7 +353,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { headers: protocol_headers(&headers), body: None, timeout_ms: None, - redirect_policy: HttpRedirectPolicy::Follow, + redirect_policy, request_id: "buffered-request".to_string(), stream_response: false, }) @@ -326,6 +416,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { StreamableHttpClientAdapterError::Header, )?; } + let redirect_policy = mcp_redirect_policy(&headers); let (response, body_stream) = self .http_client @@ -335,7 +426,7 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { headers: protocol_headers(&headers), body: None, timeout_ms: None, - redirect_policy: HttpRedirectPolicy::Follow, + redirect_policy, request_id: "buffered-request".to_string(), stream_response: true, }) @@ -517,7 +608,12 @@ fn retryable_post_response_status(mcp_method: Option<&str>, status: u16) -> bool is_retryable_http_status(status) && matches!( mcp_method, - Some("initialize" | "notifications/initialized" | "tools/list") + Some( + DiscoverRequestMethod::VALUE + | "initialize" + | "notifications/initialized" + | "tools/list" + ) ) } @@ -540,8 +636,141 @@ fn parse_json_rpc_error(body: &[u8]) -> Option { } } +fn mcp_redirect_policy(headers: &HeaderMap) -> HttpRedirectPolicy { + if headers + .get(HEADER_MCP_PROTOCOL_VERSION) + .and_then(|value| value.to_str().ok()) + == Some(ProtocolVersion::V_2026_07_28.as_str()) + { + HttpRedirectPolicy::Stop + } else { + HttpRedirectPolicy::Follow + } +} + +// rmcp's automatic lifecycle does not yet recognize deployed legacy discovery +// rejection shapes. Remove this compatibility shim once the SDK does: +// https://github.com/modelcontextprotocol/rust-sdk/issues/1040 +fn legacy_discovery_fallback_response( + request: &ClientJsonRpcMessage, + response: ServerJsonRpcMessage, + allow_uncorrelated_http_rejection: bool, +) -> ServerJsonRpcMessage { + let JsonRpcMessage::Request(request) = request else { + return response; + }; + if request.request.method() != DiscoverRequestMethod::VALUE { + return response; + } + + if let JsonRpcMessage::Error(error) = &response + && error.error.code == ErrorCode::METHOD_NOT_FOUND + && error.id.as_ref() != Some(&request.id) + { + return ServerJsonRpcMessage::error( + ErrorData::new( + ErrorCode::INVALID_REQUEST, + "server/discover method-not-found response did not match its request ID", + None, + ), + Some(request.id.clone()), + ); + } + + let only_known_legacy_versions = match &response { + JsonRpcMessage::Response(response) if response.id == request.id => match &response.result { + ServerResult::DiscoverResult(result) => { + only_known_legacy_protocol_versions(&result.supported_versions) + } + _ => false, + }, + JsonRpcMessage::Error(error) if error.id.as_ref() == Some(&request.id) => { + (error.error.code == ErrorCode::UNSUPPORTED_PROTOCOL_VERSION + && error + .error + .data + .as_ref() + .and_then(|data| data.get("supported")) + .and_then(|supported| { + serde_json::from_value::>(supported.clone()).ok() + }) + .is_some_and(|supported| only_known_legacy_protocol_versions(&supported))) + || matches!( + error.error.code, + ErrorCode::UNSUPPORTED_PROTOCOL_VERSION + | ErrorCode::INVALID_REQUEST + | ErrorCode::INVALID_PARAMS + ) && explicitly_rejects_modern_protocol_version(&error.error.message) + } + JsonRpcMessage::Error(error) + if error.id.is_none() + && allow_uncorrelated_http_rejection + && error.error.code.0 == -32000 => + { + rejects_modern_protocol_at_legacy_http_boundary(&error.error.message) + } + _ => false, + }; + + if only_known_legacy_versions { + ServerJsonRpcMessage::error( + ErrorData::new( + ErrorCode::METHOD_NOT_FOUND, + "MCP discovery only advertised legacy protocol versions", + None, + ), + Some(request.id.clone()), + ) + } else { + response + } +} + +fn only_known_legacy_protocol_versions(versions: &[ProtocolVersion]) -> bool { + !versions.is_empty() + && versions.iter().all(|version| { + ProtocolVersion::KNOWN_VERSIONS.contains(version) + && version < &ProtocolVersion::V_2026_07_28 + }) +} + +fn explicitly_rejects_modern_protocol_version(message: &str) -> bool { + message + .trim() + .eq_ignore_ascii_case("unsupported protocol version: 2026-07-28") +} + +fn rejects_modern_protocol_at_legacy_http_boundary(message: &str) -> bool { + if message == "Bad Request: No valid session ID provided" { + return true; + } + + let Some(supported) = message + .strip_prefix("Bad Request: Unsupported protocol version: 2026-07-28 (supported versions: ") + .and_then(|supported| supported.strip_suffix(')')) + else { + return false; + }; + + let versions = supported.split(',').map(str::trim).collect::>(); + !versions.is_empty() + && versions.contains(&"2025-06-18") + && versions.iter().all(|version| { + let bytes = version.as_bytes(); + bytes.len() == 10 + && bytes[4] == b'-' + && bytes[7] == b'-' + && bytes + .iter() + .enumerate() + .all(|(index, byte)| matches!(index, 4 | 7) || byte.is_ascii_digit()) + && *version < "2026-07-28" + }) +} + async fn collect_body( body_stream: &mut HttpResponseBodyStream, + maximum_bytes: Option, ) -> std::result::Result, StreamableHttpError> { let mut body = Vec::new(); while let Some(chunk) = body_stream @@ -550,6 +779,13 @@ async fn collect_body( .map_err(StreamableHttpClientAdapterError::from) .map_err(StreamableHttpError::Client)? { + if let Some(maximum_bytes) = maximum_bytes + && chunk.len() > maximum_bytes.saturating_sub(body.len()) + { + return Err(StreamableHttpError::Client( + StreamableHttpClientAdapterError::ResponseTooLarge { maximum_bytes }, + )); + } body.extend_from_slice(&chunk); } Ok(body) diff --git a/codex-rs/rmcp-client/src/incoming_jsonrpc.rs b/codex-rs/rmcp-client/src/incoming_jsonrpc.rs new file mode 100644 index 0000000000..6e797bbccc --- /dev/null +++ b/codex-rs/rmcp-client/src/incoming_jsonrpc.rs @@ -0,0 +1,50 @@ +//! Compatibility decoding for modern discovery responses whose server identity +//! is represented in namespaced metadata instead of a top-level field. + +use rmcp::model::DiscoverResult; +use rmcp::model::JsonRpcResponse; +use rmcp::model::ServerJsonRpcMessage; +use rmcp::model::ServerResult; +use serde_json::Value; + +// Remove this compatibility decoder once rmcp accepts namespaced server identity: +// https://github.com/modelcontextprotocol/rust-sdk/issues/1039 +const SERVER_INFO_METADATA_KEY: &str = "io.modelcontextprotocol/serverInfo"; + +/// Decode at the transport boundary, before rmcp's untagged result union can +/// mistake a metadata-only discovery response for a completed tool call. +pub(crate) fn deserialize_incoming_jsonrpc_message( + bytes: &[u8], +) -> serde_json::Result { + let mut message: Value = serde_json::from_slice(bytes)?; + let Some(result) = message.get_mut("result").and_then(Value::as_object_mut) else { + return serde_json::from_value(message); + }; + + if !result.contains_key("supportedVersions") || !result.contains_key("capabilities") { + return serde_json::from_value(message); + } + + let Some(server_info) = result + .get("_meta") + .and_then(Value::as_object) + .and_then(|metadata| metadata.get(SERVER_INFO_METADATA_KEY)) + .cloned() + else { + return serde_json::from_value(message); + }; + + if !result.contains_key("serverInfo") { + result.insert("serverInfo".to_owned(), server_info); + } + + let response: JsonRpcResponse = serde_json::from_value(message)?; + Ok(ServerJsonRpcMessage::response( + ServerResult::DiscoverResult(response.result), + response.id, + )) +} + +#[cfg(test)] +#[path = "incoming_jsonrpc_tests.rs"] +mod tests; diff --git a/codex-rs/rmcp-client/src/incoming_jsonrpc_tests.rs b/codex-rs/rmcp-client/src/incoming_jsonrpc_tests.rs new file mode 100644 index 0000000000..04e42ae2ca --- /dev/null +++ b/codex-rs/rmcp-client/src/incoming_jsonrpc_tests.rs @@ -0,0 +1,65 @@ +use pretty_assertions::assert_eq; +use rmcp::model::JsonRpcMessage; +use rmcp::model::ServerResult; +use serde_json::json; + +use super::deserialize_incoming_jsonrpc_message; + +#[test] +fn discovery_accepts_metadata_namespaced_server_identity() { + let message = json!({ + "jsonrpc": "2.0", + "id": 1, + "result": { + "resultType": "complete", + "supportedVersions": ["2026-07-28"], + "capabilities": {"tools": {}}, + "ttlMs": 0, + "cacheScope": "private", + "_meta": { + "io.modelcontextprotocol/serverInfo": { + "name": "modern-server", + "version": "1.0.0", + }, + "retained": "metadata", + }, + }, + }); + + let decoded = deserialize_incoming_jsonrpc_message(&serde_json::to_vec(&message).unwrap()) + .expect("metadata-only identity must decode as discovery"); + let JsonRpcMessage::Response(response) = decoded else { + panic!("expected a discovery response"); + }; + let ServerResult::DiscoverResult(result) = response.result else { + panic!("metadata-only identity must not become a completed tool result"); + }; + assert_eq!(result.server_info.name, "modern-server"); + assert_eq!( + result.meta.and_then(|meta| meta.get("retained").cloned()), + Some(json!("metadata")) + ); +} + +#[test] +fn extension_result_with_discovery_like_fields_remains_a_custom_result() { + let result = json!({ + "supportedVersions": ["custom-extension-v1"], + "capabilities": {"custom": true}, + }); + let message = json!({ + "jsonrpc": "2.0", + "id": 2, + "result": result, + }); + + let decoded = deserialize_incoming_jsonrpc_message(&serde_json::to_vec(&message).unwrap()) + .expect("legacy extension result must deserialize normally"); + let JsonRpcMessage::Response(response) = decoded else { + panic!("expected a legacy extension response"); + }; + let ServerResult::CustomResult(custom) = response.result else { + panic!("legacy extension response must remain a custom result"); + }; + assert_eq!(custom.0, result); +} diff --git a/codex-rs/rmcp-client/src/lib.rs b/codex-rs/rmcp-client/src/lib.rs index f54ca94185..5febefe281 100644 --- a/codex-rs/rmcp-client/src/lib.rs +++ b/codex-rs/rmcp-client/src/lib.rs @@ -3,11 +3,14 @@ mod elicitation_client_service; mod executor_process_transport; mod http_client_adapter; mod in_process_transport; +mod incoming_jsonrpc; +mod local_stdio_transport; mod logging_client_handler; mod oauth; mod oauth_http_client; mod perform_oauth_login; mod program_resolver; +mod protocol_mode; mod rmcp_client; mod startup_error; mod stdio_server_launcher; @@ -36,6 +39,7 @@ pub use perform_oauth_login::perform_oauth_login; pub use perform_oauth_login::perform_oauth_login_return_url; pub use perform_oauth_login::perform_oauth_login_return_url_with_http_client; pub use perform_oauth_login::perform_oauth_login_silent; +pub use protocol_mode::McpProtocolMode; pub use rmcp::model::ElicitationAction; pub use rmcp_client::Elicitation; pub use rmcp_client::ElicitationResponse; diff --git a/codex-rs/rmcp-client/src/local_stdio_transport.rs b/codex-rs/rmcp-client/src/local_stdio_transport.rs new file mode 100644 index 0000000000..1d69afa9d0 --- /dev/null +++ b/codex-rs/rmcp-client/src/local_stdio_transport.rs @@ -0,0 +1,172 @@ +//! Bounded, compatibility-aware transport for modern locally spawned MCP servers. + +use std::future::Future; +use std::io; +use std::process::Stdio; +use std::sync::Arc; +use std::time::Duration; + +use memchr::memchr; +use rmcp::service::RoleClient; +use rmcp::service::RxJsonRpcMessage; +use rmcp::service::TxJsonRpcMessage; +use rmcp::transport::Transport; +use tokio::io::AsyncBufReadExt; +use tokio::io::AsyncWriteExt; +use tokio::io::BufReader; +use tokio::process::Child; +use tokio::process::ChildStderr; +use tokio::process::ChildStdin; +use tokio::process::ChildStdout; +use tokio::process::Command; +use tokio::sync::Mutex; +use tracing::debug; +use tracing::warn; + +use crate::incoming_jsonrpc::deserialize_incoming_jsonrpc_message; + +/// Match the existing executor stdio transport and the production MCP limit. +pub(crate) const MAX_MCP_STDIO_LINE_BYTES: usize = 8 * 1024 * 1024; + +pub(super) struct LocalStdioTransport { + child: Child, + stdin: Arc>>, + stdout: BufReader, + pending_line: Vec, + program_name: String, +} + +impl LocalStdioTransport { + pub(super) fn spawn( + mut command: Command, + program_name: String, + ) -> io::Result<(Self, Option)> { + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = command.spawn()?; + let stdin = child + .stdin + .take() + .ok_or_else(|| io::Error::other("MCP server stdin was not piped"))?; + let stdout = child + .stdout + .take() + .ok_or_else(|| io::Error::other("MCP server stdout was not piped"))?; + let stderr = child.stderr.take(); + + Ok(( + Self { + child, + stdin: Arc::new(Mutex::new(Some(stdin))), + stdout: BufReader::new(stdout), + pending_line: Vec::new(), + program_name, + }, + stderr, + )) + } + + pub(super) fn id(&self) -> Option { + self.child.id() + } + + async fn receive_message(&mut self) -> Option> { + loop { + let bytes = match self.stdout.fill_buf().await { + Ok(bytes) => bytes, + Err(error) => { + warn!( + "Failed to read MCP server stdout ({}): {error}", + self.program_name + ); + return None; + } + }; + + if bytes.is_empty() { + if self.pending_line.is_empty() { + return None; + } + return self.decode_pending_message(); + } + + let newline = memchr(b'\n', bytes); + let content_len = newline.unwrap_or(bytes.len()); + if content_len > MAX_MCP_STDIO_LINE_BYTES.saturating_sub(self.pending_line.len()) { + warn!( + "MCP stdio line exceeds {MAX_MCP_STDIO_LINE_BYTES} bytes ({}); closing transport", + self.program_name + ); + self.pending_line.clear(); + return None; + } + + self.pending_line.extend_from_slice(&bytes[..content_len]); + let consumed = content_len + usize::from(newline.is_some()); + self.stdout.consume(consumed); + + if newline.is_some() + && let Some(message) = self.decode_pending_message() + { + return Some(message); + } + } + } + + fn decode_pending_message(&mut self) -> Option> { + let line = std::mem::take(&mut self.pending_line); + let line = line.strip_suffix(b"\r").unwrap_or(&line); + match deserialize_incoming_jsonrpc_message(line) { + Ok(message) => Some(message), + Err(error) => { + debug!( + "Failed to parse local MCP server message ({}): {error}", + self.program_name + ); + None + } + } + } +} + +impl Transport for LocalStdioTransport { + type Error = io::Error; + + #[expect( + clippy::await_holding_invalid_type, + reason = "complete JSON-RPC frames must hold the stdin lock across writes" + )] + fn send( + &mut self, + item: TxJsonRpcMessage, + ) -> impl Future> + Send + 'static { + let stdin = Arc::clone(&self.stdin); + async move { + let mut message = serde_json::to_vec(&item).map_err(io::Error::other)?; + message.push(b'\n'); + let mut guard = stdin.lock().await; + let stdin = guard + .as_mut() + .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "MCP stdin closed"))?; + stdin.write_all(&message).await?; + stdin.flush().await + } + } + + fn receive(&mut self) -> impl Future>> + Send { + self.receive_message() + } + + async fn close(&mut self) -> Result<(), Self::Error> { + self.stdin.lock().await.take(); + match tokio::time::timeout(Duration::from_secs(3), self.child.wait()).await { + Ok(status) => { + status?; + Ok(()) + } + Err(_) => self.child.kill().await, + } + } +} diff --git a/codex-rs/rmcp-client/src/protocol_mode.rs b/codex-rs/rmcp-client/src/protocol_mode.rs new file mode 100644 index 0000000000..656ba8f6e9 --- /dev/null +++ b/codex-rs/rmcp-client/src/protocol_mode.rs @@ -0,0 +1,128 @@ +use std::ffi::OsStr; +use std::io; + +use rmcp::model::ProtocolVersion; +use rmcp::service::ClientLifecycleMode; + +/// MCP compatibility policy selected once when a Codex session is created. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum McpProtocolMode { + /// Preserve the existing MCP initialization and OAuth behavior. + #[default] + Legacy, + /// Allow the MCP 2026-07-28 discovery and request lifecycle. + V20260728, +} + +impl McpProtocolMode { + /// Returns the newest protocol version this compatibility policy can use. + pub fn preferred_protocol_version(self) -> ProtocolVersion { + match self { + Self::Legacy => ProtocolVersion::V_2025_06_18, + Self::V20260728 => ProtocolVersion::V_2026_07_28, + } + } + + pub(crate) fn client_lifecycle(self) -> ClientLifecycleMode { + match self { + Self::Legacy => ClientLifecycleMode::Initialize, + Self::V20260728 => ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_06_18), + }, + } + } + + pub(crate) fn stdio_mode(self, requested_version: Option<&OsStr>) -> io::Result { + match (self, requested_version) { + (Self::Legacy, _) => Ok(Self::Legacy), + (_, None) => Ok(Self::Legacy), + (Self::V20260728, Some(version)) if version == OsStr::new("2026-07-28") => { + Ok(Self::V20260728) + } + (_, Some(version)) => Err(io::Error::new( + io::ErrorKind::InvalidInput, + format!( + "unsupported CODEX_MCP_PROTOCOL_VERSION `{}` for stdio MCP server; expected `2026-07-28`", + version.to_string_lossy() + ), + )), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn protocol_modes_select_compatible_sdk_lifecycles() { + assert_eq!( + McpProtocolMode::Legacy.preferred_protocol_version(), + ProtocolVersion::V_2025_06_18 + ); + assert_eq!( + McpProtocolMode::Legacy.client_lifecycle(), + ClientLifecycleMode::Initialize + ); + assert_eq!( + McpProtocolMode::V20260728.preferred_protocol_version(), + ProtocolVersion::V_2026_07_28 + ); + assert_eq!( + McpProtocolMode::V20260728.client_lifecycle(), + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_06_18), + } + ); + } + + #[test] + fn stdio_requires_both_the_modern_feature_and_a_server_opt_in() { + let modern_version = Some(OsStr::new("2026-07-28")); + + assert_eq!( + McpProtocolMode::Legacy + .stdio_mode(/*requested_version*/ None) + .unwrap(), + McpProtocolMode::Legacy + ); + assert_eq!( + McpProtocolMode::Legacy.stdio_mode(modern_version).unwrap(), + McpProtocolMode::Legacy + ); + assert_eq!( + McpProtocolMode::V20260728 + .stdio_mode(/*requested_version*/ None) + .unwrap(), + McpProtocolMode::Legacy + ); + assert_eq!( + McpProtocolMode::V20260728 + .stdio_mode(modern_version) + .unwrap(), + McpProtocolMode::V20260728 + ); + } + + #[test] + fn stdio_rejects_unknown_protocol_markers() { + let error = McpProtocolMode::V20260728 + .stdio_mode(Some(OsStr::new("1999-01-01"))) + .expect_err("an unknown protocol marker must fail"); + + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + assert!(error.to_string().contains("1999-01-01")); + } + + #[test] + fn legacy_stdio_does_not_interpret_existing_protocol_markers() { + assert_eq!( + McpProtocolMode::Legacy + .stdio_mode(Some(OsStr::new("1999-01-01"))) + .unwrap(), + McpProtocolMode::Legacy + ); + } +} diff --git a/codex-rs/rmcp-client/src/rmcp_client.rs b/codex-rs/rmcp-client/src/rmcp_client.rs index a02d617531..bf13566ea2 100644 --- a/codex-rs/rmcp-client/src/rmcp_client.rs +++ b/codex-rs/rmcp-client/src/rmcp_client.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::ffi::OsStr; use std::ffi::OsString; use std::future::Future; use std::io; @@ -45,9 +46,9 @@ use rmcp::model::RequestParamsMeta; use rmcp::model::ServerResult; use rmcp::model::Tool; use rmcp::service::ClientCacheConfig; +use rmcp::service::ClientServiceExt; use rmcp::service::RoleClient; use rmcp::service::RunningService; -use rmcp::service::{self}; use rmcp::transport::AuthorizationManager; use rmcp::transport::StreamableHttpClientTransport; use rmcp::transport::auth::AuthClient; @@ -75,6 +76,7 @@ use crate::oauth::ResolvedOAuthTokens; use crate::oauth::StoredOAuthTokens; use crate::oauth::resolve_oauth_tokens_from_store_policy; use crate::oauth_http_client::OAuthHttpClientAdapter; +use crate::protocol_mode::McpProtocolMode; use crate::stdio_server_launcher::StdioServerCommand; use crate::stdio_server_launcher::StdioServerLauncher; use crate::stdio_server_launcher::StdioServerProcessHandle; @@ -328,12 +330,18 @@ pub struct RmcpClient { state: Mutex, stdio_process: Option, transport_recipe: TransportRecipe, + protocol_mode: McpProtocolMode, initialize_context: Mutex>, session_recovery_lock: Semaphore, elicitation_pause_state: ElicitationPauseState, } impl RmcpClient { + /// Returns the protocol compatibility policy captured when this client was created. + pub fn protocol_mode(&self) -> McpProtocolMode { + self.protocol_mode + } + pub async fn new_in_process_client( factory: Arc, ) -> io::Result { @@ -348,6 +356,7 @@ impl RmcpClient { }), stdio_process: None, transport_recipe, + protocol_mode: McpProtocolMode::Legacy, initialize_context: Mutex::new(None), session_recovery_lock: Semaphore::new(/*permits*/ 1), elicitation_pause_state: ElicitationPauseState::new(), @@ -362,8 +371,45 @@ impl RmcpClient { cwd: Option, launcher: Arc, ) -> io::Result { + Self::new_stdio_client_with_protocol_mode( + program, + args, + env, + env_vars, + cwd, + launcher, + McpProtocolMode::Legacy, + ) + .await + } + + /// Constructs a stdio client with an explicitly selected compatibility policy. + #[allow(clippy::too_many_arguments)] + pub async fn new_stdio_client_with_protocol_mode( + program: OsString, + args: Vec, + mut env: Option>, + env_vars: &[McpServerEnvVar], + cwd: Option, + launcher: Arc, + protocol_mode: McpProtocolMode, + ) -> io::Result { + let requested_stdio_version = match protocol_mode { + McpProtocolMode::Legacy => None, + McpProtocolMode::V20260728 => env + .as_mut() + .and_then(|env| env.remove(OsStr::new("CODEX_MCP_PROTOCOL_VERSION"))), + }; + let protocol_mode = protocol_mode.stdio_mode(requested_stdio_version.as_deref())?; let transport_recipe = TransportRecipe::Stdio { - command: StdioServerCommand::new(program, args, env, env_vars.to_vec(), cwd), + command: StdioServerCommand::new( + program, + args, + env, + env_vars.to_vec(), + cwd, + protocol_mode, + ), launcher, }; let transport = Self::create_pending_transport(&transport_recipe) @@ -382,6 +428,7 @@ impl RmcpClient { }), stdio_process, transport_recipe, + protocol_mode, initialize_context: Mutex::new(None), session_recovery_lock: Semaphore::new(/*permits*/ 1), elicitation_pause_state: ElicitationPauseState::new(), @@ -399,6 +446,35 @@ impl RmcpClient { keyring_backend_kind: AuthKeyringBackendKind, http_client: Arc, auth_provider: Option, + ) -> Result { + Self::new_streamable_http_client_with_protocol_mode( + server_name, + url, + bearer_token, + http_headers, + env_http_headers, + store_mode, + keyring_backend_kind, + http_client, + auth_provider, + McpProtocolMode::Legacy, + ) + .await + } + + /// Constructs a streamable HTTP client with an explicitly selected compatibility policy. + #[allow(clippy::too_many_arguments)] + pub async fn new_streamable_http_client_with_protocol_mode( + server_name: &str, + url: &str, + bearer_token: Option, + http_headers: Option>, + env_http_headers: Option>, + store_mode: OAuthCredentialsStoreMode, + keyring_backend_kind: AuthKeyringBackendKind, + http_client: Arc, + auth_provider: Option, + protocol_mode: McpProtocolMode, ) -> Result { let transport_recipe = TransportRecipe::StreamableHttp { server_name: server_name.to_string(), @@ -419,6 +495,7 @@ impl RmcpClient { }), stdio_process: None, transport_recipe, + protocol_mode, initialize_context: Mutex::new(None), session_recovery_lock: Semaphore::new(/*permits*/ 1), elicitation_pause_state: ElicitationPauseState::new(), @@ -933,6 +1010,7 @@ impl RmcpClient { } async fn connect_pending_transport( + &self, pending_transport: PendingTransport, client_service: ElicitationClientService, timeout: Option, @@ -940,24 +1018,33 @@ impl RmcpClient { Arc>, Option, )> { + let lifecycle = self.protocol_mode.client_lifecycle(); let (transport, oauth_persistor) = match pending_transport { PendingTransport::InProcess { transport } => ( - service::serve_client(client_service, transport).boxed(), + client_service + .serve_with_lifecycle(transport, lifecycle) + .boxed(), None, ), PendingTransport::Stdio { transport } => ( - service::serve_client(client_service, *transport).boxed(), + client_service + .serve_with_lifecycle(*transport, lifecycle) + .boxed(), None, ), PendingTransport::StreamableHttp { transport } => ( - service::serve_client(client_service, transport).boxed(), + client_service + .serve_with_lifecycle(transport, lifecycle) + .boxed(), None, ), PendingTransport::StreamableHttpWithOAuth { transport, oauth_persistor, } => ( - service::serve_client(client_service, transport).boxed(), + client_service + .serve_with_lifecycle(transport, lifecycle) + .boxed(), Some(oauth_persistor), ), }; diff --git a/codex-rs/rmcp-client/src/stdio_server_launcher.rs b/codex-rs/rmcp-client/src/stdio_server_launcher.rs index 4a2c52a40f..9c84105867 100644 --- a/codex-rs/rmcp-client/src/stdio_server_launcher.rs +++ b/codex-rs/rmcp-client/src/stdio_server_launcher.rs @@ -56,7 +56,9 @@ use tracing::info; use tracing::warn; use crate::executor_process_transport::ExecutorProcessTransport; +use crate::local_stdio_transport::LocalStdioTransport; use crate::program_resolver; +use crate::protocol_mode::McpProtocolMode; use crate::utils::create_env_for_mcp_server; use crate::utils::create_env_overlay_for_remote_mcp_server; use crate::utils::remote_mcp_env_var_names; @@ -85,6 +87,7 @@ pub struct StdioServerCommand { env: Option>, env_vars: Vec, cwd: Option, + protocol_mode: McpProtocolMode, } /// Client-side rmcp transport for a launched MCP stdio server. @@ -98,7 +101,8 @@ pub struct StdioServerTransport { } enum StdioServerTransportInner { - Local(TokioChildProcess), + LocalLegacy(TokioChildProcess), + LocalModern(LocalStdioTransport), Executor(ExecutorProcessTransport), } @@ -113,7 +117,8 @@ impl Transport for StdioServerTransport { // wrapper keeps process placement private while leaving rmcp's send // semantics unchanged. match &mut self.inner { - StdioServerTransportInner::Local(transport) => transport.send(item).boxed(), + StdioServerTransportInner::LocalLegacy(transport) => transport.send(item).boxed(), + StdioServerTransportInner::LocalModern(transport) => transport.send(item).boxed(), StdioServerTransportInner::Executor(transport) => transport.send(item).boxed(), } } @@ -123,7 +128,8 @@ impl Transport for StdioServerTransport { // executor variant turns pushed process-output events back into the // line-delimited JSON stream expected by rmcp. match &mut self.inner { - StdioServerTransportInner::Local(transport) => transport.receive().boxed(), + StdioServerTransportInner::LocalLegacy(transport) => transport.receive().boxed(), + StdioServerTransportInner::LocalModern(transport) => transport.receive().boxed(), StdioServerTransportInner::Executor(transport) => transport.receive().boxed(), } } @@ -131,7 +137,8 @@ impl Transport for StdioServerTransport { async fn close(&mut self) -> std::result::Result<(), Self::Error> { self.process.terminate().await?; match &mut self.inner { - StdioServerTransportInner::Local(transport) => transport.close().await, + StdioServerTransportInner::LocalLegacy(transport) => transport.close().await, + StdioServerTransportInner::LocalModern(transport) => transport.close().await, StdioServerTransportInner::Executor(transport) => transport.close().await, } } @@ -152,6 +159,7 @@ impl StdioServerCommand { env: Option>, env_vars: Vec, cwd: Option, + protocol_mode: McpProtocolMode, ) -> Self { Self { program, @@ -159,6 +167,7 @@ impl StdioServerCommand { env, env_vars, cwd, + protocol_mode, } } } @@ -253,6 +262,7 @@ impl LocalStdioServerLauncher { env, env_vars, cwd, + protocol_mode, } = command; let program_name = program.to_string_lossy().into_owned(); let envs = create_env_for_mcp_server(env, &env_vars).map_err(io::Error::other)?; @@ -272,12 +282,32 @@ impl LocalStdioServerLauncher { #[cfg(unix)] command.process_group(0); - let (transport, stderr) = TokioChildProcess::builder(command) - .stderr(Stdio::piped()) - .spawn()?; + let (transport, stderr, process_id) = match protocol_mode { + McpProtocolMode::Legacy => { + let (transport, stderr) = TokioChildProcess::builder(command) + .stderr(Stdio::piped()) + .spawn()?; + let process_id = transport.id(); + ( + StdioServerTransportInner::LocalLegacy(transport), + stderr, + process_id, + ) + } + McpProtocolMode::V20260728 => { + let (transport, stderr) = + LocalStdioTransport::spawn(command, program_name.clone())?; + let process_id = transport.id(); + ( + StdioServerTransportInner::LocalModern(transport), + stderr, + process_id, + ) + } + }; let process = StdioServerProcessHandle::local( program_name.clone(), - transport.id().map(LocalProcessTerminator::new), + process_id.map(LocalProcessTerminator::new), ); if let Some(stderr) = stderr { @@ -299,7 +329,7 @@ impl LocalStdioServerLauncher { } Ok(StdioServerTransport { - inner: StdioServerTransportInner::Local(transport), + inner: transport, process, }) } @@ -482,6 +512,7 @@ impl ExecutorStdioServerLauncher { env, env_vars, cwd, + protocol_mode: _, } = command; let Some(cwd) = cwd else { return Err(io::Error::other( diff --git a/codex-rs/rmcp-client/src/streamable_http_retry.rs b/codex-rs/rmcp-client/src/streamable_http_retry.rs index 5c4d288815..aff433d3d4 100644 --- a/codex-rs/rmcp-client/src/streamable_http_retry.rs +++ b/codex-rs/rmcp-client/src/streamable_http_retry.rs @@ -76,12 +76,9 @@ impl RmcpClient { } let attempt_timeout = remaining_initialize_timeout(timeout, retry_deadline)?; - match Self::connect_pending_transport( - transport, - client_service.clone(), - attempt_timeout, - ) - .await + match self + .connect_pending_transport(transport, client_service.clone(), attempt_timeout) + .await { Ok(result) => return Ok(result), Err(error) if should_retry && Self::is_retryable_initialize_error(&error) => { @@ -124,7 +121,10 @@ impl RmcpClient { fn is_retryable_client_initialize_error(error: &rmcp::service::ClientInitializeError) -> bool { match error { rmcp::service::ClientInitializeError::TransportError { error, context } - if context.as_ref() == "send initialize request" => + if matches!( + context.as_ref(), + "send initialize request" | "send discover request" + ) => { error .error diff --git a/codex-rs/rmcp-client/src/streamable_http_retry_tests.rs b/codex-rs/rmcp-client/src/streamable_http_retry_tests.rs index ed0d8030e6..69b97e1a82 100644 --- a/codex-rs/rmcp-client/src/streamable_http_retry_tests.rs +++ b/codex-rs/rmcp-client/src/streamable_http_retry_tests.rs @@ -10,8 +10,9 @@ use crate::http_client_adapter::StreamableHttpClientAdapterError; use super::*; #[test] -fn retryable_initialize_error_includes_initialized_notification_context() { +fn retryable_initialize_error_includes_discovery_and_initialized_notification_context() { let contexts = [ + "send discover request", "send initialize request", "send initialized notification", "receive initialize response", @@ -21,7 +22,7 @@ fn retryable_initialize_error_includes_initialized_notification_context() { contexts.map(|context| { RmcpClient::is_retryable_client_initialize_error(&retryable_initialize_error(context)) }), - [true, true, false], + [true, true, true, false], ); } diff --git a/codex-rs/rmcp-client/tests/mcp_2026_discovery.rs b/codex-rs/rmcp-client/tests/mcp_2026_discovery.rs new file mode 100644 index 0000000000..de6e63fdfe --- /dev/null +++ b/codex-rs/rmcp-client/tests/mcp_2026_discovery.rs @@ -0,0 +1,1073 @@ +use std::collections::HashMap; +use std::ffi::OsString; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use codex_config::types::AuthKeyringBackendKind; +use codex_config::types::OAuthCredentialsStoreMode; +use codex_exec_server::Environment; +use codex_rmcp_client::ElicitationAction; +use codex_rmcp_client::ElicitationResponse; +use codex_rmcp_client::LocalStdioServerLauncher; +use codex_rmcp_client::McpProtocolMode; +use codex_rmcp_client::RmcpClient; +use futures::FutureExt; +use pretty_assertions::assert_eq; +use rmcp::model::ClientCapabilities; +use rmcp::model::Implementation; +use rmcp::model::InitializeRequestParams; +use rmcp::model::ProtocolVersion; +use serde_json::Value; +use serde_json::json; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::Request; +use wiremock::ResponseTemplate; +use wiremock::matchers::header; +use wiremock::matchers::method; +use wiremock::matchers::path; + +const MODERN_VERSION: &str = "2026-07-28"; +const LEGACY_VERSION: &str = "2025-06-18"; +const MAX_MCP_MESSAGE_BYTES: usize = 8 * 1024 * 1024; + +fn initialize_params() -> InitializeRequestParams { + InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("codex-discovery-test", "0.0.0"), + ) + .with_protocol_version(ProtocolVersion::V_2025_06_18) +} + +async fn create_client(server: &MockServer, mode: McpProtocolMode) -> anyhow::Result { + RmcpClient::new_streamable_http_client_with_protocol_mode( + "discovery-test", + &format!("{}/mcp", server.uri()), + /*bearer_token*/ None, + /*http_headers*/ None, + /*env_http_headers*/ None, + OAuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + Environment::default_for_tests().get_http_client(), + /*auth_provider*/ None, + mode, + ) + .await +} + +async fn initialize_client(client: &RmcpClient) -> anyhow::Result<()> { + client + .initialize( + initialize_params(), + Some(Duration::from_secs(5)), + Box::new(|_, _| { + async { + Ok(ElicitationResponse { + action: ElicitationAction::Accept, + content: Some(json!({})), + meta: None, + }) + } + .boxed() + }), + ) + .await?; + Ok(()) +} + +fn modern_discover_response(request: &Value) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_json(modern_discover_result(request)) +} + +fn modern_discover_result(request: &Value) -> Value { + json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { + "resultType": "complete", + "supportedVersions": [MODERN_VERSION], + "capabilities": {"tools": {}}, + "serverInfo": {"name": "modern-test", "version": "1.0.0"}, + "ttlMs": 0, + "cacheScope": "private", + }, + }) +} + +fn legacy_initialize_response(request: &Value) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { + "protocolVersion": LEGACY_VERSION, + "capabilities": {"tools": {}}, + "serverInfo": {"name": "legacy-test", "version": "1.0.0"}, + }, + })) +} + +#[tokio::test] +async fn modern_mode_uses_sdk_discovery_and_self_contained_request_metadata() -> anyhow::Result<()> +{ + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + recorded.lock().expect("requests lock").push(body.clone()); + match body["method"].as_str() { + Some("server/discover") => { + assert_eq!( + body.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion"), + Some(&json!(MODERN_VERSION)) + ); + assert_eq!( + body.pointer("/params/_meta/io.modelcontextprotocol~1clientInfo/name"), + Some(&json!("codex-discovery-test")) + ); + modern_discover_response(&body) + } + Some("tools/list") => { + assert_eq!( + body.pointer("/params/_meta/io.modelcontextprotocol~1protocolVersion"), + Some(&json!(MODERN_VERSION)) + ); + assert_eq!( + body.pointer("/params/_meta/io.modelcontextprotocol~1clientInfo/name"), + Some(&json!("codex-discovery-test")) + ); + ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": {"resultType": "complete", "tools": []}, + })) + } + other => panic!("unexpected modern lifecycle request: {other:?}"), + } + }) + .expect(2) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + let tools = client + .list_tools(/*params*/ None, Some(Duration::from_secs(5))) + .await?; + assert!(tools.tools.is_empty()); + + let methods = observed + .lock() + .expect("requests lock") + .iter() + .map(|request| request["method"].as_str().unwrap_or_default().to_owned()) + .collect::>(); + assert_eq!(methods, vec!["server/discover", "tools/list"]); + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_does_not_follow_redirects_with_sensitive_headers() -> anyhow::Result<()> { + let redirect_target = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/forwarded")) + .and(header("x-api-key", "sensitive-key")) + .respond_with(ResponseTemplate::new(200)) + .expect(0) + .mount(&redirect_target) + .await; + + let resource_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/mcp")) + .and(header("x-api-key", "sensitive-key")) + .respond_with( + ResponseTemplate::new(307) + .insert_header("location", format!("{}/forwarded", redirect_target.uri())), + ) + .expect(1) + .mount(&resource_server) + .await; + + let client = RmcpClient::new_streamable_http_client_with_protocol_mode( + "discovery-redirect-test", + &format!("{}/mcp", resource_server.uri()), + /*bearer_token*/ None, + Some(HashMap::from([( + "x-api-key".to_string(), + "sensitive-key".to_string(), + )])), + /*env_http_headers*/ None, + OAuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + Environment::default_for_tests().get_http_client(), + /*auth_provider*/ None, + McpProtocolMode::V20260728, + ) + .await?; + let error = initialize_client(&client) + .await + .expect_err("modern MCP discovery must not follow redirects"); + assert!( + error.to_string().contains("307"), + "redirect rejection should report its HTTP status: {error:#}" + ); + redirect_target.verify().await; + resource_server.verify().await; + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn legacy_mcp_requests_preserve_existing_redirect_behavior() -> anyhow::Result<()> { + let redirect_target = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/forwarded")) + .and(header("x-api-key", "sensitive-key")) + .respond_with(|request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + match body["method"].as_str() { + Some("initialize") => legacy_initialize_response(&body), + Some("notifications/initialized") => ResponseTemplate::new(202), + other => panic!("unexpected redirected legacy method: {other:?}"), + } + }) + .expect(2) + .mount(&redirect_target) + .await; + + let resource_server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with( + ResponseTemplate::new(307) + .insert_header("location", format!("{}/forwarded", redirect_target.uri())), + ) + .expect(2) + .mount(&resource_server) + .await; + + let client = RmcpClient::new_streamable_http_client_with_protocol_mode( + "legacy-redirect-test", + &format!("{}/mcp", resource_server.uri()), + /*bearer_token*/ None, + Some(HashMap::from([( + "x-api-key".to_string(), + "sensitive-key".to_string(), + )])), + /*env_http_headers*/ None, + OAuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + Environment::default_for_tests().get_http_client(), + /*auth_provider*/ None, + McpProtocolMode::Legacy, + ) + .await?; + initialize_client(&client).await?; + redirect_target.verify().await; + resource_server.verify().await; + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_accepts_metadata_namespaced_server_identity() -> anyhow::Result<()> { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(|request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + let mut response = modern_discover_result(&body); + let server_info = response["result"] + .as_object_mut() + .expect("discovery object") + .remove("serverInfo") + .expect("top-level identity"); + response["result"]["_meta"] = json!({ + "io.modelcontextprotocol/serverInfo": server_info, + }); + ResponseTemplate::new(200).set_body_json(response) + }) + .expect(1) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_http_rejects_oversized_discovery_and_tool_json_responses() -> anyhow::Result<()> { + for oversized_method in ["server/discover", "tools/list", "tools/list:error"] { + let server = MockServer::start().await; + let padding = Arc::new("x".repeat(MAX_MCP_MESSAGE_BYTES + 1)); + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + match body["method"].as_str() { + Some("server/discover") => { + let mut response = modern_discover_result(&body); + if oversized_method == "server/discover" { + response["result"]["instructions"] = json!(padding.as_str()); + } + ResponseTemplate::new(200).set_body_json(response) + } + Some("tools/list") if oversized_method == "tools/list:error" => { + ResponseTemplate::new(500).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": {"code": -32603, "message": padding.as_str()}, + })) + } + Some("tools/list") => ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "resultType": "complete", + "tools": [{ + "name": "oversized_tool", + "description": padding.as_str(), + "inputSchema": {"type": "object"}, + }], + }, + })), + other => panic!("unexpected MCP method {other:?}"), + } + }) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + let error = if oversized_method == "server/discover" { + initialize_client(&client) + .await + .expect_err("oversized discovery response must be rejected") + } else { + initialize_client(&client).await?; + client + .list_tools(/*params*/ None, Some(Duration::from_secs(5))) + .await + .expect_err("oversized tools response must be rejected") + }; + assert!( + format!("{error:#}").contains("8388608 bytes"), + "expected bounded {oversized_method} response, got {error:#}" + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test] +async fn legacy_http_keeps_existing_large_json_response_behavior() -> anyhow::Result<()> { + let server = MockServer::start().await; + let padding = Arc::new("x".repeat(MAX_MCP_MESSAGE_BYTES + 1)); + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + match body["method"].as_str() { + Some("initialize") => legacy_initialize_response(&body), + Some("notifications/initialized") => ResponseTemplate::new(202), + Some("tools/list") => ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "tools": [{ + "name": "legacy_large_tool", + "description": padding.as_str(), + "inputSchema": {"type": "object"}, + }], + }, + })), + other => panic!("unexpected legacy MCP method {other:?}"), + } + }) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::Legacy).await?; + initialize_client(&client).await?; + let tools = client + .list_tools(/*params*/ None, Some(Duration::from_secs(15))) + .await?; + assert_eq!( + tools.tools[0].description.as_deref().map(str::len), + Some(MAX_MCP_MESSAGE_BYTES + 1) + ); + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_retries_transient_service_unavailable() -> anyhow::Result<()> { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + let attempt = { + let mut requests = recorded.lock().expect("requests lock"); + requests.push(body.clone()); + requests.len() + }; + if attempt == 1 { + ResponseTemplate::new(503).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": { + "code": -32000, + "message": "service unavailable", + }, + })) + } else { + modern_discover_response(&body) + } + }) + .expect(2) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + { + let requests = observed.lock().expect("requests lock"); + assert_eq!( + requests + .iter() + .map(|request| request["method"].as_str()) + .collect::>(), + vec![Some("server/discover"), Some("server/discover")], + "transient discovery failures must retry without falling back to legacy initialization" + ); + } + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_retries_json_unsupported_protocol_errors() -> anyhow::Result<()> { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + let attempt = { + let mut requests = recorded.lock().expect("requests lock"); + requests.push(body.clone()); + requests.len() + }; + if attempt == 1 { + ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": { + "code": -32022, + "message": "unsupported protocol version", + "data": {"supported": [LEGACY_VERSION, MODERN_VERSION]}, + }, + })) + } else { + modern_discover_response(&body) + } + }) + .expect(2) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + { + let requests = observed.lock().expect("requests lock"); + assert_eq!(requests.len(), 2); + assert_ne!( + requests[0]["id"], requests[1]["id"], + "each SDK discovery attempt must use a distinct JSON-RPC request ID" + ); + } + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_sdk_emits_standard_headers_and_rejects_invalid_header_schemas() -> anyhow::Result<()> +{ + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + assert_eq!( + request + .headers + .get("mcp-protocol-version") + .and_then(|value| value.to_str().ok()), + Some(MODERN_VERSION) + ); + assert_eq!( + request + .headers + .get("mcp-method") + .and_then(|value| value.to_str().ok()), + Some(method) + ); + + match method { + "server/discover" => modern_discover_response(&body), + "tools/list" => ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "resultType": "complete", + "tools": [ + { + "name": "echo", + "inputSchema": { + "type": "object", + "properties": { + "tenant": {"type": "string", "x-mcp-header": "Tenant"}, + }, + }, + }, + { + "name": "invalid-header-schema", + "inputSchema": { + "type": "object", + "properties": { + "tenant": {"type": "string", "x-mcp-header": true}, + }, + }, + }, + ], + }, + })), + "tools/call" => { + assert_eq!( + request + .headers + .get("mcp-name") + .and_then(|value| value.to_str().ok()), + Some("echo") + ); + assert_eq!( + request + .headers + .get("mcp-param-tenant") + .and_then(|value| value.to_str().ok()), + Some("acme") + ); + ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "resultType": "complete", + "content": [{"type": "text", "text": "headers work"}], + }, + })) + } + other => panic!("unexpected modern request: {other}"), + } + }) + .expect(3) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + let tools = client + .list_tools(/*params*/ None, Some(Duration::from_secs(5))) + .await?; + assert_eq!(tools.tools.len(), 1); + assert_eq!(tools.tools[0].name.as_ref(), "echo"); + + let result = client + .call_tool( + "echo".to_string(), + Some(json!({"tenant": "acme"})), + /*meta*/ None, + Some(Duration::from_secs(5)), + ) + .await?; + assert_eq!( + result.content[0].as_text().map(|text| text.text.as_str()), + Some("headers work") + ); + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn legacy_mode_never_probes_server_discovery() -> anyhow::Result<()> { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + recorded.lock().expect("requests lock").push(method.into()); + match method { + "initialize" => { + assert_eq!(body["params"]["protocolVersion"], LEGACY_VERSION); + legacy_initialize_response(&body) + } + "notifications/initialized" => ResponseTemplate::new(202), + other => panic!("legacy mode must not use modern lifecycle: {other}"), + } + }) + .expect(2) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::Legacy).await?; + initialize_client(&client).await?; + assert_eq!( + *observed.lock().expect("requests lock"), + vec!["initialize", "notifications/initialized"] + ); + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_mode_falls_back_only_for_a_json_rpc_method_not_found() -> anyhow::Result<()> { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + recorded.lock().expect("requests lock").push(method.into()); + match method { + "server/discover" => ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": {"code": -32601, "message": "method not found"}, + })), + "initialize" => { + assert_eq!(body["params"]["protocolVersion"], LEGACY_VERSION); + legacy_initialize_response(&body) + } + "notifications/initialized" => ResponseTemplate::new(202), + other => panic!("unexpected legacy fallback request: {other}"), + } + }) + .expect(3) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + assert_eq!( + *observed.lock().expect("requests lock"), + vec!["server/discover", "initialize", "notifications/initialized"] + ); + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_mode_falls_back_when_discovery_only_advertises_legacy_versions() +-> anyhow::Result<()> { + for legacy_response in ["discover-result", "unsupported-protocol"] { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + recorded.lock().expect("requests lock").push(method.into()); + match method { + "server/discover" if legacy_response == "discover-result" => { + let mut response = modern_discover_result(&body); + response["result"]["supportedVersions"] = json!(["2025-11-25"]); + ResponseTemplate::new(200).set_body_json(response) + } + "server/discover" => ResponseTemplate::new(400).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": { + "code": -32022, + "message": "only legacy MCP protocols are supported", + "data": {"supported": [LEGACY_VERSION, "2025-11-25"]}, + }, + })), + "initialize" => { + assert_eq!(body["params"]["protocolVersion"], LEGACY_VERSION); + legacy_initialize_response(&body) + } + "notifications/initialized" => ResponseTemplate::new(202), + other => { + panic!("unexpected {legacy_response} fallback request: {other}") + } + } + }) + .expect(3) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + assert_eq!( + *observed.lock().expect("requests lock"), + vec!["server/discover", "initialize", "notifications/initialized"], + "{legacy_response} should prove the endpoint only supports legacy protocols" + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_never_downgrades_without_valid_legacy_only_version_evidence() +-> anyhow::Result<()> { + for (case, supported) in [ + ("missing-versions", None), + ("empty-versions", Some(json!([]))), + ( + "mixed-unknown-version", + Some(json!([LEGACY_VERSION, "1999-01-01"])), + ), + ("mixed-invalid-version", Some(json!([LEGACY_VERSION, 42]))), + ] { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + let mut error = json!({ + "code": -32022, + "message": "unsupported protocol version", + "data": {}, + }); + if let Some(supported) = &supported { + error["data"]["supported"] = supported.clone(); + } + ResponseTemplate::new(400).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": error, + })) + }) + .expect(1) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + assert!( + initialize_client(&client).await.is_err(), + "{case} must not silently downgrade to legacy" + ); + } + Ok(()) +} + +#[tokio::test] +async fn modern_mode_falls_back_when_legacy_server_explicitly_rejects_modern_version() +-> anyhow::Result<()> { + for error_code in [-32022, -32600, -32602] { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + recorded.lock().expect("requests lock").push(method.into()); + match method { + "server/discover" => ResponseTemplate::new(400).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": { + "code": error_code, + "message": "Unsupported protocol version: 2026-07-28", + }, + })), + "initialize" => legacy_initialize_response(&body), + "notifications/initialized" => ResponseTemplate::new(202), + other => panic!("unexpected legacy fallback method: {other}"), + } + }) + .expect(3) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + assert_eq!( + *observed.lock().expect("requests lock"), + vec!["server/discover", "initialize", "notifications/initialized"] + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_rejects_uncorrelated_legacy_version_rejection() -> anyhow::Result<()> { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(|request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + ResponseTemplate::new(400).set_body_json(json!({ + "jsonrpc": "2.0", + "id": "unrelated-request", + "error": { + "code": -32602, + "message": "Unsupported protocol version: 2026-07-28", + }, + })) + }) + .expect(1) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + assert!( + initialize_client(&client).await.is_err(), + "an unrelated response ID must never authorize protocol downgrade" + ); + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_rejects_uncorrelated_method_not_found_errors() -> anyhow::Result<()> { + for response_id in [Value::Null, json!("unrelated-request")] { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": response_id, + "error": {"code": -32601, "message": "method not found"}, + })) + }) + .expect(1) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + let error = initialize_client(&client) + .await + .expect_err("an unrelated error must not authorize protocol downgrade"); + assert!( + error.to_string().contains("did not match its request ID"), + "uncorrelated method-not-found error should explain its rejection: {error:#}" + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test] +async fn modern_mode_falls_back_for_real_legacy_http_prevalidation_errors() -> anyhow::Result<()> { + for message in [ + "Bad Request: Unsupported protocol version: 2026-07-28 (supported versions: 2025-11-25, 2025-06-18, 2025-03-26, 2024-11-05, 2024-10-07)", + "Bad Request: No valid session ID provided", + ] { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + recorded.lock().expect("requests lock").push(method.into()); + match method { + "server/discover" => ResponseTemplate::new(400).set_body_json(json!({ + "jsonrpc": "2.0", + "id": null, + "error": {"code": -32000, "message": message}, + })), + "initialize" => legacy_initialize_response(&body), + "notifications/initialized" => ResponseTemplate::new(202), + other => panic!("unexpected legacy fallback method: {other}"), + } + }) + .expect(3) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + assert_eq!( + *observed.lock().expect("requests lock"), + vec!["server/discover", "initialize", "notifications/initialized"] + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_rejects_unrecognized_or_modern_null_id_errors() -> anyhow::Result<()> { + for (status, message) in [ + (400, "Bad Request: malformed request"), + ( + 400, + "Bad Request: Unsupported protocol version: 2026-07-28 (supported versions: 2025-06-18, 2099-01-01)", + ), + ( + 400, + "Bad Request: Unsupported protocol version: 2026-07-28 (supported versions: 2025-11-25)", + ), + (200, "Bad Request: No valid session ID provided"), + ] { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + ResponseTemplate::new(status).set_body_json(json!({ + "jsonrpc": "2.0", + "id": null, + "error": {"code": -32000, "message": message}, + })) + }) + .expect(1) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + assert!( + initialize_client(&client).await.is_err(), + "unproven legacy downgrade must fail (HTTP {status}, {message})" + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test] +async fn modern_mode_falls_back_for_plain_404_and_405_discovery_responses() -> anyhow::Result<()> { + for status in [404, 405] { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + recorded.lock().expect("requests lock").push(method.into()); + match method { + "server/discover" => { + ResponseTemplate::new(status).set_body_string("legacy MCP endpoint") + } + "initialize" => legacy_initialize_response(&body), + "notifications/initialized" => ResponseTemplate::new(202), + other => panic!("unexpected HTTP {status} fallback request: {other}"), + } + }) + .expect(3) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + initialize_client(&client).await?; + assert_eq!( + *observed.lock().expect("requests lock"), + vec!["server/discover", "initialize", "notifications/initialized"], + "HTTP {status} should prove the endpoint is legacy" + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test] +async fn modern_discovery_rejection_does_not_downgrade_to_legacy() -> anyhow::Result<()> { + for (status, error_code) in [(400, -32602), (403, -32021), (200, -32020)] { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + ResponseTemplate::new(status).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": {"code": error_code, "message": "modern discovery rejected"}, + })) + }) + .expect(1) + .mount(&server) + .await; + + let client = create_client(&server, McpProtocolMode::V20260728).await?; + let error = initialize_client(&client) + .await + .expect_err("modern discovery errors must not silently downgrade"); + assert!( + error.to_string().contains("modern discovery rejected"), + "unexpected discovery rejection for HTTP {status}: {error:#}" + ); + } + Ok(()) +} + +#[tokio::test] +async fn stdio_protocol_marker_rejects_unknown_versions_before_launch() -> anyhow::Result<()> { + let env = HashMap::from([( + OsString::from("CODEX_MCP_PROTOCOL_VERSION"), + OsString::from("1999-01-01"), + )]); + let error = RmcpClient::new_stdio_client_with_protocol_mode( + OsString::from("never-launch"), + Vec::new(), + Some(env), + &[], + /*cwd*/ None, + Arc::new(LocalStdioServerLauncher::new(std::env::current_dir()?)), + McpProtocolMode::V20260728, + ) + .await + .err() + .expect("an unsupported stdio protocol must fail before launch"); + + assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput); + assert!(error.to_string().contains("1999-01-01")); + Ok(()) +} diff --git a/codex-rs/rmcp-client/tests/mcp_2026_sse_discovery.rs b/codex-rs/rmcp-client/tests/mcp_2026_sse_discovery.rs new file mode 100644 index 0000000000..92910a34a2 --- /dev/null +++ b/codex-rs/rmcp-client/tests/mcp_2026_sse_discovery.rs @@ -0,0 +1,203 @@ +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use codex_config::types::AuthKeyringBackendKind; +use codex_config::types::OAuthCredentialsStoreMode; +use codex_exec_server::Environment; +use codex_rmcp_client::ElicitationAction; +use codex_rmcp_client::ElicitationResponse; +use codex_rmcp_client::McpProtocolMode; +use codex_rmcp_client::RmcpClient; +use futures::FutureExt; +use pretty_assertions::assert_eq; +use rmcp::model::ClientCapabilities; +use rmcp::model::Implementation; +use rmcp::model::InitializeRequestParams; +use rmcp::model::ProtocolVersion; +use serde_json::Value; +use serde_json::json; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::Request; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; + +const MODERN_VERSION: &str = "2026-07-28"; +const LEGACY_VERSION: &str = "2025-06-18"; + +async fn initialize_modern_client(server: &MockServer) -> anyhow::Result { + let client = RmcpClient::new_streamable_http_client_with_protocol_mode( + "sse-discovery-test", + &format!("{}/mcp", server.uri()), + /*bearer_token*/ None, + /*http_headers*/ None, + /*env_http_headers*/ None, + OAuthCredentialsStoreMode::File, + AuthKeyringBackendKind::default(), + Environment::default_for_tests().get_http_client(), + /*auth_provider*/ None, + McpProtocolMode::V20260728, + ) + .await?; + + let params = InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("codex-sse-discovery-test", "0.0.0"), + ) + .with_protocol_version(ProtocolVersion::V_2025_06_18); + + client + .initialize( + params, + Some(Duration::from_secs(5)), + Box::new(|_, _| { + async { + Ok(ElicitationResponse { + action: ElicitationAction::Accept, + content: Some(json!({})), + meta: None, + }) + } + .boxed() + }), + ) + .await?; + + Ok(client) +} + +fn sse_response(message: Value) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_raw( + format!("event: message\ndata: {message}\n\n"), + "text/event-stream", + ) +} + +#[tokio::test] +async fn modern_sse_discovery_accepts_metadata_namespaced_server_identity() -> anyhow::Result<()> { + let server = MockServer::start().await; + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(|request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + assert_eq!(body["method"], "server/discover"); + + sse_response(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "resultType": "complete", + "supportedVersions": [MODERN_VERSION], + "capabilities": {"tools": {}}, + "_meta": { + "io.modelcontextprotocol/serverInfo": { + "name": "modern-sse-test", + "version": "1.0.0", + }, + }, + "ttlMs": 0, + "cacheScope": "private", + }, + })) + }) + .expect(1) + .mount(&server) + .await; + + let client = initialize_modern_client(&server).await?; + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_sse_discovery_falls_back_for_correlated_method_not_found() -> anyhow::Result<()> { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + recorded.lock().expect("requests lock").push(method.into()); + + match method { + "server/discover" => sse_response(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "error": {"code": -32601, "message": "method not found"}, + })), + "initialize" => { + assert_eq!(body["params"]["protocolVersion"], LEGACY_VERSION); + ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", + "id": body["id"], + "result": { + "protocolVersion": LEGACY_VERSION, + "capabilities": {"tools": {}}, + "serverInfo": { + "name": "legacy-sse-test", + "version": "1.0.0", + }, + }, + })) + } + "notifications/initialized" => ResponseTemplate::new(202), + other => panic!("unexpected legacy SSE fallback request: {other}"), + } + }) + .expect(3) + .mount(&server) + .await; + + let client = initialize_modern_client(&server).await?; + assert_eq!( + *observed.lock().expect("requests lock"), + vec!["server/discover", "initialize", "notifications/initialized"] + ); + client.shutdown().await; + Ok(()) +} + +#[tokio::test] +async fn modern_sse_discovery_rejects_uncorrelated_method_not_found() -> anyhow::Result<()> { + for (case, rejected_id) in [("null", json!(null)), ("mismatched", json!("unrelated"))] { + let server = MockServer::start().await; + let observed = Arc::new(Mutex::new(Vec::::new())); + let recorded = Arc::clone(&observed); + + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("valid JSON-RPC request"); + let method = body["method"].as_str().expect("JSON-RPC method"); + recorded.lock().expect("requests lock").push(method.into()); + assert_eq!(method, "server/discover"); + + sse_response(json!({ + "jsonrpc": "2.0", + "id": rejected_id.clone(), + "error": {"code": -32601, "message": "method not found"}, + })) + }) + .expect(1) + .mount(&server) + .await; + + assert!( + initialize_modern_client(&server).await.is_err(), + "{case} SSE discovery error must not downgrade to legacy" + ); + assert_eq!( + *observed.lock().expect("requests lock"), + vec!["server/discover"], + "{case} SSE discovery error must not initialize a legacy session" + ); + } + + Ok(()) +} diff --git a/codex-rs/rmcp-client/tests/mcp_2026_stdio_discovery.rs b/codex-rs/rmcp-client/tests/mcp_2026_stdio_discovery.rs new file mode 100644 index 0000000000..b2ff92ef6a --- /dev/null +++ b/codex-rs/rmcp-client/tests/mcp_2026_stdio_discovery.rs @@ -0,0 +1,159 @@ +use std::collections::HashMap; +use std::ffi::OsString; +use std::sync::Arc; +use std::time::Duration; + +use codex_exec_server::Environment; +use codex_rmcp_client::ElicitationAction; +use codex_rmcp_client::ElicitationResponse; +use codex_rmcp_client::ExecutorStdioServerLauncher; +use codex_rmcp_client::LocalStdioServerLauncher; +use codex_rmcp_client::McpProtocolMode; +use codex_rmcp_client::RmcpClient; +use codex_rmcp_client::StdioServerLauncher; +use futures::FutureExt; +use rmcp::model::ClientCapabilities; +use rmcp::model::Implementation; +use rmcp::model::InitializeRequestParams; +use rmcp::model::ProtocolVersion; +use serde_json::json; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn modern_local_and_executor_stdio_discover_metadata_identity_and_catalogs() +-> anyhow::Result<()> { + let server = codex_utils_cargo_bin::cargo_bin("test_mcp_2026_discovery_stdio_server")?; + + for executor in [false, true] { + let launcher: Arc = if executor { + Arc::new(ExecutorStdioServerLauncher::new( + Environment::default_for_tests().get_exec_backend(), + )) + } else { + Arc::new(LocalStdioServerLauncher::new(std::env::current_dir()?)) + }; + let client = RmcpClient::new_stdio_client_with_protocol_mode( + server.clone().into(), + Vec::new(), + Some(HashMap::from([( + OsString::from("CODEX_MCP_PROTOCOL_VERSION"), + OsString::from("2026-07-28"), + )])), + &[], + Some(std::env::current_dir()?.to_string_lossy().into_owned()), + launcher, + McpProtocolMode::V20260728, + ) + .await?; + client + .initialize( + InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("stdio-discovery-test", "1.0.0"), + ) + .with_protocol_version(ProtocolVersion::V_2025_06_18), + Some(Duration::from_secs(10)), + Box::new(|_, _| { + async { + Ok(ElicitationResponse { + action: ElicitationAction::Decline, + content: None, + meta: None, + }) + } + .boxed() + }), + ) + .await?; + + let tools = client + .list_tools(/*params*/ None, Some(Duration::from_secs(10))) + .await?; + assert_eq!(tools.tools[0].name.as_ref(), "stdio_echo"); + let resources = client + .list_resources(/*params*/ None, Some(Duration::from_secs(10))) + .await?; + assert_eq!(resources.resources[0].uri, "test://stdio/resource"); + let templates = client + .list_resource_templates(/*params*/ None, Some(Duration::from_secs(10))) + .await?; + assert_eq!( + templates.resource_templates[0].name, + "stdio resource template" + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn legacy_stdio_preserves_existing_protocol_marker_environment() -> anyhow::Result<()> { + let server = codex_utils_cargo_bin::cargo_bin("test_stdio_server")?; + + for executor in [false, true] { + for version in ["2026-07-28", "1999-01-01"] { + let launcher: Arc = if executor { + Arc::new(ExecutorStdioServerLauncher::new( + Environment::default_for_tests().get_exec_backend(), + )) + } else { + Arc::new(LocalStdioServerLauncher::new(std::env::current_dir()?)) + }; + let client = RmcpClient::new_stdio_client_with_protocol_mode( + server.clone().into(), + Vec::new(), + Some(HashMap::from([( + OsString::from("CODEX_MCP_PROTOCOL_VERSION"), + OsString::from(version), + )])), + &[], + Some(std::env::current_dir()?.to_string_lossy().into_owned()), + launcher, + McpProtocolMode::Legacy, + ) + .await?; + + client + .initialize( + InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("stdio-legacy-environment-test", "1.0.0"), + ) + .with_protocol_version(ProtocolVersion::V_2025_06_18), + Some(Duration::from_secs(10)), + Box::new(|_, _| { + async { + Ok(ElicitationResponse { + action: ElicitationAction::Decline, + content: None, + meta: None, + }) + } + .boxed() + }), + ) + .await?; + + let result = client + .call_tool( + "echo".to_string(), + Some(json!({ + "message": "legacy environment", + "env_var": "CODEX_MCP_PROTOCOL_VERSION", + })), + /*meta*/ None, + Some(Duration::from_secs(10)), + ) + .await?; + assert_eq!( + result.structured_content, + Some(json!({ + "echo": "ECHOING: legacy environment", + "env": version, + })) + ); + client.shutdown().await; + } + } + + Ok(()) +} diff --git a/codex-rs/rmcp-client/tests/stdio_message_limits.rs b/codex-rs/rmcp-client/tests/stdio_message_limits.rs new file mode 100644 index 0000000000..0ddc4c3ee6 --- /dev/null +++ b/codex-rs/rmcp-client/tests/stdio_message_limits.rs @@ -0,0 +1,88 @@ +use std::collections::HashMap; +use std::ffi::OsString; +use std::sync::Arc; +use std::time::Duration; + +use codex_exec_server::Environment; +use codex_rmcp_client::ElicitationAction; +use codex_rmcp_client::ElicitationResponse; +use codex_rmcp_client::ExecutorStdioServerLauncher; +use codex_rmcp_client::LocalStdioServerLauncher; +use codex_rmcp_client::McpProtocolMode; +use codex_rmcp_client::RmcpClient; +use codex_rmcp_client::StdioServerLauncher; +use futures::FutureExt; +use rmcp::model::ClientCapabilities; +use rmcp::model::Implementation; +use rmcp::model::InitializeRequestParams; +use rmcp::model::ProtocolVersion; + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn stdio_message_limits_preserve_legacy_local_compatibility() -> anyhow::Result<()> { + let server = codex_utils_cargo_bin::cargo_bin("test_stdio_server")?; + + for (executor, protocol_mode, accepts_oversized) in [ + (false, McpProtocolMode::Legacy, true), + (false, McpProtocolMode::V20260728, false), + (true, McpProtocolMode::Legacy, false), + ] { + let launcher: Arc = if executor { + Arc::new(ExecutorStdioServerLauncher::new( + Environment::default_for_tests().get_exec_backend(), + )) + } else { + Arc::new(LocalStdioServerLauncher::new(std::env::current_dir()?)) + }; + let mut env = HashMap::from([( + OsString::from("MCP_TEST_OVERSIZED_TOOL_DESCRIPTION"), + OsString::from("1"), + )]); + if protocol_mode == McpProtocolMode::V20260728 { + env.insert( + OsString::from("CODEX_MCP_PROTOCOL_VERSION"), + OsString::from("2026-07-28"), + ); + } + let client = RmcpClient::new_stdio_client_with_protocol_mode( + server.clone().into(), + Vec::new(), + Some(env), + &[], + Some(std::env::current_dir()?.to_string_lossy().into_owned()), + launcher, + protocol_mode, + ) + .await?; + client + .initialize( + InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("stdio-limit-test", "1.0.0"), + ) + .with_protocol_version(ProtocolVersion::V_2025_06_18), + Some(Duration::from_secs(10)), + Box::new(|_, _| { + async { + Ok(ElicitationResponse { + action: ElicitationAction::Decline, + content: None, + meta: None, + }) + } + .boxed() + }), + ) + .await?; + + let result = client + .list_tools(/*params*/ None, Some(Duration::from_secs(10))) + .await; + assert_eq!( + result.is_ok(), + accepts_oversized, + "unexpected stdio size handling (executor={executor}, mode={protocol_mode:?})" + ); + client.shutdown().await; + } + Ok(()) +}