feat(openai): support deferred resolution for configured MCP connectors

apis/src/openai/responses/agentic_loop/mod.rs

@@ -149,6 +149,11 @@ const META_STATUS: &str = "responses.status";
 /// control in `on_response_body` (end-of-stream), writing
 /// `filter_results` for `iterative_request_router` transitions.
 ///
+/// Also extracts `tool_search_call` items into
+/// `ResponsesState.tool_search_calls` so `openai_mcp_dispatch` can
+/// load deferred connectors on the next iteration without forwarding
+/// those items to the inference backend.
+///
 /// # YAML
 ///
 /// ```yaml
@@ -370,6 +375,7 @@ fn end_stream_with_error(
 /// the original client header).
 fn prepare_iteration(ctx: &mut HttpFilterContext<'_>, state: &mut ResponsesState) {
     state.tool_calls.clear();
+    state.tool_search_calls.clear();
     state.web_search_calls.clear();
     state.parallel_tool_calls = false;
     set_request_body_field(state, "parallel_tool_calls", Value::Bool(false));
@@ -406,7 +412,7 @@ fn evaluate_loop_decision(
     body: &mut Option<Bytes>,
     config: &AgenticLoopConfig,
 ) -> Result<FilterAction, FilterError> {
-    if state.tool_calls.is_empty() && state.web_search_calls.is_empty() {
+    if state.tool_calls.is_empty() && state.web_search_calls.is_empty() && state.tool_search_calls.is_empty() {
         trace!("no tool calls, signaling done");
         state.finalize_response_body(body);
         return set_done(ctx);
@@ -503,6 +509,10 @@ fn collect_output_items(response: &Value, state: &mut ResponsesState) {
                 state.web_search_calls.push(item.clone());
                 state.persisted_messages.push(item.clone());
             },
+            Some("tool_search_call") => {
+                state.tool_search_calls.push(item.clone());
+                state.persisted_messages.push(item.clone());
+            },
             _ => {},
         }
     }

apis/src/openai/responses/agentic_loop/tests.rs

@@ -1600,6 +1600,52 @@ fn web_search_call_excluded_from_messages_but_persisted() {
     );
 }
 
+#[test]
+fn tool_search_call_queued_for_deferred_discovery() {
+    let filter = make_filter();
+    let req = make_request(Method::POST, "/v1/responses");
+    let mut ctx = make_filter_context(&req);
+
+    let state = make_state_with_tool_calls(vec![]);
+    ctx.extensions.insert(state);
+
+    let response_body = json!({
+        "id": "resp_1",
+        "object": "response",
+        "output": [
+            {
+                "type": "tool_search_call",
+                "id": "tsc_1",
+                "status": "completed"
+            }
+        ]
+    });
+    let mut body = Some(Bytes::from(serde_json::to_vec(&response_body).unwrap()));
+
+    let action = filter.on_response_body(&mut ctx, &mut body, true).unwrap();
+    assert!(
+        matches!(action, FilterAction::Continue),
+        "tool_search_call should continue so dispatch can loop"
+    );
+
+    let state = ctx.extensions.get::<ResponsesState>().unwrap();
+    assert_eq!(state.tool_search_calls.len(), 1);
+    assert!(
+        state
+            .messages
+            .iter()
+            .all(|item| item.get("type").and_then(Value::as_str) != Some("tool_search_call")),
+        "tool_search_call should not enter backend messages"
+    );
+    assert!(
+        state
+            .persisted_messages
+            .iter()
+            .any(|item| item.get("type").and_then(Value::as_str) == Some("tool_search_call")),
+        "tool_search_call should be persisted"
+    );
+}
+
 #[test]
 fn web_search_call_does_not_count_as_function_call_for_limit() {
     let filter = make_filter();

apis/src/openai/responses/mcp_dispatch/mod.rs

@@ -6,12 +6,14 @@
 //! Operates in two phases within an
 //! `iterative_request_router` inference step:
 //!
-//! 1. **Response path** (`on_response_body`): after `openai_agentic_loop` extracts model-produced function calls,
-//!    identifies calls backed by [`ResponsesState::mcp_tool_map`], checks approval policies, and writes
-//!    `openai_mcp_dispatch.action = "loop"` to filter results.
-//! 2. **Request-body path** (`on_request_body`, next IRR iteration): executes pending MCP calls via
-//!    [`mcp_client::call_tool`] and appends results to `messages`, `persisted_messages`, and `output_items` before
-//!    `openai_responses_proxy` serializes the next inference request.
+//! 1. **Response path** (`on_response_body`): after `openai_agentic_loop` extracts model-produced function calls or
+//!    `tool_search_call` items, identifies MCP-backed work, checks approval policies, and writes
+//!    `openai_mcp_dispatch.action = "loop"` to filter results. Deferred connectors loop with `discover_mcp` until
+//!    `tools/list` can run on the next iteration.
+//! 2. **Request-body path** (`on_request_body`, next IRR iteration): loads deferred connector tools when a
+//!    `tool_search_call` is pending, then executes pending MCP calls via [`mcp_client::call_tool`] and appends results
+//!    to `messages`, `persisted_messages`, and `output_items` before `openai_responses_proxy` serializes the next
+//!    inference request.
 //!
 //! # Pipeline dependencies
 //!
@@ -58,7 +60,12 @@ use self::{
     approval::{parse_approval_policy, requires_approval},
     config::{McpDispatchConfig, build_config},
 };
-use super::{openai_mcp_tool_resolve::encode_function_name, state::ResponsesState};
+use super::{
+    openai_mcp_tool_resolve::{
+        discover_deferred_connectors, encode_function_name, has_pending_deferred_discovery, resolve_error_rejection,
+    },
+    state::ResponsesState,
+};
 use crate::mcp_client;
 
 /// Filter result key consumed by `iterative_request_router`.
@@ -77,6 +84,12 @@ const ACTION_DONE: &str = "done";
 /// Executes MCP tool calls against upstream MCP servers within
 /// the Responses API agentic loop.
 ///
+/// When `openai_mcp_tool_resolve` stored deferred connectors and the
+/// model returns a `tool_search_call`, this filter loads every pending
+/// deferred connector from its internally resolved endpoint on the next
+/// iteration, then dispatches later MCP calls through the existing
+/// `tools/call` path.
+///
 /// # YAML
 ///
 /// ```yaml
@@ -145,6 +158,35 @@ impl McpDispatchFilter {
 
         Ok(FilterAction::Continue)
     }
+
+    /// Execute pending MCP tool calls after optional deferred discovery.
+    async fn execute_pending_mcp_calls(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
+        let Some(state) = ctx.extensions.get::<ResponsesState>() else {
+            return Ok(FilterAction::Continue);
+        };
+        let mcp_calls = extract_mcp_tool_calls(&state.tool_calls, &state.mcp_tool_map);
+        if mcp_calls.is_empty() {
+            return Ok(FilterAction::Continue);
+        }
+
+        debug!(count = mcp_calls.len(), "executing pending MCP tool calls");
+        let parallel = state.parallel_tool_calls;
+        let tool_map = std::sync::Arc::new(state.mcp_tool_map.clone());
+        let results = execute_mcp_calls(&mcp_calls, &tool_map, parallel, self.timeout, self.allow_loopback).await;
+
+        let Some(state) = ctx.extensions.get_mut::<ResponsesState>() else {
+            warn!("ResponsesState missing when appending results");
+            return Ok(FilterAction::Continue);
+        };
+        for result in results {
+            state.messages.push(result.message.clone());
+            state.persisted_messages.push(result.message);
+            state.accumulated_output.push(result.output_item);
+        }
+        let tool_map_ref = &state.mcp_tool_map;
+        state.tool_calls.retain(|tc| !is_mcp_tool_call(tc, tool_map_ref));
+        Ok(FilterAction::Continue)
+    }
 }
 
 #[async_trait]
@@ -190,32 +232,18 @@ impl HttpFilter for McpDispatchFilter {
         let Some(state) = ctx.extensions.get::<ResponsesState>() else {
             return Ok(FilterAction::Continue);
         };
-
-        let mcp_calls = extract_mcp_tool_calls(&state.tool_calls, &state.mcp_tool_map);
-        if mcp_calls.is_empty() {
+        let needs_discovery = has_pending_deferred_discovery(state);
+        let has_calls = !extract_mcp_tool_calls(&state.tool_calls, &state.mcp_tool_map).is_empty();
+        if !has_calls && !needs_discovery {
             return Ok(FilterAction::Continue);
         }
-
-        debug!(count = mcp_calls.len(), "executing pending MCP tool calls");
-
-        let parallel = state.parallel_tool_calls;
-        let tool_map = std::sync::Arc::new(state.mcp_tool_map.clone());
-        let results = execute_mcp_calls(&mcp_calls, &tool_map, parallel, self.timeout, self.allow_loopback).await;
-
-        let Some(state) = ctx.extensions.get_mut::<ResponsesState>() else {
-            warn!("ResponsesState missing when appending results");
-            return Ok(FilterAction::Continue);
-        };
-        for result in results {
-            state.messages.push(result.message.clone());
-            state.persisted_messages.push(result.message);
-            state.accumulated_output.push(result.output_item);
+        if needs_discovery {
+            let action = discover_pending_connectors(ctx).await?;
+            if !matches!(action, FilterAction::Continue) {
+                return Ok(action);
+            }
         }
-
-        let tool_map_ref = &state.mcp_tool_map;
-        state.tool_calls.retain(|tc| !is_mcp_tool_call(tc, tool_map_ref));
-
-        Ok(FilterAction::Continue)
+        self.execute_pending_mcp_calls(ctx).await
     }
 
     fn on_response_body(
@@ -233,11 +261,18 @@ impl HttpFilter for McpDispatchFilter {
         };
 
         let mcp_calls = extract_mcp_tool_calls(&state.tool_calls, &state.mcp_tool_map);
-        if mcp_calls.is_empty() {
+        let needs_discovery = has_pending_deferred_discovery(state);
+        if mcp_calls.is_empty() && !needs_discovery {
             set_action(ctx, ACTION_DONE)?;
             return Ok(FilterAction::Continue);
         }
 
+        if needs_discovery && mcp_calls.is_empty() {
+            ctx.set_metadata("openai_mcp_dispatch.action".to_owned(), "discover_mcp".to_owned());
+            set_action(ctx, ACTION_LOOP)?;
+            return Ok(FilterAction::Continue);
+        }
+
         if let Some(pending) = find_approval_required(&mcp_calls, &state.mcp_tool_map) {
             return Self::handle_approval_required(ctx, body, &pending);
         }
@@ -249,6 +284,23 @@ impl HttpFilter for McpDispatchFilter {
     }
 }
 
+/// Load deferred connector tools when a `tool_search_call` is pending.
+async fn discover_pending_connectors(ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
+    let Some(state) = ctx.extensions.get_mut::<ResponsesState>() else {
+        warn!("ResponsesState missing when discovering deferred MCP connectors");
+        return Ok(FilterAction::Continue);
+    };
+    match discover_deferred_connectors(state).await {
+        Ok(()) => Ok(FilterAction::Continue),
+        Err(err) => {
+            let streaming = ctx
+                .get_metadata("openai_responses_format.stream")
+                .is_some_and(|v| v == "true");
+            Ok(resolve_error_rejection(&err, streaming))
+        },
+    }
+}
+
 /// Publish the dispatch decision for IRR transition evaluation.
 fn set_action(ctx: &mut HttpFilterContext<'_>, action: &'static str) -> Result<(), FilterError> {
     ctx.filter_results

apis/src/openai/responses/mcp_dispatch/tests.rs

@@ -21,7 +21,7 @@ use crate::{
             approval::{ApprovalPolicy, parse_approval_policy, requires_approval},
             config::{McpDispatchConfig, build_config},
         },
-        state::ResponsesState,
+        state::{DeferredMcpConnector, ResponsesState},
     },
     test_utils::{make_filter_context, make_request},
 };
@@ -1029,6 +1029,39 @@ fn on_response_body_with_mcp_calls_sets_execute_metadata() {
     assert_dispatch_action(&ctx, "loop");
 }
 
+#[test]
+fn on_response_body_deferred_tool_search_sets_loop() {
+    let filter = make_dispatch_filter();
+    let req = make_request(http::Method::POST, "/v1/responses");
+    let mut ctx = make_filter_context(&req);
+    let state = ResponsesState {
+        deferred_mcp: vec![DeferredMcpConnector {
+            allow_loopback: true,
+            authorization: None,
+            allowed_tools: None,
+            connector_id: "corp_drive".to_owned(),
+            headers: None,
+            max_rewritten_body_bytes: 67_108_864,
+            max_tools: 128,
+            require_approval: None,
+            server_label: "drive".to_owned(),
+            server_url: "https://drive.example.com/mcp".to_owned(),
+            timeout: std::time::Duration::from_secs(5),
+        }],
+        tool_search_calls: vec![json!({"type": "tool_search_call", "id": "tsc_1"})],
+        ..ResponsesState::default()
+    };
+    ctx.extensions.insert(state);
+    let mut body = None;
+    let result = filter.on_response_body(&mut ctx, &mut body, true).unwrap();
+    assert!(matches!(result, FilterAction::Continue));
+    assert_eq!(
+        ctx.filter_metadata.get("openai_mcp_dispatch.action"),
+        Some(&"discover_mcp".to_owned())
+    );
+    assert_dispatch_action(&ctx, "loop");
+}
+
 #[test]
 fn on_response_body_approval_required_sets_done_metadata() {
     let filter = make_dispatch_filter();