openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-05 15:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

[codex] align MCP status with callable thread manager

Browse Source
This commit is contained in:
Brian Brunner
2026-06-23 19:03:25 +00:00
parent 5013d10824
commit 68a359353a
6 changed files with 294 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
codex-rs/app-server/src/request_processors/mcp_processor.rs
View File

@@ -2,6 +2,18 @@ use super::*;
const MCP_TOOL_THREAD_ID_META_KEY: &str = "threadId";
enum McpServerStatusSource {
Installed {
thread: Arc<CodexThread>,
config: Box<Config>,
},
Configured {
mcp_config: Box<codex_mcp::McpConfig>,
auth: Option<CodexAuth>,
runtime_context: McpRuntimeContext,
},
}
#[derive(Clone)]
pub(crate) struct McpRequestProcessor {
auth_manager: Arc<AuthManager>,
@@ -204,7 +216,7 @@ impl McpRequestProcessor {
let request = request_id.clone();
let outgoing = Arc::clone(&self.outgoing);
let (config, thread) = match params.thread_id.as_deref() {
let source = match params.thread_id.as_deref() {
Some(thread_id) => {
let (_, thread) = self.load_thread(thread_id).await?;
let thread_config = thread.config().await;
@@ -213,37 +225,37 @@ impl McpRequestProcessor {
.load_latest_config_for_thread(thread_config.as_ref())
.await
.map_err(|err| internal_error(format!("failed to reload config: {err}")))?;
(config, Some(thread))
McpServerStatusSource::Installed {
thread,
config: Box::new(config),
}
}
None => (self.load_latest_config(/*fallback_cwd*/ None).await?, None),
};
let mcp_config = match thread {
Some(thread) => thread.runtime_mcp_config(&config).await,
None => {
self.thread_manager
let config = self.load_latest_config(/*fallback_cwd*/ None).await?;
let mcp_config = self
.thread_manager
.mcp_manager()
.runtime_config(&config)
.await
.await;
let auth = self.auth_manager.auth().await;
let environment_manager = self.thread_manager.environment_manager();
// This status path has no turn-selected environment. Use config cwd
// as the local stdio fallback; named environment stdio MCPs must
// declare their own absolute cwd.
let runtime_context = McpRuntimeContext::new(
Arc::clone(&environment_manager),
config.cwd.to_path_buf(),
);
McpServerStatusSource::Configured {
mcp_config: Box::new(mcp_config),
auth,
runtime_context,
}
}
};
let auth = self.auth_manager.auth().await;
let environment_manager = self.thread_manager.environment_manager();
// This status path has no turn-selected environment. Use config cwd
// as the local stdio fallback; named environment stdio MCPs must
// declare their own absolute cwd.
let runtime_context =
McpRuntimeContext::new(Arc::clone(&environment_manager), config.cwd.to_path_buf());
tokio::spawn(async move {
Self::list_mcp_server_status_task(
outgoing,
request,
params,
mcp_config,
auth,
runtime_context,
)
.await;
Self::list_mcp_server_status_task(outgoing, request, params, source).await;
});
Ok(())
}
@@ -252,42 +264,41 @@ impl McpRequestProcessor {
outgoing: Arc<OutgoingMessageSender>,
request_id: ConnectionRequestId,
params: ListMcpServerStatusParams,
mcp_config: codex_mcp::McpConfig,
auth: Option<CodexAuth>,
runtime_context: McpRuntimeContext,
source: McpServerStatusSource,
) {
let result = Self::list_mcp_server_status_response(
request_id.request_id.to_string(),
params,
mcp_config,
auth,
runtime_context,
)
.await;
outgoing.send_result(request_id, result).await;
}
async fn list_mcp_server_status_response(
request_id: String,
params: ListMcpServerStatusParams,
mcp_config: codex_mcp::McpConfig,
auth: Option<CodexAuth>,
runtime_context: McpRuntimeContext,
) -> Result<ListMcpServerStatusResponse, JSONRPCErrorError> {
let detail = match params.detail.unwrap_or(McpServerStatusDetail::Full) {
McpServerStatusDetail::Full => McpSnapshotDetail::Full,
McpServerStatusDetail::ToolsAndAuthOnly => McpSnapshotDetail::ToolsAndAuthOnly,
};
let snapshot = match source {
McpServerStatusSource::Installed { thread, config } => {
thread
.mcp_server_status_snapshot(config.as_ref(), detail)
.await
}
McpServerStatusSource::Configured {
mcp_config,
auth,
runtime_context,
} => {
collect_mcp_server_status_snapshot_with_detail(
mcp_config.as_ref(),
auth.as_ref(),
request_id.request_id.to_string(),
runtime_context,
detail,
)
.await
}
};
let result = Self::list_mcp_server_status_response(params, snapshot);
outgoing.send_result(request_id, result).await;
}
let snapshot = collect_mcp_server_status_snapshot_with_detail(
&mcp_config,
auth.as_ref(),
request_id,
runtime_context,
detail,
)
.await;
fn list_mcp_server_status_response(
params: ListMcpServerStatusParams,
snapshot: McpServerStatusSnapshot,
) -> Result<ListMcpServerStatusResponse, JSONRPCErrorError> {
let McpServerStatusSnapshot {
server_infos,
tools_by_server,

181
codex-rs/app-server/tests/suite/v2/mcp_server_status.rs
View File

@@ -5,21 +5,29 @@ use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::TestAppServer;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use app_test_support::to_response;
use app_test_support::write_chatgpt_auth;
use app_test_support::write_mock_responses_config_toml;
use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url;
use axum::Router;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::ListMcpServerStatusParams;
use codex_app_server_protocol::ListMcpServerStatusResponse;
use codex_app_server_protocol::McpServerStatusDetail;
use codex_app_server_protocol::McpServerToolCallParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_config::types::AuthCredentialsStoreMode;
use codex_core::config::set_project_trust_level;
use codex_protocol::config_types::TrustLevel;
use pretty_assertions::assert_eq;
use rmcp::handler::server::ServerHandler;
use rmcp::model::CallToolRequestParams;
use rmcp::model::CallToolResult;
use rmcp::model::Implementation;
use rmcp::model::JsonObject;
use rmcp::model::ListResourceTemplatesResult;
@@ -131,6 +139,17 @@ async fn mcp_server_status_list_uses_thread_project_local_config() -> Result<()>
)?;
std::fs::create_dir_all(workspace.path().join(".git"))?;
set_project_trust_level(codex_home.path(), workspace.path(), TrustLevel::Trusted)?;
let project_config_dir = workspace.path().join(".codex");
std::fs::create_dir_all(&project_config_dir)?;
std::fs::write(
project_config_dir.join("config.toml"),
format!(
r#"
[mcp_servers.project-server]
url = "{mcp_server_url}/mcp"
"#
),
)?;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
@@ -148,18 +167,6 @@ async fn mcp_server_status_list_uses_thread_project_local_config() -> Result<()>
.await??;
let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?;
let project_config_dir = workspace.path().join(".codex");
std::fs::create_dir_all(&project_config_dir)?;
std::fs::write(
project_config_dir.join("config.toml"),
format!(
r#"
[mcp_servers.project-server]
url = "{mcp_server_url}/mcp"
"#
),
)?;
let threadless_request_id = mcp
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
cursor: None,
@@ -206,6 +213,135 @@ url = "{mcp_server_url}/mcp"
Ok(())
}
#[tokio::test]
async fn mcp_server_status_list_never_advertises_uninstalled_codex_apps() -> Result<()> {
let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
let (apps_server_url, apps_server_handle) = start_apps_mcp_server("calendar_lookup").await?;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&apps_server_url,
)?;
let config_path = codex_home.path().join("config.toml");
let mut config_toml = std::fs::read_to_string(&config_path)?;
config_toml.push_str(
r#"
[features]
apps = false
"#,
);
std::fs::write(&config_path, &config_toml)?;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("chatgpt-token")
.account_id("account-123")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let stale_thread_start_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let stale_thread_start_response = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(stale_thread_start_id)),
)
.await??;
let ThreadStartResponse {
thread: stale_thread,
..
} = to_response(stale_thread_start_response)?;
std::fs::write(
&config_path,
config_toml.replace("apps = false", "apps = true"),
)?;
let stale_status_request_id = mcp
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
cursor: None,
limit: None,
detail: Some(McpServerStatusDetail::ToolsAndAuthOnly),
thread_id: Some(stale_thread.id),
})
.await?;
let stale_status_response = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(stale_status_request_id)),
)
.await??;
let stale_status_response: ListMcpServerStatusResponse = to_response(stale_status_response)?;
assert_eq!(stale_status_response.data, Vec::new());
let current_thread_start_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let current_thread_start_response = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(current_thread_start_id)),
)
.await??;
let ThreadStartResponse {
thread: current_thread,
..
} = to_response(current_thread_start_response)?;
let current_status_request_id = mcp
.send_list_mcp_server_status_request(ListMcpServerStatusParams {
cursor: None,
limit: None,
detail: Some(McpServerStatusDetail::ToolsAndAuthOnly),
thread_id: Some(current_thread.id.clone()),
})
.await?;
let current_status_response = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(current_status_request_id)),
)
.await??;
let current_status_response: ListMcpServerStatusResponse =
to_response(current_status_response)?;
assert_eq!(current_status_response.next_cursor, None);
assert_eq!(current_status_response.data.len(), 1);
let status = &current_status_response.data[0];
assert_eq!(status.name, "codex_apps");
assert_eq!(
status.tools.keys().cloned().collect::<BTreeSet<_>>(),
BTreeSet::from(["calendar_lookup".to_string()])
);
let tool_call_id = mcp
.send_mcp_server_tool_call_request(McpServerToolCallParams {
thread_id: current_thread.id,
server: "codex_apps".to_string(),
tool: "missing_tool".to_string(),
arguments: Some(json!({})),
meta: None,
})
.await?;
let error: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(tool_call_id)),
)
.await??;
assert!(
error.error.message.contains("unknown tool"),
"a listed server must receive the call; got: {error:?}"
);
assert!(!error.error.message.contains("unknown MCP server"));
apps_server_handle.abort();
let _ = apps_server_handle.await;
Ok(())
}
#[derive(Clone)]
struct McpStatusServer {
tool_name: Arc<String>,
@@ -242,6 +378,17 @@ impl ServerHandler for McpStatusServer {
meta: None,
})
}
async fn call_tool(
&self,
request: CallToolRequestParams,
_context: RequestContext<rmcp::service::RoleServer>,
) -> Result<CallToolResult, rmcp::ErrorData> {
Err(rmcp::ErrorData::invalid_params(
format!("unknown tool: {}", request.name),
None,
))
}
}
#[derive(Clone)]
@@ -451,6 +598,14 @@ url = "{underscore_server_url}/mcp"
}
async fn start_mcp_server(tool_name: &str) -> Result<(String, JoinHandle<()>)> {
start_mcp_server_at(tool_name, "/mcp").await
}
async fn start_apps_mcp_server(tool_name: &str) -> Result<(String, JoinHandle<()>)> {
start_mcp_server_at(tool_name, "/api/codex/ps/mcp").await
}
async fn start_mcp_server_at(tool_name: &str, route: &str) -> Result<(String, JoinHandle<()>)> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let addr = listener.local_addr()?;
let tool_name = Arc::new(tool_name.to_string());
@@ -463,7 +618,7 @@ async fn start_mcp_server(tool_name: &str) -> Result<(String, JoinHandle<()>)> {
Arc::new(LocalSessionManager::default()),
StreamableHttpServerConfig::default(),
);
let router = Router::new().nest_service("/mcp", mcp_service);
let router = Router::new().nest_service(route, mcp_service);
let handle = tokio::spawn(async move {
let _ = axum::serve(listener, router).await;

4
codex-rs/codex-mcp/src/connection_manager.rs
View File

@@ -338,6 +338,10 @@ impl McpConnectionManager {
!self.clients.is_empty()
}
pub(crate) fn server_names(&self) -> Vec<String> {
self.clients.keys().cloned().collect()
}
pub(crate) fn contains_server(&self, server_name: &str) -> bool {
self.clients.contains_key(server_name)
}

1
codex-rs/codex-mcp/src/lib.rs
View File

@@ -51,6 +51,7 @@ pub use plugin_config::parse_plugin_mcp_config;
pub use mcp::McpServerStatusSnapshot;
pub use mcp::McpSnapshotDetail;
pub use mcp::collect_mcp_server_status_snapshot_from_manager_with_detail;
pub use mcp::collect_mcp_server_status_snapshot_with_detail;
pub use mcp::read_mcp_resource;

33
codex-rs/codex-mcp/src/mcp/mod.rs
View File

@@ -396,6 +396,39 @@ pub async fn collect_mcp_server_status_snapshot_with_detail(
snapshot
}
/// Collects status from one installed manager generation.
///
/// Latest config is used only to refresh authentication status for servers
/// that are actually present in the manager's callable client map.
pub async fn collect_mcp_server_status_snapshot_from_manager_with_detail(
mcp_connection_manager: &McpConnectionManager,
config: &McpConfig,
auth: Option<&CodexAuth>,
detail: McpSnapshotDetail,
) -> McpServerStatusSnapshot {
let server_names = mcp_connection_manager.server_names();
let installed_servers = server_names.iter().cloned().collect::<HashSet<_>>();
let mcp_servers = effective_mcp_servers(config, auth)
.into_iter()
.filter(|(name, _)| installed_servers.contains(name))
.collect::<HashMap<_, _>>();
let auth_status_entries = compute_auth_statuses(
mcp_servers.iter(),
config.mcp_oauth_credentials_store_mode,
config.auth_keyring_backend_kind,
auth,
)
.await;
collect_mcp_server_status_snapshot_from_manager(
mcp_connection_manager,
auth_status_entries,
server_names,
detail,
)
.await
}
/// The Responses API requires tool names to match `^[a-zA-Z0-9_-]+$`.
/// MCP server/tool names are user-controlled, so sanitize the fully-qualified
/// name we expose to the model by replacing any disallowed character with `_`.

23
codex-rs/core/src/codex_thread.rs
View File

@@ -593,6 +593,29 @@ impl CodexThread {
self.codex.session.runtime_mcp_config(config).await
}
/// Returns the MCP inventory installed for this thread.
pub async fn mcp_server_status_snapshot(
&self,
config: &crate::config::Config,
detail: codex_mcp::McpSnapshotDetail,
) -> codex_mcp::McpServerStatusSnapshot {
let mcp_config = self.runtime_mcp_config(config).await;
let auth = self.codex.session.services.auth_manager.auth().await;
let manager = self
.codex
.session
.services
.mcp_connection_manager
.load_full();
codex_mcp::collect_mcp_server_status_snapshot_from_manager_with_detail(
manager.as_ref(),
&mcp_config,
auth.as_ref(),
detail,
)
.await
}
pub fn multi_agent_version(&self) -> Option<MultiAgentVersion> {
self.codex.session.multi_agent_version()
}