From b2268999fe62818246690f90bb098ca35e96a297 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Thu, 7 May 2026 10:36:32 +0200 Subject: [PATCH] feat: make built-in MCPs first-class runtime servers (#21356) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- codex-rs/Cargo.lock | 3 +- codex-rs/builtin-mcps/Cargo.toml | 2 +- codex-rs/builtin-mcps/src/lib.rs | 175 +++++++----------- codex-rs/cli/Cargo.toml | 1 - codex-rs/cli/src/main.rs | 20 -- codex-rs/cli/src/mcp_cmd.rs | 11 +- codex-rs/codex-mcp/Cargo.toml | 2 +- codex-rs/codex-mcp/src/builtin.rs | 39 ++++ codex-rs/codex-mcp/src/connection_manager.rs | 86 +++++---- .../codex-mcp/src/connection_manager_tests.rs | 37 +--- codex-rs/codex-mcp/src/lib.rs | 7 +- codex-rs/codex-mcp/src/mcp/auth.rs | 48 +++-- codex-rs/codex-mcp/src/mcp/mod.rs | 39 +++- codex-rs/codex-mcp/src/mcp/mod_tests.rs | 72 ++++++- codex-rs/codex-mcp/src/rmcp_client.rs | 39 +++- codex-rs/codex-mcp/src/server.rs | 108 +++++++++++ codex-rs/core/src/config/config_tests.rs | 72 +++---- codex-rs/core/src/config/mod.rs | 11 +- codex-rs/core/src/connectors.rs | 3 +- codex-rs/core/src/mcp.rs | 3 +- codex-rs/core/src/mcp_skill_dependencies.rs | 13 +- codex-rs/core/src/mcp_tool_call.rs | 19 +- codex-rs/core/src/session/mcp.rs | 3 +- codex-rs/core/src/session/mod.rs | 2 +- codex-rs/core/src/session/session.rs | 5 +- codex-rs/core/src/session/turn.rs | 13 +- codex-rs/core/src/tools/code_mode/mod.rs | 24 +-- codex-rs/core/tests/suite/sqlite_state.rs | 87 +++++++++ codex-rs/memories/mcp/Cargo.toml | 1 + codex-rs/memories/mcp/src/lib.rs | 1 + codex-rs/memories/mcp/src/server.rs | 12 +- codex-rs/rmcp-client/Cargo.toml | 1 + .../rmcp-client/src/in_process_transport.rs | 14 ++ codex-rs/rmcp-client/src/lib.rs | 2 + codex-rs/rmcp-client/src/rmcp_client.rs | 38 +++- 35 files changed, 668 insertions(+), 345 deletions(-) create mode 100644 codex-rs/codex-mcp/src/builtin.rs create mode 100644 codex-rs/codex-mcp/src/server.rs create mode 100644 codex-rs/rmcp-client/src/in_process_transport.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 29178947b..b4ea56033 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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", diff --git a/codex-rs/builtin-mcps/Cargo.toml b/codex-rs/builtin-mcps/Cargo.toml index de64ec703..ee61f9675 100644 --- a/codex-rs/builtin-mcps/Cargo.toml +++ b/codex-rs/builtin-mcps/Cargo.toml @@ -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 } diff --git a/codex-rs/builtin-mcps/src/lib.rs b/codex-rs/builtin-mcps/src/lib.rs index abecc994d..cf5cb7488 100644 --- a/codex-rs/builtin-mcps/src/lib.rs +++ b/codex-rs/builtin-mcps/src/lib.rs @@ -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(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 { - 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 { + 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::::new() + ); } } diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index b644734f4..cdee241b4 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -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 } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 6a1d08e56..70cd0294a 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -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(), diff --git a/codex-rs/cli/src/mcp_cmd.rs b/codex-rs/cli/src/mcp_cmd.rs index 858ef442a..af7599916 100644 --- a/codex-rs/cli/src/mcp_cmd.rs +++ b/codex-rs/cli/src/mcp_cmd.rs @@ -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); diff --git a/codex-rs/codex-mcp/Cargo.toml b/codex-rs/codex-mcp/Cargo.toml index c47bd947e..e22257706 100644 --- a/codex-rs/codex-mcp/Cargo.toml +++ b/codex-rs/codex-mcp/Cargo.toml @@ -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 } diff --git a/codex-rs/codex-mcp/src/builtin.rs b/codex-rs/codex-mcp/src/builtin.rs new file mode 100644 index 000000000..9441b644d --- /dev/null +++ b/codex-rs/codex-mcp/src/builtin.rs @@ -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> { + 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() + } +} diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index f93ac5e08..f85d28d6d 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -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, - server_origins: HashMap, + server_metadata: HashMap, 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 + 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 { + 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, + mcp_servers: &HashMap, store_mode: OAuthCredentialsStoreMode, auth_entries: HashMap, approval_policy: &Constrained, @@ -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 { - 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, }, diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index f2dc85cde..2d370fcef 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -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())); -} diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index 83dc65dd0..c9ea32657 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -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; diff --git a/codex-rs/codex-mcp/src/mcp/auth.rs b/codex-rs/codex-mcp/src/mcp/auth.rs index 6a97b5278..12f832f9e 100644 --- a/codex-rs/codex-mcp/src/mcp/auth.rs +++ b/codex-rs/codex-mcp/src/mcp/auth.rs @@ -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, pub auth_status: McpAuthStatus, } @@ -131,29 +133,37 @@ pub async fn compute_auth_statuses<'a, I>( auth: Option<&CodexAuth>, ) -> HashMap where - I: IntoIterator, + I: IntoIterator, { - 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, diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index ae667c698..1be06e4ba 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -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, + /// Product-owned built-ins enabled for this runtime config. + pub builtin_mcp_servers: Vec, /// Plugin metadata used to attribute MCP tools/connectors to plugin display names. pub plugin_capability_summaries: Vec, } @@ -197,14 +199,14 @@ impl ToolPluginProvenance { } pub fn with_codex_apps_mcp( - mut servers: HashMap, + mut servers: HashMap, auth: Option<&CodexAuth>, config: &McpConfig, -) -> HashMap { +) -> HashMap { 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, -) -> HashMap { - let servers = configured_mcp_servers(config); +) -> HashMap { + effective_mcp_servers_from_configured(configured_mcp_servers(config), config, auth) +} + +pub fn effective_mcp_servers_from_configured( + configured_servers: HashMap, + config: &McpConfig, + auth: Option<&CodexAuth>, +) -> HashMap { + let mut servers = configured_servers + .into_iter() + .map(|(name, server)| (name, EffectiveMcpServer::configured(server))) + .collect::>(); + 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) } diff --git a/codex-rs/codex-mcp/src/mcp/mod_tests.rs b/codex-rs/codex-mcp/src/mcp/mod_tests.rs index 94ddd43b3..491341c3e 100644 --- a/codex-rs/codex-mcp/src/mcp/mod_tests.rs +++ b/codex-rs/codex-mcp/src/mcp/mod_tests.rs @@ -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::>(), + ["list".to_string(), "read".to_string(), "search".to_string()] + .into_iter() + .collect() + ); +} diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index c6a781bb8..c9a8ca8c3 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -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, @@ -141,9 +146,13 @@ impl AsyncManagedClient { codex_apps_tools_cache_context: Option, tool_plugin_provenance: Arc, runtime_environment: McpRuntimeEnvironment, + codex_home: PathBuf, runtime_auth_provider: Option, ) -> 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, ) -> Result { + let config = match server.launch() { + McpServerLaunch::Configured(config) => config.as_ref().clone(), + McpServerLaunch::Builtin(builtin_server) => { + let factory: Arc = + 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, diff --git a/codex-rs/codex-mcp/src/server.rs b/codex-rs/codex-mcp/src/server.rs new file mode 100644 index 000000000..a57aceccb --- /dev/null +++ b/codex-rs/codex-mcp/src/server.rs @@ -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), + 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 { + 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, + 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(), + }, + } + } +} diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 9b785ff33..bc9af28a6 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -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(()) } diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index ed355a0ae..b2dfbcb2b 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -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(), } } diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index 0bd53a50e..a2ec6e93c 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -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) diff --git a/codex-rs/core/src/mcp.rs b/codex-rs/core/src/mcp.rs index 60e325c4e..91ebce4ad 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -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 { + ) -> HashMap { let mcp_config = config.to_mcp_config(self.plugins_manager.as_ref()).await; effective_mcp_servers(&mcp_config, auth) } diff --git a/codex-rs/core/src/mcp_skill_dependencies.rs b/codex-rs/core/src/mcp_skill_dependencies.rs index aefc783cb..44764e006 100644 --- a/codex-rs/core/src/mcp_skill_dependencies.rs +++ b/codex-rs/core/src/mcp_skill_dependencies.rs @@ -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()) diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index c7e665cb0..9c01ad48b 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -324,9 +324,8 @@ async fn handle_approved_mcp_tool_call( request_meta: Option, mcp_app_resource_uri: Option, ) -> 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, diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index ad04d30c6..a7d7a965a 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -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 = diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 0910a3f87..1f40438cd 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -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; diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index e12f1fdee..11fa4109b 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -936,11 +936,12 @@ impl Session { sess.start_skills_watcher_listener(); let mut required_mcp_servers: Vec = 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 diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index effbf5372..171ca10d6 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -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::>(); - Ok(Arc::new(ToolRouter::from_config( &turn_context.tools_config, ToolRouterParams { diff --git a/codex-rs/core/src/tools/code_mode/mod.rs b/codex-rs/core/src/tools/code_mode/mod.rs index 77cbd72b0..4bafd2650 100644 --- a/codex-rs/core/src/tools/code_mode/mod.rs +++ b/codex-rs/core/src/tools/code_mode/mod.rs @@ -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::>(); + 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, diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index b2a277804..34d60dd4a 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -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; diff --git a/codex-rs/memories/mcp/Cargo.toml b/codex-rs/memories/mcp/Cargo.toml index 4e91472d4..42789a54d 100644 --- a/codex-rs/memories/mcp/Cargo.toml +++ b/codex-rs/memories/mcp/Cargo.toml @@ -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"] } diff --git a/codex-rs/memories/mcp/src/lib.rs b/codex-rs/memories/mcp/src/lib.rs index 004d48ee7..a643bf665 100644 --- a/codex-rs/memories/mcp/src/lib.rs +++ b/codex-rs/memories/mcp/src/lib.rs @@ -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; diff --git a/codex-rs/memories/mcp/src/server.rs b/codex-rs/memories/mcp/src/server.rs index f25e75d37..749726993 100644 --- a/codex-rs/memories/mcp/src/server.rs +++ b/codex-rs/memories/mcp/src/server.rs @@ -185,19 +185,27 @@ impl ServerHandler for MemoriesMcpServer { } } -pub async fn run_stdio_server(codex_home: &AbsolutePathBuf) -> anyhow::Result<()> { +pub async fn run_server(codex_home: &AbsolutePathBuf, transport: T) -> anyhow::Result<()> +where + T: rmcp::transport::IntoTransport, + 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), diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index c4f056892..c0c32c4b9 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -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", diff --git a/codex-rs/rmcp-client/src/in_process_transport.rs b/codex-rs/rmcp-client/src/in_process_transport.rs new file mode 100644 index 000000000..f78d4ce0b --- /dev/null +++ b/codex-rs/rmcp-client/src/in_process_transport.rs @@ -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>; +} diff --git a/codex-rs/rmcp-client/src/lib.rs b/codex-rs/rmcp-client/src/lib.rs index 57e9f0e80..e1ee18c75 100644 --- a/codex-rs/rmcp-client/src/lib.rs +++ b/codex-rs/rmcp-client/src/lib.rs @@ -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; diff --git a/codex-rs/rmcp-client/src/rmcp_client.rs b/codex-rs/rmcp-client/src/rmcp_client.rs index 6f38acadd..5079570f3 100644 --- a/codex-rs/rmcp-client/src/rmcp_client.rs +++ b/codex-rs/rmcp-client/src/rmcp_client.rs @@ -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, + }, Stdio { command: StdioServerCommand, launcher: Arc, @@ -275,6 +282,26 @@ pub struct RmcpClient { } impl RmcpClient { + pub async fn new_in_process_client( + factory: Arc, + ) -> io::Result { + 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, @@ -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 { 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, )> { 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,