From e7557ada57ad649fc0b0063aa2d89e1f7186f979 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Mon, 21 Sep 2026 14:56:02 -0700 Subject: [PATCH] fix(responses): list executed MCP calls as completed mcp_call items in the final output --- .../responses/mcp/mcp_streaming_iterator.py | 23 ++++++++- .../mcp/test_mcp_streaming_iterator.py | 48 +++++++++++++++---- 2 files changed, 60 insertions(+), 11 deletions(-) diff --git a/litellm/responses/mcp/mcp_streaming_iterator.py b/litellm/responses/mcp/mcp_streaming_iterator.py index fb05e7bff99..101000b14a9 100644 --- a/litellm/responses/mcp/mcp_streaming_iterator.py +++ b/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, diff --git a/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py b/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py index 1587557f3d2..1e869745dc3 100644 --- a/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py +++ b/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."""