"""
Handles transforming from Responses API -> LiteLLM completion (Chat Completion API)
"""
import json
import re
import uuid
from collections.abc import Iterable, Iterator, Mapping, Sequence
from types import MappingProxyType
from typing import (
TYPE_CHECKING,
Any,
Final,
Literal,
Protocol,
TypeAlias,
cast,
runtime_checkable,
)
from openai.types.chat.chat_completion_named_tool_choice_param import (
ChatCompletionNamedToolChoiceParam,
)
from openai.types.chat.chat_completion_named_tool_choice_param import (
Function as NamedToolChoiceFunction,
)
from openai.types.responses import ResponseFunctionToolCall
from openai.types.responses.response_create_params import ResponseInputParam
from openai.types.responses.tool_param import FunctionToolParam
from pydantic import TypeAdapter
from typing_extensions import ReadOnly, TypedDict
from litellm._logging import verbose_logger
from litellm.caching import InMemoryCache
from litellm.constants import REDACTED_BY_LITELLM, REDACTED_TOOL_CALL_ARGUMENTS_PLACEHOLDER
from litellm.litellm_core_utils.get_supported_openai_params import (
get_supported_openai_params,
)
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.responses.litellm_completion_transformation.session_handler import (
ResponsesSessionHandler,
)
from litellm.types.llms.openai import (
AllMessageValues,
ChatCompletionImageObject,
ChatCompletionImageUrlObject,
ChatCompletionRedactedThinkingBlock,
ChatCompletionResponseMessage,
ChatCompletionSystemMessage,
ChatCompletionTextObject,
ChatCompletionThinkingBlock,
ChatCompletionToolCallChunk,
ChatCompletionToolCallFunctionChunk,
ChatCompletionToolMessage,
ChatCompletionToolParam,
ChatCompletionToolParamFunctionChunk,
ChatCompletionUserMessage,
GenericChatCompletionMessage,
InputTokensDetails,
OpenAIChatCompletionTextObject,
OpenAIMcpServerTool,
OpenAIWebSearchOptions,
OpenAIWebSearchUserLocation,
OutputTokensDetails,
Reasoning,
ResponseAPIUsage,
ResponsesAPIOptionalRequestParams,
ResponsesAPIResponse,
ResponsesAPIStatus,
ValidChatCompletionMessageContentTypes,
ValidChatCompletionMessageContentTypesLiteral,
)
from litellm.types.responses.main import (
CustomToolCallOutputItem,
GenericResponseOutputItem,
GenericResponseOutputItemContentAnnotation,
OutputCodeInterpreterCall,
OutputFunctionToolCall,
OutputImageGenerationCall,
OutputText,
)
from litellm.types.utils import (
ChatCompletionAnnotation,
ChatCompletionMessageToolCall,
Choices,
Function,
Message,
ModelResponse,
Usage,
)
from .custom_tools import (
convert_custom_tool_to_function_tool,
extract_custom_tool_names,
is_custom_tool_call,
unwrap_custom_tool_arguments,
validated_allowed_callers,
)
NamespaceNameMap: TypeAlias = Mapping[str, tuple[str, str]]
NamespaceTool: TypeAlias = Mapping[str, object]
ResponseTools: TypeAlias = Sequence[Mapping[str, object]] | None
if TYPE_CHECKING:
from openai.types.responses.response_apply_patch_tool_call import (
ResponseApplyPatchToolCall,
)
########### Initialize Classes used for Responses API ###########
TOOL_CALLS_CACHE: Final = InMemoryCache()
_ANY_KEY_DICT_ADAPTER: Final = TypeAdapter(dict[object, object])
_STR_KEY_DICT_ADAPTER: Final = TypeAdapter(dict[str, object])
_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
_DICT_ITEMS_LIST_ADAPTER: Final = TypeAdapter(list[dict[object, object]])
_TEXT_ADAPTER: Final = TypeAdapter(str)
@runtime_checkable
class _SupportsIter(Protocol):
def __iter__(self) -> Iterator[object]: ...
@runtime_checkable
class _HasToolCalls(Protocol):
tool_calls: object
@runtime_checkable
class _HasId(Protocol):
id: object
class _ResponsesToolCallItem(Protocol):
name: object
arguments: object
def get(self, key: str, /) -> object: ...
class _ToolFunctionDefinition(TypedDict, total=False):
name: ReadOnly[str]
description: ReadOnly[str]
parameters: ReadOnly[dict[str, object]]
strict: ReadOnly[bool | None]
def _attribute_fields(value: object) -> dict[str, object]:
if not hasattr(value, "__dict__"):
return {} # mutable-ok: provider_specific_fields payload
return dict(cast("Iterable[tuple[str, object]]", value)) # cast-ok: dict() raises on non-pair values, as before
def _input_item_role(input_item: Mapping[str, object]) -> str:
return cast(str, input_item.get("role") or "user") # cast-ok: client-supplied role forwarded verbatim, unvalidated
class ChatCompletionSession(TypedDict, total=False):
messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
| Message
]
litellm_session_id: str | None
########### End of Initialize Classes used for Responses API ###########
class LiteLLMCompletionResponsesConfig:
@staticmethod
def get_supported_openai_params(model: str) -> list:
"""
LiteLLM Adapter from OpenAI Responses API to Chat Completion API supports a subset of OpenAI Responses API params
"""
return [
"input",
"model",
"instructions",
"max_output_tokens",
"metadata",
"parallel_tool_calls",
"previous_response_id",
"reasoning",
"stream",
"temperature",
"text",
"tool_choice",
"tools",
"top_p",
"user",
]
@staticmethod
def _transform_tool_choice(
tool_choice: Any,
) -> str | dict[str, Any] | None:
"""
Transform tool_choice from various formats to OpenAI Chat Completion format.
Handles:
- String values: "auto", "none", "required" -> pass through as-is
- Dict with type only (Cursor IDE format):
- {"type": "auto"} -> "auto"
- {"type": "none"} -> "none"
- {"type": "required"} -> "required"
- {"type": "tool"} -> "required" (force tool use without specific tool)
- Dict with function (OpenAI format):
- {"type": "function", "function": {"name": "..."}} -> pass through as-is
This normalization is needed because some clients (like Cursor IDE) send
tool_choice in a dict format like {"type": "tool"} which is not valid for
providers like Anthropic that require a tool name when forcing tool use.
"""
if tool_choice is None:
return None
if isinstance(tool_choice, str):
return tool_choice
if isinstance(tool_choice, dict):
tool_choice_type: Final = tool_choice.get("type")
# If it has a function with name, it's standard OpenAI format - pass through
if tool_choice.get("function") and tool_choice.get("function", {}).get("name"):
return tool_choice
# Handle Cursor IDE dict formats without function name
if tool_choice_type == "auto":
return "auto"
elif tool_choice_type == "none":
return "none"
elif tool_choice_type in ["required", "tool", "any"]:
# "tool" without a specific function name means "use any tool"
# which is equivalent to "required" in OpenAI format
return "required"
elif tool_choice_type == "function":
function_name: Final = tool_choice.get("name")
if function_name:
return ChatCompletionNamedToolChoiceParam(
type="function", function=NamedToolChoiceFunction(name=function_name)
)
return "required"
elif tool_choice_type == "custom":
custom: Final = tool_choice.get("custom")
custom_name = tool_choice.get("name") or (custom.get("name") if isinstance(custom, dict) else None)
if custom_name:
return ChatCompletionNamedToolChoiceParam(
type="function", function=NamedToolChoiceFunction(name=custom_name)
)
return "required"
# Return as-is for unknown formats
return tool_choice
@staticmethod
def _should_drop_derived_web_search_options(model: str, custom_llm_provider: str | None) -> bool:
"""
A Responses ``web_search`` built-in tool is derived into a ``web_search_options`` param.
When the resolved provider/model does not support it (e.g. Bedrock Anthropic, where only
Nova maps it to a nova_grounding systemTool), the derived param is dropped here instead of
raising UnsupportedParamsError downstream. Providers that support it keep it untouched.
Support is read from each provider's own ``get_supported_openai_params`` so this bridge
stays provider-agnostic; an unmapped provider (``None``) is treated as "keep".
"""
supported_params: Final[list[str] | None] = get_supported_openai_params(
model=model, custom_llm_provider=custom_llm_provider
)
return supported_params is not None and "web_search_options" not in supported_params
@staticmethod
def transform_responses_api_request_to_chat_completion_request(
model: str,
input: str | ResponseInputParam,
responses_api_request: ResponsesAPIOptionalRequestParams,
custom_llm_provider: str | None = None,
stream: bool | None = None,
extra_headers: Mapping[str, object] | None = None,
**kwargs,
) -> dict:
"""
Transform a Responses API request into a Chat Completion request
"""
(
tools,
web_search_options,
) = LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools(
responses_api_request.get("tools") or []
)
if web_search_options is not None and LiteLLMCompletionResponsesConfig._should_drop_derived_web_search_options(
model=model, custom_llm_provider=custom_llm_provider
):
web_search_options = None
response_format = None
text_param: Final = responses_api_request.get("text")
if text_param:
response_format = LiteLLMCompletionResponsesConfig._transform_text_format_to_response_format(text_param)
# Extract reasoning_effort from reasoning parameter
reasoning_effort: Reasoning | str | None = None
reasoning_param: Final = responses_api_request.get("reasoning")
if reasoning_param:
if isinstance(reasoning_param, dict):
# reasoning can be {"effort": "low|medium|high", "summary": "detailed"}
# Keep the full dict when summary is set so the responses API bridge can
# forward it; otherwise use the effort string for chat completion (e.g. Gemini).
if "summary" in reasoning_param:
reasoning_effort = reasoning_param
elif "effort" in reasoning_param:
reasoning_effort = reasoning_param.get("effort")
else:
reasoning_effort = reasoning_param
elif isinstance(reasoning_param, str):
# reasoning could be a string directly
reasoning_effort = reasoning_param
litellm_completion_request: dict = {
"messages": LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=input,
responses_api_request=responses_api_request,
replay_reasoning=True,
),
"model": model,
"tool_choice": LiteLLMCompletionResponsesConfig._transform_tool_choice(
responses_api_request.get("tool_choice")
),
"tools": tools,
"top_p": responses_api_request.get("top_p"),
"user": responses_api_request.get("user"),
"temperature": responses_api_request.get("temperature"),
"parallel_tool_calls": responses_api_request.get("parallel_tool_calls"),
"max_tokens": responses_api_request.get("max_output_tokens"),
"stream": stream,
"metadata": kwargs.get("metadata"),
"service_tier": kwargs.get("service_tier"),
"web_search_options": web_search_options,
"response_format": response_format,
"reasoning_effort": reasoning_effort,
"context_management": responses_api_request.get("context_management"),
# litellm specific params
"custom_llm_provider": custom_llm_provider,
"extra_headers": extra_headers,
}
if not tools:
litellm_completion_request.pop("tool_choice", None)
litellm_completion_request.pop("tools", None)
# Responses API `Completed` events require usage, we pass `stream_options` to litellm.completion to include usage
if stream is True:
stream_options: Final = {
"include_usage": True,
}
litellm_completion_request["stream_options"] = stream_options
litellm_logging_obj: Final[LiteLLMLoggingObj | None] = kwargs.get("litellm_logging_obj")
if litellm_logging_obj:
litellm_logging_obj.stream_options = stream_options
# only pass non-None values
litellm_completion_request = {k: v for k, v in litellm_completion_request.items() if v is not None}
return litellm_completion_request
@staticmethod
def transform_responses_api_input_to_messages(
input: str | ResponseInputParam,
responses_api_request: ResponsesAPIOptionalRequestParams | dict,
replay_reasoning: bool = False,
) -> list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
| Message
]:
"""
Transform a Responses API input into a list of messages
``replay_reasoning`` belongs to callers whose messages are about to be
sent to a model: prior-turn ``reasoning`` items are then rebuilt as
assistant ``reasoning_content`` and signed ``thinking_blocks`` so the
provider gets its own chain-of-thought back instead of reading it as
visible text.
Callers that only inspect the messages (token counting, rate limiting,
guardrail scanning) leave it off, because they need every piece of text
in the request to stay readable as message ``content``.
"""
messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
| Message
] = []
if responses_api_request.get("instructions"):
messages.append(
LiteLLMCompletionResponsesConfig.transform_instructions_to_system_message(
responses_api_request.get("instructions")
)
)
messages.extend(
LiteLLMCompletionResponsesConfig._transform_response_input_param_to_chat_completion_message(
input=input,
replay_reasoning=replay_reasoning,
)
)
return messages
@staticmethod
async def async_responses_api_session_handler(
previous_response_id: str,
litellm_completion_request: dict,
) -> dict:
"""
Async hook to get the chain of previous input and output pairs and return a list of Chat Completion messages
"""
chat_completion_session = ChatCompletionSession(messages=[], litellm_session_id=None)
if previous_response_id:
chat_completion_session = (
await ResponsesSessionHandler.get_chat_completion_message_history_for_previous_response_id(
previous_response_id=previous_response_id
)
)
_messages: Final = litellm_completion_request.get("messages") or []
session_messages: Final = chat_completion_session.get("messages") or []
# If session messages are empty (e.g., no database in test environment),
# we still need to process the new input messages
# Store original _messages before combining for safety check
original_new_messages: Final = _messages.copy() if _messages else []
combined_messages = session_messages + _messages
# Fix: Ensure tool_results have corresponding tool_calls in previous assistant message
# Pass tools parameter to help reconstruct tool_calls if not in cache
tools: Final = litellm_completion_request.get("tools") or []
combined_messages = LiteLLMCompletionResponsesConfig._ensure_tool_results_have_corresponding_tool_calls(
messages=combined_messages, tools=tools
)
# Safety check: Ensure we don't end up with empty messages
# This can happen when using previous_response_id without a database (e.g., in tests)
# and session messages are empty but new input messages exist
if not combined_messages:
# If we end up with empty messages, try to restore from original inputs
if original_new_messages:
# If we had new input messages but they got filtered out,
# restore them (better to have messages than empty list)
# This can happen when tool_call_id is empty and can't be recovered
combined_messages = original_new_messages
elif session_messages:
# If we had session messages but they got filtered out,
# restore them
combined_messages = session_messages
else:
# Both are empty - this likely means function_call_output had empty/invalid call_id
# Provide a helpful error message
import litellm
raise litellm.BadRequestError(
message=(
f"Unable to create messages for completion request. "
f"This can happen when: "
f"1) Using previous_response_id without a session database, AND "
f"2) Input contains only function_call_output with empty or invalid call_id. "
f"Please ensure function_call_output has a valid call_id from a previous response. "
f"Original request: previous_response_id={previous_response_id}"
),
model=litellm_completion_request.get("model", ""),
llm_provider=litellm_completion_request.get("custom_llm_provider", ""),
)
litellm_completion_request["messages"] = combined_messages
litellm_completion_request["litellm_trace_id"] = chat_completion_session.get("litellm_session_id")
return litellm_completion_request
@staticmethod
def _transform_response_input_param_to_chat_completion_message(
input: str | ResponseInputParam,
replay_reasoning: bool = False,
) -> list[
AllMessageValues | GenericChatCompletionMessage | ChatCompletionMessageToolCall | ChatCompletionResponseMessage
]:
"""
Transform a ResponseInputParam into a Chat Completion message
See ``transform_responses_api_input_to_messages`` for what
``replay_reasoning`` means.
"""
messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
] = []
if isinstance(input, str):
messages.append(ChatCompletionUserMessage(role="user", content=input))
elif isinstance(input, list):
existing_tool_call_ids: Final[set[str]] = set()
for _input in input:
chat_completion_messages = (
LiteLLMCompletionResponsesConfig._transform_responses_api_input_item_to_chat_completion_message(
input_item=_input,
replay_reasoning=replay_reasoning,
)
)
if LiteLLMCompletionResponsesConfig._is_input_item_function_call(input_item=_input):
call_id_raw = _input.get("call_id") or _input.get("id") or ""
if call_id_raw:
existing_tool_call_ids.add(str(call_id_raw))
#########################################################
# Merge consecutive function_call items into a single assistant
# message. Anthropic requires that all tool_use blocks appear in
# ONE assistant message immediately followed by the tool_result
# blocks. Without this merging, each function_call creates its own
# assistant message, producing back-to-back assistant messages that
# Anthropic rejects with "tool_use ids were found without
# tool_result blocks immediately after".
#########################################################
if messages:
last_msg = messages[-1]
last_role = (
last_msg.get("role") if isinstance(last_msg, dict) else getattr(last_msg, "role", None)
)
if last_role == "assistant":
for new_msg in chat_completion_messages:
new_role = (
new_msg.get("role") if isinstance(new_msg, dict) else getattr(new_msg, "role", None)
)
if new_role == "assistant":
_raw_tcs = (
new_msg.get("tool_calls")
if isinstance(new_msg, dict)
else getattr(new_msg, "tool_calls", None)
)
new_tcs: list = _raw_tcs if isinstance(_raw_tcs, list) else []
for tc in new_tcs:
LiteLLMCompletionResponsesConfig._add_tool_call_to_assistant(last_msg, tc)
continue
#########################################################
# If Input Item is a Tool Call Output, add it to the tool_call_output_messages list
# preserving the ordering of tool call outputs. Some models require the tool
# result to immediately follow the assistant tool call.
#########################################################
if LiteLLMCompletionResponsesConfig._is_input_item_tool_call_output(input_item=_input):
if not chat_completion_messages:
continue
deduped_in_place: list[
AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage
] = []
for m in chat_completion_messages:
role = ""
if isinstance(m, dict):
role = str(m.get("role") or "")
else:
role = str(getattr(m, "role", "") or "")
# Drop assistant tool_calls wrappers if we already have this call_id
if role == "assistant":
tool_calls: object = (
m.get("tool_calls") if isinstance(m, dict) else getattr(m, "tool_calls", None)
)
call_id = ""
if (
isinstance(tool_calls, Sequence)
and not isinstance(tool_calls, (str, bytes))
and len(tool_calls) > 0
):
first_call = tool_calls[0]
call_id_raw = (
first_call.get("id")
if isinstance(first_call, dict)
else getattr(first_call, "id", None)
)
if call_id_raw:
call_id = str(call_id_raw)
if call_id and call_id in existing_tool_call_ids:
continue
if call_id:
existing_tool_call_ids.add(call_id)
deduped_in_place.append(m)
messages.extend(deduped_in_place)
continue
merged_assistant = LiteLLMCompletionResponsesConfig._merged_trailing_assistant_message(
messages=messages,
chat_completion_messages=chat_completion_messages,
)
if merged_assistant is not None:
messages[-1] = merged_assistant
continue
messages.extend(chat_completion_messages)
if not replay_reasoning:
return messages
return LiteLLMCompletionResponsesConfig._merge_reasoning_only_assistant_messages(messages)
@staticmethod
def _reasoning_only_assistant_message(
reasoning_text: str | None,
thinking_blocks: Sequence[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock] | None,
) -> ChatCompletionResponseMessage:
"""
Build the assistant message that carries a prior turn's reasoning and
nothing else, so a reasoning item never reaches the provider as visible
assistant ``content``.
"""
message: Final = ChatCompletionResponseMessage(role="assistant", content=None)
if reasoning_text:
message["reasoning_content"] = reasoning_text
if thinking_blocks:
message["thinking_blocks"] = list( # mutable-ok: thinking_blocks is a list on the message contract
thinking_blocks
)
return message
@staticmethod
def _merge_reasoning_only_assistant_messages(
messages: list[ # mutable-ok: input sequence
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
],
) -> list[ # mutable-ok: fresh merged list
AllMessageValues | GenericChatCompletionMessage | ChatCompletionMessageToolCall | ChatCompletionResponseMessage
]:
"""
Responses API emits prior-turn reasoning as its own ``reasoning`` input
item, which becomes a standalone assistant message with
``content=None`` + ``reasoning_content``. Chat-completions providers
(e.g. DeepSeek V4, Kimi K2.6) expect the chain-of-thought on the
assistant message that carries the answer or tool calls. This pass
merges standalone reasoning-only assistant messages into the
immediately following assistant message.
Signed ``thinking_blocks`` decoded from ``encrypted_content`` travel the
same way and are placed ahead of any thinking blocks the target message
already carries, because Anthropic and Bedrock verify signatures against
the original block order.
If the reasoning item is not followed by an assistant message (e.g. a
stateless chain replays ``reasoning`` + ``user``), the standalone
reasoning message is preserved so the reasoning is still passed back.
"""
def _role(msg: object) -> str:
if isinstance(msg, dict):
return str(msg.get("role") or "")
return str(getattr(msg, "role", "") or "")
def _reasoning_text(msg: object) -> str | None:
if isinstance(msg, dict):
value = msg.get("reasoning_content") # rebind-ok: branch lookup
else:
value = getattr(msg, "reasoning_content", None) # rebind-ok: branch lookup
return value if isinstance(value, str) and value else None
def _thinking_blocks(
msg: object,
) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...] | None:
if isinstance(msg, dict):
value = msg.get("thinking_blocks") # rebind-ok: branch lookup
else:
value = getattr(msg, "thinking_blocks", None) # rebind-ok: branch lookup
return tuple(value) if isinstance(value, list) and value else None
def _content(msg: object) -> object | None:
if isinstance(msg, dict):
return msg.get("content")
return getattr(msg, "content", None)
def _tool_calls(msg: object) -> object | None:
if isinstance(msg, dict):
return msg.get("tool_calls")
return getattr(msg, "tool_calls", None)
def _apply_pending(
msg: object,
pending_items: Sequence[
tuple[
str | None,
tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...] | None,
]
],
) -> None:
pending_texts: Final = tuple(text for text, _ in pending_items if text)
pending_blocks: Final = tuple(block for _, blocks in pending_items for block in blocks or ())
if pending_texts:
existing_text: Final = _reasoning_text(msg)
combined: Final = "\n".join(pending_texts + ((existing_text,) if existing_text else ()))
if isinstance(msg, dict):
cast(dict[str, object], msg)["reasoning_content"] = combined # cast-ok: mutable reasoning carrier
else:
setattr(msg, "reasoning_content", combined) # noqa: B010 # attribute name is fixed, not dynamic
if pending_blocks:
replayed: Final = list( # mutable-ok: thinking_blocks is a list on the message contract
pending_blocks + (_thinking_blocks(msg) or ())
)
if isinstance(msg, dict):
cast(dict[str, object], msg)["thinking_blocks"] = replayed # cast-ok: mutable reasoning carrier
else:
setattr(msg, "thinking_blocks", replayed) # noqa: B010 # attribute name is fixed, not dynamic
_standalone: Final = LiteLLMCompletionResponsesConfig._reasoning_only_assistant_message
merged: list[ # mutable-ok: accumulator # rebind-ok: accumulator
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
] = [] # mutable-ok: accumulator
pending: list[ # mutable-ok: accumulator # rebind-ok: accumulator
tuple[
str | None,
tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...] | None,
]
] = [] # mutable-ok: accumulator
for msg in messages:
if (
_role(msg) == "assistant"
and _content(msg) is None
and not _tool_calls(msg)
and (_reasoning_text(msg) is not None or _thinking_blocks(msg) is not None)
):
pending.append((_reasoning_text(msg), _thinking_blocks(msg)))
continue
if pending and _role(msg) == "assistant":
_apply_pending(msg, pending)
pending = [] # mutable-ok: reset accumulator
elif pending:
# Not followed by an assistant message — keep the reasoning
# standalone instead of dropping it.
merged.extend( # mutable-ok: append reasoning messages
[_standalone(text, blocks) for text, blocks in pending] # mutable-ok: append reasoning messages
)
pending = [] # mutable-ok: reset accumulator
merged.append(msg)
merged.extend( # mutable-ok: append trailing reasoning
[_standalone(text, blocks) for text, blocks in pending] # mutable-ok: append trailing reasoning
)
return merged
@staticmethod
def _merged_trailing_assistant_message(
messages: Sequence[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
],
chat_completion_messages: Sequence[
AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage
],
) -> ChatCompletionResponseMessage | None:
"""Fold an assistant content message into a directly preceding assistant
tool_calls message. Providers like DeepSeek and Anthropic require tool
results immediately after the tool_calls message, so an assistant message
between them is rejected."""
if not messages or len(chat_completion_messages) != 1:
return None
last_message = messages[-1]
new_message = chat_completion_messages[0]
if not isinstance(last_message, dict):
return None
if last_message.get("role") != "assistant" or new_message.get("role") != "assistant":
return None
if not last_message.get("tool_calls") or last_message.get("content") or new_message.get("tool_calls"):
return None
new_content = new_message.get("content")
if new_content is None:
return None
merged: Final = { # mutable-ok: json.dumps rejects MappingProxyType in outbound chat messages
**last_message,
"content": new_content,
}
return cast(ChatCompletionResponseMessage, merged) # cast-ok: TypedDict spread widens to dict[str, object]
@staticmethod
def _deduplicate_tool_call_output_messages(
tool_call_output_messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
],
existing_tool_call_ids: set[str],
) -> list[
AllMessageValues | GenericChatCompletionMessage | ChatCompletionMessageToolCall | ChatCompletionResponseMessage
]:
"""Return tool call outputs after dropping assistant entries with duplicate call_ids."""
if not tool_call_output_messages:
return []
filtered_messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionMessageToolCall
| ChatCompletionResponseMessage
] = []
seen_tool_call_ids: Final[set[str]] = set(existing_tool_call_ids)
for tool_call_message in tool_call_output_messages:
if isinstance(tool_call_message, dict):
role = tool_call_message.get("role", "")
else:
role = getattr(tool_call_message, "role", "")
call_id = ""
if role == "assistant":
tool_calls: object = None
if isinstance(tool_call_message, dict):
tool_calls = tool_call_message.get("tool_calls")
else:
tool_calls = getattr(tool_call_message, "tool_calls", None)
if (
isinstance(tool_calls, Sequence)
and not isinstance(tool_calls, (str, bytes))
and len(tool_calls) > 0
):
first_call = tool_calls[0]
call_id_raw = None
if isinstance(first_call, dict):
call_id_raw = first_call.get("id")
else:
call_id_raw = getattr(first_call, "id", None)
if call_id_raw:
call_id = str(call_id_raw)
if call_id and call_id in seen_tool_call_ids and role == "assistant":
continue
if call_id and role == "assistant":
seen_tool_call_ids.add(call_id)
filtered_messages.append(tool_call_message)
return filtered_messages
@staticmethod
def _ensure_tool_call_output_has_corresponding_tool_call(
messages: list[AllMessageValues | GenericChatCompletionMessage],
) -> bool:
"""
If any tool call output is present, ensure there is a corresponding tool call/tool_use block
"""
for message in messages:
if message.get("role") == "tool":
return True
return False
@staticmethod
def _find_previous_assistant_idx(
messages: Sequence[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message
],
current_idx: int,
) -> int | None:
"""Find the index of the previous assistant message."""
for j in range(current_idx - 1, -1, -1):
if messages[j].get("role") == "assistant":
return j
return None
@staticmethod
def _recover_tool_call_id_from_assistant(
assistant_message: AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message,
message: AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message,
) -> str:
"""Try to recover empty tool_call_id from assistant message's tool_calls."""
tool_calls_raw: Final = (
assistant_message.get("tool_calls")
if isinstance(assistant_message, dict)
else getattr(assistant_message, "tool_calls", None)
)
if tool_calls_raw and isinstance(tool_calls_raw, list) and len(tool_calls_raw) > 0:
first_tool_call: Final = _OBJECT_LIST_ADAPTER.validate_python(tool_calls_raw)[0]
if isinstance(first_tool_call, dict):
tool_call_id_raw = _ANY_KEY_DICT_ADAPTER.validate_python(first_tool_call).get("id", "")
return str(tool_call_id_raw) if tool_call_id_raw is not None else ""
elif isinstance(first_tool_call, _HasId):
tool_call_id_raw = first_tool_call.id
return str(tool_call_id_raw) if tool_call_id_raw is not None else ""
return ""
@staticmethod
def _get_tool_calls_list(
assistant_message: AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message,
) -> Sequence[object]:
"""Extract tool_calls as a list from assistant message."""
tool_calls_raw: Final = (
assistant_message.get("tool_calls")
if isinstance(assistant_message, dict)
else getattr(assistant_message, "tool_calls", None)
)
if tool_calls_raw is None:
return []
if isinstance(tool_calls_raw, list):
return _OBJECT_LIST_ADAPTER.validate_python(tool_calls_raw)
if isinstance(tool_calls_raw, _SupportsIter) and not isinstance(tool_calls_raw, (str, bytes)):
return list(tool_calls_raw)
return []
@staticmethod
def _check_tool_call_exists(tool_calls: Sequence[object], tool_call_id: str) -> bool:
"""Check if a tool_call with the given ID exists in the list."""
for tool_call in tool_calls:
tool_call_id_to_check: object = None
if isinstance(tool_call, dict):
tool_call_id_to_check = _ANY_KEY_DICT_ADAPTER.validate_python(tool_call).get("id")
elif hasattr(tool_call, "id"):
tool_call_id_to_check = getattr(tool_call, "id", None)
if tool_call_id_to_check == tool_call_id:
return True
return False
@staticmethod
def _reconstruct_tool_call_from_tools(tool_call_id: str, tools: Sequence[object]) -> dict[str, object] | None:
"""Reconstruct a minimal tool_call definition from tools list."""
for tool in tools:
if isinstance(tool, dict):
tool_map = _ANY_KEY_DICT_ADAPTER.validate_python(tool)
tool_function = _ANY_KEY_DICT_ADAPTER.validate_python(tool_map.get("function") or {})
tool_name = tool_function.get("name") or tool_map.get("name") or ""
if tool_name:
return {
"id": tool_call_id,
"type": "function",
"function": {
"name": tool_name,
"arguments": "{}", # We don't know the arguments, use empty
},
}
return None
@staticmethod
def _get_mapping_or_attr_value(obj: object, key: str, default: object = None) -> object:
"""
Safely read a field from dict-like or attribute-based objects.
"""
if obj is None:
return default
if isinstance(obj, dict):
return _ANY_KEY_DICT_ADAPTER.validate_python(obj).get(key, default)
getter: Final = getattr(obj, "get", None)
if callable(getter):
try:
return getter(key, default)
except (TypeError, AttributeError):
pass
return getattr(obj, key, default)
@staticmethod
def _create_tool_call_chunk(
tool_use_definition: Mapping[object, object], tool_call_id: str, index: int
) -> ChatCompletionToolCallChunk:
"""Create a ChatCompletionToolCallChunk from tool_use_definition."""
function_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "function")
function_name_raw: Final = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(function_raw, "name")
function_arguments_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(function_raw, "arguments")
function: Final[dict[str, object]] = {
"name": function_name_raw or "",
"arguments": function_arguments_raw or "{}",
}
tool_use_id_raw: Final = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "id")
tool_use_id: Final[str] = str(tool_use_id_raw) if tool_use_id_raw is not None else str(tool_call_id)
tool_use_type_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "type")
tool_use_type: Final[str] = str(tool_use_type_raw) if tool_use_type_raw is not None else "function"
return ChatCompletionToolCallChunk(
id=tool_use_id,
type=cast(Literal["function"], tool_use_type),
function=ChatCompletionToolCallFunctionChunk(
name=str(function.get("name", "")),
arguments=str(function.get("arguments", "{}")),
),
index=index,
)
@staticmethod
def _normalize_tool_use_definition(tool_use_definition: object, tool_call_id: str) -> dict[object, object] | None:
"""
Normalize cached tool_call definitions to a dict-like shape consumed by _create_tool_call_chunk.
"""
if not tool_use_definition:
return None
if isinstance(tool_use_definition, dict):
normalized_definition: dict[object, object] = _ANY_KEY_DICT_ADAPTER.validate_python(tool_use_definition)
else:
tool_use_id_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "id")
tool_use_type_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "type")
function_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(tool_use_definition, "function")
# Object does not expose the expected tool_call fields.
if tool_use_id_raw is None and tool_use_type_raw is None and function_raw is None:
return None
normalized_definition = {
"id": tool_use_id_raw,
"type": tool_use_type_raw,
"function": function_raw,
}
function_raw = normalized_definition.get("function")
if function_raw is not None and not isinstance(function_raw, dict):
function_name_raw: Final = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(function_raw, "name")
function_arguments_raw: Final = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(
function_raw, "arguments"
)
if function_name_raw is not None or function_arguments_raw is not None:
normalized_definition["function"] = {
"name": function_name_raw,
"arguments": function_arguments_raw,
}
normalized_definition["id"] = normalized_definition.get("id") or tool_call_id
normalized_definition["type"] = normalized_definition.get("type") or "function"
return normalized_definition
@staticmethod
def _add_tool_call_to_assistant(assistant_message: object, tool_call_chunk: ChatCompletionToolCallChunk) -> None:
"""Add a tool_call to an assistant message."""
if isinstance(assistant_message, dict):
prev_assistant_dict: Final = cast(dict[str, object], assistant_message)
if "tool_calls" not in prev_assistant_dict:
prev_assistant_dict["tool_calls"] = []
tool_calls_list: Final = prev_assistant_dict["tool_calls"]
if isinstance(tool_calls_list, list):
tool_calls_list.append(tool_call_chunk)
elif isinstance(assistant_message, _HasToolCalls):
if assistant_message.tool_calls is None:
assistant_message.tool_calls = []
if isinstance(assistant_message.tool_calls, list):
assistant_message.tool_calls.append(tool_call_chunk)
@staticmethod
def _ensure_tool_results_have_corresponding_tool_calls(
messages: Sequence[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message
],
tools: Sequence[object] | None = None,
) -> list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message
]:
"""
Ensure that tool_result messages have corresponding tool_calls in the previous assistant message.
This is critical for Anthropic API which requires that each tool_result block has a
corresponding tool_use block in the previous assistant message.
Args:
messages: List of messages that may include tool_result messages
tools: Optional list of tools that can be used to reconstruct tool_calls if not in cache
Returns:
List of messages with tool_calls added to assistant messages when needed
"""
if not messages:
return list(messages)
# Create a deep copy to avoid modifying the original (use list() so we can mutate and return List)
import copy
fixed_messages: list[
AllMessageValues
| GenericChatCompletionMessage
| ChatCompletionResponseMessage
| ChatCompletionMessageToolCall
| Message
] = list(copy.deepcopy(messages))
messages_to_remove: Final = []
# Count non-tool messages to avoid removing all messages
# This prevents empty messages list when using previous_response_id without a database
non_tool_messages_count: Final = sum(1 for msg in fixed_messages if msg.get("role") != "tool")
for i, message in enumerate(fixed_messages):
# Only process tool messages - check role first to narrow the type
if message.get("role") != "tool":
continue
# At this point, we know it's a tool message, so it should have tool_call_id
# Use get() with default to safely access tool_call_id
tool_call_id_raw = (
message.get("tool_call_id") if isinstance(message, dict) else getattr(message, "tool_call_id", None)
)
tool_call_id: str = str(tool_call_id_raw) if tool_call_id_raw is not None else ""
prev_assistant_idx = LiteLLMCompletionResponsesConfig._find_previous_assistant_idx(fixed_messages, i)
# Try to recover empty tool_call_id from previous assistant message
if not tool_call_id and prev_assistant_idx is not None:
prev_assistant = fixed_messages[prev_assistant_idx]
tool_call_id = LiteLLMCompletionResponsesConfig._recover_tool_call_id_from_assistant(
prev_assistant, message
)
if tool_call_id:
# Type-safe way to set tool_call_id on tool message
if isinstance(message, dict):
# Cast to dict to allow setting tool_call_id
message_dict = cast(dict[str, object], message)
message_dict["tool_call_id"] = tool_call_id
elif hasattr(message, "tool_call_id"):
setattr(message, "tool_call_id", tool_call_id)
# Only remove messages with empty tool_call_id if we have other non-tool messages
# This prevents ending up with an empty messages list when using previous_response_id
# without a database (e.g., in tests where session messages are empty)
if not tool_call_id:
# If we have non-tool messages, we can safely remove this tool message
# But if removing it would leave us with no messages, keep it to avoid empty list
if non_tool_messages_count > 0:
messages_to_remove.append(i)
# If no non-tool messages, keep the tool message even with empty call_id
# The API will return a proper error message about the missing tool_use block
continue
# Check if the previous assistant message has the corresponding tool_call
# This needs to run for ALL tool messages with a valid tool_call_id,
# not just those that had an empty tool_call_id initially
if prev_assistant_idx is not None and tool_call_id:
prev_assistant = fixed_messages[prev_assistant_idx]
tool_calls = LiteLLMCompletionResponsesConfig._get_tool_calls_list(prev_assistant)
if not LiteLLMCompletionResponsesConfig._check_tool_call_exists(tool_calls, tool_call_id):
_tool_use_definition: object = TOOL_CALLS_CACHE.get_cache(key=tool_call_id)
if not _tool_use_definition and tools:
_tool_use_definition = LiteLLMCompletionResponsesConfig._reconstruct_tool_call_from_tools(
tool_call_id, tools
)
normalized_tool_use_definition = LiteLLMCompletionResponsesConfig._normalize_tool_use_definition(
_tool_use_definition, tool_call_id
)
if normalized_tool_use_definition:
tool_call_chunk = LiteLLMCompletionResponsesConfig._create_tool_call_chunk(
normalized_tool_use_definition,
tool_call_id,
len(tool_calls),
)
LiteLLMCompletionResponsesConfig._add_tool_call_to_assistant(prev_assistant, tool_call_chunk)
# Remove messages with empty tool_call_id that couldn't be fixed
for idx in reversed(messages_to_remove):
fixed_messages.pop(idx)
return fixed_messages
@staticmethod
def _transform_responses_api_input_item_to_chat_completion_message(
input_item: Mapping[str, object],
replay_reasoning: bool = False,
) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage]:
"""
Transform a Responses API input item into a Chat Completion message
- EasyInputMessageParam
- Message
- ResponseOutputMessageParam
- ResponseFileSearchToolCallParam
- ResponseComputerToolCallParam
- ComputerCallOutput
- ResponseFunctionWebSearchParam
- ResponseFunctionToolCallParam
- FunctionCallOutput
- ResponseReasoningItemParam
- ItemReference
"""
if LiteLLMCompletionResponsesConfig._is_input_item_tool_call_output(input_item):
# handle executed tool call results
return (
LiteLLMCompletionResponsesConfig._transform_responses_api_tool_call_output_to_chat_completion_message(
tool_call_output=input_item
)
)
elif LiteLLMCompletionResponsesConfig._is_input_item_function_call(input_item):
# handle function call input items
return LiteLLMCompletionResponsesConfig._transform_responses_api_function_call_to_chat_completion_message(
function_call=cast( # cast-ok: callee coerces every field it reads with `or ""` / str()
Mapping[str, str], input_item
)
)
elif input_item.get("type") == "reasoning":
# A ResponseReasoningItemParam carries the prior-turn chain-of-thought.
# Chat-completions providers (DeepSeek V4, Kimi K2.6, ...) expect this
# to be replayed as `reasoning_content` on an assistant message, not as
# visible `content` (prompt pollution) and not dropped (DeepSeek V4
# rejects multi-turn requests with a missing `reasoning_content`).
# Callers that only inspect the request keep reading the text as
# message `content`, summary-only items included: whatever the
# provider-bound branch below replays must stay scannable.
if not replay_reasoning:
# `content` wins only when it is what the provider-bound branch
# would replay; an empty or block-only `content` falls back to
# the summary text, which is what that branch replays instead.
inspectable: Final[object] = (
input_item.get("content")
if LiteLLMCompletionResponsesConfig._reasoning_text_from_content(input_item) is not None
else LiteLLMCompletionResponsesConfig._reasoning_text_from_summary(input_item)
or input_item.get("content")
)
if inspectable is None:
return [] # mutable-ok: empty drop result
return [ # mutable-ok: single message result
GenericChatCompletionMessage(
role=_input_item_role(input_item),
content=LiteLLMCompletionResponsesConfig._transform_responses_api_content_to_chat_completion_content(
inspectable
),
)
]
reasoning_text = LiteLLMCompletionResponsesConfig._extract_reasoning_text_from_input_item( # rebind-ok: extraction result
input_item
)
thinking_blocks = LiteLLMCompletionResponsesConfig._decode_thinking_blocks_from_input_item( # rebind-ok: extraction result
input_item
)
if not reasoning_text and not thinking_blocks:
return [] # mutable-ok: empty drop result
return [ # mutable-ok: single message result
LiteLLMCompletionResponsesConfig._reasoning_only_assistant_message(
reasoning_text=reasoning_text,
thinking_blocks=thinking_blocks,
)
]
else:
content: Final[object] = input_item.get("content")
# Handle None content: Responses API allows None content, but GenericChatCompletionMessage requires content
# Since guardrails skip None content anyway, we return empty list to exclude it from structured messages
if content is None:
return []
return [
GenericChatCompletionMessage(
role=_input_item_role(input_item),
content=LiteLLMCompletionResponsesConfig._transform_responses_api_content_to_chat_completion_content(
content
),
)
]
@staticmethod
def _reasoning_text_from_content(input_item: Mapping[str, object]) -> str | None:
"""
Plaintext a ResponseReasoningItemParam carries in ``content``.
Handles content as a string and content as a list of blocks
(output_text / summary_text / text). Returns None when the item has
no content, or only opaque blocks (e.g. encrypted_content).
"""
content: Final[object] = input_item.get("content")
if isinstance(content, str) and content.strip():
return content
if isinstance(content, list):
text_parts: Final = tuple(
text.strip()
for block in content
if isinstance(block, Mapping)
and block.get("type") not in ("encrypted_content", "redacted_thinking")
and isinstance(text := block.get("text"), str)
and text.strip()
)
if text_parts:
return "\n".join(text_parts)
return None
@staticmethod
def _reasoning_text_from_summary(input_item: Mapping[str, object]) -> str | None:
"""
Plaintext a ResponseReasoningItemParam carries in ``summary``.
Guardrail traversal in litellm/proxy/guardrails/_content_utils.py
inspects and rewrites these summary blocks before they are forwarded.
"""
summary: Final[object] = input_item.get("summary")
if not isinstance(summary, list):
return None
text_parts: Final = tuple(
text.strip()
for block in summary
if isinstance(block, Mapping) and isinstance(text := block.get("text"), str) and text.strip()
)
return "\n".join(text_parts) if text_parts else None
@staticmethod
def _extract_reasoning_text_from_input_item(input_item: Mapping[str, object]) -> str | None:
"""
Extract plaintext reasoning from a ResponseReasoningItemParam.
``content`` wins, ``summary`` is the fallback. Returns None when only
opaque forms (e.g. encrypted_content) are present.
"""
return LiteLLMCompletionResponsesConfig._reasoning_text_from_content(
input_item
) or LiteLLMCompletionResponsesConfig._reasoning_text_from_summary(input_item)
@staticmethod
def _decode_thinking_blocks_from_input_item(
input_item: Mapping[str, object],
) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...] | None:
"""
Decode ``encrypted_content`` written by ``_encode_thinking_blocks`` back
into the signed thinking blocks it serialized.
LiteLLM writes this field itself for providers whose reasoning is signed
(Anthropic, Bedrock converse): it is a JSON array of the provider's own
``thinking`` / ``redacted_thinking`` blocks, not an opaque OpenAI blob.
Replaying the blocks on the assistant message is what lets the provider
verify the signature and keep the prior chain-of-thought.
Returns None for anything this deployment did not write, so a genuinely
opaque blob is still skipped rather than forwarded as garbage.
"""
encrypted_content: Final[object] = input_item.get("encrypted_content")
if not isinstance(encrypted_content, str) or not encrypted_content.strip():
return None
try:
decoded: Final[object] = cast(object, json.loads(encrypted_content)) # cast-ok: json.loads returns Any
except ValueError:
return None
if not isinstance(decoded, list):
return None
blocks: Final = tuple(
cast( # cast-ok: shape validated by _is_replayable_thinking_block
ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock,
block,
)
for block in decoded
if isinstance(block, Mapping) and LiteLLMCompletionResponsesConfig._is_replayable_thinking_block(block)
)
return blocks or None
@staticmethod
def _is_replayable_thinking_block(block: Mapping[str, object]) -> bool:
"""
A thinking block is only worth replaying when the provider can verify
it: a ``thinking`` block needs its signature, a ``redacted_thinking``
block needs its opaque data.
"""
block_type: Final[object] = block.get("type")
if block_type == "thinking":
return bool(block.get("signature"))
if block_type == "redacted_thinking":
return bool(block.get("data"))
return False
@staticmethod
def _is_input_item_tool_call_output(input_item: Mapping[str, object]) -> bool:
"""
Check if the input item is a tool call output
"""
return input_item.get("type") in [
"function_call_output",
"custom_tool_call_output",
"web_search_call",
"computer_call_output",
"tool_result", # Anthropic/MCP format
]
@staticmethod
def _is_input_item_function_call(input_item: Mapping[str, object]) -> bool:
"""
Check if the input item is a function call or custom tool call.
Both need to be reconstructed as assistant tool_calls for Chat
Completions providers.
"""
return input_item.get("type") in ("function_call", "custom_tool_call")
@staticmethod
def _transform_responses_api_tool_call_output_to_chat_completion_message(
tool_call_output: Mapping[str, object],
) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage]:
"""
ChatCompletionToolMessage is used to indicate the output from a tool call
"""
call_id: Final = tool_call_output.get("call_id")
# If call_id is missing or empty, skip this message
# Empty call_id means we can't create a valid tool message
if not call_id:
return []
def _normalize_function_call_output_to_tool_content(
output: object,
) -> str | list[ChatCompletionTextObject | ChatCompletionImageObject]:
"""
Normalize Responses API function_call_output.output into a shape that downstream
chat adapters (esp. Gemini) can reliably consume.
OpenAI Responses API typically uses:
- output: string
Some clients/adapters send:
- output: [{"type": "input_text", "text": "..."}, {"type": "input_image", ...}]
For chat tool messages we normalize to either:
- string (preferred)
- list of {"type": "text"|"image_url", ...} blocks (for multimodal tool outputs)
"""
if output is None:
return ""
if isinstance(output, str):
return output
# Some adapters represent tool output as a list of "input_*" parts
if isinstance(output, list):
normalized_blocks: Final[list[ChatCompletionTextObject | ChatCompletionImageObject]] = []
text_acc: Final[list[str]] = []
for part in output:
if not isinstance(part, dict):
continue
part_type = part.get("type")
if part_type in ("input_text", "output_text", "text"):
txt = part.get("text")
if isinstance(txt, str) and txt:
text_acc.append(txt)
normalized_blocks.append({"type": "text", "text": txt})
elif part_type in ("input_image", "image_url"):
image_url_val = part.get("image_url") or part.get("url")
if isinstance(image_url_val, dict):
url = image_url_val.get("url")
if isinstance(url, str) and url:
normalized_blocks.append({"type": "image_url", "image_url": {"url": url}})
elif isinstance(image_url_val, str) and image_url_val:
normalized_blocks.append(
{
"type": "image_url",
"image_url": {"url": image_url_val},
}
)
# Prefer structured blocks if we have images; otherwise return a string.
if any(b.get("type") == "image_url" for b in normalized_blocks):
# Ensure we include any accumulated text as text blocks too
return normalized_blocks
if text_acc:
return "".join(text_acc)
try:
# last resort: keep something meaningful for providers that require a string
import json as _json
return _json.dumps(output)
except Exception:
return str(output)
# Fallback for dict/number/etc.
try:
import json as _json
return _json.dumps(output)
except Exception:
return str(output)
tool_output_message: Final = ChatCompletionToolMessage(
role="tool",
content=_normalize_function_call_output_to_tool_content(tool_call_output.get("output")),
tool_call_id=str(call_id),
)
_tool_use_definition: Final = TOOL_CALLS_CACHE.get_cache(
key=tool_call_output.get("call_id") or "",
)
if _tool_use_definition:
"""
Append the tool use definition to the list of messages
Providers like Anthropic require the tool use definition to be included with the tool output
- Input:
{'function':
arguments:'{"command": ["echo","\\n\\n Hello\\n\\n\\n Hi
\\n\\n",">","index.html"]}',
name='shell',
'id': 'toolu_018KFWsEySHjdKZPdUzXpymJ',
'type': 'function'
}
- Output:
{
"id": "toolu_018KFWsEySHjdKZPdUzXpymJ",
"type": "function",
"function": {
"name": "get_weather",
"arguments": "{\"latitude\":48.8566,\"longitude\":2.3522}"
}
}
"""
function: Final[dict] = _tool_use_definition.get("function") or {}
tool_call_chunk: Final = ChatCompletionToolCallChunk(
id=_tool_use_definition.get("id") or "",
type=cast(Literal["function"], _tool_use_definition.get("type") or "function"),
function=ChatCompletionToolCallFunctionChunk(
name=function.get("name") or "",
arguments=str(function.get("arguments") or ""),
),
index=0,
)
chat_completion_response_message: Final = ChatCompletionResponseMessage(
tool_calls=[tool_call_chunk],
role="assistant",
)
return [chat_completion_response_message, tool_output_message]
return [tool_output_message]
@staticmethod
def _transform_responses_api_function_call_to_chat_completion_message(
function_call: Mapping[str, str],
) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage]:
"""
Transform a Responses API function_call into a Chat Completion message with tool calls
Handles Input items of this type:
function_call:
```json
{
"type": "function_call",
"arguments":"{\"location\": \"São Paulo, Brazil\"}",
"call_id": "call_v2wlBzrlTIFl9FxPeY774GHZ",
"name": "get_weather",
"id": "fc_685c42deefc0819a822b6936faaa30be0c76bc1491ab6619",
"status": "completed"
}
```
"""
# Create a tool call for the function call. Custom tool calls
# store their payload in "input" (raw string) rather than
# "arguments" (JSON string), so normalize to arguments here.
raw_arguments = function_call.get("arguments")
if raw_arguments == REDACTED_BY_LITELLM:
# redaction stores the bare sentinel (invalid JSON) in arguments
raw_arguments = REDACTED_TOOL_CALL_ARGUMENTS_PLACEHOLDER
if not raw_arguments and function_call.get("type") == "custom_tool_call":
raw_input: Final = function_call.get("input") or ""
raw_arguments = json.dumps({"content": raw_input}) if raw_input else ""
raw_name: Final = function_call.get("name") or ""
namespace: Final = function_call.get("namespace") or ""
qualify: Final = bool(namespace) and function_call.get("type") != "custom_tool_call"
tool_call: Final = ChatCompletionToolCallChunk(
id=function_call.get("call_id") or function_call.get("id") or "",
type="function",
function=ChatCompletionToolCallFunctionChunk(
name=f"{namespace}__{raw_name}" if qualify else raw_name,
arguments=str(raw_arguments or ""),
),
index=0,
)
# Create an assistant message with the tool call
chat_completion_response_message: Final = ChatCompletionResponseMessage(
tool_calls=[tool_call],
role="assistant",
content=None, # Function calls don't have content
)
return [chat_completion_response_message]
@staticmethod
def _resolve_file_id(item: Mapping[str, object]) -> object:
"""
Return the effective file_id for a Responses API input_file item.
Explicit file_id takes precedence; file_url is used as fallback so
downstream providers (Anthropic, Gemini) can handle the URL natively.
"""
return item.get("file_id") or item.get("file_url") or None
@staticmethod
def _transform_input_file_item_to_file_item(item: Mapping[str, object]) -> dict[str, object]:
"""
Transform a Responses API input_file item to a Chat Completion file item
Args:
item: Dictionary containing input_file type with file_id, file_data, and/or file_url
Returns:
Dictionary with transformed file structure for Chat Completion
"""
file_dict: Final[dict[str, object]] = {}
file_id: Final = LiteLLMCompletionResponsesConfig._resolve_file_id(item)
if file_id:
file_dict["file_id"] = file_id
if item.get("file_data"):
file_dict["file_data"] = item["file_data"]
new_item: Final[dict[str, object]] = {"type": "file", "file": file_dict}
if "cache_control" in item:
new_item["cache_control"] = item["cache_control"]
return new_item
@staticmethod
def _transform_input_image_item_to_image_item(
item: Mapping[str, str],
) -> ChatCompletionImageObject:
"""
Transform a Responses API input_image item to a Chat Completion image item
"""
image_url_obj: Final = ChatCompletionImageUrlObject(
url=item.get("image_url") or "", detail=item.get("detail") or "auto"
)
return ChatCompletionImageObject(type="image_url", image_url=image_url_obj)
@staticmethod
def _transform_responses_api_content_to_chat_completion_content(
content: object,
) -> str | list[str | dict[str, object]]:
"""
Transform a Responses API content into a Chat Completion content
Note: This function should not be called with None content.
Callers should check for None before calling this function.
"""
if content is None:
# Defensive check: should not happen if callers check first
# Return empty string as fallback to avoid type errors
return ""
elif isinstance(content, str):
return content
elif isinstance(content, list):
content_list: Final[list[str | dict[str, object]]] = []
for item in content:
if isinstance(item, str):
content_list.append(item)
elif isinstance(item, dict):
if item.get("type") == "input_file":
content_list.append(
LiteLLMCompletionResponsesConfig._transform_input_file_item_to_file_item(item)
)
elif item.get("type") == "input_image":
image_block = _STR_KEY_DICT_ADAPTER.validate_python(
dict(LiteLLMCompletionResponsesConfig._transform_input_image_item_to_image_item(item))
)
if "cache_control" in item:
image_block["cache_control"] = item["cache_control"]
content_list.append(image_block)
elif item.get("type") == "encrypted_content":
encrypted_content = item.get("encrypted_content")
if encrypted_content is not None:
content_list.append(
OpenAIChatCompletionTextObject(type="text", text=str(encrypted_content))
)
else:
# Skip text blocks with None text to avoid downstream errors
text_value = item.get("text")
if text_value is None:
continue
content_block: dict[str, object] = {
"type": LiteLLMCompletionResponsesConfig._get_chat_completion_request_content_type(
item.get("type") or "text"
),
"text": text_value,
}
if "cache_control" in item:
content_block["cache_control"] = item["cache_control"]
content_list.append(content_block)
return content_list
else:
raise ValueError(f"Invalid content type: {type(content)}")
@staticmethod
def _get_chat_completion_request_content_type(
content_type: str,
) -> ValidChatCompletionMessageContentTypesLiteral:
"""
Transform Responses API content type to valid Chat Completion content type.
Returns one of ValidChatCompletionMessageContentTypes:
- User: "text", "image_url", "input_audio", "audio_url", "document",
"guarded_text", "video_url", "file"
- Assistant: "text", "thinking", "redacted_thinking"
"""
# Responses API content has `input_` prefix, if it exists, remove it
if content_type.startswith("input_"):
stripped: Final = content_type[len("input_") :]
# Validate stripped type is valid, otherwise default to "text"
if stripped in ValidChatCompletionMessageContentTypes:
return stripped
# Handle input_audio -> input_audio (it's already valid)
if stripped == "audio":
return "input_audio"
return "text"
# Map Responses API specific types to valid Chat Completion types
if content_type in ["tool_result", "output_text"]:
return "text"
# Return as-is if it's a valid type, otherwise default to "text"
if content_type in ValidChatCompletionMessageContentTypes:
return content_type
return "text"
@staticmethod
def transform_instructions_to_system_message(
instructions: str | None,
) -> ChatCompletionSystemMessage:
"""
Transform a Instructions into a system message
"""
return ChatCompletionSystemMessage(role="system", content=instructions or "")
@staticmethod
def _build_ns_chat_tool(
namespace: str,
namespace_description: str,
namespace_tool: NamespaceTool,
nested: bool,
) -> ChatCompletionToolParam | None:
if nested and namespace_tool.get("type") != "function":
return None
raw_parameters: Final = namespace_tool.get("parameters")
parameters: Final = (
MappingProxyType(raw_parameters) if isinstance(raw_parameters, Mapping) else MappingProxyType({})
)
normalized_parameters: Final = (
parameters if parameters and "type" in parameters else MappingProxyType({**parameters, "type": "object"})
)
tool_name: Final = str(namespace_tool.get("name") or "")
raw_description: Final = str(namespace_tool.get("description") or "")
description: Final = (
f"{namespace_description}\n\n{raw_description}"
if nested and namespace_description and raw_description
else namespace_description
if nested and namespace_description
else raw_description
)
chat_tool_name: Final = f"{namespace}__{tool_name}" if nested else tool_name
function: Final = ChatCompletionToolParamFunctionChunk(
name=chat_tool_name,
description=description,
parameters=dict( # mutable-ok: json.dumps rejects MappingProxyType in the outbound payload
normalized_parameters
),
strict=bool(namespace_tool.get("strict", False)),
)
allowed_callers: Final = validated_allowed_callers(namespace_tool.get("allowed_callers"))
if allowed_callers is None:
return ChatCompletionToolParam(type="function", function=function)
return ChatCompletionToolParam(type="function", function=function, allowed_callers=allowed_callers)
@staticmethod
def _namespace_chat_tools(tool: NamespaceTool) -> tuple[ChatCompletionToolParam, ...]:
namespace: Final = str(tool.get("name") or "")
namespace_description: Final = str(tool.get("description") or "")
namespace_tools: Final = tool.get("tools")
if isinstance(namespace_tools, Sequence) and not isinstance(namespace_tools, (str, bytes)):
return tuple(
chat_tool
for raw_tool in namespace_tools
if isinstance(raw_tool, Mapping)
if (
chat_tool := LiteLLMCompletionResponsesConfig._build_ns_chat_tool(
namespace,
namespace_description,
raw_tool,
True,
)
)
is not None
)
flat_tool: Final = LiteLLMCompletionResponsesConfig._build_ns_chat_tool(
namespace, namespace_description, tool, False
)
return (flat_tool,) if flat_tool is not None else ()
@staticmethod
def _validate_namespace_name_collisions(tools: ResponseTools) -> None:
top_level_function_names: Final = frozenset(
str(tool.get("name") or "") for tool in tools or () if tool.get("type") == "function"
)
flattened_namespace_names: Final = frozenset(
f"{(tool.get('name') or '')!s}__{(namespace_tool.get('name') or '')!s}"
for tool in tools or ()
if tool.get("type") == "namespace"
for namespace_tools in (tool.get("tools"),)
if isinstance(namespace_tools, Sequence) and not isinstance(namespace_tools, (str, bytes))
for namespace_tool in namespace_tools
if isinstance(namespace_tool, Mapping) and namespace_tool.get("type") == "function"
)
conflicting_tool_names: Final = top_level_function_names & flattened_namespace_names
if conflicting_tool_names:
raise ValueError(
"Top-level function names conflict with flattened namespace tools: "
+ ", ".join(sorted(conflicting_tool_names))
)
@staticmethod
def transform_responses_api_tools_to_chat_completion_tools(
tools: list[FunctionToolParam | OpenAIMcpServerTool] | None,
) -> tuple[
list[ChatCompletionToolParam | OpenAIMcpServerTool],
OpenAIWebSearchOptions | None,
]:
"""
Transform a Responses API tools into a Chat Completion tools
"""
if tools is None:
return [], None
LiteLLMCompletionResponsesConfig._validate_namespace_name_collisions(tools)
chat_completion_tools: Final[list[ChatCompletionToolParam | OpenAIMcpServerTool]] = []
web_search_options: OpenAIWebSearchOptions | None = None
for tool in tools:
if tool.get("type") == "mcp":
chat_completion_tools.append(cast(OpenAIMcpServerTool, tool))
elif tool.get("type") == "web_search_preview" or tool.get("type") == "web_search":
_search_context_size: Literal["low", "medium", "high"] = cast(
Literal["low", "medium", "high"], tool.get("search_context_size")
)
_user_location: OpenAIWebSearchUserLocation | None = cast(
OpenAIWebSearchUserLocation | None,
tool.get("user_location") or None,
)
web_search_options = OpenAIWebSearchOptions(
search_context_size=_search_context_size,
user_location=_user_location,
)
elif tool.get("type") == "function":
typed_tool = cast(FunctionToolParam, tool)
# Ensure parameters has "type": "object" as required by providers like Anthropic
parameters = dict(typed_tool.get("parameters", {}) or {})
if not parameters or "type" not in parameters:
parameters["type"] = "object"
chat_completion_tool: dict[str, object] = {
"type": "function",
"function": {
"name": typed_tool.get("name") or "",
"description": typed_tool.get("description") or "",
"parameters": parameters,
"strict": typed_tool.get("strict", False) or False,
},
}
if tool.get("cache_control"):
chat_completion_tool["cache_control"] = tool.get("cache_control")
if tool.get("defer_loading"):
chat_completion_tool["defer_loading"] = tool.get("defer_loading")
if tool.get("allowed_callers"):
chat_completion_tool["allowed_callers"] = tool.get("allowed_callers")
if tool.get("input_examples"):
chat_completion_tool["input_examples"] = tool.get("input_examples")
chat_completion_tools.append(cast(ChatCompletionToolParam, chat_completion_tool))
elif tool.get("type") == "namespace":
chat_completion_tools.extend(LiteLLMCompletionResponsesConfig._namespace_chat_tools(tool))
elif tool.get("type") == "custom":
converted = convert_custom_tool_to_function_tool(tool)
if converted is not None:
chat_completion_tools.append(converted)
else:
_tool_type = tool.get("type")
if _tool_type in ("computer_use", "image_generation", "shell"):
# Drop unsupported Responses-API-only tool types that have no
# Chat Completions equivalent. Passing them through verbatim
# causes providers to reject the request with "'function' is a
# required property".
verbose_logger.warning(
"Dropping Responses API tool of type '%s': it has no Chat Completions "
"equivalent and the target provider would reject the request.",
_tool_type,
)
continue
chat_completion_tools.append(cast(ChatCompletionToolParam | OpenAIMcpServerTool, tool))
return chat_completion_tools, web_search_options
@staticmethod
def transform_chat_completion_tool_params_to_responses_api_tools(
chat_completion_tools: list[ChatCompletionToolParam | OpenAIMcpServerTool] | None,
) -> list[dict[str, object]]:
"""
Transform Chat Completion tool params (e.g. from guardrail output) back to
Responses API request tool format. Inverse of
transform_responses_api_tools_to_chat_completion_tools for the tools list.
"""
if chat_completion_tools is None or not chat_completion_tools:
return []
result: Final[list[dict[str, object]]] = []
for tool in chat_completion_tools:
if not isinstance(tool, dict):
result.append(tool)
continue
if tool.get("type") == "function":
fn = cast(_ToolFunctionDefinition, tool.get("function") or {})
parameters = dict(fn.get("parameters", {}) or {})
if not parameters or "type" not in parameters:
parameters["type"] = "object"
responses_tool: dict[str, object] = {
"type": "function",
"name": fn.get("name") or "",
"description": fn.get("description") or "",
"parameters": parameters,
"strict": fn.get("strict", False) or False,
}
if tool.get("cache_control") is not None:
responses_tool["cache_control"] = tool.get("cache_control")
if tool.get("defer_loading") is not None:
responses_tool["defer_loading"] = tool.get("defer_loading")
if tool.get("allowed_callers") is not None:
responses_tool["allowed_callers"] = tool.get("allowed_callers")
if tool.get("input_examples") is not None:
responses_tool["input_examples"] = tool.get("input_examples")
result.append(responses_tool)
else:
# mcp or other: pass through unchanged
result.append(dict(tool))
return result
@staticmethod
def namespace_tool_name_map(tools: ResponseTools) -> NamespaceNameMap:
namespace_entries: Final = tuple(
(str(tool.get("name") or ""), str(namespace_tool.get("name") or ""))
for tool in tools or ()
if tool.get("type") == "namespace"
for namespace_tools in (tool.get("tools"),)
if isinstance(namespace_tools, Sequence) and not isinstance(namespace_tools, (str, bytes))
for namespace_tool in namespace_tools
if isinstance(namespace_tool, Mapping) and namespace_tool.get("type") == "function"
)
top_level_function_names: Final = frozenset(
str(tool.get("name") or "") for tool in tools or () if tool.get("type") == "function"
)
unqualified_counts: Final = MappingProxyType(
{
tool_name: sum(1 for _, candidate_name in namespace_entries if candidate_name == tool_name)
for tool_name in frozenset(tool_name for _, tool_name in namespace_entries)
}
)
unambiguous_entries: Final = tuple(
(tool_name, (namespace, tool_name))
for namespace, tool_name in namespace_entries
if tool_name not in top_level_function_names and unqualified_counts[tool_name] == 1
)
qualified_entries: Final = tuple(
(f"{namespace}__{tool_name}", (namespace, tool_name)) for namespace, tool_name in namespace_entries
)
return MappingProxyType(dict(qualified_entries + unambiguous_entries))
@staticmethod
def _restore_namespace_tool_name(tool_name: str, names: NamespaceNameMap) -> tuple[str, str | None]:
mapped = names.get(tool_name)
if mapped is None:
return tool_name, None
namespace, restored_tool_name = mapped
return restored_tool_name, namespace
@staticmethod
def transform_chat_completion_tools_to_responses_tools(
chat_completion_response: ModelResponse,
responses_api_request: ResponsesAPIOptionalRequestParams | None = None,
) -> list[ResponseFunctionToolCall | CustomToolCallOutputItem]:
"""
Transform a Chat Completion tools into a Responses API tools.
For custom tools (e.g. apply_patch), returns CustomToolCallOutputItem
with ``type="custom_tool_call"``. For regular function tools, returns
``ResponseFunctionToolCall`` with ``type="function_call"``.
"""
all_chat_completion_tools: Final[list[ChatCompletionMessageToolCall]] = []
for choice in chat_completion_response.choices:
if isinstance(choice, Choices):
if choice.message.tool_calls:
all_chat_completion_tools.extend(choice.message.tool_calls)
for tool_call in choice.message.tool_calls:
TOOL_CALLS_CACHE.set_cache(
key=tool_call.id,
value=tool_call,
)
request_tools: Final = responses_api_request.get("tools") if responses_api_request is not None else None
custom_tool_names: Final = extract_custom_tool_names(request_tools)
namespace_tool_names: Final = LiteLLMCompletionResponsesConfig.namespace_tool_name_map(request_tools)
responses_tools: Final[list[ResponseFunctionToolCall | CustomToolCallOutputItem]] = []
for tool in all_chat_completion_tools:
if tool.type == "function":
function_definition = tool.function
tool_name = function_definition.name or ""
tool_id = tool.id or ""
tool_arguments = function_definition.get("arguments") or ""
# Check if this is a custom tool
if is_custom_tool_call(tool_name, custom_tool_names):
# Build custom_tool_call output item
input_str = unwrap_custom_tool_arguments(tool_arguments)
custom_item = CustomToolCallOutputItem(
type="custom_tool_call",
call_id=tool_id,
id=tool_id,
name=tool_name,
input=input_str,
status=function_definition.get("status") or "completed",
)
responses_tools.append(custom_item)
else:
# Build regular function_call output item
restore_name = LiteLLMCompletionResponsesConfig._restore_namespace_tool_name
tool_name, namespace = restore_name(tool_name, namespace_tool_names)
provider_specific_fields: dict | None = None
if hasattr(tool, "provider_specific_fields") and getattr(tool, "provider_specific_fields", None):
provider_specific_fields = getattr(tool, "provider_specific_fields")
if not isinstance(provider_specific_fields, dict):
provider_specific_fields = (
dict(provider_specific_fields) if hasattr(provider_specific_fields, "__dict__") else {}
)
elif hasattr(function_definition, "provider_specific_fields") and getattr(
function_definition, "provider_specific_fields", None
):
provider_specific_fields = getattr(function_definition, "provider_specific_fields")
if not isinstance(provider_specific_fields, dict):
provider_specific_fields = (
dict(provider_specific_fields) if hasattr(provider_specific_fields, "__dict__") else {}
)
output_tool_call: ResponseFunctionToolCall = ResponseFunctionToolCall(
name=tool_name,
arguments=tool_arguments,
call_id=tool_id,
id=tool_id,
type="function_call",
status=function_definition.get("status") or "completed",
)
if namespace:
output_tool_call.namespace = namespace
# Pass through provider_specific_fields as-is if present
if provider_specific_fields:
setattr(
output_tool_call,
"provider_specific_fields",
provider_specific_fields,
)
responses_tools.append(output_tool_call)
return responses_tools
@staticmethod
def _map_chat_completion_finish_reason_to_responses_status(
finish_reason: str | None,
) -> ResponsesAPIStatus:
"""
Map chat completion finish_reason to responses API status.
Chat completion finish_reason values include: "stop", "length", "tool_calls", "content_filter", "function_call", "refusal"
Responses API status values are: "completed", "failed", "in_progress", "cancelled", "queued", "incomplete"
Args:
finish_reason: The finish_reason from a chat completion response
Returns:
The corresponding responses API status value (one of ResponsesAPIStatus)
"""
if finish_reason is None:
return "completed"
# Map finish reasons to status
if finish_reason in ["stop", "tool_calls", "function_call"]:
return "completed"
elif finish_reason in ["length", "content_filter", "refusal"]:
return "incomplete"
else:
# Default to completed for unknown finish reasons
return "completed"
@staticmethod
def _tool_call_id_from_responses_item(item_id: str | None, call_id: str | None) -> str:
"""Bedrock Mantle returns a non-unique, index-based ``call_id`` (``call_0``,
``call_1``, ... that resets every response) alongside a unique ``id``
(``fc_...``). ``call_id`` is the canonical Responses API correlation key, so
prefer it; fall back to the unique ``id`` only when ``call_id`` is absent or
in that degenerate index form, otherwise multi-turn tool calls collide and an
agent cannot correlate its tool results."""
if call_id and re.fullmatch(r"call_\d+", call_id) is None:
return call_id
return item_id or call_id or ""
@staticmethod
def convert_response_function_tool_call_to_chat_completion_tool_call(
tool_call_item: object,
index: int = 0,
) -> dict[str, object]:
"""
Convert ResponseFunctionToolCall to ChatCompletionToolCallChunk format.
Args:
tool_call_item: ResponseFunctionToolCall object or similar with name, arguments, call_id
index: The index of this tool call
Returns:
Dictionary in ChatCompletionToolCallChunk format
"""
item: Final = cast( # cast-ok: duck-typed tool call item, .get access guarded by hasattr below
_ResponsesToolCallItem, tool_call_item
)
# Extract provider_specific_fields if present
provider_specific_fields: object = getattr(tool_call_item, "provider_specific_fields", None)
if provider_specific_fields and not isinstance(provider_specific_fields, dict):
provider_specific_fields = _attribute_fields(provider_specific_fields)
elif hasattr(tool_call_item, "get") and callable(item.get):
provider_fields: Final = item.get("provider_specific_fields")
if provider_fields:
provider_specific_fields = (
cast("dict[str, object]", provider_fields) # cast-ok: passed through as-is, keys unvalidated
if isinstance(provider_fields, dict)
else _attribute_fields(provider_fields)
)
function_dict: Final[dict[str, object]] = {
"name": item.name,
"arguments": item.arguments,
}
if provider_specific_fields:
function_dict["provider_specific_fields"] = provider_specific_fields
tool_call_dict: Final[dict[str, object]] = {
"id": LiteLLMCompletionResponsesConfig._tool_call_id_from_responses_item(
getattr(tool_call_item, "id", None),
getattr(tool_call_item, "call_id", None),
),
"function": function_dict,
"type": "function",
"index": 0,
}
if provider_specific_fields:
tool_call_dict["provider_specific_fields"] = provider_specific_fields
return tool_call_dict
@staticmethod
def convert_apply_patch_tool_call_to_chat_completion_tool_call(
tool_call_item: "ResponseApplyPatchToolCall",
index: int = 0,
) -> dict[str, object]:
"""
Convert ResponseApplyPatchToolCall to ChatCompletionToolCallChunk format.
The operation (create_file / update_file / delete_file) is serialised
as JSON so it appears in function.arguments, just like any other
tool call.
Args:
tool_call_item: ResponseApplyPatchToolCall object with call_id and operation
index: The index of this tool call
Returns:
Dictionary in ChatCompletionToolCallChunk format
"""
import json
operation_dict: Final = tool_call_item.operation.model_dump()
tool_call_dict: Final[dict[str, object]] = {
"id": tool_call_item.call_id,
"function": {
"name": "apply_patch",
"arguments": json.dumps(operation_dict),
},
"type": "function",
"index": index,
}
return tool_call_dict
@staticmethod
def transform_chat_completion_response_to_responses_api_response(
request_input: str | ResponseInputParam,
responses_api_request: ResponsesAPIOptionalRequestParams,
chat_completion_response: ModelResponse | dict,
) -> ResponsesAPIResponse:
"""
Transform a Chat Completion response into a Responses API response
"""
if isinstance(chat_completion_response, dict):
chat_completion_response = ModelResponse(**chat_completion_response)
# Get finish_reason from the first choice to determine overall status
finish_reason: str | None = None
choices: Final[list[Choices]] = getattr(chat_completion_response, "choices", [])
if choices and len(choices) > 0:
finish_reason = choices[0].finish_reason
responses_api_response: Final[ResponsesAPIResponse] = ResponsesAPIResponse(
id=chat_completion_response.id,
created_at=chat_completion_response.created,
model=chat_completion_response.model,
object="response",
error=getattr(chat_completion_response, "error", None),
incomplete_details=getattr(chat_completion_response, "incomplete_details", None),
instructions=getattr(chat_completion_response, "instructions", None),
metadata=getattr(chat_completion_response, "metadata", {}),
output=LiteLLMCompletionResponsesConfig._transform_chat_completion_choices_to_responses_output(
chat_completion_response=chat_completion_response,
choices=getattr(chat_completion_response, "choices", []),
responses_api_request=responses_api_request,
),
parallel_tool_calls=getattr(chat_completion_response, "parallel_tool_calls", False),
temperature=getattr(chat_completion_response, "temperature", 0),
tool_choice=getattr(chat_completion_response, "tool_choice", "auto"),
tools=getattr(chat_completion_response, "tools", []),
top_p=getattr(chat_completion_response, "top_p", None),
max_output_tokens=getattr(chat_completion_response, "max_output_tokens", None),
previous_response_id=getattr(chat_completion_response, "previous_response_id", None),
reasoning=None,
status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
finish_reason
),
text={},
truncation=getattr(chat_completion_response, "truncation", None),
usage=LiteLLMCompletionResponsesConfig._transform_chat_completion_usage_to_responses_usage(
chat_completion_response=chat_completion_response
),
user=getattr(chat_completion_response, "user", None),
)
responses_api_response._hidden_params = getattr(chat_completion_response, "_hidden_params", {})
# Surface provider-specific fields (generic passthrough from any provider)
provider_fields: Final = responses_api_response._hidden_params.get("provider_specific_fields")
if provider_fields:
setattr(responses_api_response, "provider_specific_fields", provider_fields)
return responses_api_response
@staticmethod
def _transform_chat_completion_choices_to_responses_output(
chat_completion_response: ModelResponse,
choices: list[Choices],
responses_api_request: ResponsesAPIOptionalRequestParams | None = None,
) -> list[
GenericResponseOutputItem
| OutputCodeInterpreterCall
| OutputFunctionToolCall
| OutputImageGenerationCall
| ResponseFunctionToolCall
| CustomToolCallOutputItem
]:
responses_output: list[
GenericResponseOutputItem
| OutputCodeInterpreterCall
| OutputFunctionToolCall
| OutputImageGenerationCall
| ResponseFunctionToolCall
| CustomToolCallOutputItem
] = []
responses_output.extend(
LiteLLMCompletionResponsesConfig._extract_reasoning_output_items(chat_completion_response, choices)
)
responses_output.extend(
LiteLLMCompletionResponsesConfig._extract_message_output_items(chat_completion_response, choices)
)
responses_output.extend(
LiteLLMCompletionResponsesConfig.transform_chat_completion_tools_to_responses_tools(
chat_completion_response=chat_completion_response,
responses_api_request=responses_api_request,
)
)
# Convert server-side tool results (e.g. Anthropic code execution)
# into code_interpreter_call output items, replacing the corresponding
# function_call items so the output matches OpenAI's native shape.
tool_result_items = LiteLLMCompletionResponsesConfig._extract_tool_result_output_items(chat_completion_response)
if tool_result_items:
result_by_id: Final = {item.id: item for item in tool_result_items}
replaced_ids: Final = set(result_by_id.keys())
responses_output = [
(
result_by_id[getattr(item, "call_id", None)]
if (
getattr(item, "type", None) == "function_call"
and getattr(item, "call_id", None) in replaced_ids
)
else item
)
for item in responses_output
]
return responses_output
@staticmethod
def _extract_tool_result_output_items(
chat_completion_response: ModelResponse,
) -> list:
"""Extract pre-built code_interpreter_call output items from provider_specific_fields.
Provider transformers (e.g. Anthropic) convert their native tool results
into OutputCodeInterpreterCall objects and store them in
provider_specific_fields["code_interpreter_results"]. This method
simply retrieves them — no provider-specific parsing here.
"""
output_items: Final[list] = []
for choice in chat_completion_response.choices or []:
message: object = getattr(choice, "message", None)
if not message:
continue
psf = getattr(message, "provider_specific_fields", None)
if not psf or not isinstance(psf, dict):
continue
results = psf.get("code_interpreter_results")
if results and isinstance(results, list):
for item in results:
# In the streaming path, items are plain dicts after
# model_dump() in stream_chunk_builder. Reconstruct
# Pydantic objects so responses_output has a uniform type.
if isinstance(item, dict):
output_items.append(OutputCodeInterpreterCall(**item))
else:
output_items.append(item)
return output_items
@staticmethod
def _encode_thinking_blocks(message: Message) -> str | None:
thinking_blocks: Final[Sequence[Mapping[str, object]]] = getattr(message, "thinking_blocks", None) or ()
preserved: Final = tuple(block for block in thinking_blocks if block.get("signature") or block.get("data"))
return json.dumps(preserved, separators=(",", ":")) if preserved else None
@staticmethod
def _extract_reasoning_output_items(
chat_completion_response: ModelResponse,
choices: list[Choices],
) -> list[GenericResponseOutputItem]:
for choice in choices:
if hasattr(choice, "message") and choice.message:
message = choice.message
reasoning_content: str = getattr(message, "reasoning_content", None) or ""
encrypted_content = LiteLLMCompletionResponsesConfig._encode_thinking_blocks(message)
if reasoning_content or encrypted_content:
# Only check the first choice for reasoning content
return [
GenericResponseOutputItem(
type="reasoning",
id=f"rs_{uuid.uuid4()}",
status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
choice.finish_reason
),
role="assistant",
content=[
OutputText(
type="output_text",
text=text,
annotations=[],
)
for text in (reasoning_content,)
if text
],
encrypted_content=encrypted_content,
)
]
return []
@staticmethod
def _extract_image_generation_output_items(
choice: Choices,
) -> list[OutputImageGenerationCall]:
"""
Extract image generation outputs from a choice that contains images.
Transforms message.images from chat completion format:
{
'image_url': {'url': 'data:image/png;base64,iVBORw0...'},
'type': 'image_url',
'index': 0
}
To Responses API format:
{
'type': 'image_generation_call',
'id': 'ig_...',
'status': 'completed',
'result': 'iVBORw0...' # Pure base64 without data: prefix
}
"""
image_generation_items: Final[list[OutputImageGenerationCall]] = []
images: Final = getattr(choice.message, "images", [])
if not images:
return image_generation_items
for image_item in _DICT_ITEMS_LIST_ADAPTER.validate_python(images):
# Extract base64 from data URL
image_url = _TEXT_ADAPTER.validate_python(
_ANY_KEY_DICT_ADAPTER.validate_python(image_item.get("image_url", {})).get("url", "")
)
base64_data = LiteLLMCompletionResponsesConfig._extract_base64_from_data_url(image_url)
if base64_data:
image_generation_items.append(
OutputImageGenerationCall(
type="image_generation_call",
id=f"ig_{uuid.uuid4()}",
status=LiteLLMCompletionResponsesConfig._map_finish_reason_to_image_generation_status(
choice.finish_reason
),
result=base64_data,
)
)
return image_generation_items
@staticmethod
def _map_finish_reason_to_image_generation_status(
finish_reason: str | None,
) -> Literal["in_progress", "completed", "incomplete", "failed"]:
"""
Map finish_reason to image generation status.
Image generation status only supports: in_progress, completed, incomplete, failed
(does not support: cancelled, queued like general ResponsesAPIStatus)
"""
if finish_reason == "stop":
return "completed"
elif finish_reason == "length":
return "incomplete"
elif finish_reason in ["content_filter", "error"]:
return "failed"
else:
# Default to completed for other cases
return "completed"
@staticmethod
def _extract_base64_from_data_url(data_url: str) -> str | None:
"""
Extract pure base64 string from a data URL.
Input: 'data:image/png;base64,iVBORw0KGgoAAAANS...'
Output: 'iVBORw0KGgoAAAANS...'
If input is already pure base64 (no prefix), return as-is.
"""
if not data_url:
return None
# Check if it's a data URL with prefix
if data_url.startswith("data:"):
# Split by comma to separate prefix from base64 data
parts: Final = data_url.split(",", 1)
if len(parts) == 2:
return parts[1] # Return the base64 part
return None
else:
# Already pure base64
return data_url
@staticmethod
def _extract_message_output_items(
chat_completion_response: ModelResponse,
choices: list[Choices],
) -> list[GenericResponseOutputItem | OutputImageGenerationCall]:
message_output_items: Final[list[GenericResponseOutputItem | OutputImageGenerationCall]] = []
for choice in choices:
# Check if message has images (image generation)
if hasattr(choice.message, "images") and choice.message.images:
# Extract image generation output
image_generation_items = LiteLLMCompletionResponsesConfig._extract_image_generation_output_items(
choice=choice,
)
message_output_items.extend(image_generation_items)
else:
# Regular message output
message_output_items.append(
GenericResponseOutputItem(
type="message",
id=f"msg_{uuid.uuid4()}",
status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
choice.finish_reason
),
role=choice.message.role,
content=[
LiteLLMCompletionResponsesConfig._transform_chat_message_to_response_output_text(
choice.message
)
],
)
)
return message_output_items
@staticmethod
def _transform_responses_api_outputs_to_chat_completion_messages(
responses_api_output: ResponsesAPIResponse,
) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionMessageToolCall]:
messages: Final[list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionMessageToolCall]] = []
output_items: Final = responses_api_output.output
for _output_item in output_items:
output_item: dict = dict(_output_item)
if output_item.get("type") == "function_call":
# handle function call output
messages.append(
LiteLLMCompletionResponsesConfig._transform_responses_output_tool_call_to_chat_completion_output_tool_call(
tool_call=output_item
)
)
else:
# transform as generic ResponseOutputItem
content = output_item.get("content")
# Skip if content is None (GenericChatCompletionMessage requires content)
if content is not None:
messages.append(
GenericChatCompletionMessage(
role=str(output_item.get("role")) or "user",
content=LiteLLMCompletionResponsesConfig._transform_responses_api_content_to_chat_completion_content(
content
),
)
)
return messages
@staticmethod
def _transform_responses_output_tool_call_to_chat_completion_output_tool_call(
tool_call: dict,
) -> ChatCompletionMessageToolCall:
return ChatCompletionMessageToolCall(
id=tool_call.get("id") or "",
type="function",
function=Function(
name=tool_call.get("name") or "",
arguments=tool_call.get("arguments") or "",
),
)
@staticmethod
def _transform_chat_message_to_response_output_text(
message: Message,
) -> OutputText:
annotations: Final = getattr(message, "annotations", None)
transformed_annotations: Final = (
LiteLLMCompletionResponsesConfig._transform_chat_completion_annotations_to_response_output_annotations(
annotations=annotations
)
)
return OutputText(
type="output_text",
text=message.content,
annotations=transformed_annotations,
)
@staticmethod
def _transform_chat_completion_annotations_to_response_output_annotations(
annotations: list[ChatCompletionAnnotation] | None,
) -> list[GenericResponseOutputItemContentAnnotation]:
response_output_annotations: Final[list[GenericResponseOutputItemContentAnnotation]] = []
if annotations is None:
return response_output_annotations
for annotation in annotations:
annotation_type = annotation.get("type")
if annotation_type == "url_citation" and "url_citation" in annotation:
url_citation = annotation["url_citation"]
response_output_annotations.append(
GenericResponseOutputItemContentAnnotation(
type=annotation_type,
start_index=url_citation.get("start_index"),
end_index=url_citation.get("end_index"),
url=url_citation.get("url"),
title=url_citation.get("title"),
)
)
# Handle other annotation types here
return response_output_annotations
@staticmethod
def _transform_chat_completion_usage_to_responses_usage(
chat_completion_response: ModelResponse | Usage,
) -> ResponseAPIUsage:
if isinstance(chat_completion_response, ModelResponse):
usage: Usage | None = getattr(chat_completion_response, "usage", None)
else:
usage = chat_completion_response
if usage is None:
return ResponseAPIUsage(
input_tokens=0,
output_tokens=0,
total_tokens=0,
)
response_usage: Final = ResponseAPIUsage(
input_tokens=usage.prompt_tokens,
output_tokens=usage.completion_tokens,
total_tokens=usage.total_tokens,
)
# Preserve cost field if it exists (for streaming usage with cost calculation)
if hasattr(usage, "cost") and usage.cost is not None:
setattr(response_usage, "cost", usage.cost)
# Translate prompt_tokens_details to input_tokens_details
if hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details is not None:
prompt_details: Final = usage.prompt_tokens_details
input_details_dict: Final[dict[str, int]] = {}
if hasattr(prompt_details, "cached_tokens") and prompt_details.cached_tokens is not None:
input_details_dict["cached_tokens"] = prompt_details.cached_tokens
else:
input_details_dict["cached_tokens"] = 0
if hasattr(prompt_details, "text_tokens") and prompt_details.text_tokens is not None:
input_details_dict["text_tokens"] = prompt_details.text_tokens
if hasattr(prompt_details, "audio_tokens") and prompt_details.audio_tokens is not None:
input_details_dict["audio_tokens"] = prompt_details.audio_tokens
cache_write_tokens = getattr(prompt_details, "cache_write_tokens", None) or getattr(
prompt_details, "cache_creation_tokens", None
)
if cache_write_tokens is not None:
input_details_dict["cache_write_tokens"] = cache_write_tokens
if input_details_dict:
response_usage.input_tokens_details = InputTokensDetails(**input_details_dict)
# Translate completion_tokens_details to output_tokens_details
if hasattr(usage, "completion_tokens_details") and usage.completion_tokens_details is not None:
completion_details: Final = usage.completion_tokens_details
reasoning_token_count: Final = getattr(completion_details, "reasoning_tokens", None)
optional_output_details: Final[dict[str, int]] = {
field: value
for field, value in (
("audio_tokens", getattr(completion_details, "audio_tokens", None)),
("text_tokens", getattr(completion_details, "text_tokens", None)),
("image_tokens", getattr(completion_details, "image_tokens", None)),
)
if value is not None
}
response_usage.output_tokens_details = OutputTokensDetails(
reasoning_tokens=reasoning_token_count if reasoning_token_count is not None else 0,
**optional_output_details,
)
return response_usage
@staticmethod
def _transform_text_format_to_response_format(
text_param: object,
) -> dict[str, object] | None:
"""
Transform Responses API text.format parameter to Chat Completion response_format parameter.
Responses API text parameter structure:
{
"format": {
"type": "json_schema",
"name": "schema_name",
"schema": {...},
"strict": True
}
}
Chat Completion response_format structure:
{
"type": "json_schema",
"json_schema": {
"name": "schema_name",
"schema": {...},
"strict": True
}
}
"""
if not text_param:
return None
if isinstance(text_param, dict):
format_param: Final = text_param.get("format")
if format_param and isinstance(format_param, dict):
format_type: Final = format_param.get("type")
if format_type == "json_schema":
return {
"type": "json_schema",
"json_schema": {
"name": format_param.get("name", "response_schema"),
"schema": format_param.get("schema", {}),
"strict": format_param.get("strict", False),
},
}
elif format_type == "json_object":
return {"type": "json_object"}
elif format_type == "text":
return None
return None