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

Add app-server MCP event streaming (#39761)

Browse Source 
## What changed

- Add experimental `mcpServer/event/stream/start` and `mcpServer/event/stream/stop` requests for hosted apps, plus `mcpServer/event/stream/notification` forwarding.
- Scope subscriptions to the owning app-server connection and subscribed thread, enforce unique IDs and a per-connection limit, and clean them up when the thread is unsubscribed or the connection closes.
- Wait for the MCP active notification before completing startup, retry streams that close immediately, and terminate them when authentication or hosted runtime ownership changes.

## Testing

- Add an app-server integration test covering activation, event forwarding, duplicate subscription rejection, and explicit cancellation.

GitOrigin-RevId: 8a6fc1615adfc5af7e67def3b824fa5909ab3e7b
This commit is contained in:
victor-openai
2026-08-20 17:49:35 +00:00
committed by copyberry
parent 8a40095ea3
commit 097825f75a
30 changed files with 1043 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+48
View File
@@ -2472,6 +2472,34 @@
"LegacyAppPathString": {
"type": "string"
},
"McpServerEventNotification": {
"properties": {
"method": {
"type": "string"
},
"params": true
},
"required": [
"method",
"params"
],
"type": "object"
},
"McpServerEventStreamNotification": {
"properties": {
"notification": {
"$ref": "#/definitions/McpServerEventNotification"
},
"subscriptionId": {
"type": "string"
}
},
"required": [
"notification",
"subscriptionId"
],
"type": "object"
},
"McpServerOauthLoginCompletedNotification": {
"properties": {
"error": {
@@ -6987,6 +7015,26 @@
"title": "McpServer/startupStatus/updatedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"mcpServer/event/stream/notification"
],
"title": "McpServer/event/stream/notificationNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/McpServerEventStreamNotification"
}
},
"required": [
"method",
"params"
],
"title": "McpServer/event/stream/notificationNotification",
"type": "object"
},
{
"properties": {
"method": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+50
View File
@@ -5202,6 +5202,26 @@
"title": "McpServer/startupStatus/updatedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"mcpServer/event/stream/notification"
],
"title": "McpServer/event/stream/notificationNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/v2/McpServerEventStreamNotification"
}
},
"required": [
"method",
"params"
],
"title": "McpServer/event/stream/notificationNotification",
"type": "object"
},
{
"properties": {
"method": {
@@ -13384,6 +13404,36 @@
"title": "McpResourceReadResponse",
"type": "object"
},
"McpServerEventNotification": {
"properties": {
"method": {
"type": "string"
},
"params": true
},
"required": [
"method",
"params"
],
"type": "object"
},
"McpServerEventStreamNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"notification": {
"$ref": "#/definitions/v2/McpServerEventNotification"
},
"subscriptionId": {
"type": "string"
}
},
"required": [
"notification",
"subscriptionId"
],
"title": "McpServerEventStreamNotification",
"type": "object"
},
"McpServerInfo": {
"description": "Presentation metadata advertised by an initialized MCP server.",
"properties": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+50
View File
@@ -9614,6 +9614,36 @@
"title": "McpResourceReadResponse",
"type": "object"
},
"McpServerEventNotification": {
"properties": {
"method": {
"type": "string"
},
"params": true
},
"required": [
"method",
"params"
],
"type": "object"
},
"McpServerEventStreamNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"notification": {
"$ref": "#/definitions/McpServerEventNotification"
},
"subscriptionId": {
"type": "string"
}
},
"required": [
"notification",
"subscriptionId"
],
"title": "McpServerEventStreamNotification",
"type": "object"
},
"McpServerInfo": {
"description": "Presentation metadata advertised by an initialized MCP server.",
"properties": {
@@ -15422,6 +15452,26 @@
"title": "McpServer/startupStatus/updatedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"mcpServer/event/stream/notification"
],
"title": "McpServer/event/stream/notificationNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/McpServerEventStreamNotification"
}
},
"required": [
"method",
"params"
],
"title": "McpServer/event/stream/notificationNotification",
"type": "object"
},
{
"properties": {
"method": {
codex-rs/app-server-protocol/schema/json/v2/McpServerEventStreamNotification.json
Generated
+32
View File
@@ -0,0 +1,32 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"McpServerEventNotification": {
"properties": {
"method": {
"type": "string"
},
"params": true
},
"required": [
"method",
"params"
],
"type": "object"
}
},
"properties": {
"notification": {
"$ref": "#/definitions/McpServerEventNotification"
},
"subscriptionId": {
"type": "string"
}
},
"required": [
"notification",
"subscriptionId"
],
"title": "McpServerEventStreamNotification",
"type": "object"
}
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts
Generated
+2 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts
Generated
+2 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/v2/McpServerEventNotification.ts
Generated
+6
View File
@@ -0,0 +1,6 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { JsonValue } from "../serde_json/JsonValue";
export type McpServerEventNotification = { method: string, params: JsonValue, };
codex-rs/app-server-protocol/schema/typescript/v2/McpServerEventStreamNotification.ts
Generated
+6
View File
@@ -0,0 +1,6 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { McpServerEventNotification } from "./McpServerEventNotification";
export type McpServerEventStreamNotification = { subscriptionId: string, notification: McpServerEventNotification, };
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
Generated
+2
View File
@@ -258,6 +258,8 @@ export type { McpResourceReadResponse } from "./McpResourceReadResponse";
export type { McpServerElicitationAction } from "./McpServerElicitationAction";
export type { McpServerElicitationRequestParams } from "./McpServerElicitationRequestParams";
export type { McpServerElicitationRequestResponse } from "./McpServerElicitationRequestResponse";
export type { McpServerEventNotification } from "./McpServerEventNotification";
export type { McpServerEventStreamNotification } from "./McpServerEventStreamNotification";
export type { McpServerMigration } from "./McpServerMigration";
export type { McpServerOauthClientRegistration } from "./McpServerOauthClientRegistration";
export type { McpServerOauthLoginCompletedNotification } from "./McpServerOauthLoginCompletedNotification";
codex-rs/app-server-protocol/src/protocol/common.rs
+16
View File
@@ -1144,6 +1144,20 @@ client_request_definitions! {
response: v2::McpResourceReadResponse,
},
#[experimental("mcpServer/event/stream/start")]
McpServerEventStreamStart => "mcpServer/event/stream/start" {
params: v2::McpServerEventStreamStartParams,
serialization: None,
response: v2::McpServerEventStreamStartResponse,
},
#[experimental("mcpServer/event/stream/stop")]
McpServerEventStreamStop => "mcpServer/event/stream/stop" {
params: v2::McpServerEventStreamStopParams,
serialization: None,
response: v2::McpServerEventStreamStopResponse,
},
McpServerToolCall => "mcpServer/tool/call" {
params: v2::McpServerToolCallParams,
serialization: thread_id(params.thread_id),
@@ -1865,6 +1879,8 @@ server_notification_definitions! {
McpToolCallProgress => "item/mcpToolCall/progress" (v2::McpToolCallProgressNotification),
McpServerOauthLoginCompleted => "mcpServer/oauthLogin/completed" (v2::McpServerOauthLoginCompletedNotification),
McpServerStatusUpdated => "mcpServer/startupStatus/updated" (v2::McpServerStatusUpdatedNotification),
#[experimental("mcpServer/event/stream/notification")]
McpServerEventStream => "mcpServer/event/stream/notification" (v2::McpServerEventStreamNotification),
AccountUpdated => "account/updated" (v2::AccountUpdatedNotification),
AccountRateLimitsUpdated => "account/rateLimits/updated" (v2::AccountRateLimitsUpdatedNotification),
AppListUpdated => "app/list/updated" (v2::AppListUpdatedNotification),
codex-rs/app-server-protocol/src/protocol/v2/mcp.rs
+47
View File
@@ -103,6 +103,53 @@ pub struct McpResourceReadResponse {
pub origin_call_id: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct McpServerEventStreamStartParams {
pub thread_id: String,
pub server: String,
pub subscription_id: String,
pub name: String,
pub arguments: JsonValue,
#[serde(rename = "_meta")]
#[ts(rename = "_meta", optional = nullable)]
pub meta: Option<JsonValue>,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct McpServerEventStreamStartResponse {}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct McpServerEventStreamStopParams {
pub subscription_id: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct McpServerEventStreamStopResponse {}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct McpServerEventNotification {
pub method: String,
pub params: JsonValue,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct McpServerEventStreamNotification {
pub subscription_id: String,
pub notification: McpServerEventNotification,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
codex-rs/app-server/README.md
+3
View File
@@ -279,6 +279,8 @@ Example with notification opt-out:
- `config/mcpServer/reload` — reload MCP server config from disk and queue a refresh for loaded threads (applied on each thread's next active turn); returns `{}`. Use this after editing `config.toml` without restarting the server.
- `mcpServerStatus/list` — enumerate configured MCP servers with their tools, auth status, server info, owning `pluginId` (`null` for servers not contributed by a plugin), plus resources/resource templates for `full` detail; supports optional `threadId` and cursor+limit pagination. If `threadId` is omitted, the server reads from the latest global config directly. If `detail` is omitted, the server defaults to `full`. An `unknown` auth status means OAuth support could not be determined; `unsupported` means OAuth is known not to be supported.
- `mcpServer/resource/read` — read a resource from a configured MCP server by optional `threadId`, `server`, and `uri`, returning text/blob resource `contents`. Pass `originCallId` with `threadId` to scope a Codex app widget to the app and account of the completed tool call that produced it; successful scoped reads return the same `originCallId`. Optional `connectorId` restricts other hosted app resources to their originating connector. If `threadId` is omitted, the server reads from the latest MCP config directly.
- `mcpServer/event/stream/start` (experimental) — subscribe to an MCP event by `threadId`, `server`, `subscriptionId`, event `name`, `arguments`, and optional `_meta`.
- `mcpServer/event/stream/stop` (experimental) — stop the caller's event subscription by `subscriptionId`.
- `mcpServer/tool/call` — call a tool on a thread's configured MCP server by `threadId`, `server`, `tool`, optional `arguments`, and optional `_meta`, returning the MCP tool result.
- `windowsSandbox/setupStart` — start Windows sandbox setup for the selected mode (`elevated` or `unelevated`); accepts an optional absolute `cwd` to target setup for a specific workspace, returns `{ started: true }` immediately, and later emits `windowsSandbox/setupCompleted`.
- `feedback/upload` — submit a feedback report (classification + optional reason/logs, conversation_id, and optional `extraLogFiles` attachments array); returns the tracking thread id.
@@ -2256,6 +2258,7 @@ Codex supports these authentication modes. The current mode is surfaced in `acco
- `account/sendAddCreditsNudgeEmail` — ask ChatGPT to email the workspace owner about depleted credits or a reached usage limit.
- `mcpServer/oauthLogin/completed` (notify) — emitted after a `mcpServer/oauth/login` flow finishes for a server; payload includes `{ name, threadId, success, error? }`.
- `mcpServer/startupStatus/updated` (notify) — emitted when a configured MCP server's startup status changes; payload includes `{ threadId, name, status, error, failureReason }`, where `threadId` is the owning thread when startup is thread-scoped and `null` when it is app-scoped, and `status` is `starting`, `ready`, `failed`, or `cancelled`. `failureReason` is `reauthenticationRequired` when stored OAuth credentials have expired and cannot be refreshed, so clients can prompt the user to reconnect the named server.
- `mcpServer/event/stream/notification` (experimental, notify) — forwards `{ subscriptionId, notification: { method, params } }` to the connection that owns the subscription.
### 1) Check auth state
codex-rs/app-server/src/in_process.rs
+1
View File
@@ -727,6 +727,7 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult<InProcessClie
match send_error {
mpsc::error::TrySendError::Full(_) => {
warn!("dropping in-process server notification (queue full)");
continue;
}
mpsc::error::TrySendError::Closed(_) => {
break;
codex-rs/app-server/src/lib.rs
+5 -5
View File
@@ -1176,11 +1176,11 @@ pub async fn run_main_with_transport_options(
};
if !shutdown_state.forced() {
futures::future::join_all(
connections
.values()
.map(|connection_state| connection_state.session.rpc_gate.shutdown()),
)
futures::future::join_all(connections.iter().map(
|(&connection_id, connection_state)| {
processor.connection_closed(connection_id, &connection_state.session)
},
))
.await;
connection_cleanup_tasks.drain().await;
processor.drain_background_tasks().await;
codex-rs/app-server/src/message_processor.rs
+53 -12
View File
@@ -8,6 +8,7 @@ use crate::attestation::app_server_attestation_provider;
use crate::config_manager::ConfigManager;
use crate::connection_rpc_gate::ConnectionRpcGate;
use crate::current_time::app_server_time_provider;
use crate::error_code::internal_error;
use crate::error_code::invalid_params;
use crate::error_code::invalid_request;
use crate::extensions::ThreadExtensionDependencies;
@@ -32,6 +33,8 @@ use crate::request_processors::FsRequestProcessor;
use crate::request_processors::GitRequestProcessor;
use crate::request_processors::InitializeRequestProcessor;
use crate::request_processors::MarketplaceRequestProcessor;
use crate::request_processors::McpEventStreamReady;
use crate::request_processors::McpEventStreams;
use crate::request_processors::McpRequestProcessor;
use crate::request_processors::PluginRequestProcessor;
use crate::request_processors::ProcessExecRequestProcessor;
@@ -163,6 +166,7 @@ pub(crate) struct MessageProcessor {
#[derive(Debug)]
pub(crate) struct ConnectionSessionState {
pub(crate) rpc_gate: Arc<ConnectionRpcGate>,
pub(crate) mcp_event_streams: McpEventStreams,
initialized: OnceLock<InitializedConnectionSessionState>,
}
@@ -186,6 +190,7 @@ impl ConnectionSessionState {
pub(crate) fn new() -> Self {
Self {
rpc_gate: Arc::new(ConnectionRpcGate::new()),
mcp_event_streams: McpEventStreams::default(),
initialized: OnceLock::new(),
}
}
@@ -447,6 +452,7 @@ impl MessageProcessor {
let mcp_processor = McpRequestProcessor::new(
auth_manager.clone(),
Arc::clone(&thread_manager),
thread_state_manager.clone(),
outgoing.clone(),
config_manager.clone(),
);
@@ -784,6 +790,8 @@ impl MessageProcessor {
connection_id: ConnectionId,
session_state: &ConnectionSessionState,
) {
session_state.rpc_gate.close().await;
session_state.mcp_event_streams.clear().await;
if timeout(
CONNECTION_RPC_DRAIN_TIMEOUT,
session_state.rpc_gate.shutdown(),
@@ -893,10 +901,16 @@ impl MessageProcessor {
&codex_request,
);
let event_stream_ready = match &codex_request {
ClientRequest::McpServerEventStreamStart { params, .. } => Some(
session
.mcp_event_streams
.start(connection_id, params.clone(), self.mcp_processor.clone())
.await?,
),
_ => None,
};
let serialization_scope = codex_request.serialization_scope();
let app_server_client_name = session.app_server_client_name().map(str::to_string);
let client_version = session.client_version().map(str::to_string);
let client_mcp_extensions = session.client_mcp_extensions();
let error_request_id = connection_request_id.clone();
let rpc_gate = Arc::clone(&session.rpc_gate);
let processor = Arc::clone(self);
@@ -910,9 +924,8 @@ impl MessageProcessor {
connection_request_id,
codex_request,
request_context,
app_server_client_name,
client_version,
client_mcp_extensions,
session,
event_stream_ready,
)
.await;
if let Err(error) = result {
@@ -940,16 +953,17 @@ impl MessageProcessor {
connection_request_id: ConnectionRequestId,
codex_request: ClientRequest,
request_context: RequestContext,
app_server_client_name: Option<String>,
client_version: Option<String>,
client_mcp_extensions: ClientMcpExtensions,
session: Arc<ConnectionSessionState>,
event_stream_ready: Option<McpEventStreamReady>,
) -> Result<(), JSONRPCErrorError> {
let connection_id = connection_request_id.connection_id;
let app_server_client_name = session.app_server_client_name().map(str::to_string);
let client_version = session.client_version().map(str::to_string);
let client_mcp_extensions = session.client_mcp_extensions();
let request_id = ConnectionRequestId {
connection_id,
request_id: codex_request.id().clone(),
};
let result: Result<Option<ClientResponsePayload>, JSONRPCErrorError> = match codex_request {
ClientRequest::Initialize { .. } => {
panic!("Initialize should be handled before initialized request dispatch");
@@ -1113,9 +1127,15 @@ impl MessageProcessor {
.await
}
ClientRequest::ThreadUnsubscribe { params, .. } => {
self.thread_processor
let thread_id = params.thread_id.clone();
let response = self
.thread_processor
.thread_unsubscribe(&request_id, params)
.await
.await?;
if let Ok(thread_id) = ThreadId::from_string(&thread_id) {
session.mcp_event_streams.stop_thread(thread_id).await;
}
Ok(response)
}
ClientRequest::ThreadResume { params, .. } => {
self.thread_processor
@@ -1486,6 +1506,27 @@ impl MessageProcessor {
.mcp_resource_read(&request_id, params)
.await
}
ClientRequest::McpServerEventStreamStart { params, .. } => {
let ready = event_stream_ready.ok_or_else(|| {
internal_error("MCP event subscription was not reserved before startup")
})?;
session
.mcp_event_streams
.wait_for_activation(&params.subscription_id, ready)
.await?;
Ok(Some(
codex_app_server_protocol::McpServerEventStreamStartResponse {}.into(),
))
}
ClientRequest::McpServerEventStreamStop { params, .. } => {
session
.mcp_event_streams
.stop(&params.subscription_id)
.await;
Ok(Some(
codex_app_server_protocol::McpServerEventStreamStopResponse {}.into(),
))
}
ClientRequest::McpServerToolCall { params, .. } => {
self.mcp_processor
.mcp_server_tool_call(&request_id, params)
codex-rs/app-server/src/outgoing_message.rs
+2 -2
View File
@@ -634,7 +634,7 @@ impl OutgoingMessageSender {
&self,
connection_id: ConnectionId,
notification: ServerNotification,
) {
) -> bool {
tracing::trace!("app-server event: {notification}");
let outgoing_message = timestamped_server_notification(notification);
let (write_complete_tx, write_complete_rx) = oneshot::channel();
@@ -649,7 +649,7 @@ impl OutgoingMessageSender {
{
warn!("failed to send server notification to client: {err:?}");
}
let _ = write_complete_rx.await;
write_complete_rx.await.is_ok()
}
pub(crate) async fn send_error(
codex-rs/app-server/src/request_processors.rs
+3
View File
@@ -543,6 +543,7 @@ mod fs_processor;
mod git_processor;
mod initialize_processor;
mod marketplace_processor;
mod mcp_event_stream;
mod mcp_processor;
mod persisted_resume_settings;
mod plugins;
@@ -571,6 +572,8 @@ pub(crate) use fs_processor::FsRequestProcessor;
pub(crate) use git_processor::GitRequestProcessor;
pub(crate) use initialize_processor::InitializeRequestProcessor;
pub(crate) use marketplace_processor::MarketplaceRequestProcessor;
pub(crate) use mcp_event_stream::McpEventStreamReady;
pub(crate) use mcp_event_stream::McpEventStreams;
pub(crate) use mcp_processor::McpRequestProcessor;
pub(crate) use plugins::PluginRequestProcessor;
pub(crate) use process_exec_processor::ProcessExecRequestProcessor;
codex-rs/app-server/src/request_processors/mcp_event_stream.rs
+335
View File
@@ -0,0 +1,335 @@
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::McpServerEventNotification;
use codex_app_server_protocol::McpServerEventStreamNotification;
use codex_app_server_protocol::McpServerEventStreamStartParams;
use codex_app_server_protocol::ServerNotification;
use codex_core::CodexThread;
use codex_login::AuthManager;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_mcp::McpEventStream;
use codex_protocol::ThreadId;
use serde_json::Value;
use tokio::sync::Mutex;
use tokio::sync::oneshot;
use tokio::sync::watch;
use tokio::task::JoinHandle;
use super::mcp_processor::McpRequestProcessor;
use crate::error_code::OVERLOADED_ERROR_CODE;
use crate::error_code::internal_error;
use crate::error_code::invalid_request;
use crate::outgoing_message::ConnectionId;
use crate::outgoing_message::OutgoingMessageSender;
const MAX_MCP_EVENT_STREAMS_PER_CONNECTION: usize = 64;
const MAX_MCP_EVENT_STREAM_RECONNECT_ATTEMPTS: u32 = 3;
const MCP_EVENT_STREAM_RECONNECT_DELAY: Duration = Duration::from_secs(/*secs*/ 1);
const MCP_EVENT_STREAM_STARTUP_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 90);
pub(crate) type McpEventStreamReady = oneshot::Receiver<Result<(), JSONRPCErrorError>>;
#[derive(Debug)]
struct McpEventStreamTask {
thread_id: ThreadId,
task: JoinHandle<()>,
}
#[derive(Debug, Default)]
pub(crate) struct McpEventStreams {
tasks: Mutex<HashMap<String, McpEventStreamTask>>,
}
impl McpEventStreams {
pub(crate) async fn start(
&self,
connection_id: ConnectionId,
params: McpServerEventStreamStartParams,
processor: McpRequestProcessor,
) -> Result<McpEventStreamReady, JSONRPCErrorError> {
if params.server != CODEX_APPS_MCP_SERVER_NAME {
return Err(invalid_request(
"MCP event subscriptions are only supported for hosted apps",
));
}
let thread_id = ThreadId::from_string(&params.thread_id)
.map_err(|error| invalid_request(format!("invalid thread id: {error}")))?;
let subscription_id = params.subscription_id.clone();
let (ready_tx, ready_rx) = oneshot::channel();
{
let mut tasks = self.tasks.lock().await;
tasks.retain(|_, task| !task.task.is_finished());
if tasks.contains_key(&subscription_id) {
return Err(invalid_request(format!(
"MCP event subscription '{subscription_id}' already exists"
)));
}
if tasks.len() >= MAX_MCP_EVENT_STREAMS_PER_CONNECTION {
return Err(JSONRPCErrorError {
code: OVERLOADED_ERROR_CODE,
message: format!(
"MCP event subscription limit of {MAX_MCP_EVENT_STREAMS_PER_CONNECTION} reached"
),
data: None,
});
}
let task = tokio::spawn(async move {
let mut auth_changes =
McpEventStreamAuthChanges::new(Arc::clone(&processor.auth_manager));
let opened = tokio::select! {
() = auth_changes.changed() => Err(internal_error(
"MCP event subscription authentication changed during startup",
)),
result = async {
if !processor
.thread_state_manager
.subscribed_connection_ids(thread_id)
.await
.contains(&connection_id)
{
return Err(invalid_request(format!(
"connection is not subscribed to thread '{thread_id}'"
)));
}
let (_, thread) = processor.load_thread(&params.thread_id).await?;
let stream = thread
.start_mcp_event_stream(
&params.name,
params.arguments.clone(),
params.meta.clone(),
)
.await
.map_err(|error| internal_error(format!(
"failed to start MCP event stream for '{}': {error:#}",
params.server
)))?;
Ok((stream, thread))
} => result,
};
match opened {
Ok((stream, thread)) => {
forward_events(
connection_id,
&params,
processor.outgoing.as_ref(),
thread,
stream,
auth_changes,
ready_tx,
)
.await;
}
Err(error) => {
let _ = ready_tx.send(Err(error));
}
}
});
tasks.insert(subscription_id, McpEventStreamTask { thread_id, task });
}
Ok(ready_rx)
}
pub(crate) async fn wait_for_activation(
&self,
subscription_id: &str,
ready: McpEventStreamReady,
) -> Result<(), JSONRPCErrorError> {
match tokio::time::timeout(MCP_EVENT_STREAM_STARTUP_TIMEOUT, ready).await {
Ok(Ok(result)) => result,
Ok(Err(_)) => Err(internal_error(
"MCP event stream ended before becoming active",
)),
Err(_) => {
self.stop(subscription_id).await;
Err(internal_error("MCP event stream startup timed out"))
}
}
}
pub(crate) async fn stop(&self, subscription_id: &str) {
let task = self.tasks.lock().await.remove(subscription_id);
if let Some(task) = task {
task.task.abort();
let _ = task.task.await;
}
}
pub(crate) async fn stop_thread(&self, thread_id: ThreadId) {
self.stop_matching(Some(thread_id)).await;
}
pub(crate) async fn clear(&self) {
self.stop_matching(/*thread_id*/ None).await;
}
async fn stop_matching(&self, thread_id: Option<ThreadId>) {
let tasks = self
.tasks
.lock()
.await
.extract_if(|_, task| thread_id.is_none_or(|id| task.thread_id == id))
.map(|(_, task)| task)
.collect::<Vec<_>>();
for task in tasks {
task.task.abort();
let _ = task.task.await;
}
}
}
async fn forward_events(
connection_id: ConnectionId,
params: &McpServerEventStreamStartParams,
outgoing: &OutgoingMessageSender,
thread: Arc<CodexThread>,
mut stream: McpEventStream,
mut auth_changes: McpEventStreamAuthChanges,
ready: oneshot::Sender<Result<(), JSONRPCErrorError>>,
) {
let mut ready = Some(ready);
let mut reconnect_attempts = 0;
let mut reconnect_deadline = None;
let send = |notification| {
outgoing.send_server_notification_to_connection_and_wait(
connection_id,
ServerNotification::McpServerEventStream(McpServerEventStreamNotification {
subscription_id: params.subscription_id.clone(),
notification,
}),
)
};
loop {
let notification = tokio::select! {
biased;
() = auth_changes.changed() => break,
() = thread.wait_until_terminated() => break,
result = stream.recv() => match result {
Ok(Some(notification)) => notification,
Ok(None)
if ready.is_none()
&& reconnect_attempts < MAX_MCP_EVENT_STREAM_RECONNECT_ATTEMPTS =>
{
reconnect_attempts += 1;
let deadline = *reconnect_deadline.get_or_insert_with(|| {
tokio::time::Instant::now() + MCP_EVENT_STREAM_STARTUP_TIMEOUT
});
tokio::select! {
biased;
() = auth_changes.changed() => break,
() = thread.wait_until_terminated() => break,
result = tokio::time::timeout_at(deadline, async {
tokio::time::sleep(
MCP_EVENT_STREAM_RECONNECT_DELAY
* (1 << (reconnect_attempts - 1)),
)
.await;
let mut stream = thread
.start_mcp_event_stream(
&params.name,
params.arguments.clone(),
params.meta.clone(),
)
.await?;
let notification = stream.recv().await?;
Ok::<_, anyhow::Error>((stream, notification))
}) => match result {
Ok(Ok((reconnected, Some(notification)))) => {
stream = reconnected;
notification
}
Ok(Ok((_, None)) | Err(_)) => continue,
Err(_) => break,
},
}
}
Ok(None) => break,
Err(error) => {
if let Some(ready) = ready.take() {
let _ = ready.send(Err(internal_error(format!("{error:#}"))));
}
break;
}
},
};
let active = notification.method == "notifications/events/active";
let terminated = notification.method == "notifications/events/terminated";
if !send(McpServerEventNotification {
method: notification.method,
params: notification
.params
.unwrap_or_else(|| Value::Object(Default::default())),
})
.await
{
break;
}
if active && let Some(ready) = ready.take() {
let _ = ready.send(Ok(()));
}
if terminated {
return;
}
if !active {
reconnect_attempts = 0;
reconnect_deadline = None;
}
}
if let Some(ready) = ready {
let _ = ready.send(Err(internal_error(
"MCP event stream ended before becoming active",
)));
} else {
send(McpServerEventNotification {
method: "notifications/events/terminated".to_string(),
params: Value::Object(Default::default()),
})
.await;
}
}
struct McpEventStreamAuthChanges {
manager: Arc<AuthManager>,
receiver: watch::Receiver<u64>,
revision: u64,
principal: Option<(Option<String>, Option<String>)>,
}
impl McpEventStreamAuthChanges {
fn new(manager: Arc<AuthManager>) -> Self {
let receiver = manager.auth_change_receiver();
let revision = *receiver.borrow();
let principal = manager
.auth_cached()
.map(|auth| (auth.get_account_id(), auth.get_chatgpt_user_id()));
Self {
manager,
receiver,
revision,
principal,
}
}
async fn changed(&mut self) {
loop {
if self
.manager
.auth_cached()
.map(|auth| (auth.get_account_id(), auth.get_chatgpt_user_id()))
!= self.principal
{
return;
}
if self.receiver.changed().await.is_err() || self.principal == Some((None, None)) {
return;
}
let revision = *self.receiver.borrow_and_update();
if revision.wrapping_sub(self.revision) > 1 {
return;
}
self.revision = revision;
}
}
}
codex-rs/app-server/src/request_processors/mcp_processor.rs
+8 -3
View File
@@ -3,20 +3,24 @@ use codex_core::McpManager;
use codex_mcp::McpServerSource;
use codex_mcp::ReadResourceRequestParams;
use crate::thread_state::ThreadStateManager;
const MCP_TOOL_THREAD_ID_META_KEY: &str = "threadId";
#[derive(Clone)]
pub(crate) struct McpRequestProcessor {
auth_manager: Arc<AuthManager>,
pub(super) auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
pub(super) outgoing: Arc<OutgoingMessageSender>,
config_manager: ConfigManager,
pub(super) thread_state_manager: ThreadStateManager,
}
impl McpRequestProcessor {
pub(crate) fn new(
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
thread_state_manager: ThreadStateManager,
outgoing: Arc<OutgoingMessageSender>,
config_manager: ConfigManager,
) -> Self {
@@ -25,6 +29,7 @@ impl McpRequestProcessor {
thread_manager,
outgoing,
config_manager,
thread_state_manager,
}
}
@@ -96,7 +101,7 @@ impl McpRequestProcessor {
.map_err(|err| internal_error(format!("failed to reload config: {err}")))
}
async fn load_thread(
pub(super) async fn load_thread(
&self,
thread_id: &str,
) -> Result<(ThreadId, Arc<CodexThread>), JSONRPCErrorError> {
codex-rs/app-server/tests/suite/v2/mcp_event_stream.rs
+246
View File
@@ -0,0 +1,246 @@
use std::convert::Infallible;
use std::sync::Arc;
use std::time::Duration;
use anyhow::Context as _;
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::MockResponsesConfig;
use app_test_support::TestAppServer;
use app_test_support::write_chatgpt_auth;
use axum::Json;
use axum::Router;
use axum::http::StatusCode;
use axum::response::IntoResponse;
use axum::response::sse::Event;
use axum::response::sse::Sse;
use axum::routing::post;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::McpServerEventNotification;
use codex_app_server_protocol::McpServerEventStreamNotification;
use codex_app_server_protocol::McpServerEventStreamStartResponse;
use codex_app_server_protocol::McpServerEventStreamStopParams;
use codex_app_server_protocol::McpServerEventStreamStopResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_config::types::AuthCredentialsStoreMode;
use codex_features::Feature;
use core_test_support::responses;
use futures::StreamExt as _;
use futures::stream;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::Notify;
use tokio::sync::mpsc;
use tokio::time::timeout;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mcp_event_stream_waits_for_activation_forwards_events_and_cancels() -> Result<()> {
let responses_server = responses::start_mock_server().await;
let listener = TcpListener::bind("127.0.0.1:0").await?;
let apps_url = format!("http://{}", listener.local_addr()?);
let (stream_started_tx, mut stream_started_rx) = mpsc::unbounded_channel::<Value>();
let allow_activation = Arc::new(Notify::new());
let activation_gate = Arc::clone(&allow_activation);
let router = Router::new().route(
"/api/codex/ps/mcp",
post(move |Json(message): Json<Value>| {
let stream_started_tx = stream_started_tx.clone();
let allow_activation = Arc::clone(&activation_gate);
async move {
match message["method"].as_str() {
Some("initialize") => (
[("mcp-session-id", "event-test-session")],
Json(json!({
"jsonrpc": "2.0",
"id": message["id"],
"result": {
"protocolVersion": "2025-06-18",
"capabilities": {},
"serverInfo": {
"name": "event-test-server",
"version": "1.0.0",
},
},
})),
)
.into_response(),
Some("notifications/initialized") => StatusCode::ACCEPTED.into_response(),
Some("tools/list") => Json(json!({
"jsonrpc": "2.0",
"id": message["id"],
"result": { "tools": [] },
}))
.into_response(),
Some("events/stream") => {
stream_started_tx
.send(message.clone())
.expect("stream-start receiver must remain open");
let metadata = json!({
"io.modelcontextprotocol/subscriptionId": message["id"],
"provider": "event-test-server",
});
let active = json!({
"jsonrpc": "2.0",
"method": "notifications/events/active",
"params": { "_meta": metadata, "status": "active" },
});
let event = json!({
"jsonrpc": "2.0",
"method": "notifications/events/event",
"params": {
"_meta": metadata,
"name": "issue.updated",
"data": { "issue": 42 },
},
});
let events = stream::once(async move {
allow_activation.notified().await;
Ok::<_, Infallible>(
Event::default().event("message").data(active.to_string()),
)
})
.chain(stream::once(async move {
Ok::<_, Infallible>(
Event::default().event("message").data(event.to_string()),
)
}))
.chain(stream::pending());
Sse::new(events).into_response()
}
method => panic!("unexpected MCP request: {method:?}"),
}
}
}),
);
let mcp_server = tokio::spawn(async move { axum::serve(listener, router).await });
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&responses_server.uri())
.with_root_config(&format!("chatgpt_base_url = \"{apps_url}\""))
.enable_feature(Feature::Apps)
.write(codex_home.path())?;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("chatgpt-token")
.account_id("account-123")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
let mut app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let ThreadStartResponse { thread, .. } = app_server
.start_thread(ThreadStartParams::default())
.await?;
let start_params = json!({
"threadId": thread.id,
"server": "codex_apps",
"subscriptionId": "subscription-1",
"name": "issue.updated",
"arguments": { "project": "codex" },
"_meta": { "source": "desktop" },
});
let start_request_id = app_server
.send_raw_request("mcpServer/event/stream/start", Some(start_params.clone()))
.await?;
let stream_request = timeout(Duration::from_secs(5), stream_started_rx.recv())
.await?
.context("MCP event stream was not requested")?;
assert_eq!(
stream_request["params"],
json!({
"name": "issue.updated",
"arguments": { "project": "codex" },
"_meta": {
"progressToken": stream_request["params"]["_meta"]["progressToken"],
"source": "desktop",
},
})
);
let duplicate_request_id = app_server
.send_raw_request("mcpServer/event/stream/start", Some(start_params))
.await?;
let duplicate_error = timeout(
Duration::from_secs(5),
app_server.read_stream_until_error_message(RequestId::Integer(duplicate_request_id)),
)
.await??;
assert!(duplicate_error.error.message.contains("already exists"));
assert!(
timeout(
Duration::from_millis(100),
app_server.read_stream_until_response_message(RequestId::Integer(start_request_id)),
)
.await
.is_err(),
"subscription start must not complete before the active notification"
);
allow_activation.notify_one();
let _: McpServerEventStreamStartResponse = timeout(
Duration::from_secs(5),
app_server.read_response(start_request_id),
)
.await??;
let metadata = json!({
"io.modelcontextprotocol/subscriptionId": stream_request["id"],
"provider": "event-test-server",
});
let active: McpServerEventStreamNotification = timeout(
Duration::from_secs(5),
app_server.read_notification("mcpServer/event/stream/notification"),
)
.await??;
assert_eq!(
active,
McpServerEventStreamNotification {
subscription_id: "subscription-1".to_string(),
notification: McpServerEventNotification {
method: "notifications/events/active".to_string(),
params: json!({ "_meta": metadata, "status": "active" }),
},
}
);
let event: McpServerEventStreamNotification = timeout(
Duration::from_secs(5),
app_server.read_notification("mcpServer/event/stream/notification"),
)
.await??;
assert_eq!(
event,
McpServerEventStreamNotification {
subscription_id: "subscription-1".to_string(),
notification: McpServerEventNotification {
method: "notifications/events/event".to_string(),
params: json!({
"_meta": metadata,
"name": "issue.updated",
"data": { "issue": 42 },
}),
},
}
);
let _: McpServerEventStreamStopResponse = app_server
.request(|request_id| ClientRequest::McpServerEventStreamStop {
request_id,
params: McpServerEventStreamStopParams {
subscription_id: "subscription-1".to_string(),
},
})
.await?;
mcp_server.abort();
let _ = mcp_server.await;
Ok(())
}
codex-rs/app-server/tests/suite/v2/mod.rs
+1
View File
@@ -43,6 +43,7 @@ mod initialize;
mod marketplace_add;
mod marketplace_remove;
mod marketplace_upgrade;
mod mcp_event_stream;
mod mcp_resource;
mod mcp_resource_origin;
mod mcp_server_elicitation;
codex-rs/codex-mcp/src/connection_manager/resources.rs
+12 -6
View File
@@ -16,6 +16,7 @@ use tokio::task::JoinSet;
use tracing::warn;
use super::McpConnectionSet;
use super::McpServerConnection;
use crate::pagination::collect_paginated;
use crate::rmcp_client::ManagedClient;
@@ -160,15 +161,20 @@ impl McpConnectionSet {
&self,
name: &str,
) -> Result<(ManagedClient, Option<Duration>)> {
let (client, timeout, _connection) = self.client_with_connection_by_name(name).await?;
Ok((client, timeout))
}
pub(crate) async fn client_with_connection_by_name(
&self,
name: &str,
) -> Result<(ManagedClient, Option<Duration>, Arc<McpServerConnection>)> {
let view = self
.servers
.get(name)
.ok_or_else(|| anyhow!("unknown MCP server '{name}'"))?;
let client = view
.connection
.client()
.await
.context("failed to get client")?;
Ok((client, view.tool_timeout))
let connection = Arc::clone(&view.connection);
let client = connection.client().await.context("failed to get client")?;
Ok((client, view.tool_timeout, connection))
}
}
codex-rs/codex-mcp/src/resource_client.rs
+42 -28
View File
@@ -8,6 +8,7 @@ use anyhow::anyhow;
use codex_protocol::mcp::Resource;
use codex_protocol::mcp::ResourceContent;
use codex_rmcp_client::CancellableEventStreamRequest;
use rmcp::model::GetMeta;
use rmcp::model::PaginatedRequestParams;
use rmcp::model::ReadResourceRequestParams;
use rmcp::model::ServerResult;
@@ -17,9 +18,11 @@ use serde_json::Map;
use serde_json::Value;
use serde_json::json;
use tokio::runtime::Handle;
use tokio::sync::watch;
use crate::McpRuntime;
use crate::connection_manager::McpConnectionSet;
use crate::connection_manager::McpServerConnection;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
/// One page of resources returned by an MCP server.
@@ -66,7 +69,8 @@ pub struct McpEventNotification {
pub struct McpEventStream {
request: Option<CancellableEventStreamRequest>,
runtime_handle: Handle,
_connections: Arc<McpConnectionSet>,
connection: Option<Arc<McpServerConnection>>,
hosted_event_server_removals: watch::Receiver<()>,
}
impl McpEventStream {
@@ -79,57 +83,62 @@ impl McpEventStream {
tokio::select! {
biased;
notification = request.notifications.recv() => {
match notification {
Some(notification) => Ok(Some(McpEventNotification {
method: notification.method,
params: notification.params,
})),
None => {
let response = (&mut request.handle.rx).await;
self.request = None;
match response {
Ok(Ok(_)) | Ok(Err(ServiceError::Cancelled { .. })) => Ok(None),
Ok(Err(error)) => Err(error.into()),
Err(error) => Err(error.into()),
}
}
Ok(()) = self.hosted_event_server_removals.changed() => {
self.cancel();
Err(anyhow!("hosted MCP event server was removed"))
}
Some(notification) = request.notifications.recv() => {
let metadata = notification.get_meta().0.0.clone();
let mut params = notification.params;
if !metadata.is_empty() {
params.get_or_insert_with(|| json!({}))["_meta"] = Value::Object(metadata);
}
Ok(Some(McpEventNotification {
method: notification.method,
params,
}))
}
response = &mut request.handle.rx => {
self.request = None;
self.connection = None;
match response {
Ok(Ok(_)) | Ok(Err(ServiceError::Cancelled { .. })) => Ok(None),
Ok(Ok(_))
| Ok(Err(ServiceError::Cancelled { .. }))
| Ok(Err(ServiceError::TransportClosed))
| Err(_) => Ok(None),
Ok(Err(error)) => Err(error.into()),
Err(error) => Err(error.into()),
}
}
}
}
}
impl Drop for McpEventStream {
fn drop(&mut self) {
fn cancel(&mut self) {
if let Some(CancellableEventStreamRequest {
handle,
notifications,
}) = self.request.take()
{
drop(notifications);
let connections = Arc::clone(&self._connections);
let connection = self.connection.take();
self.runtime_handle.spawn(async move {
let _ = tokio::time::timeout(
Duration::from_secs(30),
handle.cancel(Some("event subscription closed".to_string())),
)
.await;
drop(connections);
drop(connection);
});
}
}
}
impl Drop for McpEventStream {
fn drop(&mut self) {
self.cancel();
}
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct McpEventListResult {
@@ -222,7 +231,9 @@ impl McpResourceClient {
/// Lists the events advertised by the hosted Plugin Runtime.
pub async fn list_events(&self) -> Result<McpEventCatalogSnapshot> {
let connections = self.runtime.latest_connections();
let (connections, _) = self
.runtime
.latest_connections_for_event_server(CODEX_APPS_MCP_SERVER_NAME)?;
let cache_key = McpResourceClientCacheKey(Arc::downgrade(&connections));
let (managed, request_timeout) = connections
.client_by_name(CODEX_APPS_MCP_SERVER_NAME)
@@ -260,9 +271,11 @@ impl McpResourceClient {
params["_meta"] = Value::Object(request_meta.clone());
}
let connections = self.runtime.latest_connections();
let (managed, _) = connections
.client_by_name(CODEX_APPS_MCP_SERVER_NAME)
let (connections, hosted_event_server_removals) = self
.runtime
.latest_connections_for_event_server(CODEX_APPS_MCP_SERVER_NAME)?;
let (managed, _, connection) = connections
.client_with_connection_by_name(CODEX_APPS_MCP_SERVER_NAME)
.await?;
let request = managed
.client
@@ -273,7 +286,8 @@ impl McpResourceClient {
Ok(McpEventStream {
request: Some(request),
runtime_handle: Handle::current(),
_connections: connections,
connection: Some(connection),
hosted_event_server_removals,
})
}
}
codex-rs/codex-mcp/src/runtime.rs
+40
View File
@@ -49,6 +49,7 @@ use crate::connection_manager::McpConnectionSet;
use crate::elicitation::ElicitationLifecycle;
use crate::elicitation::ElicitationRequestRouter;
use crate::elicitation::ElicitationReviewerHandle;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::resource_origin::ResourceOrigins;
use crate::server::EffectiveMcpServer;
use crate::tool_catalog_cache::McpToolCatalogCache;
@@ -90,6 +91,7 @@ pub struct McpRuntimeInput {
/// their exact connections and configuration for as long as they are needed.
pub struct McpRuntime {
current: ArcSwap<PublishedMcpRuntime>,
hosted_event_server_removals: watch::Sender<()>,
reconnect_pending: AtomicBool,
elicitation_router: ElicitationRequestRouter,
resource_origins: Mutex<ResourceOrigins>,
@@ -174,6 +176,7 @@ impl McpRuntime {
ready_selected_capability_roots: Vec::new(),
cached_binding: Mutex::new(None),
}),
hosted_event_server_removals: watch::channel(()).0,
reconnect_pending: AtomicBool::new(false),
elicitation_router: ElicitationRequestRouter::default(),
resource_origins: Mutex::default(),
@@ -276,6 +279,15 @@ impl McpRuntime {
)
.await,
);
let hosted_event_server_retained = connections.contains_server(CODEX_APPS_MCP_SERVER_NAME)
&& config
.mcp_server_catalog
.server(CODEX_APPS_MCP_SERVER_NAME)
.is_some_and(|registration| {
registration
.source()
.is_host_owned_apps(CODEX_APPS_MCP_SERVER_NAME, registration.config())
});
self.current.store(Arc::new(PublishedMcpRuntime {
connections,
config: Some(config),
@@ -286,6 +298,9 @@ impl McpRuntime {
cached_binding: Mutex::new(None),
}));
let _ = publish.send(true);
if !hosted_event_server_retained {
self.hosted_event_server_removals.send_replace(());
}
}
/// Ensures the next refresh creates fresh connections for every configured server.
@@ -505,6 +520,31 @@ impl McpRuntime {
Arc::clone(&self.current.load().connections)
}
pub(crate) fn latest_connections_for_event_server(
&self,
server: &str,
) -> anyhow::Result<(Arc<McpConnectionSet>, watch::Receiver<()>)> {
let hosted_event_server_removals = self.hosted_event_server_removals.subscribe();
let current = self.current.load();
if server == CODEX_APPS_MCP_SERVER_NAME
&& !current
.config
.as_ref()
.and_then(|config| config.mcp_server_catalog.server(server))
.is_some_and(|registration| {
registration
.source()
.is_host_owned_apps(server, registration.config())
})
{
anyhow::bail!("MCP server '{server}' is not registered by the hosted runtime");
}
Ok((
Arc::clone(&current.connections),
hosted_event_server_removals,
))
}
pub async fn shutdown(&self) {
self.latest_connections().shutdown().await;
}
codex-rs/core/src/codex_thread.rs
+20
View File
@@ -751,6 +751,26 @@ impl CodexThread {
Ok(serde_json::to_value(result)?)
}
pub async fn start_mcp_event_stream(
&self,
name: &str,
arguments: serde_json::Value,
meta: Option<serde_json::Value>,
) -> anyhow::Result<codex_mcp::McpEventStream> {
let meta = match meta.as_ref() {
Some(serde_json::Value::Object(meta)) => Some(meta),
Some(other) => {
anyhow::bail!("MCP event request _meta must be a JSON object, got {other}")
}
None => None,
};
let _ = self.session.services.auth_manager.auth().await;
self.session.refresh_mcp_if_dirty().await;
codex_mcp::McpResourceClient::new(Arc::clone(&self.session.services.mcp_runtime))
.open_event_stream(name, &arguments, meta)
.await
}
pub async fn call_mcp_tool(
&self,
server: &str,
codex-rs/rmcp-client/src/event_notification_transport.rs
+9 -8
View File
@@ -55,15 +55,10 @@ pub(crate) fn event_notification_channel() -> (EventNotificationSender, EventNot
}
impl EventNotificationSender {
async fn send(
&self,
notification: CustomNotification,
notification_bytes: usize,
) -> Result<(), ()> {
fn send(&self, notification: CustomNotification, notification_bytes: usize) -> Result<(), ()> {
let bytes = u32::try_from(notification_bytes).map_err(|_| ())?;
let permit = Arc::clone(&self.available_bytes)
.acquire_many_owned(bytes)
.await
.try_acquire_many_owned(bytes)
.map_err(|_| ())?;
self.notifications
.send(QueuedEventNotification {
@@ -231,7 +226,6 @@ where
}
if route
.send(notification.clone(), notification_bytes)
.await
.is_err()
&& let Some(route) = self
.routes
@@ -239,6 +233,13 @@ where
.unwrap_or_else(PoisonError::into_inner)
.remove(&subscription_id)
{
let _ = route.send(
CustomNotification::new(
"notifications/events/terminated",
/*params*/ None,
),
/*notification_bytes*/ 0,
);
route.close();
}
}
codex-rs/tui/src/app/app_server_event_targets.rs
+1
View File
@@ -192,6 +192,7 @@ pub(super) fn server_notification_thread_target(
| ServerNotification::CommandExecOutputDelta(_)
| ServerNotification::ProcessOutputDelta(_)
| ServerNotification::ProcessExited(_)
| ServerNotification::McpServerEventStream(_)
| ServerNotification::FsChanged(_)
| ServerNotification::WindowsWorldWritableWarning(_)
| ServerNotification::WindowsSandboxSetupCompleted(_)
codex-rs/tui/src/chatwidget/protocol.rs
+1
View File
@@ -222,6 +222,7 @@ impl ChatWidget {
| ServerNotification::CommandExecOutputDelta(_)
| ServerNotification::ProcessOutputDelta(_)
| ServerNotification::ProcessExited(_)
| ServerNotification::McpServerEventStream(_)
| ServerNotification::FileChangePatchUpdated(_)
| ServerNotification::McpToolCallProgress(_)
| ServerNotification::McpServerOauthLoginCompleted(_)