From 00a7b888b23715989db19b74f6cb623ca46be620 Mon Sep 17 00:00:00 2001 From: jif Date: Fri, 21 Aug 2026 14:15:46 +0000 Subject: [PATCH] Discover HTTP MCP servers from selected executors (#39941) ## What changed - Read `mcp_servers` configuration and requirements from each selected remote executor and add eligible HTTP servers to the thread's MCP runtime. - Bind discovered servers to the thread's concrete executor snapshot and apply environment MCP policy and requirements. Discovery is best effort, and executor-local servers are not treated as required at startup. - Ignore unsupported stdio servers and HTTP configurations that depend on environment-provided headers or header helpers. ## Testing - Added an app-server integration test covering discovery, authenticated HTTP tool invocation, requirements enforcement, and exclusion of stdio servers. GitOrigin-RevId: 6e1cdcebdbb1cc21a5a2285fbc5617d8d5997182 --- .../src/request_processors/mcp_processor.rs | 22 +-- .../app-server/tests/suite/v2/executor_mcp.rs | 154 ++++++++++++++++- codex-rs/codex-mcp/src/runtime.rs | 20 ++- codex-rs/core/src/session/mcp.rs | 23 ++- codex-rs/core/src/session/mcp_runtime.rs | 158 ++++++++++++++++++ .../exec-server/src/environment_config.rs | 83 +++++++++ 6 files changed, 445 insertions(+), 15 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/mcp_processor.rs b/codex-rs/app-server/src/request_processors/mcp_processor.rs index 2cfb470386..5bb1933aac 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/codex-rs/app-server/src/request_processors/mcp_processor.rs @@ -275,19 +275,19 @@ impl McpRequestProcessor { }; let mcp_manager = self.thread_manager.mcp_manager(); let auth = self.auth_manager.auth().await; - let (mcp_config, runtime_context) = match thread { - Some(thread) => thread.runtime_mcp_config_and_context(&config).await, - None => { - let mcp_config = mcp_manager.runtime_config(&config).await; - let runtime_context = McpRuntimeContext::new( - self.thread_manager.environment_manager(), - config.cwd.to_path_buf(), - ); - (mcp_config, runtime_context) - } - }; + let environment_manager = self.thread_manager.environment_manager(); tokio::spawn(async move { + let (mcp_config, runtime_context) = match thread { + Some(thread) => thread.runtime_mcp_config_and_context(&config).await, + None => { + let mcp_config = mcp_manager.runtime_config(&config).await; + let runtime_context = + McpRuntimeContext::new(environment_manager, config.cwd.to_path_buf()); + (mcp_config, runtime_context) + } + }; + Self::list_mcp_server_status_task( outgoing, request, diff --git a/codex-rs/app-server/tests/suite/v2/executor_mcp.rs b/codex-rs/app-server/tests/suite/v2/executor_mcp.rs index 8e426fe7bf..1239185f74 100644 --- a/codex-rs/app-server/tests/suite/v2/executor_mcp.rs +++ b/codex-rs/app-server/tests/suite/v2/executor_mcp.rs @@ -70,9 +70,148 @@ const EXECUTOR_ENV_VALUE: &str = "executor-only"; const EXECUTOR_HTTP_AUTH_ENV_NAME: &str = "NODE_REPL_AUTH_TOKEN"; const EXECUTOR_HTTP_AUTH_ENV_VALUE: &str = "executor-only-http-token"; const EXECUTOR_ID: &str = "executor-1"; +const EXECUTOR_DISABLED_PLUGIN_SERVER_NAME: &str = "executor_disabled_plugin"; +const PROJECT_MCP_SERVER_NAME: &str = "node_repl"; +const PROJECT_MCP_BEARER_TOKEN: &str = "executor-browser-token"; const REFRESH_PROBE_SERVER_NAME: &str = "refresh_probe"; const TOOL_CALL_ID: &str = "executor-mcp-call"; +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn selected_executor_discovers_executor_local_http_mcp() -> Result<()> { + let responses_server = responses::start_mock_server().await; + let http_listener = TcpListener::bind("127.0.0.1:0").await?; + let http_addr = http_listener.local_addr()?; + let http_mcp_service = StreamableHttpService::new( + || Ok(ExecutorHttpMcpServer), + Arc::new(LocalSessionManager::default()), + StreamableHttpServerConfig::default().with_allowed_hosts(["executor-only.invalid"]), + ); + let http_router = + Router::new() + .nest_service("/mcp", http_mcp_service) + .layer(axum::middleware::from_fn( + |request: axum::extract::Request, next: axum::middleware::Next| async move { + let authorized = request + .headers() + .get(axum::http::header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value == "Bearer executor-browser-token"); + if !authorized { + return axum::http::StatusCode::UNAUTHORIZED.into_response(); + } + next.run(request).await + }, + )); + let http_server_handle = tokio::spawn(async move { + let _ = axum::serve(http_listener, http_router).await; + }); + let codex_home = TempDir::new()?; + let executor_home = TempDir::new()?; + MockResponsesConfig::new(&responses_server.uri()) + .with_sandbox_mode("danger-full-access") + .write(codex_home.path())?; + std::fs::write( + codex_home.path().join("requirements.toml"), + format!( + "[mcp_servers.{PROJECT_MCP_SERVER_NAME}.identity]\nurl = \"{EXECUTOR_HTTP_MCP_URL}\"\n" + ), + )?; + std::fs::write( + executor_home.path().join("config.toml"), + format!( + "[mcp_servers.{PROJECT_MCP_SERVER_NAME}]\nurl = \"{EXECUTOR_HTTP_MCP_URL}\"\nhttp_headers = {{ Authorization = \"Bearer {PROJECT_MCP_BEARER_TOKEN}\" }}\nstartup_timeout_sec = 10\n\n[mcp_servers.ignored_stdio]\ncommand = \"executor-local-command\"\nenv_vars = [\"EXECUTOR_ONLY_TOKEN\"]\ncwd = \"./server\"\n\n[mcp_servers.policy_unlisted]\nurl = \"{EXECUTOR_HTTP_MCP_URL}\"\n" + ), + )?; + let codex_bin = toml::Value::String( + codex_utils_cargo_bin::cargo_bin("codex")? + .to_string_lossy() + .into_owned(), + ); + let executor_home_value = + toml::Value::String(executor_home.path().to_string_lossy().into_owned()); + let http_proxy = toml::Value::String(format!("http://{http_addr}")); + std::fs::write( + codex_home.path().join("environments.toml"), + format!( + "default = \"{EXECUTOR_ID}\"\ninclude_local = false\n\n[[environments]]\nid = \"{EXECUTOR_ID}\"\nprogram = {codex_bin}\nargs = [\"exec-server\", \"--listen\", \"stdio\"]\n[environments.env]\nCODEX_HOME = {executor_home_value}\nHTTP_PROXY = {http_proxy}\n" + ), + )?; + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build_initialized_with_timeout(DEFAULT_READ_TIMEOUT) + .await?; + let thread_id = start_thread(&mut app_server, /*selected_capability_roots*/ None).await?; + let servers = mcp_server_statuses(&mut app_server, thread_id.clone()).await?; + assert!( + servers + .iter() + .any(|server| { server.name == PROJECT_MCP_SERVER_NAME && server.plugin_id.is_none() }) + ); + assert!(servers.iter().all(|server| server.name != "ignored_stdio")); + assert!( + servers + .iter() + .any(|server| server.name == "policy_unlisted" && server.tools.is_empty()) + ); + + let namespace = format!("mcp__{PROJECT_MCP_SERVER_NAME}"); + let response_mock = responses::mount_sse_sequence( + &responses_server, + vec![ + responses::sse(vec![ + responses::ev_response_created("resp-browser-mcp-call"), + responses::ev_function_call_with_namespace( + "browser-mcp-call", + &namespace, + "echo", + &json!({"message": "browser use works"}).to_string(), + ), + responses::ev_completed("resp-browser-mcp-call"), + ]), + responses::sse(vec![ + responses::ev_response_created("resp-browser-mcp-done"), + responses::ev_assistant_message("msg-browser-mcp-done", "Done"), + responses::ev_completed("resp-browser-mcp-done"), + ]), + ], + ) + .await; + let request_id = app_server + .send_turn_start_request(TurnStartParams { + thread_id, + input: vec![UserInput::Text { + text: "Use the executor browser".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let _: TurnStartResponse = + timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(request_id)).await??; + timeout( + DEFAULT_READ_TIMEOUT, + app_server.read_stream_until_notification_message("turn/completed"), + ) + .await??; + let requests = response_mock.requests(); + assert_eq!(requests.len(), 2); + assert!(requests[0].tool_by_name(&namespace, "echo").is_some()); + let output = requests[1].function_call_output("browser-mcp-call"); + assert!( + output + .get("output") + .and_then(serde_json::Value::as_str) + .is_some_and(|output| output.contains("ECHOING: browser use works")) + ); + + http_server_handle.abort(); + let _ = http_server_handle.await; + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Result<()> { assert!(std::env::var_os(EXECUTOR_HTTP_AUTH_ENV_NAME).is_none()); @@ -186,6 +325,13 @@ async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Resu let global_callback_port = global_callback_listener.local_addr()?.port(); drop(plugin_callback_listener); let codex_home = TempDir::new()?; + let executor_home = TempDir::new()?; + std::fs::write( + executor_home.path().join("config.toml"), + format!( + "[mcp_servers.{EXECUTOR_DISABLED_PLUGIN_SERVER_NAME}]\nurl = \"{EXECUTOR_HTTP_MCP_URL}\"\nenabled = false\n" + ), + )?; let root_config = format!( "compact_prompt = \"compact\"\nmodel_auto_compact_token_limit = 1024\nmcp_oauth_credentials_store = \"file\"\nmcp_oauth_callback_port = {global_callback_port}" ); @@ -215,7 +361,7 @@ async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> Resu .args(["exec-server", "--listen", "ws://127.0.0.1:0"]) .stdout(Stdio::piped()) .kill_on_drop(true) - .env("CODEX_HOME", codex_home.path()) + .env("CODEX_HOME", executor_home.path()) .env(EXECUTOR_ENV_NAME, EXECUTOR_ENV_VALUE) .env(EXECUTOR_HTTP_AUTH_ENV_NAME, EXECUTOR_HTTP_AUTH_ENV_VALUE) .env("HTTP_PROXY", format!("http://{http_addr}")) @@ -259,6 +405,11 @@ url = "{executor_url}" "bearer_token_env_var": EXECUTOR_HTTP_AUTH_ENV_NAME, "startup_timeout_sec": 10, }, + (EXECUTOR_DISABLED_PLUGIN_SERVER_NAME): { + "url": EXECUTOR_HTTP_MCP_URL, + "environment_id": "local", + "startup_timeout_sec": 10, + }, (OAUTH_MCP_SERVER_NAME): { "url": EXECUTOR_OAUTH_MCP_URL, "environment_id": "local", @@ -583,6 +734,7 @@ startup_timeout_sec = 10 assert_eq!( selected_server_owners, BTreeMap::from([ + (EXECUTOR_DISABLED_PLUGIN_SERVER_NAME.to_string(), None), ( HTTP_MCP_SERVER_NAME.to_string(), Some("executor-demo@1".to_string()), diff --git a/codex-rs/codex-mcp/src/runtime.rs b/codex-rs/codex-mcp/src/runtime.rs index daa47c29aa..56bc38df7f 100644 --- a/codex-rs/codex-mcp/src/runtime.rs +++ b/codex-rs/codex-mcp/src/runtime.rs @@ -567,6 +567,7 @@ pub struct SandboxState { #[derive(Clone)] pub struct McpRuntimeContext { environment_manager: Arc, + selected_environments: HashMap>, local_process_cwd: PathBuf, local_http_client: Arc, } @@ -610,11 +611,21 @@ impl McpRuntimeContext { ); Self { environment_manager, + selected_environments: HashMap::new(), local_process_cwd, local_http_client, } } + /// Pins the concrete environment handles captured for this thread or model step. + pub fn with_selected_environments( + mut self, + selected_environments: HashMap>, + ) -> Self { + self.selected_environments = selected_environments; + self + } + pub(crate) fn local_process_cwd(&self) -> PathBuf { self.local_process_cwd.clone() } @@ -632,8 +643,13 @@ impl McpRuntimeContext { // HTTP is the one current exception: it can use the ambient HTTP client // even when no local Environment is configured. if let Some(environment) = self - .environment_manager - .get_environment(&config.environment_id) + .selected_environments + .get(&config.environment_id) + .cloned() + .or_else(|| { + self.environment_manager + .get_environment(&config.environment_id) + }) { return Ok(Some(environment)); } diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index a46655e0d3..86aac5741c 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -120,7 +120,7 @@ impl Session { windows_sandbox_level, ) .await; - let mcp_config = self + let mcp_projection = self .services .mcp_manager .runtime_config_for_step( @@ -135,6 +135,9 @@ impl Session { &ready_selected_capability_roots, executor_capability_discovery.as_deref(), ) + .await; + let mcp_config = self + .project_selected_environment_mcp_servers(config, &environments, mcp_projection) .await .config; let local_process_cwd = environments @@ -144,6 +147,17 @@ impl Session { let runtime_context = McpRuntimeContext::new( self.services.turn_environments.environment_manager(), local_process_cwd, + ) + .with_selected_environments( + environments + .turn_environments() + .map(|environment| { + ( + environment.selection.environment_id.clone(), + Arc::clone(&environment.environment), + ) + }) + .collect(), ); (mcp_config, runtime_context) } @@ -265,6 +279,13 @@ impl Session { executor_capability_discovery.as_deref(), ) .await; + let mcp_projection = self + .project_selected_environment_mcp_servers( + &desired.config, + &desired.environments, + mcp_projection, + ) + .await; let selected_plugins = mcp_projection.selected_plugins.clone(); let input = self.build_mcp_runtime_input( &desired, diff --git a/codex-rs/core/src/session/mcp_runtime.rs b/codex-rs/core/src/session/mcp_runtime.rs index 3c1bebb4db..2f7aad4818 100644 --- a/codex-rs/core/src/session/mcp_runtime.rs +++ b/codex-rs/core/src/session/mcp_runtime.rs @@ -7,10 +7,18 @@ use super::session::SessionConfiguration; use super::*; use crate::mcp::McpRuntimeProjection; +use codex_config::McpServerDisabledReason; +use codex_config::McpServerTransportConfig; +use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::ElicitationReviewerHandle; +use codex_mcp::McpEnvironmentAuthority; +use codex_mcp::McpServerRegistration; +use codex_mcp::McpServerSource; use codex_mcp::McpStartupPolicy; use codex_mcp::PreparedMcpCall; use codex_protocol::capabilities::SelectedCapabilityRoot; +use codex_protocol::protocol::EnvironmentConfigState; +use std::collections::HashSet; pub(super) struct McpDesiredState { pub(super) config: Arc, @@ -133,6 +141,137 @@ impl Session { self.services.mcp_runtime.validate_required_servers().await } + /// Adds effective executor-owned configuration from this exact thread snapshot. + pub(super) fn project_selected_environment_mcp_servers<'a>( + &'a self, + config: &'a Config, + environments: &'a TurnEnvironmentSnapshot, + mut projection: McpRuntimeProjection, + ) -> BoxFuture<'a, McpRuntimeProjection> { + Box::pin(async move { + let mut catalog = None; + let mut registered = HashSet::new(); + for selected in environments.turn_environments() { + let environment = &selected.environment; + if !environment.is_remote() { + continue; + } + + let environment_id = &selected.selection.environment_id; + let servers = match environment + .discover_http_mcp_servers(selected.cwd().clone()) + .await + { + Ok(servers) => servers, + Err(error) => { + tracing::warn!( + environment_id, + %error, + "failed to discover executor-local MCP servers" + ); + continue; + } + }; + for (name, mut server) in servers { + if name == CODEX_APPS_MCP_SERVER_NAME + || !server.is_local_environment() + || projection + .config + .mcp_server_catalog + .server(&name) + .is_some_and(|existing| { + existing.config().environment_id != *environment_id + || !matches!( + existing.source(), + McpServerSource::Plugin(_) + | McpServerSource::SelectedPlugin(_) + ) + }) + || registered.contains(&name) + { + continue; + } + // Selected executors can attach after startup, so their MCPs are best effort. + server.required = false; + let McpServerTransportConfig::StreamableHttp { + env_http_headers, + http_headers_helper, + .. + } = &server.transport + else { + continue; + }; + if http_headers_helper.is_some() + || env_http_headers + .as_ref() + .is_some_and(|headers| !headers.is_empty()) + { + tracing::warn!( + environment_id, + server = name, + "executor-local HTTP header helpers are not supported" + ); + continue; + } + server.environment_id = environment_id.clone(); + if let Some(requirements) = config + .config_layer_stack + .requirements() + .mcp_servers + .as_ref() + && !requirements + .value + .get(&name) + .is_some_and(|requirement| server.matches_requirement(requirement)) + { + server.enabled = false; + server.disabled_reason = Some(McpServerDisabledReason::Requirements { + source: requirements.source.clone(), + }); + } + registered.insert(name.clone()); + catalog + .get_or_insert_with(|| projection.config.mcp_server_catalog.to_builder()) + .register(McpServerRegistration::from_config(name, server)); + } + } + + if let Some(catalog) = catalog { + let selections = self.services.turn_environments.selections(); + projection.config.mcp_server_catalog = + catalog.build_with_environment_authority(|environment_id| { + let Some(selection) = selections + .iter() + .find(|selection| selection.environment_id == environment_id) + else { + return if environment_id + == codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID + { + McpEnvironmentAuthority::Unrestricted + } else { + McpEnvironmentAuthority::SelectedPluginsOnly + }; + }; + match &selection.config { + EnvironmentConfigState::FromThread => { + McpEnvironmentAuthority::Unrestricted + } + EnvironmentConfigState::Pending | EnvironmentConfigState::Failed(_) => { + McpEnvironmentAuthority::Unavailable + } + EnvironmentConfigState::Ready(config) => config + .mcp_policy + .as_ref() + .map_or(McpEnvironmentAuthority::Unrestricted, |policy| { + McpEnvironmentAuthority::Restricted(policy) + }), + } + }); + } + projection + }) + } + #[tracing::instrument(name = "mcp.runtime.refresh", skip_all)] pub(super) async fn publish_mcp_runtime( &self, @@ -141,6 +280,13 @@ impl Session { ready_selected_capability_roots: &[SelectedCapabilityRoot], elicitation_reviewer: Option, ) { + let mcp_projection = self + .project_selected_environment_mcp_servers( + &desired.config, + &desired.environments, + mcp_projection, + ) + .await; let selected_plugins = mcp_projection.selected_plugins.clone(); let input = self.build_mcp_runtime_input( desired, @@ -187,6 +333,18 @@ impl Session { let runtime_context = McpRuntimeContext::new( self.services.turn_environments.environment_manager(), desired.local_process_cwd.clone(), + ) + .with_selected_environments( + desired + .environments + .turn_environments() + .map(|environment| { + ( + environment.selection.environment_id.clone(), + Arc::clone(&environment.environment), + ) + }) + .collect(), ); let codex_apps_auth_manager = codex_mcp::host_owned_codex_apps_enabled(&mcp_config, auth.as_ref()) diff --git a/codex-rs/exec-server/src/environment_config.rs b/codex-rs/exec-server/src/environment_config.rs index 7124e6f3f6..d6d32d5e1a 100644 --- a/codex-rs/exec-server/src/environment_config.rs +++ b/codex-rs/exec-server/src/environment_config.rs @@ -1,4 +1,11 @@ +use std::collections::HashMap; + use codex_config::CONFIG_TOML_FILE; +use codex_config::McpServerConfig; +use codex_config::McpServerDisabledReason; +use codex_config::RequirementSource; +use codex_config::RequirementsLayerEntry; +use codex_config::compose_requirements_for_hostname; use codex_config::format_config_layer_source; use codex_config::host_name; use codex_config::loader::LocalTomlLayerStack; @@ -11,6 +18,9 @@ use codex_file_system::ExecutorFileSystem; use codex_utils_home_dir::find_codex_home; use codex_utils_path_uri::PathUri; +use crate::Environment; +use crate::ExecServerError; + #[derive(Debug, thiserror::Error)] pub(crate) enum ReadEnvironmentConfigError { #[error("{0}")] @@ -52,6 +62,79 @@ pub(crate) async fn read_environment_config( }) } +impl Environment { + /// Reads executor-owned HTTP MCP servers from the selected project's current config. + pub async fn discover_http_mcp_servers( + &self, + cwd: PathUri, + ) -> Result, ExecServerError> { + let response = tokio::time::timeout(std::time::Duration::from_secs(10), async { + if !self.info().await?.capabilities.environment_config_read { + return Ok(None); + } + self.read_environment_config(EnvironmentConfigReadParams { + cwd, + config_paths: vec![vec!["mcp_servers".to_string()]], + requirements_paths: vec![vec!["mcp_servers".to_string()]], + }) + .await + .map(Some) + }) + .await + .map_err(|_| ExecServerError::Protocol("executor MCP discovery timed out".to_string()))??; + let Some(response) = response else { + return Ok(Vec::new()); + }; + let requirements = compose_requirements_for_hostname( + response.requirements.layers.into_iter().map(|layer| { + RequirementsLayerEntry::from_toml(RequirementSource::Unknown, layer.toml) + }), + response.hostname.as_deref(), + ) + .map_err(|error| { + ExecServerError::Protocol(format!("invalid executor-local MCP requirements: {error}")) + })? + .and_then(|requirements| requirements.mcp_servers); + let mut merged = toml::Value::Table(toml::map::Map::new()); + for layer in response.config.layers { + let config = toml::from_str::(&layer.toml).map_err(|error| { + ExecServerError::Protocol(format!("invalid executor-local MCP config: {error}")) + })?; + codex_config::merge_toml_values(&mut merged, &config); + } + let mut servers = merged + .get("mcp_servers") + .cloned() + .unwrap_or_else(|| toml::Value::Table(toml::map::Map::new())); + if let Some(servers) = servers.as_table_mut() { + servers.retain(|_, server| server.get("url").is_some()); + } + let servers = servers + .try_into::>() + .map_err(|error| { + ExecServerError::Protocol(format!("invalid executor-local MCP servers: {error}")) + })?; + + Ok(servers + .into_iter() + .map(|(name, mut server)| { + if let Some(requirements) = requirements.as_ref() + && !requirements + .value + .get(&name) + .is_some_and(|requirement| server.matches_requirement(requirement)) + { + server.enabled = false; + server.disabled_reason = Some(McpServerDisabledReason::Requirements { + source: requirements.source.clone(), + }); + } + (name, server) + }) + .collect()) + } +} + fn validate_paths(params: &EnvironmentConfigReadParams) -> Result<(), ReadEnvironmentConfigError> { if params.config_paths.is_empty() && params.requirements_paths.is_empty() { return Err(ReadEnvironmentConfigError::InvalidParams(