From 19940967bdb5ac04aec5d08ebd465481f1ac964d Mon Sep 17 00:00:00 2001 From: stevenlee-oai Date: Mon, 20 Jul 2026 20:39:07 +0000 Subject: [PATCH] Support threadless MCP connections without event channels (#34408) ## What changed - Make the MCP connection manager's event sender optional. - Skip startup notifications when a caller has no session event stream. - Decline interactive MCP elicitations in that mode while allowing the underlying operation to continue. - Update threadless resource reads, status snapshots, and connector discovery to use the channel-free path. ## Testing - Extend the app-server MCP resource test to verify that a threadless resource read declines an elicitation and still returns the resource contents. GitOrigin-RevId: 65fea152adbc02a78e09a40cae4134b6dccffd34 --- codex-rs/Cargo.lock | 1 - codex-rs/app-server/Cargo.toml | 1 - .../apps_processor/installed.rs | 5 +- .../app-server/tests/suite/v2/mcp_resource.rs | 66 ++++++- codex-rs/codex-mcp/src/connection_manager.rs | 177 ++++++++++-------- .../codex-mcp/src/connection_manager_tests.rs | 12 +- codex-rs/codex-mcp/src/elicitation.rs | 10 +- codex-rs/codex-mcp/src/mcp/mod.rs | 10 +- codex-rs/codex-mcp/src/rmcp_client.rs | 10 +- codex-rs/core/src/connectors.rs | 6 +- codex-rs/core/src/mcp_tool_call_tests.rs | 3 +- codex-rs/core/src/session/mcp.rs | 2 +- codex-rs/core/src/session/session.rs | 2 +- codex-rs/core/tests/suite/mcp_auth_refresh.rs | 4 +- 14 files changed, 185 insertions(+), 124 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index aaeeb3328c..8669268591 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1952,7 +1952,6 @@ version = "0.0.0" dependencies = [ "anyhow", "app_test_support", - "async-channel", "axum", "base64 0.22.1", "chrono", diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index 201069c10e..27eee2b837 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -26,7 +26,6 @@ workspace = true [dependencies] anyhow = { workspace = true } -async-channel = { workspace = true } base64 = { workspace = true } axum = { workspace = true, default-features = false, features = [ "http1", 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 9837f51b0f..21b260bb28 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 @@ -1,6 +1,5 @@ use super::*; -use async_channel::unbounded; use codex_connectors::ConnectorRuntimeTool; use codex_connectors::connector_runtime_context_key; use codex_connectors::connector_tool_is_synthetic; @@ -75,8 +74,6 @@ impl AppsRequestProcessor { self.thread_manager.environment_manager(), config.cwd.to_path_buf(), ); - let (tx_event, rx_event) = unbounded(); - drop(rx_event); let cancellation_token = CancellationToken::new(); let codex_apps_auth_manager = host_owned_codex_apps_enabled(&mcp_config, auth.as_ref()) @@ -87,7 +84,7 @@ impl AppsRequestProcessor { config.auth_keyring_backend_kind(), &config.permissions.approval_policy, APPS_INSTALLED_SUBMIT_ID.to_string(), - tx_event, + /*tx_event*/ None, cancellation_token.clone(), PermissionProfile::default(), runtime_context, diff --git a/codex-rs/app-server/tests/suite/v2/mcp_resource.rs b/codex-rs/app-server/tests/suite/v2/mcp_resource.rs index cd4efee8ab..f968084982 100644 --- a/codex-rs/app-server/tests/suite/v2/mcp_resource.rs +++ b/codex-rs/app-server/tests/suite/v2/mcp_resource.rs @@ -34,9 +34,15 @@ use codex_protocol::protocol::SessionSource; use core_test_support::responses; use pretty_assertions::assert_eq; use rmcp::handler::server::ServerHandler; +use rmcp::model::BooleanSchema; +use rmcp::model::CreateElicitationRequestParams; +use rmcp::model::CreateElicitationResult; +use rmcp::model::ElicitationAction; +use rmcp::model::ElicitationSchema; use rmcp::model::ListResourcesResult; use rmcp::model::Meta; use rmcp::model::PaginatedRequestParams; +use rmcp::model::PrimitiveSchema; use rmcp::model::ProtocolVersion; use rmcp::model::RawResource; use rmcp::model::ReadResourceRequestParams; @@ -61,6 +67,8 @@ const TEST_RESOURCE_URI: &str = "test://codex/resource"; const TEST_BLOB_RESOURCE_URI: &str = "test://codex/resource.bin"; const TEST_RESOURCE_BLOB: &str = "YmluYXJ5LXJlc291cmNl"; const TEST_RESOURCE_TEXT: &str = "Resource body from the MCP server."; +const TEST_ELICITATION_RESOURCE_URI: &str = "test://codex/elicitation"; +const TEST_ELICITATION_RESOURCE_TEXT: &str = "Threadless elicitation was declined."; const SKILL_NAME: &str = "demo-plugin:deploy"; const RAW_SKILL_DESCRIPTION: &str = "Deploy\nthrough the orchestrator."; const SKILL_DESCRIPTION: &str = "Deploy through the <hosted> orchestrator."; @@ -539,7 +547,8 @@ enabled = false } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn mcp_resource_read_returns_resource_contents_without_thread() -> Result<()> { +async fn mcp_resource_read_returns_contents_and_declines_elicitation_without_thread() -> Result<()> +{ let (apps_server_url, _apps_server_calls, apps_server_handle) = start_resource_apps_mcp_server().await?; @@ -550,6 +559,7 @@ async fn mcp_resource_read_returns_resource_contents_without_thread() -> Result< r#" chatgpt_base_url = "{apps_server_url}" mcp_oauth_credentials_store = "file" +approval_policy = "on-request" [features] apps = true @@ -590,6 +600,30 @@ apps = true expected_resource_read_response() ); + let read_request_id = mcp + .send_mcp_resource_read_request(McpResourceReadParams { + thread_id: None, + server: "codex_apps".to_string(), + uri: TEST_ELICITATION_RESOURCE_URI.to_string(), + }) + .await?; + let read_response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(read_request_id)), + ) + .await??; + assert_eq!( + to_response::(read_response)?, + McpResourceReadResponse { + contents: vec![McpResourceContent::Text { + uri: TEST_ELICITATION_RESOURCE_URI.to_string(), + mime_type: Some("text/plain".to_string()), + text: TEST_ELICITATION_RESOURCE_TEXT.to_string(), + meta: None, + }], + } + ); + apps_server_handle.abort(); let _ = apps_server_handle.await; Ok(()) @@ -862,9 +896,37 @@ impl ServerHandler for ResourceAppsMcpServer { async fn read_resource( &self, request: ReadResourceRequestParams, - _context: RequestContext, + context: RequestContext, ) -> Result { let uri = request.uri; + if uri == TEST_ELICITATION_RESOURCE_URI { + let requested_schema = ElicitationSchema::builder() + .required_property("confirmed", PrimitiveSchema::Boolean(BooleanSchema::new())) + .build() + .map_err(|err| rmcp::ErrorData::internal_error(err.to_string(), None))?; + let result = context + .peer + .create_elicitation(CreateElicitationRequestParams::FormElicitationParams { + meta: None, + message: "Confirm the resource read.".to_string(), + requested_schema, + }) + .await + .map_err(|err| rmcp::ErrorData::internal_error(err.to_string(), None))?; + assert_eq!( + result, + CreateElicitationResult::new(ElicitationAction::Decline) + ); + + return Ok(ReadResourceResult::new(vec![ + ResourceContents::TextResourceContents { + uri: TEST_ELICITATION_RESOURCE_URI.to_string(), + mime_type: Some("text/plain".to_string()), + text: TEST_ELICITATION_RESOURCE_TEXT.to_string(), + meta: None, + }, + ])); + } if uri == SKILL_MAIN_PROMPT_URI { self.calls.main_prompt_reads.fetch_add(1, Ordering::Relaxed); return Ok(ReadResourceResult::new(vec![ diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index eaf0719dfd..58e1d3b23d 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -124,6 +124,8 @@ pub struct McpConnectionManager { } impl McpConnectionManager { + /// Creates an MCP connection manager. Threadless callers can pass no `tx_event`; startup + /// notifications are then skipped and interactive elicitations are declined. #[allow(clippy::too_many_arguments)] pub async fn new( mcp_servers: &HashMap, @@ -131,7 +133,7 @@ impl McpConnectionManager { keyring_backend_kind: AuthKeyringBackendKind, approval_policy: &Constrained, submit_id: String, - tx_event: Sender, + tx_event: Option>, startup_cancellation_token: CancellationToken, initial_permission_profile: PermissionProfile, runtime_context: McpRuntimeContext, @@ -182,15 +184,17 @@ impl McpConnectionManager { { server_metadata.insert(server_name.clone(), McpServerMetadata::from(&server)); let cancel_token = startup_cancellation_token.child_token(); - let _ = emit_update( - startup_submit_id.as_str(), - &tx_event, - McpStartupUpdateEvent { - server: server_name.clone(), - status: McpStartupStatus::Starting, - }, - ) - .await; + if let Some(tx_event) = tx_event.as_ref() { + let _ = emit_update( + startup_submit_id.as_str(), + tx_event, + McpStartupUpdateEvent { + server: server_name.clone(), + status: McpStartupStatus::Starting, + }, + ) + .await; + } let configured_config = server.configured_config().cloned(); let resolved_environment = configured_config.as_ref().map_or_else( || Ok(None), @@ -277,68 +281,73 @@ impl McpConnectionManager { if cancel_token.is_cancelled() { outcome = Err(StartupOutcomeError::Cancelled); } - let auth_state = match &outcome { - Err(error) if error.is_authentication_required() && !has_runtime_auth => { - configured_config.as_ref().and_then(|config| { - let McpServerTransportConfig::StreamableHttp { - url, - bearer_token_env_var, - http_headers, - env_http_headers, - } = &config.transport - else { - return None; - }; - match determine_streamable_http_auth_status_from_credentials( - &server_name, - url, - bearer_token_env_var.as_deref(), - http_headers.clone(), - env_http_headers.clone(), - store_mode, - keyring_backend_kind, - ) { - Ok(auth_state) => auth_state, - Err(error) => { - warn!( - "failed to read stored auth status for MCP server `{server_name}`: {error:?}" - ); - None + if let Some(tx_event) = tx_event.as_ref() { + let auth_state = match &outcome { + Err(error) if error.is_authentication_required() && !has_runtime_auth => { + configured_config.as_ref().and_then(|config| { + let McpServerTransportConfig::StreamableHttp { + url, + bearer_token_env_var, + http_headers, + env_http_headers, + } = &config.transport + else { + return None; + }; + match determine_streamable_http_auth_status_from_credentials( + &server_name, + url, + bearer_token_env_var.as_deref(), + http_headers.clone(), + env_http_headers.clone(), + store_mode, + keyring_backend_kind, + ) { + Ok(auth_state) => auth_state, + Err(error) => { + warn!( + "failed to read stored auth status for MCP server `{server_name}`: {error:?}" + ); + None + } } - } - }) + }) + } + Ok(_) | Err(_) => None, + }; + if cancel_token.is_cancelled() { + outcome = Err(StartupOutcomeError::Cancelled); } - Ok(_) | Err(_) => None, - }; + let status = match &outcome { + Ok(_) => McpStartupStatus::Ready, + Err(StartupOutcomeError::Cancelled) => McpStartupStatus::Cancelled, + Err(error) => { + let reason = mcp_startup_failure_reason(auth_state, error); + let error_str = mcp_init_error_display( + server_name.as_str(), + configured_config.as_ref(), + error, + ); + McpStartupStatus::Failed { + error: error_str, + reason, + } + } + }; + + let _ = emit_update( + submit_id.as_str(), + tx_event, + McpStartupUpdateEvent { + server: server_name.clone(), + status, + }, + ) + .await; + } if cancel_token.is_cancelled() { outcome = Err(StartupOutcomeError::Cancelled); } - let status = match &outcome { - Ok(_) => McpStartupStatus::Ready, - Err(StartupOutcomeError::Cancelled) => McpStartupStatus::Cancelled, - Err(error) => { - let reason = mcp_startup_failure_reason(auth_state, error); - let error_str = mcp_init_error_display( - server_name.as_str(), - configured_config.as_ref(), - error, - ); - McpStartupStatus::Failed { - error: error_str, - reason, - } - } - }; - - let _ = emit_update( - submit_id.as_str(), - &tx_event, - McpStartupUpdateEvent { - server: server_name.clone(), - status, - }, - ) - .await; if matches!(&outcome, Err(StartupOutcomeError::Failed { .. })) { async_managed_client.reconnect_failed_startup().await; @@ -358,25 +367,27 @@ impl McpConnectionManager { }; tokio::spawn(async move { let outcomes = join_set.join_all().await; - let mut summary = McpStartupCompleteEvent::default(); - for (server_name, outcome) in outcomes { - match outcome { - Ok(_) => summary.ready.push(server_name), - Err(StartupOutcomeError::Cancelled) => summary.cancelled.push(server_name), - Err(StartupOutcomeError::Failed { error, .. }) => { - summary.failed.push(McpStartupFailure { - server: server_name, - error, - }) + if let Some(tx_event) = tx_event { + let mut summary = McpStartupCompleteEvent::default(); + for (server_name, outcome) in outcomes { + match outcome { + Ok(_) => summary.ready.push(server_name), + Err(StartupOutcomeError::Cancelled) => summary.cancelled.push(server_name), + Err(StartupOutcomeError::Failed { error, .. }) => { + summary.failed.push(McpStartupFailure { + server: server_name, + error, + }) + } } } + let _ = tx_event + .send(Event { + id: startup_submit_id, + msg: EventMsg::McpStartupComplete(summary), + }) + .await; } - let _ = tx_event - .send(Event { - id: startup_submit_id, - msg: EventMsg::McpStartupComplete(summary), - }) - .await; }); manager } diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index e881d9bad9..18e8afe86e 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -260,7 +260,7 @@ async fn disabled_permissions_auto_accept_elicitation_with_empty_form_schema() { ElicitationRequestRouter::default(), ); let (tx_event, _rx_event) = async_channel::bounded(1); - let sender = manager.make_sender("server".to_string(), tx_event); + let sender = manager.make_sender("server".to_string(), Some(tx_event)); let response = sender( NumberOrString::Number(1), @@ -297,7 +297,7 @@ async fn disabled_permissions_do_not_auto_accept_elicitation_with_requested_fiel ElicitationRequestRouter::default(), ); let (tx_event, _rx_event) = async_channel::bounded(1); - let sender = manager.make_sender("server".to_string(), tx_event); + let sender = manager.make_sender("server".to_string(), Some(tx_event)); let response = sender( NumberOrString::Number(1), @@ -362,8 +362,8 @@ async fn shared_elicitation_router_targets_the_exact_pending_request() { router, ); let (tx_event, rx_event) = async_channel::bounded(2); - let sender_a = manager_a.make_sender("server".to_string(), tx_event.clone()); - let sender_b = manager_b.make_sender("server".to_string(), tx_event); + let sender_a = manager_a.make_sender("server".to_string(), Some(tx_event.clone())); + let sender_b = manager_b.make_sender("server".to_string(), Some(tx_event)); let elicitation = codex_rmcp_client::Elicitation::Mcp( CreateElicitationRequestParams::FormElicitationParams { meta: None, @@ -1539,8 +1539,6 @@ fn host_owned_codex_apps_matches_reserved_name_with_server_metadata() { #[tokio::test] async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() { let approval_policy = Constrained::allow_any(AskForApproval::OnRequest); - let (tx_event, rx_event) = async_channel::unbounded(); - drop(rx_event); let codex_home = tempdir().expect("tempdir"); let mcp_servers = HashMap::from([ ( @@ -1605,7 +1603,7 @@ async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() { AuthKeyringBackendKind::default(), &approval_policy, String::new(), - tx_event, + /*tx_event*/ None, cancel_token.clone(), PermissionProfile::default(), McpRuntimeContext::new( diff --git a/codex-rs/codex-mcp/src/elicitation.rs b/codex-rs/codex-mcp/src/elicitation.rs index b6bb7fa791..de4ce1448c 100644 --- a/codex-rs/codex-mcp/src/elicitation.rs +++ b/codex-rs/codex-mcp/src/elicitation.rs @@ -163,7 +163,7 @@ impl ElicitationRequestManager { pub(crate) fn make_sender( &self, server_name: String, - tx_event: Sender, + tx_event: Option>, ) -> SendElicitation { let router = self.router.clone(); let approval_policy = self.approval_policy.clone(); @@ -233,6 +233,14 @@ impl ElicitationRequestManager { } } + let Some(tx_event) = tx_event else { + return Ok(ElicitationResponse { + action: ElicitationAction::Decline, + content: None, + meta: None, + }); + }; + let public_request_id = format!( "codex-mcp-elicitation-{}", NEXT_ELICITATION_REQUEST_ID.fetch_add(1, Ordering::Relaxed) diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index 8026703124..97f5f7061f 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -19,7 +19,6 @@ use std::env; use std::path::PathBuf; use std::time::Duration; -use async_channel::unbounded; use codex_config::Constrained; use codex_config::McpServerAuth; use codex_config::McpServerConfig; @@ -311,8 +310,6 @@ pub async fn read_mcp_resource( ) -> anyhow::Result { let mut mcp_servers = effective_mcp_servers(config, auth); mcp_servers.retain(|name, _| name == server); - let (tx_event, rx_event) = unbounded(); - drop(rx_event); let cancel_token = CancellationToken::new(); let manager = McpConnectionManager::new( &mcp_servers, @@ -320,7 +317,7 @@ pub async fn read_mcp_resource( config.auth_keyring_backend_kind, &config.approval_policy, String::new(), - tx_event, + /*tx_event*/ None, cancel_token.clone(), PermissionProfile::default(), runtime_context, @@ -390,9 +387,6 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( let server_names = mcp_servers.keys().cloned().collect(); - let (tx_event, rx_event) = unbounded(); - drop(rx_event); - let cancel_token = CancellationToken::new(); let mcp_connection_manager = McpConnectionManager::new( &mcp_servers, @@ -400,7 +394,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( config.auth_keyring_backend_kind, &config.approval_policy, submit_id, - tx_event, + /*tx_event*/ None, cancel_token.clone(), PermissionProfile::default(), runtime_context, diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 565549ef0a..aa34b47cf6 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -181,9 +181,9 @@ impl CodexAppsStartupReconnect { mut self, submit_id: String, server_name: String, - tx_event: Sender, + tx_event: Option>, ) -> Self { - self.startup_status_context = Some(CodexAppsStartupStatusContext { + self.startup_status_context = tx_event.map(|tx_event| CodexAppsStartupStatusContext { submit_id, server_name, tx_event, @@ -277,7 +277,7 @@ struct ManagedClientStartup { server: EffectiveMcpServer, store_mode: OAuthCredentialsStoreMode, keyring_backend_kind: AuthKeyringBackendKind, - tx_event: Sender, + tx_event: Option>, elicitation_requests: ElicitationRequestManager, codex_apps_tools_cache_context: Option>, tool_catalog_cache_context: Option, @@ -423,7 +423,7 @@ impl AsyncManagedClient { store_mode: OAuthCredentialsStoreMode, keyring_backend_kind: AuthKeyringBackendKind, cancel_token: CancellationToken, - tx_event: Sender, + tx_event: Option>, elicitation_requests: ElicitationRequestManager, codex_apps_tools_cache_context: Option>, tool_catalog_cache_context: Option, @@ -1003,7 +1003,7 @@ struct StartServerTaskParams { startup_timeout: Option, // TODO: cancel_token should handle this. tool_timeout: Duration, tool_filter: ToolFilter, - tx_event: Sender, + tx_event: Option>, elicitation_requests: ElicitationRequestManager, codex_apps_tools_cache_context: Option>, tool_catalog_cache_context: Option, diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 56b4623775..a7e2881d75 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -5,7 +5,6 @@ use std::sync::Mutex as StdMutex; use std::time::Duration; use std::time::Instant; -use async_channel::unbounded; pub use codex_connectors::AppBranding; pub use codex_connectors::AppInfo; pub use codex_connectors::AppMetadata; @@ -235,9 +234,6 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( let runtime_context = McpRuntimeContext::new(Arc::clone(&environment_manager), config.cwd.to_path_buf()); - let (tx_event, rx_event) = unbounded(); - drop(rx_event); - let cancel_token = CancellationToken::new(); let codex_apps_auth_manager = codex_mcp::host_owned_codex_apps_enabled(&mcp_config, auth.as_ref()) @@ -248,7 +244,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( config.auth_keyring_backend_kind(), &config.permissions.approval_policy, INITIAL_SUBMIT_ID.to_owned(), - tx_event, + /*tx_event*/ None, cancel_token.clone(), PermissionProfile::default(), // Connector discovery is threadless. Use an actually configured env if diff --git a/codex-rs/core/src/mcp_tool_call_tests.rs b/codex-rs/core/src/mcp_tool_call_tests.rs index 2699d6057d..f4beb97b72 100644 --- a/codex-rs/core/src/mcp_tool_call_tests.rs +++ b/codex-rs/core/src/mcp_tool_call_tests.rs @@ -1449,7 +1449,6 @@ async fn host_owned_codex_apps_manager( let auth = session.services.auth_manager.auth().await; let startup_cancellation_token = CancellationToken::new(); startup_cancellation_token.cancel(); - let (tx_event, _rx_event) = async_channel::unbounded(); let mcp_servers = HashMap::from([( CODEX_APPS_MCP_SERVER_NAME.to_string(), codex_mcp::EffectiveMcpServer::configured(codex_mcp::codex_apps_mcp_server_config( @@ -1464,7 +1463,7 @@ async fn host_owned_codex_apps_manager( turn_context.config.auth_keyring_backend_kind(), &turn_context.approval_policy, turn_context.sub_id.clone(), - tx_event, + /*tx_event*/ None, startup_cancellation_token, turn_context.permission_profile(), codex_mcp::McpRuntimeContext::new( diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index 157391c390..d88459a25b 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -451,7 +451,7 @@ impl Session { mcp_config.auth_keyring_backend_kind, &turn_context.approval_policy, turn_context.sub_id.clone(), - self.get_tx_event(), + Some(self.get_tx_event()), mcp_startup_cancellation_token, turn_context.permission_profile(), mcp_runtime_context.clone(), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 2c81b59720..5b136c0492 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1218,7 +1218,7 @@ impl Session { config.auth_keyring_backend_kind(), &session_configuration.approval_policy, INITIAL_SUBMIT_ID.to_owned(), - tx_event.clone(), + Some(tx_event.clone()), mcp_startup_cancellation_token, session_configuration.permission_profile(), mcp_runtime_context.clone(), diff --git a/codex-rs/core/tests/suite/mcp_auth_refresh.rs b/codex-rs/core/tests/suite/mcp_auth_refresh.rs index a8bd10cfc6..822d080524 100644 --- a/codex-rs/core/tests/suite/mcp_auth_refresh.rs +++ b/codex-rs/core/tests/suite/mcp_auth_refresh.rs @@ -83,8 +83,6 @@ async fn hosted_plugin_runtime_ps_mcp_tool_calls_use_current_auth_manager_token( CODEX_APPS_MCP_SERVER_NAME.to_string(), EffectiveMcpServer::configured(hosted_plugin_runtime_config), )]); - let (tx_event, rx_event) = async_channel::unbounded(); - drop(rx_event); let approval_policy = Constrained::allow_any(AskForApproval::Never); let manager = McpConnectionManager::new( &mcp_servers, @@ -92,7 +90,7 @@ async fn hosted_plugin_runtime_ps_mcp_tool_calls_use_current_auth_manager_token( AuthKeyringBackendKind::default(), &approval_policy, "test".to_string(), - tx_event, + /*tx_event*/ None, CancellationToken::new(), PermissionProfile::default(), McpRuntimeContext::new(