diff --git a/litellm-proxy-extras/litellm_proxy_extras/utils.py b/litellm-proxy-extras/litellm_proxy_extras/utils.py index ece2b496bf6..73065b050b7 100644 --- a/litellm-proxy-extras/litellm_proxy_extras/utils.py +++ b/litellm-proxy-extras/litellm_proxy_extras/utils.py @@ -131,7 +131,9 @@ class ProxyExtrasDBManager: ) @staticmethod - def _resolve_all_migrations(migrations_dir: str, schema_path: str): + def _resolve_all_migrations( + migrations_dir: str, schema_path: str, mark_all_applied: bool = True + ): """ 1. Compare the current database state to schema.prisma and generate a migration for the diff. 2. Run prisma migrate deploy to apply any pending migrations. @@ -210,6 +212,8 @@ class ProxyExtrasDBManager: logger.warning("Migration diff application timed out.") # 3. Mark all migrations as applied + if not mark_all_applied: + return migration_names = ProxyExtrasDBManager._get_migration_names(migrations_dir) logger.info(f"Resolving {len(migration_names)} migrations") for migration_name in migration_names: @@ -263,6 +267,13 @@ class ProxyExtrasDBManager: logger.info(f"prisma migrate deploy stdout: {result.stdout}") logger.info("prisma migrate deploy completed") + + # Run sanity check to ensure DB matches schema + logger.info("Running post-migration sanity check...") + ProxyExtrasDBManager._resolve_all_migrations( + migrations_dir, schema_path, mark_all_applied=False + ) + logger.info("✅ Post-migration sanity check completed") return True except subprocess.CalledProcessError as e: logger.info(f"prisma db error: {e.stderr}, e: {e.stdout}") diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 683f06ee0c7..b060f22d355 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -57,22 +57,22 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): ) -> Tuple[Optional[Any], int]: """ Handle raw dict response items from Responses API (e.g., GPT-5 Codex format). - + Args: item: Raw dict response item with 'type' field index: Current choice index - + Returns: Tuple of (Choice object or None, updated index) """ from litellm.types.utils import Choices, Message item_type = item.get("type") - + # Ignore reasoning items for now if item_type == "reasoning": return None, index - + # Handle message items with output_text content if item_type == "message": content_list = item.get("content", []) @@ -83,13 +83,11 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): response_text = content_item.get("text", "") msg = Message( role=item.get("role", "assistant"), - content=response_text if response_text else "" - ) - choice = Choices( - message=msg, finish_reason="stop", index=index + content=response_text if response_text else "", ) + choice = Choices(message=msg, finish_reason="stop", index=index) return choice, index + 1 - + # Unknown or unsupported type return None, index @@ -294,8 +292,8 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): if isinstance(item, ResponseReasoningItem): - for content in item.summary: - response_text = getattr(content, "text", "") + for summary_item in item.summary: + response_text = getattr(summary_item, "text", "") reasoning_content = response_text if response_text else "" elif isinstance(item, ResponseOutputMessage): @@ -340,7 +338,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): index += 1 elif isinstance(item, dict): # Handle raw dict responses (e.g., from GPT-5 Codex) - choice, index = self._handle_raw_dict_response_item(item=item, index=index) + choice, index = self._handle_raw_dict_response_item( + item=item, index=index + ) if choice is not None: choices.append(choice) else: diff --git a/litellm/proxy/_new_secret_config.yaml b/litellm/proxy/_new_secret_config.yaml index c0772f964ba..108bef4fff3 100644 --- a/litellm/proxy/_new_secret_config.yaml +++ b/litellm/proxy/_new_secret_config.yaml @@ -1,4 +1,4 @@ model_list: - model_name: gpt-5-codex litellm_params: - model: gpt-5-codex \ No newline at end of file + model: gpt-5-codex diff --git a/litellm/responses/litellm_completion_transformation/handler.py b/litellm/responses/litellm_completion_transformation/handler.py index 9317bf26178..7e0b4cfa243 100644 --- a/litellm/responses/litellm_completion_transformation/handler.py +++ b/litellm/responses/litellm_completion_transformation/handler.py @@ -58,7 +58,7 @@ class LiteLLMCompletionTransformationHandler: responses_api_request=responses_api_request, **kwargs, ) - + completion_args = {} completion_args.update(kwargs) completion_args.update(litellm_completion_request) @@ -83,6 +83,7 @@ class LiteLLMCompletionTransformationHandler: elif isinstance(litellm_completion_response, litellm.CustomStreamWrapper): return LiteLLMCompletionStreamingIterator( + model=model, litellm_custom_stream_wrapper=litellm_completion_response, request_input=input, responses_api_request=responses_api_request, @@ -106,7 +107,7 @@ class LiteLLMCompletionTransformationHandler: previous_response_id=previous_response_id, litellm_completion_request=litellm_completion_request, ) - + acompletion_args = {} acompletion_args.update(kwargs) acompletion_args.update(litellm_completion_request) @@ -130,9 +131,12 @@ class LiteLLMCompletionTransformationHandler: elif isinstance(litellm_completion_response, litellm.CustomStreamWrapper): return LiteLLMCompletionStreamingIterator( + model=litellm_completion_request.get("model") or "", litellm_custom_stream_wrapper=litellm_completion_response, request_input=request_input, responses_api_request=responses_api_request, - custom_llm_provider=litellm_completion_request.get("custom_llm_provider"), + custom_llm_provider=litellm_completion_request.get( + "custom_llm_provider" + ), litellm_metadata=kwargs.get("litellm_metadata", {}), ) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index 20822b3628e..d3a8cea3497 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -1,4 +1,6 @@ -from typing import List, Optional, Union +import time +import uuid +from typing import List, Optional, Union, cast import litellm from litellm.main import stream_chunk_builder @@ -8,11 +10,23 @@ from litellm.responses.litellm_completion_transformation.transformation import ( from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator from litellm.responses.utils import ResponsesAPIRequestUtils from litellm.types.llms.openai import ( + PART_UNION_TYPES, + BaseLiteLLMOpenAIResponseObject, + ContentPartAddedEvent, + ContentPartDoneEvent, + ContentPartDonePartOutputText, + ContentPartDonePartReasoningText, + OutputItemAddedEvent, + OutputItemDoneEvent, OutputTextDeltaEvent, + OutputTextDoneEvent, ReasoningSummaryTextDeltaEvent, ResponseCompletedEvent, + ResponseCreatedEvent, + ResponseInProgressEvent, ResponseInputParam, ResponsesAPIOptionalRequestParams, + ResponsesAPIResponse, ResponsesAPIStreamEvents, ResponsesAPIStreamingResponse, ) @@ -32,12 +46,14 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): def __init__( self, + model: str, litellm_custom_stream_wrapper: litellm.CustomStreamWrapper, request_input: Union[str, ResponseInputParam], responses_api_request: ResponsesAPIOptionalRequestParams, custom_llm_provider: Optional[str] = None, litellm_metadata: Optional[dict] = None, ): + self.model: str = model self.litellm_custom_stream_wrapper: litellm.CustomStreamWrapper = ( litellm_custom_stream_wrapper ) @@ -50,14 +66,276 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self.collected_chat_completion_chunks: List[ModelResponseStream] = [] self.finished: bool = False self.litellm_logging_obj = litellm_custom_stream_wrapper.logging_obj + self.sent_response_created_event: bool = False + self.sent_response_in_progress_event: bool = False + self.sent_output_item_added_event: bool = False + self.sent_content_part_added_event: bool = False + self.sent_output_text_done_event: bool = False + self.sent_output_content_part_done_event: bool = False + self.sent_output_item_done_event: bool = False + self.litellm_model_response: Optional[ + Union[ModelResponse, TextCompletionResponse] + ] = None + self.final_text: str = "" + + def _default_response_created_event_data(self) -> dict: + response_created_event_data = { + "id": f"resp_{str(uuid.uuid4())}", + "object": "response", + "created_at": int(time.time()), + "status": "in_progress", + "error": None, + "incomplete_details": None, + "instructions": self.request_input, + "max_output_tokens": None, + "model": self.model, + "output": [], + "parallel_tool_calls": True, + "previous_response_id": None, + "reasoning": {"effort": None, "summary": None}, + "store": True, + } + if "temperature" in self.responses_api_request: + response_created_event_data["temperature"] = self.responses_api_request[ + "temperature" + ] + if "text" in self.responses_api_request: + response_created_event_data["text"] = self.responses_api_request["text"] + if "tool_choice" in self.responses_api_request: + response_created_event_data["tool_choice"] = self.responses_api_request[ + "tool_choice" + ] + else: + response_created_event_data["tool_choice"] = "auto" + if "tools" in self.responses_api_request: + response_created_event_data["tools"] = self.responses_api_request["tools"] + else: + response_created_event_data["tools"] = [] + if "top_p" in self.responses_api_request: + response_created_event_data["top_p"] = self.responses_api_request["top_p"] + else: + response_created_event_data["top_p"] = 1.0 + if "truncation" in self.responses_api_request: + response_created_event_data["truncation"] = self.responses_api_request[ + "truncation" + ] + if "usage" in self.responses_api_request: + response_created_event_data["usage"] = self.responses_api_request["usage"] + if "user" in self.responses_api_request: + response_created_event_data["user"] = self.responses_api_request["user"] + if "metadata" in self.responses_api_request: + response_created_event_data["metadata"] = self.responses_api_request[ + "metadata" + ] + return response_created_event_data + + def create_response_created_event(self) -> ResponseCreatedEvent: + """ + data: {"type":"response.created","response":{"id":"resp_67c9fdcecf488190bdd9a0409de3a1ec07b8b0ad4e5eb654","object":"response","created_at":1741290958,"status":"in_progress","error":null,"incomplete_details":null,"instructions":"You are a helpful assistant.","max_output_tokens":null,"model":"gpt-4.1-2025-04-14","output":[],"parallel_tool_calls":true,"previous_response_id":null,"reasoning":{"effort":null,"summary":null},"store":true,"temperature":1.0,"text":{"format":{"type":"text"}},"tool_choice":"auto","tools":[],"top_p":1.0,"truncation":"disabled","usage":null,"user":null,"metadata":{}}} + + """ + response_created_event_data = self._default_response_created_event_data() + return ResponseCreatedEvent( + type=ResponsesAPIStreamEvents.RESPONSE_CREATED, + response=ResponsesAPIResponse(**response_created_event_data), + ) + + def create_response_in_progress_event(self) -> ResponseInProgressEvent: + response_in_progress_event_data = self._default_response_created_event_data() + response_in_progress_event_data["status"] = "in_progress" + return ResponseInProgressEvent( + type=ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, + response=ResponsesAPIResponse(**response_in_progress_event_data), + ) + + def create_output_item_added_event(self) -> OutputItemAddedEvent: + return OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=0, + item=BaseLiteLLMOpenAIResponseObject( + **{ + "id": f"msg_{str(uuid.uuid4())}", + "type": "message", + "status": "in_progress", + "role": "assistant", + "content": [], + } + ), + ) + + def create_content_part_added_event(self) -> ContentPartAddedEvent: + return ContentPartAddedEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED, + item_id=f"msg_{str(uuid.uuid4())}", + output_index=0, + content_index=0, + part=BaseLiteLLMOpenAIResponseObject( + **{"type": "output_text", "text": "", "annotations": []} + ), + ) + + def create_litellm_model_response( + self, + ) -> Optional[ModelResponse]: + return cast( + Optional[ModelResponse], + stream_chunk_builder( + chunks=self.collected_chat_completion_chunks, + logging_obj=self.litellm_logging_obj, + ), + ) + + def create_output_text_done_event( + self, litellm_complete_object: ModelResponse + ) -> OutputTextDoneEvent: + return OutputTextDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + item_id=f"msg_{str(uuid.uuid4())}", + output_index=0, + content_index=0, + text=getattr(litellm_complete_object.choices[0].message, "content", "") # type: ignore + or "", + ) + + def create_output_content_part_done_event( + self, litellm_complete_object: ModelResponse + ) -> ContentPartDoneEvent: + + text = getattr(litellm_complete_object.choices[0].message, "content", "") or "" # type: ignore + reasoning_content = getattr(litellm_complete_object.choices[0].message, "reasoning_content", "") or "" # type: ignore + + part: Optional[PART_UNION_TYPES] = None + if reasoning_content: + part = ContentPartDonePartReasoningText( + type="reasoning_text", + reasoning=reasoning_content, + ) + + else: + part = ContentPartDonePartOutputText( + type="output_text", + text=text, + annotations=[], + logprobs=None, + ) + + return ContentPartDoneEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_DONE, + item_id=f"msg_{str(uuid.uuid4())}", + output_index=0, + content_index=0, + part=part, + ) + + def create_output_item_done_event( + self, litellm_complete_object: ModelResponse + ) -> OutputItemDoneEvent: + text = self.litellm_model_response.choices[0].message.content or "" # type: ignore + return OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=0, + sequence_number=1, + item=BaseLiteLLMOpenAIResponseObject( + **{ + "id": f"msg_{str(uuid.uuid4())}", + "status": "completed", + "type": "message", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": text, + "annotations": [], + } + ], + } + ), + ) + + def return_default_done_events( + self, litellm_complete_object: ModelResponse + ) -> Optional[BaseLiteLLMOpenAIResponseObject]: + if self.sent_output_text_done_event is False: + self.sent_output_text_done_event = True + return self.create_output_text_done_event(litellm_complete_object) + if self.sent_output_content_part_done_event is False: + self.sent_output_content_part_done_event = True + return self.create_output_content_part_done_event(litellm_complete_object) + if self.sent_output_item_done_event is False: + self.sent_output_item_done_event = True + return self.create_output_item_done_event(litellm_complete_object) + return None + + def return_default_initial_events( + self, + ) -> Optional[BaseLiteLLMOpenAIResponseObject]: + if self.sent_response_created_event is False: + self.sent_response_created_event = True + return self.create_response_created_event() + elif self.sent_response_in_progress_event is False: + self.sent_response_in_progress_event = True + return self.create_response_in_progress_event() + elif self.sent_output_item_added_event is False: + self.sent_output_item_added_event = True + return self.create_output_item_added_event() + elif self.sent_content_part_added_event is False: + self.sent_content_part_added_event = True + return self.create_content_part_added_event() + return None + + def is_stream_finished(self) -> bool: + if ( + self.sent_output_text_done_event is True + and self.sent_output_content_part_done_event is True + and self.sent_output_item_done_event is True + ): + return True + return False + + def common_done_event_logic( + self, sync_mode: bool = True + ) -> BaseLiteLLMOpenAIResponseObject: + if not self.litellm_model_response or isinstance( + self.litellm_model_response, TextCompletionResponse + ): + self.litellm_model_response = self.create_litellm_model_response() + if self.litellm_model_response: + done_event = self.return_default_done_events(self.litellm_model_response) + if done_event: + return done_event + else: + if sync_mode: + raise StopIteration + else: + raise StopAsyncIteration + + self.finished = self.is_stream_finished() + response_completed_event = self._emit_response_completed_event( + self.litellm_model_response + ) + if response_completed_event: + return response_completed_event + else: + if sync_mode: + raise StopIteration + else: + raise StopAsyncIteration async def __anext__( self, - ) -> Union[ResponsesAPIStreamingResponse, ResponseCompletedEvent]: + ) -> Union[ + ResponsesAPIStreamingResponse, + ResponseCompletedEvent, + BaseLiteLLMOpenAIResponseObject, + ]: try: while True: if self.finished is True: raise StopAsyncIteration + + result = self.return_default_initial_events() + if result: + return result # Get the next chunk from the stream try: chunk = await self.litellm_custom_stream_wrapper.__anext__() @@ -70,12 +348,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): if response_api_chunk: return response_api_chunk except StopAsyncIteration: - self.finished = True - response_completed_event = self._emit_response_completed_event() - if response_completed_event: - return response_completed_event - else: - raise StopAsyncIteration + return self.common_done_event_logic(sync_mode=False) except Exception as e: # Handle HTTP errors @@ -87,12 +360,20 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): def __next__( self, - ) -> Union[ResponsesAPIStreamingResponse, ResponseCompletedEvent]: + ) -> Union[ + ResponsesAPIStreamingResponse, + ResponseCompletedEvent, + BaseLiteLLMOpenAIResponseObject, + ]: try: while True: if self.finished is True: raise StopIteration # Get the next chunk from the stream + + result = self.return_default_initial_events() + if result: + return result try: chunk = self.litellm_custom_stream_wrapper.__next__() self.collected_chat_completion_chunks.append(chunk) @@ -104,13 +385,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): if response_api_chunk: return response_api_chunk except StopIteration: - self.finished = True - response_completed_event = self._emit_response_completed_event() - if response_completed_event: - return response_completed_event - else: - raise StopIteration - + return self.common_done_event_logic(sync_mode=True) except Exception as e: # Handle HTTP errors self.finished = True @@ -165,26 +440,33 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): chat_completion_delta: ChatCompletionDelta = choice.delta return chat_completion_delta.content or "" - def _emit_response_completed_event(self) -> Optional[ResponseCompletedEvent]: - litellm_model_response: Optional[ - Union[ModelResponse, TextCompletionResponse] - ] = stream_chunk_builder(chunks=self.collected_chat_completion_chunks, logging_obj=self.litellm_logging_obj) - if litellm_model_response and isinstance(litellm_model_response, ModelResponse): + def _emit_response_completed_event( + self, litellm_model_response: ModelResponse + ) -> Optional[ResponseCompletedEvent]: + + if litellm_model_response: # Add cost to usage object if include_cost_in_streaming_usage is True - if litellm.include_cost_in_streaming_usage and self.litellm_logging_obj is not None: + if ( + litellm.include_cost_in_streaming_usage + and self.litellm_logging_obj is not None + ): usage = getattr(litellm_model_response, "usage", None) if usage is not None: setattr( - usage, "cost", self.litellm_logging_obj._response_cost_calculator(result=litellm_model_response) + usage, + "cost", + self.litellm_logging_obj._response_cost_calculator( + result=litellm_model_response + ), ) - + # Transform the response responses_api_response = LiteLLMCompletionResponsesConfig.transform_chat_completion_response_to_responses_api_response( request_input=self.request_input, chat_completion_response=litellm_model_response, responses_api_request=self.responses_api_request, ) - + # Encode the response ID to match non-streaming behavior encoded_response = ResponsesAPIRequestUtils._update_responses_api_response_id_with_model_id( responses_api_response=responses_api_response, diff --git a/litellm/responses/mcp/mcp_streaming_iterator.py b/litellm/responses/mcp/mcp_streaming_iterator.py index 7d53452c1c0..8801e561915 100644 --- a/litellm/responses/mcp/mcp_streaming_iterator.py +++ b/litellm/responses/mcp/mcp_streaming_iterator.py @@ -1,19 +1,10 @@ -from litellm._uuid import uuid -from typing import ( - TYPE_CHECKING, - Any, - Dict, - List, - Optional, - Union, - cast, -) +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast from litellm._logging import verbose_logger -from litellm.responses.streaming_iterator import ( - BaseResponsesAPIStreamingIterator, -) +from litellm._uuid import uuid +from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator from litellm.types.llms.openai import ( + BaseLiteLLMOpenAIResponseObject, MCPCallArgumentsDeltaEvent, MCPCallArgumentsDoneEvent, MCPCallCompletedEvent, @@ -38,22 +29,24 @@ async def create_mcp_list_tools_events( mcp_tools_with_litellm_proxy: List[ToolParam], user_api_key_auth: Any, base_item_id: str, - pre_processed_mcp_tools: List[Any] + pre_processed_mcp_tools: List[Any], ) -> List[ResponsesAPIStreamingResponse]: """Create MCP discovery events using pre-processed tools from the parent""" - + events: List[ResponsesAPIStreamingResponse] = [] - + try: # Extract MCP server names mcp_servers = [] for tool in mcp_tools_with_litellm_proxy: if isinstance(tool, dict) and "server_url" in tool: server_url = tool.get("server_url") - if isinstance(server_url, str) and server_url.startswith("litellm_proxy/mcp/"): + if isinstance(server_url, str) and server_url.startswith( + "litellm_proxy/mcp/" + ): server_name = server_url.split("/")[-1] mcp_servers.append(server_name) - + # Emit list tools in progress event in_progress_event = MCPListToolsInProgressEvent( type=ResponsesAPIStreamEvents.MCP_LIST_TOOLS_IN_PROGRESS, @@ -62,21 +55,21 @@ async def create_mcp_list_tools_events( item_id=base_item_id, ) events.append(in_progress_event) - + # Use the pre-processed MCP tools that were already fetched, filtered, and deduplicated by the parent filtered_mcp_tools = pre_processed_mcp_tools - + # Convert tools to dict format for the event mcp_tools_dict = [] for tool in filtered_mcp_tools: - if hasattr(tool, 'model_dump') and callable(getattr(tool, 'model_dump')): + if hasattr(tool, "model_dump") and callable(getattr(tool, "model_dump")): # Type cast to help mypy understand this is safe after hasattr check mcp_tools_dict.append(cast(Any, tool).model_dump()) - elif hasattr(tool, '__dict__'): + elif hasattr(tool, "__dict__"): mcp_tools_dict.append(tool.__dict__) else: - mcp_tools_dict.append({"name": getattr(tool, 'name', str(tool))}) - + mcp_tools_dict.append({"name": getattr(tool, "name", str(tool))}) + # Emit list tools completed event completed_event = MCPListToolsCompletedEvent( type=ResponsesAPIStreamEvents.MCP_LIST_TOOLS_COMPLETED, @@ -85,7 +78,7 @@ async def create_mcp_list_tools_events( item_id=base_item_id, ) events.append(completed_event) - + # Add output_item.done event with the actual tools list (matching OpenAI format) from litellm.types.llms.openai import OutputItemDoneEvent @@ -95,45 +88,50 @@ async def create_mcp_list_tools_events( first_tool = mcp_tools_with_litellm_proxy[0] if isinstance(first_tool, dict): server_label_value = first_tool.get("server_label", "") - server_label = str(server_label_value) if server_label_value is not None else "" - + server_label = ( + str(server_label_value) if server_label_value is not None else "" + ) + # Format tools for OpenAI output_item.done format formatted_tools = [] for tool in filtered_mcp_tools: tool_dict = { - "name": getattr(tool, 'name', 'unknown'), - "description": getattr(tool, 'description', ''), + "name": getattr(tool, "name", "unknown"), + "description": getattr(tool, "description", ""), "annotations": {"read_only": False}, } - + # Add input_schema if available - if hasattr(tool, 'inputSchema'): - tool_dict["input_schema"] = getattr(tool, 'inputSchema') - elif hasattr(tool, 'input_schema'): - tool_dict["input_schema"] = getattr(tool, 'input_schema') - + if hasattr(tool, "inputSchema"): + tool_dict["input_schema"] = getattr(tool, "inputSchema") + elif hasattr(tool, "input_schema"): + tool_dict["input_schema"] = getattr(tool, "input_schema") + formatted_tools.append(tool_dict) - + # Create the output_item.done event with MCP tools list output_item_done_event = OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=0, - item={ - "id": base_item_id, - "type": "mcp_list_tools", - "server_label": server_label, - "tools": formatted_tools - } + item=BaseLiteLLMOpenAIResponseObject( + **{ + "id": base_item_id, + "type": "mcp_list_tools", + "server_label": server_label, + "tools": formatted_tools, + } + ), ) events.append(output_item_done_event) - + verbose_logger.debug(f"Created {len(events)} MCP discovery events") - + except Exception as e: verbose_logger.error(f"Error creating MCP list tools events: {e}") import traceback + traceback.print_exc() - + # Emit failed event on error failed_event = MCPListToolsFailedEvent( type=ResponsesAPIStreamEvents.MCP_LIST_TOOLS_FAILED, @@ -142,37 +140,39 @@ async def create_mcp_list_tools_events( item_id=base_item_id, ) events.append(failed_event) - + # Still emit output_item.done event even on failure (with empty tools list) from litellm.types.llms.openai import OutputItemDoneEvent - + output_item_done_event = OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=0, - item={ - "id": base_item_id, - "type": "mcp_list_tools", - "server_label": "", - "tools": [] - } + item=BaseLiteLLMOpenAIResponseObject( + **{ + "id": base_item_id, + "type": "mcp_list_tools", + "server_label": "", + "tools": [], + } + ), ) events.append(output_item_done_event) - + return events def create_mcp_call_events( - tool_name: str, - tool_call_id: str, + tool_name: str, + tool_call_id: str, arguments: str, result: Optional[str] = None, base_item_id: Optional[str] = None, - sequence_start: int = 1 + sequence_start: int = 1, ) -> List[ResponsesAPIStreamingResponse]: """Create MCP call events following OpenAI's specification""" events: List[ResponsesAPIStreamingResponse] = [] item_id = base_item_id or f"mcp_{uuid.uuid4().hex[:8]}" - + # MCP call in progress event in_progress_event = MCPCallInProgressEvent( type=ResponsesAPIStreamEvents.MCP_CALL_IN_PROGRESS, @@ -181,7 +181,7 @@ def create_mcp_call_events( item_id=item_id, ) events.append(in_progress_event) - + # MCP call arguments delta event (streaming the arguments) arguments_delta_event = MCPCallArgumentsDeltaEvent( type=ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DELTA, @@ -191,7 +191,7 @@ def create_mcp_call_events( sequence_number=sequence_start + 1, ) events.append(arguments_delta_event) - + # MCP call arguments done event arguments_done_event = MCPCallArgumentsDoneEvent( type=ResponsesAPIStreamEvents.MCP_CALL_ARGUMENTS_DONE, @@ -201,7 +201,7 @@ def create_mcp_call_events( sequence_number=sequence_start + 2, ) events.append(arguments_done_event) - + # MCP call completed event (or failed if result indicates failure) if result is not None: completed_event = MCPCallCompletedEvent( @@ -211,23 +211,25 @@ def create_mcp_call_events( output_index=0, ) events.append(completed_event) - + # Add output_item.done event with the tool call result from litellm.types.llms.openai import OutputItemDoneEvent - + output_item_done_event = OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=0, - item={ - "id": item_id, - "type": "mcp_call", - "approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}", - "arguments": arguments, - "error": None, - "name": tool_name, - "output": result, - "server_label": "litellm" - }, + item=BaseLiteLLMOpenAIResponseObject( + **{ + "id": item_id, + "type": "mcp_call", + "approval_request_id": f"mcpr_{uuid.uuid4().hex[:8]}", + "arguments": arguments, + "error": None, + "name": tool_name, + "output": result, + "server_label": "litellm", + } + ), ) events.append(output_item_done_event) else: @@ -238,7 +240,7 @@ def create_mcp_call_events( output_index=0, ) events.append(failed_event) - + return events @@ -250,51 +252,60 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): 3. Handles tool execution and follow-up calls for auto-execute tools 4. Emits tool execution events in the stream """ - + def __init__( self, base_iterator: Any, # Can be None - will be created internally mcp_events: List[ResponsesAPIStreamingResponse], mcp_tools_with_litellm_proxy: Optional[List[Any]] = None, user_api_key_auth: Any = None, - original_request_params: Optional[Dict[str, Any]] = None + original_request_params: Optional[Dict[str, Any]] = None, ): # MCP setup self.mcp_tools_with_litellm_proxy = mcp_tools_with_litellm_proxy or [] self.user_api_key_auth = user_api_key_auth self.original_request_params = original_request_params or {} self.should_auto_execute = self._should_auto_execute_tools() - + # Streaming state management self.phase = "mcp_discovery" # mcp_discovery -> initial_response -> tool_execution -> follow_up_response -> finished self.finished = False - + # Event queues and generation flags - self.mcp_discovery_events: List[ResponsesAPIStreamingResponse] = mcp_events # Pre-generated MCP discovery events + self.mcp_discovery_events: List[ResponsesAPIStreamingResponse] = ( + mcp_events # Pre-generated MCP discovery events + ) self.tool_execution_events: List[ResponsesAPIStreamingResponse] = [] self.mcp_discovery_generated = True # Events are already generated - self.mcp_events = mcp_events # Store the initial MCP events for backward compatibility - + self.mcp_events = ( + mcp_events # Store the initial MCP events for backward compatibility + ) + # Iterator references - self.base_iterator: Optional[Union[Any, ResponsesAPIResponse]] = base_iterator # Will be created when needed + self.base_iterator: Optional[Union[Any, ResponsesAPIResponse]] = ( + base_iterator # Will be created when needed + ) self.follow_up_iterator: Optional[Any] = None - + # Response collection for tool execution self.collected_response: Optional[ResponsesAPIResponse] = None - + # Set up model metadata (will be updated when we get the real iterator) - self.model = self.original_request_params.get('model', 'unknown') + self.model = self.original_request_params.get("model", "unknown") self.litellm_metadata = {} - self.custom_llm_provider = self.original_request_params.get('custom_llm_provider', None) - + self.custom_llm_provider = self.original_request_params.get( + "custom_llm_provider", None + ) + # Mark as async iterator self.is_async = True - + def _should_auto_execute_tools(self) -> bool: """Check if tools should be auto-executed""" from litellm.responses.mcp.litellm_proxy_mcp_handler import ( LiteLLM_Proxy_MCP_Handler, ) + return LiteLLM_Proxy_MCP_Handler._should_auto_execute_tools( self.mcp_tools_with_litellm_proxy ) @@ -306,45 +317,49 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): """ Phase-based streaming: 1. mcp_discovery - Emit MCP discovery events - 2. initial_response - Stream the first LLM response + 2. initial_response - Stream the first LLM response 3. tool_execution - Emit tool execution events 4. follow_up_response - Stream the follow-up response 5. finished - End iteration """ - + # Phase 1: MCP Discovery Events if self.phase == "mcp_discovery": # Generate MCP discovery events if not already done # MCP discovery events are already generated and available - + # Emit MCP discovery events if self.mcp_discovery_events: return self.mcp_discovery_events.pop(0) - + # All MCP discovery events emitted, move to next phase - verbose_logger.debug("MCP discovery phase complete, transitioning to initial_response") + verbose_logger.debug( + "MCP discovery phase complete, transitioning to initial_response" + ) self.phase = "initial_response" await self._create_initial_response_iterator() # Fall through to process the initial response immediately - + # Phase 2: Initial Response Stream if self.phase == "initial_response": if self.base_iterator: # Check if base_iterator is actually iterable - if hasattr(self.base_iterator, '__anext__'): + if hasattr(self.base_iterator, "__anext__"): try: chunk = await cast(Any, self.base_iterator).__anext__() # type: ignore[attr-defined] - + # If auto-execution is enabled, check for completed responses - if self.should_auto_execute and self._is_response_completed(chunk): + if self.should_auto_execute and self._is_response_completed( + chunk + ): # Collect the response for tool execution - response_obj = getattr(chunk, 'response', None) + 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 except StopAsyncIteration: # Initial response ended, move to next phase @@ -357,24 +372,26 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): else: # base_iterator is not async iterable (likely a ResponsesAPIResponse) # Collect it for tool execution if needed - if self.should_auto_execute and isinstance(self.base_iterator, ResponsesAPIResponse): + if self.should_auto_execute and isinstance( + self.base_iterator, ResponsesAPIResponse + ): self.collected_response = self.base_iterator self.phase = "tool_execution" await self._generate_tool_execution_events() else: self.phase = "finished" raise StopAsyncIteration - + # Phase 3: Tool Execution Events if self.phase == "tool_execution": # Emit any queued tool execution events if self.tool_execution_events: return self.tool_execution_events.pop(0) - + # Move to follow-up response phase self.phase = "follow_up_response" await self._create_follow_up_iterator() - + # Phase 4: Follow-up Response Stream if self.phase == "follow_up_response": if self.follow_up_iterator: @@ -386,20 +403,22 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): else: self.phase = "finished" raise StopAsyncIteration - + # Phase 5: Finished if self.phase == "finished": raise StopAsyncIteration - + # Should not reach here raise StopAsyncIteration - + def _is_response_completed(self, chunk: ResponsesAPIStreamingResponse) -> bool: """Check if this chunk indicates the response is completed""" from litellm.types.llms.openai import ResponsesAPIStreamEvents - return getattr(chunk, 'type', None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED - - + + return ( + getattr(chunk, "type", None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED + ) + async def _create_initial_response_iterator(self) -> None: """Create the initial response iterator by making the first LLM call""" try: @@ -408,38 +427,45 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): # Make the initial response API call - but avoid the MCP wrapper params = self.original_request_params.copy() - params['stream'] = True # Ensure streaming - + params["stream"] = True # Ensure streaming + # Use the pre-fetched all_tools from original_request_params (no re-processing needed) params_for_llm = {} for key, value in params.items(): - params_for_llm[key] = value # Copy all params as-is since tools are already processed - - tools_count = len(params_for_llm.get('tools', [])) + params_for_llm[key] = ( + value # Copy all params as-is since tools are already processed + ) + + tools_count = len(params_for_llm.get("tools", [])) verbose_logger.debug(f"Making LLM call with {tools_count} tools") response = await aresponses(**params_for_llm) - + # Set the base iterator - if hasattr(response, '__aiter__') or hasattr(response, '__iter__'): + if hasattr(response, "__aiter__") or hasattr(response, "__iter__"): self.base_iterator = response # Copy metadata from the real iterator - self.model = getattr(response, 'model', self.model) - self.litellm_metadata = getattr(response, 'litellm_metadata', {}) - self.custom_llm_provider = getattr(response, 'custom_llm_provider', self.custom_llm_provider) - verbose_logger.debug(f"Created base iterator: {type(self.base_iterator)}") + self.model = getattr(response, "model", self.model) + self.litellm_metadata = getattr(response, "litellm_metadata", {}) + self.custom_llm_provider = getattr( + response, "custom_llm_provider", self.custom_llm_provider + ) + verbose_logger.debug( + f"Created base iterator: {type(self.base_iterator)}" + ) else: # Non-streaming response - this shouldn't happen but handle it verbose_logger.warning(f"Got non-streaming response: {type(response)}") self.base_iterator = None self.phase = "finished" - + except Exception as e: verbose_logger.error(f"Error creating initial response iterator: {e}") import traceback + traceback.print_exc() self.base_iterator = None self.phase = "finished" - + async def _generate_tool_execution_events(self) -> None: """Generate tool execution events and execute tools""" if not self.collected_response: @@ -447,7 +473,7 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): from litellm.responses.mcp.litellm_proxy_mcp_handler import ( LiteLLM_Proxy_MCP_Handler, ) - + try: # Extract tool calls from the response if self.collected_response is not None: @@ -456,9 +482,11 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): tool_calls = [] if not tool_calls: return - + for tool_call in tool_calls: - tool_name, tool_arguments, tool_call_id = LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call) + tool_name, tool_arguments, tool_call_id = ( + LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call) + ) if tool_name and tool_call_id: # Create MCP call events for this tool execution call_events = create_mcp_call_events( @@ -467,34 +495,35 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): arguments=tool_arguments or "{}", # JSON string with arguments result=None, # Will be set after execution base_item_id=f"mcp_{uuid.uuid4().hex[:8]}", - sequence_start=len(self.tool_execution_events) + 1 + sequence_start=len(self.tool_execution_events) + 1, ) # Add the in_progress and arguments events (not the completed event yet) self.tool_execution_events.extend(call_events[:-1]) - + # Execute the tools tool_results = await LiteLLM_Proxy_MCP_Handler._execute_tool_calls( - tool_calls=tool_calls, - user_api_key_auth=self.user_api_key_auth + tool_calls=tool_calls, user_api_key_auth=self.user_api_key_auth ) - + # Create completion events and output_item.done events for tool execution for tool_result in tool_results: tool_call_id = tool_result.get("tool_call_id", "unknown") result_text = tool_result.get("result", "") - + # Find matching tool name and arguments tool_name = "unknown" tool_arguments = "{}" for tool_call in tool_calls: - name, args, call_id = LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call) + name, args, call_id = ( + LiteLLM_Proxy_MCP_Handler._extract_tool_call_details(tool_call) + ) if call_id == tool_call_id: tool_name = name or "unknown" tool_arguments = args or "{}" break - + item_id = f"mcp_{uuid.uuid4().hex[:8]}" - + # Create the completion event completed_event = MCPCallCompletedEvent( type=ResponsesAPIStreamEvents.MCP_CALL_COMPLETED, @@ -503,79 +532,84 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): output_index=0, ) self.tool_execution_events.append(completed_event) - + # Create output_item.done event with the tool call result from litellm.types.llms.openai import OutputItemDoneEvent - + output_item_done_event = OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=0, - item={ - "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 - }, + 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 + } + ), ) self.tool_execution_events.append(output_item_done_event) - + # Store tool results for follow-up call self.tool_results = tool_results - + except Exception as e: verbose_logger.error(f"Error in tool execution: {e}") import traceback + traceback.print_exc() self.tool_results = [] - + async def _create_follow_up_iterator(self) -> None: """Create the follow-up response iterator with tool results""" - if not self.collected_response or not hasattr(self, 'tool_results'): + if not self.collected_response or not hasattr(self, "tool_results"): return - + from litellm.responses.main import aresponses from litellm.responses.mcp.litellm_proxy_mcp_handler import ( LiteLLM_Proxy_MCP_Handler, ) - + try: # Create follow-up input if self.collected_response is not None: follow_up_input = LiteLLM_Proxy_MCP_Handler._create_follow_up_input( response=self.collected_response, # type: ignore[arg-type] tool_results=self.tool_results, - original_input=self.original_request_params.get('input') + original_input=self.original_request_params.get("input"), ) - + # Make follow-up call with streaming follow_up_params = self.original_request_params.copy() - follow_up_params.update({ - 'input': follow_up_input, - 'previous_response_id': self.collected_response.id, # type: ignore[attr-defined] - 'stream': True - }) + follow_up_params.update( + { + "input": follow_up_input, + "previous_response_id": self.collected_response.id, # type: ignore[attr-defined] + "stream": True, + } + ) else: return # Remove tool_choice to avoid forcing more tool calls - follow_up_params.pop('tool_choice', None) - + follow_up_params.pop("tool_choice", None) + follow_up_response = await aresponses(**follow_up_params) - + # Set up the follow-up iterator - if hasattr(follow_up_response, '__aiter__'): + if hasattr(follow_up_response, "__aiter__"): self.follow_up_iterator = follow_up_response - + except Exception as e: verbose_logger.error(f"Error creating follow-up iterator: {e}") import traceback + traceback.print_exc() self.follow_up_iterator = None - def __iter__(self): return self @@ -583,11 +617,11 @@ class MCPEnhancedStreamingIterator(BaseResponsesAPIStreamingIterator): # First, emit any queued MCP events if self.mcp_events: # type: ignore[attr-defined] return self.mcp_events.pop(0) # type: ignore[attr-defined] - + # Then delegate to the base iterator if not self.is_async: try: - if self.base_iterator and hasattr(self.base_iterator, '__next__'): + if self.base_iterator and hasattr(self.base_iterator, "__next__"): return next(cast(Any, self.base_iterator)) # type: ignore[arg-type] else: raise StopIteration diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index eda3e6921da..b78913402ff 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -93,24 +93,35 @@ class BaseResponsesAPIStreamingIterator: # Store the completed response if ( openai_responses_api_chunk - and openai_responses_api_chunk.type + and getattr(openai_responses_api_chunk, "type", None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED ): self.completed_response = openai_responses_api_chunk # Add cost to usage object if include_cost_in_streaming_usage is True - if litellm.include_cost_in_streaming_usage and self.logging_obj is not None: - response_obj: Optional[ResponsesAPIResponse] = getattr(openai_responses_api_chunk, "response", None) + if ( + litellm.include_cost_in_streaming_usage + and self.logging_obj is not None + ): + response_obj: Optional[ResponsesAPIResponse] = getattr( + openai_responses_api_chunk, "response", None + ) if response_obj: - usage_obj: Optional[ResponseAPIUsage] = getattr(response_obj, "usage", None) + usage_obj: Optional[ResponseAPIUsage] = getattr( + response_obj, "usage", None + ) if usage_obj is not None: try: - cost: Optional[float] = self.logging_obj._response_cost_calculator(result=response_obj) + cost: Optional[float] = ( + self.logging_obj._response_cost_calculator( + result=response_obj + ) + ) if cost is not None: setattr(usage_obj, "cost", cost) except Exception: # If cost calculation fails, continue without cost pass - + self._handle_logging_completed_response() return openai_responses_api_chunk diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 005aec31e2e..c9761beeafc 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -1,3 +1,4 @@ +import uuid from enum import Enum from os import PathLike from typing import IO, Any, Iterable, List, Literal, Mapping, Optional, Tuple, Union @@ -44,7 +45,7 @@ from openai.types.responses.response import ( # Handle OpenAI SDK version compatibility for Text type try: # fmt: off - from openai.types.responses.response_create_params import ( # type: ignore[attr-defined] + from openai.types.responses.response_create_params import ( Text as ResponseText, # type: ignore[attr-defined] ) @@ -992,7 +993,9 @@ class ResponsesAPIOptionalRequestParams(TypedDict, total=False): prompt_cache_key: Optional[str] stream_options: Optional[dict] top_logprobs: Optional[int] - partial_images: Optional[int] # Number of partial images to generate (1-3) for streaming image generation + partial_images: Optional[ + int + ] # Number of partial images to generate (1-3) for streaming image generation class ResponsesAPIRequestParams(ResponsesAPIOptionalRequestParams, total=False): @@ -1056,7 +1059,9 @@ class ResponsesAPIResponse(BaseLiteLLMOpenAIResponseObject): parallel_tool_calls: Optional[bool] = None temperature: Optional[float] = None tool_choice: Optional[ToolChoice] = None - tools: Optional[Union[List[Tool], List[ResponseFunctionToolCall], List[Dict[str, Any]]]] = None + tools: Optional[ + Union[List[Tool], List[ResponseFunctionToolCall], List[Dict[str, Any]]] + ] = None top_p: Optional[float] = None max_output_tokens: Optional[int] = None previous_response_id: Optional[str] = None @@ -1180,13 +1185,27 @@ class ReasoningSummaryTextDeltaEvent(BaseLiteLLMOpenAIResponseObject): class OutputItemAddedEvent(BaseLiteLLMOpenAIResponseObject): type: Literal[ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED] output_index: int - item: Optional[dict] + item: Optional[BaseLiteLLMOpenAIResponseObject] class OutputItemDoneEvent(BaseLiteLLMOpenAIResponseObject): type: Literal[ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE] output_index: int - item: dict + sequence_number: int = 1 + item: BaseLiteLLMOpenAIResponseObject + + +class OpenAIChatCompletionLogprobsContentTopLogprobs(TypedDict, total=False): + bytes: List + logprob: Required[float] + token: Required[str] + + +class OpenAIChatCompletionLogprobsContent(TypedDict, total=False): + bytes: List + logprob: Required[float] + token: Required[str] + top_logprobs: List[OpenAIChatCompletionLogprobsContentTopLogprobs] class ContentPartAddedEvent(BaseLiteLLMOpenAIResponseObject): @@ -1194,7 +1213,31 @@ class ContentPartAddedEvent(BaseLiteLLMOpenAIResponseObject): item_id: str output_index: int content_index: int - part: dict + part: BaseLiteLLMOpenAIResponseObject + + +class ContentPartDonePartOutputText(BaseLiteLLMOpenAIResponseObject): + type: Literal["output_text"] + text: str + annotations: List[BaseLiteLLMOpenAIResponseObject] + logprobs: Optional[List[OpenAIChatCompletionLogprobsContent]] + + +class ContentPartDonePartRefusal(BaseLiteLLMOpenAIResponseObject): + type: Literal["refusal"] + refusal: str + + +class ContentPartDonePartReasoningText(BaseLiteLLMOpenAIResponseObject): + type: Literal["reasoning_text"] + reasoning: str + + +PART_UNION_TYPES = Union[ + ContentPartDonePartOutputText, + ContentPartDonePartRefusal, + ContentPartDonePartReasoningText, +] class ContentPartDoneEvent(BaseLiteLLMOpenAIResponseObject): @@ -1202,7 +1245,7 @@ class ContentPartDoneEvent(BaseLiteLLMOpenAIResponseObject): item_id: str output_index: int content_index: int - part: dict + part: PART_UNION_TYPES class OutputTextDeltaEvent(BaseLiteLLMOpenAIResponseObject): @@ -1413,6 +1456,7 @@ ResponsesAPIStreamingResponse = Annotated[ ImageGenerationPartialImageEvent, ErrorEvent, GenericEvent, + BaseLiteLLMOpenAIResponseObject, ], Discriminator("type"), ] @@ -1731,19 +1775,6 @@ class OpenAIModerationResponse(BaseLiteLLMOpenAIResponseObject): _hidden_params: dict = PrivateAttr(default_factory=dict) -class OpenAIChatCompletionLogprobsContentTopLogprobs(TypedDict, total=False): - bytes: List - logprob: Required[float] - token: Required[str] - - -class OpenAIChatCompletionLogprobsContent(TypedDict, total=False): - bytes: List - logprob: Required[float] - token: Required[str] - top_logprobs: List[OpenAIChatCompletionLogprobsContentTopLogprobs] - - class OpenAIChatCompletionLogprobs(TypedDict, total=False): content: List[OpenAIChatCompletionLogprobsContent] refusal: List[OpenAIChatCompletionLogprobsContent] diff --git a/tests/llm_translation/test_anthropic_completion.py b/tests/llm_translation/test_anthropic_completion.py index 4eeb80c4191..7d9ef56f164 100644 --- a/tests/llm_translation/test_anthropic_completion.py +++ b/tests/llm_translation/test_anthropic_completion.py @@ -361,6 +361,7 @@ def test_process_anthropic_headers_with_no_matching_headers(): def test_anthropic_tool_use(tool_type, tool_config, message_content): """Test Anthropic tool use with computer use and web fetch tools.""" from litellm import completion + litellm._turn_on_debug() tools = [tool_config] @@ -1518,3 +1519,126 @@ def test_anthropic_streaming(): role_set_count += 1 assert role_set_count == 1 + + +def test_anthropic_via_responses_api(): + from litellm.types.llms.openai import ResponsesAPIStreamEvents + + response = litellm.responses( + model="anthropic/claude-sonnet-4-5", + input="Who won the World Cup in 2022?", + max_output_tokens=100, + stream=True, + ) + + assert response is not None + + # Expected event sequence + expected_events = [ + ResponsesAPIStreamEvents.RESPONSE_CREATED, + ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, + ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + ResponsesAPIStreamEvents.CONTENT_PART_ADDED, + ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, # Can occur multiple times + ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + ResponsesAPIStreamEvents.CONTENT_PART_DONE, + ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + ResponsesAPIStreamEvents.RESPONSE_COMPLETED, + ] + + events_seen = [] + text_delta_count = 0 + + for chunk in response: + print(f"chunk: {chunk}") + + # Each chunk should have a type attribute + assert hasattr(chunk, "type"), f"Chunk missing 'type' attribute: {chunk}" + + event_type = chunk.type + + # Track events seen + if event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA: + text_delta_count += 1 + if ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA not in events_seen: + events_seen.append(event_type) + else: + events_seen.append(event_type) + + # Assert specific structures for each event type + if event_type == ResponsesAPIStreamEvents.RESPONSE_CREATED: + assert chunk.type == ResponsesAPIStreamEvents.RESPONSE_CREATED + assert hasattr(chunk, "response") + assert chunk.response.status == "in_progress" + assert hasattr(chunk.response, "id") + assert hasattr(chunk.response, "model") + + elif event_type == ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS: + assert chunk.type == ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS + assert hasattr(chunk, "response") + assert chunk.response.status == "in_progress" + + elif event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED: + assert chunk.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED + assert hasattr(chunk, "output_index") + assert hasattr(chunk, "item") + assert chunk.item.type == "message" + assert chunk.item.role == "assistant" + + elif event_type == ResponsesAPIStreamEvents.CONTENT_PART_ADDED: + assert chunk.type == ResponsesAPIStreamEvents.CONTENT_PART_ADDED + assert hasattr(chunk, "item_id") + assert hasattr(chunk, "output_index") + assert hasattr(chunk, "content_index") + assert hasattr(chunk, "part") + assert chunk.part.type == "output_text" + + elif event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA: + assert chunk.type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA + assert hasattr(chunk, "item_id") + assert hasattr(chunk, "output_index") + assert hasattr(chunk, "content_index") + assert hasattr(chunk, "delta") + assert isinstance(chunk.delta, str) + + elif event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: + assert chunk.type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE + assert hasattr(chunk, "item_id") + assert hasattr(chunk, "output_index") + assert hasattr(chunk, "content_index") + assert hasattr(chunk, "text") + + elif event_type == ResponsesAPIStreamEvents.CONTENT_PART_DONE: + assert chunk.type == ResponsesAPIStreamEvents.CONTENT_PART_DONE + assert hasattr(chunk, "item_id") + assert hasattr(chunk, "output_index") + assert hasattr(chunk, "content_index") + assert hasattr(chunk, "part") + assert chunk.part.type == "output_text" + + elif event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: + assert chunk.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE + assert hasattr(chunk, "output_index") + assert hasattr(chunk, "item") + assert chunk.item.status == "completed" + + elif event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED: + assert chunk.type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED + assert hasattr(chunk, "response") + assert chunk.response.status == "completed" + assert hasattr(chunk.response, "usage") + assert hasattr(chunk.response, "output") + + # Assert we saw all expected events + print(f"Events seen: {events_seen}") + assert ( + events_seen == expected_events + ), f"Event sequence mismatch. Expected: {expected_events}, Got: {events_seen}" + + # Assert we saw at least one text delta + assert ( + text_delta_count > 0 + ), f"Expected at least one response.output_text.delta event, got {text_delta_count}" + + print(f"✓ All {len(events_seen)} events matched expected structure") + print(f"✓ Received {text_delta_count} text delta chunks") diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_reasoning_content_transformation.py b/tests/test_litellm/responses/litellm_completion_transformation/test_reasoning_content_transformation.py index 5323589818b..d1926bbc93f 100644 --- a/tests/test_litellm/responses/litellm_completion_transformation/test_reasoning_content_transformation.py +++ b/tests/test_litellm/responses/litellm_completion_transformation/test_reasoning_content_transformation.py @@ -4,14 +4,20 @@ Test reasoning content preservation in Responses API transformation from unittest.mock import AsyncMock -from litellm.types.utils import ModelResponseStream, StreamingChoices, Delta from litellm.responses.litellm_completion_transformation.streaming_iterator import ( LiteLLMCompletionStreamingIterator, ) from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) -from litellm.types.utils import ModelResponse, Choices, Message +from litellm.types.utils import ( + Choices, + Delta, + Message, + ModelResponse, + ModelResponseStream, + StreamingChoices, +) class TestReasoningContentStreaming: @@ -41,6 +47,7 @@ class TestReasoningContentStreaming: mock_stream = AsyncMock() iterator = LiteLLMCompletionStreamingIterator( + model="test-model", litellm_custom_stream_wrapper=mock_stream, request_input="Test input", responses_api_request={}, @@ -78,6 +85,7 @@ class TestReasoningContentStreaming: mock_stream = AsyncMock() iterator = LiteLLMCompletionStreamingIterator( + model="test-model", litellm_custom_stream_wrapper=mock_stream, request_input="Test input", responses_api_request={}, @@ -114,6 +122,7 @@ class TestReasoningContentStreaming: mock_stream = AsyncMock() iterator = LiteLLMCompletionStreamingIterator( + model="test-model", litellm_custom_stream_wrapper=mock_stream, request_input="Test input", responses_api_request={},