From 391b8a265b7c5285b2640361268ec67b72c143f3 Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Mon, 7 Sep 2026 18:40:43 +0530 Subject: [PATCH 1/6] fix(responses): stream one lifecycle across MCP auto-execute rounds Each auto-executed MCP round is a distinct upstream response, but the client reads one stream. The follow-up round's response.created, response.in_progress, and the interim response.completed were forwarded as-is, so one SSE body carried two lifecycles and output_index restarted at zero, which aborts accumulating clients such as the OpenAI SDK's responses.stream() before the final answer arrives. Fold the rounds into one public lifecycle: drop the openers of follow-up rounds, hold back the completed event of a round whose tool calls the gateway executes, shift later output indexes past the items already emitted, and list every round's items on the single final response.completed. Each mcp_call item is announced with output_item.added, keeps one item id across its events, and owns its own output_index. Sequence numbers stay strictly increasing when a round or a gateway event restarts numbering. The final round's response id is kept on response.completed so previous_response_id continuation still works. --- .../responses/mcp/mcp_streaming_iterator.py | 212 ++++++++++++++---- 1 file changed, 166 insertions(+), 46 deletions(-) diff --git a/litellm/responses/mcp/mcp_streaming_iterator.py b/litellm/responses/mcp/mcp_streaming_iterator.py index ca12b3e7cc3..ac158a32371 100644 --- a/litellm/responses/mcp/mcp_streaming_iterator.py +++ b/litellm/responses/mcp/mcp_streaming_iterator.py @@ -34,6 +34,16 @@ 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 _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", @@ -170,6 +180,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]] = [] @@ -179,7 +190,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) @@ -187,7 +198,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, @@ -197,7 +208,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, @@ -210,7 +221,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) @@ -219,7 +230,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, @@ -239,7 +250,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) @@ -330,6 +341,17 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): self._error_event_emitted = False self._last_sequence_number = 0 + # Every auto-execute round is a distinct upstream response, but the + # client is reading one stream. Fold the rounds into one public + # lifecycle: one response.created, one response.completed whose + # output holds every round's items, and output indexes that are + # never reused for a different item. + 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""" @@ -415,8 +437,14 @@ 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): + # Follow-up rounds and gateway events restart their numbering. + # Keep the public stream strictly increasing. + 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: @@ -472,7 +500,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 @@ -530,17 +558,11 @@ 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 + # None means the chunk was folded into the single public + # lifecycle; fall through so phase 4 runs the follow-up. + return await self._compose_round_chunk(chunk) except StopAsyncIteration: if self.should_auto_execute and self.collected_response: self.phase = "tool_execution" @@ -566,6 +588,69 @@ 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 so the final response.completed can list them.""" + width: Final = self._round_output_width(response) + self._composed_output.extend(_output_items(response)) + 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 = [] + 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), + ] + _set_event_field(chunk, "response", response_obj.model_copy(update={"output": merged_output})) + return chunk + async def _process_base_iterator_chunk(self) -> ResponsesAPIStreamingResponse: """ Process a chunk from the base iterator with response ID consistency enforcement. @@ -593,17 +678,11 @@ 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: + # The chunk stays internal; hand the next public event back instead. + 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""" @@ -667,6 +746,16 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): return self.tool_call_round += 1 + # Each executed tool is one mcp_call output item of the single + # public response. Announce it at an output_index past the items + # this round already streamed, and keep that item id for the + # completion events below. + 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, @@ -674,14 +763,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( + { + "type": ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + "sequence_number": len(self.tool_execution_events) + 1, + "output_index": output_index, + "item": { + "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]) @@ -719,37 +830,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( + **{ + "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_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) + # The response model accepts output items as dicts, not as the generic event object. + self._pending_mcp_call_items.append(mcp_call_item.model_dump()) # Store tool results for follow-up call self.tool_results = tool_results @@ -824,6 +943,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) From 13ce30488ff6eed68c213a80a41a38979a279e59 Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Mon, 7 Sep 2026 18:40:43 +0530 Subject: [PATCH 2/6] test(responses): cover the single public lifecycle for MCP rounds Assert one response.created and one response.completed across rounds, dense output indexes for the function call, the mcp_call item, and the final message, stable mcp_call item ids, strictly increasing sequence numbers, a serializable merged output, and that a stream without auto-execution is forwarded unchanged. Update the two existing tests that counted one completed event per internal round. --- .../mcp/test_mcp_streaming_iterator.py | 128 +++++++++++++++++- 1 file changed, 122 insertions(+), 6 deletions(-) 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 5001589ce54..57ccbbfa493 100644 --- a/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py +++ b/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py @@ -57,6 +57,10 @@ def _text_message(text: str): return {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": text}]} +def _item_type(item) -> str: + return item["type"] if isinstance(item, dict) else 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)]) @@ -134,11 +138,19 @@ 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. 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] == [ + "function_call", + "mcp_call", + "function_call", + "mcp_call", + "message", + ] + assert final_output[-1]["content"][0]["text"] == "Here's what I found after retrying." @pytest.mark.asyncio @@ -207,7 +219,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" @@ -280,7 +292,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=[], @@ -316,7 +330,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=[], @@ -336,3 +352,103 @@ 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, **fields): + return SimpleNamespace(type=event_type, **fields) + + +def _lifecycle_round(response_id: str, item: dict, sequence_start: int = 0): + """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] == ["function_call", "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_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() From 146e5c492b1bee31a57c77a8fafff74122a13df9 Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Mon, 7 Sep 2026 18:49:04 +0530 Subject: [PATCH 3/6] fix(responses): keep the MCP lifecycle change within the type-discipline budget Mark the dict literals handed straight to model constructors, clear the pending mcp_call list in place, and bind the merged response before setting it on the event, so LIT002 stays at or below its base count. --- litellm/responses/mcp/mcp_streaming_iterator.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/litellm/responses/mcp/mcp_streaming_iterator.py b/litellm/responses/mcp/mcp_streaming_iterator.py index ac158a32371..1c892eec803 100644 --- a/litellm/responses/mcp/mcp_streaming_iterator.py +++ b/litellm/responses/mcp/mcp_streaming_iterator.py @@ -602,7 +602,7 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): self._composed_output.extend(_output_items(response)) 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 = [] + self._pending_mcp_call_items.clear() self._round_max_output_index = -1 async def _compose_round_chunk(self, chunk: ResponsesAPIStreamingResponse) -> ResponsesAPIStreamingResponse | None: @@ -648,7 +648,10 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): *self._composed_output, *_output_items(response_obj), ] - _set_event_field(chunk, "response", response_obj.model_copy(update={"output": merged_output})) + 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: @@ -769,11 +772,11 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): 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": { + "item": { # mutable-ok: consumed once by model_validate "id": item_id, "type": "mcp_call", "status": "in_progress", @@ -850,7 +853,7 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): 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", "approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}", From 5f84e8331689c95c333d0309e89ad566cd613122 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:48:11 -0700 Subject: [PATCH 4/6] chore(responses): drop narrative comments from the MCP streaming iterator --- litellm/responses/mcp/mcp_streaming_iterator.py | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/litellm/responses/mcp/mcp_streaming_iterator.py b/litellm/responses/mcp/mcp_streaming_iterator.py index 1c892eec803..fb05e7bff99 100644 --- a/litellm/responses/mcp/mcp_streaming_iterator.py +++ b/litellm/responses/mcp/mcp_streaming_iterator.py @@ -341,11 +341,6 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): self._error_event_emitted = False self._last_sequence_number = 0 - # Every auto-execute round is a distinct upstream response, but the - # client is reading one stream. Fold the rounds into one public - # lifecycle: one response.created, one response.completed whose - # output holds every round's items, and output indexes that are - # never reused for a different item. self._round_index = 0 self._output_index_offset = 0 self._round_max_output_index = -1 @@ -438,8 +433,6 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): chunk: Final = await self._anext_impl() sequence_number: Final = getattr(chunk, "sequence_number", None) if isinstance(sequence_number, int): - # Follow-up rounds and gateway events restart their numbering. - # Keep the public stream strictly increasing. 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) @@ -560,8 +553,6 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): self.phase = "mcp_discovery" return await self._compose_round_chunk(chunk) - # None means the chunk was folded into the single public - # lifecycle; fall through so phase 4 runs the follow-up. return await self._compose_round_chunk(chunk) except StopAsyncIteration: if self.should_auto_execute and self.collected_response: @@ -683,7 +674,6 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): composed: Final = await self._compose_round_chunk(chunk) if composed is None: - # The chunk stays internal; hand the next public event back instead. return await self._anext_impl() return composed @@ -749,10 +739,6 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): return self.tool_call_round += 1 - # Each executed tool is one mcp_call output item of the single - # public response. Announce it at an output_index past the items - # this round already streamed, and keep that item id for the - # completion events below. from litellm.types.llms.openai import OutputItemAddedEvent next_output_index = self._output_index_offset + self._round_output_width( # rebind-ok: advances per item @@ -870,7 +856,6 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): item=mcp_call_item, ) self.tool_execution_events.append(output_item_done_event) - # The response model accepts output items as dicts, not as the generic event object. self._pending_mcp_call_items.append(mcp_call_item.model_dump()) # Store tool results for follow-up call From 239bff3315ee9836559b7947157abe71c3cdcd0c Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:57:50 -0700 Subject: [PATCH 5/6] test(responses): type the MCP lifecycle test helpers --- .../responses/mcp/test_mcp_streaming_iterator.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) 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 57ccbbfa493..1587557f3d2 100644 --- a/tests/test_litellm/responses/mcp/test_mcp_streaming_iterator.py +++ b/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,8 +61,8 @@ def _text_message(text: str): return {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": text}]} -def _item_type(item) -> str: - return item["type"] if isinstance(item, dict) else item.type +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: @@ -354,11 +358,11 @@ async def test_streaming_follow_up_keeps_previous_response_id_when_stored(monkey assert not [item for item in follow_up_kwargs["input"] if item.get("type") == "reasoning"] -def _event(event_type, **fields): +def _event(event_type: ResponsesAPIStreamEvents, **fields: object) -> SimpleNamespace: return SimpleNamespace(type=event_type, **fields) -def _lifecycle_round(response_id: str, item: dict, sequence_start: int = 0): +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( 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 6/6] 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."""