openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-08-30 14:19:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Discover HTTP MCP servers from selected executors (#39941)

Browse Source 
## 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
This commit is contained in:
jif
2026-08-21 14:15:46 +00:00
committed by copyberry
parent 275ef855fa
commit 00a7b888b2
6 changed files with 445 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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,

154
codex-rs/app-server/tests/suite/v2/executor_mcp.rs
View File

@@ -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()),

20
codex-rs/codex-mcp/src/runtime.rs
View File

@@ -567,6 +567,7 @@ pub struct SandboxState {
#[derive(Clone)]
pub struct McpRuntimeContext {
environment_manager: Arc<EnvironmentManager>,
selected_environments: HashMap<String, Arc<Environment>>,
local_process_cwd: PathBuf,
local_http_client: Arc<dyn HttpClient>,
}
@@ -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<String, Arc<Environment>>,
) -> 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));
}

23
codex-rs/core/src/session/mcp.rs
View File

@@ -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,

158
codex-rs/core/src/session/mcp_runtime.rs
View File

@@ -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<Config>,
@@ -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<ElicitationReviewerHandle>,
) {
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())

83
codex-rs/exec-server/src/environment_config.rs
View File

@@ -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<Vec<(String, McpServerConfig)>, 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::<toml::Value>(&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::<HashMap<String, McpServerConfig>>()
.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(