Report runtime MCP connection status (#40068)

## Why

MCP inventory can be cached or collected separately from a thread's live
connections, so tool availability alone does not describe the current runtime
state.

## What changed

- Add a nullable `runtimeStatus` to `mcpServerStatus/list` for thread-scoped
  requests, covering not-started, starting, connected, authentication-required,
  failed, cancelled, and disabled connections.
- Observe published connection state without starting or reconnecting servers,
  and return an unknown status when no thread is supplied or the active
  configuration no longer matches the published registration.
- Show connection state and tool counts in the compact `/mcp` view while
  retaining the detailed inventory in `/mcp verbose` and compatibility with
  servers that omit `runtimeStatus`.

## Testing

- Cover runtime status transitions, deferred and disabled servers, closed
  transports, configuration changes, protocol compatibility, and TUI rendering.

GitOrigin-RevId: e3bb6efe652f0fa8b3c97d5c53e4729b3a87cd91
This commit is contained in:
Eric Traut
2026-08-22 05:54:43 +00:00
committed by copyberry
parent 4f39251a01
commit 343074d420
34 changed files with 899 additions and 48 deletions
@@ -13816,6 +13816,18 @@
"title": "McpResourceReadResponse",
"type": "object"
},
"McpServerConnectionStatus": {
"enum": [
"notStarted",
"starting",
"connected",
"authenticationRequired",
"failed",
"cancelled",
"disabled"
],
"type": "string"
},
"McpServerEventNotification": {
"properties": {
"method": {
@@ -14040,6 +14052,17 @@
},
"type": "array"
},
"runtimeStatus": {
"anyOf": [
{
"$ref": "#/definitions/v2/McpServerConnectionStatus"
},
{
"type": "null"
}
],
"description": "Current thread-runtime connection state; null when unavailable or the configuration changed."
},
"serverInfo": {
"anyOf": [
{
@@ -10026,6 +10026,18 @@
"title": "McpResourceReadResponse",
"type": "object"
},
"McpServerConnectionStatus": {
"enum": [
"notStarted",
"starting",
"connected",
"authenticationRequired",
"failed",
"cancelled",
"disabled"
],
"type": "string"
},
"McpServerEventNotification": {
"properties": {
"method": {
@@ -10250,6 +10262,17 @@
},
"type": "array"
},
"runtimeStatus": {
"anyOf": [
{
"$ref": "#/definitions/McpServerConnectionStatus"
},
{
"type": "null"
}
],
"description": "Current thread-runtime connection state; null when unavailable or the configuration changed."
},
"serverInfo": {
"anyOf": [
{
@@ -11,6 +11,18 @@
],
"type": "string"
},
"McpServerConnectionStatus": {
"enum": [
"notStarted",
"starting",
"connected",
"authenticationRequired",
"failed",
"cancelled",
"disabled"
],
"type": "string"
},
"McpServerInfo": {
"description": "Presentation metadata advertised by an initialized MCP server.",
"properties": {
@@ -78,6 +90,17 @@
},
"type": "array"
},
"runtimeStatus": {
"anyOf": [
{
"$ref": "#/definitions/McpServerConnectionStatus"
},
{
"type": "null"
}
],
"description": "Current thread-runtime connection state; null when unavailable or the configuration changed."
},
"serverInfo": {
"anyOf": [
{
@@ -0,0 +1,5 @@
// GENERATED CODE! DO NOT MODIFY BY HAND!
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
export type McpServerConnectionStatus = "notStarted" | "starting" | "connected" | "authenticationRequired" | "failed" | "cancelled" | "disabled";
@@ -6,5 +6,10 @@ import type { Resource } from "../Resource";
import type { ResourceTemplate } from "../ResourceTemplate";
import type { Tool } from "../Tool";
import type { McpAuthStatus } from "./McpAuthStatus";
import type { McpServerConnectionStatus } from "./McpServerConnectionStatus";
export type McpServerStatus = { name: string, pluginId: string | null, serverInfo: McpServerInfo | null, tools: { [key in string]?: Tool }, resources: Array<Resource>, resourceTemplates: Array<ResourceTemplate>, authStatus: McpAuthStatus, };
export type McpServerStatus = { name: string,
/**
* Current thread-runtime connection state; null when unavailable or the configuration changed.
*/
runtimeStatus: McpServerConnectionStatus | null, pluginId: string | null, serverInfo: McpServerInfo | null, tools: { [key in string]?: Tool }, resources: Array<Resource>, resourceTemplates: Array<ResourceTemplate>, authStatus: McpAuthStatus, };
@@ -267,6 +267,7 @@ export type { McpElicitationUntitledMultiSelectEnumSchema } from "./McpElicitati
export type { McpElicitationUntitledSingleSelectEnumSchema } from "./McpElicitationUntitledSingleSelectEnumSchema";
export type { McpResourceReadParams } from "./McpResourceReadParams";
export type { McpResourceReadResponse } from "./McpResourceReadResponse";
export type { McpServerConnectionStatus } from "./McpServerConnectionStatus";
export type { McpServerElicitationAction } from "./McpServerElicitationAction";
export type { McpServerElicitationRequestParams } from "./McpServerElicitationRequestParams";
export type { McpServerElicitationRequestResponse } from "./McpServerElicitationRequestResponse";
@@ -30,6 +30,19 @@ v2_enum_from_core!(
}
);
v2_enum_from_core!(
#[ts(rename_all = "camelCase")]
pub enum McpServerConnectionStatus from codex_protocol::mcp::McpServerConnectionStatus {
NotStarted,
Starting,
Connected,
AuthenticationRequired,
Failed,
Cancelled,
Disabled
}
);
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
@@ -61,6 +74,8 @@ pub enum McpServerStatusDetail {
#[ts(export_to = "v2/")]
pub struct McpServerStatus {
pub name: String,
/// Current thread-runtime connection state; null when unavailable or the configuration changed.
pub runtime_status: Option<McpServerConnectionStatus>,
pub plugin_id: Option<String>,
pub server_info: Option<McpServerInfo>,
pub tools: std::collections::HashMap<String, McpTool>,
@@ -2506,6 +2506,7 @@ fn mcp_server_status_serializes_absent_server_info_as_null() {
let response = ListMcpServerStatusResponse {
data: vec![McpServerStatus {
name: "not-ready".to_string(),
runtime_status: None,
plugin_id: None,
server_info: None,
tools: HashMap::new(),
@@ -2521,6 +2522,7 @@ fn mcp_server_status_serializes_absent_server_info_as_null() {
json!({
"data": [{
"name": "not-ready",
"runtimeStatus": null,
"pluginId": null,
"serverInfo": null,
"tools": {},
@@ -2533,6 +2535,33 @@ fn mcp_server_status_serializes_absent_server_info_as_null() {
);
}
#[test]
fn mcp_server_status_accepts_older_inventory_without_runtime_status() {
let status: McpServerStatus = serde_json::from_value(json!({
"name": "older-server",
"pluginId": null,
"serverInfo": null,
"tools": {},
"resources": [],
"resourceTemplates": [],
"authStatus": "unknown",
}))
.expect("older app-server inventory should deserialize");
assert_eq!(
status,
McpServerStatus {
name: "older-server".to_string(),
runtime_status: None,
plugin_id: None,
server_info: None,
tools: HashMap::new(),
resources: Vec::new(),
resource_templates: Vec::new(),
auth_status: McpAuthStatus::Unknown,
}
);
}
#[test]
fn mcp_server_status_updated_accepts_missing_thread_id() {
let notification: McpServerStatusUpdatedNotification = serde_json::from_value(json!({
@@ -2593,6 +2622,7 @@ fn mcp_server_status_serializes_absent_server_info_metadata_as_null() {
let response = ListMcpServerStatusResponse {
data: vec![McpServerStatus {
name: "initialized".to_string(),
runtime_status: None,
plugin_id: Some("lookup@test".to_string()),
server_info: Some(McpServerInfo {
name: "lookup-server".to_string(),
@@ -2615,6 +2645,7 @@ fn mcp_server_status_serializes_absent_server_info_metadata_as_null() {
json!({
"data": [{
"name": "initialized",
"runtimeStatus": null,
"pluginId": "lookup@test",
"serverInfo": {
"name": "lookup-server",
+1 -1
View File
@@ -277,7 +277,7 @@ Example with notification opt-out:
- `mcpServer/oauth/login` — start an OAuth login for a configured MCP server; pass `threadId` to resolve servers from that thread's selected plugins and executor, optionally pass `clientRegistration` (`auto`, `cimd`, or `dcr`) to override client registration for this login only, and receive an `authorization_url` followed by `mcpServer/oauthLogin/completed` once the browser flow finishes. Omitting `clientRegistration` automatically discovers the authorization server's supported registration methods; the override is never persisted in server configuration.
- `tool/requestUserInput` — prompt the user with 1–3 short questions for a tool call and return their answers (experimental).
- `config/mcpServer/reload` — reload MCP server config from disk and queue a refresh for loaded threads (applied on each thread's next active turn); returns `{}`. Use this after editing `config.toml` without restarting the server.
- `mcpServerStatus/list` — enumerate configured MCP servers with their tools, auth status, server info, owning `pluginId` (`null` for servers not contributed by a plugin), plus resources/resource templates for `full` detail; supports optional `threadId` and cursor+limit pagination. If `threadId` is omitted, the server reads from the latest global config directly. If `detail` is omitted, the server defaults to `full`. An `unknown` auth status means OAuth support could not be determined; `unsupported` means OAuth is known not to be supported.
- `mcpServerStatus/list` — enumerate configured MCP servers with their tools, auth status, server info, owning `pluginId` (`null` for servers not contributed by a plugin), and nullable `runtimeStatus` from the current thread’s published connections, plus resources/resource templates for `full` detail; supports optional `threadId` and cursor+limit pagination. If `threadId` is omitted, the server reads from the latest global config directly and `runtimeStatus` is `null`. Runtime status is also `null` when the latest server registration differs from the thread’s published configuration. Runtime status is observed without starting or reconnecting the thread’s servers; it can be `notStarted`, `starting`, `connected`, `authenticationRequired`, `failed`, `cancelled`, or `disabled`. Inventory may be cached or collected separately and does not prove that the thread is connected. Older servers omit `runtimeStatus`; clients should treat that as unknown. If `detail` is omitted, the server defaults to `full`. An `unknown` auth status means OAuth support could not be determined; `unsupported` means OAuth is known not to be supported.
- `mcpServer/resource/read` — read a resource from a configured MCP server by optional `threadId`, `server`, and `uri`, returning text/blob resource `contents`. Pass `originCallId` with `threadId` to scope a Codex app widget to the app and account of the completed tool call that produced it; successful scoped reads return the same `originCallId`. Optional `connectorId` restricts other hosted app resources to their originating connector. If `threadId` is omitted, the server reads from the latest MCP config directly.
- `mcpServer/event/stream/start` (experimental) — subscribe to an MCP event by `threadId`, `server`, `subscriptionId`, event `name`, `arguments`, and optional `_meta`.
- `mcpServer/event/stream/stop` (experimental) — stop the caller's event subscription by `subscriptionId`.
@@ -278,7 +278,7 @@ impl McpRequestProcessor {
let environment_manager = self.thread_manager.environment_manager();
tokio::spawn(async move {
let (mcp_config, runtime_context) = match thread {
let (mcp_config, runtime_context) = match thread.as_ref() {
Some(thread) => thread.runtime_mcp_config_and_context(&config).await,
None => {
let mcp_config = mcp_manager.runtime_config(&config).await;
@@ -288,41 +288,21 @@ impl McpRequestProcessor {
}
};
Self::list_mcp_server_status_task(
outgoing,
request,
let result = Self::list_mcp_server_status_response(
request.request_id.to_string(),
params,
mcp_config,
auth,
runtime_context,
mcp_manager,
thread,
)
.await;
outgoing.send_result(request, result).await;
});
Ok(())
}
async fn list_mcp_server_status_task(
outgoing: Arc<OutgoingMessageSender>,
request_id: ConnectionRequestId,
params: ListMcpServerStatusParams,
mcp_config: codex_mcp::McpConfig,
auth: Option<CodexAuth>,
runtime_context: McpRuntimeContext,
mcp_manager: Arc<McpManager>,
) {
let result = Self::list_mcp_server_status_response(
request_id.request_id.to_string(),
params,
mcp_config,
auth,
runtime_context,
mcp_manager,
)
.await;
outgoing.send_result(request_id, result).await;
}
async fn list_mcp_server_status_response(
request_id: String,
params: ListMcpServerStatusParams,
@@ -330,6 +310,7 @@ impl McpRequestProcessor {
auth: Option<CodexAuth>,
runtime_context: McpRuntimeContext,
mcp_manager: Arc<McpManager>,
thread: Option<Arc<codex_core::CodexThread>>,
) -> Result<ListMcpServerStatusResponse, JSONRPCErrorError> {
let detail = match params.detail.unwrap_or(McpServerStatusDetail::Full) {
McpServerStatusDetail::Full => McpSnapshotDetail::Full,
@@ -347,6 +328,10 @@ impl McpRequestProcessor {
)
.await;
let runtime_statuses = match thread {
Some(thread) => thread.mcp_connection_statuses(&mcp_config).await,
None => HashMap::new(),
};
let McpServerStatusSnapshot {
server_infos,
tools_by_server,
@@ -355,6 +340,7 @@ impl McpRequestProcessor {
auth_statuses,
mut server_names,
} = snapshot;
server_names.extend(runtime_statuses.keys().cloned());
server_names.extend(
auth_statuses
.keys()
@@ -388,6 +374,7 @@ impl McpRequestProcessor {
.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(name).and_then(
|server| match server.source() {
McpServerSource::Plugin(plugin)
@@ -21,6 +21,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::McpServerConnectionStatus;
use codex_app_server_protocol::McpServerOauthLoginCompletedNotification;
use codex_app_server_protocol::McpServerOauthLoginResponse;
use codex_app_server_protocol::McpServerStatusDetail;
@@ -30,6 +31,7 @@ use codex_app_server_protocol::ThreadStartResponse;
use codex_core::config::set_project_trust_level;
use codex_http_client::HttpClientBuilder;
use codex_protocol::config_types::TrustLevel;
use core_test_support::skip_if_remote;
use core_test_support::stdio_server_bin;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
@@ -465,6 +467,7 @@ async fn mcp_server_status_list_returns_raw_server_and_tool_names() -> Result<()
assert_eq!(response.data.len(), 1);
let status = &response.data[0];
assert_eq!(status.name, "some-server");
assert_eq!(status.runtime_status, None);
assert_eq!(status.plugin_id, None);
assert_eq!(
status.tools.keys().cloned().collect::<BTreeSet<_>>(),
@@ -626,6 +629,7 @@ url = "{mcp_server_url}/mcp"
assert_eq!(thread_response.data.len(), 1);
let status = &thread_response.data[0];
assert_eq!(status.name, "project-server");
assert_eq!(status.runtime_status, None);
assert_eq!(
status.tools.keys().cloned().collect::<BTreeSet<_>>(),
BTreeSet::from(["project_lookup".to_string()])
@@ -637,6 +641,179 @@ url = "{mcp_server_url}/mcp"
Ok(())
}
#[tokio::test]
async fn mcp_server_status_list_reports_thread_runtime_connections() -> Result<()> {
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let (mcp_server_url, mcp_server_handle) = start_mcp_server("lookup").await?;
let codex_home = TempDir::new()?;
mock_responses_config(&server.uri())
.with_extra_config(&format!(
"[mcp_servers.connected]\nurl = \"{mcp_server_url}/mcp\"\n\
[mcp_servers.disabled]\nurl = \"{mcp_server_url}/mcp\"\nenabled = false\n"
))
.write(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let ThreadStartResponse { thread, .. } = mcp.start_thread(ThreadStartParams::default()).await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_matching_notification("connected MCP server ready", |notification| {
notification.method == "mcpServer/startupStatus/updated"
&& notification.params.as_ref().is_some_and(|params| {
params["name"] == "connected" && params["status"] == "ready"
})
}),
)
.await??;
let response: ListMcpServerStatusResponse = mcp
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
cursor: None,
limit: None,
detail: Some(McpServerStatusDetail::ToolsAndAuthOnly),
thread_id: Some(thread.id.clone()),
},
})
.await?;
assert_eq!(
response
.data
.into_iter()
.map(|status| (status.name, status.runtime_status))
.collect::<Vec<_>>(),
vec![
(
"connected".to_string(),
Some(McpServerConnectionStatus::Connected)
),
(
"disabled".to_string(),
Some(McpServerConnectionStatus::Disabled)
),
]
);
// Inventory uses the latest config, but a same-name replacement is not the
// connection that this thread started. Unchanged registrations retain status.
let (replacement_url, replacement_handle) = start_mcp_server("replacement_lookup").await?;
mock_responses_config(&server.uri())
.with_extra_config(&format!(
"[mcp_servers.connected]\nurl = \"{replacement_url}/mcp\"\n\
[mcp_servers.disabled]\nurl = \"{mcp_server_url}/mcp\"\nenabled = false\n"
))
.write(codex_home.path())?;
let response: ListMcpServerStatusResponse = mcp
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
cursor: None,
limit: None,
detail: Some(McpServerStatusDetail::ToolsAndAuthOnly),
thread_id: Some(thread.id),
},
})
.await?;
assert_eq!(
response
.data
.into_iter()
.map(|status| (
status.name,
status.runtime_status,
status.tools.into_keys().collect::<BTreeSet<_>>()
))
.collect::<Vec<_>>(),
vec![
(
"connected".to_string(),
None,
BTreeSet::from(["replacement_lookup".to_string()])
),
(
"disabled".to_string(),
Some(McpServerConnectionStatus::Disabled),
BTreeSet::new()
),
]
);
replacement_handle.abort();
let _ = replacement_handle.await;
mcp_server_handle.abort();
let _ = mcp_server_handle.await;
Ok(())
}
#[tokio::test]
async fn mcp_server_status_list_reports_disconnected_stdio_transport() -> Result<()> {
skip_if_remote!(
Ok(()),
"the stdio executable and shutdown marker are host-local"
);
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let codex_home = TempDir::new()?;
let exit_file = codex_home.path().join("exit-mcp-server");
mock_responses_config(&server.uri())
.with_extra_config(&format!(
"[mcp_servers.stdio]\ncommand = {}\nstartup_timeout_sec = 2\n\
[mcp_servers.stdio.env]\nMCP_TEST_EXIT_FILE = {}\n",
toml::Value::String(stdio_server_bin()?),
toml::Value::String(exit_file.to_string_lossy().into_owned()),
))
.write(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let ThreadStartResponse { thread, .. } = mcp.start_thread(ThreadStartParams::default()).await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_matching_notification("stdio MCP server ready", |notification| {
notification.method == "mcpServer/startupStatus/updated"
&& notification
.params
.as_ref()
.is_some_and(|params| params["name"] == "stdio" && params["status"] == "ready")
}),
)
.await??;
for expected in [
McpServerConnectionStatus::Connected,
McpServerConnectionStatus::Failed,
] {
if expected == McpServerConnectionStatus::Failed {
std::fs::write(&exit_file, "exit")?;
}
timeout(DEFAULT_READ_TIMEOUT, async {
loop {
let response: ListMcpServerStatusResponse = mcp
.request(|request_id| ClientRequest::McpServerStatusList {
request_id,
params: ListMcpServerStatusParams {
cursor: None,
limit: None,
detail: Some(McpServerStatusDetail::ToolsAndAuthOnly),
thread_id: Some(thread.id.clone()),
},
})
.await?;
let statuses = response
.data
.into_iter()
.map(|status| (status.name, status.runtime_status))
.collect::<Vec<_>>();
if statuses == vec![("stdio".to_string(), Some(expected))] {
return Ok::<(), anyhow::Error>(());
}
sleep(Duration::from_millis(/*millis*/ 20)).await;
}
})
.await??;
}
Ok(())
}
#[derive(Clone)]
struct McpStatusServer {
tool_name: Arc<String>,
@@ -11,6 +11,8 @@ mod required;
mod resources;
#[path = "connection_manager/startup.rs"]
mod startup;
#[path = "connection_manager/status.rs"]
mod status;
#[path = "connection_manager/tool_catalog.rs"]
mod tool_catalog;
@@ -178,6 +180,7 @@ impl McpServerView {
/// A published view over a set of running MCP server connections.
pub(crate) struct McpConnectionSet {
servers: HashMap<String, McpServerView>,
disabled_servers: Vec<String>,
protocol_mode: crate::McpProtocolMode,
required_servers: Vec<String>,
optional_startup_deadline: OnceLock<tokio::time::Instant>,
@@ -230,6 +233,11 @@ impl McpConnectionSet {
let tool_plugin_provenance = crate::mcp::tool_plugin_provenance(&config);
let auth = auth.as_ref();
let mut servers = HashMap::new();
let disabled_servers = mcp_servers
.iter()
.filter(|(_, server)| !server.enabled())
.map(|(name, _)| name.clone())
.collect();
let mut required_servers = mcp_servers
.iter()
.filter(|(_, server)| server.enabled() && server.required())
@@ -661,6 +669,7 @@ impl McpConnectionSet {
}
let manager = Self {
servers,
disabled_servers,
protocol_mode,
required_servers,
optional_startup_deadline: OnceLock::new(),
@@ -720,6 +729,7 @@ impl McpConnectionSet {
pub fn empty(prefix_mcp_tool_names: bool) -> Self {
Self {
servers: HashMap::new(),
disabled_servers: Vec::new(),
protocol_mode: crate::McpProtocolMode::Legacy,
required_servers: Vec::new(),
optional_startup_deadline: OnceLock::new(),
@@ -91,12 +91,7 @@ pub(super) fn mcp_init_error_display(
format!(
"GitHub MCP does not support OAuth. Log in by adding a personal access token (https://github.com/settings/personal-access-tokens) to your environment and config.toml:\n[mcp_servers.{server_name}]\nbearer_token_env_var = CODEX_GITHUB_PERSONAL_ACCESS_TOKEN"
)
} else if error.is_authentication_required()
|| matches!(
error,
StartupOutcomeError::Failed { error, .. } if error.contains("Auth required")
)
{
} else if error.is_authentication_required() {
let recovery_hint = if config.is_some_and(|config| !config.is_local_environment()) {
"Use your client's MCP OAuth sign-in flow.".to_string()
} else {
@@ -0,0 +1,30 @@
//! Read-only connection state; observing a server must not start it.
use std::collections::HashMap;
use codex_protocol::mcp::McpServerConnectionStatus;
use super::McpConnectionSet;
impl McpConnectionSet {
pub(crate) async fn connection_statuses(&self) -> HashMap<String, McpServerConnectionStatus> {
use McpServerConnectionStatus as Status;
let mut statuses = self
.disabled_servers
.iter()
.map(|name| (name.clone(), Status::Disabled))
.collect::<HashMap<_, _>>();
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() {
Status::NotStarted
} else {
client.connection_status().await
};
statuses.insert(name.clone(), status);
}
statuses
}
}
@@ -92,6 +92,7 @@ impl McpConnectionSet {
) -> Self {
Self {
servers: HashMap::new(),
disabled_servers: Vec::new(),
protocol_mode: crate::McpProtocolMode::Legacy,
required_servers: Vec::new(),
optional_startup_deadline: OnceLock::new(),
@@ -441,6 +442,138 @@ async fn create_ready_async_managed_client(tools: Vec<ToolInfo>) -> AsyncManaged
}
}
#[tokio::test]
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());
let ready = create_ready_async_managed_client(Vec::new()).await;
ready.client().await.expect("ready client");
manager.insert_test_client("connected", ready);
for (name, error) in [
(
"failed",
StartupOutcomeError::Failed {
error: "broken".to_string(),
is_authentication_required: false,
},
),
(
"auth",
StartupOutcomeError::Failed {
error: "login".to_string(),
is_authentication_required: true,
},
),
(
"flattened-auth",
StartupOutcomeError::from(anyhow!("Auth required for server")),
),
("cancelled", StartupOutcomeError::Cancelled),
] {
let mut client = create_ready_async_managed_client(Vec::new()).await;
client.client = futures::future::ready(Err(error)).boxed().shared();
assert!(client.client().await.is_err());
manager.insert_test_client(name, client);
}
let mut pending = create_ready_async_managed_client(Vec::new()).await;
pending.client = futures::future::pending().boxed().shared();
pending.cached_server_info = Some(create_test_server_info("Cached"));
manager.insert_test_client("starting", pending.clone());
manager.insert_test_client("deferred", pending);
let (trigger, _receiver) = watch::channel(/*init*/ false);
Arc::get_mut(&mut manager.servers.get_mut("deferred").unwrap().connection)
.unwrap()
.startup_trigger = Some(trigger.clone());
let statuses = tokio::time::timeout(
Duration::from_millis(/*millis*/ 100),
manager.connection_statuses(),
)
.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),
]);
assert_eq!(statuses, expected);
assert!(!*trigger.borrow());
manager.test_client("connected").cancel_token.cancel();
expected.insert("connected".to_string(), Status::Cancelled);
assert_eq!(manager.connection_statuses().await, expected);
}
#[tokio::test(start_paused = true)]
async fn connection_statuses_follow_latest_reconnect_outcome() {
use codex_protocol::mcp::McpServerConnectionStatus as Status;
let recovered = create_test_managed_client(Vec::new()).await;
let attempts = Arc::new(AtomicUsize::new(0));
let started = Arc::new(Notify::new());
let release = Arc::new(Notify::new());
let finished = Arc::new(Notify::new());
let factory = {
let attempts = Arc::clone(&attempts);
let started = Arc::clone(&started);
let release = Arc::clone(&release);
let finished = Arc::clone(&finished);
Arc::new(move || {
let attempt = attempts.fetch_add(1, Ordering::SeqCst);
let recovered = recovered.clone();
let started = Arc::clone(&started);
let release = Arc::clone(&release);
let finished = Arc::clone(&finished);
async move {
started.notify_one();
release.notified().await;
finished.notify_one();
match attempt {
0 | 1 => Err(StartupOutcomeError::Failed {
error: "retry failed".to_string(),
is_authentication_required: attempt == 0,
}),
_ => Ok(recovered),
}
}
.boxed()
.shared()
})
};
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)]);
assert_eq!(
manager.connection_statuses().await,
expected(Status::Failed)
);
for status in [
Status::AuthenticationRequired,
Status::Failed,
Status::Connected,
] {
client.reconnect_failed_startup().await;
started.notified().await;
assert_eq!(
manager.connection_statuses().await,
expected(Status::Starting)
);
release.notify_one();
finished.notified().await;
assert_eq!(manager.connection_statuses().await, expected(status));
tokio::time::advance(CODEX_APPS_RECONNECT_INITIAL_BACKOFF * 2).await;
}
assert_eq!(attempts.load(Ordering::SeqCst), 3);
}
fn create_gated_async_managed_client(
client: ManagedClient,
) -> (
+8 -2
View File
@@ -6,6 +6,9 @@
//! Higher-level aggregation and resource/tool APIs live in
//! [`crate::connection_manager`].
#[path = "rmcp_client/status.rs"]
mod status;
use std::borrow::Cow;
use std::collections::BTreeMap;
use std::collections::HashMap;
@@ -150,6 +153,7 @@ pub(crate) type ManagedClientFuture =
#[derive(Default)]
struct CodexAppsStartupReconnectState {
current_client: Option<ManagedClient>,
last_error: Option<StartupOutcomeError>,
reconnect_in_flight: bool,
consecutive_failures: u32,
retry_not_before: Option<TokioInstant>,
@@ -230,11 +234,13 @@ impl CodexAppsStartupReconnect {
match result {
Ok(client) => {
state.current_client = Some(client);
state.last_error = None;
state.consecutive_failures = 0;
state.retry_not_before = None;
true
}
Err(error) => {
state.last_error = Some(error.clone());
state.consecutive_failures = state.consecutive_failures.saturating_add(1);
let retry_after = codex_apps_reconnect_backoff(state.consecutive_failures);
state.retry_not_before = Some(TokioInstant::now() + retry_after);
@@ -591,9 +597,9 @@ impl StartupOutcomeError {
match self {
Self::Cancelled => false,
Self::Failed {
error,
is_authentication_required,
..
} => *is_authentication_required,
} => *is_authentication_required || error.contains("Auth required"),
}
}
}
@@ -0,0 +1,46 @@
//! Observe the latest startup attempt without polling it or initiating a retry.
use codex_protocol::mcp::McpServerConnectionStatus as Status;
use super::AsyncManagedClient;
use super::StartupOutcomeError;
impl AsyncManagedClient {
pub(crate) async fn connection_status(&self) -> Status {
if self.cancel_token.is_cancelled() {
return Status::Cancelled;
}
let reconnect_outcome = if let Some(reconnect) = &self.startup_reconnect {
let state = reconnect
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if state.reconnect_in_flight {
return Status::Starting;
}
state
.current_client
.clone()
.map(Ok)
.or_else(|| state.last_error.clone().map(Err))
} else {
None
};
let outcome = reconnect_outcome.or_else(|| self.client.peek().cloned());
match outcome {
Some(Ok(client)) => {
if client.client.is_closed().await {
Status::Failed
} else {
Status::Connected
}
}
Some(Err(error)) if error.is_authentication_required() => {
Status::AuthenticationRequired
}
Some(Err(StartupOutcomeError::Failed { .. })) => Status::Failed,
Some(Err(StartupOutcomeError::Cancelled)) => Status::Cancelled,
None => Status::Starting,
}
}
}
+19
View File
@@ -519,6 +519,25 @@ impl McpRuntime {
self.current.load().connections.cancel_startup();
}
/// Observes matching published registrations without starting or reconnecting servers.
pub async fn connection_statuses(
&self,
config: &McpConfig,
) -> std::collections::HashMap<String, codex_protocol::mcp::McpServerConnectionStatus> {
let current = self.current.load_full();
let Some(published_config) = current.config.as_ref() else {
return HashMap::new();
};
let mut statuses = current.connections.connection_statuses().await;
statuses.retain(|name, _| {
published_config
.mcp_server_catalog
.server(name)
.is_some_and(|server| config.mcp_server_catalog.server(name) == Some(server))
});
statuses
}
pub(crate) fn latest_connections(&self) -> Arc<McpConnectionSet> {
Arc::clone(&self.current.load().connections)
}
+12
View File
@@ -775,6 +775,18 @@ impl CodexThread {
self.session.get_config().await
}
/// Observes this thread's published MCP connections that match the requested config.
pub async fn mcp_connection_statuses(
&self,
config: &codex_mcp::McpConfig,
) -> std::collections::HashMap<String, codex_protocol::mcp::McpServerConnectionStatus> {
self.session
.services
.mcp_runtime
.connection_statuses(config)
.await
}
/// Resolves MCP configuration and environment bindings from the same config snapshot.
pub async fn runtime_mcp_config_and_context(
&self,
@@ -10,6 +10,7 @@ use codex_core::StartThreadOptions;
use codex_core::TurnInputRequest;
use codex_exec_server::ExecutorFileSystem;
use codex_exec_server::RemoveOptions;
use codex_protocol::mcp::McpServerConnectionStatus;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
@@ -547,6 +548,14 @@ async fn cached_mcp_startup_is_eager_for_root_and_lazy_for_subagents() -> anyhow
&format!("Use the tools from {cached_process}."),
&format!("Echo from {cached_process}."),
);
let (mcp_config, _) = second_thread.current_mcp_config_and_runtime_context().await;
assert_eq!(
second_thread
.mcp_connection_statuses(&mcp_config)
.await
.get(SERVER_NAME),
Some(&McpServerConnectionStatus::NotStarted),
);
assert_eq!(
fs.read_file_text(&pid_file, Default::default(), /*sandbox*/ None)
.await?
+14
View File
@@ -9,6 +9,20 @@ use serde::Serialize;
use std::collections::HashMap;
use ts_rs::TS;
/// Observed state of an MCP connection in a published thread runtime.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(rename_all = "camelCase")]
pub enum McpServerConnectionStatus {
NotStarted,
Starting,
Connected,
AuthenticationRequired,
Failed,
Cancelled,
Disabled,
}
/// Extension ID for OpenAI form elicitation.
pub const OPENAI_FORM_EXTENSION_ID: &str = "openai/form";
/// Extension ID for standard MCP form elicitations that require user-entered input.
@@ -1020,8 +1020,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let service = TestToolServer::new();
let running = service.serve(stdio()).await?;
// Wait for the client to finish interacting with the server.
running.waiting().await?;
// A test can close an initialized transport without killing an arbitrary PID.
let exit_file = std::env::var_os("MCP_TEST_EXIT_FILE");
tokio::select! {
result = running.waiting() => { result?; }
_ = async {
let Some(exit_file) = exit_file else {
return std::future::pending::<()>().await;
};
while !std::path::Path::new(&exit_file).exists() {
sleep(Duration::from_millis(/*millis*/ 20)).await;
}
} => std::process::exit(0),
}
// Drain background tasks to ensure clean shutdown.
task::yield_now().await;
Ok(())
@@ -1557,6 +1557,7 @@ mod tests {
let statuses = vec![
McpServerStatus {
name: "docs".to_string(),
runtime_status: None,
plugin_id: None,
server_info: None,
tools: HashMap::from([(
@@ -1578,6 +1579,7 @@ mod tests {
},
McpServerStatus {
name: "disabled".to_string(),
runtime_status: None,
plugin_id: None,
server_info: None,
tools: HashMap::new(),
+1
View File
@@ -256,6 +256,7 @@ async fn handle_mcp_inventory_result_respects_origin_thread() {
app.handle_mcp_inventory_result(
Ok(vec![McpServerStatus {
name: "docs".to_string(),
runtime_status: None,
plugin_id: None,
server_info: None,
tools: HashMap::new(),
+54 -7
View File
@@ -5,6 +5,10 @@ use super::*;
#[path = "mcp_result.rs"]
mod result;
use crate::style::StatusTone;
use crate::style::accent_style;
use crate::style::status_style;
use codex_app_server_protocol::McpServerConnectionStatus;
use result::McpResultKind;
use result::McpToolResult;
@@ -518,9 +522,8 @@ pub(crate) fn new_mcp_tools_output(
/// alphabetically. The TUI deliberately does not enrich these rows from
/// client-local config because the app-server owns the remote MCP state.
///
/// This mirrors the layout of [`new_mcp_tools_output`] but sources data from
/// the paginated RPC response rather than the in-process `McpManager`. The
/// `detail` flag controls whether resources and resource templates are rendered.
/// Normal output is a compact connection summary. Full detail preserves the
/// tool, auth, resource, and resource-template inventory.
pub(crate) fn new_mcp_tools_output_from_statuses(
statuses: &[McpServerStatus],
detail: McpServerStatusDetail,
@@ -536,15 +539,54 @@ pub(crate) fn new_mcp_tools_output_from_statuses(
statuses.sort_by(|a, b| a.name.cmp(&b.name));
let has_any_tools = statuses.iter().any(|status| !status.tools.is_empty());
if !has_any_tools {
if !has_any_tools && matches!(detail, McpServerStatusDetail::Full) {
lines.push(" • No MCP tools available.".italic().into());
lines.push("".into());
}
for status in statuses {
let header: Vec<Span<'static>> = vec![" • ".into(), status.name.clone().into()];
lines.push(header.into());
let (label, style) = match status.runtime_status {
Some(McpServerConnectionStatus::Connected) => {
("connected", status_style(StatusTone::Success))
}
Some(McpServerConnectionStatus::Starting) => ("starting", accent_style()),
Some(McpServerConnectionStatus::AuthenticationRequired) => (
"authentication required",
status_style(StatusTone::Attention),
),
Some(McpServerConnectionStatus::Failed) => {
("failed", status_style(StatusTone::Failure))
}
Some(McpServerConnectionStatus::NotStarted) => ("not started", Style::default().dim()),
Some(McpServerConnectionStatus::Disabled) => ("disabled", Style::default().dim()),
Some(McpServerConnectionStatus::Cancelled) => ("cancelled", Style::default().dim()),
None if matches!(
status.auth_status,
codex_app_server_protocol::McpAuthStatus::NotLoggedIn
) =>
{
(
"authentication required",
status_style(StatusTone::Attention),
)
}
None => ("unknown", Style::default().dim()),
};
let count = status.tools.len();
let unit = if count == 1 { "tool" } else { "tools" };
lines.push(
vec![
" • ".set_style(style),
status.name.clone().bold(),
": ".into(),
label.set_style(style),
format!(" ({count} {unit})").dim(),
]
.into(),
);
if matches!(detail, McpServerStatusDetail::ToolsAndAuthOnly) {
continue;
}
let auth_status = match status.auth_status {
codex_app_server_protocol::McpAuthStatus::Unknown => McpAuthStatus::Unknown,
codex_app_server_protocol::McpAuthStatus::Unsupported => McpAuthStatus::Unsupported,
@@ -613,6 +655,11 @@ pub(crate) fn new_mcp_tools_output_from_statuses(
lines.push(Line::from(""));
}
if matches!(detail, McpServerStatusDetail::ToolsAndAuthOnly) {
lines.push("".into());
lines.push(" Use /mcp verbose for tools and resources.".dim().into());
}
PlainHistoryCell { lines }
}
/// A transient history cell that shows an animated spinner while the MCP
+115
View File
@@ -7,6 +7,121 @@ use serde_json::json;
const PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg==";
#[test]
fn mcp_inventory_connection_states() {
use McpServerConnectionStatus as Status;
let statuses = [
("unknown", None),
("starting", Some(Status::Starting)),
("failed", Some(Status::Failed)),
("disabled", Some(Status::Disabled)),
("deferred", Some(Status::NotStarted)),
("connected-empty", Some(Status::Connected)),
("cancelled", Some(Status::Cancelled)),
("auth", Some(Status::AuthenticationRequired)),
]
.into_iter()
.map(|(name, runtime_status)| McpServerStatus {
name: name.to_string(),
runtime_status,
plugin_id: None,
server_info: None,
tools: HashMap::new(),
resources: Vec::new(),
resource_templates: Vec::new(),
auth_status: McpAuthStatus::Unknown,
})
.collect::<Vec<_>>();
let cell =
new_mcp_tools_output_from_statuses(&statuses, McpServerStatusDetail::ToolsAndAuthOnly);
let rendered = cell
.display_lines(/*width*/ 100)
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!(rendered);
for (name, expected) in [
("auth", status_style(StatusTone::Attention)),
("cancelled", Style::default().dim()),
("connected-empty", status_style(StatusTone::Success)),
("deferred", Style::default().dim()),
("disabled", Style::default().dim()),
("failed", status_style(StatusTone::Failure)),
("starting", accent_style()),
("unknown", Style::default().dim()),
] {
let row = cell
.lines
.iter()
.find(|line| line.spans.get(1).is_some_and(|span| span.content == name))
.expect("connection-state row");
assert_eq!(
(row.spans[0].style, row.spans[3].style),
(expected, expected)
);
}
}
#[test]
fn mcp_inventory_older_app_server_authentication() {
let statuses = serde_json::from_value::<Vec<McpServerStatus>>(json!([
{
"name": "legacy-auth",
"tools": {},
"resources": [],
"resourceTemplates": [],
"authStatus": "notLoggedIn"
},
{
"name": "legacy-healthy",
"tools": {"lookup": {"name": "lookup", "inputSchema": {"type": "object"}}},
"resources": [],
"resourceTemplates": [],
"authStatus": "oAuth"
},
{
"name": "runtime-disabled",
"runtimeStatus": "disabled",
"tools": {},
"resources": [],
"resourceTemplates": [],
"authStatus": "notLoggedIn"
}
]))
.expect("mixed-version MCP statuses");
let cell =
new_mcp_tools_output_from_statuses(&statuses, McpServerStatusDetail::ToolsAndAuthOnly);
let rendered = cell
.display_lines(/*width*/ 100)
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!(rendered);
let styles = cell
.lines
.iter()
.filter(|line| {
line.spans
.get(1)
.is_some_and(|span| statuses.iter().any(|status| span.content == status.name))
})
.map(|line| (line.spans[0].style, line.spans[3].style))
.collect::<Vec<_>>();
let attention = status_style(StatusTone::Attention);
let neutral = Style::default().dim();
assert_eq!(
styles,
vec![
(attention, attention),
(neutral, neutral),
(neutral, neutral)
]
);
}
fn result(content: Vec<Value>) -> CallToolResult {
CallToolResult {
content,
@@ -0,0 +1,18 @@
---
source: tui/src/history_cell/mcp_tests.rs
expression: rendered
---
/mcp
🔌 MCP Tools
• auth: authentication required (0 tools)
• cancelled: cancelled (0 tools)
• connected-empty: connected (0 tools)
• deferred: not started (0 tools)
• disabled: disabled (0 tools)
• failed: failed (0 tools)
• starting: starting (0 tools)
• unknown: unknown (0 tools)
Use /mcp verbose for tools and resources.
@@ -0,0 +1,13 @@
---
source: tui/src/history_cell/mcp_tests.rs
expression: rendered
---
/mcp
🔌 MCP Tools
• legacy-auth: authentication required (0 tools)
• legacy-healthy: unknown (1 tool)
• runtime-disabled: disabled (0 tools)
Use /mcp verbose for tools and resources.
@@ -6,6 +6,6 @@ expression: rendered
🔌 MCP Tools
• plugin_docs
• Auth: Unknown
• Tools: lookup
• plugin_docs: unknown (1 tool)
Use /mcp verbose for tools and resources.
@@ -6,7 +6,7 @@ expression: rendered
🔌 MCP Tools
• plugin_docs
• plugin_docs: unknown (1 tool)
• Auth: Unsupported
• Tools: lookup
• Resources: Docs (file:///docs)
+2
View File
@@ -964,6 +964,7 @@ async fn mcp_tools_output_lists_tools_for_hyphenated_server_names() {
fn mcp_tools_output_from_statuses_renders_status_only_servers() {
let statuses = vec![McpServerStatus {
name: "plugin_docs".to_string(),
runtime_status: None,
plugin_id: None,
server_info: None,
tools: HashMap::from([(
@@ -995,6 +996,7 @@ fn mcp_tools_output_from_statuses_renders_status_only_servers() {
fn mcp_tools_output_from_statuses_renders_verbose_inventory() {
let statuses = vec![McpServerStatus {
name: "plugin_docs".to_string(),
runtime_status: None,
plugin_id: None,
server_info: None,
tools: HashMap::from([(
+78
View File
@@ -4,12 +4,42 @@ use crate::terminal_palette::StdoutColorLevel;
use crate::terminal_palette::best_color;
use crate::terminal_palette::default_bg;
use crate::terminal_palette::default_fg;
use crate::terminal_palette::effective_stdout_color_level;
use crate::terminal_palette::rgb_color;
use crate::terminal_palette::stdout_color_level;
use ratatui::style::Color;
use ratatui::style::Style;
const LIGHT_BG_ACCENT_RGB: (u8, u8, u8) = (0, 95, 135);
#[derive(Clone, Copy)]
pub(crate) enum StatusTone {
Success,
Attention,
Failure,
}
/// Semantic status colors that preserve the terminal's configured palette.
pub(crate) fn status_style(tone: StatusTone) -> Style {
status_style_for(tone, default_bg(), effective_stdout_color_level())
}
fn status_style_for(
tone: StatusTone,
terminal_bg: Option<(u8, u8, u8)>,
color_level: StdoutColorLevel,
) -> Style {
let light = terminal_bg.is_some_and(is_light);
let color = match (tone, color_level) {
(_, StdoutColorLevel::Unknown) => Color::Reset,
(StatusTone::Success, _) => Color::Green,
(StatusTone::Failure, _) => Color::Red,
// Yellow can disappear on light themes; use it only with a known dark background.
(StatusTone::Attention, _) if light || terminal_bg.is_none() => Color::Reset,
(StatusTone::Attention, _) => Color::Yellow,
};
Style::default().fg(color).bold()
}
// Decorative table rules should remain visible without competing with cell content.
const TABLE_SEPARATOR_FG_ALPHA: f32 = 0.20;
@@ -96,6 +126,54 @@ mod tests {
use pretty_assertions::assert_eq;
use ratatui::style::Modifier;
#[test]
fn status_colors_preserve_light_terminal_themes() {
for level in [StdoutColorLevel::TrueColor, StdoutColorLevel::Ansi256] {
for bg in [(255, 255, 255), (130, 130, 130), (220, 210, 180)] {
for (tone, color) in [
(StatusTone::Success, Color::Green),
(StatusTone::Attention, Color::Reset),
(StatusTone::Failure, Color::Red),
] {
assert_eq!(
status_style_for(tone, Some(bg), level),
Style::default().fg(color).bold(),
);
}
}
for bg in [(0, 0, 0), (0, 218, 0)] {
assert_eq!(
status_style_for(StatusTone::Attention, Some(bg), level),
Style::default().fg(Color::Yellow).bold(),
);
}
assert_eq!(
status_style_for(StatusTone::Attention, /*terminal_bg*/ None, level),
Style::default().fg(Color::Reset).bold(),
);
}
}
#[test]
fn status_colors_preserve_ansi16_and_no_color_fallbacks() {
for (tone, light, dark) in [
(StatusTone::Success, Color::Green, Color::Green),
(StatusTone::Attention, Color::Reset, Color::Yellow),
(StatusTone::Failure, Color::Red, Color::Red),
] {
for (bg, expected) in [((255, 255, 255), light), ((0, 0, 0), dark)] {
assert_eq!(
status_style_for(tone, Some(bg), StdoutColorLevel::Ansi16),
Style::default().fg(expected).bold()
);
assert_eq!(
status_style_for(tone, Some(bg), StdoutColorLevel::Unknown),
Style::default().fg(Color::Reset).bold()
);
}
}
}
#[test]
fn accent_style_uses_darker_cyan_on_light_backgrounds() {
let style = accent_style_for(Some((255, 255, 255)));