Preserve MCP registration scope across client operations (#61257)

## Why

MCP servers registered under the same name in different environments need distinct inventory, routing, and status reporting.

## What changed

- Keep MCP inventory and runtime status keyed by registration scope so same-named servers remain separate.
- Add optional `environmentId` selectors for status lookup, tool calls, resource reads, and OAuth login. Omitting the selector preserves existing name-only routing.
- Carry environment identity through startup notifications, elicitation requests, and tool-call history, and display it in the TUI.
- Support `/mcp login <name> --environment <id>` and update generated protocol schemas and SDK types.

## Testing

Add integration coverage for two environments with identically named MCP servers, tools, and resource URIs. Verify separate inventory, explicit environment routing, legacy routing when the selector is omitted, and errors for unknown environments. Update protocol and TUI tests for environment identity.

GitOrigin-RevId: 074a3db10549438cb6aff33eb7bed10791adae4d
This commit is contained in:
sayan-oai
2026-11-13 04:51:06 +00:00
committed by copyberry
parent 7eaaee4270
commit 6fcfa75a9f
190 changed files with 1821 additions and 502 deletions
+1
View File
@@ -4292,6 +4292,7 @@ mod tests {
ThreadItemPayload::McpToolCall {
id: keys[2].item_id.clone(),
server: "server".to_string(),
environment_id: None,
tool: "tool".to_string(),
status: McpToolCallStatus::Failed,
arguments: serde_json::json!({}),
@@ -237,6 +237,7 @@ async fn cua_tool_call_lifecycle_requires_a_start_and_deduplicates_replay() {
let item = |status| ThreadItemPayload::McpToolCall {
id: "cua-call".into(),
server: "codex_apps".into(),
environment_id: None,
tool: "js".into(),
status,
arguments: json!({"script": "synthetic-private-input"}),
@@ -797,6 +798,7 @@ async fn mcp_elicitation_classification_survives_turn_completion_and_preserves_c
let item = ThreadItemPayload::McpToolCall {
id: item_id.to_string(),
server: "server".to_string(),
environment_id: None,
tool: "search".to_string(),
status: McpToolCallStatus::Completed,
arguments: json!({ "token": "synthetic-private-input" }),
@@ -490,6 +490,7 @@ async fn turn_event_counts_completed_tool_items() {
let mcp_tool_call_item = |status, duration_ms| ThreadItemPayload::McpToolCall {
id: "mcp-1".to_string(),
server: "server".to_string(),
environment_id: None,
tool: "search".to_string(),
status,
arguments: json!({}),
@@ -1913,6 +1913,13 @@
],
"description": "Controls how much MCP inventory data to fetch for each server. Defaults to `Full` when omitted."
},
"environmentId": {
"description": "Select this environment's registration. Requires `serverName` and `threadId`. When omitted, a named request uses the same server as older clients.",
"type": [
"string",
"null"
]
},
"limit": {
"description": "Optional page size; defaults to a server-defined value.",
"format": "uint32",
@@ -2240,6 +2247,13 @@
"null"
]
},
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"originCallId": {
"description": "Originating MCP tool call used to select the resource's app.",
"type": [
@@ -2328,6 +2342,13 @@
],
"description": "Registration strategy for this login only; omission selects automatic discovery."
},
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"name": {
"type": "string"
},
@@ -2370,6 +2391,13 @@
"properties": {
"_meta": true,
"arguments": true,
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"server": {
"type": "string"
},
@@ -628,6 +628,13 @@
}
],
"properties": {
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"serverName": {
"type": "string"
},
@@ -3117,6 +3117,13 @@
},
"McpServerStatusUpdatedNotification": {
"properties": {
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"type": [
"string",
@@ -5292,6 +5299,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -1493,6 +1493,13 @@
}
],
"properties": {
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"serverName": {
"type": "string"
},
@@ -4349,6 +4349,13 @@
}
],
"properties": {
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"serverName": {
"type": "string"
},
@@ -14991,6 +14998,13 @@
],
"description": "Controls how much MCP inventory data to fetch for each server. Defaults to `Full` when omitted."
},
"environmentId": {
"description": "Select this environment's registration. Requires `serverName` and `threadId`. When omitted, a named request uses the same server as older clients.",
"type": [
"string",
"null"
]
},
"limit": {
"description": "Optional page size; defaults to a server-defined value.",
"format": "uint32",
@@ -15730,6 +15744,13 @@
"null"
]
},
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"originCallId": {
"description": "Originating MCP tool call used to select the resource's app.",
"type": [
@@ -15962,6 +15983,13 @@
],
"description": "Registration strategy for this login only; omission selects automatic discovery."
},
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"name": {
"type": "string"
},
@@ -16040,6 +16068,13 @@
"authStatus": {
"$ref": "#/definitions/v2/McpAuthStatus"
},
"environmentId": {
"description": "Environment that contributed this registration; null for a thread-wide server. Older servers omit this field.",
"type": [
"string",
"null"
]
},
"httpOrigin": {
"description": "HTTP origin of the effective configured endpoint, including plugin servers. Excludes credentials, path, query, and fragment; null for non-HTTP transports.",
"type": [
@@ -16128,6 +16163,13 @@
"McpServerStatusUpdatedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"type": [
"string",
@@ -16169,6 +16211,13 @@
"properties": {
"_meta": true,
"arguments": true,
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"server": {
"type": "string"
},
@@ -24111,6 +24160,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -10952,6 +10952,13 @@
],
"description": "Controls how much MCP inventory data to fetch for each server. Defaults to `Full` when omitted."
},
"environmentId": {
"description": "Select this environment's registration. Requires `serverName` and `threadId`. When omitted, a named request uses the same server as older clients.",
"type": [
"string",
"null"
]
},
"limit": {
"description": "Optional page size; defaults to a server-defined value.",
"format": "uint32",
@@ -11691,6 +11698,13 @@
"null"
]
},
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"originCallId": {
"description": "Originating MCP tool call used to select the resource's app.",
"type": [
@@ -11923,6 +11937,13 @@
],
"description": "Registration strategy for this login only; omission selects automatic discovery."
},
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"name": {
"type": "string"
},
@@ -12001,6 +12022,13 @@
"authStatus": {
"$ref": "#/definitions/McpAuthStatus"
},
"environmentId": {
"description": "Environment that contributed this registration; null for a thread-wide server. Older servers omit this field.",
"type": [
"string",
"null"
]
},
"httpOrigin": {
"description": "HTTP origin of the effective configured endpoint, including plugin servers. Excludes credentials, path, query, and fragment; null for non-HTTP transports.",
"type": [
@@ -12089,6 +12117,13 @@
"McpServerStatusUpdatedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"type": [
"string",
@@ -12130,6 +12165,13 @@
"properties": {
"_meta": true,
"arguments": true,
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"server": {
"type": "string"
},
@@ -21814,6 +21856,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -1414,6 +1414,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -1414,6 +1414,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -28,6 +28,13 @@
],
"description": "Controls how much MCP inventory data to fetch for each server. Defaults to `Full` when omitted."
},
"environmentId": {
"description": "Select this environment's registration. Requires `serverName` and `threadId`. When omitted, a named request uses the same server as older clients.",
"type": [
"string",
"null"
]
},
"limit": {
"description": "Optional page size; defaults to a server-defined value.",
"format": "uint32",
@@ -69,6 +69,13 @@
"authStatus": {
"$ref": "#/definitions/McpAuthStatus"
},
"environmentId": {
"description": "Environment that contributed this registration; null for a thread-wide server. Older servers omit this field.",
"type": [
"string",
"null"
]
},
"httpOrigin": {
"description": "HTTP origin of the effective configured endpoint, including plugin servers. Excludes credentials, path, query, and fragment; null for non-HTTP transports.",
"type": [
@@ -36,6 +36,13 @@
"null"
]
},
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"originCallId": {
"description": "Originating MCP tool call used to select the resource's app.",
"type": [
@@ -22,6 +22,13 @@
],
"description": "Registration strategy for this login only; omission selects automatic discovery."
},
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"name": {
"type": "string"
},
@@ -18,6 +18,13 @@
}
},
"properties": {
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"type": [
"string",
@@ -3,6 +3,13 @@
"properties": {
"_meta": true,
"arguments": true,
"environmentId": {
"description": "Exact registration scope; omission preserves legacy name-only selection.",
"type": [
"string",
"null"
]
},
"server": {
"type": "string"
},
@@ -1626,6 +1626,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2240,6 +2240,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -1414,6 +1414,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2031,6 +2031,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2031,6 +2031,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2031,6 +2031,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2292,6 +2292,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2031,6 +2031,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2240,6 +2240,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2031,6 +2031,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -1626,6 +1626,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -2031,6 +2031,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -1626,6 +1626,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -1626,6 +1626,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -1626,6 +1626,13 @@
"null"
]
},
"environmentId": {
"description": "Registration scope, not necessarily the server's execution environment.",
"type": [
"string",
"null"
]
},
"error": {
"anyOf": [
{
@@ -20,4 +20,9 @@ detail?: McpServerStatusDetail | null, threadId?: string | null,
/**
* Limit discovery to one server. With a thread ID, reuse that thread's MCP connection.
*/
serverName?: string | null, };
serverName?: string | null,
/**
* Select this environment's registration. Requires `serverName` and `threadId`.
* When omitted, a named request uses the same server as older clients.
*/
environmentId?: string | null, };
@@ -12,7 +12,11 @@ _meta?: { [key in string]?: JsonValue } | null, threadId?: string | null,
/**
* Originating MCP tool call used to select the resource's app.
*/
originCallId?: string | null, server: string, uri: string, connectorId?: string | null,
originCallId?: string | null, server: string,
/**
* Exact registration scope; omission preserves legacy name-only selection.
*/
environmentId?: string | null, uri: string, connectorId?: string | null,
/**
* Explicit hosted app/account. Omit to retain legacy resource discovery.
*/
@@ -13,4 +13,8 @@ export type McpServerElicitationRequestParams = { threadId: string,
* context is app-server correlation rather than part of the protocol identity of the
* elicitation itself.
*/
turnId: string | null, serverName: string, } & ({ "mode": "form", _meta: JsonValue | null, message: string, requestedSchema: McpElicitationSchema, } | { "mode": "openai/form", _meta: JsonValue | null, message: string, requestedSchema: JsonValue, } | { "mode": "openaiForm", _meta: JsonValue | null, message: string, requestedSchema: JsonValue, } | { "mode": "url", _meta: JsonValue | null, message: string, url: string, elicitationId: string, });
turnId: string | null, serverName: string,
/**
* Registration scope, not necessarily the server's execution environment.
*/
environmentId: string | null, } & ({ "mode": "form", _meta: JsonValue | null, message: string, requestedSchema: McpElicitationSchema, } | { "mode": "openai/form", _meta: JsonValue | null, message: string, requestedSchema: JsonValue, } | { "mode": "openaiForm", _meta: JsonValue | null, message: string, requestedSchema: JsonValue, } | { "mode": "url", _meta: JsonValue | null, message: string, url: string, elicitationId: string, });
@@ -3,7 +3,11 @@
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
import type { McpServerOauthClientRegistration } from "./McpServerOauthClientRegistration";
export type McpServerOauthLoginParams = { name: string, threadId?: string | null,
export type McpServerOauthLoginParams = { name: string,
/**
* Exact registration scope; omission preserves legacy name-only selection.
*/
environmentId?: string | null, threadId?: string | null,
/**
* Registration strategy for this login only; omission selects automatic discovery.
*/
@@ -10,6 +10,11 @@ import type { McpAuthStatus } from "./McpAuthStatus";
import type { McpServerConnectionStatus } from "./McpServerConnectionStatus";
export type McpServerStatus = { name: string,
/**
* Environment that contributed this registration; null for a thread-wide server.
* Older servers omit this field.
*/
environmentId?: string | null,
/**
* Current thread-runtime connection state; null when unavailable or the configuration changed.
*/
@@ -4,4 +4,8 @@
import type { McpServerStartupFailureReason } from "./McpServerStartupFailureReason";
import type { McpServerStartupState } from "./McpServerStartupState";
export type McpServerStatusUpdatedNotification = { threadId: string | null, name: string, status: McpServerStartupState, error: string | null, failureReason: McpServerStartupFailureReason | null, };
export type McpServerStatusUpdatedNotification = { threadId: string | null, name: string,
/**
* Registration scope, not necessarily the server's execution environment.
*/
environmentId: string | null, status: McpServerStartupState, error: string | null, failureReason: McpServerStartupFailureReason | null, };
@@ -3,4 +3,8 @@
// 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 McpServerToolCallParams = { threadId: string, server: string, tool: string, arguments?: JsonValue, _meta?: JsonValue, };
export type McpServerToolCallParams = { threadId: string, server: string,
/**
* Exact registration scope; omission preserves legacy name-only selection.
*/
environmentId?: string | null, tool: string, arguments?: JsonValue, _meta?: JsonValue, };
@@ -81,7 +81,11 @@ exitCode: number | null,
/**
* The duration of the command execution in milliseconds.
*/
durationMs: number | null, } | { "type": "fileChange", id: string, changes: Array<FileUpdateChange>, status: PatchApplyStatus, } | { "type": "mcpToolCall", id: string, server: string, tool: string, status: McpToolCallStatus, arguments: JsonValue, appContext: McpToolCallAppContext | null,
durationMs: number | null, } | { "type": "fileChange", id: string, changes: Array<FileUpdateChange>, status: PatchApplyStatus, } | { "type": "mcpToolCall", id: string, server: string,
/**
* Registration scope, not necessarily the server's execution environment.
*/
environmentId: string | null, tool: string, status: McpToolCallStatus, arguments: JsonValue, appContext: McpToolCallAppContext | null,
/**
* Legacy compatibility field; prefer `mcpAppUi.resourceUri` when available.
*/
@@ -2437,6 +2437,8 @@ mod tests {
&& matches!(field_prefix.trim(), "startedAtMs?" | "completedAtMs?"))
|| (path == Path::new("v2/McpServerOauthLoginCompletedNotification.ts")
&& field_prefix.trim() == "loginId?")
|| (path == Path::new("v2/McpServerStatus.ts")
&& field_prefix.trim() == "environmentId?")
|| (path == Path::new("v2/ProcessExitedNotification.ts")
&& field_prefix.trim() == "outputDrainTimedOut?")
|| (path == Path::new("v2/MisalignmentErrorDetails.ts")
@@ -2586,6 +2586,7 @@ mod tests {
let mcp_oauth = ClientRequest::McpServerOauthLogin {
request_id: request_id(),
params: v2::McpServerOauthLoginParams {
environment_id: None,
name: "server-a".to_string(),
thread_id: None,
client_registration: None,
@@ -2603,6 +2604,7 @@ mod tests {
let mcp_resource_read = ClientRequest::McpResourceRead {
request_id: request_id(),
params: v2::McpResourceReadParams {
environment_id: None,
meta: None,
thread_id: Some("thread-1".to_string()),
origin_call_id: None,
@@ -2799,6 +2801,7 @@ mod tests {
let mcp_resource_read = ClientRequest::McpResourceRead {
request_id: request_id(),
params: v2::McpResourceReadParams {
environment_id: None,
meta: None,
thread_id: None,
origin_call_id: None,
@@ -3193,6 +3196,7 @@ mod tests {
thread_id: "thr_123".to_string(),
turn_id: Some("turn_123".to_string()),
server_name: "codex_apps".to_string(),
environment_id: None,
request: v2::McpServerElicitationRequest::Form {
meta: Some(json!({
"suggestion_id": "request_plugin_install_install-github"
@@ -3214,6 +3218,7 @@ mod tests {
"threadId": "thr_123",
"turnId": "turn_123",
"serverName": "codex_apps",
"environmentId": null,
"mode": "form",
"_meta": {
"suggestion_id": "request_plugin_install_install-github"
@@ -961,6 +961,7 @@ impl ThreadHistoryBuilder {
let item = ThreadItemPayload::McpToolCall {
id: payload.call_id.clone(),
server: payload.invocation.server.clone(),
environment_id: payload.invocation.environment_id.clone(),
tool: payload.invocation.tool.clone(),
status: McpToolCallStatus::InProgress,
arguments: payload
@@ -1019,6 +1020,7 @@ impl ThreadHistoryBuilder {
let item = ThreadItemPayload::McpToolCall {
id: payload.call_id.clone(),
server: payload.invocation.server.clone(),
environment_id: payload.invocation.environment_id.clone(),
tool: payload.invocation.tool.clone(),
status,
arguments: payload
@@ -3590,6 +3592,7 @@ mod tests {
call_id: "mcp-1".into(),
invocation: McpInvocation {
server: "docs".into(),
environment_id: None,
tool: "lookup".into(),
arguments: Some(serde_json::json!({"id":"123"})),
},
@@ -3655,6 +3658,7 @@ mod tests {
ThreadItemPayload::McpToolCall {
id: "mcp-1".into(),
server: "docs".into(),
environment_id: None,
tool: "lookup".into(),
status: McpToolCallStatus::Failed,
arguments: serde_json::json!({"id":"123"}),
@@ -3689,6 +3693,7 @@ mod tests {
call_id: "mcp-1".into(),
invocation: McpInvocation {
server: "docs".into(),
environment_id: None,
tool: "lookup".into(),
arguments: Some(serde_json::json!({"id":"123"})),
},
@@ -3726,6 +3731,7 @@ mod tests {
ThreadItemPayload::McpToolCall {
id: "mcp-1".into(),
server: "docs".into(),
environment_id: None,
tool: "lookup".into(),
status: McpToolCallStatus::Completed,
arguments: serde_json::json!({"id":"123"}),
@@ -4135,6 +4141,7 @@ mod tests {
item: CoreTurnItem::McpToolCall(McpToolCallItem {
id: "mcp-1".into(),
server: "slack".into(),
environment_id: None,
tool: "search".into(),
arguments: serde_json::json!({"query": "incident"}),
connector_id: None,
@@ -422,6 +422,8 @@ pub enum ThreadItemPayload {
McpToolCall {
id: String,
server: String,
/// Registration scope, not necessarily the server's execution environment.
environment_id: Option<String>,
tool: String,
status: McpToolCallStatus,
arguments: JsonValue,
@@ -1129,6 +1131,7 @@ impl From<CoreTurnItem> for ThreadItemPayload {
Self::McpToolCall {
id: mcp.id,
server: mcp.server,
environment_id: mcp.environment_id,
tool: mcp.tool,
status: McpToolCallStatus::from(mcp.status),
arguments: mcp.arguments,
@@ -63,6 +63,10 @@ pub struct ListMcpServerStatusParams {
/// Limit discovery to one server. With a thread ID, reuse that thread's MCP connection.
#[ts(optional = nullable)]
pub server_name: Option<String>,
/// Select this environment's registration. Requires `serverName` and `threadId`.
/// When omitted, a named request uses the same server as older clients.
#[ts(optional = nullable)]
pub environment_id: Option<String>,
}
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
@@ -78,6 +82,10 @@ pub enum McpServerStatusDetail {
#[ts(export_to = "v2/")]
pub struct McpServerStatus {
pub name: String,
/// Environment that contributed this registration; null for a thread-wide server.
/// Older servers omit this field.
#[ts(optional, as = "Option<Option<String>>")]
pub environment_id: Option<String>,
/// Current thread-runtime connection state; null when unavailable or the configuration changed.
pub runtime_status: Option<McpServerConnectionStatus>,
pub plugin_id: Option<String>,
@@ -122,6 +130,9 @@ pub struct McpResourceReadParams {
#[ts(optional = nullable)]
pub origin_call_id: Option<String>,
pub server: String,
/// Exact registration scope; omission preserves legacy name-only selection.
#[ts(optional = nullable)]
pub environment_id: Option<String>,
pub uri: String,
#[ts(optional = nullable)]
pub connector_id: Option<String>,
@@ -207,6 +218,9 @@ pub struct McpServerEventStreamNotification {
pub struct McpServerToolCallParams {
pub thread_id: String,
pub server: String,
/// Exact registration scope; omission preserves legacy name-only selection.
#[ts(optional = nullable)]
pub environment_id: Option<String>,
pub tool: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[ts(optional)]
@@ -299,6 +313,9 @@ pub struct McpServerRefreshResponse {}
#[ts(export_to = "v2/")]
pub struct McpServerOauthLoginParams {
pub name: String,
/// Exact registration scope; omission preserves legacy name-only selection.
#[ts(optional = nullable)]
pub environment_id: Option<String>,
#[ts(optional = nullable)]
pub thread_id: Option<String>,
/// Registration strategy for this login only; omission selects automatic discovery.
@@ -374,6 +391,8 @@ pub enum McpServerStartupState {
pub struct McpServerStatusUpdatedNotification {
pub thread_id: Option<String>,
pub name: String,
/// Registration scope, not necessarily the server's execution environment.
pub environment_id: Option<String>,
pub status: McpServerStartupState,
pub error: Option<String>,
pub failure_reason: Option<McpServerStartupFailureReason>,
@@ -433,6 +452,8 @@ pub struct McpServerElicitationRequestParams {
/// elicitation itself.
pub turn_id: Option<String>,
pub server_name: String,
/// Registration scope, not necessarily the server's execution environment.
pub environment_id: Option<String>,
#[serde(flatten)]
#[experimental(nested)]
pub request: McpServerElicitationRequest,
@@ -2694,6 +2694,7 @@ fn mcp_server_elicitation_response_serializes_nullable_content() {
fn mcp_server_status_serializes_absent_server_info_as_null() {
let response = ListMcpServerStatusResponse {
data: vec![McpServerStatus {
environment_id: None,
tool_list_meta: None,
server_capabilities: None,
tools_error: None,
@@ -2715,6 +2716,7 @@ fn mcp_server_status_serializes_absent_server_info_as_null() {
json!({
"data": [{
"name": "not-ready",
"environmentId": null,
"runtimeStatus": null,
"pluginId": null,
"httpOrigin": null,
@@ -2747,6 +2749,7 @@ fn mcp_server_status_accepts_older_inventory_without_runtime_status() {
assert_eq!(
status,
McpServerStatus {
environment_id: None,
tool_list_meta: None,
server_capabilities: None,
tools_error: None,
@@ -2775,6 +2778,7 @@ fn mcp_server_status_updated_accepts_missing_thread_id() {
let expected = McpServerStatusUpdatedNotification {
thread_id: None,
name: "optional_broken".to_string(),
environment_id: None,
status: McpServerStartupState::Failed,
error: Some("handshake failed".to_string()),
failure_reason: None,
@@ -2785,6 +2789,7 @@ fn mcp_server_status_updated_accepts_missing_thread_id() {
json!({
"threadId": null,
"name": "optional_broken",
"environmentId": null,
"status": "failed",
"error": "handshake failed",
"failureReason": null,
@@ -2798,6 +2803,7 @@ fn mcp_server_status_updated_serializes_failure_reason() {
ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification {
thread_id: Some("thread-1".to_string()),
name: "expired-oauth".to_string(),
environment_id: Some("remote".to_string()),
status: McpServerStartupState::Failed,
error: Some("OAuth credentials expired".to_string()),
failure_reason: Some(McpServerStartupFailureReason::ReauthenticationRequired),
@@ -2810,6 +2816,7 @@ fn mcp_server_status_updated_serializes_failure_reason() {
"params": {
"threadId": "thread-1",
"name": "expired-oauth",
"environmentId": "remote",
"status": "failed",
"error": "OAuth credentials expired",
"failureReason": "reauthenticationRequired",
@@ -2822,6 +2829,7 @@ fn mcp_server_status_updated_serializes_failure_reason() {
fn mcp_server_status_serializes_absent_server_info_metadata_as_null() {
let response = ListMcpServerStatusResponse {
data: vec![McpServerStatus {
environment_id: None,
tool_list_meta: None,
server_capabilities: None,
tools_error: None,
@@ -2850,6 +2858,7 @@ fn mcp_server_status_serializes_absent_server_info_metadata_as_null() {
json!({
"data": [{
"name": "initialized",
"environmentId": null,
"runtimeStatus": null,
"pluginId": "lookup@test",
"httpOrigin": null,
@@ -3714,6 +3723,7 @@ fn core_turn_item_into_thread_item_converts_supported_variants() {
let mcp_tool_call_item = TurnItem::McpToolCall(McpToolCallItem {
id: "mcp-1".to_string(),
server: "server".to_string(),
environment_id: Some("remote".to_string()),
tool: "tool".to_string(),
arguments: json!({"arg": "value"}),
connector_id: Some("calendar".to_string()),
@@ -3735,6 +3745,7 @@ fn core_turn_item_into_thread_item_converts_supported_variants() {
ThreadItemPayload::McpToolCall {
id: "mcp-1".to_string(),
server: "server".to_string(),
environment_id: Some("remote".to_string()),
tool: "tool".to_string(),
status: McpToolCallStatus::InProgress,
arguments: json!({"arg": "value"}),
@@ -3758,6 +3769,7 @@ fn core_turn_item_into_thread_item_converts_supported_variants() {
let completed_mcp_tool_call_item = TurnItem::McpToolCall(McpToolCallItem {
id: "mcp-2".to_string(),
server: "server".to_string(),
environment_id: None,
tool: "tool".to_string(),
arguments: JsonValue::Null,
connector_id: None,
@@ -3784,6 +3796,7 @@ fn core_turn_item_into_thread_item_converts_supported_variants() {
ThreadItemPayload::McpToolCall {
id: "mcp-2".to_string(),
server: "server".to_string(),
environment_id: None,
tool: "tool".to_string(),
status: McpToolCallStatus::Completed,
arguments: JsonValue::Null,
@@ -3821,6 +3834,7 @@ fn mcp_tool_call_app_context_serializes_connector_id() {
let item = ThreadItemPayload::McpToolCall {
id: "mcp-1".to_string(),
server: "codex_apps".to_string(),
environment_id: None,
tool: "calendar.create_event".to_string(),
status: McpToolCallStatus::InProgress,
arguments: json!({}),
@@ -3846,6 +3860,7 @@ fn mcp_tool_call_app_context_serializes_connector_id() {
"type": "mcpToolCall",
"id": "mcp-1",
"server": "codex_apps",
"environmentId": null,
"tool": "calendar.create_event",
"status": "inProgress",
"arguments": {},
+1 -3
View File
@@ -6,12 +6,10 @@ codex_rust_crate(
crate_name = "codex_app_server",
extra_binaries = [
"//codex-rs/bwrap:bwrap",
"//codex-rs/cli:codex",
"//codex-rs/code-mode-host:codex-code-mode-host",
"//codex-rs/rmcp-client:test_stdio_server",
],
extra_binaries_non_windows = [
"//codex-rs/cli:codex",
],
integration_test_timeout = "long",
run_tests_with_wine_exec = True,
test_shard_counts = {
@@ -226,6 +226,7 @@ pub(crate) async fn apply_bespoke_event_handling(
let notification = McpServerStatusUpdatedNotification {
thread_id: Some(conversation_id.to_string()),
name: update.server,
environment_id: update.environment_id,
status,
error,
failure_reason,
@@ -941,6 +942,7 @@ pub(crate) async fn apply_bespoke_event_handling(
thread_id: conversation_id.to_string(),
turn_id,
server_name: request.server_name.clone(),
environment_id: request.environment_id.clone(),
request: request_body,
};
let (pending_request_id, rx) = outgoing
@@ -137,6 +137,7 @@ impl McpRequestProcessor {
) -> Result<McpServerOauthLoginResponse, JSONRPCErrorError> {
let McpServerOauthLoginParams {
name,
environment_id,
thread_id,
client_registration,
scopes,
@@ -165,13 +166,12 @@ impl McpRequestProcessor {
}
};
let effective_servers = codex_mcp::effective_mcp_servers(&mcp_config, auth.as_ref());
let Some((key, server)) = mcp_config
let key = mcp_config
.mcp_server_catalog
.resolve_server_name(&name)
.and_then(|key| effective_servers.get(key).map(|server| (key, server)))
else {
.resolve_server_key(&name, environment_id.as_deref());
let Some(server) = effective_servers.get(&key) else {
return Err(invalid_request(format!(
"No MCP server named '{name}' found."
"No MCP server named '{key}' found."
)));
};
let redirect_mode = if server.is_agent_plugin() {
@@ -417,12 +417,22 @@ impl McpRequestProcessor {
(thread.as_ref(), params.server_name.as_deref())
{
thread
.mcp_server_status_snapshot(server, detail, /*expected_registration*/ None)
.mcp_server_status_snapshot(
server,
params.environment_id.as_deref(),
detail,
/*expected_registration*/ None,
)
.await
.map_err(|error| {
internal_error(format!("failed to read MCP server status: {error:#}"))
})?
} else {
if params.environment_id.is_some() {
return Err(invalid_request(
"environmentId requires serverName and threadId",
));
}
let config = match thread.as_ref() {
Some(thread) => {
let thread_config = thread.config().await;
@@ -455,6 +465,11 @@ impl McpRequestProcessor {
),
),
};
let selected_server = params.server_name.as_deref().map(|name| {
mcp_config
.mcp_server_catalog
.resolve_server_key(name, /*environment_id*/ None)
});
let mut snapshot = collect_mcp_server_status_snapshot_with_detail(
&mcp_config,
auth.as_ref(),
@@ -464,7 +479,7 @@ impl McpRequestProcessor {
mcp_manager.codex_apps_tools_cache(),
mcp_manager.tool_catalog_cache(),
detail,
params.server_name.as_deref(),
selected_server.as_ref(),
)
.await;
if mcp_config.xaa_enabled
@@ -472,16 +487,18 @@ impl McpRequestProcessor {
{
// EMA can only use its admitted thread connection. Keep ordinary
// status discovery unchanged and replace the threadless EMA results.
for (name, server) in mcp_config.mcp_server_catalog.configured_servers() {
if !matches!(server.auth, McpServerAuth::EmaAuth) {
for (key, server) in mcp_config.mcp_server_catalog.servers() {
if !matches!(server.config().auth, McpServerAuth::EmaAuth) {
continue;
}
let Some(key) = mcp_config.mcp_server_catalog.resolve_server_name(&name) else {
continue;
};
let registration = mcp_config.mcp_server_catalog.server(key);
let (selected_config, mut selected) = thread
.mcp_server_status_snapshot(&name, detail, registration)
let registration = Some(server);
let (selected_config, selected) = thread
.mcp_server_status_snapshot(
key.name(),
key.scoped_environment_id(),
detail,
registration,
)
.await
.map_err(|error| {
internal_error(format!("failed to read MCP server status: {error:#}"))
@@ -489,27 +506,25 @@ impl McpRequestProcessor {
if selected_config.mcp_server_catalog.server(key) != registration {
continue;
}
snapshot.server_infos.remove(&name);
snapshot.server_infos.remove(key);
snapshot.server_infos.extend(selected.server_infos);
snapshot.server_capabilities.remove(&name);
if let Some(capabilities) = selected.server_capabilities.remove(&name) {
snapshot
.server_capabilities
.insert(name.clone(), capabilities);
}
snapshot.tools_by_server.remove(&name);
snapshot.server_capabilities.remove(key);
snapshot
.server_capabilities
.extend(selected.server_capabilities);
snapshot.tools_by_server.remove(key);
snapshot.tools_by_server.extend(selected.tools_by_server);
snapshot.tool_list_meta.remove(&name);
snapshot.tool_list_meta.remove(key);
snapshot.tool_list_meta.extend(selected.tool_list_meta);
snapshot.tools_errors.remove(&name);
snapshot.tools_errors.remove(key);
snapshot.tools_errors.extend(selected.tools_errors);
snapshot.resources.remove(&name);
snapshot.resources.remove(key);
snapshot.resources.extend(selected.resources);
snapshot.resource_templates.remove(&name);
snapshot.resource_templates.remove(key);
snapshot
.resource_templates
.extend(selected.resource_templates);
snapshot.auth_statuses.remove(&name);
snapshot.auth_statuses.remove(key);
snapshot.auth_statuses.extend(selected.auth_statuses);
}
}
@@ -529,26 +544,27 @@ impl McpRequestProcessor {
mut resources,
mut resource_templates,
mut auth_statuses,
mut server_names,
mut server_keys,
} = snapshot;
server_names.extend(runtime_statuses.keys().cloned());
server_names.extend(
server_keys.extend(runtime_statuses.keys().cloned());
server_keys.extend(
auth_statuses
.keys()
.cloned()
.chain(resources.keys().cloned())
.chain(resource_templates.keys().cloned()),
);
server_names.sort();
server_names.dedup();
server_names.retain(|name| {
params
.server_name
.as_deref()
.is_none_or(|server| name == server)
server_keys.sort();
server_keys.dedup();
server_keys.retain(|key| {
params.server_name.as_deref().is_none_or(|server| {
*key == mcp_config
.mcp_server_catalog
.resolve_server_key(server, params.environment_id.as_deref())
})
});
let total = server_names.len();
let total = server_keys.len();
let limit = params.limit.unwrap_or(total as u32).max(1) as usize;
let effective_limit = limit.min(total);
let start = match params.cursor {
@@ -567,15 +583,14 @@ impl McpRequestProcessor {
let end = start.saturating_add(effective_limit).min(total);
let data: Vec<McpServerStatus> = server_names[start..end]
let data: Vec<McpServerStatus> = server_keys[start..end]
.iter()
.map(|name| McpServerStatus {
name: name.clone(),
runtime_status: runtime_statuses.get(name).copied().map(Into::into),
plugin_id: mcp_config
.mcp_server_catalog
.server_by_name(name)
.and_then(|server| match server.source() {
.map(|key| McpServerStatus {
name: key.name().to_owned(),
environment_id: key.scoped_environment_id().map(str::to_owned),
runtime_status: runtime_statuses.get(key).copied().map(Into::into),
plugin_id: mcp_config.mcp_server_catalog.server(key).and_then(
|server| match server.source() {
McpServerSource::Plugin(plugin)
| McpServerSource::SelectedPlugin(plugin) => {
Some(plugin.plugin_id().to_owned())
@@ -583,9 +598,12 @@ impl McpRequestProcessor {
McpServerSource::Config
| McpServerSource::Compatibility { .. }
| McpServerSource::Extension { .. } => None,
}),
http_origin: mcp_config.mcp_server_catalog.server_by_name(name).and_then(
|server| match &server.config().transport {
},
),
http_origin: mcp_config
.mcp_server_catalog
.server(key)
.and_then(|server| match &server.config().transport {
McpServerTransportConfig::StreamableHttp { url, .. } => {
url::Url::parse(url).ok().and_then(|url| {
matches!(url.scheme(), "http" | "https")
@@ -593,17 +611,16 @@ impl McpRequestProcessor {
})
}
McpServerTransportConfig::Stdio { .. } => None,
},
),
server_info: server_infos.remove(name),
server_capabilities: server_capabilities.remove(name),
tool_list_meta: tool_list_meta.remove(name).map(Arc::unwrap_or_clone),
tools: tools_by_server.remove(name).unwrap_or_default(),
tools_error: tools_errors.remove(name),
resources: resources.remove(name).unwrap_or_default(),
resource_templates: resource_templates.remove(name).unwrap_or_default(),
}),
server_info: server_infos.remove(key),
server_capabilities: server_capabilities.remove(key),
tool_list_meta: tool_list_meta.remove(key).map(Arc::unwrap_or_clone),
tools: tools_by_server.remove(key).unwrap_or_default(),
tools_error: tools_errors.remove(key),
resources: resources.remove(key).unwrap_or_default(),
resource_templates: resource_templates.remove(key).unwrap_or_default(),
auth_status: auth_statuses
.remove(name)
.remove(key)
.unwrap_or(CoreMcpAuthStatus::Unsupported)
.into(),
})
@@ -629,6 +646,7 @@ impl McpRequestProcessor {
thread_id,
origin_call_id,
server,
environment_id,
uri,
connector_id,
target,
@@ -636,6 +654,11 @@ impl McpRequestProcessor {
let mut resource_params = ReadResourceRequestParams::new(uri);
let mut meta = meta.unwrap_or_default();
let is_codex_apps = server == codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
if is_codex_apps && origin_call_id.is_some() && environment_id.is_some() {
return Err(invalid_request(
"originCallId requires the unscoped codex_apps registration",
));
}
let reserved_key = meta.iter().find_map(|(key, value)| match key.as_str() {
"x-codex-turn-metadata"
| MCP_THREAD_ID_META_KEY
@@ -664,6 +687,7 @@ impl McpRequestProcessor {
&& let Some(target) = target
{
if !is_codex_apps
|| environment_id.is_some()
|| target.connector_id.trim().is_empty()
|| target
.link_id
@@ -671,7 +695,7 @@ impl McpRequestProcessor {
.is_some_and(|id| id.trim().is_empty() || id.starts_with("synthetic_link::"))
{
return Err(invalid_request(
"target requires codex_apps, a connectorId, and a real linkId or null",
"target requires unscoped codex_apps, a connectorId, and a real linkId or null",
));
}
meta.insert(
@@ -696,7 +720,11 @@ impl McpRequestProcessor {
.read_mcp_resource_for_call(call_id, resource_params)
.await
}
None => thread.read_mcp_resource(&server, resource_params).await,
None => {
thread
.read_mcp_resource(&server, environment_id.as_deref(), resource_params)
.await
}
};
Self::send_mcp_resource_read_response(
outgoing,
@@ -741,6 +769,7 @@ impl McpRequestProcessor {
codex_apps_tools_cache,
tool_catalog_cache,
&server,
environment_id.as_deref(),
resource_params,
)
.await
@@ -793,7 +822,13 @@ impl McpRequestProcessor {
let task = async move {
let result = thread
.call_mcp_tool(&params.server, &params.tool, params.arguments, meta)
.call_mcp_tool(
&params.server,
params.environment_id.as_deref(),
&params.tool,
params.arguments,
meta,
)
.await
.map(McpServerToolCallResponse::from)
.map_err(mcp_operation_error);
@@ -83,6 +83,7 @@ mod tests {
ThreadItemPayload::McpToolCall {
id: "mcp-1".to_string(),
server: "docs".to_string(),
environment_id: None,
tool: "lookup".to_string(),
status: McpToolCallStatus::Completed,
arguments: serde_json::json!({"secret":"argument"}),
@@ -147,6 +148,7 @@ mod tests {
ThreadItemPayload::McpToolCall {
id: "mcp-1".to_string(),
server: "docs".to_string(),
environment_id: None,
tool: "lookup".to_string(),
status: McpToolCallStatus::Completed,
arguments: JsonValue::String(REDACTED_PAYLOAD.to_string()),
@@ -173,6 +175,7 @@ mod tests {
let mut thread = test_thread(vec![ThreadItemPayload::McpToolCall {
id: "mcp-1".to_string(),
server: "docs".to_string(),
environment_id: None,
tool: "lookup".to_string(),
status: McpToolCallStatus::Failed,
arguments: serde_json::json!({"secret":"argument"}),
@@ -195,6 +198,7 @@ mod tests {
ThreadItemPayload::McpToolCall {
id: "mcp-1".to_string(),
server: "docs".to_string(),
environment_id: None,
tool: "lookup".to_string(),
status: McpToolCallStatus::Failed,
arguments: JsonValue::String(REDACTED_PAYLOAD.to_string()),
@@ -77,6 +77,7 @@ async fn user_verification_initialize_owns_advertisement_and_eligibility() -> Re
thread_id: thread_id.to_string(),
turn_id: None,
server_name: "codex_apps".into(),
environment_id: None,
request: rpc::McpServerElicitationRequest::UserVerification {
meta: None,
challenge: "AQ".into(),
@@ -23,6 +23,7 @@ fn request() -> rpc::ServerRequestPayload {
thread_id: ThreadId::new().to_string(),
turn_id: None,
server_name: "codex_apps".into(),
environment_id: None,
request: rpc::McpServerElicitationRequest::UserVerification {
meta: None,
title: "Approve".into(),
@@ -17,6 +17,7 @@ fn request(thread_id: ThreadId) -> ServerRequestPayload {
thread_id: thread_id.to_string(),
turn_id: Some("turn-1".into()),
server_name: "plugin-service".into(),
environment_id: None,
request: McpServerElicitationRequest::UserVerification {
meta: None,
title: "Approve".into(),
@@ -239,6 +239,7 @@ async fn existing_thread_loads_api_curated_mcp_after_auth_switch_sync() -> Resul
.request(|request_id| ClientRequest::McpServerToolCall {
request_id,
params: McpServerToolCallParams {
environment_id: None,
thread_id: thread_id.clone(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -205,6 +205,7 @@ async fn daybreak_access_respects_plugin_provenance(case: AccessCall) -> Result<
.request(|request_id| ClientRequest::McpServerToolCall {
request_id,
params: McpServerToolCallParams {
environment_id: None,
thread_id: thread.id.clone(),
server: "sample".to_string(),
tool: "get_daybreak_access".to_string(),
@@ -1382,6 +1382,7 @@ startup_timeout_sec = 10
let request_id = app_server
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: selected_thread.clone(),
server: HTTP_MCP_SERVER_NAME.to_string(),
tool: "echo".to_string(),
@@ -1398,6 +1399,7 @@ startup_timeout_sec = 10
let request_id = app_server
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: selected_thread.clone(),
server: OAUTH_MCP_SERVER_NAME.to_string(),
tool: "echo".to_string(),
@@ -1443,6 +1445,7 @@ startup_timeout_sec = 10
let request_id = app_server
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: selected_thread.clone(),
server: REFRESH_PROBE_SERVER_NAME.to_string(),
tool: "echo".to_string(),
@@ -1562,6 +1565,7 @@ async fn mcp_server_statuses(
) -> Result<Vec<McpServerStatus>> {
let request_id = app_server
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -230,6 +230,7 @@ async fn spawned_agent_inherits_selected_executor_skill(
}
let status_id = app_server
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
environment_id: None,
thread_id: Some(request.header("thread-id").expect("agent thread ID")),
server_name: None,
cursor: None,
@@ -155,6 +155,7 @@ async fn call_servers(
.request(|request_id| ClientRequest::McpServerToolCall {
request_id,
params: McpServerToolCallParams {
environment_id: None,
thread_id: thread_id.to_string(),
server: name.to_string(),
tool: "echo".to_string(),
@@ -50,6 +50,7 @@ async fn enabling_hosted_apps_protocol_refreshes_an_existing_thread() -> Result<
.id;
let call_tool = |thread_id: String| McpServerToolCallParams {
environment_id: None,
thread_id,
server: "codex_apps".to_string(),
tool: "calendar_list_events".to_string(),
@@ -190,6 +190,7 @@ async fn mcp_resource_read_returns_resource_contents() -> Result<()> {
.request(|request_id| ClientRequest::McpResourceRead {
request_id,
params: McpResourceReadParams {
environment_id: None,
meta: Some(serde_json::Map::from_iter([(
"example/trace".into(),
metadata_value.clone(),
@@ -239,6 +240,7 @@ async fn mcp_resource_read_returns_resource_contents() -> Result<()> {
] {
let request_id = mcp
.send_mcp_resource_read_request(McpResourceReadParams {
environment_id: None,
thread_id: Some(thread.id.clone()),
origin_call_id,
meta: Some(serde_json::Map::from_iter([(key.into(), value)])),
@@ -314,6 +316,7 @@ async fn mcp_resource_read_preserves_protocol_errors(protocol: ProtocolVersion)
for thread_id in [Some(thread.id.clone()), None] {
let request_id = mcp
.send_mcp_resource_read_request(McpResourceReadParams {
environment_id: None,
meta: None,
thread_id,
origin_call_id: None,
@@ -531,6 +534,7 @@ apps = true
.request(|request_id| ClientRequest::McpResourceRead {
request_id,
params: McpResourceReadParams {
environment_id: None,
meta: None,
thread_id: None,
origin_call_id: None,
@@ -546,6 +550,7 @@ apps = true
.request(|request_id| ClientRequest::McpResourceRead {
request_id,
params: McpResourceReadParams {
environment_id: None,
meta: None,
thread_id: None,
origin_call_id: None,
@@ -589,6 +594,7 @@ async fn mcp_resource_read_returns_error_for_unknown_thread() -> Result<()> {
.request(ClientRequest::McpResourceRead {
request_id: RequestId::Integer(1),
params: McpResourceReadParams {
environment_id: None,
meta: None,
thread_id: Some("00000000-0000-4000-8000-000000000000".to_string()),
origin_call_id: None,
@@ -717,6 +723,7 @@ async fn metadata_and_mcp_requests_complete_while_unrelated_resume_loads_config(
sender.request(ClientRequest::McpResourceRead {
request_id: RequestId::Integer(7),
params: McpResourceReadParams {
environment_id: None,
meta: None,
thread_id: Some(thread.id.clone()),
origin_call_id: None,
@@ -729,6 +736,7 @@ async fn metadata_and_mcp_requests_complete_while_unrelated_resume_loads_config(
sender.request(ClientRequest::McpServerToolCall {
request_id: RequestId::Integer(8),
params: McpServerToolCallParams {
environment_id: None,
thread_id: thread.id.clone(),
server: "resource_server".to_string(),
tool: "best_buy_product_search".to_string(),
@@ -247,6 +247,7 @@ async fn list_plugin_skills(
async fn read_mcp_skill_resource(app_server: &mut TestAppServer, thread_id: &str) -> Result<()> {
let request = app_server
.send_mcp_resource_read_request(McpResourceReadParams {
environment_id: None,
meta: None,
thread_id: Some(thread_id.to_string()),
origin_call_id: None,
@@ -173,6 +173,7 @@ async fn widget_reads_survive_history_modes_compaction_restarts_and_app_only_vis
] {
let request_id = app_server
.send_mcp_resource_read_request(McpResourceReadParams {
environment_id: None,
meta: None,
thread_id,
origin_call_id: Some(call_id.to_string()),
@@ -298,6 +299,7 @@ async fn read_widget(
.request(|request_id| ClientRequest::McpResourceRead {
request_id,
params: McpResourceReadParams {
environment_id: None,
meta: Some(serde_json::Map::from_iter([(
"example/trace".into(),
json!(call_id),
@@ -200,6 +200,7 @@ async fn assert_standard_form_elicitation_round_trip(
thread_id: fixture.thread_id.clone(),
turn_id: Some(fixture.turn_id.clone()),
server_name: "codex_apps".to_string(),
environment_id: None,
request: McpServerElicitationRequest::Form {
meta: None,
message: ELICITATION_MESSAGE.to_string(),
@@ -246,6 +247,7 @@ async fn mcp_server_openai_form_elicitation_round_trip() -> Result<()> {
thread_id: fixture.thread_id.clone(),
turn_id: Some(fixture.turn_id.clone()),
server_name: "codex_apps".to_string(),
environment_id: None,
request: McpServerElicitationRequest::OpenAiForm {
meta: None,
message: OPENAI_FORM_MESSAGE.to_string(),
@@ -285,6 +287,7 @@ async fn mcp_server_openai_elicitation_form_round_trip() -> Result<()> {
thread_id: fixture.thread_id.clone(),
turn_id: Some(fixture.turn_id.clone()),
server_name: "codex_apps".to_string(),
environment_id: None,
request: McpServerElicitationRequest::OpenAiElicitationForm {
meta: Some(json!({ "example/request": "template-picker" })),
message: OPENAI_FORM_MESSAGE.to_string(),
@@ -23,6 +23,7 @@ use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::ListMcpServerStatusParams;
use codex_app_server_protocol::ListMcpServerStatusResponse;
use codex_app_server_protocol::McpAuthStatus;
use codex_app_server_protocol::McpServerConnectionStatus;
use codex_app_server_protocol::McpServerOauthLoginCompletedNotification;
use codex_app_server_protocol::McpServerOauthLoginResponse;
@@ -32,7 +33,9 @@ use codex_app_server_protocol::PluginInstallResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnEnvironmentParams;
use codex_core::config::set_project_trust_level;
use codex_features::Feature;
use codex_http_client::HttpClientBuilder;
use codex_protocol::config_types::TrustLevel;
use codex_rmcp_client::McpOAuthCallbackMode;
@@ -43,12 +46,18 @@ use core_test_support::skip_if_remote;
use core_test_support::stdio_server_bin;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
use rmcp::model::CallToolRequestParams;
use rmcp::model::CallToolResult;
use rmcp::model::ContentBlock;
use rmcp::model::Implementation;
use rmcp::model::JsonObject;
use rmcp::model::ListResourceTemplatesResult;
use rmcp::model::ListResourcesResult;
use rmcp::model::ListToolsResult;
use rmcp::model::PaginatedRequestParams;
use rmcp::model::ReadResourceRequestParams;
use rmcp::model::ReadResourceResult;
use rmcp::model::ResourceContents;
use rmcp::model::ServerCapabilities;
use rmcp::model::ServerConfig;
use rmcp::model::Tool;
@@ -142,6 +151,7 @@ async fn mcp_server_status_list_reuses_thread_connection() -> Result<()> {
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
thread_id: Some(thread.id.clone()),
server_name: Some("widget".to_string()),
detail: Some(detail),
@@ -203,6 +213,7 @@ async fn full_mcp_server_status_skips_codex_apps_resource_templates(
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
cursor: None,
limit: None,
detail: Some(McpServerStatusDetail::Full),
@@ -558,11 +569,13 @@ async fn oauth_login_automatically_selects_callback_specific_cimd_without_metada
let (token_request_tx, mut token_request_rx) = mpsc::unbounded_channel();
let (mcp_authorization_tx, mut mcp_authorization_rx) = mpsc::unbounded_channel();
let tool_name = Arc::new("cimd".to_string());
let mcp_origin = base_url.clone();
let mcp_service = StreamableHttpService::new(
move || {
Ok(McpStatusServer {
tool_name: Arc::clone(&tool_name),
tools_error: None,
origin: mcp_origin.clone(),
})
},
Arc::new(LocalSessionManager::default()),
@@ -728,6 +741,7 @@ async fn oauth_login_automatically_selects_callback_specific_cimd_without_metada
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -818,6 +832,7 @@ async fn mcp_server_status_list_returns_raw_server_and_tool_names(plugin: bool)
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -925,6 +940,7 @@ MCP_TEST_PID_FILE = {}
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -939,6 +955,7 @@ MCP_TEST_PID_FILE = {}
std::fs::remove_file(&barrier_file)?;
let second_request_id = mcp
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1003,6 +1020,7 @@ url = "{mcp_server_url}/mcp"
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1017,6 +1035,7 @@ url = "{mcp_server_url}/mcp"
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1073,6 +1092,7 @@ async fn mcp_server_status_list_reports_thread_runtime_connections() -> Result<(
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1112,6 +1132,7 @@ async fn mcp_server_status_list_reports_thread_runtime_connections() -> Result<(
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1196,6 +1217,7 @@ async fn mcp_server_status_list_reports_disconnected_stdio_transport() -> Result
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1244,6 +1266,7 @@ async fn mcp_server_status_retains_capabilities_when_tool_discovery_fails() -> R
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1283,11 +1306,15 @@ async fn mcp_server_status_retains_capabilities_when_tool_discovery_fails() -> R
struct McpStatusServer {
tool_name: Arc<String>,
tools_error: Option<rmcp::ErrorData>,
origin: String,
}
impl ServerHandler for McpStatusServer {
fn get_info(&self) -> ServerConfig {
let mut capabilities = ServerCapabilities::builder().enable_tools().build();
let mut capabilities = ServerCapabilities::builder()
.enable_tools()
.enable_resources()
.build();
capabilities.extensions = Some(BTreeMap::from([(
"openai/settings".to_string(),
serde_json::from_value(
@@ -1323,6 +1350,31 @@ impl ServerHandler for McpStatusServer {
Ok(ListToolsResult::with_all_items(vec![tool]))
}
async fn call_tool(
&self,
request: CallToolRequestParams,
_context: RequestContext<rmcp::service::RoleServer>,
) -> Result<rmcp::model::CallToolResponse, rmcp::ErrorData> {
assert_eq!(request.name.as_ref(), self.tool_name.as_str());
Ok(CallToolResult::success(vec![ContentBlock::text(self.origin.clone())]).into())
}
async fn read_resource(
&self,
request: ReadResourceRequestParams,
_context: RequestContext<rmcp::service::RoleServer>,
) -> Result<rmcp::model::ReadResourceResponse, rmcp::ErrorData> {
Ok(
ReadResourceResult::new(vec![ResourceContents::TextResourceContents {
uri: request.uri,
mime_type: None,
text: self.origin.clone(),
meta: None,
}])
.into(),
)
}
}
#[derive(Clone)]
@@ -1399,6 +1451,7 @@ async fn mcp_server_status_list_tools_and_auth_only_skips_slow_inventory_calls()
let request_id = mcp
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1454,6 +1507,7 @@ url = "{underscore_server_url}/mcp"
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -1494,6 +1548,210 @@ url = "{underscore_server_url}/mcp"
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mcp_server_status_keeps_same_named_environment_inventory_separate() -> Result<()> {
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let (first_url, first_handle) = start_mcp_server("lookup", /*tools_error*/ None).await?;
let (second_url, second_handle) = start_mcp_server("lookup", /*tools_error*/ None).await?;
let codex_home = TempDir::new()?;
mock_responses_config(&server.uri())
.enable_feature(Feature::MultiEnvironmentMcp)
.enable_feature(Feature::ExecutorCapabilityDiscovery)
.write(codex_home.path())?;
let executors = [
("env-a", TempDir::new()?, &first_url),
("env-b", TempDir::new()?, &second_url),
];
let program = toml::Value::String(
codex_utils_cargo_bin::cargo_bin("codex")?
.to_string_lossy()
.into_owned(),
);
let mut environment_config = "default = \"env-a\"\ninclude_local = false\n".to_string();
let mut environments = Vec::new();
for (id, home, url) in &executors {
let headers = if *id == "env-b" {
"http_headers = { Authorization = \"Bearer test-token\" }\n"
} else {
""
};
std::fs::write(
home.path().join("config.toml"),
format!("[mcp_servers.notes]\nurl = \"{url}/mcp\"\n{headers}"),
)?;
let home_value = toml::Value::String(home.path().to_string_lossy().into_owned());
environment_config.push_str(&format!(
"\n[[environments]]\nid = \"{id}\"\nprogram = {program}\nargs = [\"exec-server\", \"--listen\", \"stdio\"]\n[environments.env]\nCODEX_HOME = {home_value}\n"
));
environments.push(TurnEnvironmentParams {
environment_id: (*id).to_string(),
cwd: AbsolutePathBuf::try_from(home.path().to_path_buf())?.into(),
runtime_workspace_roots: None,
});
}
std::fs::write(
codex_home.path().join("environments.toml"),
environment_config,
)?;
let mut app = TestAppServer::builder()
.with_codex_home(codex_home.path())
// This test owns both executor environments and their configurations.
.without_auto_env()
.build_initialized()
.await?;
let request = app
.send_thread_start_request(ThreadStartParams {
environments: Some(environments),
..Default::default()
})
.await?;
let ThreadStartResponse { thread, .. } = app.read_response(request).await?;
let notification = app
.read_stream_until_matching_notification("notes MCP ready", |notification| {
notification.method == "mcpServer/startupStatus/updated"
&& notification.params.as_ref().is_some_and(|params| {
params["threadId"] == thread.id
&& params["name"] == "notes"
&& params["status"] == "ready"
})
})
.await?;
assert_eq!(
notification.params,
Some(serde_json::json!({
"threadId": thread.id,
"name": "notes",
"environmentId": "env-a",
"status": "ready",
"error": null,
"failureReason": null,
}))
);
let response: ListMcpServerStatusResponse = app
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
thread_id: Some(thread.id.clone()),
server_name: None,
detail: Some(McpServerStatusDetail::Full),
cursor: None,
limit: None,
},
})
.await?;
assert_eq!(response.next_cursor, None);
assert_eq!(
response
.data
.iter()
.map(|status| (
status.name.as_str(),
status.environment_id.as_deref(),
status
.tools
.keys()
.map(String::as_str)
.collect::<BTreeSet<_>>(),
status.tools_error.as_deref(),
status.auth_status,
status.http_origin.as_deref(),
))
.collect::<Vec<_>>(),
vec![
(
"notes",
Some("env-a"),
BTreeSet::from(["lookup"]),
None,
McpAuthStatus::Unsupported,
Some(first_url.as_str())
),
(
"notes",
Some("env-b"),
BTreeSet::from(["lookup"]),
None,
McpAuthStatus::BearerToken,
Some(second_url.as_str())
),
],
);
for (environment_id, expected) in [
(Some("env-a"), &response.data[0]),
(Some("env-b"), &response.data[1]),
// Omitting the selector keeps the original first-registration lookup.
(None, &response.data[0]),
] {
let selected: ListMcpServerStatusResponse = app
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
thread_id: Some(thread.id.clone()),
server_name: Some("notes".to_string()),
environment_id: environment_id.map(str::to_owned),
detail: Some(McpServerStatusDetail::Full),
cursor: None,
limit: None,
},
})
.await?;
// The selected read waits for the thread connection; the bulk read only observed it.
let mut expected = expected.clone();
expected.runtime_status = Some(McpServerConnectionStatus::Connected);
let origin = expected.http_origin.clone().expect("HTTP MCP origin");
assert_eq!(
selected,
ListMcpServerStatusResponse {
data: vec![expected],
next_cursor: None
}
);
// Identical tool names and resource URIs must reach the selected registration.
for (method, mut params, content_key, expected_content) in [
(
"mcpServer/tool/call",
json!({"tool": "lookup"}),
"content",
json!([{"type": "text", "text": origin}]),
),
(
"mcpServer/resource/read",
json!({"uri": "memo://shared"}),
"contents",
json!([{"uri": "memo://shared", "text": origin}]),
),
] {
params["threadId"] = json!(thread.id);
params["server"] = json!("notes");
if let Some(environment_id) = environment_id {
params["environmentId"] = json!(environment_id);
}
let request = app.send_raw_request(method, Some(params)).await?;
let result: serde_json::Value = app.read_response(request).await?;
assert_eq!(result[content_key], expected_content);
}
}
for (method, mut params) in [
("mcpServer/tool/call", json!({"tool": "lookup"})),
("mcpServer/resource/read", json!({"uri": "memo://shared"})),
] {
params["threadId"] = json!(thread.id);
params["server"] = json!("notes");
params["environmentId"] = json!("missing");
let request = app.send_raw_request(method, Some(params)).await?;
let error = app
.read_stream_until_error_message(RequestId::Integer(request))
.await?;
assert!(error.error.message.contains("unknown MCP server"));
}
first_handle.abort();
let _ = first_handle.await;
second_handle.abort();
let _ = second_handle.await;
Ok(())
}
async fn start_mcp_server(
tool_name: &str,
tools_error: Option<rmcp::ErrorData>,
@@ -1506,6 +1764,7 @@ async fn start_mcp_server(
Ok(McpStatusServer {
tool_name: Arc::clone(&tool_name),
tools_error: tools_error.clone(),
origin: format!("http://{addr}"),
})
},
Arc::new(LocalSessionManager::default()),
@@ -124,6 +124,7 @@ async fn mcp_server_tool_call_returns_tool_result() -> Result<()> {
.request(|request_id| ClientRequest::McpServerToolCall {
request_id,
params: McpServerToolCallParams {
environment_id: None,
thread_id: thread_id.clone(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -211,6 +212,7 @@ async fn mcp_server_tool_call_preserves_protocol_errors(protocol: ProtocolVersio
] {
let request_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: thread.id.clone(),
server: server.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -288,6 +290,7 @@ async fn mcp_server_tool_call_forwards_only_server_extensions() -> Result<()> {
.request(|request_id| ClientRequest::McpServerToolCall {
request_id,
params: McpServerToolCallParams {
environment_id: None,
thread_id: thread_id.clone(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -433,6 +436,7 @@ async fn mcp_server_tool_call_returns_error_for_unknown_thread() -> Result<()> {
let request_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: "00000000-0000-4000-8000-000000000000".to_string(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -602,6 +606,7 @@ async fn assert_full_access_form_elicitation_is_declined(
let request_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: thread.id,
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -700,6 +705,7 @@ async fn mcp_server_tool_call_round_trips_elicitation_for_thread(
let tool_call_request_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: thread.id.clone(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -733,6 +739,7 @@ async fn mcp_server_tool_call_round_trips_elicitation_for_thread(
thread_id: thread.id,
turn_id: None,
server_name: TEST_SERVER_NAME.to_string(),
environment_id: None,
request: McpServerElicitationRequest::Form {
meta: None,
message: ELICITATION_MESSAGE.to_string(),
@@ -858,6 +865,7 @@ async fn mcp_server_elicitation_survives_environment_runtime_refresh() -> Result
let tool_call_request_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: thread.id.clone(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -942,6 +950,7 @@ async fn mcp_server_tool_call_forwards_url_elicitation() -> Result<()> {
let tool_call_request_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: thread.id.clone(),
server: TEST_SERVER_NAME.to_string(),
tool: TEST_TOOL_NAME.to_string(),
@@ -966,6 +975,7 @@ async fn mcp_server_tool_call_forwards_url_elicitation() -> Result<()> {
thread_id: thread.id,
turn_id: None,
server_name: TEST_SERVER_NAME.to_string(),
environment_id: None,
request: McpServerElicitationRequest::Url {
meta: None,
message: URL_ELICITATION_MESSAGE.to_string(),
@@ -1090,6 +1100,7 @@ async fn mcp_tool_call_completion_notification_contains_truncated_large_result()
let serialized_item = serde_json::to_string(&ThreadItemPayload::McpToolCall {
id,
server,
environment_id: None,
tool,
status,
arguments: json!({ "message": LARGE_RESPONSE_MESSAGE }),
@@ -1240,6 +1251,7 @@ async fn mcp_tool_call_hint_survives_mid_call_thread_read_and_resume() -> Result
let expected_item = ThreadItemPayload::McpToolCall {
id: call_id.to_string(),
server: TEST_SERVER_NAME.to_string(),
environment_id: None,
tool: TEST_TOOL_NAME.to_string(),
status: McpToolCallStatus::InProgress,
arguments: json!({ "message": ELICITATION_TRIGGER_MESSAGE }),
@@ -2563,6 +2563,7 @@ async fn plugin_install_makes_bundled_mcp_servers_available_to_followup_requests
let request_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
environment_id: None,
thread_id: thread.id,
server: "sample-mcp".to_string(),
tool: "echo".to_string(),
@@ -2579,6 +2580,7 @@ async fn plugin_install_makes_bundled_mcp_servers_available_to_followup_requests
let request_id = mcp
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -3095,6 +3095,7 @@ fn append_resume_redaction_history(
call_id: "mcp-1".to_string(),
invocation: McpInvocation {
server: "docs".to_string(),
environment_id: None,
tool: "lookup".to_string(),
arguments: Some(json!({"secret":"argument"})),
},
@@ -1345,6 +1345,7 @@ async fn thread_start_emits_mcp_server_status_updated_notifications() -> Result<
McpServerStatusUpdatedNotification {
thread_id: Some(start_response.thread.id.clone()),
name: "optional_broken".to_string(),
environment_id: None,
status: McpServerStartupState::Starting,
error: None,
failure_reason: None,
@@ -1657,6 +1658,7 @@ required = true
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
environment_id: None,
server_name: None,
cursor: None,
limit: None,
@@ -166,6 +166,7 @@ async fn user_verification_mcp_round_trip_requires_proof_in_full_access() -> Res
caller
.call_mcp_tool(
CODEX_APPS_MCP_SERVER_NAME,
/*environment_id*/ None,
"verify_action",
/*arguments*/ None,
/*meta*/ None,
+5 -3
View File
@@ -710,8 +710,7 @@ async fn run_list(config: &Config, list_args: ListArgs) -> Result<()> {
runtime_config
.mcp_server_catalog
.resolve_server_name(name)
.and_then(|key| effective_mcp_servers.get(key))
.map(|server| (name, server))
.and_then(|key| effective_mcp_servers.get_key_value(key))
}),
config.mcp_oauth_credentials_store_mode,
config.auth_keyring_backend_kind(),
@@ -720,7 +719,10 @@ async fn run_list(config: &Config, list_args: ListArgs) -> Result<()> {
&runtime_config.ema_state,
&runtime_context,
)
.await;
.await
.into_iter()
.map(|(key, status)| (key.name().to_owned(), status))
.collect::<HashMap<_, _>>();
if list_args.json {
let json_entries: Vec<_> = entries
+14 -1
View File
@@ -801,7 +801,20 @@ impl ResolvedMcpCatalog {
})
}
pub(crate) fn servers(&self) -> impl Iterator<Item = (&McpServerKey, &ResolvedMcpServer)> {
/// Resolves a client target, preserving name-only routing when the scope is omitted.
/// The operation using this key checks whether the registration exists.
pub fn resolve_server_key(&self, name: &str, environment_id: Option<&str>) -> McpServerKey {
match environment_id {
Some(environment_id) => McpServerKey::in_environment(environment_id, name),
None => self
.resolve_server_name(name)
.cloned()
.unwrap_or_else(|| McpServerKey::unscoped(name)),
}
}
/// Iterates every resolved registration, including same-named scoped peers.
pub fn servers(&self) -> impl Iterator<Item = (&McpServerKey, &ResolvedMcpServer)> {
self.servers.iter()
}
+41 -47
View File
@@ -372,7 +372,7 @@ pub(crate) struct McpConnectionSet {
network_policy_metadata: Option<Arc<dyn NetworkPolicyMetadataProvider>>,
servers: HashMap<McpServerKey, McpServerView>,
pub(crate) event_stream_connection: Option<Arc<EventStreamConnectionSettings>>,
disabled_servers: Vec<String>,
disabled_servers: Vec<McpServerKey>,
required_servers: Vec<McpServerKey>,
optional_startup_deadline: OnceLock<tokio::time::Instant>,
tool_plugin_context: Arc<ToolPluginContext>,
@@ -402,20 +402,6 @@ impl McpConnectionSet {
self.server(self.resolve_server_name(name)?)
}
/// Name-only status and resource APIs still describe the original first server.
fn legacy_servers(&self) -> impl Iterator<Item = (&str, &McpServerView)> {
self.servers
.iter()
.filter_map(|(key, view)| self.legacy_server_name(key).map(|name| (name, view)))
}
pub(crate) fn legacy_server_name<'a>(&'a self, key: &'a McpServerKey) -> Option<&'a str> {
match &self.config {
Some(config) => config.mcp_server_catalog.legacy_server_name(key),
None => Some(key.name()),
}
}
/// Creates an MCP connection manager. Threadless callers can pass no `tx_event`; startup
/// notifications are then skipped and interactive elicitations are declined.
pub async fn new(
@@ -466,12 +452,7 @@ impl McpConnectionSet {
let disabled_servers = mcp_servers
.iter()
.filter(|(_, server)| !server.enabled())
.filter_map(|(key, _)| {
config
.mcp_server_catalog
.legacy_server_name(key)
.map(str::to_owned)
})
.map(|(key, _)| key.clone())
.collect();
let mut required_servers = mcp_servers
.iter()
@@ -535,6 +516,8 @@ impl McpConnectionSet {
let server_name = registration
.map_or(server_key.name(), |server| server.name())
.to_owned();
// Name-only clients still route startup errors by server name; publish only that
// name's default registration.
let report_startup = config
.mcp_server_catalog
.resolve_server_name(&server_name)
@@ -912,7 +895,7 @@ impl McpConnectionSet {
);
if let Some(error) = unchanged_auth_failure {
if report_startup {
eager_starts.push(server_name.clone());
eager_starts.push(server_key.clone());
}
let reason = connection_identity
.oauth_credentials()
@@ -938,6 +921,9 @@ impl McpConnectionSet {
tx_event,
McpStartupUpdateEvent {
server: server_name.clone(),
environment_id: server_key
.scoped_environment_id()
.map(str::to_owned),
status,
},
)
@@ -1067,7 +1053,7 @@ impl McpConnectionSet {
let submit_id = startup_submit_id.clone();
if !defer_startup && report_startup {
eager_starts.push(server_name.clone());
eager_starts.push(server_key.clone());
}
let startup = async move {
if let Some(mut startup_receiver) = startup_receiver
@@ -1084,6 +1070,7 @@ impl McpConnectionSet {
tx_event,
McpStartupUpdateEvent {
server: server_name.clone(),
environment_id: server_key.scoped_environment_id().map(str::to_owned),
status: McpStartupStatus::Starting,
},
)
@@ -1159,6 +1146,7 @@ impl McpConnectionSet {
tx_event,
McpStartupUpdateEvent {
server: server_name.clone(),
environment_id: server_key.scoped_environment_id().map(str::to_owned),
status,
},
)
@@ -1210,7 +1198,8 @@ impl McpConnectionSet {
startup_submit_id.as_str(),
tx_event,
McpStartupUpdateEvent {
server,
server: server.name().to_owned(),
environment_id: server.scoped_environment_id().map(str::to_owned),
status: McpStartupStatus::Starting,
},
)
@@ -1228,16 +1217,18 @@ impl McpConnectionSet {
.map(str::to_owned)
};
if let Some(tx_event) = tx_event {
let mut summary = McpStartupCompleteEvent {
ready: reused_ready.into_iter().filter_map(&legacy_name).collect(),
..Default::default()
};
for server_name in &summary.ready {
let mut summary = McpStartupCompleteEvent::default();
for key in reused_ready {
let Some(server_name) = legacy_name(key.clone()) else {
continue;
};
summary.ready.push(server_name);
let _ = emit_update(
startup_submit_id.as_str(),
&tx_event,
McpStartupUpdateEvent {
server: server_name.clone(),
server: key.name().to_owned(),
environment_id: key.scoped_environment_id().map(str::to_owned),
status: McpStartupStatus::Ready,
},
)
@@ -1468,14 +1459,21 @@ impl McpConnectionSet {
view.metadata.environment_id
);
}
if !view.tool_filter.allows(tool) {
return Err(anyhow!(
"tool '{tool}' is disabled for MCP server '{server}'"
));
}
self.validate_tool_call(server_key, tool)?;
Ok(server_key)
}
/// Validates an exact client target before startup or execution.
pub(crate) fn validate_tool_call(&self, server: &McpServerKey, tool: &str) -> Result<()> {
let view = self
.server(server)
.ok_or_else(|| anyhow!("unknown MCP server '{server}'"))?;
if !view.tool_filter.allows(tool) {
bail!("tool '{tool}' is disabled for MCP server '{server}'");
}
Ok(())
}
/// Invoke the tool indicated by the (server, tool) pair.
#[allow(clippy::too_many_arguments)]
pub async fn call_tool(
@@ -1487,12 +1485,10 @@ impl McpConnectionSet {
requested_timeout: Option<Duration>,
wait_for_server: bool,
) -> Result<CallToolResult> {
self.validate_tool_call(server_key, tool)?;
let view = self
.server(server_key)
.ok_or_else(|| anyhow!("unknown MCP server '{server_key}'"))?;
if !view.tool_filter.allows(tool) {
bail!("tool '{tool}' is disabled for MCP server '{server_key}'");
}
let client = if wait_for_server {
view.connection
.client()
@@ -1543,9 +1539,10 @@ impl McpConnectionSet {
/// Capabilities belong to the initialized connection, never a shared tool cache.
pub(crate) fn list_available_server_capabilities(
&self,
include_server: impl Fn(&str) -> bool,
) -> HashMap<String, serde_json::Value> {
self.legacy_servers()
include_server: impl Fn(&McpServerKey) -> bool,
) -> HashMap<McpServerKey, serde_json::Value> {
self.servers
.iter()
.filter(|(name, _)| include_server(name))
.filter_map(|(name, view)| {
view.connection
@@ -1566,13 +1563,10 @@ impl McpConnectionSet {
/// connection-specific, so pending regular clients are awaited.
pub(crate) async fn list_available_server_infos(
&self,
include_server: impl Fn(&str) -> bool,
) -> HashMap<String, McpServerInfo> {
include_server: impl Fn(&McpServerKey) -> bool,
) -> HashMap<McpServerKey, McpServerInfo> {
let mut server_infos = HashMap::new();
for (server_name, view) in self
.legacy_servers()
.filter(|(name, _)| include_server(name))
{
for (server_name, view) in self.servers.iter().filter(|(name, _)| include_server(name)) {
let server_name = server_name.to_owned();
let client = &view.connection.client;
if !client.startup_complete.load(Ordering::Acquire)
@@ -45,11 +45,12 @@ impl McpConnectionSet {
/// Returns resources from servers selected by `include_server`.
pub async fn list_all_resources(
&self,
include_server: impl Fn(&str) -> bool,
) -> HashMap<String, Vec<Resource>> {
include_server: impl Fn(&McpServerKey) -> bool,
) -> HashMap<McpServerKey, Vec<Resource>> {
let mut join_set = JoinSet::new();
for (server_name, view) in self
.legacy_servers()
.servers
.iter()
.filter(|(server_name, _)| include_server(server_name))
{
let server_name = server_name.to_owned();
@@ -91,10 +92,10 @@ impl McpConnectionSet {
/// Returns resource templates from servers selected by `include_server`.
pub async fn list_all_resource_templates(
&self,
include_server: impl Fn(&str) -> bool,
) -> HashMap<String, Vec<ResourceTemplate>> {
include_server: impl Fn(&McpServerKey) -> bool,
) -> HashMap<McpServerKey, Vec<ResourceTemplate>> {
let mut join_set = JoinSet::new();
for (server_name, view) in self.legacy_servers().filter(|(server_name, view)| {
for (server_name, view) in self.servers.iter().filter(|(server_name, view)| {
include_server(server_name)
&& !self.skips_resource_template_listing(&view.metadata.server_key)
}) {
@@ -5,9 +5,12 @@ use std::collections::HashMap;
use codex_protocol::mcp::McpServerConnectionStatus;
use super::McpConnectionSet;
use crate::McpServerKey;
impl McpConnectionSet {
pub(crate) async fn connection_statuses(&self) -> HashMap<String, McpServerConnectionStatus> {
pub(crate) async fn connection_statuses(
&self,
) -> HashMap<McpServerKey, McpServerConnectionStatus> {
use McpServerConnectionStatus as Status;
let mut statuses = self
@@ -15,7 +18,7 @@ impl McpConnectionSet {
.iter()
.map(|name| (name.clone(), Status::Disabled))
.collect::<HashMap<_, _>>();
for (name, view) in self.legacy_servers() {
for (name, view) in &self.servers {
let connection = &view.connection;
let client = &connection.client;
let status = if connection.startup_is_dormant() && !client.cancel_token.is_cancelled() {
@@ -198,19 +198,19 @@ impl McpConnectionSet {
#[instrument(level = "trace", skip_all, fields(mcp_server_count = self.servers.len()))]
pub(crate) async fn list_tools_with_errors(
&self,
include_server: impl Fn(&str) -> bool,
) -> (Vec<ToolInfo>, HashMap<String, String>) {
include_server: impl Fn(&McpServerKey) -> bool,
) -> (Vec<ToolInfo>, HashMap<McpServerKey, String>) {
let (tools, errors, _) = self.list_catalogs_with_errors(include_server).await;
(tools, errors)
}
pub(crate) async fn list_catalogs_with_errors(
&self,
include_server: impl Fn(&str) -> bool,
include_server: impl Fn(&McpServerKey) -> bool,
) -> (
Vec<ToolInfo>,
HashMap<String, String>,
HashMap<String, Arc<serde_json::Value>>,
HashMap<McpServerKey, String>,
HashMap<McpServerKey, Arc<serde_json::Value>>,
) {
let mut metadata = HashMap::new();
let mut tools = Vec::new();
@@ -223,7 +223,8 @@ impl McpConnectionSet {
.map(|config| config.mcp_server_catalog.exposure())
.unwrap_or_default();
let server_results = join_all(
self.legacy_servers()
self.servers
.iter()
.filter(|(name, _)| include_server(name))
.map(|(server_name, view)| async move {
view.connection.client.reconnect_failed_startup().await;
@@ -234,7 +235,7 @@ impl McpConnectionSet {
.startup_complete
.load(Ordering::Acquire);
let server_tools = view
.listed_catalog(server_name, &self.tool_plugin_context, exposure)
.listed_catalog(server_name.name(), &self.tool_plugin_context, exposure)
.instrument(trace_span!(
"list_tools_for_server",
server_name = %server_name,
@@ -688,7 +688,9 @@ async fn connection_statuses_observe_clients_without_starting_them() {
use codex_protocol::mcp::McpServerConnectionStatus as Status;
let mut manager = McpConnectionSet::empty(/*prefix_mcp_tool_names*/ true);
manager.disabled_servers.push("disabled".to_string());
manager
.disabled_servers
.push(McpServerKey::unscoped("disabled"));
let ready = create_ready_async_managed_client(Vec::new()).await;
ready.client().await.expect("ready client");
manager.insert_test_client("connected", ready);
@@ -727,16 +729,17 @@ async fn connection_statuses_observe_clients_without_starting_them() {
pending.cached_server_info = Some(create_test_server_info("Cached"));
manager.insert_test_client("starting", pending.clone());
manager.insert_test_client("deferred", pending);
let scoped_key = McpServerKey::in_environment("remote", "connected");
let mut deferred = manager
.servers
.remove(&McpServerKey::unscoped("deferred"))
.unwrap();
deferred.metadata.server_key = scoped_key.clone();
manager.servers.insert(scoped_key.clone(), deferred);
let (trigger, _receiver) = watch::channel(/*init*/ false);
Arc::get_mut(
&mut manager
.servers
.get_mut(&McpServerKey::unscoped("deferred"))
.unwrap()
.connection,
)
.unwrap()
.startup_trigger = Some(trigger.clone());
Arc::get_mut(&mut manager.servers.get_mut(&scoped_key).unwrap().connection)
.unwrap()
.startup_trigger = Some(trigger.clone());
let statuses = tokio::time::timeout(
Duration::from_millis(/*millis*/ 100),
@@ -745,19 +748,25 @@ async fn connection_statuses_observe_clients_without_starting_them() {
.await
.expect("status must not await startup");
let mut expected = HashMap::from([
("connected".to_string(), Status::Connected),
("failed".to_string(), Status::Failed),
("auth".to_string(), Status::AuthenticationRequired),
("flattened-auth".to_string(), Status::AuthenticationRequired),
("cancelled".to_string(), Status::Cancelled),
("starting".to_string(), Status::Starting),
("deferred".to_string(), Status::NotStarted),
("disabled".to_string(), Status::Disabled),
(McpServerKey::unscoped("connected"), Status::Connected),
(McpServerKey::unscoped("failed"), Status::Failed),
(
McpServerKey::unscoped("auth"),
Status::AuthenticationRequired,
),
(
McpServerKey::unscoped("flattened-auth"),
Status::AuthenticationRequired,
),
(McpServerKey::unscoped("cancelled"), Status::Cancelled),
(McpServerKey::unscoped("starting"), Status::Starting),
(scoped_key, Status::NotStarted),
(McpServerKey::unscoped("disabled"), Status::Disabled),
]);
assert_eq!(statuses, expected);
assert!(!*trigger.borrow());
manager.test_client("connected").cancel_token.cancel();
expected.insert("connected".to_string(), Status::Cancelled);
expected.insert(McpServerKey::unscoped("connected"), Status::Cancelled);
assert_eq!(manager.connection_statuses().await, expected);
}
@@ -802,7 +811,8 @@ async fn connection_statuses_follow_latest_reconnect_outcome() {
let manager = create_test_manager_with_failed_apps_startup(Vec::new(), factory);
let client = manager.test_client(CODEX_APPS_MCP_SERVER_NAME);
assert!(client.client().await.is_err());
let expected = |status| HashMap::from([(CODEX_APPS_MCP_SERVER_NAME.to_string(), status)]);
let expected =
|status| HashMap::from([(McpServerKey::unscoped(CODEX_APPS_MCP_SERVER_NAME), status)]);
assert_eq!(
manager.connection_statuses().await,
expected(Status::Failed)
@@ -3063,7 +3073,7 @@ async fn list_available_server_infos_uses_cache_while_client_is_pending() {
.await;
let server_infos = timeout_result.expect("server info lookup should not block on startup");
assert_eq!(
server_infos.get(CODEX_APPS_MCP_SERVER_NAME),
server_infos.get(&McpServerKey::unscoped(CODEX_APPS_MCP_SERVER_NAME)),
Some(&server_info)
);
}
@@ -4502,7 +4512,7 @@ async fn list_all_tools_uses_shared_codex_apps_cache_when_client_startup_fails()
manager
.list_available_server_infos(|_| true)
.await
.get(CODEX_APPS_MCP_SERVER_NAME),
.get(&McpServerKey::unscoped(CODEX_APPS_MCP_SERVER_NAME)),
Some(&server_info)
);
}
@@ -5170,7 +5180,7 @@ async fn executor_owned_chatgpt_mcp_accepts_only_safe_explicit_authorization() -
"{case}: executor-owned servers must not retain hosted OAuth credentials"
);
let auth_statuses = crate::compute_auth_statuses(
[(&"fake-first-party".to_string(), remote_server)],
[(&McpServerKey::unscoped("fake-first-party"), remote_server)],
OAuthCredentialsStoreMode::default(),
AuthKeyringBackendKind::default(),
Some(&hosted_auth),
@@ -5185,7 +5195,8 @@ async fn executor_owned_chatgpt_mcp_accepts_only_safe_explicit_authorization() -
McpAuthState::Unsupported
};
assert_eq!(
auth_statuses["fake-first-party"].auth_state, expected_auth_state,
auth_statuses[&McpServerKey::unscoped("fake-first-party")].auth_state,
expected_auth_state,
"{case}: auth status must only accept safe executor-owned authorization"
);
+4 -2
View File
@@ -275,9 +275,10 @@ impl ElicitationRequestRouter {
&self,
events: Option<Sender<Event>>,
authority: &ElicitationAuthority,
server_name: String,
server_key: McpServerKey,
request: ElicitationRequest,
) -> Result<ElicitationResponse> {
let server_name = server_key.name().to_owned();
let Some(events) = events else {
return Ok(ElicitationResponse {
action: ElicitationAction::Decline,
@@ -319,6 +320,7 @@ impl ElicitationRequestRouter {
msg: EventMsg::ElicitationRequest(ElicitationRequestEvent {
turn_id: None,
server_name,
environment_id: server_key.scoped_environment_id().map(str::to_owned),
id: ProtocolRequestId::String(public_request_id),
request,
}),
@@ -684,7 +686,7 @@ impl ElicitationRequestManager {
},
};
router
.request_user_interaction(tx_event, &authority, server_name, request)
.request_user_interaction(tx_event, &authority, server_key, request)
.await
}
.boxed()
+26 -8
View File
@@ -31,13 +31,13 @@ async fn abandonment_is_ordered_and_exactly_once_when_the_native_waiter_is_dropp
let mut first = Box::pin(router.request_user_interaction(
Some(tx),
&authority,
"server".into(),
McpServerKey::in_environment("first", "server"),
request.clone(),
));
let mut sibling = Box::pin(router.request_user_interaction(
Some(replacement_tx),
&authority,
"server".into(),
McpServerKey::in_environment("second", "server"),
request,
));
assert!(futures::poll!(&mut first).is_pending());
@@ -50,6 +50,19 @@ async fn abandonment_is_ordered_and_exactly_once_when_the_native_waiter_is_dropp
else {
panic!("expected sibling request");
};
assert_eq!(
[
(
first_request.server_name.as_str(),
first_request.environment_id.as_deref()
),
(
sibling_request.server_name.as_str(),
sibling_request.environment_id.as_deref()
),
],
[("server", Some("first")), ("server", Some("second"))],
);
drop(first);
let EventMsg::ElicitationAbandoned(abandoned) = events.recv().await.unwrap().msg else {
panic!("expected abandonment");
@@ -107,7 +120,7 @@ async fn bounded_sources_preserve_backpressure_across_close() {
let mut pending = Box::pin(router.request_user_interaction(
Some(tx),
&authority,
"server".into(),
McpServerKey::unscoped("server"),
ElicitationRequest::Form {
meta: None,
message: "Continue".into(),
@@ -150,7 +163,7 @@ async fn abandonment_closed_receiver_cleans_up_without_delivery_or_replay() {
let mut pending = Box::pin(router.request_user_interaction(
Some(tx),
&authority,
"server".into(),
McpServerKey::unscoped("server"),
ElicitationRequest::Form {
meta: None,
message: "Continue".into(),
@@ -188,7 +201,7 @@ async fn abandonment_close_joins_elected_actions_and_rejects_retained_router_adm
let mut pending = Box::pin(router.request_user_interaction(
Some(tx.clone()),
&authority,
"server".into(),
McpServerKey::unscoped("server"),
request.clone(),
));
assert!(futures::poll!(&mut pending).is_pending());
@@ -213,7 +226,7 @@ async fn abandonment_close_joins_elected_actions_and_rejects_retained_router_adm
.request_user_interaction(
Some(tx.clone()),
&authority,
"server".into(),
McpServerKey::unscoped("server"),
request.clone(),
)
.await
@@ -240,7 +253,12 @@ async fn abandonment_close_joins_elected_actions_and_rejects_retained_router_adm
router.close().await;
assert!(
retained_router
.request_user_interaction(Some(tx), &authority, "server".into(), request,)
.request_user_interaction(
Some(tx),
&authority,
McpServerKey::unscoped("server"),
request
)
.await
.is_err()
);
@@ -256,7 +274,7 @@ async fn abandonment_close_invalidates_a_published_responder_before_returning()
let mut pending = Box::pin(router.request_user_interaction(
Some(tx),
&authority,
"server".into(),
McpServerKey::unscoped("server"),
ElicitationRequest::Form {
meta: None,
message: "Continue".into(),
+4 -3
View File
@@ -25,6 +25,7 @@ use futures::future::join_all;
use tokio::sync::OnceCell;
use tracing::warn;
use crate::McpServerKey;
use crate::runtime::McpRuntimeContext;
use crate::server::EffectiveMcpServer;
use crate::server::has_explicit_http_authorization;
@@ -205,9 +206,9 @@ pub async fn compute_auth_statuses<'a, I>(
auth_manager: Option<&std::sync::Arc<codex_login::AuthManager>>,
ema_state: &crate::EmaState,
runtime_context: &McpRuntimeContext,
) -> HashMap<String, McpAuthStatusEntry>
) -> HashMap<McpServerKey, McpAuthStatusEntry>
where
I: IntoIterator<Item = (&'a String, &'a EffectiveMcpServer)>,
I: IntoIterator<Item = (&'a McpServerKey, &'a EffectiveMcpServer)>,
{
// Share successful, missing, and failed reads without starting storage work for non-EMA servers.
let account_grant = OnceCell::new();
@@ -223,7 +224,7 @@ where
let runtime_context = runtime_context.clone();
async move {
let auth_state = match compute_auth_status(
&name,
name.name(),
&config,
store_mode,
keyring_backend_kind,
+37 -56
View File
@@ -482,10 +482,14 @@ pub async fn read_mcp_resource(
codex_apps_tools_cache: ConnectorRuntimeManager<ToolInfo>,
tool_catalog_cache: crate::McpToolCatalogCache,
server: &str,
environment_id: Option<&str>,
params: ReadResourceRequestParams,
) -> anyhow::Result<ReadResourceResult> {
let server_key = config
.mcp_server_catalog
.resolve_server_key(server, environment_id);
let mut mcp_servers = effective_mcp_servers(config, auth);
mcp_servers.retain(|key, _| config.mcp_server_catalog.legacy_server_name(key) == Some(server));
mcp_servers.retain(|key, _| key == &server_key);
let cancel_token = CancellationToken::new();
let runtime_config = config.for_threadless_operations(&mcp_servers);
let manager = McpConnectionSet::new(
@@ -518,25 +522,22 @@ pub async fn read_mcp_resource(
)
.await;
let result = match manager.resolve_server_name(server) {
Some(key) => manager.read_resource(key, params).await,
None => Err(anyhow::anyhow!("unknown MCP server '{server}'")),
};
let result = manager.read_resource(&server_key, params).await;
cancel_token.cancel();
result
}
#[derive(Debug, Clone)]
pub struct McpServerStatusSnapshot {
pub server_infos: HashMap<String, McpServerInfo>,
pub server_capabilities: HashMap<String, serde_json::Value>,
pub tool_list_meta: HashMap<String, Arc<serde_json::Value>>,
pub tools_by_server: HashMap<String, HashMap<String, Tool>>,
pub tools_errors: HashMap<String, String>,
pub resources: HashMap<String, Vec<Resource>>,
pub resource_templates: HashMap<String, Vec<ResourceTemplate>>,
pub auth_statuses: HashMap<String, McpAuthStatus>,
pub server_names: Vec<String>,
pub server_infos: HashMap<McpServerKey, McpServerInfo>,
pub server_capabilities: HashMap<McpServerKey, serde_json::Value>,
pub tool_list_meta: HashMap<McpServerKey, Arc<serde_json::Value>>,
pub tools_by_server: HashMap<McpServerKey, HashMap<String, Tool>>,
pub tools_errors: HashMap<McpServerKey, String>,
pub resources: HashMap<McpServerKey, Vec<Resource>>,
pub resource_templates: HashMap<McpServerKey, Vec<ResourceTemplate>>,
pub auth_statuses: HashMap<McpServerKey, McpAuthStatus>,
pub server_keys: Vec<McpServerKey>,
}
#[allow(clippy::too_many_arguments)]
@@ -552,15 +553,10 @@ pub async fn collect_mcp_server_status_snapshot_with_detail(
codex_apps_tools_cache: ConnectorRuntimeManager<ToolInfo>,
tool_catalog_cache: crate::McpToolCatalogCache,
detail: McpSnapshotDetail,
server_name: Option<&str>,
server: Option<&McpServerKey>,
) -> McpServerStatusSnapshot {
let mut mcp_servers = effective_mcp_servers(config, auth);
mcp_servers.retain(|key, _| {
config
.mcp_server_catalog
.legacy_server_name(key)
.is_some_and(|name| server_name.is_none_or(|selected| name == selected))
});
mcp_servers.retain(|key, _| server.is_none_or(|selected| key == selected));
if mcp_servers.is_empty() {
return McpServerStatusSnapshot {
server_infos: HashMap::new(),
@@ -571,21 +567,12 @@ pub async fn collect_mcp_server_status_snapshot_with_detail(
resources: HashMap::new(),
resource_templates: HashMap::new(),
auth_statuses: HashMap::new(),
server_names: Vec::new(),
server_keys: Vec::new(),
};
}
let named_servers = mcp_servers
.iter()
.filter_map(|(key, server)| {
config
.mcp_server_catalog
.legacy_server_name(key)
.map(|name| (name.to_owned(), server.clone()))
})
.collect::<HashMap<_, _>>();
let auth_status_entries = compute_auth_statuses(
named_servers.iter(),
mcp_servers.iter(),
config.mcp_oauth_credentials_store_mode,
config.auth_keyring_backend_kind,
auth,
@@ -595,7 +582,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail(
)
.await;
let server_names = named_servers.into_keys().collect();
let server_keys = mcp_servers.keys().cloned().collect();
let cancel_token = CancellationToken::new();
let runtime_config = config.for_threadless_operations(&mcp_servers);
@@ -632,7 +619,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail(
let snapshot = collect_mcp_server_status_snapshot_from_manager(
&mcp_connection_manager,
auth_status_entries,
server_names,
server_keys,
detail,
)
.await;
@@ -774,8 +761,8 @@ fn protocol_tool_from_rmcp_tool(name: &str, tool: &rmcp::model::Tool) -> Option<
}
fn auth_statuses_from_entries(
auth_status_entries: &HashMap<String, crate::mcp::auth::McpAuthStatusEntry>,
) -> HashMap<String, McpAuthStatus> {
auth_status_entries: &HashMap<McpServerKey, crate::mcp::auth::McpAuthStatusEntry>,
) -> HashMap<McpServerKey, McpAuthStatus> {
auth_status_entries
.iter()
.map(|(name, entry)| (name.clone(), McpAuthStatus::from(entry.auth_state)))
@@ -783,8 +770,8 @@ fn auth_statuses_from_entries(
}
fn convert_mcp_resources(
resources: HashMap<String, Vec<rmcp::model::Resource>>,
) -> HashMap<String, Vec<Resource>> {
resources: HashMap<McpServerKey, Vec<rmcp::model::Resource>>,
) -> HashMap<McpServerKey, Vec<Resource>> {
resources
.into_iter()
.map(|(name, resources)| {
@@ -822,8 +809,8 @@ fn convert_mcp_resources(
}
fn convert_mcp_resource_templates(
resource_templates: HashMap<String, Vec<rmcp::model::ResourceTemplate>>,
) -> HashMap<String, Vec<ResourceTemplate>> {
resource_templates: HashMap<McpServerKey, Vec<rmcp::model::ResourceTemplate>>,
) -> HashMap<McpServerKey, Vec<ResourceTemplate>> {
resource_templates
.into_iter()
.map(|(name, templates)| {
@@ -863,12 +850,12 @@ fn convert_mcp_resource_templates(
pub(crate) async fn collect_mcp_server_status_snapshot_from_manager(
mcp_connection_manager: &McpConnectionSet,
auth_status_entries: HashMap<String, crate::mcp::auth::McpAuthStatusEntry>,
server_names: Vec<String>,
auth_status_entries: HashMap<McpServerKey, crate::mcp::auth::McpAuthStatusEntry>,
server_keys: Vec<McpServerKey>,
detail: McpSnapshotDetail,
) -> McpServerStatusSnapshot {
let selected_servers: HashSet<&str> = server_names.iter().map(String::as_str).collect();
let include_server = |name: &str| selected_servers.contains(name);
let selected_servers: HashSet<&McpServerKey> = server_keys.iter().collect();
let include_server = |key: &McpServerKey| selected_servers.contains(key);
let ((server_infos, (tools, tools_errors, tool_list_meta)), resources, resource_templates) = tokio::join!(
async {
let server_infos = mcp_connection_manager
@@ -901,7 +888,7 @@ pub(crate) async fn collect_mcp_server_status_snapshot_from_manager(
let server_capabilities =
mcp_connection_manager.list_available_server_capabilities(include_server);
let mut tools_by_server = HashMap::<String, HashMap<String, Tool>>::new();
let mut tools_by_server = HashMap::<McpServerKey, HashMap<String, Tool>>::new();
for tool_info in tools {
let raw_tool_name = tool_info.tool.name.to_string();
let Some(tool) = protocol_tool_from_rmcp_tool(&raw_tool_name, &tool_info.tool) else {
@@ -909,7 +896,7 @@ pub(crate) async fn collect_mcp_server_status_snapshot_from_manager(
};
let tool_name = tool.name.clone();
tools_by_server
.entry(tool_info.server_name().to_owned())
.entry(tool_info.server_key)
.or_default()
.insert(tool_name, tool);
}
@@ -918,19 +905,13 @@ pub(crate) async fn collect_mcp_server_status_snapshot_from_manager(
// connection attempt instead of retaining the credential-presence status read beforehand.
let mut auth_statuses = auth_statuses_from_entries(&auth_status_entries);
for key in mcp_connection_manager.authentication_failed_servers().await {
let Some(server_name) = mcp_connection_manager
.legacy_server_name(&key)
.map(str::to_owned)
else {
continue;
};
if auth_statuses.get(&server_name) == Some(&McpAuthStatus::OAuth)
if auth_statuses.get(&key) == Some(&McpAuthStatus::OAuth)
|| auth_status_entries
.get(&server_name)
.get(&key)
.and_then(|entry| entry.config.as_ref())
.is_some_and(|config| matches!(config.auth, McpServerAuth::EmaAuth))
{
auth_statuses.insert(server_name, McpAuthStatus::NotLoggedIn);
auth_statuses.insert(key, McpAuthStatus::NotLoggedIn);
}
}
@@ -943,7 +924,7 @@ pub(crate) async fn collect_mcp_server_status_snapshot_from_manager(
resources: convert_mcp_resources(resources),
resource_templates: convert_mcp_resource_templates(resource_templates),
auth_statuses,
server_names,
server_keys,
}
}
+18 -15
View File
@@ -91,7 +91,7 @@ async fn ema_auth_statuses_do_not_wait_for_account_refresh() -> anyhow::Result<(
})
.map_err(anyhow::Error::msg)?;
servers.insert(
name.to_string(),
McpServerKey::unscoped(name),
EffectiveMcpServer::from_host_config(server),
);
}
@@ -119,8 +119,8 @@ async fn ema_auth_statuses_do_not_wait_for_account_refresh() -> anyhow::Result<(
.map(|(name, entry)| (name, entry.auth_state))
.collect::<HashMap<_, _>>(),
HashMap::from([
("first".to_string(), McpAuthState::BearerToken),
("second".to_string(), McpAuthState::BearerToken),
(McpServerKey::unscoped("first"), McpAuthState::BearerToken),
(McpServerKey::unscoped("second"), McpAuthState::BearerToken),
]),
);
drop(lease);
@@ -173,7 +173,7 @@ async fn status_snapshot_only_downgrades_oauth_and_ema_authentication_failures()
}
manager.insert_test_client(name, client);
auth_status_entries.insert(
name.to_string(),
McpServerKey::unscoped(name),
McpAuthStatusEntry {
config: name.starts_with("ema-").then(|| {
serde_json::from_value(serde_json::json!({
@@ -186,26 +186,29 @@ async fn status_snapshot_only_downgrades_oauth_and_ema_authentication_failures()
);
}
let server_names = auth_status_entries.keys().cloned().collect();
let server_keys = auth_status_entries.keys().cloned().collect();
let snapshot = collect_mcp_server_status_snapshot_from_manager(
&manager,
auth_status_entries,
server_names,
server_keys,
McpSnapshotDetail::ToolsAndAuthOnly,
)
.await;
assert_eq!(
snapshot.auth_statuses,
HashMap::from([
("oauth-failed".to_string(), McpAuthStatus::NotLoggedIn),
("oauth-ready".to_string(), McpAuthStatus::OAuth),
("oauth-provider-error".to_string(), McpAuthStatus::OAuth),
("ema-failed".to_string(), McpAuthStatus::NotLoggedIn),
("ema-ready".to_string(), McpAuthStatus::BearerToken),
("ema-provider-error".to_string(), McpAuthStatus::BearerToken),
("bearer".to_string(), McpAuthStatus::BearerToken),
]),
HashMap::from(
[
("oauth-failed", McpAuthStatus::NotLoggedIn),
("oauth-ready", McpAuthStatus::OAuth),
("oauth-provider-error", McpAuthStatus::OAuth),
("ema-failed", McpAuthStatus::NotLoggedIn),
("ema-ready", McpAuthStatus::BearerToken),
("ema-provider-error", McpAuthStatus::BearerToken),
("bearer", McpAuthStatus::BearerToken),
]
.map(|(name, status)| (McpServerKey::unscoped(name), status)),
),
);
}
+4 -1
View File
@@ -346,7 +346,10 @@ impl McpResourceClient {
/// Reads one resource from the named server.
pub async fn read_resource(&self, server: &str, uri: &str) -> Result<McpResourceReadResult> {
let params = ReadResourceRequestParams::new(uri.to_string());
let result = self.runtime.latest_read_resource(server, params).await?;
let result = self
.runtime
.latest_read_resource(server, /*environment_id*/ None, params)
.await?;
let contents = result
.contents
.into_iter()
+10 -6
View File
@@ -151,7 +151,7 @@ struct CodexAppsStartupReconnectState {
#[derive(Clone)]
struct CodexAppsStartupStatusContext {
submit_id: String,
server_name: String,
server_key: McpServerKey,
tx_event: Sender<Event>,
}
@@ -173,12 +173,12 @@ impl CodexAppsStartupReconnect {
fn with_startup_status_context(
mut self,
submit_id: String,
server_name: String,
server_key: McpServerKey,
tx_event: Option<Sender<Event>>,
) -> Self {
self.startup_status_context = tx_event.map(|tx_event| CodexAppsStartupStatusContext {
submit_id,
server_name,
server_key,
tx_event,
});
self
@@ -247,7 +247,11 @@ impl CodexAppsStartupReconnect {
.send(Event {
id: context.submit_id,
msg: EventMsg::McpStartupUpdate(McpStartupUpdateEvent {
server: context.server_name,
server: context.server_key.name().to_owned(),
environment_id: context
.server_key
.scoped_environment_id()
.map(str::to_owned),
status: McpStartupStatus::Ready,
}),
})
@@ -505,7 +509,7 @@ impl AsyncManagedClient {
) -> Self {
let server_name = server_key.name().to_owned();
let is_codex_apps_mcp_server = server_name == CODEX_APPS_MCP_SERVER_NAME;
let reconnect_server_name = server_name;
let reconnect_server_key = server_key.clone();
let reconnect_tx_event = tx_event.clone();
let cached_server_info = if is_codex_apps_mcp_server {
codex_apps_tools_cache_context
@@ -614,7 +618,7 @@ impl AsyncManagedClient {
}))
.with_startup_status_context(
startup_submit_id,
reconnect_server_name,
reconnect_server_key,
reconnect_tx_event,
),
)
+41 -16
View File
@@ -4,7 +4,8 @@
//! environment in which MCP servers execute. Transport startup lives in
//! [`crate::rmcp_client`] and connection-set behavior lives in
//! [`crate::connection_manager`]. Registry removal revokes EMA authority before
//! the thread becomes untracked.
//! the thread becomes untracked. Client targets resolve against the same runtime
//! publication whose connections execute the operation.
#[cfg(test)]
#[path = "runtime/apps_tool_refresh_tests.rs"]
@@ -829,29 +830,53 @@ impl McpRuntime {
pub async fn latest_read_resource(
&self,
server: &str,
environment_id: Option<&str>,
params: ReadResourceRequestParams,
) -> anyhow::Result<ReadResourceResult> {
let connections = self.latest_connections();
let key = connections
.resolve_server_name(server)
let current = self.current.load_full();
let key = current
.config
.as_ref()
.map(|config| {
config
.mcp_server_catalog
.resolve_server_key(server, environment_id)
})
.ok_or_else(|| anyhow::anyhow!("unknown MCP server '{server}'"))?;
connections.read_resource(key, params).await
current.connections.read_resource(&key, params).await
}
pub async fn latest_call_tool_without_attribution(
&self,
server: &str,
environment_id: Option<&str>,
tool: &str,
arguments: Option<serde_json::Value>,
meta: Option<serde_json::Value>,
) -> anyhow::Result<CallToolResult> {
let connections = self.latest_connections();
let server_key =
connections.resolve_tool_call_server(server, tool, /*environment_id*/ None)?;
connections.wait_for_server_startup(server_key).await;
connections
let current = self.current.load_full();
let server_key = current
.config
.as_ref()
.map(|config| {
config
.mcp_server_catalog
.resolve_server_key(server, environment_id)
})
.ok_or_else(|| anyhow::anyhow!("unknown MCP server '{server}'"))?;
current.connections.validate_tool_call(&server_key, tool)?;
current
.connections
.wait_for_server_startup(&server_key)
.await;
current
.connections
.call_tool(
server_key, tool, arguments, meta, /*requested_timeout*/ None,
&server_key,
tool,
arguments,
meta,
/*requested_timeout*/ None,
/*wait_for_server*/ true,
)
.await
@@ -967,7 +992,8 @@ impl McpRuntime {
pub async fn connection_statuses(
&self,
config: &McpConfig,
) -> std::collections::HashMap<String, codex_protocol::mcp::McpServerConnectionStatus> {
) -> std::collections::HashMap<McpServerKey, codex_protocol::mcp::McpServerConnectionStatus>
{
let current = self.current.load_full();
let Some(published_config) = current.config.as_ref() else {
return HashMap::new();
@@ -976,10 +1002,8 @@ impl McpRuntime {
statuses.retain(|name, _| {
published_config
.mcp_server_catalog
.server_by_name(name)
.is_some_and(|server| {
config.mcp_server_catalog.server_by_name(name) == Some(server)
})
.server(name)
.is_some_and(|server| config.mcp_server_catalog.server(name) == Some(server))
});
statuses
}
@@ -1342,6 +1366,7 @@ mod tests {
runtime
.latest_call_tool_without_attribution(
CODEX_APPS_MCP_SERVER_NAME,
/*environment_id*/ None,
"search",
/*arguments*/ None,
/*meta*/ None,
+9 -9
View File
@@ -14,11 +14,12 @@ use crate::mcp::effective_mcp_servers;
impl McpRuntime {
/// Reuses the selected connection and catalog without starting unrelated servers.
/// Configuration and metadata belong to the same captured runtime publication.
/// Resolves the selector, configuration, and metadata in one captured runtime publication.
/// An expected registration excludes mismatched connections before discovery.
pub async fn server_status_snapshot(
&self,
server: &str,
environment_id: Option<&str>,
detail: McpSnapshotDetail,
expected_registration: Option<&ResolvedMcpServer>,
runtime_context: &McpRuntimeContext,
@@ -34,16 +35,15 @@ impl McpRuntime {
.as_ref()
.ok_or_else(|| anyhow::anyhow!("MCP runtime is not configured"))?,
);
let server = config
.mcp_server_catalog
.resolve_server_key(server, environment_id);
let mut servers = effective_mcp_servers(&config, current.auth.as_ref());
servers.retain(|name, _| {
Some(name) == config.mcp_server_catalog.resolve_server_name(server)
servers.retain(|key, _| {
key == &server
&& expected_registration
.is_none_or(|expected| config.mcp_server_catalog.server(name) == Some(expected))
.is_none_or(|expected| config.mcp_server_catalog.server(key) == Some(expected))
});
let servers = servers
.into_values()
.map(|value| (server.to_owned(), value))
.collect::<std::collections::HashMap<_, _>>();
if config.xaa_enabled
&& servers
.values()
@@ -52,7 +52,7 @@ impl McpRuntime {
// Finish selected startup and any rejected-grant cleanup before reading auth.
current
.connections
.list_available_server_infos(|name| name == server)
.list_available_server_infos(|name| name == &server)
.await;
}
let auth_statuses = compute_auth_statuses(
@@ -28,11 +28,6 @@ pub(super) async fn route(
});
};
router
.request_user_interaction(
Some(events),
&authority,
server_key.name().to_owned(),
request,
)
.request_user_interaction(Some(events), &authority, server_key, request)
.await
}
@@ -76,6 +76,7 @@ async fn forward_events_filters_private_events_before_blocked_send_is_cancelled(
for msg in [
EventMsg::McpStartupUpdate(McpStartupUpdateEvent {
server: "pending".to_string(),
environment_id: None,
status: McpStartupStatus::Starting,
}),
EventMsg::McpStartupComplete(McpStartupCompleteEvent::default()),
+16 -4
View File
@@ -1264,7 +1264,10 @@ impl CodexThread {
pub async fn mcp_connection_statuses(
&self,
config: &codex_mcp::McpConfig,
) -> std::collections::HashMap<String, codex_protocol::mcp::McpServerConnectionStatus> {
) -> std::collections::HashMap<
codex_mcp::McpServerKey,
codex_protocol::mcp::McpServerConnectionStatus,
> {
self.session
.services
.mcp_runtime
@@ -1388,6 +1391,7 @@ impl CodexThread {
pub async fn read_mcp_resource(
&self,
server: &str,
environment_id: Option<&str>,
params: ReadResourceRequestParams,
) -> anyhow::Result<serde_json::Value> {
self.session.refresh_mcp_if_dirty().await;
@@ -1395,7 +1399,7 @@ impl CodexThread {
.session
.services
.mcp_runtime
.latest_read_resource(server, params)
.latest_read_resource(server, environment_id, params)
.await?;
Ok(serde_json::to_value(result)?)
@@ -1405,6 +1409,7 @@ impl CodexThread {
pub async fn mcp_server_status_snapshot(
&self,
server: &str,
environment_id: Option<&str>,
detail: codex_mcp::McpSnapshotDetail,
expected_registration: Option<&codex_mcp::ResolvedMcpServer>,
) -> anyhow::Result<(
@@ -1416,7 +1421,13 @@ impl CodexThread {
self.session
.services
.mcp_runtime
.server_status_snapshot(server, detail, expected_registration, &runtime_context)
.server_status_snapshot(
server,
environment_id,
detail,
expected_registration,
&runtime_context,
)
.await
}
@@ -1460,6 +1471,7 @@ impl CodexThread {
pub async fn call_mcp_tool(
&self,
server: &str,
environment_id: Option<&str>,
tool: &str,
arguments: Option<serde_json::Value>,
meta: Option<serde_json::Value>,
@@ -1468,7 +1480,7 @@ impl CodexThread {
self.session
.services
.mcp_runtime
.latest_call_tool_without_attribution(server, tool, arguments, meta)
.latest_call_tool_without_attribution(server, environment_id, tool, arguments, meta)
.await
}
+10 -2
View File
@@ -184,6 +184,10 @@ pub(crate) async fn handle_mcp_tool_call(
let invocation = McpInvocation {
server: server.clone(),
environment_id: advertised_tool
.server_key()
.scoped_environment_id()
.map(str::to_owned),
tool: tool_name.clone(),
arguments: arguments_value.clone(),
};
@@ -959,7 +963,7 @@ async fn maybe_request_codex_apps_auth_elicitation(
let response = sess
.request_mcp_server_elicitation(
turn_context,
CODEX_APPS_MCP_SERVER_NAME.to_string(),
McpServerKey::unscoped(CODEX_APPS_MCP_SERVER_NAME),
request_id,
request,
)
@@ -1136,12 +1140,14 @@ async fn notify_mcp_tool_call_started(
) {
let McpInvocation {
server,
environment_id,
tool,
arguments,
} = invocation;
let item = TurnItem::McpToolCall(McpToolCallItem {
id: call_id.to_string(),
server,
environment_id,
tool,
arguments: arguments.unwrap_or(JsonValue::Null),
connector_id: item_metadata.connector_id,
@@ -1182,12 +1188,14 @@ async fn notify_mcp_tool_call_completed(
};
let McpInvocation {
server,
environment_id,
tool,
arguments,
} = invocation;
let item = TurnItem::McpToolCall(McpToolCallItem {
id: call_id.to_string(),
server,
environment_id,
tool,
arguments: arguments.unwrap_or(JsonValue::Null),
connector_id: item_metadata.connector_id,
@@ -1926,7 +1934,7 @@ pub(crate) async fn request_mcp_tool_user_approval(
prompt_options,
});
let response = sess
.request_mcp_server_elicitation(turn_context, server.clone(), request_id, request)
.request_mcp_server_elicitation(turn_context, server_key.clone(), request_id, request)
.await;
// Missing replies are interruptions. A received Cancel needs provenance
// from its source; RPC success alone does not establish a user action.
@@ -32,6 +32,10 @@ pub(crate) async fn prepare_approval_action(
.map_err(|err| FunctionCallError::RespondToModel(format!("{err:#}")))?;
let mcp_invocation = McpInvocation {
server: advertised_tool.server_name().to_owned(),
environment_id: advertised_tool
.server_key()
.scoped_environment_id()
.map(str::to_owned),
tool: advertised_tool.tool.name.to_string(),
arguments,
};
+17
View File
@@ -987,6 +987,7 @@ fn custom_mcp_tool_question_offers_session_remember_and_always_allow() {
fn custom_servers_support_session_and_persistent_approval() {
let invocation = McpInvocation {
server: "custom_server".to_string(),
environment_id: None,
tool: "run_action".to_string(),
arguments: None,
};
@@ -1042,6 +1043,7 @@ fn custom_servers_support_session_and_persistent_approval() {
fn codex_apps_connectors_support_persistent_approval() {
let invocation = McpInvocation {
server: CODEX_APPS_MCP_SERVER_NAME.to_string(),
environment_id: None,
tool: "calendar/list_events".to_string(),
arguments: None,
};
@@ -1592,6 +1594,7 @@ async fn mcp_tool_call_item_includes_app_identity() {
"call-plugin",
McpInvocation {
server: CODEX_APPS_MCP_SERVER_NAME.to_string(),
environment_id: None,
tool: "echo".to_string(),
arguments: None,
},
@@ -2104,6 +2107,7 @@ fn approval_elicitation_meta_merges_session_and_always_persist_for_custom_server
fn guardian_mcp_review_request_includes_invocation_metadata() {
let invocation = McpInvocation {
server: CODEX_APPS_MCP_SERVER_NAME.to_string(),
environment_id: None,
tool: "browser_navigate".to_string(),
arguments: Some(serde_json::json!({
"url": "https://example.com",
@@ -2145,6 +2149,7 @@ fn guardian_mcp_review_request_includes_invocation_metadata() {
fn guardian_mcp_review_request_includes_annotations_when_present() {
let invocation = McpInvocation {
server: "custom_server".to_string(),
environment_id: None,
tool: "dangerous_tool".to_string(),
arguments: None,
};
@@ -2193,6 +2198,7 @@ fn guardian_mcp_review_request_includes_annotations_when_present() {
fn guardian_mcp_review_request_ignores_untrusted_connected_account_email() {
let invocation = McpInvocation {
server: "custom_server".to_string(),
environment_id: None,
tool: "dangerous_tool".to_string(),
arguments: None,
};
@@ -2972,6 +2978,7 @@ async fn approve_mode_skips_when_annotations_do_not_require_approval() {
let turn_context = Arc::new(turn_context);
let invocation = McpInvocation {
server: "custom_server".to_string(),
environment_id: None,
tool: "read_only_tool".to_string(),
arguments: None,
};
@@ -3056,6 +3063,7 @@ async fn guardian_mode_skips_auto_when_annotations_do_not_require_approval() {
let turn_context = Arc::new(turn_context);
let invocation = McpInvocation {
server: "custom_server".to_string(),
environment_id: None,
tool: "read_only_tool".to_string(),
arguments: None,
};
@@ -3117,6 +3125,7 @@ async fn permission_request_hook_allows_mcp_tool_call() {
let turn_context = Arc::new(turn_context);
let invocation = McpInvocation {
server: "memory".to_string(),
environment_id: None,
tool: "create_entities".to_string(),
arguments: Some(serde_json::json!({
"entities": [{
@@ -3216,6 +3225,7 @@ async fn permission_request_hook_uses_hook_tool_name_without_metadata() {
let turn_context = Arc::new(turn_context);
let invocation = McpInvocation {
server: "memory".to_string(),
environment_id: None,
tool: "create_entities".to_string(),
arguments: Some(serde_json::json!({ "entities": [] })),
};
@@ -3292,6 +3302,7 @@ async fn permission_request_hook_runs_after_remembered_mcp_approval() {
);
let invocation = McpInvocation {
server: "memory".to_string(),
environment_id: None,
tool: "create_entities".to_string(),
arguments: Some(serde_json::json!({ "entities": [] })),
};
@@ -3409,6 +3420,7 @@ async fn strict_auto_review_forces_guardian_for_mcp_policy_skip(
let turn_context = Arc::new(turn_context);
let invocation = McpInvocation {
server: "custom_server".to_string(),
environment_id: None,
tool: "dangerous_tool".to_string(),
arguments: Some(serde_json::json!({ "calendar_id": "primary" })),
};
@@ -3501,6 +3513,7 @@ async fn assert_mcp_user_approval_persistence(
*session.active_turn.lock().await = Some(ActiveTurn::default());
let invocation = McpInvocation {
server: "memory".to_string(),
environment_id: None,
tool: "create_entities".to_string(),
arguments: Some(serde_json::json!({ "entities": [] })),
};
@@ -3594,6 +3607,7 @@ async fn prompt_mode_waits_for_approval_when_annotations_do_not_require_approval
}
let invocation = McpInvocation {
server: "custom_server".to_string(),
environment_id: None,
tool: "read_only_tool".to_string(),
arguments: None,
};
@@ -3662,6 +3676,7 @@ async fn full_access_mode_skips_mcp_tool_approval_for_all_approval_modes() {
let turn_context = Arc::new(turn_context);
let invocation = McpInvocation {
server: CODEX_APPS_MCP_SERVER_NAME.to_string(),
environment_id: None,
tool: "dangerous_tool".to_string(),
arguments: Some(serde_json::json!({ "id": 1 })),
};
@@ -3724,6 +3739,7 @@ async fn approve_mode_skips_guardian_in_every_permission_mode() {
let invocation = McpInvocation {
server: CODEX_APPS_MCP_SERVER_NAME.to_string(),
environment_id: None,
tool: "dangerous_tool".to_string(),
arguments: Some(serde_json::json!({ "id": 1 })),
};
@@ -3810,6 +3826,7 @@ async fn approval_metadata_is_released_when_the_invocation_future_is_dropped() {
let review_context = GuardianReviewContext::from(Arc::new(turn_context));
let invocation = McpInvocation {
server: CODEX_APPS_MCP_SERVER_NAME.to_string(),
environment_id: None,
tool: "write_record".to_string(),
arguments: Some(serde_json::json!({"value": 42})),
};
@@ -93,6 +93,7 @@ fn mcp_tool_call(id: &str, server: &str, status: McpToolCallStatus) -> TurnItem
TurnItem::McpToolCall(McpToolCallItem {
id: id.to_string(),
server: server.to_string(),
environment_id: None,
tool: "create_thread".to_string(),
arguments: serde_json::Value::Null,
connector_id: None,
+4 -1
View File
@@ -22,6 +22,7 @@ use codex_mcp::ElicitationReviewRequest;
use codex_mcp::ElicitationReviewer;
use codex_mcp::ElicitationReviewerHandle;
use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY;
use codex_mcp::McpServerKey;
use codex_otel::auth_storage::AuthStorageOriginator;
use codex_protocol::capabilities::CapabilityRootLocation;
use codex_protocol::capabilities::SelectedCapabilityRoot;
@@ -782,10 +783,11 @@ impl Session {
pub async fn request_mcp_server_elicitation(
&self,
turn_context: &TurnContext,
server_name: String,
server_key: McpServerKey,
request_id: RequestId,
request: ElicitationRequest,
) -> Option<ElicitationResponse> {
let server_name = server_key.name().to_owned();
// This callback carries a request ID, not a reliable tool invocation owner.
// Do not report its human wait as ordinary tool execution.
if turn_context
@@ -827,6 +829,7 @@ impl Session {
let event = EventMsg::ElicitationRequest(ElicitationRequestEvent {
turn_id: Some(turn_context.sub_id.clone()),
server_name,
environment_id: server_key.scoped_environment_id().map(str::to_owned),
id,
request,
});
+2
View File
@@ -7,6 +7,7 @@ fn plugin_install_elicitation_telemetry_metadata_requires_install_tool_suggestio
let event = EventMsg::ElicitationRequest(ElicitationRequestEvent {
turn_id: Some("turn-1".to_string()),
server_name: "codex_apps".to_string(),
environment_id: None,
id: codex_protocol::mcp::RequestId::String("request-1".to_string()),
request: codex_protocol::approvals::ElicitationRequest::Form {
meta: Some(json!({
@@ -36,6 +37,7 @@ fn plugin_install_elicitation_telemetry_metadata_requires_install_tool_suggestio
let enable_event = EventMsg::ElicitationRequest(ElicitationRequestEvent {
turn_id: Some("turn-1".to_string()),
server_name: "codex_apps".to_string(),
environment_id: None,
id: codex_protocol::mcp::RequestId::String("request-2".to_string()),
request: codex_protocol::approvals::ElicitationRequest::Form {
meta: Some(json!({
+2 -1
View File
@@ -731,7 +731,7 @@ async fn request_mcp_server_elicitation_waits_for_user_response(
session
.request_mcp_server_elicitation(
&turn_context,
"codex_apps".to_string(),
codex_mcp::McpServerKey::unscoped("codex_apps"),
RequestId::String("request-1".into()),
request,
)
@@ -750,6 +750,7 @@ async fn request_mcp_server_elicitation_waits_for_user_response(
codex_protocol::approvals::ElicitationRequestEvent {
turn_id: Some(turn_context.sub_id.clone()),
server_name: "codex_apps".to_string(),
environment_id: None,
id: codex_protocol::mcp::RequestId::String("request-1".to_string()),
request,
}
+1
View File
@@ -164,6 +164,7 @@ fn realtime_user_verification_notice_excludes_request_payload() {
let event = EventMsg::ElicitationRequest(codex_protocol::approvals::ElicitationRequestEvent {
turn_id: None,
server_name: "private-server-name".to_string(),
environment_id: None,
id: codex_protocol::mcp::RequestId::String("private-request-id".to_string()),
request: codex_protocol::approvals::ElicitationRequest::UserVerification {
meta: None,
@@ -410,12 +410,14 @@ async fn emit_tool_call_begin(
) {
let McpInvocation {
server,
environment_id,
tool,
arguments,
} = invocation;
let item = TurnItem::McpToolCall(McpToolCallItem {
id: call_id.to_string(),
server,
environment_id,
tool,
arguments: arguments.unwrap_or(Value::Null),
connector_id: None,
@@ -455,12 +457,14 @@ async fn emit_tool_call_end(
};
let McpInvocation {
server,
environment_id,
tool,
arguments,
} = invocation;
let item = TurnItem::McpToolCall(McpToolCallItem {
id: call_id.to_string(),
server,
environment_id,
tool,
arguments: arguments.unwrap_or(Value::Null),
connector_id: None,
@@ -66,15 +66,22 @@ impl ListMcpResourceTemplatesHandler {
let arguments = resource_arguments(&payload)?;
let args = ListResourceArgs::parse(arguments.clone())?;
let target = args.target(&step_context);
let invocation = McpInvocation {
server: args.server.clone().unwrap_or_else(|| "codex".to_string()),
environment_id: target
.as_ref()
.ok()
.and_then(Option::as_ref)
.and_then(|(key, _)| key.scoped_environment_id())
.map(str::to_owned),
tool: "list_mcp_resource_templates".to_string(),
arguments: arguments.clone(),
};
run_resource_operation(&session, &step_context, &call_id, invocation, async {
if let Some((server_key, params)) = args.target(&step_context)? {
if let Some((server_key, params)) = target? {
let result = mcp
.list_resource_templates(server_key, params)
.await

Some files were not shown because too many files have changed in this diff Show More