openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support threadless MCP connections without event channels (#34408)

Browse Source 
## 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
This commit is contained in:
stevenlee-oai
2026-07-20 20:43:36 +00:00
committed by copyberry
parent b7e39aa316
commit 19940967bd
14 changed files with 185 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
-1
View File
@@ -1952,7 +1952,6 @@ version = "0.0.0"
dependencies = [
"anyhow",
"app_test_support",
"async-channel",
"axum",
"base64 0.22.1",
"chrono",
codex-rs/app-server/Cargo.toml
-1
View File
@@ -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",
codex-rs/app-server/src/request_processors/apps_processor/installed.rs
+1 -4
View File
@@ -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,
codex-rs/app-server/tests/suite/v2/mcp_resource.rs
+64 -2
View File
@@ -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 <hosted> orchestrator.";
const SKILL_DESCRIPTION: &str = "Deploy through the &lt;hosted&gt; 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::<McpResourceReadResponse>(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<RoleServer>,
context: RequestContext<RoleServer>,
) -> Result<ReadResourceResult, rmcp::ErrorData> {
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![
codex-rs/codex-mcp/src/connection_manager.rs
+94 -83
View File
@@ -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<String, EffectiveMcpServer>,
@@ -131,7 +133,7 @@ impl McpConnectionManager {
keyring_backend_kind: AuthKeyringBackendKind,
approval_policy: &Constrained<AskForApproval>,
submit_id: String,
tx_event: Sender<Event>,
tx_event: Option<Sender<Event>>,
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
}
codex-rs/codex-mcp/src/connection_manager_tests.rs
+5 -7
View File
@@ -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(
codex-rs/codex-mcp/src/elicitation.rs
+9 -1
View File
@@ -163,7 +163,7 @@ impl ElicitationRequestManager {
pub(crate) fn make_sender(
&self,
server_name: String,
tx_event: Sender<Event>,
tx_event: Option<Sender<Event>>,
) -> 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)
codex-rs/codex-mcp/src/mcp/mod.rs
+2 -8
View File
@@ -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<ReadResourceResult> {
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,
codex-rs/codex-mcp/src/rmcp_client.rs
+5 -5
View File
@@ -181,9 +181,9 @@ impl CodexAppsStartupReconnect {
mut self,
submit_id: String,
server_name: String,
tx_event: Sender<Event>,
tx_event: Option<Sender<Event>>,
) -> 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<Event>,
tx_event: Option<Sender<Event>>,
elicitation_requests: ElicitationRequestManager,
codex_apps_tools_cache_context: Option<ConnectorRuntimeContext<ToolInfo>>,
tool_catalog_cache_context: Option<McpToolCatalogCacheContext>,
@@ -423,7 +423,7 @@ impl AsyncManagedClient {
store_mode: OAuthCredentialsStoreMode,
keyring_backend_kind: AuthKeyringBackendKind,
cancel_token: CancellationToken,
tx_event: Sender<Event>,
tx_event: Option<Sender<Event>>,
elicitation_requests: ElicitationRequestManager,
codex_apps_tools_cache_context: Option<ConnectorRuntimeContext<ToolInfo>>,
tool_catalog_cache_context: Option<McpToolCatalogCacheContext>,
@@ -1003,7 +1003,7 @@ struct StartServerTaskParams {
startup_timeout: Option<Duration>, // TODO: cancel_token should handle this.
tool_timeout: Duration,
tool_filter: ToolFilter,
tx_event: Sender<Event>,
tx_event: Option<Sender<Event>>,
elicitation_requests: ElicitationRequestManager,
codex_apps_tools_cache_context: Option<ConnectorRuntimeContext<ToolInfo>>,
tool_catalog_cache_context: Option<McpToolCatalogCacheContext>,
codex-rs/core/src/connectors.rs
+1 -5
View File
@@ -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
codex-rs/core/src/mcp_tool_call_tests.rs
+1 -2
View File
@@ -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(
codex-rs/core/src/session/mcp.rs
+1 -1
View File
@@ -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(),
codex-rs/core/src/session/session.rs
+1 -1
View File
@@ -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(),
codex-rs/core/tests/suite/mcp_auth_refresh.rs
+1 -3
View File
@@ -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(