app-server: add turn-scoped MCP metadata targets

codex-rs/app-server-protocol/src/protocol/v2/tests.rs

@@ -3560,6 +3560,8 @@ fn turn_start_params_preserve_explicit_null_service_tier() {
         thread_id: "thread_123".to_string(),
         input: vec![],
         responsesapi_client_metadata: None,
+        mcp_meta_by_server: None,
+        mcp_meta_by_connector: None,
         environments: None,
         cwd: None,
         runtime_workspace_roots: None,
@@ -3690,6 +3692,76 @@ fn turn_start_params_round_trip_environments() {
     );
 }
 
+#[test]
+fn turn_start_params_round_trip_mcp_meta_by_server() {
+    let params: TurnStartParams = serde_json::from_value(json!({
+        "threadId": "thread_123",
+        "input": [],
+        "mcpMetaByServer": {
+            "search_service": {
+                "client/location": {
+                    "country": "US"
+                }
+            }
+        },
+    }))
+    .expect("params should deserialize");
+
+    assert_eq!(
+        params.mcp_meta_by_server,
+        Some(HashMap::from([(
+            "search_service".to_string(),
+            HashMap::from([("client/location".to_string(), json!({ "country": "US" }),)]),
+        )]))
+    );
+    assert_eq!(
+        crate::experimental_api::ExperimentalApi::experimental_reason(&params),
+        Some("turn/start.mcpMetaByServer")
+    );
+
+    let serialized = serde_json::to_value(&params).expect("params should serialize");
+    assert_eq!(
+        serialized.pointer("/mcpMetaByServer/search_service/client~1location/country"),
+        Some(&json!("US"))
+    );
+}
+
+#[test]
+fn turn_start_params_round_trip_mcp_meta_by_connector() {
+    let params: TurnStartParams = serde_json::from_value(json!({
+        "threadId": "thread_123",
+        "input": [],
+        "mcpMetaByConnector": {
+            "connector_openai_search_service": {
+                "client/location": {
+                    "country": "US"
+                }
+            }
+        },
+    }))
+    .expect("params should deserialize");
+
+    assert_eq!(
+        params.mcp_meta_by_connector,
+        Some(HashMap::from([(
+            "connector_openai_search_service".to_string(),
+            HashMap::from([("client/location".to_string(), json!({ "country": "US" }),)]),
+        )]))
+    );
+    assert_eq!(
+        crate::experimental_api::ExperimentalApi::experimental_reason(&params),
+        Some("turn/start.mcpMetaByConnector")
+    );
+
+    let serialized = serde_json::to_value(&params).expect("params should serialize");
+    assert_eq!(
+        serialized.pointer(
+            "/mcpMetaByConnector/connector_openai_search_service/client~1location/country"
+        ),
+        Some(&json!("US"))
+    );
+}
+
 #[test]
 fn turn_start_params_preserve_empty_environments() {
     let params: TurnStartParams = serde_json::from_value(json!({

codex-rs/app-server-protocol/src/protocol/v2/turn.rs

@@ -53,6 +53,22 @@ pub struct TurnStartParams {
     #[experimental("turn/start.responsesapiClientMetadata")]
     #[ts(optional = nullable)]
     pub responsesapi_client_metadata: Option<HashMap<String, String>>,
+    /// Optional turn-scoped MCP request metadata keyed by configured custom MCP server name.
+    ///
+    /// Values are forwarded only to model-initiated tool calls for the
+    /// matching custom server. The host-routed `codex_apps` server cannot be
+    /// targeted through this field; use `mcpMetaByConnector` for its tools.
+    #[experimental("turn/start.mcpMetaByServer")]
+    #[ts(optional = nullable)]
+    pub mcp_meta_by_server: Option<HashMap<String, HashMap<String, JsonValue>>>,
+    /// Optional turn-scoped MCP request metadata keyed by app connector id.
+    ///
+    /// Values are forwarded only to model-initiated `codex_apps` tool calls
+    /// whose resolved connector id matches the key. Codex-owned `_meta` fields
+    /// take precedence.
+    #[experimental("turn/start.mcpMetaByConnector")]
+    #[ts(optional = nullable)]
+    pub mcp_meta_by_connector: Option<HashMap<String, HashMap<String, JsonValue>>>,
     /// Optional turn-scoped environments.
     ///
     /// Omitted uses the thread sticky environments. Empty disables

codex-rs/app-server/README.md

@@ -158,7 +158,7 @@ Example with notification opt-out:
 - `thread/shellCommand` — run a user-initiated `!` shell command against a thread; this runs unsandboxed with full access rather than inheriting the thread sandbox policy. Returns `{}` immediately while progress streams through standard turn/item notifications and any active turn receives the formatted output in its message stream.
 - `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
 - `thread/rollback` — drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success.
-- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; relative paths resolve against the effective turn cwd. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
+- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. Experimental `runtimeWorkspaceRoots` replaces the thread-scoped runtime workspace roots used to materialize `:workspace_roots`; relative paths resolve against the effective turn cwd. Experimental `mcpMetaByServer` adds request `_meta` to model-initiated calls routed through matching custom MCP server names, while experimental `mcpMetaByConnector` addresses individual app/connector calls by connector id; Codex-owned metadata fields take precedence. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode".
 - `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success.
 - `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. Review and manual compaction turns reject `turn/steer`.
 - `turn/interrupt` — request cancellation of an in-flight turn by `(thread_id, turn_id)`; success is an empty `{}` response and the turn finishes with `status: "interrupted"`.
@@ -634,6 +634,8 @@ Turns attach user input (text or images) to a thread and trigger Codex generatio
 
 You can optionally specify config overrides on the new turn. If specified, these settings become the default for subsequent turns on the same thread. `outputSchema` applies only to the current turn. Experimental `environments` is turn-scoped: omit it to inherit the thread's sticky environments, pass `[]` to run the turn with no environments, or pass explicit environment ids to override the sticky selection for this turn only.
 
+Experimental `mcpMetaByServer` and `mcpMetaByConnector` are also turn-scoped. Each `mcpMetaByServer` key is a configured custom MCP server name, and its object is merged into `_meta` only on model-initiated calls to that server. Each `mcpMetaByConnector` key is an app/connector id, and its object is merged only on model-initiated `codex_apps` calls whose resolved connector id matches. The aggregate `codex_apps` server cannot be targeted through `mcpMetaByServer`; use `mcpMetaByConnector` so metadata is not exposed to unrelated connectors. Codex-owned `_meta` fields cannot be replaced through either parameter. `responsesapiClientMetadata` remains Responses API request metadata and is not forwarded as MCP request `_meta`.
+
 `approvalsReviewer` accepts:
 
 - `"user"` — default. Review approval requests directly in the client.
@@ -649,6 +651,18 @@ You can optionally specify config overrides on the new turn. If specified, these
     "environments": [
         { "environmentId": "local", "cwd": "/Users/me/project" }
     ],
+    // Experimental: metadata for model-initiated calls to this custom MCP server.
+    "mcpMetaByServer": {
+        "my_custom_server": {
+            "request-id": "request-123"
+        }
+    },
+    // Experimental: metadata for model-initiated calls to this app/connector.
+    "mcpMetaByConnector": {
+        "connector_openai_search_service": {
+            "client/location": { "country": "US" }
+        }
+    },
     "approvalPolicy": "unlessTrusted",
     "sandboxPolicy": {
         "type": "workspaceWrite",

codex-rs/app-server/src/message_processor_tracing_tests.rs

@@ -658,6 +658,8 @@ async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> {
                         text_elements: Vec::new(),
                     }],
                     responsesapi_client_metadata: None,
+                    mcp_meta_by_server: None,
+                    mcp_meta_by_connector: None,
                     cwd: None,
                     runtime_workspace_roots: None,
                     approval_policy: None,

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -367,6 +367,15 @@ impl TurnRequestProcessor {
             );
             return Err(error);
         }
+        if params
+            .mcp_meta_by_server
+            .as_ref()
+            .is_some_and(|metadata| metadata.contains_key(codex_mcp::CODEX_APPS_MCP_SERVER_NAME))
+        {
+            return Err(invalid_request(
+                "`mcpMetaByServer` cannot target `codex_apps`; use `mcpMetaByConnector` to target an individual connector",
+            ));
+        }
         let (thread_id, thread) =
             self.load_thread(&params.thread_id)
                 .await
@@ -419,6 +428,8 @@ impl TurnRequestProcessor {
             environments: environment_selections,
             final_output_json_schema: params.output_schema,
             responsesapi_client_metadata: params.responsesapi_client_metadata,
+            mcp_meta_by_server: params.mcp_meta_by_server.map(Box::new),
+            mcp_meta_by_connector: params.mcp_meta_by_connector.map(Box::new),
             thread_settings,
         };
         let turn_id = self
@@ -752,6 +763,8 @@ impl TurnRequestProcessor {
                 mapped_items,
                 Some(&params.expected_turn_id),
                 params.responsesapi_client_metadata,
+                /*mcp_meta_by_server*/ None,
+                /*mcp_meta_by_connector*/ None,
             )
             .await
             .map_err(|err| {

codex-rs/app-server/tests/suite/v2/turn_start.rs

@@ -1024,6 +1024,61 @@ async fn turn_start_rejects_invalid_permission_selection_before_starting_turn()
     Ok(())
 }
 
+#[tokio::test]
+async fn turn_start_rejects_codex_apps_server_scoped_mcp_metadata() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    create_config_toml(
+        codex_home.path(),
+        "http://localhost/unused",
+        "never",
+        &BTreeMap::default(),
+    )?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let thread_req = mcp
+        .send_thread_start_request(ThreadStartParams {
+            model: Some("mock-model".to_string()),
+            ..Default::default()
+        })
+        .await?;
+    let thread_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
+
+    let turn_req = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id,
+            input: vec![V2UserInput::Text {
+                text: "Hello".to_string(),
+                text_elements: Vec::new(),
+            }],
+            mcp_meta_by_server: Some(HashMap::from([(
+                "codex_apps".to_string(),
+                HashMap::from([("client/location".to_string(), json!("US"))]),
+            )])),
+            ..Default::default()
+        })
+        .await?;
+    let err: JSONRPCError = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_error_message(RequestId::Integer(turn_req)),
+    )
+    .await??;
+
+    assert_eq!(err.error.code, INVALID_REQUEST_ERROR_CODE);
+    assert_eq!(
+        err.error.message,
+        "`mcpMetaByServer` cannot target `codex_apps`; use `mcpMetaByConnector` to target an individual connector"
+    );
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn turn_start_rejects_unknown_environment_before_starting_turn() -> Result<()> {
     let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -2065,6 +2120,8 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
                 text_elements: Vec::new(),
             }],
             responsesapi_client_metadata: None,
+            mcp_meta_by_server: None,
+            mcp_meta_by_connector: None,
             cwd: Some(first_cwd.clone()),
             runtime_workspace_roots: None,
             approval_policy: Some(codex_app_server_protocol::AskForApproval::Never),
@@ -2107,6 +2164,8 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> {
                 text_elements: Vec::new(),
             }],
             responsesapi_client_metadata: None,
+            mcp_meta_by_server: None,
+            mcp_meta_by_connector: None,
             cwd: Some(second_cwd.clone()),
             runtime_workspace_roots: None,
             approval_policy: Some(codex_app_server_protocol::AskForApproval::Never),

codex-rs/core/src/agent/control_tests.rs

@@ -443,6 +443,8 @@ async fn send_input_submits_user_message() {
             }],
             final_output_json_schema: None,
             responsesapi_client_metadata: None,
+            mcp_meta_by_server: None,
+            mcp_meta_by_connector: None,
             thread_settings: Default::default(),
         },
     );
@@ -597,6 +599,8 @@ async fn spawn_agent_creates_thread_and_sends_prompt() {
             }],
             final_output_json_schema: None,
             responsesapi_client_metadata: None,
+            mcp_meta_by_server: None,
+            mcp_meta_by_connector: None,
             thread_settings: Default::default(),
         },
     );
@@ -769,6 +773,8 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() {
             }],
             final_output_json_schema: None,
             responsesapi_client_metadata: None,
+            mcp_meta_by_server: None,
+            mcp_meta_by_connector: None,
             thread_settings: Default::default(),
         },
     );

codex-rs/core/src/codex_delegate.rs

@@ -192,6 +192,8 @@ pub(crate) async fn run_codex_thread_one_shot(
         items: input,
         final_output_json_schema,
         responsesapi_client_metadata: None,
+        mcp_meta_by_server: None,
+        mcp_meta_by_connector: None,
         thread_settings: Default::default(),
     })
     .await?;

codex-rs/core/src/codex_thread.rs

@@ -238,9 +238,17 @@ impl CodexThread {
         input: Vec<UserInput>,
         expected_turn_id: Option<&str>,
         responsesapi_client_metadata: Option<HashMap<String, String>>,
+        mcp_meta_by_server: Option<HashMap<String, HashMap<String, serde_json::Value>>>,
+        mcp_meta_by_connector: Option<HashMap<String, HashMap<String, serde_json::Value>>>,
     ) -> Result<String, SteerInputError> {
         self.codex
-            .steer_input(input, expected_turn_id, responsesapi_client_metadata)
+            .steer_input(
+                input,
+                expected_turn_id,
+                responsesapi_client_metadata,
+                mcp_meta_by_server,
+                mcp_meta_by_connector,
+            )
             .await
     }
 

codex-rs/core/src/guardian/review_session.rs

@@ -714,6 +714,8 @@ async fn run_review_on_session(
             environments: None,
             final_output_json_schema: Some(params.schema.clone()),
             responsesapi_client_metadata: None,
+            mcp_meta_by_server: None,
+            mcp_meta_by_connector: None,
             thread_settings: codex_protocol::protocol::ThreadSettingsOverrides {
                 #[allow(deprecated)]
                 cwd: Some(params.parent_turn.cwd.to_path_buf()),

codex-rs/core/src/mcp_tool_call.rs

@@ -971,6 +971,18 @@ const MCP_TOOL_UI_RESOURCE_URI_META_KEY: &str = "ui/resourceUri";
 const MCP_TOOL_PLUGIN_ID_META_KEY: &str = "plugin_id";
 const MCP_TOOL_THREAD_ID_META_KEY: &str = "threadId";
 
+fn is_codex_owned_mcp_request_meta_key(key: &str) -> bool {
+    [
+        crate::X_CODEX_TURN_METADATA_HEADER,
+        MCP_TOOL_CODEX_APPS_META_KEY,
+        MCP_TOOL_PLUGIN_ID_META_KEY,
+        MCP_TOOL_THREAD_ID_META_KEY,
+        codex_mcp::MCP_SANDBOX_STATE_META_CAPABILITY,
+        codex_rollout_trace::MCP_CALL_ID_META_KEY,
+    ]
+    .contains(&key)
+}
+
 async fn custom_mcp_tool_approval_mode(
     sess: &Session,
     turn_context: &TurnContext,
@@ -1028,6 +1040,24 @@ fn build_mcp_tool_call_request_meta(
 ) -> Option<serde_json::Value> {
     let mut request_meta = serde_json::Map::new();
 
+    let client_meta = if server == CODEX_APPS_MCP_SERVER_NAME {
+        metadata
+            .and_then(|metadata| metadata.connector_id.as_deref())
+            .and_then(|connector_id| {
+                turn_context
+                    .turn_metadata_state
+                    .mcp_meta_for_connector(connector_id)
+            })
+    } else {
+        turn_context.turn_metadata_state.mcp_meta_for_server(server)
+    };
+    if let Some(meta) = client_meta {
+        request_meta.extend(
+            meta.into_iter()
+                .filter(|(key, _)| !is_codex_owned_mcp_request_meta_key(key)),
+        );
+    }
+
     if let Some(turn_metadata) = turn_context
         .turn_metadata_state
         .current_meta_value_for_mcp_request(McpTurnMetadataContext {