fix(responses): list executed MCP calls as completed mcp_call items in the final output

litellm/responses/mcp/mcp_streaming_iterator.py

@@ -39,6 +39,19 @@ def _output_items(response: ResponsesAPIResponse) -> Sequence[object]:
     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)
@@ -588,9 +601,14 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
         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 so the final response.completed can list them."""
+        """Bank a finished round's items, each function_call the gateway answered replaced by its mcp_call."""
         width: Final = self._round_output_width(response)
-        self._composed_output.extend(_output_items(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()
@@ -842,6 +860,7 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator):
                     **{  # 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,

tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py

@@ -143,17 +143,12 @@ async def test_second_round_tool_call_is_executed_and_reaches_final_text(monkeyp
 
     # The stream reached round 3 and produced the final text response instead
     # of stopping after round 1 or round 2. The client sees one lifecycle whose
-    # final output lists every round's items in order.
+    # 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) == 1
     final_output = completed_chunks[-1].response.output
-    assert [_item_type(item) for item in final_output] == [
-        "function_call",
-        "mcp_call",
-        "function_call",
-        "mcp_call",
-        "message",
-    ]
+    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."
 
 
@@ -422,7 +417,7 @@ async def test_auto_execute_rounds_share_one_public_lifecycle(monkeypatch):
     # 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] == ["function_call", "mcp_call", "message"]
+    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)
@@ -433,6 +428,41 @@ async def test_auto_execute_rounds_share_one_public_lifecycle(monkeypatch):
     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."""