""" Handler for transforming /chat/completions api requests to litellm.responses requests """ import json import os from collections.abc import AsyncIterator, Callable, Iterable, Iterator, Mapping, Sequence from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict, Union, cast, get_args from openai.types.responses.custom_tool_param import CustomToolParam from openai.types.responses.response_input_param import ( FunctionCallOutput, ResponseCustomToolCallOutputParam, ResponseCustomToolCallParam, ) from openai.types.responses.tool_choice_custom_param import ToolChoiceCustomParam from openai.types.responses.tool_choice_function_param import ToolChoiceFunctionParam from openai.types.responses.tool_param import FunctionToolParam from pydantic import BaseModel import litellm from litellm import ModelResponse from litellm._logging import verbose_logger from litellm.litellm_core_utils.prompt_templates.common_utils import ( responses_reasoning_item_from_thinking_blocks, ) from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator from litellm.llms.base_llm.bridges.completion_transformation import ( CompletionTransformationBridge, ) from litellm.responses.sse_output_recovery import ( parse_sse_json_chunk, record_output_item_chunk, record_output_text_chunk, ) from litellm.responses.utils import normalize_responses_api_stream_options from litellm.types.llms.openai import ( REASONING_EFFORT, ChatCompletionAnnotation, ChatCompletionReasoningItem, ChatCompletionToolCallChunk, ChatCompletionToolCallFunctionChunk, ChatCompletionToolParamFunctionChunk, Reasoning, ResponsesAPIOptionalRequestParams, ResponsesAPIStreamEvents, ) from litellm.types.utils import GenericStreamingChunk, ModelResponseStream if TYPE_CHECKING: from openai.types.responses import ResponseInputImageParam from openai.types.responses.response_text_config_param import ( ResponseTextConfigParam as ResponseText, ) from pydantic import BaseModel from litellm import LiteLLMLoggingObj, ModelResponse from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator from litellm.types.llms.openai import ( ALL_RESPONSES_API_TOOL_PARAMS, AllMessageValues, ChatCompletionFileObject, ChatCompletionImageObject, ChatCompletionRedactedThinkingBlock, ChatCompletionThinkingBlock, ChatCompletionToolReferenceObject, OpenAIMessageContentListBlock, ) from litellm.types.utils import Choices class _ReasoningSummaryText(TypedDict): type: str text: str class _BuiltReasoningItem(TypedDict): type: Literal["reasoning"] id: str encrypted_content: str | None summary: Sequence[_ReasoningSummaryText] def _get_reasoning_items( msg: "AllMessageValues", ) -> list[ChatCompletionReasoningItem]: """Extract reasoning_items from a message dict with proper typing.""" items: Final = msg.get("reasoning_items") if items: return items return [] def _reasoning_input_items(msg: "AllMessageValues") -> list[dict[str, object]]: # mutable-ok: API message payload """Reasoning input items for an assistant message. Stored reasoning items win because they carry an id the Responses API minted; thinking blocks are the fallback for turns that arrived over another API surface. """ items: Final = _get_reasoning_items(msg) stored: Final = [_reasoning_item_to_response_input(item) for item in items] # mutable-ok: API message payload if stored: return stored raw_blocks: Final = msg.get("thinking_blocks") or () blocks: Final = cast("Iterable[ChatCompletionThinkingBlock]", raw_blocks) # cast-ok: untyped client json from_thinking: Final = responses_reasoning_item_from_thinking_blocks(blocks) return [] if from_thinking is None else [dict(from_thinking)] # mutable-ok: API message payload def _build_reasoning_item( item_id: str, encrypted_content: str | None, summary_raw: Iterable[object] | None, ) -> _BuiltReasoningItem: """Build a ChatCompletionReasoningItem-shaped dict from raw response data. Handles both pydantic objects (attribute access) and plain dicts. """ summary: Final[list[_ReasoningSummaryText]] = [] for s in summary_raw or []: if isinstance(s, dict): summary.append({"type": s.get("type", "summary_text"), "text": s.get("text", "")}) else: summary.append( { "type": getattr(s, "type", "summary_text"), "text": getattr(s, "text", ""), } ) return { "id": item_id, "type": "reasoning", "encrypted_content": encrypted_content, "summary": summary, } def _reasoning_item_from_output_item(item: object) -> _BuiltReasoningItem | None: from openai.types.responses import ResponseReasoningItem if isinstance(item, ResponseReasoningItem): return _build_reasoning_item( item_id=item.id, encrypted_content=getattr(item, "encrypted_content", None), summary_raw=item.summary, ) if isinstance(item, dict) and item.get("type") == "reasoning": return _build_reasoning_item( item_id=item.get("id", ""), encrypted_content=item.get("encrypted_content"), summary_raw=item.get("summary"), ) return None def _reasoning_items_from_output_items(output_items: Sequence[object]) -> tuple[_BuiltReasoningItem, ...]: return tuple( reasoning_item for reasoning_item in (_reasoning_item_from_output_item(item) for item in output_items) if reasoning_item is not None ) def _as_chat_reasoning_items( reasoning_items: Sequence[_BuiltReasoningItem], ) -> list[ChatCompletionReasoningItem] | None: if not reasoning_items: return None # cast-ok: _BuiltReasoningItem is the structural shape ChatCompletionReasoningItem # describes, and TypedDict invariance is what stops the two from unifying here. return cast(list[ChatCompletionReasoningItem], list(reasoning_items)) def _map_incomplete_reason_to_finish_reason(incomplete_reason: str | None) -> Literal["length", "content_filter"]: if incomplete_reason == "content_filter": return "content_filter" return "length" def _input_file_from_file_value(file_value: object) -> dict[str, object]: if not isinstance(file_value, dict): return {"type": "input_file"} file_dict: Final = cast("dict[str, object]", file_value) # cast-ok: runtime dict checked return { "type": "input_file", **{key: file_dict[key] for key in ("file_id", "file_data", "filename") if key in file_dict}, } def _incomplete_reason_from_response_payload(response_payload: object) -> str | None: if not isinstance(response_payload, Mapping): return None incomplete_details: Final = response_payload.get("incomplete_details") if not isinstance(incomplete_details, Mapping): return None reason: Final = incomplete_details.get("reason") return reason if isinstance(reason, str) else None class _ChatToolCallDict(ChatCompletionToolCallChunk, total=False): provider_specific_fields: Mapping[str, object] def _tool_call_dict_from_output_item(item: Mapping[str, Any], index: int) -> _ChatToolCallDict: """Convert a ``function_call`` or ``custom_tool_call`` output item dict to a chat completions tool_call dict. Custom (grammar/freeform) tool calls carry their raw string payload in ``input`` rather than ``arguments``; both map to ``function.arguments`` so chat clients (e.g. Cursor agent mode) receive them like any other tool call. The single conversion rule shared by the non-streaming accumulator and the streaming ``output_item.added`` branch.""" from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) is_custom: Final = item.get("type") == "custom_tool_call" arguments: Final = (item.get("input") if is_custom else item.get("arguments")) or "" name: Final = item.get("name") or ("custom_tool" if is_custom else "") function_chunk: Final = ChatCompletionToolCallFunctionChunk(name=name, arguments=arguments) tool_call_dict: Final = _ChatToolCallDict( id=LiteLLMCompletionResponsesConfig._tool_call_id_from_responses_item(item.get("id"), item.get("call_id")), type="function", function=function_chunk, index=index, ) raw_provider_fields: Final = item.get("provider_specific_fields") if isinstance(raw_provider_fields, dict): provider_specific_fields = raw_provider_fields elif raw_provider_fields and hasattr(raw_provider_fields, "__dict__"): provider_specific_fields = vars(raw_provider_fields) else: provider_specific_fields = None if provider_specific_fields: tool_call_dict["provider_specific_fields"] = provider_specific_fields function_chunk["provider_specific_fields"] = provider_specific_fields return tool_call_dict def _flat_responses_tool_choice(choice_type: str, name: str) -> ToolChoiceFunctionParam | ToolChoiceCustomParam: if choice_type == "custom": return ToolChoiceCustomParam(type="custom", name=name) return ToolChoiceFunctionParam(type="function", name=name) def _reasoning_item_to_response_input( r_item: ChatCompletionReasoningItem, ) -> dict[str, object]: """Convert a stored ChatCompletionReasoningItem back to a Responses API input item.""" r_input: Final[dict[str, object]] = { "type": "reasoning", "id": r_item.get("id") or f"rs_{id(r_item)}", # summary is always required by the Responses API, even when empty "summary": r_item.get("summary") or [], } if r_item.get("encrypted_content"): r_input["encrypted_content"] = r_item["encrypted_content"] return r_input class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): """ Handler for transforming /chat/completions api requests to litellm.responses requests """ def __init__(self): pass def _normalize_tool_choice_for_responses_api(self, tool_choice: Any) -> Any: """Chat tool_choice nests the name under function/custom; Responses API expects top-level name.""" if not isinstance(tool_choice, dict): return tool_choice choice_type: Final = tool_choice.get("type") if isinstance(choice_type, str) and choice_type in ("auto", "none", "required"): return choice_type if choice_type not in ("function", "custom"): return tool_choice if isinstance(tool_choice.get("name"), str) and tool_choice.get("name"): # Return only Responses shape so stray chat ``function``/``custom`` keys are not sent upstream. return _flat_responses_tool_choice(choice_type, tool_choice["name"]) nested: Final = tool_choice.get(choice_type) if isinstance(nested, dict): nested_name: Final = nested.get("name") if isinstance(nested_name, str) and nested_name: return _flat_responses_tool_choice(choice_type, nested_name) return tool_choice def _handle_raw_dict_response_item(self, item: dict[str, Any], index: int) -> tuple["Choices | None", 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: Final = 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: Final = item.get("content", []) for content_item in content_list: if isinstance(content_item, dict): content_type = content_item.get("type") if content_type == "output_text": response_text = content_item.get("text", "") # Extract annotations from content if present annotations = LiteLLMResponsesTransformationHandler._convert_annotations_to_chat_format( content_item.get("annotations", None) ) msg = Message( role=item.get("role", "assistant"), content=response_text if response_text else "", annotations=annotations, ) choice = Choices(message=msg, finish_reason="stop", index=index) return choice, index + 1 # function_call / custom_tool_call dicts are intercepted and accumulated by # _convert_response_output_to_choices before this callback is reached # Unknown or unsupported type return None, index def convert_chat_completion_messages_to_responses_api( self, messages: list["AllMessageValues"] ) -> tuple[list[object], str | None]: input_items: Final[list[object]] = [] instructions: str | None = None custom_tool_call_ids: Final = frozenset( tool_call["id"] for msg in messages if msg.get("role") == "assistant" and isinstance(msg.get("tool_calls"), list) for tool_call in msg.get("tool_calls") or () if isinstance(tool_call, dict) and not tool_call.get("function") and isinstance(tool_call.get("custom"), dict) ) for msg in messages: role = msg.get("role") content = msg.get("content", "") tool_calls = msg.get("tool_calls") tool_call_id = msg.get("tool_call_id") if role == "system": # Extract system message as instructions if isinstance(content, str): if instructions: # Concatenate multiple system prompts with a space instructions = f"{instructions} {content}" else: instructions = content else: input_items.append( { "type": "message", "role": role, "content": self._convert_content_to_responses_format( content, role, ), } ) elif role == "tool": # Convert tool message to function call output format # The Responses API expects 'output' to be a list with input_text/input_image types # Using list format for consistency across text and multimodal content tool_output: list[dict[str, object]] if content is None: tool_output = [] elif isinstance(content, str): # Convert string to list with input_text tool_output = [{"type": "input_text", "text": content}] elif isinstance(content, list): # Transform list content to Responses API format tool_output = self._convert_content_to_responses_format( content, "user", # Use "user" role to get input_* types ) else: # Fallback: convert unexpected types to input_text tool_output = [{"type": "input_text", "text": str(content)}] if tool_call_id in custom_tool_call_ids: input_items.append( ResponseCustomToolCallOutputParam( type="custom_tool_call_output", call_id=tool_call_id, output=content if isinstance(content, str) else tool_output, ) ) else: input_items.append( FunctionCallOutput( type="function_call_output", call_id=tool_call_id, output=tool_output, ) ) elif role == "assistant" and tool_calls and isinstance(tool_calls, list): input_items.extend(_reasoning_input_items(msg)) if content: input_items.append( { # mutable-ok: API message payload "type": "message", "role": "assistant", "content": self._convert_content_to_responses_format(content, "assistant"), } ) for tool_call in tool_calls: function = tool_call.get("function") custom = tool_call.get("custom") if function: input_tool_call: dict[str, object] = { "type": "function_call", "call_id": tool_call["id"], } if "name" in function: input_tool_call["name"] = function["name"] if "arguments" in function: input_tool_call["arguments"] = function["arguments"] input_items.append(input_tool_call) elif isinstance(custom, dict): input_items.append( ResponseCustomToolCallParam( type="custom_tool_call", call_id=tool_call["id"], name=custom.get("name", ""), input=custom.get("input", ""), ) ) else: raise ValueError(f"tool call not supported: {tool_call}") elif content is not None: if role == "assistant": input_items.extend(_reasoning_input_items(msg)) input_items.append( { # mutable-ok: API message payload "type": "message", "role": role, "content": self._convert_content_to_responses_format(content, cast(str, role)), } ) elif role == "assistant": input_items.extend(_reasoning_input_items(msg)) return input_items, instructions def _map_optional_params_to_responses_api_request( self, optional_params: dict, responses_api_request: "ResponsesAPIOptionalRequestParams", ) -> None: """Map optional_params into responses_api_request (mutates in place).""" for key, value in optional_params.items(): if value is None: continue if key in ("max_tokens", "max_completion_tokens"): responses_api_request["max_output_tokens"] = value elif key == "tools" and value is not None: responses_api_request["tools"] = self._convert_tools_to_responses_format( cast(list[dict[str, Any]], value) ) elif key == "response_format": text_format = self._transform_response_format_to_text_format(value) if text_format: responses_api_request["text"] = text_format elif key == "tool_choice": responses_api_request["tool_choice"] = self._normalize_tool_choice_for_responses_api(value) elif key == "stream_options": stream_options = normalize_responses_api_stream_options(value) if stream_options is not None: responses_api_request["stream_options"] = stream_options elif key in ResponsesAPIOptionalRequestParams.__annotations__: responses_api_request[key] = value elif key == "previous_response_id": responses_api_request["previous_response_id"] = value elif key == "reasoning_effort": responses_api_request["reasoning"] = self._map_reasoning_effort(value) elif key == "web_search_options": self._add_web_search_tool(responses_api_request, value) def _build_sanitized_litellm_params(self, litellm_params: dict) -> dict[str, object]: """Build sanitized litellm_params with merged metadata.""" responses_optional_param_keys: Final = set(ResponsesAPIOptionalRequestParams.__annotations__.keys()) sanitized: Final[dict[str, object]] = { key: value for key, value in litellm_params.items() if key not in responses_optional_param_keys } legacy_metadata: Final = litellm_params.get("metadata") existing_litellm_metadata: Final = litellm_params.get("litellm_metadata") merged_litellm_metadata: Final[dict[str, object]] = {} if isinstance(legacy_metadata, dict): merged_litellm_metadata.update(legacy_metadata) if isinstance(existing_litellm_metadata, dict): merged_litellm_metadata.update(existing_litellm_metadata) if merged_litellm_metadata: sanitized["litellm_metadata"] = merged_litellm_metadata else: sanitized.pop("litellm_metadata", None) return sanitized def _merge_responses_api_request_into_request_data( self, request_data: dict[str, Any], responses_api_request: "ResponsesAPIOptionalRequestParams", instructions: str | None, ) -> None: """Add non-None values from responses_api_request into request_data.""" for key, value in responses_api_request.items(): if value is None: continue if key == "instructions" and instructions: request_data["instructions"] = instructions elif key == "user" and isinstance(value, str): # OpenAI API requires user param to be max 64 chars - truncate if longer if len(value) <= 64: request_data["user"] = value else: request_data["user"] = value[:64] else: request_data[key] = value def transform_request( self, model: str, messages: list["AllMessageValues"], optional_params: dict, litellm_params: dict, headers: dict, litellm_logging_obj: "LiteLLMLoggingObj", client: object | None = None, ) -> dict: ( input_items, instructions, ) = self.convert_chat_completion_messages_to_responses_api(messages) # OpenAI's Responses API rejects an empty input. For a system-only # request, carry the system message as a system-role input item instead # of instructions, mirroring how non-string system content is already # handled in convert_chat_completion_messages_to_responses_api. if not input_items and instructions is not None: input_items = [ { "type": "message", "role": "system", "content": [{"type": "input_text", "text": instructions}], } ] instructions = None optional_params = self._extract_extra_body_params(optional_params) # Build responses API request using the reverse transformation logic responses_api_request: Final = ResponsesAPIOptionalRequestParams() # Set instructions if we found a system message if instructions: responses_api_request["instructions"] = instructions self._map_optional_params_to_responses_api_request(optional_params, responses_api_request) stream: Final = optional_params.get("stream") or litellm_params.get("stream", False) verbose_logger.debug("Chat provider: Stream parameter: %s", stream) # Ensure stream is properly set in the request if stream: responses_api_request["stream"] = True # Handle session management if previous_response_id is provided previous_response_id: Final = optional_params.get("previous_response_id") if previous_response_id: # Use the existing session handler for responses API verbose_logger.debug("Chat provider: Warning ignoring previous response ID: %s", previous_response_id) # Convert back to responses API format for the actual request api_model: Final = model from litellm.types.utils import CallTypes setattr(litellm_logging_obj, "call_type", CallTypes.responses.value) sanitized_litellm_params: Final = self._build_sanitized_litellm_params(litellm_params) request_data: Final = { "model": api_model, "input": input_items, "litellm_logging_obj": litellm_logging_obj, **sanitized_litellm_params, "client": client, } verbose_logger.debug("Chat provider: Final request model=%s, input_items=%s", api_model, len(input_items)) self._merge_responses_api_request_into_request_data(request_data, responses_api_request, instructions) if headers: request_data["extra_headers"] = headers return request_data @staticmethod def _convert_response_output_to_choices( output_items: Sequence[object], handle_raw_dict_callback: Callable[..., tuple["Choices | None", int]] | None = None, ) -> list["Choices"]: """ Convert Responses API output items to chat completion choices. Args: output_items: List of items from ResponsesAPIResponse.output handle_raw_dict_callback: Optional callback for handling raw dict items Returns: List of Choices objects """ from openai.types.responses import ( ResponseFunctionToolCall, ResponseOutputMessage, ResponseReasoningItem, ) try: from openai.types.responses.response_output_item import ( ResponseApplyPatchToolCall, ) except ImportError: ResponseApplyPatchToolCall = None from litellm.types.utils import Choices, Message choices: Final[list[Choices]] = [] index = 0 reasoning_content: str | None = None pending_reasoning_item: _BuiltReasoningItem | None = None # Collect all tool calls to put them in a single choice # (Chat Completions API expects all tool calls in one message) accumulated_tool_calls: Final[list[Mapping[str, object]]] = [] tool_call_index = 0 for item in output_items: if isinstance(item, ResponseReasoningItem): pending_reasoning_item = _build_reasoning_item( item_id=item.id, encrypted_content=getattr(item, "encrypted_content", None), summary_raw=item.summary, ) reasoning_content = " ".join(s["text"] for s in pending_reasoning_item["summary"] if s.get("text")) elif isinstance(item, ResponseOutputMessage): for content in item.content: response_text = getattr(content, "text", "") # Extract annotations from content if present raw_annotations = getattr(content, "annotations", None) annotations = LiteLLMResponsesTransformationHandler._convert_annotations_to_chat_format( raw_annotations ) msg = Message( role=item.role, content=response_text if response_text else "", reasoning_content=reasoning_content, annotations=annotations, reasoning_items=cast( list[ChatCompletionReasoningItem] | None, ([pending_reasoning_item] if pending_reasoning_item is not None else None), ), ) choices.append( Choices( message=msg, finish_reason="stop", index=index, ) ) reasoning_content = None # flush pending_reasoning_item = None # flush index += 1 elif isinstance(item, ResponseFunctionToolCall): from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) tool_call_dict = ( LiteLLMCompletionResponsesConfig.convert_response_function_tool_call_to_chat_completion_tool_call( tool_call_item=item, index=tool_call_index, ) ) accumulated_tool_calls.append(tool_call_dict) tool_call_index += 1 elif ResponseApplyPatchToolCall is not None and isinstance(item, ResponseApplyPatchToolCall): from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) tool_call_dict = ( LiteLLMCompletionResponsesConfig.convert_apply_patch_tool_call_to_chat_completion_tool_call( tool_call_item=item, index=tool_call_index, ) ) accumulated_tool_calls.append(tool_call_dict) tool_call_index += 1 elif isinstance(item, (dict, BaseModel)): # Raw dict items (e.g., from GPT-5 Codex) and pydantic items matching no # openai SDK class above: typed ResponseCustomToolCall and litellm's own # GenericResponseOutputItem from the completion bridge both land here raw_item = item if isinstance(item, dict) else item.model_dump() if raw_item.get("type") in ("function_call", "custom_tool_call"): # Tool calls accumulate into the single trailing tool_calls choice # like the typed branches above; a choice per call would hide every # call after choices[0] from chat clients accumulated_tool_calls.append(_tool_call_dict_from_output_item(raw_item, tool_call_index)) tool_call_index += 1 elif handle_raw_dict_callback is not None: choice, index = handle_raw_dict_callback(item=raw_item, index=index) if choice is not None: choices.append(choice) else: pass # don't fail request if item in list is not supported # If we accumulated tool calls, create a single choice with all of them if accumulated_tool_calls: msg = Message( content=None, tool_calls=accumulated_tool_calls, reasoning_content=reasoning_content, reasoning_items=cast( list[ChatCompletionReasoningItem] | None, ([pending_reasoning_item] if pending_reasoning_item is not None else None), ), ) choices.append(Choices(message=msg, finish_reason="tool_calls", index=index)) reasoning_content = None pending_reasoning_item = None return choices @staticmethod def _build_empty_incomplete_choice( output_items: Sequence[object], finish_reason: Literal["length", "content_filter"], ) -> "Choices": from litellm.types.utils import Choices, Message reasoning_items: Final = _reasoning_items_from_output_items(output_items) reasoning_content: Final = " ".join( summary_block["text"] for reasoning_item in reasoning_items for summary_block in reasoning_item["summary"] if summary_block.get("text") ) message: Final = Message( content="", reasoning_content=reasoning_content if reasoning_content else None, reasoning_items=_as_chat_reasoning_items(reasoning_items), ) return Choices(message=message, finish_reason=finish_reason, index=0) @classmethod def _extract_output_from_completed_event(cls, parsed_chunk: Mapping[str, object]) -> list[dict[str, object]] | None: response_payload: Final = parsed_chunk.get("response") if not isinstance(response_payload, dict): return None response_output: Final = response_payload.get("output") if not isinstance(response_output, list) or len(response_output) == 0: return None return cast(list[dict[str, Any]], response_output) @classmethod def _recover_output_items_from_raw_sse(cls, raw_sse: str | None) -> list[dict[str, object]]: if not raw_sse or not isinstance(raw_sse, str): return [] recovered_output_items: Final[dict[int, dict[str, object]]] = {} recovered_text_only_items: Final[dict[int, dict[str, object]]] = {} for chunk in raw_sse.splitlines(): parsed_chunk = parse_sse_json_chunk(chunk) if parsed_chunk is None: continue event_type = parsed_chunk.get("type") if event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED: recovered_output = cls._extract_output_from_completed_event(parsed_chunk) if recovered_output is not None: return recovered_output continue if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: record_output_item_chunk( parsed_chunk=parsed_chunk, output_items=recovered_output_items, ) continue if event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: record_output_text_chunk( parsed_chunk=parsed_chunk, output_items=recovered_output_items, text_only_items=recovered_text_only_items, ) continue # Merge text-only items into the recovered output items. Real # OUTPUT_ITEM_DONE events take precedence at any given output_index, # but text-only items at indices without a matching OUTPUT_ITEM_DONE # must still be preserved (e.g. multi-output responses where some # indices only emitted OUTPUT_TEXT_DONE). merged_items: Final[dict[int, dict[str, object]]] = {**recovered_text_only_items} merged_items.update(recovered_output_items) if merged_items: return [item for _, item in sorted(merged_items.items())] return [] @classmethod def _recover_output_items_from_logging(cls, logging_obj: "LiteLLMLoggingObj") -> list[dict[str, object]]: model_call_details: Final = getattr(logging_obj, "model_call_details", {}) or {} original_response: Final = model_call_details.get("original_response") return cls._recover_output_items_from_raw_sse(original_response) def transform_response( self, model: str, raw_response: "BaseModel", model_response: "ModelResponse", logging_obj: "LiteLLMLoggingObj", request_data: dict, messages: list["AllMessageValues"], optional_params: dict, litellm_params: dict, encoding: object, api_key: str | None = None, json_mode: bool | None = None, ) -> "ModelResponse": """Transform Responses API response to chat completion response""" from litellm.responses.utils import ResponseAPILoggingUtils from litellm.types.llms.openai import ResponsesAPIResponse if not isinstance(raw_response, ResponsesAPIResponse): raise ValueError(f"Unexpected response type: {type(raw_response)}") if raw_response.error is not None: raise ValueError(f"Error in response: {raw_response.error}") output_items = raw_response.output if len(output_items) == 0: recovered_output_items: Final = self._recover_output_items_from_logging(logging_obj) if recovered_output_items: output_items = cast(Any, recovered_output_items) raw_response.output = cast(Any, recovered_output_items) verbose_logger.warning( "Recovered empty Responses API output from raw SSE for model=%s", model, ) # Convert response output to choices using the static helper choices: Final = self._convert_response_output_to_choices( output_items=output_items, handle_raw_dict_callback=self._handle_raw_dict_response_item, ) response_is_incomplete: Final = raw_response.status == "incomplete" or ( raw_response.incomplete_details is not None and raw_response.incomplete_details.reason is not None ) if len(choices) == 0 and not response_is_incomplete: raise ValueError(f"Unknown items in responses API response: {output_items}") if response_is_incomplete: incomplete_finish_reason: Final = _map_incomplete_reason_to_finish_reason( raw_response.incomplete_details.reason if raw_response.incomplete_details is not None else None ) if len(choices) == 0: choices.append(self._build_empty_incomplete_choice(output_items, incomplete_finish_reason)) else: for choice in choices: choice.finish_reason = incomplete_finish_reason setattr(model_response, "choices", choices) model_response.model = model setattr( model_response, "usage", ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(raw_response.usage), ) # Preserve hidden params from the ResponsesAPIResponse, especially the headers # which contain important provider information like x-request-id raw_response_hidden_params: Final = getattr(raw_response, "_hidden_params", {}) if raw_response_hidden_params: if not hasattr(model_response, "_hidden_params") or model_response._hidden_params is None: model_response._hidden_params = {} # Merge the raw_response hidden params with model_response hidden params # Preserve existing keys in model_response but add/override with raw_response params for key, value in raw_response_hidden_params.items(): if key == "additional_headers" and key in model_response._hidden_params: # Merge additional_headers to preserve both sets existing_additional_headers = model_response._hidden_params.get("additional_headers", {}) merged_headers = {**value, **existing_additional_headers} model_response._hidden_params[key] = merged_headers else: model_response._hidden_params[key] = value return model_response def get_model_response_iterator( self, streaming_response: Union[Iterator[str], AsyncIterator[str], "ModelResponse", "BaseModel"], sync_stream: bool, json_mode: bool | None = False, ) -> BaseModelResponseIterator: return OpenAiResponsesToChatCompletionStreamIterator(streaming_response, sync_stream, json_mode) def _convert_content_str_to_input_text(self, content: str, role: str) -> dict[str, object]: if role == "user" or role == "system" or role == "tool": return {"type": "input_text", "text": content} else: return {"type": "output_text", "text": content} def _convert_content_to_responses_format_image( self, content: "ChatCompletionImageObject", role: str ) -> "ResponseInputImageParam": from openai.types.responses import ResponseInputImageParam content_image_url: Final = content.get("image_url") actual_image_url: str | None = None detail: Literal["low", "high", "auto"] | None = None if isinstance(content_image_url, str): actual_image_url = content_image_url elif isinstance(content_image_url, dict): actual_image_url = content_image_url.get("url") detail = cast( Literal["low", "high", "auto"] | None, content_image_url.get("detail"), ) if actual_image_url is None: raise ValueError(f"Invalid image URL: {content_image_url}") image_param: Final = ResponseInputImageParam(image_url=actual_image_url, detail="auto", type="input_image") if detail: image_param["detail"] = detail return image_param def _convert_content_to_responses_format( self, content: str | list[object] | Iterable[ Union[ "OpenAIMessageContentListBlock", "ChatCompletionThinkingBlock", "ChatCompletionRedactedThinkingBlock", "ChatCompletionToolReferenceObject", ] ] | None, role: str, ) -> list[dict[str, object]]: """Convert chat completion content to responses API format""" from litellm.types.llms.openai import ChatCompletionImageObject verbose_logger.debug("Chat provider: Converting content to responses format - input type: %s", type(content)) if content is None: return [self._convert_content_str_to_input_text("", role)] elif isinstance(content, str): result = [self._convert_content_str_to_input_text(content, role)] verbose_logger.debug("Chat provider: String content -> %s", result) return result elif isinstance(content, list): result = [] for i, item in enumerate(content): verbose_logger.debug("Chat provider: Processing content item %s: %s = %s", i, type(item), item) if isinstance(item, str): converted = self._convert_content_str_to_input_text(item, role) result.append(converted) verbose_logger.debug("Chat provider: -> %s", converted) elif isinstance(item, dict): # Handle multimodal content original_type = item.get("type") if original_type == "text": converted = self._convert_content_str_to_input_text(item.get("text", ""), role) result.append(converted) verbose_logger.debug("Chat provider: text -> %s", converted) elif original_type == "image_url": # Map to responses API image format converted = cast( dict, self._convert_content_to_responses_format_image( cast(ChatCompletionImageObject, item), role ), ) result.append(converted) verbose_logger.debug("Chat provider: image_url -> %s", converted) else: # Try to map other types to responses API format item_type = original_type or "input_text" if item_type == "image": converted = {"type": "input_image", **item} result.append(converted) verbose_logger.debug("Chat provider: image -> %s", converted) elif item_type == "file": converted = _input_file_from_file_value( cast("ChatCompletionFileObject", item).get("file"), # cast-ok: type tag checked ) result.append(converted) verbose_logger.debug("Chat provider: file -> %s", converted) elif item_type == "tool_reference": verbose_logger.debug( "Chat provider: tool_reference has no responses API equivalent; skipped" ) elif item_type in [ "input_text", "input_image", "output_text", "refusal", "input_file", "computer_screenshot", "summary_text", ]: # Already in responses API format result.append(item) verbose_logger.debug("Chat provider: passthrough -> %s", item) else: # Default to input_text for unknown types converted = self._convert_content_str_to_input_text(str(item.get("text", item)), role) result.append(converted) verbose_logger.debug("Chat provider: unknown(%s) -> %s", original_type, converted) verbose_logger.debug("Chat provider: Final converted content: %s", result) return result else: result = [self._convert_content_str_to_input_text(str(content), role)] verbose_logger.debug("Chat provider: Other content type -> %s", result) return result def _convert_tools_to_responses_format(self, tools: list[dict[str, Any]]) -> list["ALL_RESPONSES_API_TOOL_PARAMS"]: """Convert chat completion tools to responses API tools format""" responses_tools: Final[list[ALL_RESPONSES_API_TOOL_PARAMS]] = [] for tool in tools: # convert function tool from chat completion to responses API format if tool.get("type") == "function": function_tool = cast(ChatCompletionToolParamFunctionChunk, tool.get("function")) responses_tools.append( FunctionToolParam( name=function_tool["name"], parameters=function_tool.get("parameters"), strict=function_tool.get("strict"), type="function", description=function_tool.get("description"), ) ) elif tool.get("type") == "custom" and isinstance(tool.get("custom"), dict): from litellm.litellm_core_utils.prompt_templates.common_utils import ( convert_custom_tool_format_to_responses_shape, ) custom_payload = tool["custom"] flat_custom = CustomToolParam(type="custom", name=custom_payload.get("name", "")) if custom_payload.get("description") is not None: flat_custom["description"] = custom_payload["description"] if isinstance(custom_payload.get("format"), dict): flat_custom["format"] = convert_custom_tool_format_to_responses_shape(custom_payload["format"]) responses_tools.append(flat_custom) else: responses_tools.append(tool) return cast(list["ALL_RESPONSES_API_TOOL_PARAMS"], responses_tools) def _extract_extra_body_params(self, optional_params: dict): """ Extract extra_body from optional_params and separate supported Responses API params from unsupported ones. Supported params are moved to top-level optional_params, unsupported params remain in extra_body. """ # Extract extra_body and separate supported params from unsupported ones extra_body: Final = optional_params.pop("extra_body", None) or {} if not extra_body: return optional_params supported_responses_api_params: Final = set(ResponsesAPIOptionalRequestParams.__annotations__.keys()) # Also include params we handle specially supported_responses_api_params.update( { "previous_response_id", "reasoning_effort", # We map this to "reasoning" } ) # Extract supported params from extra_body and merge into optional_params extra_body_copy: Final = extra_body.copy() for key, value in extra_body_copy.items(): if key in supported_responses_api_params: # Prefer extra_body value if it exists (may have more complete info like summary in reasoning_effort) optional_params[key] = extra_body.pop(key) return optional_params def _map_reasoning_effort(self, reasoning_effort: str | Reasoning) -> Reasoning | None: # If dict is passed, convert it directly to Reasoning object if isinstance(reasoning_effort, dict): return Reasoning(**reasoning_effort) # Check if auto-summary is enabled via flag or environment variable # Priority: litellm.reasoning_auto_summary flag > LITELLM_REASONING_AUTO_SUMMARY env var auto_summary_enabled: Final = ( litellm.reasoning_auto_summary or os.getenv("LITELLM_REASONING_AUTO_SUMMARY", "false").lower() == "true" ) if reasoning_effort in get_args(REASONING_EFFORT): return ( Reasoning(effort=reasoning_effort, summary="detailed") if auto_summary_enabled else Reasoning(effort=reasoning_effort) ) return None def _add_web_search_tool( self, responses_api_request: ResponsesAPIOptionalRequestParams, web_search_options: object, ) -> None: """ Add web search tool to responses API request. Args: responses_api_request: The responses API request dict to modify web_search_options: Web search configuration (dict or other value) """ if "tools" not in responses_api_request or responses_api_request["tools"] is None: responses_api_request["tools"] = [] # Get the tools list with proper type narrowing tools = responses_api_request["tools"] if tools is None: tools = [] responses_api_request["tools"] = tools web_search_tool: Final[dict[str, object]] = {"type": "web_search"} if isinstance(web_search_options, dict): web_search_tool.update(web_search_options) # Cast to Any to match the expected union type for tools list items tools.append(cast(Any, web_search_tool)) def _transform_response_format_to_text_format(self, response_format: object) -> "ResponseText | None": """ Transform Chat Completion response_format parameter to Responses API text.format parameter. Chat Completion response_format structure: { "type": "json_schema", "json_schema": { "name": "schema_name", "schema": {...}, "strict": True } } Responses API text parameter structure: { "format": { "type": "json_schema", "name": "schema_name", "schema": {...}, "strict": True } } """ if not response_format: return None if isinstance(response_format, dict): format_type: Final = response_format.get("type") if format_type == "json_schema": json_schema: Final = response_format.get("json_schema", {}) return { "format": { "type": "json_schema", "name": json_schema.get("name", "response_schema"), "schema": json_schema.get("schema", {}), "strict": json_schema.get("strict", False), } } elif format_type == "json_object": return {"format": {"type": "json_object"}} elif format_type == "text": return {"format": {"type": "text"}} return None @staticmethod def _convert_annotations_to_chat_format( annotations: list[Any] | None, ) -> list[ChatCompletionAnnotation] | None: """ Convert annotations from Responses API to Chat Completions format. Annotations are already in compatible format between both APIs, so we just need to convert Pydantic models to dicts. """ if not annotations: return None result: Final[list[ChatCompletionAnnotation]] = [] for annotation in annotations: try: # Convert Pydantic models to dicts (handles both v1 and v2) if hasattr(annotation, "model_dump"): annotation_dict = annotation.model_dump() elif hasattr(annotation, "dict"): annotation_dict = annotation.dict() elif isinstance(annotation, dict): annotation_dict = annotation else: # Skip unsupported annotation types verbose_logger.debug("Skipping unsupported annotation type: %s", type(annotation)) continue result.append(annotation_dict) except Exception as e: # Skip malformed annotations verbose_logger.debug("Skipping malformed annotation: %s, error: %s", annotation, e) continue return result if result else None def _map_responses_status_to_finish_reason(self, status: str | None) -> str: """Map responses API status to chat completion finish_reason""" if not status: return "stop" status_mapping: Final = { "completed": "stop", "incomplete": "length", "failed": "stop", "cancelled": "stop", } return status_mapping.get(status, "stop") class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator): def __init__( self, streaming_response: Union[Iterator[str], AsyncIterator[str], "ModelResponse", "BaseModel"], sync_stream: bool, json_mode: bool | None = False, ): super().__init__(streaming_response, sync_stream, json_mode) self._chat_completion_id: str | None = None self._tool_call_index_map: dict[int, int] = {} # mutable-ok: per-stream accumulator state def _handle_string_chunk( self, str_line: Union[str, "BaseModel"] ) -> Union["GenericStreamingChunk", "ModelResponseStream"]: from pydantic import BaseModel if isinstance(str_line, BaseModel): return self.chunk_parser(str_line.model_dump()) if not str_line or str_line.startswith("event:"): # ignore. return GenericStreamingChunk(text="", tool_use=None, is_finished=False, finish_reason="", usage=None) index: Final = str_line.find("data:") if index != -1: str_line = str_line[index + 5 :] return self.chunk_parser(json.loads(str_line)) @staticmethod def _sequential_tool_call_index( tool_call_index_map: dict[int, int] | None, # mutable-ok: per-stream state, remapped in place output_index: int, ) -> int: """Chat-completions tool_call indices must be 0-based and sequential, but Responses API ``output_index`` counts every output item (reasoning, message, ...), so the first tool call of a reasoning model arrives at output_index >= 1 and strict SSE accumulators (e.g. Cursor agent mode) misplace it. When a per-stream map is provided, remap each distinct output_index to the next sequential slot; without a map (stateless callers), fall back to the raw output_index.""" if tool_call_index_map is None: return output_index if output_index not in tool_call_index_map: tool_call_index_map[output_index] = len(tool_call_index_map) # mutable-ok: per-stream accumulator state return tool_call_index_map[output_index] @staticmethod def translate_responses_chunk_to_openai_stream( parsed_chunk: dict | BaseModel, tool_call_index_map: dict[int, int] | None = None, # mutable-ok: per-stream state, remapped in place ) -> "ModelResponseStream": """ Translate a Responses API streaming chunk to OpenAI chat completion streaming format. Args: parsed_chunk: Dict containing the Responses API event chunk tool_call_index_map: Per-stream output_index -> sequential tool_call index map Returns: ModelResponseStream: OpenAI-formatted streaming chunk Raises: ValueError: If chunk is invalid or missing required fields """ from litellm.types.llms.openai import ChatCompletionToolCallFunctionChunk from litellm.types.utils import ( ChatCompletionToolCallChunk, Delta, ModelResponseStream, StreamingChoices, ) if not parsed_chunk: raise ValueError("Chat provider: Empty parsed_chunk") if isinstance(parsed_chunk, BaseModel): parsed_chunk = parsed_chunk.model_dump() if not isinstance(parsed_chunk, dict): raise ValueError(f"Chat provider: Invalid chunk type {type(parsed_chunk)}") # Handle different event types from responses API event_type = parsed_chunk.get("type") if isinstance(event_type, ResponsesAPIStreamEvents): event_type = event_type.value if parsed_chunk.get("object") == "chat.completion.chunk" or ( event_type is None and isinstance(parsed_chunk.get("choices"), list) and parsed_chunk.get("choices") ): return ModelResponseStream(**parsed_chunk) verbose_logger.debug("Chat provider: Processing event type: %s", event_type) if event_type == "response.created": # Initial response creation event verbose_logger.debug("Chat provider: response.created -> %s", parsed_chunk) return ModelResponseStream( choices=[ StreamingChoices( index=0, delta=Delta(content=""), finish_reason=None, ) ] ) elif event_type == "response.output_item.added": # New output item added output_item = parsed_chunk.get("item", {}) if output_item.get("type") in ("function_call", "custom_tool_call"): converted: Final = _tool_call_dict_from_output_item(output_item, parsed_chunk.get("output_index", 0)) provider_specific_fields: Final = converted.get("provider_specific_fields") function_chunk: Final = ChatCompletionToolCallFunctionChunk( name=converted["function"]["name"] or None, arguments=converted["function"]["arguments"] or parsed_chunk.get("arguments") or "", ) if provider_specific_fields: function_chunk["provider_specific_fields"] = provider_specific_fields tool_call_index = OpenAiResponsesToChatCompletionStreamIterator._sequential_tool_call_index( tool_call_index_map, parsed_chunk.get("output_index", 0) ) tool_call_chunk: Final = ChatCompletionToolCallChunk( id=converted["id"], index=tool_call_index, type="function", function=function_chunk, ) if provider_specific_fields: tool_call_chunk.provider_specific_fields = provider_specific_fields return ModelResponseStream( choices=[ StreamingChoices( index=0, delta=Delta(tool_calls=[tool_call_chunk]), finish_reason=None, ) ] ) elif event_type in ( ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DELTA, ResponsesAPIStreamEvents.CUSTOM_TOOL_CALL_INPUT_DELTA, ): content_part: str | None = parsed_chunk.get("delta", None) if content_part: tool_call_index = OpenAiResponsesToChatCompletionStreamIterator._sequential_tool_call_index( tool_call_index_map, parsed_chunk.get("output_index", 0) ) return ModelResponseStream( choices=[ StreamingChoices( index=0, delta=Delta( tool_calls=[ ChatCompletionToolCallChunk( id=None, index=tool_call_index, type="function", function=ChatCompletionToolCallFunctionChunk(name=None, arguments=content_part), ) ] ), finish_reason=None, ) ] ) else: raise ValueError(f"Chat provider: Invalid function argument delta {parsed_chunk}") elif event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: # New output item added output_item = parsed_chunk.get("item", {}) if output_item.get("type") in ("function_call", "custom_tool_call"): if tool_call_index_map is None: # Stateless callers (the responses guardrail handler extracting # tool calls from a buffered output_item.done) get the complete # tool call; per-stream callers already received it via # output_item.added and the argument delta events return ModelResponseStream( choices=[ # mutable-ok: ModelResponseStream coerces only list choices StreamingChoices( index=0, delta=Delta( tool_calls=( _tool_call_dict_from_output_item( output_item, parsed_chunk.get("output_index", 0) ), ) ), finish_reason=None, ) ] ) # Do NOT emit finish_reason here — response.completed handles the terminal # finish_reason. Emitting "tool_calls" here would prematurely terminate # the stream before subsequent tool calls arrive (same fix as #17246 for # the message-type branch). The item's fields were already streamed via # output_item.added and the argument delta events. return ModelResponseStream( choices=[ StreamingChoices( index=0, delta=Delta(), finish_reason=None, ) ] ) elif output_item.get("type") == "message": # Message completion should NOT emit finish_reason # This is the fix for issue #17246 - don't end stream prematurely return ModelResponseStream( choices=[ StreamingChoices( index=0, delta=Delta(content=""), finish_reason=None, ) ] ) elif event_type == "response.output_text.delta": # Content part added to output content_part = parsed_chunk.get("delta", None) if content_part is not None: return ModelResponseStream( choices=[ StreamingChoices( index=0, delta=Delta(content=content_part), finish_reason=None, ) ] ) else: raise ValueError(f"Chat provider: Invalid text delta {parsed_chunk}") elif event_type == "response.reasoning_summary_text.delta": content_part = parsed_chunk.get("delta", None) if content_part: return ModelResponseStream( choices=[ StreamingChoices( index=cast(int, parsed_chunk.get("summary_index")), delta=Delta(reasoning_content=content_part), ) ] ) elif event_type in ("response.completed", "response.incomplete"): response_data: Final = parsed_chunk.get("response", {}) output_items: Final = response_data.get("output", []) if response_data else [] has_function_calls: Final = any( item.get("type") in ("function_call", "custom_tool_call") for item in output_items if isinstance(item, dict) ) finish_reason: Final = ( _map_incomplete_reason_to_finish_reason(_incomplete_reason_from_response_payload(response_data)) if event_type == "response.incomplete" else ("tool_calls" if has_function_calls else "stop") ) terminal_reasoning_items_typed: Final = _as_chat_reasoning_items( _reasoning_items_from_output_items(output_items) ) usage = None if response_data.get("usage"): from litellm.responses.utils import ResponseAPILoggingUtils usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(response_data.get("usage")) return ModelResponseStream( choices=[ StreamingChoices( index=0, delta=Delta( content="", reasoning_items=terminal_reasoning_items_typed, ), finish_reason=finish_reason, ) ], usage=usage, ) else: pass # For any unhandled event types, create a minimal valid chunk or skip verbose_logger.debug("Chat provider: Unhandled event type '%s', creating empty chunk", event_type) # Return a minimal valid chunk for unknown events return ModelResponseStream( choices=[ StreamingChoices( index=0, delta=Delta(content=""), finish_reason=None, ) ] ) def chunk_parser(self, chunk: dict[str, object]) -> "ModelResponseStream": """ Parse a Responses API streaming chunk and convert to OpenAI format. Args: chunk: Dict containing the Responses API event chunk Returns: ModelResponseStream: OpenAI-formatted streaming chunk """ verbose_logger.debug("Chat provider: transform_streaming_response called with chunk: %s", chunk) return self._with_stream_scoped_id( OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream( chunk, tool_call_index_map=self._tool_call_index_map ) ) def _with_stream_scoped_id(self, chunk: "ModelResponseStream") -> "ModelResponseStream": if self._chat_completion_id is None: self._chat_completion_id = chunk.id else: chunk.id = self._chat_completion_id return chunk