diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index def2f72437..dd7936059a 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -21,6 +21,8 @@ from litellm.types.llms.openai import ( OutputTextAnnotationAddedEvent, OutputTextDeltaEvent, OutputTextDoneEvent, + FunctionCallArgumentsDeltaEvent, + FunctionCallArgumentsDoneEvent, ReasoningSummaryTextDeltaEvent, ResponseCompletedEvent, ResponseCreatedEvent, @@ -79,6 +81,161 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): Union[ModelResponse, TextCompletionResponse] ] = None self.final_text: str = "" + self._pending_tool_events: List[BaseLiteLLMOpenAIResponseObject] = [] + self._tool_output_index_by_call_id: dict[str, int] = {} + self._tool_args_by_call_id: dict[str, str] = {} + self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item + self._final_tool_events_queued: bool = False + + def _get_or_assign_tool_output_index(self, call_id: str) -> int: + existing = self._tool_output_index_by_call_id.get(call_id) + if existing is not None: + return existing + idx = self._next_tool_output_index + self._next_tool_output_index += 1 + self._tool_output_index_by_call_id[call_id] = idx + return idx + + def _queue_tool_call_delta_events(self, tool_calls: object) -> None: + """ + Convert chat-completions streaming `tool_calls` deltas into Responses API streaming events. + + We emit: + - response.output_item.added (function_call) + - response.function_call_arguments.delta + """ + if not isinstance(tool_calls, list): + return + + for tc in tool_calls: + call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None) + if not call_id_raw: + continue + call_id = str(call_id_raw) + + fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) + fn_name = "" + fn_args_delta = "" + if isinstance(fn, dict): + fn_name = str(fn.get("name") or "") + fn_args_delta = str(fn.get("arguments") or "") + else: + fn_name = str(getattr(fn, "name", "") or "") + fn_args_delta = str(getattr(fn, "arguments", "") or "") + + output_index = self._get_or_assign_tool_output_index(call_id) + + if call_id not in self._tool_args_by_call_id: + self._tool_args_by_call_id[call_id] = "" + self._pending_tool_events.append( + OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=output_index, + item=BaseLiteLLMOpenAIResponseObject( + **{ + "type": "function_call", + "id": call_id, + "call_id": call_id, + "name": fn_name, + "arguments": "", + "status": "in_progress", + } + ), + ) + ) + + if fn_args_delta: + self._tool_args_by_call_id[call_id] += fn_args_delta + self._pending_tool_events.append( + FunctionCallArgumentsDeltaEvent( + type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, + item_id=call_id, + output_index=output_index, + delta=fn_args_delta, + ) + ) + + def _queue_final_tool_call_done_events(self, litellm_complete_object: ModelResponse) -> None: + """ + Ensure tool calls that were not streamed as deltas still get emitted before response.completed. + """ + if self._final_tool_events_queued: + return + self._final_tool_events_queued = True + + try: + message = litellm_complete_object.choices[0].message # type: ignore + tool_calls = getattr(message, "tool_calls", None) + except Exception: + tool_calls = None + + if not tool_calls or not isinstance(tool_calls, list): + return + + for tc in tool_calls: + call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None) + if not call_id_raw: + continue + call_id = str(call_id_raw) + output_index = self._get_or_assign_tool_output_index(call_id) + + fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None) + fn_name = "" + fn_args = "" + if isinstance(fn, dict): + fn_name = str(fn.get("name") or "") + fn_args = str(fn.get("arguments") or "") + else: + fn_name = str(getattr(fn, "name", "") or "") + fn_args = str(getattr(fn, "arguments", "") or "") + + # If we never sent output_item.added for this call_id, emit it now. + if call_id not in self._tool_args_by_call_id: + self._tool_args_by_call_id[call_id] = "" + self._pending_tool_events.append( + OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=output_index, + item=BaseLiteLLMOpenAIResponseObject( + **{ + "type": "function_call", + "id": call_id, + "call_id": call_id, + "name": fn_name, + "arguments": "", + "status": "in_progress", + } + ), + ) + ) + + final_args = fn_args or self._tool_args_by_call_id.get(call_id, "") + self._pending_tool_events.append( + FunctionCallArgumentsDoneEvent( + type=ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE, + item_id=call_id, + output_index=output_index, + arguments=final_args, + ) + ) + + self._pending_tool_events.append( + OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=output_index, + sequence_number=1, + item=BaseLiteLLMOpenAIResponseObject( + **{ + "type": "function_call", + "id": call_id, + "call_id": call_id, + "name": fn_name, + "arguments": final_args, + "status": "completed", + } + ), + ) + ) def _default_response_created_event_data(self) -> dict: response_created_event_data = { @@ -310,6 +467,12 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): ): self.litellm_model_response = self.create_litellm_model_response() if self.litellm_model_response: + # If tool calls exist, emit tool events before finishing/response.completed. + if isinstance(self.litellm_model_response, ModelResponse): + self._queue_final_tool_call_done_events(self.litellm_model_response) + if self._pending_tool_events: + return self._pending_tool_events.pop(0) + done_event = self.return_default_done_events(self.litellm_model_response) if done_event: return done_event @@ -462,13 +625,27 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): content_index=0, delta=delta_content, ) + + # Priority 3: Handle tool call deltas (if any) -> queue events and emit them + if ( + chunk.choices + and hasattr(chunk.choices[0].delta, "tool_calls") + and chunk.choices[0].delta.tool_calls + ): + self._queue_tool_call_delta_events(chunk.choices[0].delta.tool_calls) + if self._pending_tool_events: + return self._pending_tool_events.pop(0) - # Priority 3: If we have pending annotation events, emit the next one + # Priority 4: If we have pending annotation events, emit the next one # This happens when the current chunk has no text/reasoning content if hasattr(self, '_pending_annotation_events') and self._pending_annotation_events: event = self._pending_annotation_events.pop(0) return event + # Priority 5: If we have pending tool events (from earlier chunk), emit the next one + if self._pending_tool_events: + return self._pending_tool_events.pop(0) + return None def _get_delta_string_from_streaming_choices( diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_tool_call_streaming_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_tool_call_streaming_transformation.py new file mode 100644 index 0000000000..51150383b0 --- /dev/null +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_tool_call_streaming_transformation.py @@ -0,0 +1,116 @@ +""" +Tests for streaming tool-calls in Responses API transformation. + +Ensures that when the underlying chat-completions stream includes tool_calls deltas, +LiteLLM emits Responses API streaming events (output_item.added + function_call_arguments.*). + +Also ensures that tool calls that only appear in the final built response still get emitted +before response.completed. +""" + +from unittest.mock import AsyncMock + +from litellm.responses.litellm_completion_transformation.streaming_iterator import ( + LiteLLMCompletionStreamingIterator, +) +from litellm.types.llms.openai import ResponsesAPIStreamEvents +from litellm.types.utils import Delta, ModelResponse, ModelResponseStream, StreamingChoices + + +def test_tool_call_delta_is_emitted_as_responses_events(): + iterator = LiteLLMCompletionStreamingIterator( + model="test-model", + litellm_custom_stream_wrapper=AsyncMock(), + request_input="Test input", + responses_api_request={}, + ) + + # A streaming chunk with tool_calls delta but no text + chunk = ModelResponseStream( + id="chunk-1", + created=123, + model="test-model", + object="chat.completion.chunk", + choices=[ + StreamingChoices( + finish_reason=None, + index=0, + delta=Delta( + role="assistant", + content="", + tool_calls=[ + { + "id": "call_1", + "type": "function", + "function": {"name": "do_thing", "arguments": '{"x":1}'}, + } + ], + ), + ) + ], + ) + + evt1 = iterator._transform_chat_completion_chunk_to_response_api_chunk(chunk) + assert evt1 is not None + assert evt1.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED + assert evt1.output_index == 1 + + evt2 = iterator._transform_chat_completion_chunk_to_response_api_chunk(chunk) + assert evt2 is not None + assert evt2.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA + assert evt2.item_id == "call_1" + assert evt2.output_index == 1 + assert evt2.delta == '{"x":1}' + + +def test_tool_calls_present_only_in_final_response_are_emitted_before_completed(): + iterator = LiteLLMCompletionStreamingIterator( + model="test-model", + litellm_custom_stream_wrapper=AsyncMock(), + request_input="Test input", + responses_api_request={}, + ) + + # Construct a final ModelResponse with tool_calls on the message. + # We bypass the stream builder and directly set iterator.litellm_model_response. + response = ModelResponse( + id="resp-1", + created=123, + model="test-model", + object="chat.completion", + choices=[ + { + "index": 0, + "finish_reason": "tool_calls", + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_2", + "type": "function", + "function": {"name": "do_thing", "arguments": '{"y":2}'}, + "index": 0, + } + ], + }, + } + ], + ) + iterator.litellm_model_response = response + + # First common_done_event_logic call should yield tool events, not response.completed. + evt1 = iterator.common_done_event_logic(sync_mode=True) + assert evt1.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED + assert evt1.output_index == 1 + + evt2 = iterator.common_done_event_logic(sync_mode=True) + assert evt2.type == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE + assert evt2.item_id == "call_2" + assert evt2.output_index == 1 + assert evt2.arguments == '{"y":2}' + + evt3 = iterator.common_done_event_logic(sync_mode=True) + assert evt3.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE + assert evt3.output_index == 1 +