Merge pull request #40121 from Atharva-Kanherkar/fix/mcp-responses-stream-single-lifecycle

fix(responses): stream one lifecycle across MCP auto-execute rounds

litellm/responses/mcp/mcp_streaming_iterator.py

@@ -35,6 +35,29 @@ else:
 MAX_MCP_TOOL_CALL_ROUNDS: Final = 5
 
 
+def _output_items(response: ResponsesAPIResponse) -> Sequence[object]:
+    """Read a response's output items as plain objects; the field is a wide union of item models."""
+    return tuple(cast("Sequence[object]", response.output))  # cast-ok: items are only carried, never inspected
+
+
+def _function_call_id(item: object) -> str | None:
+    """The call id of a function_call item, None for every other item kind."""
+    item_type: Final[object] = item.get("type") if isinstance(item, dict) else getattr(item, "type", None)
+    if item_type != "function_call":
+        return None
+    call_id: Final[object] = (
+        item.get("call_id") or item.get("id")
+        if isinstance(item, dict)
+        else getattr(item, "call_id", None) or getattr(item, "id", None)
+    )
+    return call_id if isinstance(call_id, str) else None
+
+
+def _set_event_field(event: ResponsesAPIStreamingResponse, name: str, value: object) -> None:
+    """Events are pydantic models with extra fields allowed, so any event type can carry the field."""
+    setattr(event, name, value)
+
+
 async def create_mcp_list_tools_events(
     mcp_tools_with_litellm_proxy: Sequence[Mapping[str, object]],
     user_api_key_auth: "UserAPIKeyAuth | None",
@@ -171,6 +194,7 @@ def create_mcp_call_events(
     result: str | None = None,
     base_item_id: str | None = None,
     sequence_start: int = 1,
+    output_index: int = 0,
 ) -> list[ResponsesAPIStreamingResponse]:
     """Create MCP call events following OpenAI's specification"""
     events: Final[list[ResponsesAPIStreamingResponse]] = []
@@ -180,7 +204,7 @@ def create_mcp_call_events(
     in_progress_event: Final = MCPCallInProgressEvent(
         type=ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS,
         sequence_number=sequence_start,
-        output_index=0,
+        output_index=output_index,
         item_id=item_id,
     )
     events.append(in_progress_event)
@@ -188,7 +212,7 @@ def create_mcp_call_events(
     # MCP call arguments delta event (streaming the arguments)
     arguments_delta_event: Final = MCPCallArgumentsDeltaEvent(
         type=ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DELTA,
-        output_index=0,
+        output_index=output_index,
         item_id=item_id,
         delta=arguments,  # JSON string with arguments
         sequence_number=sequence_start + 1,
@@ -198,7 +222,7 @@ def create_mcp_call_events(
     # MCP call arguments done event
     arguments_done_event: Final = MCPCallArgumentsDoneEvent(
         type=ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DONE,
-        output_index=0,
+        output_index=output_index,
         item_id=item_id,
         arguments=arguments,  # Complete JSON string with finalized arguments
         sequence_number=sequence_start + 2,
@@ -211,7 +235,7 @@ def create_mcp_call_events(
             type=ResponsesAPIStreamEvents.MCP_CALL_COMPLETED,
             sequence_number=sequence_start + 3,
             item_id=item_id,
-            output_index=0,
+            output_index=output_index,
         )
         events.append(completed_event)
 
@@ -220,7 +244,7 @@ def create_mcp_call_events(
 
         output_item_done_event: Final = OutputItemDoneEvent(
             type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
-            output_index=0,
+            output_index=output_index,
             item=BaseLiteLLMOpenAIResponseObject(
                 **{
                     "id": item_id,
@@ -240,7 +264,7 @@ def create_mcp_call_events(
             type=ResponsesAPIStreamEvents.MCP_CALL_FAILED,
             sequence_number=sequence_start + 3,
             item_id=item_id,
-            output_index=0,
+            output_index=output_index,
         )
         events.append(failed_event)
 
@@ -331,6 +355,12 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
         self._error_event_emitted = False
         self._last_sequence_number = 0
 
+        self._round_index = 0
+        self._output_index_offset = 0
+        self._round_max_output_index = -1
+        self._composed_output: list[object] = []  # mutable-ok: grows as each round finishes
+        self._pending_mcp_call_items: list[dict[str, object]] = []  # mutable-ok: grows per executed tool
+
     def _extract_mcp_headers_from_params(self) -> None:
         """Extract MCP headers from original request params to pass to tool calls"""
 
@@ -416,8 +446,12 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
     async def __anext__(self) -> ResponsesAPIStreamingResponse:
         chunk: Final = await self._anext_impl()
         sequence_number: Final = getattr(chunk, "sequence_number", None)
-        if isinstance(sequence_number, int) and sequence_number > self._last_sequence_number:
-            self._last_sequence_number = sequence_number
+        if isinstance(sequence_number, int):
+            if sequence_number <= self._last_sequence_number and self._last_sequence_number > 0:
+                self._last_sequence_number += 1
+                _set_event_field(chunk, "sequence_number", self._last_sequence_number)
+            else:
+                self._last_sequence_number = max(self._last_sequence_number, sequence_number)
         return chunk
 
     async def _anext_impl(self) -> ResponsesAPIStreamingResponse:
@@ -473,7 +507,7 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
             await self._create_follow_up_iterator()
             if self.base_iterator is not None:
                 self.phase = "continue_initial_response"
-                return await self.__anext__()
+                return await self._anext_impl()
             self.phase = "finished"
             if self._stream_error is not None and not self._error_event_emitted:
                 self._error_event_emitted = True
@@ -531,17 +565,9 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
                         if chunk_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED:
                             self.initial_events_emitted = True
                             self.phase = "mcp_discovery"
-                            return chunk
+                            return await self._compose_round_chunk(chunk)
 
-                    # If auto-execution is enabled, check for completed responses
-                    if self.should_auto_execute and self._is_response_completed(chunk):
-                        response_obj = getattr(chunk, "response", None)
-                        if isinstance(response_obj, ResponsesAPIResponse):
-                            self.collected_response = response_obj
-                        self.phase = "tool_execution"
-                        await self._generate_tool_execution_events()
-
-                    return chunk
+                    return await self._compose_round_chunk(chunk)
                 except StopAsyncIteration:
                     if self.should_auto_execute and self.collected_response:
                         self.phase = "tool_execution"
@@ -567,6 +593,77 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
         chunk_type: Final[object] = getattr(chunk, "type", None)
         return chunk_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED
 
+    def _follow_up_pending(self) -> bool:
+        """True when the current round's tool calls were executed and a follow-up round will run."""
+        return self.collected_response is not None and self.collected_response is self._tool_results_for_response
+
+    def _round_output_width(self, response: ResponsesAPIResponse) -> int:
+        """How many output indexes this round used, counting items it streamed but never listed."""
+        return max(len(_output_items(response)), self._round_max_output_index + 1)
+
+    def _absorb_round(self, response: ResponsesAPIResponse) -> None:
+        """Bank a finished round's items, each function_call the gateway answered replaced by its mcp_call."""
+        width: Final = self._round_output_width(response)
+        answered_call_ids: Final = frozenset(
+            call_id for result in self.tool_results if (call_id := result.get("tool_call_id")) is not None
+        )
+        self._composed_output.extend(
+            item for item in _output_items(response) if _function_call_id(item) not in answered_call_ids
+        )
+        self._composed_output.extend(self._pending_mcp_call_items)
+        self._output_index_offset += width + len(self._pending_mcp_call_items)
+        self._pending_mcp_call_items.clear()
+        self._round_max_output_index = -1
+
+    async def _compose_round_chunk(self, chunk: ResponsesAPIStreamingResponse) -> ResponsesAPIStreamingResponse | None:
+        """
+        Fold one round's event into the single public lifecycle.
+
+        Returns None when the event must not reach the client: the lifecycle
+        openers of a follow-up round, and the response.completed of a round
+        whose tool calls the gateway executes itself. Shifts output_index on
+        follow-up rounds past the items already emitted, and lists every
+        round's items on the final response.completed.
+        """
+        chunk_type: Final[object] = getattr(chunk, "type", None)
+        if self._round_index > 0 and chunk_type in (
+            ResponsesAPIStreamEvents.RESPONSE_CREATED,
+            ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS,
+        ):
+            return None
+
+        output_index: Final[object] = getattr(chunk, "output_index", None)
+        if isinstance(output_index, int):
+            self._round_max_output_index = max(self._round_max_output_index, output_index)
+            if self._output_index_offset:
+                _set_event_field(chunk, "output_index", output_index + self._output_index_offset)
+
+        if not (self.should_auto_execute and self._is_response_completed(chunk)):
+            return chunk
+
+        response_obj: Final[object] = getattr(chunk, "response", None)
+        if isinstance(response_obj, ResponsesAPIResponse):
+            self.collected_response = response_obj
+        # Move to tool execution phase after this chunk
+        self.phase = "tool_execution"
+        await self._generate_tool_execution_events()
+
+        if not isinstance(response_obj, ResponsesAPIResponse):
+            return chunk
+        if self._follow_up_pending():
+            self._absorb_round(response_obj)
+            return None
+        if self._composed_output:
+            merged_output: Final[list[object]] = [  # mutable-ok: the response model declares output as a list
+                *self._composed_output,
+                *_output_items(response_obj),
+            ]
+            merged_response: Final = response_obj.model_copy(
+                update={"output": merged_output}  # mutable-ok: pydantic's update argument must be a dict
+            )
+            _set_event_field(chunk, "response", merged_response)
+        return chunk
+
     async def _process_base_iterator_chunk(self) -> ResponsesAPIStreamingResponse:
         """
         Process a chunk from the base iterator with response ID consistency enforcement.
@@ -594,17 +691,10 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
                     )
                     response_obj.id = self._cached_response_id
 
-        # If auto-execution is enabled, check for completed responses
-        if self.should_auto_execute and self._is_response_completed(chunk):
-            # Collect the response for tool execution
-            response_obj = getattr(chunk, "response", None)
-            if isinstance(response_obj, ResponsesAPIResponse):
-                self.collected_response = response_obj
-            # Move to tool execution phase after emitting this chunk
-            self.phase = "tool_execution"
-            await self._generate_tool_execution_events()
-
-        return chunk
+        composed: Final = await self._compose_round_chunk(chunk)
+        if composed is None:
+            return await self._anext_impl()
+        return composed
 
     async def _create_initial_response_iterator(self) -> None:
         """Create the initial response iterator by making the first LLM call"""
@@ -668,6 +758,12 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
                 return
             self.tool_call_round += 1
 
+            from litellm.types.llms.openai import OutputItemAddedEvent
+
+            next_output_index = self._output_index_offset + self._round_output_width(  # rebind-ok: advances per item
+                self.collected_response
+            )
+            call_items: Final[dict[str, tuple[str, int]]] = {}  # mutable-ok: filled per tool call as events queue
             for tool_call in tool_calls:
                 (
                     tool_name,
@@ -675,14 +771,36 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
                     tool_call_id,
                 ) = LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call)
                 if tool_name and tool_call_id:
+                    item_id = f"mcp_{uuid.uuid4().hex[:8]}"
+                    output_index = next_output_index
+                    next_output_index += 1
+                    call_items[tool_call_id] = (item_id, output_index)
+                    self.tool_execution_events.append(
+                        OutputItemAddedEvent.model_validate(
+                            {  # mutable-ok: consumed once by model_validate
+                                "type": ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
+                                "sequence_number": len(self.tool_execution_events) + 1,
+                                "output_index": output_index,
+                                "item": {  # mutable-ok: consumed once by model_validate
+                                    "id": item_id,
+                                    "type": "mcp_call",
+                                    "status": "in_progress",
+                                    "arguments": tool_arguments or "{}",
+                                    "name": tool_name,
+                                    "server_label": "litellm",
+                                },
+                            }
+                        )
+                    )
                     # Create MCP call events for this tool execution
                     call_events = create_mcp_call_events(
                         tool_name=tool_name,
                         tool_call_id=tool_call_id,
                         arguments=tool_arguments or "{}",  # JSON string with arguments
                         result=None,  # Will be set after execution
-                        base_item_id=f"mcp_{uuid.uuid4().hex[:8]}",
+                        base_item_id=item_id,
                         sequence_start=len(self.tool_execution_events) + 1,
+                        output_index=output_index,
                     )
                     # Add the in_progress and arguments events (not the completed event yet)
                     self.tool_execution_events.extend(call_events[:-1])
@@ -721,37 +839,45 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
                         tool_arguments = args or "{}"
                         break
 
-                item_id = f"mcp_{uuid.uuid4().hex[:8]}"
+                if tool_call_id in call_items:
+                    item_id, output_index = call_items[tool_call_id]
+                else:
+                    item_id = f"mcp_{uuid.uuid4().hex[:8]}"
+                    output_index = next_output_index
+                    next_output_index += 1
 
                 # Create the completion event
                 completed_event = MCPCallCompletedEvent(
                     type=ResponsesAPIStreamEvents.MCP_CALL_COMPLETED,
                     sequence_number=len(self.tool_execution_events) + 1,
                     item_id=item_id,
-                    output_index=0,
+                    output_index=output_index,
                 )
                 self.tool_execution_events.append(completed_event)
 
                 # Create output_item.done event with the tool call result
                 from litellm.types.llms.openai import OutputItemDoneEvent
 
+                mcp_call_item = BaseLiteLLMOpenAIResponseObject(
+                    **{  # mutable-ok: consumed once by the model constructor
+                        "id": item_id,
+                        "type": "mcp_call",
+                        "status": "completed",
+                        "approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}",
+                        "arguments": tool_arguments,
+                        "error": None,
+                        "name": tool_name,
+                        "output": result_text,
+                        "server_label": "litellm",  # or extract from tool config
+                    }
+                )
                 output_item_done_event = OutputItemDoneEvent(
                     type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
-                    output_index=0,
-                    item=BaseLiteLLMOpenAIResponseObject(
-                        **{
-                            "id": item_id,
-                            "type": "mcp_call",
-                            "approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}",
-                            "arguments": tool_arguments,
-                            "error": None,
-                            "name": tool_name,
-                            "output": result_text,
-                            "server_label": "litellm",  # or extract from tool config
-                        }
-                    ),
+                    output_index=output_index,
+                    item=mcp_call_item,
                 )
                 self.tool_execution_events.append(output_item_done_event)
+                self._pending_mcp_call_items.append(mcp_call_item.model_dump())
 
             # Store tool results for follow-up call
             self.tool_results = tool_results
@@ -826,6 +952,7 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
                 self.base_iterator = follow_up_response
                 self.collected_response = None
                 self._cached_response_id = None
+                self._round_index += 1
 
         except Exception as e:
             verbose_logger.error("Error creating follow-up iterator: %s", e)

tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py

@@ -11,7 +11,11 @@ from litellm.responses.mcp.mcp_streaming_iterator import (
     MAX_MCP_TOOL_CALL_ROUNDS,
     MCPEnhancedStreamingIterator,
 )
-from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesAPIStreamEvents
+from litellm.types.llms.openai import (
+    BaseLiteLLMOpenAIResponseObject,
+    ResponsesAPIResponse,
+    ResponsesAPIStreamEvents,
+)
 
 # `litellm.__init__` re-exports a function named `responses`, which shadows the
 # `litellm.responses` subpackage as an attribute — `import litellm.responses.main`
@@ -57,6 +61,10 @@ def _text_message(text: str):
     return {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": text}]}
 
 
+def _item_type(item: dict[str, object] | BaseLiteLLMOpenAIResponseObject) -> str:
+    return str(item["type"]) if isinstance(item, dict) else str(item.type)
+
+
 def _tool_call_stream(call_id: str, tool_name: str, response_id: str = "resp-1") -> _FakeAsyncStream:
     return _FakeAsyncStream([_completed_chunk([_function_call(call_id, tool_name)], response_id=response_id)])
 
@@ -136,11 +144,14 @@ async def test_second_round_tool_call_is_executed_and_reaches_final_text(monkeyp
     assert iterator.tool_call_round == 2
 
     # The stream reached round 3 and produced the final text response instead
-    # of stopping after round 1 or round 2.
+    # of stopping after round 1 or round 2. The client sees one lifecycle whose
+    # final output lists every round's items in order, each executed call as
+    # the gateway's mcp_call rather than the function_call the model emitted.
     completed_chunks = [c for c in chunks if getattr(c, "type", None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED]
-    assert len(completed_chunks) == 3
+    assert len(completed_chunks) == 1
     final_output = completed_chunks[-1].response.output
-    assert final_output[0]["content"][0]["text"] == "Here's what I found after retrying."
+    assert [_item_type(item) for item in final_output] == ["mcp_call", "mcp_call", "message"]
+    assert final_output[-1]["content"][0]["text"] == "Here's what I found after retrying."
 
 
 @pytest.mark.asyncio
@@ -209,7 +220,7 @@ async def test_continuation_id_is_final_round_not_interim_tool_call(monkeypatch)
     chunks = [chunk async for chunk in iterator]
     completed = [c for c in chunks if getattr(c, "type", None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED]
 
-    assert completed[-1].response.output[0]["content"][0]["text"] == "The first item is Alpha."
+    assert completed[-1].response.output[-1]["content"][0]["text"] == "The first item is Alpha."
     assert completed[-1].response.id == "resp-final"
     assert completed[-1].response.id != "resp-interim"
 
@@ -282,7 +293,9 @@ async def test_streaming_follow_up_replays_reasoning_when_store_is_false(monkeyp
         base_iterator=_FakeAsyncStream(
             [
                 _output_item_added_chunk(),
-                _completed_chunk([_reasoning_item("gAAAAA-opaque-blob"), _function_call("call_1", "read_wiki_contents")]),
+                _completed_chunk(
+                    [_reasoning_item("gAAAAA-opaque-blob"), _function_call("call_1", "read_wiki_contents")]
+                ),
             ]
         ),
         mcp_events=[],
@@ -318,7 +331,9 @@ async def test_streaming_follow_up_keeps_previous_response_id_when_stored(monkey
         base_iterator=_FakeAsyncStream(
             [
                 _output_item_added_chunk(),
-                _completed_chunk([_reasoning_item("gAAAAA-opaque-blob"), _function_call("call_1", "read_wiki_contents")]),
+                _completed_chunk(
+                    [_reasoning_item("gAAAAA-opaque-blob"), _function_call("call_1", "read_wiki_contents")]
+                ),
             ]
         ),
         mcp_events=[],
@@ -338,3 +353,138 @@ async def test_streaming_follow_up_keeps_previous_response_id_when_stored(monkey
     follow_up_kwargs = aresponses_mock.call_args_list[0].kwargs
     assert follow_up_kwargs["previous_response_id"] == "resp_prev"
     assert not [item for item in follow_up_kwargs["input"] if item.get("type") == "reasoning"]
+
+
+def _event(event_type: ResponsesAPIStreamEvents, **fields: object) -> SimpleNamespace:
+    return SimpleNamespace(type=event_type, **fields)
+
+
+def _lifecycle_round(response_id: str, item: dict[str, object], sequence_start: int = 0) -> list[SimpleNamespace]:
+    """One upstream Responses round as a provider streams it: its own id, indexes from 0, numbering from 0."""
+    return [
+        _event(
+            ResponsesAPIStreamEvents.RESPONSE_CREATED,
+            response=ResponsesAPIResponse(id=response_id, created_at=0, output=[]),
+            sequence_number=sequence_start,
+        ),
+        _event(
+            ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, output_index=0, item=item, sequence_number=sequence_start + 1
+        ),
+        _event(
+            ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=0, item=item, sequence_number=sequence_start + 2
+        ),
+        _completed_chunk([item], response_id=response_id),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_auto_execute_rounds_share_one_public_lifecycle(monkeypatch):
+    """
+    Every auto-execute round is a distinct upstream response, but the client
+    reads one stream. It must see one response.created, one response.completed,
+    and no output_index reused for a different item, otherwise accumulating
+    clients such as the OpenAI SDK's responses.stream() abort mid-stream.
+    """
+    _mock_mcp_environment(monkeypatch)
+
+    follow_up = _FakeAsyncStream(_lifecycle_round("resp-final", _text_message("Alpha.")))
+    monkeypatch.setattr(responses_main_module, "aresponses", AsyncMock(side_effect=[follow_up]))
+
+    iterator = _make_iterator(_lifecycle_round("resp-interim", _function_call("call_1", "read_wiki_contents")))
+    chunks = [chunk async for chunk in iterator]
+    types = [chunk.type for chunk in chunks]
+
+    assert types.count(ResponsesAPIStreamEvents.RESPONSE_CREATED) == 1
+    assert types.count(ResponsesAPIStreamEvents.RESPONSE_COMPLETED) == 1
+    assert types[-1] == ResponsesAPIStreamEvents.RESPONSE_COMPLETED
+
+    # The function call, the gateway's mcp_call, and the final message each own an index.
+    added = [c for c in chunks if c.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED]
+    assert [(c.output_index, _item_type(c.item)) for c in added] == [
+        (0, "function_call"),
+        (1, "mcp_call"),
+        (2, "message"),
+    ]
+    mcp_item_ids = {c.item_id for c in chunks if c.type == ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS}
+    mcp_done = [
+        c for c in chunks if c.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE and _item_type(c.item) == "mcp_call"
+    ]
+    assert [c.output_index for c in mcp_done] == [1]
+    assert {c.item.id for c in mcp_done} == mcp_item_ids
+    round_two = [
+        c for c in chunks if getattr(c, "item_id", None) is None and getattr(c, "output_index", None) is not None
+    ]
+    assert max(c.output_index for c in round_two) == 2
+
+    # The single completed event lists every round's items and keeps the final round's id for continuation.
+    completed = chunks[-1]
+    assert completed.response.id == "resp-final"
+    assert [_item_type(item) for item in completed.response.output] == ["mcp_call", "message"]
+    assert completed.response.output[-1]["content"][0]["text"] == "Alpha."
+    # The proxy serializes every chunk; the merged output must still be a valid response.
+    assert '"type":"mcp_call"' in completed.response.model_dump_json(exclude_none=True, exclude_unset=True)
+
+    # Numbering stays strictly increasing across rounds and gateway events.
+    sequence_numbers = [c.sequence_number for c in chunks if getattr(c, "sequence_number", None) is not None]
+    assert sequence_numbers == sorted(sequence_numbers)
+    assert len(set(sequence_numbers)) == len(sequence_numbers)
+
+
+@pytest.mark.asyncio
+async def test_final_output_lists_executed_call_as_completed_mcp_call(monkeypatch):
+    """
+    A function_call the gateway executed must not reach the final output: an
+    agent framework reading it (the OpenAI Agents SDK) tries to run a tool the
+    caller never declared and aborts the run. The final output lists the
+    gateway's completed mcp_call in its place, next to the round's other items.
+    """
+    _mock_mcp_environment(monkeypatch)
+
+    follow_up = _FakeAsyncStream(_lifecycle_round("resp-final", _text_message("Alpha.")))
+    monkeypatch.setattr(responses_main_module, "aresponses", AsyncMock(side_effect=[follow_up]))
+
+    reasoning = {"type": "reasoning", "id": "rs_1", "summary": []}
+    iterator = _make_iterator(
+        [
+            _created_chunk("resp-interim"),
+            _completed_chunk([reasoning, _function_call("call_1", "read_wiki_contents")], response_id="resp-interim"),
+        ]
+    )
+    chunks = [chunk async for chunk in iterator]
+
+    final_output = chunks[-1].response.output
+    assert [_item_type(item) for item in final_output] == ["reasoning", "mcp_call", "message"]
+    executed_call = final_output[1]
+    assert executed_call["status"] == "completed"
+    assert executed_call["name"] == "read_wiki_contents"
+    assert executed_call["arguments"] == "{}"
+
+    done_mcp_items = [
+        c.item for c in chunks if c.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE and _item_type(c.item) == "mcp_call"
+    ]
+    assert [item.status for item in done_mcp_items] == ["completed"]
+
+
+@pytest.mark.asyncio
+async def test_stream_without_auto_execute_is_forwarded_unchanged(monkeypatch):
+    """With approval required there is one round, and it passes through untouched."""
+    _mock_mcp_environment(monkeypatch)
+    aresponses_mock = AsyncMock()
+    monkeypatch.setattr(responses_main_module, "aresponses", aresponses_mock)
+
+    upstream = _lifecycle_round("resp-1", _function_call("call_1", "read_wiki_contents"))
+    iterator = MCPEnhancedStreamingIterator(
+        base_iterator=_FakeAsyncStream(list(upstream)),
+        mcp_events=[],
+        tool_server_map={"read_wiki_contents": "deepwiki"},
+        mcp_tools_with_litellm_proxy=[{"require_approval": "always"}],
+        user_api_key_auth=None,
+        original_request_params={"model": "gpt-4", "input": "hi", "tools": [{"type": "mcp"}]},
+    )
+
+    chunks = [chunk async for chunk in iterator]
+
+    assert chunks == upstream
+    assert [c.output_index for c in chunks if hasattr(c, "output_index")] == [0, 0]
+    assert [c.sequence_number for c in chunks if hasattr(c, "sequence_number")] == [0, 1, 2]
+    aresponses_mock.assert_not_called()