Support HTTP MCP servers from selected executor plugins (#28522)

## Why Selected executor plugins can declare both stdio and Streamable HTTP MCP servers, but only stdio registrations were retained. That silently drops part of the plugin's tool surface and prevents HTTP traffic from using the owning executor's network. ## What changed - retain selected-plugin Streamable HTTP MCP declarations alongside stdio declarations - route their HTTP clients through the owning executor environment - preserve local auth-header environment references while rejecting them for executor-hosted declarations - cover thread isolation, refresh, and an executor-only HTTP route end to end
2026-07-01 00:31:56 +08:00 · 2026-06-25 10:10:36 +01:00
parent 5579792b3b
commit 6368937939
6 changed files with 308 additions and 59 deletions
@@ -137,7 +137,7 @@ Example with notification opt-out:

 ## API Overview

- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. Deprecated experimental `multiAgentMode` is ignored; use Ultra reasoning effort for proactive multi-agent behavior. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd`. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots using environment-native absolute paths. Skills found below those roots are listed and read through the owning environment. Stdio MCP servers declared by selected plugins are also started in that environment; HTTP MCP declarations remain inactive.
+- `thread/start` — create a new thread; emits `thread/started` (including the current `thread.status`) and auto-subscribes you to turn/item events for that thread. When the request includes a `cwd` and the resolved sandbox is `workspace-write` or full access, app-server also marks that project as trusted in the user `config.toml`. Pass `sessionStartSource: "clear"` when starting a replacement thread after clearing the current session so `SessionStart` hooks receive `source: "clear"` instead of the default `"startup"`. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; paths must be absolute. For permissions, prefer experimental `permissions` profile selection by id; the legacy `sandbox` shorthand is still accepted but cannot be combined with `permissions`. Deprecated experimental `multiAgentMode` is ignored; use Ultra reasoning effort for proactive multi-agent behavior. Experimental `environments` selects the sticky execution environments for turns on the thread; omit it to use the server default, pass `[]` to disable environments, or pass explicit environment ids with per-environment `cwd`. Experimental `selectedCapabilityRoots` selects environment-owned plugin or standalone-skill roots using environment-native absolute paths. Skills found below those roots are listed and read through the owning environment. Stdio MCP servers declared by selected plugins are started in that environment, and HTTP MCP connections use that environment's HTTP client.
 - `thread/resume` — reopen an existing thread by id so subsequent `turn/start` calls append to it. Accepts the same permission override rules as `thread/start`.
 - `thread/fork` — fork an existing thread into a new thread id by copying the stored history; if the source thread is currently mid-turn, the fork records the same interruption marker as `turn/interrupt` instead of inheriting an unmarked partial turn suffix. The returned `thread.forkedFromId` points at the source thread when known. Accepts `ephemeral: true` for an in-memory temporary fork, emits `thread/started` (including the current `thread.status`), and auto-subscribes you to turn/item events for the new thread. Experimental clients can pass `excludeTurns: true` when they plan to page fork history via `thread/turns/list` instead of receiving the full turn array immediately. Accepts the same permission override rules as `thread/start`.
 - `thread/start`, `thread/resume`, and `thread/fork` responses include the legacy `sandbox` compatibility projection. `instructionSources` lists loaded instruction files using each source environment's native absolute path syntax, including files loaded from remote environments. Experimental clients can read `runtimeWorkspaceRoots` for the thread-scoped runtime roots and `activePermissionProfile` for the named or implicit built-in profile identity/provenance when known. Their deprecated experimental `multiAgentMode` field, and the corresponding thread setting, always report `explicitRequestOnly`; Ultra reasoning effort is the source of proactive multi-agent behavior.
@@ -2,6 +2,7 @@ use anyhow::Result;
 use app_test_support::TestAppServer;
 use app_test_support::to_response;
 use app_test_support::write_mock_responses_config_toml;
+use axum::Router;
 use codex_app_server_protocol::CapabilityRootLocation;
 use codex_app_server_protocol::ListMcpServerStatusParams;
 use codex_app_server_protocol::ListMcpServerStatusResponse;
@@ -17,13 +18,32 @@ use codex_utils_path_uri::PathUri;
 use core_test_support::responses;
 use core_test_support::stdio_server_bin;
 use pretty_assertions::assert_eq;
+use rmcp::handler::server::ServerHandler;
+use rmcp::model::CallToolRequestParams;
+use rmcp::model::CallToolResult;
+use rmcp::model::JsonObject;
+use rmcp::model::ListToolsResult;
+use rmcp::model::ServerCapabilities;
+use rmcp::model::ServerInfo;
+use rmcp::model::Tool;
+use rmcp::model::ToolAnnotations;
+use rmcp::service::RequestContext;
+use rmcp::service::RoleServer;
+use rmcp::transport::StreamableHttpServerConfig;
+use rmcp::transport::StreamableHttpService;
+use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
 use serde_json::json;
+use std::borrow::Cow;
 use std::collections::BTreeMap;
+use std::sync::Arc;
 use std::time::Duration;
 use tempfile::TempDir;
+use tokio::net::TcpListener;
 use tokio::time::timeout;

 const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(20);
+const EXECUTOR_HTTP_MCP_URL: &str = "http://executor-only.invalid/mcp";
+const HTTP_MCP_SERVER_NAME: &str = "executor_http";
 const MCP_SERVER_NAME: &str = "executor_demo";
 const EXECUTOR_ENV_NAME: &str = "MCP_EXECUTOR_MARKER";
 const EXECUTOR_ENV_VALUE: &str = "executor-only";
@@ -32,8 +52,19 @@ 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_plugin_exposes_its_stdio_mcp_only_to_that_thread() -> Result<()> {
+async fn selected_executor_plugin_exposes_its_mcps_only_to_that_thread() -> 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);
+    let http_server_handle = tokio::spawn(async move {
+        let _ = axum::serve(http_listener, http_router).await;
+    });
    let codex_home = TempDir::new()?;
    write_mock_responses_config_toml(
        codex_home.path(),
@@ -44,6 +75,12 @@ async fn selected_executor_plugin_exposes_its_stdio_mcp_only_to_that_thread() ->
        "mock_provider",
        "compact",
    )?;
+    let codex_bin = toml::Value::String(
+        codex_utils_cargo_bin::cargo_bin("codex")?
+            .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!(
@@ -52,16 +89,12 @@ include_local = true

 [[environments]]
 id = "{EXECUTOR_ID}"
-program = {}
+program = {codex_bin}
 args = ["exec-server", "--listen", "stdio"]
 [environments.env]
 {EXECUTOR_ENV_NAME} = "{EXECUTOR_ENV_VALUE}"
-"#,
-            toml::Value::String(
-                codex_utils_cargo_bin::cargo_bin("codex")?
-                    .to_string_lossy()
-                    .into_owned()
-            )
+HTTP_PROXY = {http_proxy}
+"#
        ),
    )?;

@@ -79,6 +112,11 @@ args = ["exec-server", "--listen", "stdio"]
                    "command": stdio_server_bin()?,
                    "env_vars": [EXECUTOR_ENV_NAME],
                    "startup_timeout_sec": 10,
+                },
+                (HTTP_MCP_SERVER_NAME): {
+                    "url": EXECUTOR_HTTP_MCP_URL,
+                    "environment_id": "local",
+                    "startup_timeout_sec": 10,
                }
            }
        }))?,
@@ -178,6 +216,26 @@ startup_timeout_sec = 10
    assert!(output.contains("ECHOING: hello from executor"));
    assert!(output.contains(EXECUTOR_ENV_VALUE));

+    let request_id = app_server
+        .send_mcp_server_tool_call_request(McpServerToolCallParams {
+            thread_id: selected_thread.clone(),
+            server: HTTP_MCP_SERVER_NAME.to_string(),
+            tool: "echo".to_string(),
+            arguments: Some(json!({"message": "hello over executor HTTP"})),
+            meta: None,
+        })
+        .await?;
+    let response = timeout(
+        DEFAULT_READ_TIMEOUT,
+        app_server.read_stream_until_response_message(RequestId::Integer(request_id)),
+    )
+    .await??;
+    let response: McpServerToolCallResponse = to_response(response)?;
+    assert_eq!(
+        response.structured_content,
+        Some(json!({"echo": "ECHOING: hello over executor HTTP"}))
+    );
+
    let request_id = app_server
        .send_mcp_server_tool_call_request(McpServerToolCallParams {
            thread_id: selected_thread.clone(),
@@ -200,25 +258,84 @@ startup_timeout_sec = 10
        Some(json!("ECHOING: refresh applied"))
    );

+    let selected_server_names = mcp_server_names(&mut app_server, selected_thread).await?;
    assert!(
-        mcp_server_names(&mut app_server, selected_thread)
-            .await?
+        selected_server_names
            .iter()
            .any(|name| name == MCP_SERVER_NAME)
    );
+    assert!(
+        selected_server_names
+            .iter()
+            .any(|name| name == HTTP_MCP_SERVER_NAME)
+    );

    let unselected_thread =
        start_thread(&mut app_server, /*selected_capability_roots*/ None).await?;
+    let unselected_server_names = mcp_server_names(&mut app_server, unselected_thread).await?;
    assert!(
-        mcp_server_names(&mut app_server, unselected_thread)
-            .await?
+        unselected_server_names
            .iter()
-            .all(|name| name != MCP_SERVER_NAME)
+            .all(|name| { name != MCP_SERVER_NAME && name != HTTP_MCP_SERVER_NAME })
    );

+    http_server_handle.abort();
+    let _ = http_server_handle.await;
+
    Ok(())
 }

+#[derive(Clone, Copy)]
+struct ExecutorHttpMcpServer;
+
+impl ServerHandler for ExecutorHttpMcpServer {
+    fn get_info(&self) -> ServerInfo {
+        ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
+    }
+
+    async fn list_tools(
+        &self,
+        _request: Option<rmcp::model::PaginatedRequestParams>,
+        _context: RequestContext<RoleServer>,
+    ) -> Result<ListToolsResult, rmcp::ErrorData> {
+        let input_schema: JsonObject = serde_json::from_value(json!({
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+            "required": ["message"],
+            "additionalProperties": false
+        }))
+        .map_err(|err| rmcp::ErrorData::internal_error(err.to_string(), None))?;
+        let mut tool = Tool::new(
+            Cow::Borrowed("echo"),
+            Cow::Borrowed("Echo a message."),
+            Arc::new(input_schema),
+        );
+        tool.annotations = Some(ToolAnnotations::new().read_only(true));
+
+        Ok(ListToolsResult {
+            tools: vec![tool],
+            next_cursor: None,
+            meta: None,
+        })
+    }
+
+    async fn call_tool(
+        &self,
+        request: CallToolRequestParams,
+        _context: RequestContext<RoleServer>,
+    ) -> Result<CallToolResult, rmcp::ErrorData> {
+        let message = request
+            .arguments
+            .as_ref()
+            .and_then(|arguments| arguments.get("message"))
+            .and_then(serde_json::Value::as_str)
+            .unwrap_or_default();
+        Ok(CallToolResult::structured(json!({
+            "echo": format!("ECHOING: {message}")
+        })))
+    }
+}
+
 async fn mcp_server_names(
    app_server: &mut TestAppServer,
    thread_id: String,
@@ -173,8 +173,35 @@ fn environment_cwd(

 fn bind_environment_env_vars(config: &mut McpServerConfig) -> Result<(), String> {
    let is_local_environment = config.is_local_environment();
-    let McpServerTransportConfig::Stdio { env_vars, .. } = &mut config.transport else {
-        return Ok(());
+    let env_vars = match &mut config.transport {
+        McpServerTransportConfig::Stdio { env_vars, .. } => env_vars,
+        // Never resolve executor-owned environment references in the host process.
+        // Remove this rejection once the owning executor resolves these fields.
+        McpServerTransportConfig::StreamableHttp {
+            bearer_token_env_var,
+            env_http_headers,
+            ..
+        } => {
+            if is_local_environment {
+                return Ok(());
+            }
+            if bearer_token_env_var.is_some() {
+                return Err(
+                    "`bearer_token_env_var` requires executor-side environment resolution for an executor-owned HTTP MCP"
+                        .to_string(),
+                );
+            }
+            if env_http_headers
+                .as_ref()
+                .is_some_and(|headers| !headers.is_empty())
+            {
+                return Err(
+                    "`env_http_headers` requires executor-side environment resolution for an executor-owned HTTP MCP"
+                        .to_string(),
+                );
+            }
+            return Ok(());
+        }
    };
    for env_var in env_vars {
        match env_var {
@@ -275,6 +275,98 @@ fn environment_placement_rejects_orchestrator_env_vars() {
    );
 }

+#[test]
+fn remote_environment_placement_rejects_http_env_references() {
+    let plugin_root = plugin_root();
+    let outcome = parse_executor_plugin_mcp_config(
+        &plugin_root_uri(&plugin_root),
+        r#"{
+            "bearer": {
+                "url": "https://example.com/bearer",
+                "bearer_token_env_var": "TOKEN"
+            },
+            "headers": {
+                "url": "https://example.com/headers",
+                "env_http_headers": {"Authorization": "TOKEN"}
+            }
+        }"#,
+        "executor-1",
+    )
+    .expect("parse plugin MCP config");
+
+    assert_eq!(
+        outcome,
+        PluginMcpConfigParseOutcome {
+            servers: BTreeMap::new(),
+            errors: vec![
+                PluginMcpServerParseError {
+                    name: "bearer".to_string(),
+                    message: "`bearer_token_env_var` requires executor-side environment resolution for an executor-owned HTTP MCP"
+                        .to_string(),
+                },
+                PluginMcpServerParseError {
+                    name: "headers".to_string(),
+                    message: "`env_http_headers` requires executor-side environment resolution for an executor-owned HTTP MCP"
+                        .to_string(),
+                },
+            ],
+        }
+    );
+}
+
+#[test]
+fn local_environment_placement_preserves_http_env_references() {
+    let plugin_root = plugin_root();
+    let outcome = parse_executor_plugin_mcp_config(
+        &plugin_root_uri(&plugin_root),
+        r#"{
+            "demo": {
+                "url": "https://example.com/mcp",
+                "bearer_token_env_var": "TOKEN",
+                "env_http_headers": {"X-Account": "ACCOUNT_ID"}
+            }
+        }"#,
+        DEFAULT_MCP_SERVER_ENVIRONMENT_ID,
+    )
+    .expect("parse plugin MCP config");
+
+    assert_eq!(
+        outcome,
+        PluginMcpConfigParseOutcome {
+            servers: BTreeMap::from([(
+                "demo".to_string(),
+                McpServerConfig {
+                    auth: Default::default(),
+                    transport: McpServerTransportConfig::StreamableHttp {
+                        url: "https://example.com/mcp".to_string(),
+                        bearer_token_env_var: Some("TOKEN".to_string()),
+                        http_headers: None,
+                        env_http_headers: Some(HashMap::from([(
+                            "X-Account".to_string(),
+                            "ACCOUNT_ID".to_string(),
+                        )])),
+                    },
+                    environment_id: DEFAULT_MCP_SERVER_ENVIRONMENT_ID.to_string(),
+                    enabled: true,
+                    required: false,
+                    supports_parallel_tool_calls: false,
+                    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: None,
+                    oauth_resource: None,
+                    tools: HashMap::new(),
+                },
+            )]),
+            errors: Vec::new(),
+        }
+    );
+}
+
 #[test]
 fn local_environment_placement_preserves_local_env_vars() {
    let plugin_root = plugin_root();
@@ -1,5 +1,4 @@
 use codex_config::McpServerConfig;
-use codex_config::McpServerTransportConfig;
 use codex_core_plugins::ResolvedExecutorPlugin;
 use codex_exec_server::ExecutorFileSystem;
 use codex_mcp::parse_executor_plugin_mcp_config;
@@ -48,7 +47,7 @@ pub(super) enum ExecutorPluginMcpProviderError {
 }

 impl ExecutorPluginMcpProvider {
-    /// Returns stdio servers declared by `plugin`, bound to its environment.
+    /// Returns MCP servers declared by `plugin`, bound to its environment.
    pub(super) async fn load(
        &self,
        plugin: &ResolvedExecutorPlugin,
@@ -131,21 +130,7 @@ async fn load_from_file_system(
        );
    }

-    Ok(parsed
-        .servers
-        .into_iter()
-        .filter_map(|(name, config)| match &config.transport {
-            McpServerTransportConfig::Stdio { .. } => Some((name, config)),
-            McpServerTransportConfig::StreamableHttp { .. } => {
-                tracing::warn!(
-                    plugin = plugin_id,
-                    server = name,
-                    "ignoring HTTP MCP server from executor plugin"
-                );
-                None
-            }
-        })
-        .collect())
+    Ok(parsed.servers.into_iter().collect())
 }

 #[cfg(test)]
@@ -163,33 +163,61 @@ async fn reads_declared_config_only_through_executor_file_system() {

    assert_eq!(
        servers,
-        vec![(
-            "demo".to_string(),
-            McpServerConfig {
-                auth: Default::default(),
-                transport: McpServerTransportConfig::Stdio {
-                    command: "demo-mcp".to_string(),
-                    args: Vec::new(),
-                    env: None,
-                    env_vars: Vec::new(),
-                    cwd: Some(LegacyAppPathString::from_path(plugin_root.as_path())),
+        vec![
+            (
+                "demo".to_string(),
+                McpServerConfig {
+                    auth: Default::default(),
+                    transport: McpServerTransportConfig::Stdio {
+                        command: "demo-mcp".to_string(),
+                        args: Vec::new(),
+                        env: None,
+                        env_vars: Vec::new(),
+                        cwd: Some(LegacyAppPathString::from_path(plugin_root.as_path())),
+                    },
+                    environment_id: "executor-test".to_string(),
+                    enabled: true,
+                    required: false,
+                    supports_parallel_tool_calls: false,
+                    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: None,
+                    oauth_resource: None,
+                    tools: HashMap::new(),
                },
-                environment_id: "executor-test".to_string(),
-                enabled: true,
-                required: false,
-                supports_parallel_tool_calls: false,
-                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: None,
-                oauth_resource: None,
-                tools: HashMap::new(),
-            },
-        )]
+            ),
+            (
+                "hosted".to_string(),
+                McpServerConfig {
+                    auth: Default::default(),
+                    transport: McpServerTransportConfig::StreamableHttp {
+                        url: "https://example.com/mcp".to_string(),
+                        bearer_token_env_var: None,
+                        http_headers: None,
+                        env_http_headers: None,
+                    },
+                    environment_id: "executor-test".to_string(),
+                    enabled: true,
+                    required: false,
+                    supports_parallel_tool_calls: false,
+                    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: None,
+                    oauth_resource: None,
+                    tools: HashMap::new(),
+                },
+            ),
+        ]
    );
    assert_eq!(reads(&file_system), vec![config_path]);
 }