[connectors] Support connectors part 1 - App server & MCP (#9667)

In order to make Codex work with connectors, we add a built-in gateway
MCP that acts as a transparent proxy between the client and the
connectors. The gateway MCP collects actions that are accessible to the
user and sends them down to the user, when a connector action is chosen
to be called, the client invokes the action through the gateway MCP as
well.

 - [x] Add the system built-in gateway MCP to list and run connectors.
 - [x] Add the app server methods and protocol

codex-rs/core/src/codex.rs

@@ -17,6 +17,7 @@ use crate::compact;
 use crate::compact::run_inline_auto_compact_task;
 use crate::compact::should_use_remote_compact_task;
 use crate::compact_remote::run_inline_remote_auto_compact_task;
+use crate::connectors;
 use crate::exec_policy::ExecPolicyManager;
 use crate::features::Feature;
 use crate::features::Features;
@@ -103,7 +104,10 @@ use crate::exec::StreamOutput;
 use crate::exec_policy::ExecPolicyUpdateError;
 use crate::feedback_tags;
 use crate::instructions::UserInstructions;
+use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
 use crate::mcp::auth::compute_auth_statuses;
+use crate::mcp::effective_mcp_servers;
+use crate::mcp::with_codex_apps_mcp;
 use crate::mcp_connection_manager::McpConnectionManager;
 use crate::model_provider_info::CHAT_WIRE_API_DEPRECATION_SUMMARY;
 use crate::project_doc::get_user_instructions;
@@ -647,14 +651,25 @@ impl Session {
         let rollout_fut = RolloutRecorder::new(&config, rollout_params);
 
         let history_meta_fut = crate::message_history::history_metadata(&config);
-        let auth_statuses_fut = compute_auth_statuses(
-            config.mcp_servers.iter(),
-            config.mcp_oauth_credentials_store_mode,
-        );
+        let auth_manager_clone = Arc::clone(&auth_manager);
+        let config_for_mcp = Arc::clone(&config);
+        let auth_and_mcp_fut = async move {
+            let auth = auth_manager_clone.auth().await;
+            let mcp_servers = effective_mcp_servers(&config_for_mcp, auth.as_ref());
+            let auth_statuses = compute_auth_statuses(
+                mcp_servers.iter(),
+                config_for_mcp.mcp_oauth_credentials_store_mode,
+            )
+            .await;
+            (auth, mcp_servers, auth_statuses)
+        };
 
         // Join all independent futures.
-        let (rollout_recorder, (history_log_id, history_entry_count), auth_statuses) =
-            tokio::join!(rollout_fut, history_meta_fut, auth_statuses_fut);
+        let (
+            rollout_recorder,
+            (history_log_id, history_entry_count),
+            (auth, mcp_servers, auth_statuses),
+        ) = tokio::join!(rollout_fut, history_meta_fut, auth_and_mcp_fut);
 
         let rollout_recorder = rollout_recorder.map_err(|e| {
             error!("failed to initialize rollout recorder: {e:#}");
@@ -694,7 +709,6 @@ impl Session {
         }
         maybe_push_chat_wire_api_deprecation(&config, &mut post_session_configured_events);
 
-        let auth = auth_manager.auth().await;
         let auth = auth.as_ref();
         let otel_manager = OtelManager::new(
             conversation_id,
@@ -729,7 +743,7 @@ impl Session {
             config.model_auto_compact_token_limit,
             config.approval_policy.value(),
             config.sandbox_policy.get().clone(),
-            config.mcp_servers.keys().map(String::as_str).collect(),
+            mcp_servers.keys().map(String::as_str).collect(),
             config.active_profile.clone(),
         );
 
@@ -813,7 +827,7 @@ impl Session {
             .write()
             .await
             .initialize(
-                &config.mcp_servers,
+                &mcp_servers,
                 config.mcp_oauth_credentials_store_mode,
                 auth_statuses.clone(),
                 tx_event.clone(),
@@ -1987,6 +2001,14 @@ impl Session {
             }
         };
 
+        let auth = self.services.auth_manager.auth().await;
+        let config = self.get_config().await;
+        let mcp_servers = with_codex_apps_mcp(
+            mcp_servers,
+            self.features.enabled(Feature::Connectors),
+            auth.as_ref(),
+            config.as_ref(),
+        );
         let auth_statuses = compute_auth_statuses(mcp_servers.iter(), store_mode).await;
         let sandbox_state = SandboxState {
             sandbox_policy: turn_context.sandbox_policy.clone(),
@@ -2168,6 +2190,7 @@ mod handlers {
 
     use crate::mcp::auth::compute_auth_statuses;
     use crate::mcp::collect_mcp_snapshot_from_manager;
+    use crate::mcp::effective_mcp_servers;
     use crate::review_prompts::resolve_review_request;
     use crate::tasks::CompactTask;
     use crate::tasks::RegularTask;
@@ -2481,13 +2504,12 @@ mod handlers {
 
     pub async fn list_mcp_tools(sess: &Session, config: &Arc<Config>, sub_id: String) {
         let mcp_connection_manager = sess.services.mcp_connection_manager.read().await;
+        let auth = sess.services.auth_manager.auth().await;
+        let mcp_servers = effective_mcp_servers(config, auth.as_ref());
         let snapshot = collect_mcp_snapshot_from_manager(
             &mcp_connection_manager,
-            compute_auth_statuses(
-                config.mcp_servers.iter(),
-                config.mcp_oauth_credentials_store_mode,
-            )
-            .await,
+            compute_auth_statuses(mcp_servers.iter(), config.mcp_oauth_credentials_store_mode)
+                .await,
         )
         .await;
         let event = Event {
@@ -3001,6 +3023,60 @@ async fn run_auto_compact(sess: &Arc<Session>, turn_context: &Arc<TurnContext>)
     }
 }
 
+fn filter_connectors_for_input(
+    connectors: Vec<connectors::ConnectorInfo>,
+    input: &[ResponseItem],
+) -> Vec<connectors::ConnectorInfo> {
+    let user_messages = collect_user_messages(input);
+    if user_messages.is_empty() {
+        return Vec::new();
+    }
+
+    connectors
+        .into_iter()
+        .filter(|connector| connector_inserted_in_messages(connector, &user_messages))
+        .collect()
+}
+
+fn connector_inserted_in_messages(
+    connector: &connectors::ConnectorInfo,
+    user_messages: &[String],
+) -> bool {
+    let label = connectors::connector_display_label(connector);
+    let needle = label.to_lowercase();
+    let legacy = format!("{label} connector").to_lowercase();
+    user_messages.iter().any(|message| {
+        let message = message.to_lowercase();
+        message.contains(&needle) || message.contains(&legacy)
+    })
+}
+
+fn filter_codex_apps_mcp_tools(
+    mut mcp_tools: HashMap<String, crate::mcp_connection_manager::ToolInfo>,
+    connectors: &[connectors::ConnectorInfo],
+) -> HashMap<String, crate::mcp_connection_manager::ToolInfo> {
+    let allowed: HashSet<&str> = connectors
+        .iter()
+        .map(|connector| connector.connector_id.as_str())
+        .collect();
+
+    mcp_tools.retain(|_, tool| {
+        if tool.server_name != CODEX_APPS_MCP_SERVER_NAME {
+            return true;
+        }
+        let Some(connector_id) = codex_apps_connector_id(tool) else {
+            return false;
+        };
+        allowed.contains(connector_id)
+    });
+
+    mcp_tools
+}
+
+fn codex_apps_connector_id(tool: &crate::mcp_connection_manager::ToolInfo) -> Option<&str> {
+    tool.connector_id.as_deref()
+}
+
 #[instrument(level = "trace",
     skip_all,
     fields(
@@ -3017,7 +3093,7 @@ async fn run_sampling_request(
     input: Vec<ResponseItem>,
     cancellation_token: CancellationToken,
 ) -> CodexResult<SamplingRequestResult> {
-    let mcp_tools = sess
+    let mut mcp_tools = sess
         .services
         .mcp_connection_manager
         .read()
@@ -3025,6 +3101,20 @@ async fn run_sampling_request(
         .list_all_tools()
         .or_cancel(&cancellation_token)
         .await?;
+    let connectors_for_tools = if turn_context
+        .client
+        .config()
+        .features
+        .enabled(Feature::Connectors)
+    {
+        let connectors = connectors::accessible_connectors_from_mcp_tools(&mcp_tools);
+        Some(filter_connectors_for_input(connectors, &input))
+    } else {
+        None
+    };
+    if let Some(connectors) = connectors_for_tools.as_ref() {
+        mcp_tools = filter_codex_apps_mcp_tools(mcp_tools, connectors);
+    }
     let router = Arc::new(ToolRouter::from_config(
         &turn_context.tools_config,
         Some(