fix(responses): surface MCP gateway initial-call failures instead of emitting a broken stream

When the initial LLM call inside MCPEnhancedStreamingIterator fails (e.g.
an invalid previous_response_id -> provider 400 'No tool output found for
function call ...'), the proxy returned HTTP 200 and the stream emitted the
pre-generated mcp_list_tools discovery events with no response.created
before them. That violates the Responses API streaming contract and crashes
SDK stream accumulators (openai-node: "expected 'response.created' event,
got response.mcp_list_tools.in_progress").

- aresponses_api_with_mcp now makes the initial call eagerly, before any
  SSE bytes are written, and re-raises the stashed failure so the client
  gets a real 4xx/5xx with the provider error body.
- If a creation failure still surfaces during iteration, the stream emits a
  single terminal 'error' event instead of discovery events.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

litellm/proxy/dev_config.yaml

@@ -211,3 +211,8 @@ litellm_settings:
     sandbox_tool_name: e2b_sandbox
   callbacks:
     - code_interpreter_interception
+
+mcp_servers:
+  deepwiki:
+    url: "https://mcp.deepwiki.com/mcp"
+    transport: "http"

litellm/responses/main.py

@@ -261,7 +261,7 @@ async def aresponses_api_with_mcp(
             pre_processed_mcp_tools=original_mcp_tools,
         )
 
-        return LiteLLM_Proxy_MCP_Handler._create_mcp_streaming_response(
+        mcp_streaming_response = LiteLLM_Proxy_MCP_Handler._create_mcp_streaming_response(
             input=input,
             model=model,
             all_tools=all_tools,
@@ -272,6 +272,16 @@ async def aresponses_api_with_mcp(
             tool_server_map=tool_server_map,
             **kwargs,
         )
+        # Make the initial LLM call eagerly, before any SSE bytes are written,
+        # so a pre-stream failure (e.g. an invalid previous_response_id ->
+        # provider 400 "No tool output found for function call ...") surfaces
+        # as a normal HTTP error instead of an HTTP 200 whose stream emits
+        # mcp_list_tools events with no response.created (which crashes SDK
+        # stream accumulators).
+        await mcp_streaming_response._create_initial_response_iterator()
+        if mcp_streaming_response._initial_creation_error is not None:
+            raise mcp_streaming_response._initial_creation_error
+        return mcp_streaming_response
 
     # Determine if we should auto-execute tools
     should_auto_execute = bool(mcp_tools_with_litellm_proxy) and LiteLLM_Proxy_MCP_Handler._should_auto_execute_tools(

litellm/responses/mcp/mcp_streaming_iterator.py

@@ -5,6 +5,8 @@ from litellm._uuid import uuid
 from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
 from litellm.types.llms.openai import (
     BaseLiteLLMOpenAIResponseObject,
+    ErrorEvent,
+    ErrorEventError,
     MCPCallArgumentsDeltaEvent,
     MCPCallArgumentsDoneEvent,
     MCPCallCompletedEvent,
@@ -304,6 +306,14 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
         # Cache the response ID to ensure consistency across all events
         self._cached_response_id: Optional[str] = None
 
+        # Initial-LLM-call failures are stashed here so they can be surfaced
+        # to the client as an `error` stream event (lazy path) or re-raised
+        # before any SSE bytes are written (eager path in
+        # aresponses_api_with_mcp).
+        self._initial_creation_error: Optional[Exception] = None
+        self._stream_error: Optional[Exception] = None
+        self._error_event_emitted = False
+
     def _extract_mcp_headers_from_params(self) -> None:
         """Extract MCP headers from original request params to pass to tool calls"""
         from typing import Dict, Optional
@@ -368,6 +378,23 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
 
         return LiteLLM_Proxy_MCP_Handler._should_auto_execute_tools(self.mcp_tools_with_litellm_proxy)
 
+    def _make_stream_error_event(self) -> ResponsesAPIStreamingResponse:
+        """Build an OpenAI-style `error` stream event from the stashed internal
+        failure, so clients receive a real terminal error instead of a stream
+        that silently ends mid-flow."""
+        err = self._stream_error
+        status_code = getattr(err, "status_code", None)
+        return ErrorEvent(
+            type=ResponsesAPIStreamEvents.ERROR,
+            sequence_number=1,
+            error=ErrorEventError(
+                type="mcp_gateway_error",
+                code=str(status_code) if status_code is not None else "internal_error",
+                message=str(err) if err is not None else "MCP gateway stream failed",
+                param=None,
+            ),
+        )
+
     def __aiter__(self):
         return self
 
@@ -451,13 +478,17 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
             await self._create_initial_response_iterator()
 
         if self.base_iterator is None:
-            # LLM call failed — still emit MCP discovery events before finishing
-            if self.mcp_discovery_events:
-                self.phase = "mcp_discovery"
-            else:
-                self.phase = "finished"
-                raise StopAsyncIteration
-            return None
+            # The initial LLM call failed. Do NOT emit MCP discovery events: a
+            # stream that starts with mcp_list_tools events and no
+            # response.created violates the Responses API streaming contract
+            # and crashes SDK stream accumulators (openai-node: "expected
+            # 'response.created' event, got response.mcp_list_tools.in_progress").
+            # Surface the failure as an `error` event instead.
+            self.phase = "finished"
+            if self._stream_error is not None:
+                self._error_event_emitted = True
+                return self._make_stream_error_event()
+            raise StopAsyncIteration
 
         if self.base_iterator:
             if hasattr(self.base_iterator, "__anext__"):
@@ -580,8 +611,11 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
 
             traceback.print_exc()
             self.base_iterator = None
-            # Don't set phase to "finished" here — let __anext__ emit any
-            # pre-generated MCP discovery events before ending the iteration.
+            # Stash the failure so aresponses_api_with_mcp can re-raise it
+            # before any SSE bytes are written (eager creation), or so
+            # __anext__ can emit an `error` event instead of ending silently.
+            self._initial_creation_error = e
+            self._stream_error = e
 
     async def _generate_tool_execution_events(self) -> None:
         """Generate tool execution events and execute tools"""

tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py

@@ -0,0 +1,145 @@
+import sys
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+import litellm
+from litellm.responses.mcp.mcp_streaming_iterator import MCPEnhancedStreamingIterator
+from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesAPIStreamEvents
+
+# `litellm.__init__` re-exports a function named `responses`, which shadows the
+# `litellm.responses` subpackage as an attribute — `import litellm.responses.main`
+# can resolve to the unrelated third-party `responses` package instead. Look the
+# real submodule up in sys.modules directly to sidestep the shadowing.
+responses_main_module = sys.modules["litellm.responses.main"]
+
+
+class _FakeAsyncStream:
+    """Minimal async iterator yielding pre-built chunks, one per __anext__ call."""
+
+    def __init__(self, chunks):
+        self._chunks = list(chunks)
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if not self._chunks:
+            raise StopAsyncIteration
+        return self._chunks.pop(0)
+
+
+def _completed_chunk(output):
+    response = ResponsesAPIResponse(id="resp-1", created_at=0, output=output)
+    return SimpleNamespace(type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=response)
+
+
+def _text_message(text: str):
+    return {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": text}]}
+
+
+def _text_only_stream(text: str) -> _FakeAsyncStream:
+    return _FakeAsyncStream([_completed_chunk([_text_message(text)])])
+
+
+def _make_lazy_iterator(mcp_events=None) -> MCPEnhancedStreamingIterator:
+    """Iterator with no base_iterator: the initial LLM call happens lazily on iteration."""
+    return MCPEnhancedStreamingIterator(
+        base_iterator=None,
+        mcp_events=list(mcp_events or []),
+        tool_server_map={"read_wiki_contents": "deepwiki"},
+        mcp_tools_with_litellm_proxy=[{"require_approval": "never"}],
+        user_api_key_auth=None,
+        original_request_params={
+            "model": "gpt-4",
+            "input": "what is berriai/litellm?",
+            "tools": [{"type": "mcp"}],
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_initial_call_failure_emits_error_event_not_discovery_events(monkeypatch):
+    """
+    Regression test: when the initial LLM call fails (e.g. an invalid
+    previous_response_id -> provider 400 "No tool output found for function
+    call ..."), the stream used to emit the pre-generated mcp_list_tools
+    discovery events with no response.created before them — which violates
+    the Responses API streaming contract and crashes SDK stream accumulators
+    (openai-node: "expected 'response.created' event, got
+    response.mcp_list_tools.in_progress"). The stream must instead surface a
+    single terminal `error` event and end.
+    """
+    aresponses_mock = AsyncMock(
+        side_effect=litellm.BadRequestError(
+            message="No tool output found for function call call_x.",
+            model="gpt-4",
+            llm_provider="openai",
+        )
+    )
+    monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
+
+    discovery_event = SimpleNamespace(type=ResponsesAPIStreamEvents.MCP_LIST_TOOLS_IN_PROGRESS)
+    iterator = _make_lazy_iterator(mcp_events=[discovery_event])
+
+    chunks = [chunk async for chunk in iterator]
+
+    assert len(chunks) == 1
+    error_event = chunks[0]
+    assert error_event.type == ResponsesAPIStreamEvents.ERROR
+    assert error_event.error.code == "400"
+    assert "No tool output found" in error_event.error.message
+    # No discovery events leaked before/after the error.
+    assert discovery_event not in chunks
+
+
+@pytest.mark.asyncio
+async def test_eager_creation_reraises_pre_stream_failure_as_http_error(monkeypatch):
+    """
+    aresponses_api_with_mcp creates the initial response eagerly and re-raises
+    the stashed creation failure, so the proxy returns a real 4xx/5xx before
+    any SSE bytes are written instead of an HTTP 200 with a broken stream.
+    """
+    from litellm.responses.mcp.litellm_proxy_mcp_handler import LiteLLM_Proxy_MCP_Handler
+
+    monkeypatch.setattr(
+        LiteLLM_Proxy_MCP_Handler,
+        "_process_mcp_tools_without_openai_transform",
+        AsyncMock(return_value=([], {})),
+    )
+    boom = litellm.BadRequestError(
+        message="Previous response with id 'resp_bogus' not found.",
+        model="gpt-4",
+        llm_provider="openai",
+    )
+    monkeypatch.setattr(responses_main_module, "aresponses", AsyncMock(side_effect=boom))
+
+    with pytest.raises(litellm.BadRequestError) as excinfo:
+        await responses_main_module.aresponses_api_with_mcp(
+            input="hi",
+            model="gpt-4",
+            stream=True,
+            previous_response_id="resp_bogus",
+            tools=[{"type": "mcp", "server_url": "litellm_proxy/mcp/deepwiki", "require_approval": "never"}],
+        )
+
+    assert "resp_bogus" in str(excinfo.value)
+
+
+@pytest.mark.asyncio
+async def test_initial_call_success_does_not_emit_error_event(monkeypatch):
+    """Happy path is unchanged: no error event, stream flows as before."""
+    monkeypatch.setattr(
+        responses_main_module,
+        "aresponses",
+        AsyncMock(return_value=_text_only_stream("all good")),
+    )
+
+    iterator = _make_lazy_iterator()
+    chunks = [chunk async for chunk in iterator]
+
+    assert all(getattr(c, "type", None) != ResponsesAPIStreamEvents.ERROR for c in chunks)
+    completed = [c for c in chunks if getattr(c, "type", None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED]
+    assert len(completed) == 1
+    assert iterator._initial_creation_error is None