agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: make built-in MCPs first-class runtime servers (#21356)

Browse Source 
## DISCLAIMER
This is experimental and no production service must rely on this

## Why

Built-in MCPs are product-owned runtime capabilities, but they were
previously flattened into the same config-backed stdio path as
user-configured servers. That made them depend on a hidden `codex
builtin-mcp` re-exec path, exposed them through config-oriented CLI
flows, and erased distinctions the runtime needs to preserve—most
notably whether an MCP call should count as external context for
memory-mode pollution.

## What changed

- Model product-owned built-ins separately from config-backed MCP
servers via `BuiltinMcpServer` and `EffectiveMcpServer`.
- Launch built-ins in process through a reusable async transport instead
of the hidden `builtin-mcp` stdio subcommand.
- Keep config-oriented CLI operations such as `codex mcp
list/get/login/logout` scoped to configured servers, while merging
built-ins only into the effective runtime server set.
- Retain server metadata after launch so parallel-tool support and
context classification come from the live server set; built-in
`memories` is now classified as local Codex state rather than external
context.

## Test plan

- `cargo test -p codex-mcp`
- `cargo test -p codex-core --test suite
builtin_memories_mcp_call_does_not_mark_thread_memory_mode_polluted_when_configured`

---------

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
jif-oai
2026-05-07 10:36:32 +02:00
committed by GitHub
Unverified
parent 40e282849c
commit b2268999fe
35 changed files with 668 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+1 -2
View File
@@ -2131,10 +2131,10 @@ name = "codex-builtin-mcps"
version = "0.0.0"
dependencies = [
"anyhow",
"codex-config",
"codex-memories-mcp",
"codex-utils-absolute-path",
"pretty_assertions",
"tokio",
]
[[package]]
@@ -2182,7 +2182,6 @@ dependencies = [
"codex-app-server-protocol",
"codex-app-server-test-client",
"codex-arg0",
"codex-builtin-mcps",
"codex-chatgpt",
"codex-cloud-tasks",
"codex-config",
codex-rs/builtin-mcps/Cargo.toml
+1 -1
View File
@@ -13,9 +13,9 @@ workspace = true
[dependencies]
anyhow = { workspace = true }
codex-config = { workspace = true }
codex-memories-mcp = { workspace = true }
codex-utils-absolute-path = { workspace = true }
tokio = { workspace = true, features = ["io-util"] }
[dev-dependencies]
pretty_assertions = { workspace = true }
codex-rs/builtin-mcps/src/lib.rs
+72 -103
View File
@@ -1,132 +1,101 @@
//! Built-in MCP servers shipped with Codex.
//!
//! Built-ins use the same stdio MCP path as user-configured servers, but are
//! declared here so product-owned MCPs do not need to live in `codex-core`.
//! This crate owns the catalog of product-owned MCP servers and the small
//! amount of server-specific dispatch needed to run them. Runtime placement is
//! chosen by `codex-mcp`; built-ins should not be flattened into user-facing
//! MCP server config just to make them launchable.
use codex_config::McpServerConfig;
use codex_config::McpServerTransportConfig;
use codex_utils_absolute_path::AbsolutePathBuf;
use std::collections::HashMap;
use std::path::Path;
use tokio::io::AsyncRead;
use tokio::io::AsyncWrite;
pub const MEMORIES_MCP_SERVER_NAME: &str = "memories";
const BUILTIN_MCP_SUBCOMMAND: &str = "builtin-mcp";
/// Product-owned MCP servers that Codex can provide without user config.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum BuiltinMcpServer {
Memories,
}
#[derive(Debug, Clone, Copy)]
pub struct BuiltinMcpServerOptions<'a> {
pub codex_self_exe: Option<&'a Path>,
pub codex_home: &'a Path,
struct BuiltinMcpServerMetadata {
name: &'static str,
supports_parallel_tool_calls: bool,
pollutes_memory: bool,
}
impl BuiltinMcpServer {
const fn metadata(self) -> BuiltinMcpServerMetadata {
match self {
Self::Memories => BuiltinMcpServerMetadata {
name: MEMORIES_MCP_SERVER_NAME,
supports_parallel_tool_calls: true,
pollutes_memory: false,
},
}
}
pub const fn name(self) -> &'static str {
self.metadata().name
}
pub const fn supports_parallel_tool_calls(self) -> bool {
self.metadata().supports_parallel_tool_calls
}
pub const fn pollutes_memory(self) -> bool {
self.metadata().pollutes_memory
}
pub async fn serve<T>(self, codex_home: &Path, transport: T) -> anyhow::Result<()>
where
T: AsyncRead + AsyncWrite + Send + 'static,
{
match self {
Self::Memories => {
let codex_home = codex_utils_absolute_path::AbsolutePathBuf::try_from(codex_home)?;
codex_memories_mcp::run_server(&codex_home, transport).await
}
}
}
}
#[derive(Debug, Clone, Copy)]
pub struct BuiltinMcpServerOptions {
pub memories_enabled: bool,
}
pub fn configured_builtin_mcp_servers(
options: BuiltinMcpServerOptions<'_>,
) -> HashMap<String, McpServerConfig> {
let Some(codex_self_exe) = options.codex_self_exe else {
return HashMap::new();
};
let mut servers = HashMap::new();
pub fn enabled_builtin_mcp_servers(options: BuiltinMcpServerOptions) -> Vec<BuiltinMcpServer> {
let mut servers = Vec::new();
if options.memories_enabled {
servers.insert(
MEMORIES_MCP_SERVER_NAME.to_string(),
builtin_stdio_server_config(
codex_self_exe,
options.codex_home,
MEMORIES_MCP_SERVER_NAME,
),
);
servers.push(BuiltinMcpServer::Memories);
}
servers
}
pub async fn run_builtin_mcp_server(
name: &str,
codex_home: &AbsolutePathBuf,
) -> anyhow::Result<()> {
match name {
MEMORIES_MCP_SERVER_NAME => codex_memories_mcp::run_stdio_server(codex_home).await,
_ => anyhow::bail!("unknown built-in MCP server: {name}"),
}
}
fn builtin_stdio_server_config(
codex_self_exe: &Path,
codex_home: &Path,
name: &str,
) -> McpServerConfig {
McpServerConfig {
transport: McpServerTransportConfig::Stdio {
command: codex_self_exe.to_string_lossy().into_owned(),
args: vec![
BUILTIN_MCP_SUBCOMMAND.to_string(),
name.to_string(),
"--codex-home".to_string(),
codex_home.to_string_lossy().into_owned(),
],
env: None,
env_vars: Vec::new(),
cwd: None,
},
experimental_environment: None,
enabled: true,
required: false,
supports_parallel_tool_calls: true,
disabled_reason: None,
startup_timeout_sec: None,
tool_timeout_sec: None,
default_tools_approval_mode: None,
enabled_tools: None,
disabled_tools: None,
scopes: None,
oauth_resource: None,
tools: HashMap::new(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn configured_builtin_mcp_servers_adds_memories_when_enabled() {
let codex_home = AbsolutePathBuf::try_from("/tmp/codex-home").expect("absolute codex home");
let servers = configured_builtin_mcp_servers(BuiltinMcpServerOptions {
codex_self_exe: Some(Path::new("/tmp/codex")),
codex_home: codex_home.as_path(),
memories_enabled: true,
});
let server = servers
.get(MEMORIES_MCP_SERVER_NAME)
.expect("memories server should exist");
fn enabled_builtin_mcp_servers_adds_memories_when_enabled() {
assert_eq!(
server.transport,
McpServerTransportConfig::Stdio {
command: "/tmp/codex".to_string(),
args: vec![
"builtin-mcp".to_string(),
"memories".to_string(),
"--codex-home".to_string(),
codex_home.as_path().to_string_lossy().into_owned(),
],
env: None,
env_vars: Vec::new(),
cwd: None,
}
enabled_builtin_mcp_servers(BuiltinMcpServerOptions {
memories_enabled: true,
}),
vec![BuiltinMcpServer::Memories]
);
}
#[test]
fn configured_builtin_mcp_servers_requires_reexec_path() {
let codex_home = AbsolutePathBuf::try_from("/tmp/codex-home").expect("absolute codex home");
let servers = configured_builtin_mcp_servers(BuiltinMcpServerOptions {
codex_self_exe: None,
codex_home: codex_home.as_path(),
memories_enabled: true,
});
assert!(servers.is_empty());
fn enabled_builtin_mcp_servers_omits_memories_when_disabled() {
assert_eq!(
enabled_builtin_mcp_servers(BuiltinMcpServerOptions {
memories_enabled: false,
}),
Vec::<BuiltinMcpServer>::new()
);
}
}
codex-rs/cli/Cargo.toml
-1
View File
@@ -24,7 +24,6 @@ codex-app-server = { workspace = true }
codex-app-server-protocol = { workspace = true }
codex-app-server-test-client = { workspace = true }
codex-arg0 = { workspace = true }
codex-builtin-mcps = { workspace = true }
codex-chatgpt = { workspace = true }
codex-cloud-tasks = { path = "../cloud-tasks" }
codex-utils-cli = { workspace = true }
codex-rs/cli/src/main.rs
-20
View File
@@ -123,10 +123,6 @@ enum Subcommand {
/// Start Codex as an MCP server (stdio).
McpServer,
/// Internal: start a Codex-shipped MCP server (stdio).
#[clap(hide = true, name = "builtin-mcp")]
BuiltinMcp(BuiltinMcpCommand),
/// [experimental] Run the app server or related tooling.
AppServer(AppServerCommand),
@@ -179,13 +175,6 @@ enum Subcommand {
Features(FeaturesCli),
}
#[derive(Debug, Args)]
struct BuiltinMcpCommand {
name: String,
#[arg(long)]
codex_home: PathBuf,
}
#[derive(Debug, Parser)]
#[command(bin_name = "codex plugin")]
struct PluginCli {
@@ -820,15 +809,6 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
)?;
codex_mcp_server::run_main(arg0_paths.clone(), root_config_overrides).await?;
}
Some(Subcommand::BuiltinMcp(command)) => {
reject_remote_mode_for_subcommand(
root_remote.as_deref(),
root_remote_auth_token_env.as_deref(),
"builtin-mcp",
)?;
let codex_home = AbsolutePathBuf::try_from(command.codex_home)?;
codex_builtin_mcps::run_builtin_mcp_server(&command.name, &codex_home).await?;
}
Some(Subcommand::Mcp(mut mcp_cli)) => {
reject_remote_mode_for_subcommand(
root_remote.as_deref(),
codex-rs/cli/src/mcp_cmd.rs
+6 -5
View File
@@ -397,7 +397,7 @@ async fn run_login(config_overrides: &CliConfigOverrides, login_args: LoginArgs)
let mcp_manager = McpManager::new(Arc::new(PluginsManager::new(
config.codex_home.to_path_buf(),
)));
let mcp_servers = mcp_manager.effective_servers(&config, /*auth*/ None).await;
let mcp_servers = mcp_manager.configured_servers(&config).await;
let LoginArgs { name, scopes } = login_args;
@@ -450,7 +450,7 @@ async fn run_logout(config_overrides: &CliConfigOverrides, logout_args: LogoutAr
let mcp_manager = McpManager::new(Arc::new(PluginsManager::new(
config.codex_home.to_path_buf(),
)));
let mcp_servers = mcp_manager.effective_servers(&config, /*auth*/ None).await;
let mcp_servers = mcp_manager.configured_servers(&config).await;
let LogoutArgs { name } = logout_args;
@@ -482,12 +482,13 @@ async fn run_list(config_overrides: &CliConfigOverrides, list_args: ListArgs) ->
let mcp_manager = McpManager::new(Arc::new(PluginsManager::new(
config.codex_home.to_path_buf(),
)));
let mcp_servers = mcp_manager.effective_servers(&config, /*auth*/ None).await;
let mcp_servers = mcp_manager.configured_servers(&config).await;
let effective_mcp_servers = mcp_manager.effective_servers(&config, /*auth*/ None).await;
let mut entries: Vec<_> = mcp_servers.iter().collect();
entries.sort_by(|(a, _), (b, _)| a.cmp(b));
let auth_statuses = compute_auth_statuses(
mcp_servers.iter(),
effective_mcp_servers.iter(),
config.mcp_oauth_credentials_store_mode,
/*auth*/ None,
)
@@ -737,7 +738,7 @@ async fn run_get(config_overrides: &CliConfigOverrides, get_args: GetArgs) -> Re
let mcp_manager = McpManager::new(Arc::new(PluginsManager::new(
config.codex_home.to_path_buf(),
)));
let mcp_servers = mcp_manager.effective_servers(&config, /*auth*/ None).await;
let mcp_servers = mcp_manager.configured_servers(&config).await;
let Some(server) = mcp_servers.get(&get_args.name) else {
bail!("No MCP server named '{name}' found.", name = get_args.name);
codex-rs/codex-mcp/Cargo.toml
+1 -1
View File
@@ -33,7 +33,7 @@ serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
sha1 = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
tokio = { workspace = true, features = ["io-util", "macros", "rt-multi-thread"] }
tokio-util = { workspace = true, features = ["rt"] }
tracing = { workspace = true }
url = { workspace = true }
codex-rs/codex-mcp/src/builtin.rs
+39
View File
@@ -0,0 +1,39 @@
use std::io;
use std::path::PathBuf;
use codex_builtin_mcps::BuiltinMcpServer;
use codex_rmcp_client::InProcessTransportFactory;
use futures::FutureExt;
use futures::future::BoxFuture;
#[derive(Clone)]
pub(crate) struct BuiltinMcpServerFactory {
server: BuiltinMcpServer,
codex_home: PathBuf,
}
impl BuiltinMcpServerFactory {
pub(crate) fn new(server: BuiltinMcpServer, codex_home: PathBuf) -> Self {
Self { server, codex_home }
}
}
impl InProcessTransportFactory for BuiltinMcpServerFactory {
fn open(&self) -> BoxFuture<'static, io::Result<tokio::io::DuplexStream>> {
let server = self.server;
let codex_home = self.codex_home.clone();
async move {
let (client_transport, server_transport) = tokio::io::duplex(64 * 1024);
tokio::spawn(async move {
if let Err(err) = server.serve(&codex_home, server_transport).await {
tracing::warn!(
server = server.name(),
"built-in MCP server exited: {err:#}"
);
}
});
Ok(client_transport)
}
.boxed()
}
}
codex-rs/codex-mcp/src/connection_manager.rs
+53 -33
View File
@@ -7,6 +7,7 @@
//! `codex-core`.
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
@@ -29,6 +30,8 @@ use crate::rmcp_client::StartupOutcomeError;
use crate::rmcp_client::list_tools_for_client_uncached;
use crate::runtime::McpRuntimeEnvironment;
use crate::runtime::emit_duration;
use crate::server::EffectiveMcpServer;
use crate::server::McpServerMetadata;
use crate::tools::ToolInfo;
use crate::tools::filter_tools;
use crate::tools::qualify_tools;
@@ -38,7 +41,6 @@ use anyhow::Result;
use anyhow::anyhow;
use async_channel::Sender;
use codex_config::Constrained;
use codex_config::McpServerConfig;
use codex_config::McpServerTransportConfig;
use codex_config::types::OAuthCredentialsStoreMode;
use codex_login::CodexAuth;
@@ -65,12 +67,11 @@ use tokio::task::JoinSet;
use tokio_util::sync::CancellationToken;
use tracing::instrument;
use tracing::warn;
use url::Url;
/// A thin wrapper around a set of running [`RmcpClient`] instances.
pub struct McpConnectionManager {
clients: HashMap<String, AsyncManagedClient>,
server_origins: HashMap<String, String>,
server_metadata: HashMap<String, McpServerMetadata>,
host_owned_codex_apps_enabled: bool,
elicitation_requests: ElicitationRequestManager,
startup_cancellation_token: CancellationToken,
@@ -83,7 +84,7 @@ impl McpConnectionManager {
) -> Self {
Self {
clients: HashMap::new(),
server_origins: HashMap::new(),
server_metadata: HashMap::new(),
host_owned_codex_apps_enabled: false,
elicitation_requests: ElicitationRequestManager::new(
approval_policy.value(),
@@ -103,7 +104,7 @@ impl McpConnectionManager {
pub fn begin_shutdown(&mut self) -> impl std::future::Future<Output = ()> + Send + 'static {
self.startup_cancellation_token.cancel();
let clients = std::mem::take(&mut self.clients);
self.server_origins.clear();
self.server_metadata.clear();
async move {
for client in clients.into_values() {
client.shutdown().await;
@@ -117,7 +118,27 @@ impl McpConnectionManager {
}
pub fn server_origin(&self, server_name: &str) -> Option<&str> {
self.server_origins.get(server_name).map(String::as_str)
self.server_metadata
.get(server_name)
.and_then(|metadata| metadata.origin.as_ref())
.map(super::server::McpServerOrigin::as_str)
}
pub fn server_pollutes_memory(&self, server_name: &str) -> bool {
self.server_metadata
.get(server_name)
.is_none_or(|metadata| metadata.pollutes_memory)
}
pub fn parallel_tool_call_server_names(&self) -> HashSet<String> {
self.server_metadata
.iter()
.filter_map(|(name, metadata)| {
metadata
.supports_parallel_tool_calls
.then_some(name.clone())
})
.collect()
}
pub fn is_host_owned_codex_apps_server(&self, server_name: &str) -> bool {
@@ -146,7 +167,7 @@ impl McpConnectionManager {
#[allow(clippy::new_ret_no_self, clippy::too_many_arguments)]
pub async fn new(
mcp_servers: &HashMap<String, McpServerConfig>,
mcp_servers: &HashMap<String, EffectiveMcpServer>,
store_mode: OAuthCredentialsStoreMode,
auth_entries: HashMap<String, McpAuthStatusEntry>,
approval_policy: &Constrained<AskForApproval>,
@@ -163,7 +184,7 @@ impl McpConnectionManager {
) -> (Self, CancellationToken) {
let cancel_token = CancellationToken::new();
let mut clients = HashMap::new();
let mut server_origins = HashMap::new();
let mut server_metadata = HashMap::new();
let mut join_set = JoinSet::new();
let elicitation_requests = ElicitationRequestManager::new(
approval_policy.value(),
@@ -176,10 +197,11 @@ impl McpConnectionManager {
.filter(|auth| auth.uses_codex_backend())
.map(codex_model_provider::auth_provider_from_auth);
let mcp_servers = mcp_servers.clone();
for (server_name, cfg) in mcp_servers.into_iter().filter(|(_, cfg)| cfg.enabled) {
if let Some(origin) = transport_origin(&cfg.transport) {
server_origins.insert(server_name.clone(), origin);
}
for (server_name, server) in mcp_servers
.into_iter()
.filter(|(_, server)| server.enabled())
{
server_metadata.insert(server_name.clone(), McpServerMetadata::from(&server));
let cancel_token = cancel_token.child_token();
let _ = emit_update(
startup_submit_id.as_str(),
@@ -198,13 +220,16 @@ impl McpConnectionManager {
} else {
None
};
let uses_env_bearer_token = match &cfg.transport {
McpServerTransportConfig::StreamableHttp {
bearer_token_env_var,
..
} => bearer_token_env_var.is_some(),
McpServerTransportConfig::Stdio { .. } => false,
};
let uses_env_bearer_token =
server
.configured_config()
.is_some_and(|config| match &config.transport {
McpServerTransportConfig::StreamableHttp {
bearer_token_env_var,
..
} => bearer_token_env_var.is_some(),
McpServerTransportConfig::Stdio { .. } => false,
});
let runtime_auth_provider =
if server_name == CODEX_APPS_MCP_SERVER_NAME && !uses_env_bearer_token {
codex_apps_auth_provider.clone()
@@ -213,7 +238,7 @@ impl McpConnectionManager {
};
let async_managed_client = AsyncManagedClient::new(
server_name.clone(),
cfg,
server,
store_mode,
cancel_token.clone(),
tx_event.clone(),
@@ -221,6 +246,7 @@ impl McpConnectionManager {
codex_apps_tools_cache_context,
Arc::clone(&tool_plugin_provenance),
runtime_environment.clone(),
codex_home.clone(),
runtime_auth_provider,
);
clients.insert(server_name.clone(), async_managed_client.clone());
@@ -260,7 +286,7 @@ impl McpConnectionManager {
}
let manager = Self {
clients,
server_origins,
server_metadata,
host_owned_codex_apps_enabled,
elicitation_requests: elicitation_requests.clone(),
startup_cancellation_token: cancel_token.clone(),
@@ -672,16 +698,6 @@ async fn emit_update(
.await
}
fn transport_origin(transport: &McpServerTransportConfig) -> Option<String> {
match transport {
McpServerTransportConfig::StreamableHttp { url, .. } => {
let parsed = Url::parse(url).ok()?;
Some(parsed.origin().ascii_serialization())
}
McpServerTransportConfig::Stdio { .. } => Some("stdio".to_string()),
}
}
fn mcp_init_error_display(
server_name: &str,
entry: Option<&McpAuthStatusEntry>,
@@ -692,7 +708,7 @@ fn mcp_init_error_display(
bearer_token_env_var,
http_headers,
..
}) = &entry.map(|entry| &entry.config.transport)
}) = entry.and_then(|entry| entry.config.as_ref().map(|config| &config.transport))
&& url == "https://api.githubcopilot.com/mcp/"
&& bearer_token_env_var.is_none()
&& http_headers.as_ref().map(HashMap::is_empty).unwrap_or(true)
@@ -706,7 +722,11 @@ fn mcp_init_error_display(
)
} else if is_mcp_client_startup_timeout_error(err) {
let startup_timeout_secs = match entry {
Some(entry) => match entry.config.startup_timeout_sec {
Some(entry) => match entry
.config
.as_ref()
.and_then(|config| config.startup_timeout_sec)
{
Some(timeout) => timeout,
None => DEFAULT_STARTUP_TIMEOUT,
},
codex-rs/codex-mcp/src/connection_manager_tests.rs
+5 -32
View File
@@ -17,6 +17,7 @@ use crate::tools::filter_tools;
use crate::tools::qualify_tools;
use crate::tools::tool_with_model_visible_input_schema;
use codex_config::Constrained;
use codex_config::McpServerConfig;
use codex_protocol::ToolName;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::GranularApprovalConfig;
@@ -820,7 +821,7 @@ fn elicitation_capability_uses_2025_06_18_shape_for_all_servers() {
fn mcp_init_error_display_prompts_for_github_pat() {
let server_name = "github";
let entry = McpAuthStatusEntry {
config: McpServerConfig {
config: Some(McpServerConfig {
transport: McpServerTransportConfig::StreamableHttp {
url: "https://api.githubcopilot.com/mcp/".to_string(),
bearer_token_env_var: None,
@@ -840,7 +841,7 @@ fn mcp_init_error_display_prompts_for_github_pat() {
scopes: None,
oauth_resource: None,
tools: HashMap::new(),
},
}),
auth_status: McpAuthStatus::Unsupported,
};
let err: StartupOutcomeError = anyhow::anyhow!("OAuth is unsupported").into();
@@ -872,7 +873,7 @@ fn mcp_init_error_display_prompts_for_login_when_auth_required() {
fn mcp_init_error_display_reports_generic_errors() {
let server_name = "custom";
let entry = McpAuthStatusEntry {
config: McpServerConfig {
config: Some(McpServerConfig {
transport: McpServerTransportConfig::StreamableHttp {
url: "https://example.com".to_string(),
bearer_token_env_var: Some("TOKEN".to_string()),
@@ -892,7 +893,7 @@ fn mcp_init_error_display_reports_generic_errors() {
scopes: None,
oauth_resource: None,
tools: HashMap::new(),
},
}),
auth_status: McpAuthStatus::Unsupported,
};
let err: StartupOutcomeError = anyhow::anyhow!("boom").into();
@@ -916,31 +917,3 @@ fn mcp_init_error_display_includes_startup_timeout_hint() {
display
);
}
#[test]
fn transport_origin_extracts_http_origin() {
let transport = McpServerTransportConfig::StreamableHttp {
url: "https://example.com:8443/path?query=1".to_string(),
bearer_token_env_var: None,
http_headers: None,
env_http_headers: None,
};
assert_eq!(
transport_origin(&transport),
Some("https://example.com:8443".to_string())
);
}
#[test]
fn transport_origin_is_stdio_for_stdio_transport() {
let transport = McpServerTransportConfig::Stdio {
command: "server".to_string(),
args: Vec::new(),
env: None,
env_vars: Vec::new(),
cwd: None,
};
assert_eq!(transport_origin(&transport), Some("stdio".to_string()));
}
codex-rs/codex-mcp/src/lib.rs
+6 -1
View File
@@ -10,6 +10,7 @@ pub use tools::ToolInfo;
pub use mcp::CODEX_APPS_MCP_SERVER_NAME;
pub use mcp::McpConfig;
pub use mcp::ToolPluginProvenance;
pub use server::EffectiveMcpServer;
pub use auth_elicitation::CodexAppsAuthElicitation;
pub use auth_elicitation::CodexAppsAuthElicitationPlan;
@@ -22,12 +23,14 @@ pub use auth_elicitation::build_auth_elicitation_plan;
pub use auth_elicitation::connector_auth_failure_from_tool_result;
pub use codex_apps::CodexAppsToolsCacheKey;
pub use codex_apps::codex_apps_tools_cache_key;
pub use codex_builtin_mcps::BuiltinMcpServer;
pub use codex_builtin_mcps::BuiltinMcpServerOptions;
pub use codex_builtin_mcps::MEMORIES_MCP_SERVER_NAME;
pub use codex_builtin_mcps::configured_builtin_mcp_servers;
pub use codex_builtin_mcps::enabled_builtin_mcp_servers;
pub use mcp::configured_mcp_servers;
pub use mcp::effective_mcp_servers;
pub use mcp::effective_mcp_servers_from_configured;
pub use mcp::host_owned_codex_apps_enabled;
pub use mcp::tool_plugin_provenance;
pub use mcp::with_codex_apps_mcp;
@@ -55,10 +58,12 @@ pub use mcp::qualified_mcp_tool_name_prefix;
pub use tools::declared_openai_file_input_param_names;
pub(crate) mod auth_elicitation;
pub(crate) mod builtin;
pub(crate) mod codex_apps;
pub(crate) mod connection_manager;
pub(crate) mod elicitation;
pub(crate) mod mcp;
pub(crate) mod rmcp_client;
pub(crate) mod runtime;
pub(crate) mod server;
pub(crate) mod tools;
codex-rs/codex-mcp/src/mcp/auth.rs
+29 -19
View File
@@ -12,6 +12,8 @@ use codex_rmcp_client::discover_streamable_http_oauth;
use futures::future::join_all;
use tracing::warn;
use crate::server::EffectiveMcpServer;
use super::CODEX_APPS_MCP_SERVER_NAME;
#[derive(Debug, Clone)]
@@ -45,7 +47,7 @@ pub struct ResolvedMcpOAuthScopes {
#[derive(Debug, Clone)]
pub struct McpAuthStatusEntry {
pub config: McpServerConfig,
pub config: Option<McpServerConfig>,
pub auth_status: McpAuthStatus,
}
@@ -131,29 +133,37 @@ pub async fn compute_auth_statuses<'a, I>(
auth: Option<&CodexAuth>,
) -> HashMap<String, McpAuthStatusEntry>
where
I: IntoIterator<Item = (&'a String, &'a McpServerConfig)>,
I: IntoIterator<Item = (&'a String, &'a EffectiveMcpServer)>,
{
let futures = servers.into_iter().map(|(name, config)| {
let futures = servers.into_iter().map(|(name, server)| {
let name = name.clone();
let config = config.clone();
let config = server.configured_config().cloned();
let has_runtime_auth = name == CODEX_APPS_MCP_SERVER_NAME
&& auth.is_some_and(CodexAuth::uses_codex_backend)
&& matches!(
&config.transport,
McpServerTransportConfig::StreamableHttp {
bearer_token_env_var: None,
..
}
);
async move {
let auth_status =
match compute_auth_status(&name, &config, store_mode, has_runtime_auth).await {
Ok(status) => status,
Err(error) => {
warn!("failed to determine auth status for MCP server `{name}`: {error:?}");
McpAuthStatus::Unsupported
&& config.as_ref().is_some_and(|config| {
matches!(
&config.transport,
McpServerTransportConfig::StreamableHttp {
bearer_token_env_var: None,
..
}
};
)
});
async move {
let auth_status = match config.as_ref() {
Some(config) => {
match compute_auth_status(&name, config, store_mode, has_runtime_auth).await {
Ok(status) => status,
Err(error) => {
warn!(
"failed to determine auth status for MCP server `{name}`: {error:?}"
);
McpAuthStatus::Unsupported
}
}
}
None => McpAuthStatus::Unsupported,
};
let entry = McpAuthStatusEntry {
config,
auth_status,
codex-rs/codex-mcp/src/mcp/mod.rs
+29 -10
View File
@@ -38,6 +38,7 @@ use serde_json::Value;
use crate::codex_apps::codex_apps_tools_cache_key;
use crate::connection_manager::McpConnectionManager;
use crate::runtime::McpRuntimeEnvironment;
use crate::server::EffectiveMcpServer;
pub const CODEX_APPS_MCP_SERVER_NAME: &str = "codex_apps";
const MCP_TOOL_NAME_PREFIX: &str = "mcp";
@@ -99,8 +100,8 @@ pub struct McpPermissionPromptAutoApproveContext {
/// approval/sandbox policy, locate OAuth state, and merge plugin-provided MCP
/// servers. Request-scoped or auth-scoped state should not be stored here;
/// thread those values explicitly into runtime entry points such as
/// [`with_codex_apps_mcp`] and snapshot collection helpers so config objects do
/// not go stale when auth changes.
/// [`effective_mcp_servers`] and snapshot collection helpers so config objects
/// do not go stale when auth changes.
#[derive(Debug, Clone)]
pub struct McpConfig {
/// Base URL for ChatGPT-hosted app MCP servers, copied from the root config.
@@ -128,12 +129,13 @@ pub struct McpConfig {
/// ChatGPT auth is checked separately at runtime before the built-in apps
/// MCP server is added.
pub apps_enabled: bool,
/// Configured MCP servers keyed by server name.
/// Config-backed MCP servers keyed by server name.
///
/// This includes product-owned built-ins, user-configured servers, and
/// plugin-provided servers. Runtime-only additions belong in
/// Product-owned built-ins and runtime-only additions are merged later by
/// [`effective_mcp_servers`].
pub configured_mcp_servers: HashMap<String, McpServerConfig>,
/// Product-owned built-ins enabled for this runtime config.
pub builtin_mcp_servers: Vec<codex_builtin_mcps::BuiltinMcpServer>,
/// Plugin metadata used to attribute MCP tools/connectors to plugin display names.
pub plugin_capability_summaries: Vec<PluginCapabilitySummary>,
}
@@ -197,14 +199,14 @@ impl ToolPluginProvenance {
}
pub fn with_codex_apps_mcp(
mut servers: HashMap<String, McpServerConfig>,
mut servers: HashMap<String, EffectiveMcpServer>,
auth: Option<&CodexAuth>,
config: &McpConfig,
) -> HashMap<String, McpServerConfig> {
) -> HashMap<String, EffectiveMcpServer> {
if host_owned_codex_apps_enabled(config, auth) {
servers.insert(
CODEX_APPS_MCP_SERVER_NAME.to_string(),
codex_apps_mcp_server_config(config),
EffectiveMcpServer::configured(codex_apps_mcp_server_config(config)),
);
} else {
servers.remove(CODEX_APPS_MCP_SERVER_NAME);
@@ -223,8 +225,25 @@ pub fn configured_mcp_servers(config: &McpConfig) -> HashMap<String, McpServerCo
pub fn effective_mcp_servers(
config: &McpConfig,
auth: Option<&CodexAuth>,
) -> HashMap<String, McpServerConfig> {
let servers = configured_mcp_servers(config);
) -> HashMap<String, EffectiveMcpServer> {
effective_mcp_servers_from_configured(configured_mcp_servers(config), config, auth)
}
pub fn effective_mcp_servers_from_configured(
configured_servers: HashMap<String, McpServerConfig>,
config: &McpConfig,
auth: Option<&CodexAuth>,
) -> HashMap<String, EffectiveMcpServer> {
let mut servers = configured_servers
.into_iter()
.map(|(name, server)| (name, EffectiveMcpServer::configured(server)))
.collect::<HashMap<_, _>>();
for builtin_server in &config.builtin_mcp_servers {
servers.insert(
builtin_server.name().to_string(),
EffectiveMcpServer::builtin(*builtin_server),
);
}
with_codex_apps_mcp(servers, auth, config)
}
codex-rs/codex-mcp/src/mcp/mod_tests.rs
+70 -2
View File
@@ -13,6 +13,7 @@ use codex_protocol::protocol::GranularApprovalConfig;
use pretty_assertions::assert_eq;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
fn test_mcp_config(codex_home: PathBuf) -> McpConfig {
McpConfig {
@@ -28,6 +29,7 @@ fn test_mcp_config(codex_home: PathBuf) -> McpConfig {
use_legacy_landlock: false,
apps_enabled: false,
configured_mcp_servers: HashMap::new(),
builtin_mcp_servers: Vec::new(),
plugin_capability_summaries: Vec::new(),
}
}
@@ -217,7 +219,10 @@ fn codex_apps_server_config_uses_legacy_codex_apps_path() {
let server = servers
.get(CODEX_APPS_MCP_SERVER_NAME)
.expect("codex apps should be present when apps is enabled");
let url = match &server.transport {
let config = server
.configured_config()
.expect("codex apps should use configured transport");
let url = match &config.transport {
McpServerTransportConfig::StreamableHttp { url, .. } => url,
_ => panic!("expected streamable http transport for codex apps"),
};
@@ -236,7 +241,10 @@ fn codex_apps_server_config_uses_configured_apps_mcp_path_override() {
let server = servers
.get(CODEX_APPS_MCP_SERVER_NAME)
.expect("codex apps should be present when apps is enabled");
let url = match &server.transport {
let config = server
.configured_config()
.expect("codex apps should use configured transport");
let url = match &config.transport {
McpServerTransportConfig::StreamableHttp { url, .. } => url,
_ => panic!("expected streamable http transport for codex apps"),
};
@@ -310,6 +318,16 @@ async fn effective_mcp_servers_preserve_user_servers_and_add_codex_apps() {
.get(CODEX_APPS_MCP_SERVER_NAME)
.expect("codex apps server should exist");
let sample = sample
.configured_config()
.expect("configured server should retain transport");
let docs = docs
.configured_config()
.expect("configured server should retain transport");
let codex_apps = codex_apps
.configured_config()
.expect("codex apps should use configured transport");
match &sample.transport {
McpServerTransportConfig::StreamableHttp { url, .. } => {
assert_eq!(url, "https://user.example/mcp");
@@ -329,3 +347,53 @@ async fn effective_mcp_servers_preserve_user_servers_and_add_codex_apps() {
other => panic!("expected streamable http transport, got {other:?}"),
}
}
#[test]
fn effective_mcp_servers_preserve_builtin_runtime_shape() {
let mut config = test_mcp_config(PathBuf::from("/tmp"));
config.builtin_mcp_servers = vec![codex_builtin_mcps::BuiltinMcpServer::Memories];
let effective = effective_mcp_servers(&config, /*auth*/ None);
let memories = effective
.get(codex_builtin_mcps::MEMORIES_MCP_SERVER_NAME)
.expect("memories server should exist");
assert!(!crate::server::McpServerMetadata::from(memories).pollutes_memory);
assert!(matches!(
memories.launch(),
crate::server::McpServerLaunch::Builtin(codex_builtin_mcps::BuiltinMcpServer::Memories)
));
}
#[tokio::test]
async fn builtin_memories_server_runs_in_process() {
let codex_home = tempfile::tempdir().expect("tempdir");
let mut config = test_mcp_config(codex_home.path().to_path_buf());
config.builtin_mcp_servers = vec![codex_builtin_mcps::BuiltinMcpServer::Memories];
let snapshot = collect_mcp_server_status_snapshot_with_detail(
&config,
/*auth*/ None,
"builtin-memories-test".to_string(),
McpRuntimeEnvironment::new(
Arc::new(codex_exec_server::Environment::default_for_tests()),
codex_home.path().to_path_buf(),
),
McpSnapshotDetail::ToolsAndAuthOnly,
)
.await;
let tools = snapshot
.tools_by_server
.get(codex_builtin_mcps::MEMORIES_MCP_SERVER_NAME)
.expect("memories tools should be listed");
assert_eq!(
tools
.keys()
.cloned()
.collect::<std::collections::BTreeSet<_>>(),
["list".to_string(), "read".to_string(), "search".to_string()]
.into_iter()
.collect()
);
}
codex-rs/codex-mcp/src/rmcp_client.rs
+32 -7
View File
@@ -10,12 +10,14 @@ use std::borrow::Cow;
use std::collections::HashMap;
use std::env;
use std::ffi::OsString;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use crate::builtin::BuiltinMcpServerFactory;
use crate::codex_apps::CachedCodexAppsToolsLoad;
use crate::codex_apps::CodexAppsToolsCacheContext;
use crate::codex_apps::filter_disallowed_codex_apps_tools;
@@ -30,6 +32,8 @@ use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::mcp::ToolPluginProvenance;
use crate::runtime::McpRuntimeEnvironment;
use crate::runtime::emit_duration;
use crate::server::EffectiveMcpServer;
use crate::server::McpServerLaunch;
use crate::tools::ToolFilter;
use crate::tools::ToolInfo;
use crate::tools::filter_tools;
@@ -47,6 +51,7 @@ use codex_exec_server::HttpClient;
use codex_exec_server::ReqwestHttpClient;
use codex_protocol::protocol::Event;
use codex_rmcp_client::ExecutorStdioServerLauncher;
use codex_rmcp_client::InProcessTransportFactory;
use codex_rmcp_client::LocalStdioServerLauncher;
use codex_rmcp_client::RmcpClient;
use codex_rmcp_client::StdioServerLauncher;
@@ -133,7 +138,7 @@ impl AsyncManagedClient {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
server_name: String,
config: McpServerConfig,
server: EffectiveMcpServer,
store_mode: OAuthCredentialsStoreMode,
cancel_token: CancellationToken,
tx_event: Sender<Event>,
@@ -141,9 +146,13 @@ impl AsyncManagedClient {
codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
tool_plugin_provenance: Arc<ToolPluginProvenance>,
runtime_environment: McpRuntimeEnvironment,
codex_home: PathBuf,
runtime_auth_provider: Option<SharedAuthProvider>,
) -> Self {
let tool_filter = ToolFilter::from_config(&config);
let tool_filter = server
.configured_config()
.map(ToolFilter::from_config)
.unwrap_or_default();
let startup_snapshot = load_startup_cached_codex_apps_tools_snapshot(
&server_name,
codex_apps_tools_cache_context.as_ref(),
@@ -162,9 +171,10 @@ impl AsyncManagedClient {
let client = Arc::new(
make_rmcp_client(
&server_name,
config.clone(),
server.clone(),
store_mode,
runtime_environment,
codex_home,
runtime_auth_provider,
)
.await?,
@@ -173,10 +183,14 @@ impl AsyncManagedClient {
server_name,
client,
StartServerTaskParams {
startup_timeout: config
.startup_timeout_sec
startup_timeout: server
.configured_config()
.and_then(|config| config.startup_timeout_sec)
.or(Some(DEFAULT_STARTUP_TIMEOUT)),
tool_timeout: config.tool_timeout_sec.unwrap_or(DEFAULT_TOOL_TIMEOUT),
tool_timeout: server
.configured_config()
.and_then(|config| config.tool_timeout_sec)
.unwrap_or(DEFAULT_TOOL_TIMEOUT),
tool_filter: startup_tool_filter,
tx_event,
elicitation_requests,
@@ -552,11 +566,22 @@ struct StartServerTaskParams {
async fn make_rmcp_client(
server_name: &str,
config: McpServerConfig,
server: EffectiveMcpServer,
store_mode: OAuthCredentialsStoreMode,
runtime_environment: McpRuntimeEnvironment,
codex_home: PathBuf,
runtime_auth_provider: Option<SharedAuthProvider>,
) -> Result<RmcpClient, StartupOutcomeError> {
let config = match server.launch() {
McpServerLaunch::Configured(config) => config.as_ref().clone(),
McpServerLaunch::Builtin(builtin_server) => {
let factory: Arc<dyn InProcessTransportFactory> =
Arc::new(BuiltinMcpServerFactory::new(*builtin_server, codex_home));
return RmcpClient::new_in_process_client(factory)
.await
.map_err(|err| StartupOutcomeError::from(anyhow!(err)));
}
};
let McpServerConfig {
transport,
experimental_environment,
codex-rs/codex-mcp/src/server.rs
+108
View File
@@ -0,0 +1,108 @@
use codex_builtin_mcps::BuiltinMcpServer;
use codex_config::McpServerConfig;
use codex_config::McpServerTransportConfig;
/// The runtime launch strategy for an effective MCP server.
#[derive(Debug, Clone)]
pub(crate) enum McpServerLaunch {
Configured(Box<McpServerConfig>),
Builtin(BuiltinMcpServer),
}
/// MCP server after product-owned runtime additions have been applied.
#[derive(Debug, Clone)]
pub struct EffectiveMcpServer {
launch: McpServerLaunch,
}
impl EffectiveMcpServer {
pub fn configured(config: McpServerConfig) -> Self {
Self {
launch: McpServerLaunch::Configured(Box::new(config)),
}
}
pub fn builtin(server: BuiltinMcpServer) -> Self {
Self {
launch: McpServerLaunch::Builtin(server),
}
}
pub(crate) fn launch(&self) -> &McpServerLaunch {
&self.launch
}
pub fn configured_config(&self) -> Option<&McpServerConfig> {
match &self.launch {
McpServerLaunch::Configured(config) => Some(config.as_ref()),
McpServerLaunch::Builtin(_) => None,
}
}
pub fn enabled(&self) -> bool {
match &self.launch {
McpServerLaunch::Configured(config) => config.enabled,
McpServerLaunch::Builtin(_) => true,
}
}
pub fn required(&self) -> bool {
match &self.launch {
McpServerLaunch::Configured(config) => config.required,
McpServerLaunch::Builtin(_) => false,
}
}
}
/// Transport origin retained for metrics and diagnostics after server launch.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum McpServerOrigin {
InProcess,
Stdio,
StreamableHttp(String),
}
impl McpServerOrigin {
pub fn as_str(&self) -> &str {
match self {
Self::InProcess => "in_process",
Self::Stdio => "stdio",
Self::StreamableHttp(origin) => origin,
}
}
fn from_transport(transport: &McpServerTransportConfig) -> Option<Self> {
match transport {
McpServerTransportConfig::StreamableHttp { url, .. } => {
let parsed = url::Url::parse(url).ok()?;
Some(Self::StreamableHttp(parsed.origin().ascii_serialization()))
}
McpServerTransportConfig::Stdio { .. } => Some(Self::Stdio),
}
}
}
/// Semantic metadata that must survive after the server is launched.
#[derive(Debug, Clone)]
pub(crate) struct McpServerMetadata {
pub pollutes_memory: bool,
pub origin: Option<McpServerOrigin>,
pub supports_parallel_tool_calls: bool,
}
impl From<&EffectiveMcpServer> for McpServerMetadata {
fn from(server: &EffectiveMcpServer) -> Self {
match server.launch() {
McpServerLaunch::Configured(config) => Self {
pollutes_memory: true,
origin: McpServerOrigin::from_transport(&config.transport),
supports_parallel_tool_calls: config.supports_parallel_tool_calls,
},
McpServerLaunch::Builtin(server) => Self {
pollutes_memory: server.pollutes_memory(),
origin: Some(McpServerOrigin::InProcess),
supports_parallel_tool_calls: server.supports_parallel_tool_calls(),
},
}
}
}
codex-rs/core/src/config/config_tests.rs
+22 -50
View File
@@ -3360,7 +3360,6 @@ async fn to_mcp_config_empty_mcp_requirements_preserve_builtin_mcps() -> anyhow:
}))
.build()
.await?;
config.codex_self_exe = Some(PathBuf::from("/tmp/codex"));
let _ = config.features.enable(Feature::BuiltInMcp);
let _ = config.features.enable(Feature::MemoryTool);
let plugins_manager = PluginsManager::new(codex_home.path().to_path_buf());
@@ -3368,11 +3367,8 @@ async fn to_mcp_config_empty_mcp_requirements_preserve_builtin_mcps() -> anyhow:
let mcp_config = config.to_mcp_config(&plugins_manager).await;
assert_eq!(
mcp_config
.configured_mcp_servers
.get(codex_mcp::MEMORIES_MCP_SERVER_NAME)
.map(|server| (server.enabled, server.disabled_reason.clone())),
Some((true, None))
mcp_config.builtin_mcp_servers,
vec![codex_mcp::BuiltinMcpServer::Memories]
);
Ok(())
@@ -3399,7 +3395,6 @@ async fn to_mcp_config_nonempty_mcp_requirements_preserve_builtin_mcps() -> anyh
}))
.build()
.await?;
config.codex_self_exe = Some(PathBuf::from("/tmp/codex"));
let _ = config.features.enable(Feature::BuiltInMcp);
let _ = config.features.enable(Feature::MemoryTool);
let plugins_manager = PluginsManager::new(codex_home.path().to_path_buf());
@@ -3407,11 +3402,8 @@ async fn to_mcp_config_nonempty_mcp_requirements_preserve_builtin_mcps() -> anyh
let mcp_config = config.to_mcp_config(&plugins_manager).await;
assert_eq!(
mcp_config
.configured_mcp_servers
.get(codex_mcp::MEMORIES_MCP_SERVER_NAME)
.map(|server| (server.enabled, server.disabled_reason.clone())),
Some((true, None))
mcp_config.builtin_mcp_servers,
vec![codex_mcp::BuiltinMcpServer::Memories]
);
Ok(())
@@ -4244,10 +4236,7 @@ async fn to_mcp_config_includes_enabled_builtin_mcps() -> std::io::Result<()> {
let codex_home = TempDir::new()?;
let mut config = Config::load_from_base_config_with_overrides(
ConfigToml::default(),
ConfigOverrides {
codex_self_exe: Some(PathBuf::from("/tmp/codex")),
..ConfigOverrides::default()
},
ConfigOverrides::default(),
codex_home.abs(),
)
.await?;
@@ -4258,22 +4247,13 @@ async fn to_mcp_config_includes_enabled_builtin_mcps() -> std::io::Result<()> {
let mcp_config = config.to_mcp_config(&plugins_manager).await;
assert_eq!(
mcp_config
mcp_config.builtin_mcp_servers,
vec![codex_mcp::BuiltinMcpServer::Memories]
);
assert!(
!mcp_config
.configured_mcp_servers
.get(codex_mcp::MEMORIES_MCP_SERVER_NAME)
.map(|server| &server.transport),
Some(&McpServerTransportConfig::Stdio {
command: "/tmp/codex".to_string(),
args: vec![
"builtin-mcp".to_string(),
"memories".to_string(),
"--codex-home".to_string(),
codex_home.path().display().to_string(),
],
env: None,
env_vars: Vec::new(),
cwd: None,
})
.contains_key(codex_mcp::MEMORIES_MCP_SERVER_NAME)
);
Ok(())
@@ -4284,10 +4264,7 @@ async fn to_mcp_config_omits_builtin_mcps_when_feature_is_disabled() -> std::io:
let codex_home = TempDir::new()?;
let mut config = Config::load_from_base_config_with_overrides(
ConfigToml::default(),
ConfigOverrides {
codex_self_exe: Some(PathBuf::from("/tmp/codex")),
..ConfigOverrides::default()
},
ConfigOverrides::default(),
codex_home.abs(),
)
.await?;
@@ -4296,11 +4273,7 @@ async fn to_mcp_config_omits_builtin_mcps_when_feature_is_disabled() -> std::io:
let mcp_config = config.to_mcp_config(&plugins_manager).await;
assert!(
!mcp_config
.configured_mcp_servers
.contains_key(codex_mcp::MEMORIES_MCP_SERVER_NAME)
);
assert!(mcp_config.builtin_mcp_servers.is_empty());
Ok(())
}
@@ -4316,10 +4289,7 @@ async fn to_mcp_config_reserves_enabled_builtin_mcp_names() -> std::io::Result<(
)]),
..ConfigToml::default()
},
ConfigOverrides {
codex_self_exe: Some(PathBuf::from("/tmp/codex")),
..ConfigOverrides::default()
},
ConfigOverrides::default(),
codex_home.abs(),
)
.await?;
@@ -4329,13 +4299,15 @@ async fn to_mcp_config_reserves_enabled_builtin_mcp_names() -> std::io::Result<(
let mcp_config = config.to_mcp_config(&plugins_manager).await;
assert!(matches!(
mcp_config
assert_eq!(
mcp_config.builtin_mcp_servers,
vec![codex_mcp::BuiltinMcpServer::Memories]
);
assert!(
!mcp_config
.configured_mcp_servers
.get(codex_mcp::MEMORIES_MCP_SERVER_NAME)
.map(|server| &server.transport),
Some(McpServerTransportConfig::Stdio { .. })
));
.contains_key(codex_mcp::MEMORIES_MCP_SERVER_NAME)
);
Ok(())
}
codex-rs/core/src/config/mod.rs
+6 -5
View File
@@ -72,7 +72,7 @@ use codex_git_utils::resolve_root_git_project_for_trust;
use codex_login::AuthManagerConfig;
use codex_mcp::BuiltinMcpServerOptions;
use codex_mcp::McpConfig;
use codex_mcp::configured_builtin_mcp_servers;
use codex_mcp::enabled_builtin_mcp_servers;
use codex_memories_read::memory_root;
use codex_model_provider_info::LEGACY_OLLAMA_CHAT_PROVIDER_ID;
use codex_model_provider_info::ModelProviderInfo;
@@ -1091,9 +1091,7 @@ impl Config {
) -> McpConfig {
let plugins_input = self.plugins_config_input();
let loaded_plugins = plugins_manager.plugins_for_config(&plugins_input).await;
let builtin_mcp_servers = configured_builtin_mcp_servers(BuiltinMcpServerOptions {
codex_self_exe: self.codex_self_exe.as_deref(),
codex_home: self.codex_home.as_path(),
let builtin_mcp_servers = enabled_builtin_mcp_servers(BuiltinMcpServerOptions {
memories_enabled: self.features.enabled(Feature::BuiltInMcp)
&& self.features.enabled(Feature::MemoryTool)
&& self.memories.use_memories,
@@ -1122,7 +1120,9 @@ impl Config {
// allowlists.
filter_mcp_servers_by_requirements(&mut configured_mcp_servers, Some(mcp_requirements));
}
configured_mcp_servers.extend(builtin_mcp_servers);
for builtin_server in &builtin_mcp_servers {
configured_mcp_servers.remove(builtin_server.name());
}
McpConfig {
chatgpt_base_url: self.chatgpt_base_url.clone(),
@@ -1139,6 +1139,7 @@ impl Config {
use_legacy_landlock: self.features.use_legacy_landlock(),
apps_enabled: self.features.enabled(Feature::Apps),
configured_mcp_servers,
builtin_mcp_servers,
plugin_capability_summaries: loaded_plugins.capability_summaries().to_vec(),
}
}
codex-rs/core/src/connectors.rs
+2 -1
View File
@@ -321,7 +321,8 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_environment_manager(
true
} else if tools.is_empty() {
let timeout = cfg
.startup_timeout_sec
.configured_config()
.and_then(|config| config.startup_timeout_sec)
.unwrap_or(CONNECTORS_READY_TIMEOUT_ON_EMPTY_TOOLS);
let ready = mcp_connection_manager
.wait_for_server_ready(CODEX_APPS_MCP_SERVER_NAME, timeout)
codex-rs/core/src/mcp.rs
+2 -1
View File
@@ -5,6 +5,7 @@ use crate::config::Config;
use codex_config::McpServerConfig;
use codex_core_plugins::PluginsManager;
use codex_login::CodexAuth;
use codex_mcp::EffectiveMcpServer;
use codex_mcp::ToolPluginProvenance;
use codex_mcp::configured_mcp_servers;
use codex_mcp::effective_mcp_servers;
@@ -29,7 +30,7 @@ impl McpManager {
&self,
config: &Config,
auth: Option<&CodexAuth>,
) -> HashMap<String, McpServerConfig> {
) -> HashMap<String, EffectiveMcpServer> {
let mcp_config = config.to_mcp_config(self.plugins_manager.as_ref()).await;
effective_mcp_servers(&mcp_config, auth)
}
codex-rs/core/src/mcp_skill_dependencies.rs
+5 -8
View File
@@ -187,14 +187,11 @@ pub(crate) async fn maybe_install_mcp_dependencies(
}
}
// Refresh from the effective merged MCP map (global + repo + managed) and
// overlay the updated global servers so we don't drop repo-scoped servers.
let auth = sess.services.auth_manager.auth().await;
let mut refresh_servers = sess
.services
.mcp_manager
.effective_servers(config, auth.as_ref())
.await;
// Refresh from the config-backed merged MCP map (global + repo + managed)
// and overlay the updated global servers so we don't drop repo-scoped
// servers. Runtime additions such as built-ins are rebuilt by the refresh
// path from the current config.
let mut refresh_servers = sess.services.mcp_manager.configured_servers(config).await;
for (name, server_config) in &servers {
refresh_servers
.entry(name.clone())
codex-rs/core/src/mcp_tool_call.rs
+16 -3
View File
@@ -324,9 +324,8 @@ async fn handle_approved_mcp_tool_call(
request_meta: Option<JsonValue>,
mcp_app_resource_uri: Option<String>,
) -> HandledMcpToolCall {
maybe_mark_thread_memory_mode_polluted(sess, turn_context).await;
let server = invocation.server.clone();
maybe_mark_thread_memory_mode_polluted(sess, turn_context, &server).await;
let tool_name = invocation.tool.clone();
let arguments_value = invocation.arguments.clone();
let connector_id = metadata.and_then(|metadata| metadata.connector_id.as_deref());
@@ -465,6 +464,7 @@ fn mcp_tool_call_span(
) -> Span {
let transport = match fields.server_origin {
Some("stdio") => "stdio",
Some("in_process") => "in_process",
Some(_) => "streamable_http",
None => "",
};
@@ -752,10 +752,23 @@ async fn augment_mcp_tool_request_meta_with_sandbox_state(
Ok(meta)
}
async fn maybe_mark_thread_memory_mode_polluted(sess: &Session, turn_context: &TurnContext) {
async fn maybe_mark_thread_memory_mode_polluted(
sess: &Session,
turn_context: &TurnContext,
server: &str,
) {
if !turn_context.config.memories.disable_on_external_context {
return;
}
let pollutes_memory = sess
.services
.mcp_connection_manager
.read()
.await
.server_pollutes_memory(server);
if !pollutes_memory {
return;
}
state_db::mark_thread_memory_mode_polluted(
sess.services.state_db.as_deref(),
sess.conversation_id,
codex-rs/core/src/session/mcp.rs
+2 -1
View File
@@ -293,7 +293,8 @@ impl Session {
.mcp_manager
.tool_plugin_provenance(config.as_ref())
.await;
let mcp_servers = with_codex_apps_mcp(mcp_servers, auth.as_ref(), &mcp_config);
let mcp_servers =
effective_mcp_servers_from_configured(mcp_servers, &mcp_config, auth.as_ref());
let host_owned_codex_apps_enabled =
host_owned_codex_apps_enabled(&mcp_config, auth.as_ref());
let auth_statuses =
codex-rs/core/src/session/mod.rs
+1 -1
View File
@@ -307,8 +307,8 @@ use crate::windows_sandbox::WindowsSandboxLevelExt;
use codex_core_plugins::PluginsManager;
use codex_git_utils::get_git_repo_root;
use codex_mcp::compute_auth_statuses;
use codex_mcp::effective_mcp_servers_from_configured;
use codex_mcp::host_owned_codex_apps_enabled;
use codex_mcp::with_codex_apps_mcp;
use codex_otel::SessionTelemetry;
use codex_otel::THREAD_STARTED_METRIC;
use codex_otel::TelemetryAuthMode;
codex-rs/core/src/session/session.rs
+3 -2
View File
@@ -936,11 +936,12 @@ impl Session {
sess.start_skills_watcher_listener();
let mut required_mcp_servers: Vec<String> = mcp_servers
.iter()
.filter(|(_, server)| server.enabled && server.required)
.filter(|(_, server)| server.enabled() && server.required())
.map(|(name, _)| name.clone())
.collect();
required_mcp_servers.sort();
let enabled_mcp_server_count = mcp_servers.values().filter(|server| server.enabled).count();
let enabled_mcp_server_count =
mcp_servers.values().filter(|server| server.enabled()).count();
let required_mcp_server_count = required_mcp_servers.len();
let tool_plugin_provenance = mcp_manager.tool_plugin_provenance(config.as_ref()).await;
let host_owned_codex_apps_enabled = config
codex-rs/core/src/session/turn.rs
+1 -12
View File
@@ -1155,6 +1155,7 @@ pub(crate) async fn built_tools(
.list_all_tools()
.or_cancel(cancellation_token)
.await?;
let parallel_mcp_server_names = mcp_connection_manager.parallel_tool_call_server_names();
drop(mcp_connection_manager);
let loaded_plugins = sess
.services
@@ -1254,18 +1255,6 @@ pub(crate) async fn built_tools(
Vec::new()
};
let parallel_mcp_server_names = turn_context
.config
.mcp_servers
.get()
.iter()
.filter_map(|(server_name, server_config)| {
server_config
.supports_parallel_tool_calls
.then_some(server_name.clone())
})
.collect::<HashSet<_>>();
Ok(Arc::new(ToolRouter::from_config(
&turn_context.tools_config,
ToolRouterParams {
codex-rs/core/src/tools/code_mode/mod.rs
+3 -21
View File
@@ -4,7 +4,6 @@ mod response_adapter;
mod wait_handler;
pub(crate) mod wait_spec;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
@@ -274,26 +273,9 @@ pub(super) async fn build_enabled_tools(
)]
async fn build_nested_router(exec: &ExecContext) -> ToolRouter {
let nested_tools_config = exec.turn.tools_config.for_code_mode_nested_tools();
let listed_mcp_tools = exec
.session
.services
.mcp_connection_manager
.read()
.await
.list_all_tools()
.await;
let parallel_mcp_server_names = exec
.turn
.config
.mcp_servers
.get()
.iter()
.filter_map(|(server_name, server_config)| {
server_config
.supports_parallel_tool_calls
.then_some(server_name.clone())
})
.collect::<HashSet<_>>();
let mcp_connection_manager = exec.session.services.mcp_connection_manager.read().await;
let listed_mcp_tools = mcp_connection_manager.list_all_tools().await;
let parallel_mcp_server_names = mcp_connection_manager.parallel_tool_call_server_names();
ToolRouter::from_config(
&nested_tools_config,
codex-rs/core/tests/suite/sqlite_state.rs
+87
View File
@@ -2,6 +2,7 @@ use anyhow::Result;
use codex_config::types::McpServerConfig;
use codex_config::types::McpServerTransportConfig;
use codex_features::Feature;
use codex_mcp::MEMORIES_MCP_SERVER_NAME;
use codex_protocol::ThreadId;
use codex_protocol::dynamic_tools::DynamicToolSpec;
use codex_protocol::models::PermissionProfile;
@@ -447,6 +448,92 @@ async fn mcp_call_marks_thread_memory_mode_polluted_when_configured() -> Result<
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn builtin_memories_mcp_call_does_not_mark_thread_memory_mode_polluted_when_configured()
-> Result<()> {
let server = start_mock_server().await;
let call_id = "call-123";
let namespace = format!("mcp__{MEMORIES_MCP_SERVER_NAME}__");
mount_sse_once(
&server,
responses::sse(vec![
ev_response_created("resp-1"),
responses::ev_function_call_with_namespace(call_id, &namespace, "list", "{}"),
ev_completed("resp-1"),
]),
)
.await;
mount_sse_once(
&server,
responses::sse(vec![
responses::ev_assistant_message("msg-1", "memories list tool completed."),
ev_completed("resp-2"),
]),
)
.await;
let mut builder = test_codex().with_config(|config| {
config
.features
.enable(Feature::Sqlite)
.expect("test config should allow feature update");
config
.features
.enable(Feature::BuiltInMcp)
.expect("test config should allow feature update");
config
.features
.enable(Feature::MemoryTool)
.expect("test config should allow feature update");
config.memories.use_memories = true;
config.memories.disable_on_external_context = true;
});
let test = builder.build(&server).await?;
let db = test.codex.state_db().expect("state db enabled");
let thread_id = test.session_configured.thread_id;
let cwd = test.cwd_path().to_path_buf();
let (sandbox_policy, permission_profile) =
turn_permission_fields(PermissionProfile::read_only(), cwd.as_path());
test.codex
.submit(Op::UserTurn {
environments: None,
items: vec![UserInput::Text {
text: "call the memories list tool".to_string(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
cwd,
approval_policy: AskForApproval::Never,
approvals_reviewer: None,
sandbox_policy,
permission_profile,
model: test.session_configured.model.clone(),
effort: None,
summary: None,
service_tier: None,
collaboration_mode: None,
personality: None,
})
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::McpToolCallEnd(_))
})
.await;
wait_for_event_match(&test.codex, |event| match event {
EventMsg::Error(err) => Some(Err(anyhow::anyhow!(err.message.clone()))),
EventMsg::TurnComplete(_) => Some(Ok(())),
_ => None,
})
.await?;
assert_ne!(
db.get_thread_memory_mode(thread_id).await?.as_deref(),
Some("polluted")
);
Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn tool_call_logs_include_thread_id() -> Result<()> {
let server = start_mock_server().await;
codex-rs/memories/mcp/Cargo.toml
+1
View File
@@ -18,6 +18,7 @@ codex-utils-output-truncation = { workspace = true }
rmcp = { workspace = true, default-features = false, features = [
"schemars",
"server",
"transport-async-rw",
] }
schemars = { workspace = true }
serde = { workspace = true, features = ["derive"] }
codex-rs/memories/mcp/src/lib.rs
+1
View File
@@ -11,4 +11,5 @@ mod server;
pub use local::LocalMemoriesBackend;
pub use server::MemoriesMcpServer;
pub use server::run_server;
pub use server::run_stdio_server;
codex-rs/memories/mcp/src/server.rs
+10 -2
View File
@@ -185,19 +185,27 @@ impl<B: MemoriesBackend> ServerHandler for MemoriesMcpServer<B> {
}
}
pub async fn run_stdio_server(codex_home: &AbsolutePathBuf) -> anyhow::Result<()> {
pub async fn run_server<T, E, A>(codex_home: &AbsolutePathBuf, transport: T) -> anyhow::Result<()>
where
T: rmcp::transport::IntoTransport<rmcp::RoleServer, E, A>,
E: std::error::Error + Send + Sync + 'static,
{
let backend = LocalMemoriesBackend::from_codex_home(codex_home);
tokio::fs::create_dir_all(backend.root())
.await
.with_context(|| format!("create memories root at {}", backend.root().display()))?;
MemoriesMcpServer::new(backend)
.serve((tokio::io::stdin(), tokio::io::stdout()))
.serve(transport)
.await?
.waiting()
.await?;
Ok(())
}
pub async fn run_stdio_server(codex_home: &AbsolutePathBuf) -> anyhow::Result<()> {
run_server(codex_home, (tokio::io::stdin(), tokio::io::stdout())).await
}
fn list_tool() -> Tool {
let mut tool = Tool::new(
Cow::Borrowed(LIST_TOOL_NAME),
codex-rs/rmcp-client/Cargo.toml
+1
View File
@@ -37,6 +37,7 @@ rmcp = { workspace = true, default-features = false, features = [
"macros",
"schemars",
"server",
"transport-async-rw",
"transport-child-process",
"transport-streamable-http-client-reqwest",
"transport-streamable-http-server",
codex-rs/rmcp-client/src/in_process_transport.rs
+14
View File
@@ -0,0 +1,14 @@
use std::io;
use futures::future::BoxFuture;
use tokio::io::DuplexStream;
/// Recreates a fresh in-process MCP byte stream whenever the client needs one.
///
/// Implementations are expected to start the paired server side before
/// returning the client stream. The factory is retained by [`crate::RmcpClient`]
/// so reconnects can rebuild the transport without knowing which built-in
/// server produced it.
pub trait InProcessTransportFactory: Send + Sync {
fn open(&self) -> BoxFuture<'static, io::Result<DuplexStream>>;
}
codex-rs/rmcp-client/src/lib.rs
+2
View File
@@ -2,6 +2,7 @@ mod auth_status;
mod elicitation_client_service;
mod executor_process_transport;
mod http_client_adapter;
mod in_process_transport;
mod logging_client_handler;
mod oauth;
mod perform_oauth_login;
@@ -15,6 +16,7 @@ pub use auth_status::determine_streamable_http_auth_status;
pub use auth_status::discover_streamable_http_oauth;
pub use auth_status::supports_oauth_login;
pub use codex_protocol::protocol::McpAuthStatus;
pub use in_process_transport::InProcessTransportFactory;
pub use oauth::StoredOAuthTokens;
pub use oauth::WrappedOAuthTokenResponse;
pub use oauth::delete_oauth_tokens;
codex-rs/rmcp-client/src/rmcp_client.rs
+37 -1
View File
@@ -62,6 +62,7 @@ use tracing::warn;
use crate::elicitation_client_service::ElicitationClientService;
use crate::http_client_adapter::StreamableHttpClientAdapter;
use crate::http_client_adapter::StreamableHttpClientAdapterError;
use crate::in_process_transport::InProcessTransportFactory;
use crate::load_oauth_tokens;
use crate::oauth::OAuthPersistor;
use crate::oauth::StoredOAuthTokens;
@@ -74,6 +75,9 @@ use crate::utils::build_default_headers;
use codex_config::types::OAuthCredentialsStoreMode;
enum PendingTransport {
InProcess {
transport: tokio::io::DuplexStream,
},
Stdio {
transport: StdioServerTransport,
},
@@ -99,6 +103,9 @@ enum ClientState {
#[derive(Clone)]
enum TransportRecipe {
InProcess {
factory: Arc<dyn InProcessTransportFactory>,
},
Stdio {
command: StdioServerCommand,
launcher: Arc<dyn StdioServerLauncher>,
@@ -275,6 +282,26 @@ pub struct RmcpClient {
}
impl RmcpClient {
pub async fn new_in_process_client(
factory: Arc<dyn InProcessTransportFactory>,
) -> io::Result<Self> {
let transport_recipe = TransportRecipe::InProcess { factory };
let transport = Self::create_pending_transport(&transport_recipe)
.await
.map_err(io::Error::other)?;
Ok(Self {
state: Mutex::new(ClientState::Connecting {
transport: Some(transport),
}),
stdio_process: None,
transport_recipe,
initialize_context: Mutex::new(None),
session_recovery_lock: Semaphore::new(/*permits*/ 1),
elicitation_pause_state: ElicitationPauseState::new(),
})
}
pub async fn new_stdio_client(
program: OsString,
args: Vec<OsString>,
@@ -292,7 +319,8 @@ impl RmcpClient {
.map_err(io::Error::other)?;
let stdio_process = match &transport {
PendingTransport::Stdio { transport } => Some(transport.process_handle()),
PendingTransport::StreamableHttp { .. }
PendingTransport::InProcess { .. }
| PendingTransport::StreamableHttp { .. }
| PendingTransport::StreamableHttpWithOAuth { .. } => None,
};
@@ -690,6 +718,10 @@ impl RmcpClient {
transport_recipe: &TransportRecipe,
) -> Result<PendingTransport> {
match transport_recipe {
TransportRecipe::InProcess { factory } => {
let transport = factory.open().await?;
Ok(PendingTransport::InProcess { transport })
}
TransportRecipe::Stdio { command, launcher } => {
let transport = launcher.launch(command.clone()).await?;
Ok(PendingTransport::Stdio { transport })
@@ -798,6 +830,10 @@ impl RmcpClient {
Option<OAuthPersistor>,
)> {
let (transport, oauth_persistor) = match pending_transport {
PendingTransport::InProcess { transport } => (
service::serve_client(client_service, transport).boxed(),
None,
),
PendingTransport::Stdio { transport } => (
service::serve_client(client_service, transport).boxed(),
None,