diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 2d65f17ecd..e97d56bbad 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -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": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 4c89f78e1c..15d25e6bb6 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -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": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 0514ad4ad7..fac84497ae 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -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": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/McpServerEventStreamNotification.json b/codex-rs/app-server-protocol/schema/json/v2/McpServerEventStreamNotification.json new file mode 100644 index 0000000000..2b65bda01c --- /dev/null +++ b/codex-rs/app-server-protocol/schema/json/v2/McpServerEventStreamNotification.json @@ -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" +} \ No newline at end of file diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 8b5cafce9a..946916b6ab 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 90f598fc41..935759250b 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts index 82a48530f3..c0179dc46e 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts @@ -27,6 +27,7 @@ import type { ItemCompletedNotification } from "./v2/ItemCompletedNotification"; import type { ItemGuardianApprovalReviewCompletedNotification } from "./v2/ItemGuardianApprovalReviewCompletedNotification"; import type { ItemGuardianApprovalReviewStartedNotification } from "./v2/ItemGuardianApprovalReviewStartedNotification"; import type { ItemStartedNotification } from "./v2/ItemStartedNotification"; +import type { McpServerEventStreamNotification } from "./v2/McpServerEventStreamNotification"; import type { McpServerOauthLoginCompletedNotification } from "./v2/McpServerOauthLoginCompletedNotification"; import type { McpServerStatusUpdatedNotification } from "./v2/McpServerStatusUpdatedNotification"; import type { McpToolCallProgressNotification } from "./v2/McpToolCallProgressNotification"; @@ -81,4 +82,4 @@ import type { WindowsWorldWritableWarningNotification } from "./v2/WindowsWorldW /** * Notification sent from the server to the client. */ -export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/reverted", "params": ThreadRevertedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/queue/changed", "params": ThreadQueueChangedNotification } | { "method": "project/changed", "params": ProjectChangedNotification } | { "method": "thread/project/updated", "params": ThreadProjectUpdatedNotification } | { "method": "thread/environment/connected", "params": EnvironmentConnectionNotification } | { "method": "thread/environment/disconnected", "params": EnvironmentConnectionNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "autoApprovalReview/strictReviewRequired", "params": StrictReviewRequiredNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/progress", "params": ExternalAgentConfigImportProgressNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }; +export type ServerNotification = { "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/reverted", "params": ThreadRevertedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/queue/changed", "params": ThreadQueueChangedNotification } | { "method": "project/changed", "params": ProjectChangedNotification } | { "method": "thread/project/updated", "params": ThreadProjectUpdatedNotification } | { "method": "thread/environment/connected", "params": EnvironmentConnectionNotification } | { "method": "thread/environment/disconnected", "params": EnvironmentConnectionNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "autoApprovalReview/strictReviewRequired", "params": StrictReviewRequiredNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "mcpServer/event/stream/notification", "params": McpServerEventStreamNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/progress", "params": ExternalAgentConfigImportProgressNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }; diff --git a/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts b/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts index 9d5535fe8b..a8c657713d 100644 --- a/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts +++ b/codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts @@ -27,6 +27,7 @@ import type { ItemCompletedNotification } from "./v2/ItemCompletedNotification"; import type { ItemGuardianApprovalReviewCompletedNotification } from "./v2/ItemGuardianApprovalReviewCompletedNotification"; import type { ItemGuardianApprovalReviewStartedNotification } from "./v2/ItemGuardianApprovalReviewStartedNotification"; import type { ItemStartedNotification } from "./v2/ItemStartedNotification"; +import type { McpServerEventStreamNotification } from "./v2/McpServerEventStreamNotification"; import type { McpServerOauthLoginCompletedNotification } from "./v2/McpServerOauthLoginCompletedNotification"; import type { McpServerStatusUpdatedNotification } from "./v2/McpServerStatusUpdatedNotification"; import type { McpToolCallProgressNotification } from "./v2/McpToolCallProgressNotification"; @@ -91,4 +92,4 @@ export type ServerNotificationEnvelope = { * Optional so clients can decode notifications from older app-server * versions. Current app-server versions always populate it. */ -emittedAtMs?: number, } & ({ "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/reverted", "params": ThreadRevertedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/queue/changed", "params": ThreadQueueChangedNotification } | { "method": "project/changed", "params": ProjectChangedNotification } | { "method": "thread/project/updated", "params": ThreadProjectUpdatedNotification } | { "method": "thread/environment/connected", "params": EnvironmentConnectionNotification } | { "method": "thread/environment/disconnected", "params": EnvironmentConnectionNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "autoApprovalReview/strictReviewRequired", "params": StrictReviewRequiredNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/progress", "params": ExternalAgentConfigImportProgressNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }); +emittedAtMs?: number, } & ({ "method": "error", "params": ErrorNotification } | { "method": "thread/started", "params": ThreadStartedNotification } | { "method": "thread/status/changed", "params": ThreadStatusChangedNotification } | { "method": "thread/archived", "params": ThreadArchivedNotification } | { "method": "thread/deleted", "params": ThreadDeletedNotification } | { "method": "thread/unarchived", "params": ThreadUnarchivedNotification } | { "method": "thread/closed", "params": ThreadClosedNotification } | { "method": "thread/reverted", "params": ThreadRevertedNotification } | { "method": "skills/changed", "params": SkillsChangedNotification } | { "method": "thread/name/updated", "params": ThreadNameUpdatedNotification } | { "method": "thread/goal/updated", "params": ThreadGoalUpdatedNotification } | { "method": "thread/goal/cleared", "params": ThreadGoalClearedNotification } | { "method": "thread/queue/changed", "params": ThreadQueueChangedNotification } | { "method": "project/changed", "params": ProjectChangedNotification } | { "method": "thread/project/updated", "params": ThreadProjectUpdatedNotification } | { "method": "thread/environment/connected", "params": EnvironmentConnectionNotification } | { "method": "thread/environment/disconnected", "params": EnvironmentConnectionNotification } | { "method": "thread/settings/updated", "params": ThreadSettingsUpdatedNotification } | { "method": "thread/tokenUsage/updated", "params": ThreadTokenUsageUpdatedNotification } | { "method": "turn/started", "params": TurnStartedNotification } | { "method": "hook/started", "params": HookStartedNotification } | { "method": "turn/completed", "params": TurnCompletedNotification } | { "method": "hook/completed", "params": HookCompletedNotification } | { "method": "turn/diff/updated", "params": TurnDiffUpdatedNotification } | { "method": "turn/plan/updated", "params": TurnPlanUpdatedNotification } | { "method": "item/started", "params": ItemStartedNotification } | { "method": "item/autoApprovalReview/started", "params": ItemGuardianApprovalReviewStartedNotification } | { "method": "item/autoApprovalReview/completed", "params": ItemGuardianApprovalReviewCompletedNotification } | { "method": "autoApprovalReview/strictReviewRequired", "params": StrictReviewRequiredNotification } | { "method": "item/completed", "params": ItemCompletedNotification } | { "method": "rawResponseItem/completed", "params": RawResponseItemCompletedNotification } | { "method": "rawResponse/completed", "params": RawResponseCompletedNotification } | { "method": "item/agentMessage/delta", "params": AgentMessageDeltaNotification } | { "method": "item/plan/delta", "params": PlanDeltaNotification } | { "method": "command/exec/outputDelta", "params": CommandExecOutputDeltaNotification } | { "method": "process/outputDelta", "params": ProcessOutputDeltaNotification } | { "method": "process/exited", "params": ProcessExitedNotification } | { "method": "item/commandExecution/outputDelta", "params": CommandExecutionOutputDeltaNotification } | { "method": "item/commandExecution/terminalInteraction", "params": TerminalInteractionNotification } | { "method": "item/fileChange/outputDelta", "params": FileChangeOutputDeltaNotification } | { "method": "item/fileChange/patchUpdated", "params": FileChangePatchUpdatedNotification } | { "method": "serverRequest/resolved", "params": ServerRequestResolvedNotification } | { "method": "item/mcpToolCall/progress", "params": McpToolCallProgressNotification } | { "method": "mcpServer/oauthLogin/completed", "params": McpServerOauthLoginCompletedNotification } | { "method": "mcpServer/startupStatus/updated", "params": McpServerStatusUpdatedNotification } | { "method": "mcpServer/event/stream/notification", "params": McpServerEventStreamNotification } | { "method": "account/updated", "params": AccountUpdatedNotification } | { "method": "account/rateLimits/updated", "params": AccountRateLimitsUpdatedNotification } | { "method": "app/list/updated", "params": AppListUpdatedNotification } | { "method": "remoteControl/status/changed", "params": RemoteControlStatusChangedNotification } | { "method": "externalAgentConfig/import/progress", "params": ExternalAgentConfigImportProgressNotification } | { "method": "externalAgentConfig/import/completed", "params": ExternalAgentConfigImportCompletedNotification } | { "method": "fs/changed", "params": FsChangedNotification } | { "method": "item/reasoning/summaryTextDelta", "params": ReasoningSummaryTextDeltaNotification } | { "method": "item/reasoning/summaryPartAdded", "params": ReasoningSummaryPartAddedNotification } | { "method": "item/reasoning/textDelta", "params": ReasoningTextDeltaNotification } | { "method": "thread/compacted", "params": ContextCompactedNotification } | { "method": "model/rerouted", "params": ModelReroutedNotification } | { "method": "model/verification", "params": ModelVerificationNotification } | { "method": "turn/moderationMetadata", "params": TurnModerationMetadataNotification } | { "method": "model/safetyBuffering/updated", "params": ModelSafetyBufferingUpdatedNotification } | { "method": "warning", "params": WarningNotification } | { "method": "guardianWarning", "params": GuardianWarningNotification } | { "method": "deprecationNotice", "params": DeprecationNoticeNotification } | { "method": "configWarning", "params": ConfigWarningNotification } | { "method": "fuzzyFileSearch/sessionUpdated", "params": FuzzyFileSearchSessionUpdatedNotification } | { "method": "fuzzyFileSearch/sessionCompleted", "params": FuzzyFileSearchSessionCompletedNotification } | { "method": "thread/realtime/started", "params": ThreadRealtimeStartedNotification } | { "method": "thread/realtime/itemAdded", "params": ThreadRealtimeItemAddedNotification } | { "method": "thread/realtime/transcript/delta", "params": ThreadRealtimeTranscriptDeltaNotification } | { "method": "thread/realtime/transcript/done", "params": ThreadRealtimeTranscriptDoneNotification } | { "method": "thread/realtime/outputAudio/delta", "params": ThreadRealtimeOutputAudioDeltaNotification } | { "method": "thread/realtime/sdp", "params": ThreadRealtimeSdpNotification } | { "method": "thread/realtime/error", "params": ThreadRealtimeErrorNotification } | { "method": "thread/realtime/closed", "params": ThreadRealtimeClosedNotification } | { "method": "windows/worldWritableWarning", "params": WindowsWorldWritableWarningNotification } | { "method": "windowsSandbox/setupCompleted", "params": WindowsSandboxSetupCompletedNotification } | { "method": "account/login/completed", "params": AccountLoginCompletedNotification }); diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/McpServerEventNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/McpServerEventNotification.ts new file mode 100644 index 0000000000..fd01cbe0b1 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/McpServerEventNotification.ts @@ -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, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/McpServerEventStreamNotification.ts b/codex-rs/app-server-protocol/schema/typescript/v2/McpServerEventStreamNotification.ts new file mode 100644 index 0000000000..81f12340a8 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/McpServerEventStreamNotification.ts @@ -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, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index d46190b14f..3013132f43 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -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"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index ab5759e017..d8394635b1 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -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), diff --git a/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs b/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs index 283c1be360..5172434811 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/mcp.rs @@ -103,6 +103,53 @@ pub struct McpResourceReadResponse { pub origin_call_id: Option, } +#[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, +} + +#[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/")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 68b5bfc3d9..10aa1e6675 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -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 diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 31fc421532..12f82e6b9e 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -727,6 +727,7 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult { warn!("dropping in-process server notification (queue full)"); + continue; } mpsc::error::TrySendError::Closed(_) => { break; diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index 423194b50c..e253481a98 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -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; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 4c7ca2e8de..62afbca593 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -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, + pub(crate) mcp_event_streams: McpEventStreams, initialized: OnceLock, } @@ -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, - client_version: Option, - client_mcp_extensions: ClientMcpExtensions, + session: Arc, + event_stream_ready: Option, ) -> 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, 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(¶ms.subscription_id, ready) + .await?; + Ok(Some( + codex_app_server_protocol::McpServerEventStreamStartResponse {}.into(), + )) + } + ClientRequest::McpServerEventStreamStop { params, .. } => { + session + .mcp_event_streams + .stop(¶ms.subscription_id) + .await; + Ok(Some( + codex_app_server_protocol::McpServerEventStreamStopResponse {}.into(), + )) + } ClientRequest::McpServerToolCall { params, .. } => { self.mcp_processor .mcp_server_tool_call(&request_id, params) diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 262c674ae6..c5dd088121 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -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( diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index e1e359f173..34407e7fde 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -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; diff --git a/codex-rs/app-server/src/request_processors/mcp_event_stream.rs b/codex-rs/app-server/src/request_processors/mcp_event_stream.rs new file mode 100644 index 0000000000..90b36ae7d3 --- /dev/null +++ b/codex-rs/app-server/src/request_processors/mcp_event_stream.rs @@ -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>; + +#[derive(Debug)] +struct McpEventStreamTask { + thread_id: ThreadId, + task: JoinHandle<()>, +} + +#[derive(Debug, Default)] +pub(crate) struct McpEventStreams { + tasks: Mutex>, +} + +impl McpEventStreams { + pub(crate) async fn start( + &self, + connection_id: ConnectionId, + params: McpServerEventStreamStartParams, + processor: McpRequestProcessor, + ) -> Result { + 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(¶ms.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(¶ms.thread_id).await?; + let stream = thread + .start_mcp_event_stream( + ¶ms.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, + ¶ms, + 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) { + let tasks = self + .tasks + .lock() + .await + .extract_if(|_, task| thread_id.is_none_or(|id| task.thread_id == id)) + .map(|(_, task)| task) + .collect::>(); + for task in tasks { + task.task.abort(); + let _ = task.task.await; + } + } +} + +async fn forward_events( + connection_id: ConnectionId, + params: &McpServerEventStreamStartParams, + outgoing: &OutgoingMessageSender, + thread: Arc, + mut stream: McpEventStream, + mut auth_changes: McpEventStreamAuthChanges, + ready: oneshot::Sender>, +) { + 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( + ¶ms.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, + receiver: watch::Receiver, + revision: u64, + principal: Option<(Option, Option)>, +} + +impl McpEventStreamAuthChanges { + fn new(manager: Arc) -> 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; + } + } +} diff --git a/codex-rs/app-server/src/request_processors/mcp_processor.rs b/codex-rs/app-server/src/request_processors/mcp_processor.rs index e5b2217c2e..2cfb470386 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/codex-rs/app-server/src/request_processors/mcp_processor.rs @@ -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, + pub(super) auth_manager: Arc, thread_manager: Arc, - outgoing: Arc, + pub(super) outgoing: Arc, config_manager: ConfigManager, + pub(super) thread_state_manager: ThreadStateManager, } impl McpRequestProcessor { pub(crate) fn new( auth_manager: Arc, thread_manager: Arc, + thread_state_manager: ThreadStateManager, outgoing: Arc, 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), JSONRPCErrorError> { diff --git a/codex-rs/app-server/tests/suite/v2/mcp_event_stream.rs b/codex-rs/app-server/tests/suite/v2/mcp_event_stream.rs new file mode 100644 index 0000000000..e7599c43f8 --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/mcp_event_stream.rs @@ -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::(); + 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| { + 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(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 2d058cfaa8..874d02c01f 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -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; diff --git a/codex-rs/codex-mcp/src/connection_manager/resources.rs b/codex-rs/codex-mcp/src/connection_manager/resources.rs index c84cc6b43f..2495e64912 100644 --- a/codex-rs/codex-mcp/src/connection_manager/resources.rs +++ b/codex-rs/codex-mcp/src/connection_manager/resources.rs @@ -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)> { + 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, Arc)> { 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)) } } diff --git a/codex-rs/codex-mcp/src/resource_client.rs b/codex-rs/codex-mcp/src/resource_client.rs index 89ec66eb91..ba10d2c3e3 100644 --- a/codex-rs/codex-mcp/src/resource_client.rs +++ b/codex-rs/codex-mcp/src/resource_client.rs @@ -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, runtime_handle: Handle, - _connections: Arc, + connection: Option>, + 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 { - 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, }) } } diff --git a/codex-rs/codex-mcp/src/runtime.rs b/codex-rs/codex-mcp/src/runtime.rs index ca85a03501..daa47c29aa 100644 --- a/codex-rs/codex-mcp/src/runtime.rs +++ b/codex-rs/codex-mcp/src/runtime.rs @@ -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, + hosted_event_server_removals: watch::Sender<()>, reconnect_pending: AtomicBool, elicitation_router: ElicitationRequestRouter, resource_origins: Mutex, @@ -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, 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(¤t.connections), + hosted_event_server_removals, + )) + } + pub async fn shutdown(&self) { self.latest_connections().shutdown().await; } diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 2f5403c604..7098fb5308 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -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, + ) -> anyhow::Result { + 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, diff --git a/codex-rs/rmcp-client/src/event_notification_transport.rs b/codex-rs/rmcp-client/src/event_notification_transport.rs index a0ff2cd5bb..f997376b4c 100644 --- a/codex-rs/rmcp-client/src/event_notification_transport.rs +++ b/codex-rs/rmcp-client/src/event_notification_transport.rs @@ -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(); } } diff --git a/codex-rs/tui/src/app/app_server_event_targets.rs b/codex-rs/tui/src/app/app_server_event_targets.rs index 757fa8898a..19bb5db25f 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/codex-rs/tui/src/app/app_server_event_targets.rs @@ -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(_) diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index 1f1408f13a..8d81edbe0b 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -222,6 +222,7 @@ impl ChatWidget { | ServerNotification::CommandExecOutputDelta(_) | ServerNotification::ProcessOutputDelta(_) | ServerNotification::ProcessExited(_) + | ServerNotification::McpServerEventStream(_) | ServerNotification::FileChangePatchUpdated(_) | ServerNotification::McpToolCallProgress(_) | ServerNotification::McpServerOauthLoginCompleted(_)