fix(responses): emit terminal error event on MCP tool-execution / follow-up failures

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.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
 from litellm.types.llms.openai import (
     BaseLiteLLMOpenAIResponseObject,
+    ErrorEvent,
+    ErrorEventError,
     MCPCallArgumentsDeltaEvent,
     MCPCallArgumentsDoneEvent,
     MCPCallCompletedEvent,
@@ -304,6 +306,14 @@ def __init__(
         # Cache the response ID to ensure consistency across all events
         self._cached_response_id: Optional[str] = None
 
+        # Internal failures (initial LLM call, tool execution, follow-up call)
+        # are stashed here so they can be surfaced to the client as an `error`
+        # stream event, or re-raised before any SSE bytes are written (eager
+        # path in aresponses_api_with_mcp for the initial call).
+        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 @@ def _should_auto_execute_tools(self) -> bool:
 
         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
 
@@ -429,6 +456,12 @@ async def __anext__(self) -> ResponsesAPIStreamingResponse:
                     raise
             else:
                 self.phase = "finished"
+                # Tool execution or the follow-up call failed: emit a terminal
+                # `error` event so the client can distinguish a failed stream
+                # from a completed one.
+                if self._stream_error is not None and not self._error_event_emitted:
+                    self._error_event_emitted = True
+                    return self._make_stream_error_event()
                 raise StopAsyncIteration
 
         # Phase 6: Finished
@@ -451,13 +484,17 @@ async def _handle_initial_response_phase(
             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 +617,11 @@ async def _create_initial_response_iterator(self) -> None:
 
             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"""
@@ -693,11 +733,21 @@ async def _generate_tool_execution_events(self) -> None:
 
             traceback.print_exc()
             self.tool_results = []
+            # Remember the failure. Without this, the follow-up call is made
+            # with function_call items but no function_call_output items and
+            # the provider rejects it with "No tool output found for function
+            # call ...".
+            self._stream_error = e
 
     async def _create_follow_up_iterator(self) -> None:
         """Create the follow-up response iterator with tool results"""
         if not self.collected_response or not hasattr(self, "tool_results"):
             return
+        # Tool execution already failed; skip the doomed follow-up call (it
+        # would be rejected with "No tool output found for function call ...")
+        # and let __anext__ emit the terminal error event.
+        if self._stream_error is not None:
+            return
 
         from litellm.responses.main import aresponses
         from litellm.responses.mcp.litellm_proxy_mcp_handler import (
@@ -738,6 +788,9 @@ async def _create_follow_up_iterator(self) -> None:
 
             traceback.print_exc()
             self.follow_up_iterator = None
+            # Surface via a terminal `error` event in __anext__ instead of
+            # silently ending the stream with no terminal event.
+            self._stream_error = e
 
     def __iter__(self):
         return self