diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 51855f9993..0ff995088b 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3932,6 +3932,7 @@ dependencies = [ "codex-exec-server-test-support", "codex-login", "codex-model-provider", + "codex-network-proxy", "codex-otel", "codex-plugin", "codex-protocol", diff --git a/codex-rs/app-server/src/request_processors/apps_processor/installed.rs b/codex-rs/app-server/src/request_processors/apps_processor/installed.rs index 00f4f86c49..7631f24266 100644 --- a/codex-rs/app-server/src/request_processors/apps_processor/installed.rs +++ b/codex-rs/app-server/src/request_processors/apps_processor/installed.rs @@ -98,6 +98,7 @@ impl AppsRequestProcessor { host_owned_codex_apps_enabled(&mcp_config, auth.as_ref()) .then(|| Arc::clone(&self.auth_manager)); let runtime = McpRuntime::new(McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::clone(&mcp_config), plugins_available: false, diff --git a/codex-rs/codex-mcp/Cargo.toml b/codex-rs/codex-mcp/Cargo.toml index f54c6f1cbd..49adaabe6d 100644 --- a/codex-rs/codex-mcp/Cargo.toml +++ b/codex-rs/codex-mcp/Cargo.toml @@ -24,6 +24,7 @@ codex-diagnostics = { workspace = true } codex-exec-server = { workspace = true } codex-login = { workspace = true } codex-model-provider = { workspace = true } +codex-network-proxy = { workspace = true } codex-otel = { workspace = true } codex-protocol = { workspace = true } codex-rmcp-client = { workspace = true } diff --git a/codex-rs/codex-mcp/src/binding.rs b/codex-rs/codex-mcp/src/binding.rs index e19001d8af..be40ac75f9 100644 --- a/codex-rs/codex-mcp/src/binding.rs +++ b/codex-rs/codex-mcp/src/binding.rs @@ -427,7 +427,17 @@ impl PreparedMcpCall { }; self.client .client - .call_tool(tool_name.clone(), arguments, meta, remaining_timeout) + .call_tool( + tool_name.clone(), + arguments, + self.connections.network_policy_meta_for_call( + &self.client, + Some(self.permission_profile()), + self.server_environment_id(), + meta, + ), + remaining_timeout, + ) .await .with_context(|| format!("tool call failed for `{}/{tool_name}`", self.server_name)) }) diff --git a/codex-rs/codex-mcp/src/binding_tests.rs b/codex-rs/codex-mcp/src/binding_tests.rs index 363f8760e7..d816befd1b 100644 --- a/codex-rs/codex-mcp/src/binding_tests.rs +++ b/codex-rs/codex-mcp/src/binding_tests.rs @@ -98,6 +98,7 @@ async fn test_step( tool_timeout: None, server_instructions: None, server_supports_sandbox_state_meta_capability: supports_sandbox_state_meta, + client_supports_network_policy_meta_capability: false, codex_apps_tools_cache_context: None, }); let clients = Arc::new(McpBindingClients::new(HashMap::from([( diff --git a/codex-rs/codex-mcp/src/client_capabilities.rs b/codex-rs/codex-mcp/src/client_capabilities.rs index 582891dbda..8e514d6c1c 100644 --- a/codex-rs/codex-mcp/src/client_capabilities.rs +++ b/codex-rs/codex-mcp/src/client_capabilities.rs @@ -8,27 +8,46 @@ use codex_protocol::mcp::OPENAI_STANDARD_FORM_INPUT_EXTENSION_ID; use serde_json::Map; use serde_json::Value; -/// Restricts verification to the host-owned plugin service, even when another server uses its name. +/// Projects privileged capabilities only to their intended recipients. +/// Verification requires the host-owned plugin service; network policy requires a trusted +/// ChatGPT origin. Server names and caller-supplied capabilities do not grant either privilege. pub(crate) fn server_mcp_extensions( extensions: &ClientMcpExtensions, server_name: &str, server: Option<&crate::catalog::ResolvedMcpServer>, + advertise_network_policy: bool, + chatgpt_base_url: &str, ) -> ClientMcpExtensions { let plugin_service = server.is_some_and(|server| { server .source() .is_host_owned_apps(server_name, server.config()) }); - ClientMcpExtensions::new(extensions.iter().map(|(id, settings)| { - let mut settings = settings.clone(); - if !plugin_service - && id == OPENAI_ELICITATION_EXTENSION_ID - && let Some(settings) = settings.as_object_mut() - { - settings.remove("userVerification"); - } - (id.to_string(), settings) - })) + let network_policy = advertise_network_policy + && server.is_some_and(|server| { + crate::mcp::is_trusted_chatgpt_mcp_server(&server.config().transport, chatgpt_base_url) + }); + ClientMcpExtensions::new( + extensions + .iter() + .filter(|(id, _)| *id != crate::MCP_NETWORK_POLICY_META_CAPABILITY) + .map(|(id, settings)| { + let mut settings = settings.clone(); + if !plugin_service + && id == OPENAI_ELICITATION_EXTENSION_ID + && let Some(settings) = settings.as_object_mut() + { + settings.remove("userVerification"); + } + (id.to_string(), settings) + }) + .chain(network_policy.then(|| { + ( + crate::MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + Value::Object(Map::new()), + ) + })), + ) } /// Selects the MCP extensions Codex supports from those declared by the app-server host. diff --git a/codex-rs/codex-mcp/src/client_capabilities_tests.rs b/codex-rs/codex-mcp/src/client_capabilities_tests.rs index 17545a16e7..176917352d 100644 --- a/codex-rs/codex-mcp/src/client_capabilities_tests.rs +++ b/codex-rs/codex-mcp/src/client_capabilities_tests.rs @@ -75,10 +75,29 @@ fn user_verification_is_projected_only_to_the_host_owned_plugin_service() { /*apps_mcp_product_sku*/ None, /*originator*/ None, ); - let extensions = ClientMcpExtensions::new([( - OPENAI_ELICITATION_EXTENSION_ID.to_string(), - json!({"form": {}, "userVerification": {}}), - )]); + let extensions = ClientMcpExtensions::new([ + ( + OPENAI_ELICITATION_EXTENSION_ID.to_string(), + json!({"form": {}, "userVerification": {}}), + ), + ( + crate::MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + json!({}), + ), + ]); + assert_eq!( + server_mcp_extensions( + &extensions, + CODEX_APPS_MCP_SERVER_NAME, + /*server*/ None, + /*advertise_network_policy*/ true, + "https://chatgpt.com", + ), + ClientMcpExtensions::new([( + OPENAI_ELICITATION_EXTENSION_ID.to_string(), + json!({"form": {}}), + )]), + ); for (name, registration, settings) in [ ( CODEX_APPS_MCP_SERVER_NAME, @@ -125,7 +144,13 @@ fn user_verification_is_projected_only_to_the_host_owned_plugin_service() { catalog.register(registration); let catalog = catalog.build(); assert_eq!( - server_mcp_extensions(&extensions, name, catalog.server(name)), + server_mcp_extensions( + &extensions, + name, + catalog.server(name), + /*advertise_network_policy*/ false, + "https://chatgpt.com", + ), ClientMcpExtensions::new([(OPENAI_ELICITATION_EXTENSION_ID.to_string(), settings)]), ); } diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index 3115ea9472..125d934cb0 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -7,8 +7,12 @@ #[path = "connection_manager/catalog_telemetry.rs"] mod catalog_telemetry; +#[path = "connection_manager/network_policy.rs"] +mod network_policy; #[path = "connection_manager/required.rs"] mod required; +pub use network_policy::NetworkPolicyMetadata; +pub use network_policy::NetworkPolicyMetadataProvider; #[path = "connection_manager/resources.rs"] mod resources; #[path = "connection_manager/startup.rs"] @@ -266,6 +270,8 @@ impl McpServerView { /// A published view over a set of running MCP server connections. pub(crate) struct McpConnectionSet { + config: Option>, + network_policy_metadata: Option>, servers: HashMap, pub(crate) event_stream_connection: Option>, disabled_servers: Vec, @@ -293,6 +299,7 @@ impl McpConnectionSet { let McpRuntimeInput { startup_policy, config, + network_policy_metadata, plugins_available: _, ready_selected_capability_roots: _, mcp_servers, @@ -391,6 +398,8 @@ impl McpConnectionSet { &client_mcp_extensions, &server_name, registration, + network_policy_metadata.is_some(), + &config.chatgpt_base_url, ); let is_host_owned_codex_apps = registration.is_some_and(|server| { server @@ -951,6 +960,8 @@ impl McpConnectionSet { } } let manager = Self { + config: Some(config), + network_policy_metadata, servers, event_stream_connection, disabled_servers, @@ -1009,6 +1020,8 @@ impl McpConnectionSet { pub fn empty(prefix_mcp_tool_names: bool) -> Self { Self { + config: None, + network_policy_metadata: None, servers: HashMap::new(), event_stream_connection: None, disabled_servers: Vec::new(), @@ -1210,7 +1223,19 @@ impl McpConnectionSet { } let result: rmcp::model::CallToolResult = client .client - .call_tool(tool.to_string(), arguments, meta, effective_timeout) + .call_tool( + tool.to_string(), + arguments, + self.network_policy_meta_for_call( + &client, + self.config + .as_ref() + .and_then(|config| config.permission_profile_for_server(server)), + &view.metadata.environment_id, + meta, + ), + effective_timeout, + ) .await .with_context(|| format!("tool call failed for `{server}/{tool}`"))?; diff --git a/codex-rs/codex-mcp/src/connection_manager/network_policy.rs b/codex-rs/codex-mcp/src/connection_manager/network_policy.rs new file mode 100644 index 0000000000..b8ef47790f --- /dev/null +++ b/codex-rs/codex-mcp/src/connection_manager/network_policy.rs @@ -0,0 +1,85 @@ +//! Adds negotiated policy metadata using the same publication as the MCP call. + +use super::McpConnectionSet; +use crate::MCP_NETWORK_POLICY_META_CAPABILITY; +use crate::rmcp_client::ManagedClient; +use anyhow::Context; +use codex_network_proxy::NetworkMode; +use codex_protocol::models::PermissionProfile; +use rmcp::model::ServerCapabilities; +use serde::Serialize; +use serde_json::Value; +use std::sync::OnceLock; + +/// Effective network restrictions delivered with an MCP call, without proxy endpoints or credentials. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct NetworkPolicyMetadata { + pub enabled: bool, + pub mode: NetworkMode, + pub allowed_domains: Vec, + pub denied_domains: Vec, + pub hard_deny_allowlist_misses: bool, +} + +/// Produces trusted network policy metadata from an MCP publication's frozen policy sources. +/// Implementations must use the supplied call authority and capture mutable grants once per call. +pub trait NetworkPolicyMetadataProvider: Send + Sync { + fn metadata_for_call( + &self, + permission_profile: &PermissionProfile, + environment_id: &str, + ) -> anyhow::Result; +} + +impl McpConnectionSet { + pub(crate) fn network_policy_meta_for_call( + &self, + client: &ManagedClient, + permission_profile: Option<&PermissionProfile>, + environment_id: &str, + mut meta: Option, + ) -> impl Fn(&ServerCapabilities) -> anyhow::Result> + Send + Sync { + if let Some(Value::Object(meta)) = meta.as_mut() { + meta.remove(MCP_NETWORK_POLICY_META_CAPABILITY); + } + // A retry can negotiate policy support for the first time. Capture mutable grants + // only once, when an attempt needs them, and keep that snapshot through recovery. + let policy = OnceLock::new(); + move |capabilities| { + let mut meta = meta.clone(); + if client.client_supports_network_policy_meta_capability + && capabilities + .experimental + .as_ref() + .is_some_and(|experimental| { + experimental.contains_key(MCP_NETWORK_POLICY_META_CAPABILITY) + }) + { + let policy = policy.get_or_init(|| -> anyhow::Result { + let provider = self + .network_policy_metadata + .as_ref() + .context("MCP network policy metadata source is unavailable")?; + let permission_profile = permission_profile + .context("MCP server has no captured permission authority")?; + serde_json::to_value( + provider.metadata_for_call(permission_profile, environment_id)?, + ) + .map_err(Into::into) + }); + let policy = policy + .as_ref() + .map_err(|error| anyhow::anyhow!("{error:#}"))?; + meta.get_or_insert_with(|| Value::Object(Default::default())) + .as_object_mut() + .context("MCP request metadata must be an object")? + .insert( + MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + policy.clone(), + ); + } + Ok(meta) + } + } +} diff --git a/codex-rs/codex-mcp/src/connection_manager/network_policy_tests.rs b/codex-rs/codex-mcp/src/connection_manager/network_policy_tests.rs new file mode 100644 index 0000000000..774f7aa183 --- /dev/null +++ b/codex-rs/codex-mcp/src/connection_manager/network_policy_tests.rs @@ -0,0 +1,282 @@ +//! Exercises policy metadata through real in-process MCP requests on both dispatch paths. + +use super::*; +use crate::MCP_NETWORK_POLICY_META_CAPABILITY; +use crate::NetworkPolicyMetadata; +use crate::NetworkPolicyMetadataProvider; +use crate::catalog::McpServerRegistration; +use crate::catalog::ResolvedMcpCatalog; +use codex_network_proxy::NetworkMode; +use pretty_assertions::assert_eq; +use rmcp::model::CallToolRequestParams; +use rmcp::model::CallToolResponse; +use serde_json::Value; +use serde_json::json; + +#[derive(Clone)] +struct MetadataEchoServer { + supports_policy: bool, +} + +impl ServerHandler for MetadataEchoServer { + fn get_info(&self) -> ServerConfig { + let mut capabilities = ServerCapabilities::builder().enable_tools().build(); + if self.supports_policy { + capabilities.experimental = Some(std::collections::BTreeMap::from([( + MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + JsonObject::new(), + )])); + } + ServerConfig::new(capabilities) + } + + async fn call_tool( + &self, + _request: CallToolRequestParams, + context: RequestContext, + ) -> Result { + let mut result = rmcp::model::CallToolResult::success(Vec::new()); + result.structured_content = Some(Value::Object(context.meta.0.0)); + Ok(CallToolResponse::Complete(result)) + } +} + +impl InProcessTransportFactory for MetadataEchoServer { + fn open(&self) -> BoxFuture<'static, io::Result> { + let server = self.clone(); + async move { + let (client_stream, server_stream) = tokio::io::duplex(4096); + tokio::spawn(async move { + let server = server + .serve(server_stream) + .await + .expect("serve metadata echo"); + server.waiting().await + }); + Ok(client_stream) + } + .boxed() + } +} + +#[derive(Default)] +struct CapturedPolicy(AtomicUsize); + +impl NetworkPolicyMetadataProvider for CapturedPolicy { + fn metadata_for_call( + &self, + profile: &PermissionProfile, + environment: &str, + ) -> anyhow::Result { + assert_eq!( + (profile, environment), + (&PermissionProfile::read_only(), "owner-environment") + ); + self.0.fetch_add(1, Ordering::SeqCst); + Ok(NetworkPolicyMetadata { + enabled: false, + mode: NetworkMode::Limited, + allowed_domains: vec!["allowed.example.com".to_string()], + denied_domains: vec!["blocked.example.com".to_string()], + hard_deny_allowlist_misses: true, + }) + } +} + +#[tokio::test] +async fn tool_calls_deliver_only_negotiated_trusted_network_policy() -> anyhow::Result<()> { + for (url, provider_available, supports_policy, delivers_policy) in [ + (Some("https://chatgpt.com/mcp"), true, true, true), + (Some("https://chatgpt.com/mcp"), true, false, false), + (Some("https://chatgpt.com/mcp"), false, true, false), + (Some("https://gov.chatgpt.com/mcp"), true, true, true), + ( + Some("https://chatgpt.com.attacker.example/mcp"), + true, + true, + false, + ), + (Some("http://chatgpt.com/mcp"), true, true, false), + (Some("https://third-party.example/mcp"), true, true, false), + (None, true, true, false), + ] { + let mut server_config = reusable_server_config(url.unwrap_or("https://chatgpt.com/mcp")); + if url.is_none() { + server_config.transport = McpServerTransportConfig::Stdio { + command: "metadata-echo".to_string(), + args: Vec::new(), + env: None, + env_vars: Vec::new(), + cwd: None, + }; + } + let mut catalog = ResolvedMcpCatalog::builder(); + catalog.register(McpServerRegistration::from_config( + "docs".into(), + server_config, + )); + let catalog = catalog.build(); + let extensions = crate::client_capabilities::server_mcp_extensions( + &ClientMcpExtensions::new([( + MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + json!({"forged": true}), + )]), + "docs", + catalog.server("docs"), + provider_available, + "https://chatgpt.com", + ); + let client_supports_policy = extensions + .iter() + .any(|(id, _)| id == MCP_NETWORK_POLICY_META_CAPABILITY); + let client = Arc::new( + RmcpClient::new_in_process_client(Arc::new(MetadataEchoServer { supports_policy })) + .await?, + ); + client + .initialize( + crate::rmcp_client::mcp_initialize_request_params( + ElicitationCapability::default(), + extensions, + ), + Some(Duration::from_secs(5)), + Box::new(|_, _| async { Err(anyhow!("unexpected elicitation")) }.boxed()), + ) + .await?; + let mut managed = create_test_managed_client(vec![create_test_tool("docs", "echo")]).await; + managed.client = Arc::clone(&client); + managed.client_supports_network_policy_meta_capability = client_supports_policy; + let (managed, _started, release) = create_gated_async_managed_client(managed); + release.send(()).expect("release initialized client"); + let policy = Arc::new(CapturedPolicy::default()); + let mut config = crate::mcp::tests::test_mcp_config(std::env::temp_dir()); + config.permission_profile = PermissionProfile::Disabled; + config + .server_permission_profiles + .insert("docs".to_string(), PermissionProfile::read_only()); + let config = Arc::new(config); + let mut manager = McpConnectionSet::empty(/*prefix_mcp_tool_names*/ true); + manager.config = Some(Arc::clone(&config)); + manager.network_policy_metadata = + provider_available.then(|| policy.clone() as Arc); + manager.insert_test_client("docs", managed); + let view = manager.servers.get_mut("docs").expect("docs server"); + view.metadata.environment_id = "owner-environment".to_string(); + let manager = Arc::new(manager); + let binding = manager + .capture_binding_with_metadata( + config, + /*plugins_available*/ false, + /*required_servers*/ &[], + /*required_plugins*/ &HashSet::new(), + ) + .await; + let prepared = binding + .prepare_call("docs", "echo") + .expect("prepare advertised tool"); + let forged_meta = || { + Some(json!({ + "codex/network-policy-meta": {"enabled": true, "forged": true}, + "keep": "caller metadata", + })) + }; + let mut expected = json!({"keep": "caller metadata"}); + if delivers_policy { + expected[MCP_NETWORK_POLICY_META_CAPABILITY] = json!({ + "enabled": false, + "mode": "limited", + "allowedDomains": ["allowed.example.com"], + "deniedDomains": ["blocked.example.com"], + "hardDenyAllowlistMisses": true, + }); + } + for result in [ + prepared + .call( + /*arguments*/ None, + forged_meta(), + /*timeout*/ None, + ) + .await?, + manager + .call_tool( + "docs", + "echo", + /*environment_id*/ None, + /*arguments*/ None, + forged_meta(), + /*requested_timeout*/ None, + /*wait_for_server*/ true, + ) + .await?, + ] { + let mut received = result.structured_content.expect("echoed wire metadata"); + received + .as_object_mut() + .expect("metadata object") + .remove("progressToken"); + assert_eq!(received, expected); + } + assert_eq!( + policy.0.load(Ordering::SeqCst), + if delivers_policy { 2 } else { 0 } + ); + client.shutdown().await; + } + Ok(()) +} + +#[tokio::test] +async fn recovered_session_policy_keeps_one_captured_call_authority() -> anyhow::Result<()> { + let mut client = create_test_managed_client(Vec::new()).await; + client.client_supports_network_policy_meta_capability = true; + let policy = Arc::new(CapturedPolicy::default()); + let mut manager = McpConnectionSet::empty(/*prefix_mcp_tool_names*/ true); + manager.network_policy_metadata = Some(policy.clone()); + let profile = PermissionProfile::read_only(); + let metadata = manager.network_policy_meta_for_call( + &client, + Some(&profile), + "owner-environment", + Some(json!({"codex/network-policy-meta": {"forged": true}, "keep": "caller"})), + ); + let legacy = ServerCapabilities::default(); + let mut negotiated = legacy.clone(); + negotiated.experimental = Some(std::collections::BTreeMap::from([( + MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + JsonObject::new(), + )])); + + assert_eq!(metadata(&legacy)?, Some(json!({"keep": "caller"}))); + assert_eq!(policy.0.load(Ordering::SeqCst), 0); + let expected = Some(json!({ + "keep": "caller", + "codex/network-policy-meta": { + "enabled": false, + "mode": "limited", + "allowedDomains": ["allowed.example.com"], + "deniedDomains": ["blocked.example.com"], + "hardDenyAllowlistMisses": true, + }, + })); + assert_eq!(metadata(&negotiated)?, expected); + assert_eq!(metadata(&legacy)?, Some(json!({"keep": "caller"}))); + assert_eq!(metadata(&negotiated)?, expected); + assert_eq!(policy.0.load(Ordering::SeqCst), 1); + + let missing_provider = McpConnectionSet::empty(/*prefix_mcp_tool_names*/ true); + let metadata = missing_provider.network_policy_meta_for_call( + &client, + Some(&profile), + "owner-environment", + /*meta*/ None, + ); + assert_eq!(metadata(&legacy)?, None); + assert!( + metadata(&negotiated) + .unwrap_err() + .to_string() + .contains("source is unavailable") + ); + Ok(()) +} diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index 449fbb9e65..06e4c3b8d5 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -1,4 +1,6 @@ use super::*; +#[path = "connection_manager/network_policy_tests.rs"] +mod network_policy_tests; use crate::McpBinding; use crate::client_tool_catalog::ClientToolCatalog; use crate::elicitation::ElicitationLifecycle; @@ -103,6 +105,8 @@ impl McpConnectionSet { prefix_mcp_tool_names: bool, ) -> Self { Self { + config: None, + network_policy_metadata: None, servers: HashMap::new(), event_stream_connection: None, disabled_servers: Vec::new(), @@ -493,6 +497,7 @@ async fn create_test_managed_client(tools: Vec) -> ManagedClient { tool_timeout: None, server_instructions: None, server_supports_sandbox_state_meta_capability: false, + client_supports_network_policy_meta_capability: false, codex_apps_tools_cache_context: None, } } @@ -810,6 +815,7 @@ pub(crate) async fn create_test_manager_with_ready_apps_client( tool_timeout: Some(Duration::from_secs(5)), server_instructions: None, server_supports_sandbox_state_meta_capability: false, + client_supports_network_policy_meta_capability: false, codex_apps_tools_cache_context: Some(cache_context.clone()), }; let approval_policy = Constrained::allow_any(AskForApproval::OnRequest); @@ -2072,6 +2078,7 @@ async fn read_only_apps_discovery_never_uses_a_shared_writable_catalog() -> anyh /*previous*/ None, McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(config), plugins_available: false, @@ -2147,6 +2154,7 @@ async fn hosted_apps_protocol_mode_is_independent_of_generic_mode() -> anyhow::R /*previous*/ None, McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(config), plugins_available: false, @@ -2255,6 +2263,7 @@ async fn codex_apps_extension_does_not_share_host_owned_tools_cache() -> anyhow: /*previous*/ None, McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(config), plugins_available: false, @@ -4845,6 +4854,7 @@ async fn executor_owned_chatgpt_mcp_accepts_only_safe_explicit_authorization() - /*previous*/ None, McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(runtime_config.clone()), plugins_available: false, @@ -4965,6 +4975,7 @@ async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() { /*previous*/ None, McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(crate::mcp::tests::test_mcp_config( codex_home.path().to_path_buf(), @@ -5418,6 +5429,7 @@ async fn reconcile_reusable_server_with_mcp_config( Some(previous), McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(mcp_config), plugins_available: false, @@ -5789,6 +5801,7 @@ async fn reconciliation_reuses_connection_without_relisting_regular_tools() -> a tool_timeout: None, server_instructions: initialize.instructions, server_supports_sandbox_state_meta_capability: false, + client_supports_network_policy_meta_capability: false, codex_apps_tools_cache_context: None, }; let runtime_context = reusable_server_runtime_context(); @@ -5919,7 +5932,7 @@ async fn reconciliation_reuses_connection_without_relisting_regular_tools() -> a .call_tool( "notify".into(), /*arguments*/ None, - /*meta*/ None, + |_| Ok(None), /*timeout*/ None, ) .await?; @@ -6379,6 +6392,7 @@ async fn reconciliation_replaces_connection_when_protocol_mode_changes() { Some(&previous), McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(mcp_config), plugins_available: false, @@ -6439,6 +6453,7 @@ async fn reconciliation_reuses_legacy_stdio_server_when_modern_protocol_is_enabl Some(&previous), McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(mcp_config), plugins_available: false, diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index 5768aa71cd..71db3cb5a7 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -3,6 +3,8 @@ pub use binding::PreparedMcpCall; pub use client_capabilities::client_mcp_extensions; pub use client_tool_catalog::CodexAppsToolSnapshot; pub use codex_rmcp_client::McpProtocolMode; +pub use connection_manager::NetworkPolicyMetadata; +pub use connection_manager::NetworkPolicyMetadataProvider; pub use connection_manager::tool_is_model_visible; pub use elicitation::ElicitationLifecycle; pub use elicitation::ElicitationReviewRequest; @@ -22,6 +24,7 @@ pub use resource_client::McpResourcePage; pub use resource_client::McpResourceReadResult; pub use resource_client::McpResourceServerCacheKey; pub use rmcp::model::ReadResourceRequestParams; +pub use rmcp_client::MCP_NETWORK_POLICY_META_CAPABILITY; pub use runtime::McpRuntime; pub use runtime::McpRuntimeContext; pub use runtime::McpRuntimeInput; diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index d73825b6f6..fa5242e3e7 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -366,7 +366,7 @@ pub fn effective_mcp_servers( effective_mcp_servers_from_configured(configured_mcp_servers(config), config, auth) } -fn is_trusted_chatgpt_mcp_server( +pub(crate) fn is_trusted_chatgpt_mcp_server( transport: &McpServerTransportConfig, chatgpt_base_url: &str, ) -> bool { @@ -470,6 +470,7 @@ pub async fn read_mcp_resource( /*previous*/ None, McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(runtime_config), plugins_available: false, @@ -558,6 +559,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( /*previous*/ None, McpPublicationGate::already_published(), McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::new(runtime_config), plugins_available: false, diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index b60e35438f..54d5d4dfe7 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -99,6 +99,8 @@ use tracing::Instrument; use tracing::instrument; use tracing::warn; +/// Experimental client/server capability for call-scoped network policy metadata. +pub const MCP_NETWORK_POLICY_META_CAPABILITY: &str = "codex/network-policy-meta"; /// Experimental MCP server capability for development and testing only; production servers should /// not use it. Its `cacheable: false` property disables sharing tool definitions across connections. const MCP_TOOL_CATALOG_CACHE_CAPABILITY: &str = "codex/tool-catalog-cache"; @@ -132,6 +134,7 @@ pub(crate) struct ManagedClient { pub(crate) tool_timeout: Option, pub(crate) server_instructions: Option, pub(crate) server_supports_sandbox_state_meta_capability: bool, + pub(crate) client_supports_network_policy_meta_capability: bool, pub(crate) codex_apps_tools_cache_context: Option>, } @@ -1020,6 +1023,8 @@ async fn start_server_task( } = params; let send_elicitation = elicitation_requests.make_sender(server_name.clone(), tx_event, &client_mcp_extensions); + let client_supports_network_policy_meta = + client_mcp_extensions.contains(MCP_NETWORK_POLICY_META_CAPABILITY); let mut params = mcp_initialize_request_params(client_elicitation_capability, client_mcp_extensions); if auth_changes.is_some() { @@ -1162,6 +1167,7 @@ async fn start_server_task( tool_timeout: None, server_instructions: initialize_result.instructions, server_supports_sandbox_state_meta_capability, + client_supports_network_policy_meta_capability: client_supports_network_policy_meta, codex_apps_tools_cache_context, }; @@ -1205,8 +1211,15 @@ pub(crate) fn mcp_initialize_request_params( ) -> InitializeRequestParams { let mut capabilities = ClientCapabilities::default(); capabilities.elicitation = Some(client_elicitation_capability); + if client_mcp_extensions.contains(MCP_NETWORK_POLICY_META_CAPABILITY) { + capabilities.experimental = Some(BTreeMap::from([( + MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + serde_json::Map::new(), + )])); + } let extensions = client_mcp_extensions .iter() + .filter(|(id, _)| *id != MCP_NETWORK_POLICY_META_CAPABILITY) .filter_map(|(id, settings)| { settings .as_object() @@ -1508,6 +1521,7 @@ mod tests { ClientMcpExtensions::default(), ); assert_eq!(unsupported.capabilities.extensions, None); + assert_eq!(unsupported.capabilities.experimental, None); let app_ui = serde_json::json!({ "mimeTypes": ["text/html;profile=mcp-app"], @@ -1516,10 +1530,21 @@ mod tests { let supported = mcp_initialize_request_params( ElicitationCapability::default(), ClientMcpExtensions::new([ + ( + MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + serde_json::json!({}), + ), (OPENAI_FORM_EXTENSION_ID.to_string(), serde_json::json!({})), (MCP_APP_UI_EXTENSION_ID.to_string(), app_ui.clone()), ]), ); + assert_eq!( + supported.capabilities.experimental, + Some(BTreeMap::from([( + MCP_NETWORK_POLICY_META_CAPABILITY.to_string(), + JsonObject::new() + )])), + ); assert_eq!( supported.capabilities.extensions, Some(BTreeMap::from([ diff --git a/codex-rs/codex-mcp/src/runtime.rs b/codex-rs/codex-mcp/src/runtime.rs index 2cb29fb685..52ab17007d 100644 --- a/codex-rs/codex-mcp/src/runtime.rs +++ b/codex-rs/codex-mcp/src/runtime.rs @@ -83,6 +83,8 @@ pub struct McpToolCallResult { /// Configuration and owning-thread state needed to materialize an MCP runtime. pub struct McpRuntimeInput { + /// Policy source captured with the configuration and connections in this publication. + pub network_policy_metadata: Option>, pub startup_policy: McpStartupPolicy, pub config: Arc, pub plugins_available: bool, diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 00640bcb75..9b1a34a572 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -252,6 +252,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( codex_mcp::host_owned_codex_apps_enabled(&mcp_config, auth.as_ref()) .then(|| Arc::clone(&auth_manager)); let mcp_runtime = McpRuntime::new(McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::clone(&mcp_config), plugins_available: false, diff --git a/codex-rs/core/src/session/mcp_runtime.rs b/codex-rs/core/src/session/mcp_runtime.rs index a2b28fadd7..d00bb6af41 100644 --- a/codex-rs/core/src/session/mcp_runtime.rs +++ b/codex-rs/core/src/session/mcp_runtime.rs @@ -367,6 +367,7 @@ impl Session { desired.environments.ready_environment_handles(), ); McpRuntimeInput { + network_policy_metadata: None, startup_policy: if matches!(desired.session_source, SessionSource::SubAgent(_)) { McpStartupPolicy::LazyWhenCached } else { diff --git a/codex-rs/core/tests/mcp_trace_propagation.rs b/codex-rs/core/tests/mcp_trace_propagation.rs index 1024b90027..dfdd79fefd 100644 --- a/codex-rs/core/tests/mcp_trace_propagation.rs +++ b/codex-rs/core/tests/mcp_trace_propagation.rs @@ -121,7 +121,7 @@ async fn mcp_requests_preserve_trace_context_across_workers_and_continuations() .call_tool( "test".to_string(), /*arguments*/ None, - Some(json!({"application":"tool-meta"})), + |_| Ok(Some(json!({"application":"tool-meta"}))), Some(Duration::from_secs(5)), ) .instrument(tool) diff --git a/codex-rs/core/tests/suite/mcp_auth_refresh.rs b/codex-rs/core/tests/suite/mcp_auth_refresh.rs index 03d42e2707..a21b0c5272 100644 --- a/codex-rs/core/tests/suite/mcp_auth_refresh.rs +++ b/codex-rs/core/tests/suite/mcp_auth_refresh.rs @@ -91,6 +91,7 @@ async fn hosted_plugin_runtime_ps_mcp_tool_calls_use_current_auth_manager_token( let plugins_manager = plugins_manager_for_config(&config, Arc::clone(&auth_manager)); let mcp_config = Arc::new(config.to_mcp_config(&plugins_manager).await); let runtime = McpRuntime::new(McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: mcp_config, plugins_available: false, diff --git a/codex-rs/ext/mcp/src/event_stream_tests.rs b/codex-rs/ext/mcp/src/event_stream_tests.rs index a6403cbb32..fdf5c6074a 100644 --- a/codex-rs/ext/mcp/src/event_stream_tests.rs +++ b/codex-rs/ext/mcp/src/event_stream_tests.rs @@ -155,6 +155,7 @@ impl Fixture { config.mcp_server_catalog = catalog.build(); let config = Arc::new(config); let runtime_input = || McpRuntimeInput { + network_policy_metadata: None, startup_policy: McpStartupPolicy::Eager, config: Arc::clone(&config), plugins_available: false, diff --git a/codex-rs/rmcp-client/src/rmcp_client.rs b/codex-rs/rmcp-client/src/rmcp_client.rs index c9d22f23c7..d5cdf26aa1 100644 --- a/codex-rs/rmcp-client/src/rmcp_client.rs +++ b/codex-rs/rmcp-client/src/rmcp_client.rs @@ -831,13 +831,18 @@ impl RmcpClient { Ok(result) } - pub async fn call_tool( + /// Builds metadata against the session that sends each attempt, including a recovery retry. + /// The resolver must be safe to invoke again and must retain the call's captured authority. + pub async fn call_tool( &self, name: String, - arguments: Option, - meta: Option, + arguments: Option, + meta_for_session: F, timeout: Option, - ) -> Result { + ) -> Result + where + F: Fn(&rmcp::model::ServerCapabilities) -> Result> + Send + Sync, + { let authentication_required_result = |error| { if !is_authentication_required_error(&error) { return Err(error); @@ -868,32 +873,46 @@ impl RmcpClient { } None => None, }; - let mut meta = match meta { - Some(Value::Object(map)) => Some(RequestMetaObject::from(map)), - Some(other) => { - return Err(anyhow!( - "MCP tool request _meta must be a JSON object, got {other}" - )); - } - None => None, - }; - self.apply_read_only_tools_meta(&mut meta); - let meta = crate::trace_context::with_current_trace(meta); let mut rmcp_params = CallToolRequestParams::new(name); rmcp_params.arguments = arguments; let requested_modern = self.protocol_mode == McpProtocolMode::V20260728; + let meta_for_session = &meta_for_session; match self .run_service_operation("tools/call", timeout, move |service| { let mut rmcp_params = rmcp_params.clone(); - let meta = meta.clone(); async move { + // Resolve from this service, not a cached startup capability or another + // service observed before recovery. Local policy failures never send a call. + let meta = (|| { + let info = service + .peer() + .peer_info() + .context("MCP session has no negotiated server capabilities")?; + let mut meta = match meta_for_session(&info.capabilities)? { + Some(Value::Object(map)) => Some(RequestMetaObject::from(map)), + Some(other) => { + return Err(anyhow!( + "MCP tool request _meta must be a JSON object, got {other}" + )); + } + None => None, + }; + self.apply_read_only_tools_meta(&mut meta); + Ok(crate::trace_context::with_current_trace(meta)) + })(); + let meta = match meta { + Ok(meta) => meta, + Err(error) => return Ok(Err(error)), + }; let modern_session = requested_modern && service.peer().peer_info().is_some_and(|info| { info.protocol_version == ProtocolVersion::V_2026_07_28 }); if modern_session { rmcp_params.meta = meta; - return crate::tool_input::call_tool(&service, rmcp_params).await; + return crate::tool_input::call_tool(&service, rmcp_params) + .await + .map(Ok); } let mut options = rmcp::service::PeerRequestOptions::no_options(); options.meta = meta; @@ -901,18 +920,19 @@ impl RmcpClient { crate::tool_request::call_tool(service.peer(), rmcp_params, options) .await?; match result { - ServerResult::CallToolResult(result) => Ok(result), - _ => Err(rmcp::service::ServiceError::UnexpectedResponse), + ServerResult::CallToolResult(result) => Ok(Ok(result)), + _ => Err(ServiceError::UnexpectedResponse), } } .boxed() }) .await { - Ok(result) => { + Ok(Ok(result)) => { self.persist_oauth_tokens().await; Ok(result) } + Ok(Err(error)) => Err(error), Err(error) => { let Some(ClientOperationError::Service(ServiceError::TransportSend(transport))) = error.downcast_ref() diff --git a/codex-rs/rmcp-client/tests/mcp_2026_discovery.rs b/codex-rs/rmcp-client/tests/mcp_2026_discovery.rs index 8ec5b12764..0adfe630b4 100644 --- a/codex-rs/rmcp-client/tests/mcp_2026_discovery.rs +++ b/codex-rs/rmcp-client/tests/mcp_2026_discovery.rs @@ -34,6 +34,9 @@ const MODERN_VERSION: &str = "2026-07-28"; const LEGACY_VERSION: &str = "2025-06-18"; const MAX_MCP_MESSAGE_BYTES: usize = 8 * 1024 * 1024; +#[path = "mcp_2026_discovery/tool_call_session_meta_tests.rs"] +mod tool_call_session_meta_tests; + fn initialize_params() -> InitializeRequestParams { InitializeRequestParams::new( ClientCapabilities::default(), @@ -152,7 +155,7 @@ async fn read_only_tool_requests_preserve_host_policy_and_caller_metadata() -> a .call_tool( "write".to_string(), Some(json!({"message": "hello"})), - Some(caller_meta), + |_| Ok(Some(caller_meta.clone())), /*timeout*/ None, ) .await?; @@ -392,7 +395,7 @@ async fn legacy_fallback_does_not_follow_cross_origin_tool_redirects() -> anyhow .call_tool( "query_logs_sql".to_string(), Some(json!({"query": "sensitive-query"})), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(5)), ) .await @@ -812,7 +815,7 @@ async fn modern_sdk_emits_standard_headers_and_rejects_invalid_header_schemas() .call_tool( "echo".to_string(), Some(json!({"tenant": "acme"})), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(5)), ) .await?; diff --git a/codex-rs/rmcp-client/tests/mcp_2026_discovery/tool_call_session_meta_tests.rs b/codex-rs/rmcp-client/tests/mcp_2026_discovery/tool_call_session_meta_tests.rs new file mode 100644 index 0000000000..c4e8e96eb4 --- /dev/null +++ b/codex-rs/rmcp-client/tests/mcp_2026_discovery/tool_call_session_meta_tests.rs @@ -0,0 +1,140 @@ +//! Verifies metadata on the actual HTTP session-expiry retry and subsequent tool calls. + +use super::*; +use pretty_assertions::assert_eq; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +#[tokio::test] +async fn recovered_tool_calls_use_current_session_capabilities() -> anyhow::Result<()> { + for (initial_support, recovered_support, fail_policy) in [ + (false, true, false), + (true, false, false), + (true, true, false), + (false, false, false), + (false, true, true), + ] { + let server = MockServer::start().await; + let handshakes = Arc::new(AtomicUsize::new(0)); + let calls = Arc::new(AtomicUsize::new(0)); + let observed_handshakes = Arc::clone(&handshakes); + let observed_calls = Arc::clone(&calls); + Mock::given(method("POST")) + .and(path("/mcp")) + .respond_with(move |request: &Request| { + let body: Value = request.body_json().expect("JSON-RPC request"); + match body["method"].as_str() { + Some("server/discover" | "initialize") => { + let initial = observed_handshakes.fetch_add(1, Ordering::SeqCst) == 0; + let supports_policy = if initial { + initial_support + } else { + recovered_support + }; + let mut capabilities = json!({"tools": {}}); + if supports_policy { + capabilities["experimental"] = json!({"codex/network-policy-meta": {}}); + } + let mut result = if body["method"] == "server/discover" { + modern_discover_result(&body) + } else { + json!({"jsonrpc": "2.0", "id": body["id"], "result": { + "protocolVersion": LEGACY_VERSION, + "serverInfo": {"name": "recovery-test", "version": "1.0"}, + }}) + }; + result["result"]["capabilities"] = capabilities; + ResponseTemplate::new(200) + .insert_header( + "mcp-session-id", + if initial { "initial" } else { "recovered" }, + ) + .set_body_json(result) + } + Some("notifications/initialized") => ResponseTemplate::new(202), + Some("tools/call") => { + // The first call succeeds; the next expires the old session. + if observed_calls.fetch_add(1, Ordering::SeqCst) == 1 { + return ResponseTemplate::new(404); + } + ResponseTemplate::new(200).set_body_json(json!({ + "jsonrpc": "2.0", "id": body["id"], "result": { + "resultType": "complete", "content": [], + }, + })) + } + other => panic!("unexpected request: {other:?}"), + } + }) + .mount(&server) + .await; + let client = create_client(&server, McpProtocolMode::Legacy) + .await? + .with_read_only_tools(/*requires_read_only_tools*/ true); + initialize_client(&client).await?; + for phase in ["warmup", "recovery", "later"] { + let result = client + .call_tool( + "echo".to_string(), + Some(json!({"message": phase})), + |capabilities| { + let mut meta = json!({"callId": phase, "openai/readOnly": false}); + if capabilities + .experimental + .as_ref() + .is_some_and(|experimental| { + experimental.contains_key("codex/network-policy-meta") + }) + { + anyhow::ensure!(!fail_policy, "policy source unavailable"); + meta["codex/network-policy-meta"] = json!({"enabled": false}); + } + Ok(Some(meta)) + }, + /*timeout*/ None, + ) + .await; + if fail_policy && phase != "warmup" { + assert!( + result + .unwrap_err() + .to_string() + .contains("policy source unavailable") + ); + } else { + result?; + } + } + let requests = server.received_requests().await.expect("recorded requests") + .into_iter() + .filter(|request| request.method == "POST") + .filter_map(|request| { + let body: Value = request.body_json().expect("JSON-RPC request"); + (body["method"] == "tools/call").then(|| { + let meta = &body["params"]["_meta"]; + json!({ + "session": request.headers.get("mcp-session-id").map(|value| value.to_str().unwrap()), + "callId": meta["callId"], + "policy": meta["codex/network-policy-meta"], + "readOnly": meta["openai/readOnly"], + }) + }) + }).collect::>(); + let initial_policy = initial_support.then(|| json!({"enabled": false})); + let recovered_policy = recovered_support.then(|| json!({"enabled": false})); + let mut expected = vec![ + json!({"session": "initial", "callId": "warmup", "policy": initial_policy, "readOnly": true}), + json!({"session": "initial", "callId": "recovery", "policy": initial_policy, "readOnly": true}), + ]; + if !fail_policy { + expected.extend([ + json!({"session": "recovered", "callId": "recovery", "policy": recovered_policy, "readOnly": true}), + json!({"session": "recovered", "callId": "later", "policy": recovered_policy, "readOnly": true}), + ]); + } + assert_eq!(requests, expected); + assert_eq!(handshakes.load(Ordering::SeqCst), 2); + client.shutdown().await; + } + Ok(()) +} diff --git a/codex-rs/rmcp-client/tests/mcp_2026_mrtr.rs b/codex-rs/rmcp-client/tests/mcp_2026_mrtr.rs index 887139ea00..5890e3f753 100644 --- a/codex-rs/rmcp-client/tests/mcp_2026_mrtr.rs +++ b/codex-rs/rmcp-client/tests/mcp_2026_mrtr.rs @@ -295,7 +295,7 @@ async fn modern_tool_mrtr_drives_form_and_url_elicitation_and_preserves_metadata .call_tool( "confirm".into(), Some(json!({"mode": mode})), - Some(json!({"requestContext": "caller-context"})), + |_| Ok(Some(json!({"requestContext": "caller-context"}))), Some(Duration::from_secs(5)), ) .await?; @@ -460,7 +460,11 @@ async fn modern_tool_mrtr_uses_recovered_protocol_after_legacy_session_expiry() .call_tool( "confirm".into(), Some(json!({})), - Some(json!({"requestContext": "caller-context", "openai/readOnly": false})), + |_| { + Ok(Some( + json!({"requestContext": "caller-context", "openai/readOnly": false}), + )) + }, Some(Duration::from_secs(5)), ) .await?; @@ -639,7 +643,7 @@ async fn modern_tool_mrtr_accepts_sse_discovery_and_metadata_bearing_rounds() -> .call_tool( "confirm".to_owned(), Some(json!({})), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(5)), ) .await?; @@ -715,7 +719,7 @@ async fn modern_tool_mrtr_preserves_state_only_rounds_without_elicitation() -> a .call_tool( "state-only".into(), Some(json!({})), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(5)), ) .await?; diff --git a/codex-rs/rmcp-client/tests/mcp_2026_mrtr/native_verification_tests.rs b/codex-rs/rmcp-client/tests/mcp_2026_mrtr/native_verification_tests.rs index cd4959d979..705867acef 100644 --- a/codex-rs/rmcp-client/tests/mcp_2026_mrtr/native_verification_tests.rs +++ b/codex-rs/rmcp-client/tests/mcp_2026_mrtr/native_verification_tests.rs @@ -79,7 +79,7 @@ async fn call(client: &RmcpClient) -> anyhow::Result { .call_tool( "verified_operation".into(), Some(json!({"resource": "example"})), - Some(json!({"caller": "preserved"})), + |_| Ok(Some(json!({"caller": "preserved"}))), Some(Duration::from_secs(/*secs*/ 5)), ) .await?, diff --git a/codex-rs/rmcp-client/tests/mcp_2026_stdio.rs b/codex-rs/rmcp-client/tests/mcp_2026_stdio.rs index 9658b81ecd..20f1c4eb45 100644 --- a/codex-rs/rmcp-client/tests/mcp_2026_stdio.rs +++ b/codex-rs/rmcp-client/tests/mcp_2026_stdio.rs @@ -170,7 +170,7 @@ async fn exercise_stdio_server( .call_tool( "echo".to_owned(), Some(json!({"message": "hello stdio"})), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(5)), ) .await?; diff --git a/codex-rs/rmcp-client/tests/mcp_2026_stdio_discovery.rs b/codex-rs/rmcp-client/tests/mcp_2026_stdio_discovery.rs index 8cf62a2697..da2f02cedf 100644 --- a/codex-rs/rmcp-client/tests/mcp_2026_stdio_discovery.rs +++ b/codex-rs/rmcp-client/tests/mcp_2026_stdio_discovery.rs @@ -130,7 +130,7 @@ async fn local_stdio_inherits_ca_certificate_variables_child() -> anyhow::Result .call_tool( "echo".to_string(), Some(json!({ "message": "ca inheritance", "env_var": env_name })), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(10)), ) .await?; @@ -266,7 +266,7 @@ async fn legacy_stdio_preserves_existing_protocol_marker_environment() -> anyhow "message": "legacy environment", "env_var": "CODEX_MCP_PROTOCOL_VERSION", })), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(10)), ) .await?; diff --git a/codex-rs/rmcp-client/tests/process_group_cleanup.rs b/codex-rs/rmcp-client/tests/process_group_cleanup.rs index 704310583a..8da66c3b9b 100644 --- a/codex-rs/rmcp-client/tests/process_group_cleanup.rs +++ b/codex-rs/rmcp-client/tests/process_group_cleanup.rs @@ -165,7 +165,7 @@ async fn shutdown_kills_initialized_stdio_server_with_in_flight_operation() -> R .call_tool( "sync".to_string(), Some(json!({ "sleep_after_ms": 300_000 })), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(300)), ) .await diff --git a/codex-rs/rmcp-client/tests/stdio_descriptor_boundary.rs b/codex-rs/rmcp-client/tests/stdio_descriptor_boundary.rs index 988b75fe50..b082d737e6 100644 --- a/codex-rs/rmcp-client/tests/stdio_descriptor_boundary.rs +++ b/codex-rs/rmcp-client/tests/stdio_descriptor_boundary.rs @@ -140,7 +140,7 @@ for line in sys.stdin: .call_tool( "probe".to_string(), /*arguments*/ None, - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(/*secs*/ 5)), ) .await?; diff --git a/codex-rs/rmcp-client/tests/streamable_http_oauth_startup.rs b/codex-rs/rmcp-client/tests/streamable_http_oauth_startup.rs index 57604d3bad..ddb85046d2 100644 --- a/codex-rs/rmcp-client/tests/streamable_http_oauth_startup.rs +++ b/codex-rs/rmcp-client/tests/streamable_http_oauth_startup.rs @@ -230,7 +230,7 @@ async fn oauth_tool_call_child() -> anyhow::Result<()> { .call_tool( "probe".to_string(), /*arguments*/ None, - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(/*secs*/ 5)), ) .await?; diff --git a/codex-rs/rmcp-client/tests/streamable_http_oauth_tool_call.rs b/codex-rs/rmcp-client/tests/streamable_http_oauth_tool_call.rs index c620410d14..2db981f93c 100644 --- a/codex-rs/rmcp-client/tests/streamable_http_oauth_tool_call.rs +++ b/codex-rs/rmcp-client/tests/streamable_http_oauth_tool_call.rs @@ -312,7 +312,7 @@ async fn runtime_oauth_child() -> anyhow::Result<()> { .call_tool( "echo".to_string(), /*arguments*/ None, - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(5)), ) .await; diff --git a/codex-rs/rmcp-client/tests/streamable_http_test_support.rs b/codex-rs/rmcp-client/tests/streamable_http_test_support.rs index e7d6e2c52f..4b32bf3df9 100644 --- a/codex-rs/rmcp-client/tests/streamable_http_test_support.rs +++ b/codex-rs/rmcp-client/tests/streamable_http_test_support.rs @@ -179,7 +179,7 @@ pub(crate) async fn call_echo_tool( .call_tool( "echo".to_string(), Some(json!({ "message": message })), - /*meta*/ None, + |_| Ok(None), Some(Duration::from_secs(5)), ) .await