scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-09 22:31:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 86

fix(responses-bridge): custom tool round-trip and allowlist preservation for Codex CLI (#32258)

Browse source 
* fix(responses-bridge): custom tool round-trip and allowlist preservation for Codex CLI

Convert Responses API custom tools to Chat Completions function tools and map
function_call responses back to custom_tool_call output items so Codex CLI gets
the apply_patch round-trip it expects. Preserve and validate allowed_callers
during the custom->function conversion so the Anthropic adapter's caller
allowlist is not silently dropped, which would let a tool meant to be callable
only by another tool be invoked directly by the model. Use modern type
annotations (list/dict/set/X | None) throughout to keep the ruff strict budget
within its ratcheted ceilings.

* fix(responses-bridge): address review feedback on custom tool bridge

Type convert_custom_tool_to_function_tool against Mapping/ChatCompletionToolParam
and validate allowed_callers with a strict TypeAdapter so the two new cast()
calls that tripped the LIT006 ceiling are gone. Warn when dropping Responses-only
tool types (computer_use, image_generation, namespace, shell) instead of
discarding them silently. Return output items as Pydantic models instead of
model_dump()ing every item to a dict, matching the declared return type. Apply
the same None-safe metadata pattern to the request_data paths that still used
setdefault, and drop the unused build_custom_tool_call_item helper.

* fix(responses-bridge): recover custom tool input when arguments is empty

* fix(auth): extract custom tool names for allowlist enforcement on responses route

The Responses guardrail translation handler only extracted function and mcp
tool names, so a key or team restricted by metadata.allowed_tools could invoke
a disallowed tool by declaring it with type custom now that the bridge converts
custom tools into callable Chat Completions function tools. Extract custom tool
names through the same path so check_tools_allowlist rejects them.

* fix(responses-bridge): scope input payload recovery to custom_tool_call items

Recovering tool arguments from the input field on any falsy arguments value
made plain function_call input items with empty arguments and a stray input
key get rewritten into a {"content": ...} envelope, corrupting multi-turn
replay for normal function tools. Gate the recovery on the item type so it
only applies to custom_tool_call items, which are the ones that store their
payload in input.

* fix(responses-bridge): default missing function_call arguments to empty string

With input recovery scoped to custom_tool_call items, a plain function_call
input item without an arguments key left raw_arguments as None and the
downstream str() turned it into the literal string None. Coerce to an empty
string instead, matching the pre-bridge behavior.

---------

Co-authored-by: duanhongyi <duanhongyi@doopai.com>
This commit is contained in:
Mateo Wang 2026-07-06 17:34:27 -07:00 • committed by GitHub
parent 0855fa02b2
commit 2f0cdb35bf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 1267 additions and 346 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
litellm/llms/openai/responses/guardrail_translation/handler.py
View file

@ -197,12 +197,13 @@ class OpenAIResponsesHandler(BaseTranslation):
return data
def extract_request_tool_names(self, data: dict) -> List[str]:
"""Extract tool names from Responses API request (tools[].name for function, tools[].server_label for mcp)."""
"""Extract tool names from Responses API request (tools[].name for function
and custom, tools[].server_label for mcp)."""
names: List[str] = []
for tool in data.get("tools") or []:
if not isinstance(tool, dict):
continue
if tool.get("type") == "function" and tool.get("name"):
if tool.get("type") in ("function", "custom") and tool.get("name"):
names.append(str(tool["name"]))
elif tool.get("type") == "mcp" and tool.get("server_label"):
names.append(str(tool["server_label"]))

35
litellm/proxy/common_request_processing.py
View file

@ -91,7 +91,9 @@ _CLIENT_DISCONNECTED_ERROR_INFORMATION: StandardLoggingPayloadErrorInformation =
}
def _apply_client_disconnect_metadata(target_metadata: dict[str, object]) -> None:
def _apply_client_disconnect_metadata(target_metadata: Optional[dict[str, object]]) -> None:
if target_metadata is None:
return
target_metadata["client_disconnected"] = True
target_metadata["error_information"] = dict(_CLIENT_DISCONNECTED_ERROR_INFORMATION)
@ -114,12 +116,33 @@ async def _record_streaming_client_disconnect_if_needed(
logging_obj = request_data.get("litellm_logging_obj")
if logging_obj is not None:
litellm_params = logging_obj.model_call_details.setdefault("litellm_params", {})
_apply_client_disconnect_metadata(litellm_params.setdefault("metadata", {}))
_apply_client_disconnect_metadata(logging_obj.model_call_details.setdefault("metadata", {}))
_lp_metadata = litellm_params.get("metadata")
if _lp_metadata is None:
_lp_metadata = {}
litellm_params["metadata"] = _lp_metadata
_apply_client_disconnect_metadata(_lp_metadata)
_apply_client_disconnect_metadata(request_data.setdefault("metadata", {}))
litellm_params = request_data.setdefault("litellm_params", {})
_apply_client_disconnect_metadata(litellm_params.setdefault("metadata", {}))
_mcd_metadata = logging_obj.model_call_details.get("metadata")
if _mcd_metadata is None:
_mcd_metadata = {}
logging_obj.model_call_details["metadata"] = _mcd_metadata
_apply_client_disconnect_metadata(_mcd_metadata)
_rd_metadata = request_data.get("metadata")
if _rd_metadata is None:
_rd_metadata = {}
request_data["metadata"] = _rd_metadata
_apply_client_disconnect_metadata(_rd_metadata)
_rd_litellm_params = request_data.get("litellm_params")
if _rd_litellm_params is None:
_rd_litellm_params = {}
request_data["litellm_params"] = _rd_litellm_params
_rd_lp_metadata = _rd_litellm_params.get("metadata")
if _rd_lp_metadata is None:
_rd_lp_metadata = {}
_rd_litellm_params["metadata"] = _rd_lp_metadata
_apply_client_disconnect_metadata(_rd_lp_metadata)
verbose_proxy_logger.debug(
"Recorded streaming client disconnect with error_code=499 for litellm_call_id=%s",

163
litellm/responses/litellm_completion_transformation/custom_tools.py Normal file
View file

@ -0,0 +1,163 @@
"""
Utilities for handling OpenAI Responses API 'custom' tools (freeform/grammar tools)
when bridging to Chat Completions providers.
Custom tools are defined with ``type: "custom"`` and a grammar/format specification.
Since most Chat Completions providers only support standard ``function`` tools,
the bridge converts them to ``function`` tools with a single ``content`` string
parameter. When the model responds with a ``function_call`` for such a tool, this
module converts it back to the ``custom_tool_call`` format expected by clients like
Codex CLI.
The forward direction (custom -> function) and reverse direction (function_call ->
custom_tool_call) are both handled here so future custom tool types can be added by
extending this module without touching the streaming iterator or transformation
logic.
"""
import json
from collections.abc import Mapping
from typing import Any
from pydantic import BaseModel, TypeAdapter, ValidationError
from litellm.types.llms.openai import (
ChatCompletionToolParam,
ChatCompletionToolParamFunctionChunk,
)
_MAX_ARGUMENTS_LEN = 1_000_000
def extract_custom_tool_names(tools: list[Any] | None) -> set[str]:
"""Extract names of tools originally defined as ``type: "custom"``."""
if not tools:
return set()
names: set[str] = set()
for tool in tools:
if isinstance(tool, dict) and tool.get("type") == "custom" and "name" in tool:
names.add(tool["name"])
return names
def is_custom_tool_call(tool_name: str, custom_tool_names: set[str]) -> bool:
"""Check if a tool call name corresponds to a custom tool."""
return tool_name in custom_tool_names
def unwrap_custom_tool_arguments(arguments: str) -> str:
"""Extract the raw content string from JSON-wrapped arguments.
The bridge converts custom tools to function tools with schema
``{"properties": {"content": {"type": "string"}}}``, so the model returns
arguments like ``{"content": "*** Begin Patch\\n..."}``. This function
extracts just the content string. If the arguments are not valid JSON or do
not contain a ``content`` key, the original string is returned unchanged.
"""
if not arguments:
return ""
if len(arguments) > _MAX_ARGUMENTS_LEN:
return arguments
try:
parsed = json.loads(arguments)
if isinstance(parsed, dict) and "content" in parsed:
return str(parsed["content"])
except (json.JSONDecodeError, TypeError, ValueError):
pass
return arguments
def build_tool_call_item_kwargs(
call_id: str,
name: str,
arguments_or_input: str,
status: str,
custom_tool_names: set[str],
) -> dict[str, Any]:
"""Build kwargs for an output item dict that is either a ``function_call``
or a ``custom_tool_call`` depending on whether *name* is in
*custom_tool_names*.
For custom tools the ``arguments`` JSON is unwrapped into the ``input``
field. For regular function tools the raw ``arguments`` string is kept.
This centralises the branching logic so the streaming iterator and the
non-streaming transformation share a single code path.
"""
custom = is_custom_tool_call(name, custom_tool_names)
item_type = "custom_tool_call" if custom else "function_call"
kwargs: dict[str, Any] = {
"type": item_type,
"id": call_id,
"call_id": call_id,
"name": name,
"status": status,
}
if custom:
if status == "completed":
kwargs["input"] = unwrap_custom_tool_arguments(arguments_or_input)
else:
kwargs["input"] = ""
else:
kwargs["arguments"] = arguments_or_input
return kwargs
class _CustomToolFormat(BaseModel):
syntax: str = ""
definition: str = ""
_ALLOWED_CALLERS_ADAPTER = TypeAdapter(list[str] | None)
def _validated_allowed_callers(value: object) -> list[str] | None:
try:
return _ALLOWED_CALLERS_ADAPTER.validate_python(value, strict=True)
except ValidationError as exc:
raise ValueError("allowed_callers must be a list of strings") from exc
def _grammar_suffix(fmt: object) -> str:
try:
parsed = _CustomToolFormat.model_validate(fmt)
except ValidationError:
return ""
if not parsed.definition:
return ""
return f"\n\nFormat:\n```{parsed.syntax}\n{parsed.definition}\n```"
def convert_custom_tool_to_function_tool(tool: Mapping[str, object]) -> ChatCompletionToolParam | None:
"""Convert a Responses API ``custom`` tool to a Chat Completions ``function``
tool.
The grammar definition is embedded in the description so the model can
produce correctly-formatted output. Returns ``None`` if the tool is not a
custom tool. Raises ``ValueError`` if ``allowed_callers`` is not a list of
strings.
"""
if tool.get("type") != "custom":
return None
raw_name = tool.get("name")
name = raw_name if isinstance(raw_name, str) else ""
raw_description = tool.get("description")
description = (raw_description if isinstance(raw_description, str) else "") + _grammar_suffix(tool.get("format"))
allowed_callers = _validated_allowed_callers(tool.get("allowed_callers"))
function_chunk = ChatCompletionToolParamFunctionChunk(
name=name,
description=description,
parameters={
"type": "object",
"properties": {
"content": {
"type": "string",
"description": f"The {name} content following the specified format",
}
},
"required": ["content"],
},
)
if allowed_callers is None:
return ChatCompletionToolParam(type="function", function=function_chunk)
return ChatCompletionToolParam(type="function", function=function_chunk, allowed_callers=allowed_callers)

112
litellm/responses/litellm_completion_transformation/streaming_iterator.py
View file

@ -1,9 +1,13 @@
import time
import uuid
from typing import Any, Dict, List, Optional, Union, cast
from typing import Any, cast
import litellm
from litellm.main import stream_chunk_builder
from litellm.responses.litellm_completion_transformation.custom_tools import (
build_tool_call_item_kwargs,
extract_custom_tool_names,
)
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
)
@ -53,20 +57,20 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self,
model: str,
litellm_custom_stream_wrapper: litellm.CustomStreamWrapper,
request_input: Union[str, ResponseInputParam],
request_input: str | ResponseInputParam,
responses_api_request: ResponsesAPIOptionalRequestParams,
custom_llm_provider: Optional[str] = None,
litellm_metadata: Optional[dict] = None,
custom_llm_provider: str | None = None,
litellm_metadata: dict | None = None,
):
self.model: str = model
self.litellm_custom_stream_wrapper: litellm.CustomStreamWrapper = litellm_custom_stream_wrapper
self.request_input: Union[str, ResponseInputParam] = request_input
self.request_input: str | ResponseInputParam = request_input
self.responses_api_request: ResponsesAPIOptionalRequestParams = responses_api_request
self.custom_llm_provider: Optional[str] = custom_llm_provider
self.litellm_metadata: Optional[dict] = litellm_metadata or {}
self.custom_llm_provider: str | None = custom_llm_provider
self.litellm_metadata: dict | None = litellm_metadata or {}
# Store lightweight dict snapshots for stream_chunk_builder to reduce
# repeated Pydantic attribute access in end-of-stream assembly.
self.collected_chat_completion_chunks: List[Dict[str, Any]] = []
self.collected_chat_completion_chunks: list[dict[str, Any]] = []
self.finished: bool = False
self.litellm_logging_obj = litellm_custom_stream_wrapper.logging_obj
self.sent_response_created_event: bool = False
@ -77,11 +81,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self.sent_output_content_part_done_event: bool = False
self.sent_output_item_done_event: bool = False
self.sent_annotation_events: bool = False
self.litellm_model_response: Optional[Union[ModelResponse, TextCompletionResponse]] = None
self.litellm_model_response: ModelResponse | TextCompletionResponse | None = None
self.final_text: str = ""
self._cached_item_id: Optional[str] = None
self._cached_response_id: Optional[str] = None
self._pending_tool_events: List[BaseLiteLLMOpenAIResponseObject] = []
self._cached_item_id: str | None = None
self._cached_response_id: str | None = None
self._pending_tool_events: list[BaseLiteLLMOpenAIResponseObject] = []
self._tool_output_index_by_call_id: dict[str, int] = {}
self._tool_args_by_call_id: dict[str, str] = {}
self._tool_call_id_by_index: dict[int, str] = {}
@ -89,17 +93,18 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item
self._final_tool_events_queued: bool = False
self._sequence_number: int = 0
self._cached_reasoning_item_id: Optional[str] = None
self._cached_reasoning_item_id: str | None = None
self._sent_reasoning_summary_text_done_event: bool = False
self._sent_reasoning_summary_part_done_event: bool = False
self._reasoning_summary_text: str = ""
# -- GENERIC RESPONSE-EVENTS PENDING QUEUE as required by fix --
self._pending_response_events: List[BaseLiteLLMOpenAIResponseObject] = []
self._pending_response_events: list[BaseLiteLLMOpenAIResponseObject] = []
self._reasoning_active = False
self._reasoning_done_emitted = False
self._reasoning_item_id: Optional[str] = None
self._accumulated_reasoning_content_parts: List[str] = []
self._accumulated_provider_specific_fields: Dict[str, Any] = {}
self._reasoning_item_id: str | None = None
self._accumulated_reasoning_content_parts: list[str] = []
self._accumulated_provider_specific_fields: dict[str, Any] = {}
self._custom_tool_names: set[str] = extract_custom_tool_names(self.responses_api_request.get("tools"))
def _get_or_assign_tool_output_index(self, call_id: str) -> int:
existing = self._tool_output_index_by_call_id.get(call_id)
@ -110,7 +115,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self._tool_output_index_by_call_id[call_id] = idx
return idx
def _normalize_tool_call_index(self, tool_call: object) -> Optional[int]:
def _normalize_tool_call_index(self, tool_call: object) -> int | None:
idx_raw = tool_call.get("index") if isinstance(tool_call, dict) else getattr(tool_call, "index", None)
if idx_raw is None:
return None
@ -183,19 +188,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
if call_id not in self._tool_args_by_call_id:
self._tool_args_by_call_id[call_id] = ""
self._sequence_number += 1
item_kwargs = build_tool_call_item_kwargs(call_id, fn_name, "", "in_progress", self._custom_tool_names)
event = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=output_index,
item=BaseLiteLLMOpenAIResponseObject(
**{
"type": "function_call",
"id": call_id,
"call_id": call_id,
"name": fn_name,
"arguments": "",
"status": "in_progress",
}
),
item=BaseLiteLLMOpenAIResponseObject(**item_kwargs),
)
event.__dict__["sequence_number"] = self._sequence_number
self._pending_tool_events.append(event)
@ -260,19 +257,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
if is_new_tool_call:
self._tool_args_by_call_id[call_id] = ""
self._sequence_number += 1
item_kwargs = build_tool_call_item_kwargs(call_id, fn_name, "", "in_progress", self._custom_tool_names)
event = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=output_index,
item=BaseLiteLLMOpenAIResponseObject(
**{
"type": "function_call",
"id": call_id,
"call_id": call_id,
"name": fn_name,
"arguments": "",
"status": "in_progress",
}
),
item=BaseLiteLLMOpenAIResponseObject(**item_kwargs),
)
event.__dict__["sequence_number"] = self._sequence_number
self._pending_tool_events.append(event)
@ -310,20 +299,14 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self._pending_tool_events.append(done_event)
self._sequence_number += 1
item_kwargs = build_tool_call_item_kwargs(
call_id, fn_name, final_args, "completed", self._custom_tool_names
)
item_done_event = OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=output_index,
sequence_number=self._sequence_number,
item=BaseLiteLLMOpenAIResponseObject(
**{
"type": "function_call",
"id": call_id,
"call_id": call_id,
"name": fn_name,
"arguments": final_args,
"status": "completed",
}
),
item=BaseLiteLLMOpenAIResponseObject(**item_kwargs),
)
self._pending_tool_events.append(item_done_event)
@ -449,9 +432,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
for key, val in src.items():
self._accumulated_provider_specific_fields[key] = val
def create_litellm_model_response(self) -> Optional[ModelResponse]:
def create_litellm_model_response(self) -> ModelResponse | None:
response = cast(
Optional[ModelResponse],
ModelResponse | None,
stream_chunk_builder(
chunks=self.collected_chat_completion_chunks,
logging_obj=self.litellm_logging_obj,
@ -468,7 +451,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
@staticmethod
def _snapshot_chunk_for_stream_chunk_builder(
chunk: ModelResponseStream,
) -> Dict[str, Any]:
) -> dict[str, Any]:
"""
Convert a streaming chunk into a plain dict for end-of-stream assembly.
Keep _hidden_params so downstream usage/header behavior is preserved.
@ -564,7 +547,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
reasoning_content = getattr(litellm_complete_object.choices[0].message, "reasoning_content", "") or "" # type: ignore
annotations = getattr(litellm_complete_object.choices[0].message, "annotations", None) # type: ignore
part: Optional[PART_UNION_TYPES] = None
part: PART_UNION_TYPES | None = None
if reasoning_content:
part = ContentPartDonePartReasoningText(
type="reasoning_text",
@ -671,7 +654,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def return_default_done_events(
self, litellm_complete_object: ModelResponse
) -> Optional[BaseLiteLLMOpenAIResponseObject]:
) -> BaseLiteLLMOpenAIResponseObject | None:
if self.sent_output_text_done_event is False:
self.sent_output_text_done_event = True
return self.create_output_text_done_event(litellm_complete_object)
@ -685,7 +668,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def return_default_initial_events(
self,
) -> Optional[BaseLiteLLMOpenAIResponseObject]:
) -> BaseLiteLLMOpenAIResponseObject | None:
if self.sent_response_created_event is False:
self.sent_response_created_event = True
return self.create_response_created_event()
@ -725,6 +708,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self.finished = self.is_stream_finished()
response_completed_event = self._emit_response_completed_event(self.litellm_model_response)
if response_completed_event:
# Latch so wrappers (FallbackResponsesStreamWrapper) + proxy
# container-ownership hook can read completed_response.
self.completed_response = response_completed_event
return response_completed_event
else:
if sync_mode:
@ -800,11 +786,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
async def __anext__(
self,
) -> Union[
ResponsesAPIStreamingResponse,
ResponseCompletedEvent,
BaseLiteLLMOpenAIResponseObject,
]:
) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject:
try:
while True:
if self.finished is True:
@ -906,11 +888,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def __next__(
self,
) -> Union[
ResponsesAPIStreamingResponse,
ResponseCompletedEvent,
BaseLiteLLMOpenAIResponseObject,
]:
) -> ResponsesAPIStreamingResponse | ResponseCompletedEvent | BaseLiteLLMOpenAIResponseObject:
try:
while True:
if self.finished is True:
@ -961,7 +939,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def _transform_chat_completion_chunk_to_response_api_chunk(
self, chunk: ModelResponseStream
) -> Optional[ResponsesAPIStreamingResponse]:
) -> ResponsesAPIStreamingResponse | None:
"""
Transform a chat completion chunk to a response API chunk.
@ -1047,7 +1025,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return None
def _get_delta_string_from_streaming_choices(self, choices: List[StreamingChoices]) -> str:
def _get_delta_string_from_streaming_choices(self, choices: list[StreamingChoices]) -> str:
"""
Get the delta string from the streaming choices
@ -1059,7 +1037,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
chat_completion_delta: ChatCompletionDelta = choice.delta
return chat_completion_delta.content or ""
def _emit_response_completed_event(self, litellm_model_response: ModelResponse) -> Optional[ResponseCompletedEvent]:
def _emit_response_completed_event(self, litellm_model_response: ModelResponse) -> ResponseCompletedEvent | None:
if litellm_model_response:
# Add cost to usage object if include_cost_in_streaming_usage is True
if litellm.include_cost_in_streaming_usage and self.litellm_logging_obj is not None:

539
litellm/responses/litellm_completion_transformation/transformation.py
View file

File diff suppressed because it is too large Load diff

15
litellm/router.py
View file

@ -2361,7 +2361,20 @@ class Router:
return self
async def __anext__(self):
chunk = await self._async_generator.__anext__()
try:
chunk = await self._async_generator.__anext__()
except StopAsyncIteration:
# The inner generator is exhausted. If we never sniffed a
# terminal event off a chunk (the bridge path emits the
# final response.completed via common_done_event_logic,
# which raises StopAsyncIteration after returning it),
# fall back to whatever the source iterator latched so
# the proxy's container-ownership hook still sees a
# completed_response instead of logging a spurious
# "no completed_response" warning.
if self.completed_response is None:
self.completed_response = getattr(source_iterator, "completed_response", None)
raise
# Sniff the terminal stream event off each forwarded chunk
# so ``self.completed_response`` is populated regardless of
# which inner iterator produced it (source_iterator,

3
litellm/types/llms/openai.py
View file

@ -90,6 +90,7 @@ from typing_extensions import (
from litellm.types.llms.base import BaseLiteLLMOpenAIResponseObject
from litellm.types.responses.main import (
CustomToolCallOutputItem,
GenericResponseOutputItem,
OutputCodeInterpreterCall,
OutputFunctionToolCall,
@ -914,6 +915,7 @@ class OpenAIChatCompletionToolParam(TypedDict):
class ChatCompletionToolParam(OpenAIChatCompletionToolParam, total=False):
cache_control: ChatCompletionCachedContent
allowed_callers: List[str]
class Function(TypedDict, total=False):
@ -1253,6 +1255,7 @@ class ResponsesAPIResponse(BaseLiteLLMOpenAIResponseObject):
OutputFunctionToolCall,
OutputImageGenerationCall,
ResponseFunctionToolCall,
CustomToolCallOutputItem,
]
],
]

16
litellm/types/responses/main.py
View file

@ -85,6 +85,22 @@ def build_code_interpreter_log_outputs(
return [OutputCodeInterpreterCallLog(type="logs", logs=logs)] if logs else None
class CustomToolCallOutputItem(BaseLiteLLMOpenAIResponseObject):
"""A custom/freeform tool call output item (e.g. apply_patch).
Mirrors the ``custom_tool_call`` variant of OpenAI's Responses API output.
Unlike ``OutputFunctionToolCall`` which uses ``arguments`` (JSON string),
this uses ``input`` (raw string) for the tool payload.
"""
type: Literal["custom_tool_call"]
call_id: str
id: Optional[str] = None
name: str
input: str
status: Optional[Literal["in_progress", "completed", "incomplete"]] = None
class GenericResponseOutputItem(BaseLiteLLMOpenAIResponseObject):
"""
Generic response API output item

69
tests/test_litellm/proxy/test_common_request_processing.py
View file

@ -3304,6 +3304,75 @@ class TestStreamingClientDisconnectLogging:
assert recorded is False
assert "client_disconnected" not in request_data["metadata"]
@pytest.mark.asyncio
async def test_record_streaming_client_disconnect_handles_none_metadata(self):
from litellm.proxy.common_request_processing import (
_record_streaming_client_disconnect_if_needed,
)
mock_logging_obj = MagicMock()
mock_logging_obj.model_call_details = {
"litellm_params": {"metadata": None},
"metadata": None,
}
mock_request = MagicMock(spec=Request)
mock_request.is_disconnected = AsyncMock(return_value=True)
request_data = {
"litellm_call_id": "test-call-id",
"litellm_logging_obj": mock_logging_obj,
"metadata": {},
"litellm_params": {"metadata": {}},
}
recorded = await _record_streaming_client_disconnect_if_needed(
mock_request, request_data
)
assert recorded is True
assert request_data["metadata"]["client_disconnected"] is True
assert (
mock_logging_obj.model_call_details["litellm_params"]["metadata"][
"client_disconnected"
]
is True
)
assert (
mock_logging_obj.model_call_details["metadata"]["client_disconnected"]
is True
)
@pytest.mark.asyncio
async def test_record_streaming_client_disconnect_handles_none_request_data_metadata(self):
from litellm.proxy.common_request_processing import (
_record_streaming_client_disconnect_if_needed,
)
mock_request = MagicMock(spec=Request)
mock_request.is_disconnected = AsyncMock(return_value=True)
request_data = {
"litellm_call_id": "test-call-id",
"metadata": None,
"litellm_params": {"metadata": None},
}
recorded = await _record_streaming_client_disconnect_if_needed(
mock_request, request_data
)
assert recorded is True
assert request_data["metadata"]["client_disconnected"] is True
assert (
request_data["litellm_params"]["metadata"]["client_disconnected"] is True
)
@pytest.mark.asyncio
async def test_apply_client_disconnect_metadata_none_returns_early(self):
from litellm.proxy.common_request_processing import (
_apply_client_disconnect_metadata,
)
_apply_client_disconnect_metadata(None)
@pytest.mark.asyncio
async def test_finalize_streaming_generator_cleanup_fires_deferred_logging(
self, monkeypatch

30
tests/test_litellm/proxy/test_tools_allowlist_enforcement.py
View file

@ -70,6 +70,22 @@ class TestExtractRequestToolNames:
}
assert extract_request_tool_names("/v1/responses", data) == ["dmcp"]
def test_openai_responses_custom_tools(self):
"""Custom tools become callable function tools on the Chat Completions
bridge, so their names must be extracted for allowlist enforcement;
otherwise a restricted key could invoke a disallowed tool by declaring
it with type "custom" (VERIA finding on PR #32258)."""
data = {
"tools": [
{"type": "custom", "name": "apply_patch", "description": "x"},
{"type": "function", "name": "get_current_weather"},
]
}
assert extract_request_tool_names("/v1/responses", data) == [
"apply_patch",
"get_current_weather",
]
def test_anthropic_tools(self):
data = {"tools": [{"name": "get_weather"}, {"name": "run_sql"}]}
assert extract_request_tool_names("/v1/messages", data) == [
@ -143,6 +159,20 @@ class TestCheckToolsAllowlist:
assert exc_info.value.type == ProxyErrorTypes.tool_access_denied
assert "get_weather" in str(exc_info.value.message)
@pytest.mark.asyncio
async def test_disallowed_custom_tool_raises_on_responses_route(self):
token = _token(metadata={"allowed_tools": ["other_tool"]})
body = {"tools": [{"type": "custom", "name": "restricted_tool"}]}
with pytest.raises(ProxyException) as exc_info:
await check_tools_allowlist(
request_body=body,
valid_token=token,
team_object=None,
route="/v1/responses",
)
assert exc_info.value.type == ProxyErrorTypes.tool_access_denied
assert "restricted_tool" in str(exc_info.value.message)
@pytest.mark.asyncio
async def test_team_allowlist_used_when_key_empty(self):
token = _token(

283
tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py
View file

@ -1,6 +1,8 @@
import os
import sys
import pytest
sys.path.insert(
0, os.path.abspath("../../..")
) # Adds the parent directory to the system path
@ -1066,7 +1068,13 @@ class TestToolTransformation:
assert web_search_options is None
def test_transform_computer_use_tools(self):
"""Test that computer_use tools are passed through as-is"""
"""Test that computer_use tools are dropped (no Chat Completions equivalent).
This deliberately reverses the previous pass-through regression guard:
forwarding computer_use verbatim made Chat Completions providers reject
the whole request with "'function' is a required property", so the
bridge now drops such tools (with a warning log) instead.
"""
computer_use_tool = {
"type": "computer_use",
"display_width_px": 1024,
@ -1083,11 +1091,129 @@ class TestToolTransformation:
tools=tools
)
# Assert - computer_use has no Chat Completions equivalent, so it is dropped
assert len(result_tools) == 0
assert web_search_options is None
def test_transform_custom_tools_to_function_tools(self):
"""Test that custom (freeform/grammar) tools are converted to function tools"""
custom_tool = {
"type": "custom",
"name": "apply_patch",
"description": "Apply a patch to files",
"format": {
"type": "grammar",
"syntax": "lark",
"definition": "start: begin_patch hunk+ end_patch",
},
}
tools = [custom_tool]
# Execute
(
result_tools,
web_search_options,
) = LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools(
tools=tools
)
# Assert - custom tool is converted to a function tool
assert len(result_tools) == 1
assert result_tools[0]["type"] == "function"
assert result_tools[0]["function"]["name"] == "apply_patch"
assert "content" in result_tools[0]["function"]["parameters"]["properties"]
assert result_tools[0]["function"]["parameters"]["required"] == ["content"]
assert "begin_patch" in result_tools[0]["function"]["description"]
assert web_search_options is None
def test_transform_custom_tools_without_format(self):
"""Test that custom tools without format info are still converted"""
custom_tool = {
"type": "custom",
"name": "exec",
"description": "Execute code",
}
tools = [custom_tool]
# Execute
(
result_tools,
web_search_options,
) = LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools(
tools=tools
)
# Assert
assert len(result_tools) == 1
assert result_tools[0] == computer_use_tool
assert result_tools[0]["type"] == "computer_use"
assert web_search_options is None
assert result_tools[0]["type"] == "function"
assert result_tools[0]["function"]["name"] == "exec"
assert result_tools[0]["function"]["description"] == "Execute code"
def test_transform_custom_tools_preserves_allowed_callers(self):
"""allowed_callers on a custom tool gates direct model invocation in the
Anthropic adapter, so it must survive the custom->function conversion."""
custom_tool = {
"type": "custom",
"name": "apply_patch",
"description": "Apply a patch to files",
"allowed_callers": ["exec"],
}
tools = [custom_tool]
# Execute
(
result_tools,
_,
) = LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools(
tools=tools
)
# Assert
assert len(result_tools) == 1
assert result_tools[0]["type"] == "function"
assert result_tools[0]["allowed_callers"] == ["exec"]
def test_transform_custom_tools_without_allowed_callers(self):
"""A custom tool without allowed_callers must not synthesize the field."""
custom_tool = {
"type": "custom",
"name": "apply_patch",
"description": "Apply a patch to files",
}
tools = [custom_tool]
# Execute
(
result_tools,
_,
) = LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools(
tools=tools
)
# Assert
assert len(result_tools) == 1
assert "allowed_callers" not in result_tools[0]
def test_transform_custom_tools_rejects_invalid_allowed_callers(self):
"""Invalid allowed_callers must raise rather than silently dropping the
provider-side allowlist."""
custom_tool = {
"type": "custom",
"name": "apply_patch",
"description": "Apply a patch to files",
"allowed_callers": "exec",
}
tools = [custom_tool]
with pytest.raises(ValueError, match="allowed_callers must be a list of strings"):
LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools(
tools=tools
)
def test_transform_web_search_tools_to_web_search_options(self):
"""Test that web_search tools are converted to web_search_options"""
@ -2185,6 +2311,155 @@ class TestStreamingIDConsistency:
assert tool_calls is not None and len(tool_calls) == 1
class TestCompletedResponseLatchedOnStreamEnd:
"""Regression: LiteLLMCompletionStreamingIterator (the Chat Completions
bridge path) never set ``self.completed_response`` because it overrides
__anext__ and bypasses the base class's _process_chunk where that
attribute is normally latched. FallbackResponsesStreamWrapper reads
``completed_response`` via getattr to record container ownership; when
it stays None the proxy logs a "Container ownership recording skipped"
warning and follow-up /v1/containers/<id>/files calls 403 for non-admin
keys. Codex CLI's apply_patch tool also surfaces as "aborted" because
the terminal response.completed event never propagates correctly."""
def _make_iterator_with_stop(self, model_response):
"""Build a LiteLLMCompletionStreamingIterator whose underlying
CustomStreamWrapper raises StopAsyncIteration immediately (simulating
a stream that already delivered all content chunks)."""
from unittest.mock import Mock
import litellm
from litellm.responses.litellm_completion_transformation.streaming_iterator import (
LiteLLMCompletionStreamingIterator,
)
mock_wrapper = Mock(spec=litellm.CustomStreamWrapper)
mock_wrapper.logging_obj = Mock()
mock_wrapper.logging_obj._response_cost_calculator = Mock(return_value=0.0)
mock_wrapper.__aiter__ = Mock(return_value=mock_wrapper)
mock_wrapper.__anext__ = Mock(side_effect=StopAsyncIteration)
iterator = LiteLLMCompletionStreamingIterator(
model="deepseek/deepseek-chat",
litellm_custom_stream_wrapper=mock_wrapper,
request_input="test",
responses_api_request={},
)
iterator.litellm_model_response = model_response
return iterator
def test_completed_response_set_after_common_done_event_logic(self):
"""common_done_event_logic builds a ResponseCompletedEvent and must
latch it onto self.completed_response so downstream wrappers can
read it. Before the fix the event was returned but
completed_response stayed None."""
from litellm.types.utils import Choices, Message, ModelResponse
complete_response = ModelResponse(
id="resp_test",
created=1234567890,
model="deepseek-chat",
object="chat.completion",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(content="hello", role="assistant"),
)
],
)
iterator = self._make_iterator_with_stop(complete_response)
import asyncio
async def drain():
results = []
try:
async for chunk in iterator:
results.append(chunk)
except StopAsyncIteration:
pass
return results
results = asyncio.run(drain())
assert len(results) > 0
assert iterator.completed_response is not None, (
"LiteLLMCompletionStreamingIterator.completed_response is still None "
"after common_done_event_logic ran — downstream wrappers and the "
"proxy container-ownership hook will see no terminal event"
)
assert iterator.completed_response.type == "response.completed"
class TestFallbackWrapperStopAsyncIterationFallback:
"""Regression: FallbackResponsesStreamWrapper.__anext__ only sniffed
terminal events off forwarded chunks. When the inner generator raises
StopAsyncIteration without a sniffable chunk (the bridge path ends this
way), the wrapper re-raised without checking source_iterator for a
latched completed_response, leaving its own completed_response None."""
def test_falls_back_to_source_completed_response_on_stop(self):
"""When the inner async generator raises StopAsyncIteration and the
wrapper never sniffed a terminal chunk, it must copy
source_iterator.completed_response so the proxy ownership hook
still sees the terminal event."""
import asyncio
from types import SimpleNamespace
from litellm.router import Router
source = SimpleNamespace(
response=None,
model="deepseek/deepseek-chat",
logging_obj=None,
responses_api_provider_config=None,
start_time=None,
litellm_metadata=None,
custom_llm_provider="deepseek",
request_data={},
call_type="aresponses",
_hidden_params={},
completed_response=SimpleNamespace(
type="response.completed",
response=SimpleNamespace(id="resp_src", output=[], container=None),
),
)
async def empty_gen():
return
yield # pragma: no cover
async def _drive():
router = Router(
model_list=[
{
"model_name": "deepseek/deepseek-chat",
"litellm_params": {"model": "deepseek/deepseek-chat", "api_key": "sk-test"},
}
]
)
wrapper = await router._aresponses_streaming_iterator(
response=source, # type: ignore[arg-type]
initial_kwargs={},
)
wrapper._async_generator = empty_gen()
out = []
try:
async for chunk in wrapper:
out.append(chunk)
except StopAsyncIteration:
pass
return wrapper, out
wrapper, _ = asyncio.run(_drive())
assert wrapper.completed_response is not None, (
"FallbackResponsesStreamWrapper.completed_response is None after "
"StopAsyncIteration even though source_iterator had one — the "
"proxy ownership hook will log a spurious warning"
)
assert wrapper.completed_response.type == "response.completed"
class TestEnsureOutputItemContentPartAdded:
"""Test that _ensure_output_item_for_chunk emits content_part.added after
output_item.added for message items."""

343
tests/test_litellm/responses/test_custom_tool_call.py Normal file
View file

@ -0,0 +1,343 @@
"""
Test custom_tool_call adaptation for apply_patch and other custom tools.
This test verifies that when Codex sends custom tools (type="custom"),
LiteLLM bridge correctly:
1. Converts them to function tools for Chat Completions providers
2. Converts function_call responses back to custom_tool_call output items
3. Unwraps the JSON-wrapping arguments to extract the actual input content
"""
import json
import pytest
from typing import Dict, Any, List
from openai.types.responses import ResponseFunctionToolCall
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
)
from litellm.responses.litellm_completion_transformation.custom_tools import (
extract_custom_tool_names,
is_custom_tool_call,
unwrap_custom_tool_arguments,
build_tool_call_item_kwargs,
convert_custom_tool_to_function_tool,
_MAX_ARGUMENTS_LEN,
)
from litellm.types.responses.main import CustomToolCallOutputItem
class TestCustomToolUtilities:
"""Test the custom_tools utility functions."""
def test_extract_custom_tool_names(self):
"""Test extraction of custom tool names from tools list."""
tools = [
{"type": "function", "name": "regular_tool"},
{"type": "custom", "name": "apply_patch"},
{"type": "function", "name": "another_tool"},
{"type": "custom", "name": "custom_format"},
]
names = extract_custom_tool_names(tools)
assert names == {"apply_patch", "custom_format"}
def test_extract_custom_tool_names_empty(self):
"""Test extraction with no custom tools."""
tools = [
{"type": "function", "name": "tool1"},
{"type": "function", "name": "tool2"},
]
names = extract_custom_tool_names(tools)
assert names == set()
def test_extract_custom_tool_names_none(self):
"""Test extraction with None input."""
names = extract_custom_tool_names(None)
assert names == set()
def test_is_custom_tool_call_true(self):
"""Test identification of custom tool call."""
custom_names = {"apply_patch", "custom_format"}
assert is_custom_tool_call("apply_patch", custom_names) is True
assert is_custom_tool_call("custom_format", custom_names) is True
def test_is_custom_tool_call_false(self):
"""Test identification of non-custom tool call."""
custom_names = {"apply_patch"}
assert is_custom_tool_call("regular_tool", custom_names) is False
assert is_custom_tool_call("unknown_tool", custom_names) is False
def test_unwrap_custom_tool_arguments(self):
"""Test unwrapping of JSON-wrapped arguments."""
# Test with valid JSON
wrapped = json.dumps({"content": "*** Begin Patch\n*** Add File: test.py\n+hello\n*** End Patch"})
unwrapped = unwrap_custom_tool_arguments(wrapped)
assert unwrapped == "*** Begin Patch\n*** Add File: test.py\n+hello\n*** End Patch"
def test_unwrap_custom_tool_arguments_invalid_json(self):
"""Test unwrapping with invalid JSON returns original."""
raw = "*** Begin Patch\n*** Add File: test.py\n+hello\n*** End Patch"
unwrapped = unwrap_custom_tool_arguments(raw)
assert unwrapped == raw
def test_unwrap_custom_tool_arguments_no_content_key(self):
"""Test unwrapping with JSON but no content key."""
wrapped = json.dumps({"other_key": "value"})
unwrapped = unwrap_custom_tool_arguments(wrapped)
assert unwrapped == wrapped
def test_build_tool_call_item_kwargs_custom_completed(self):
"""A completed custom tool call unwraps the content into `input`."""
wrapped = json.dumps({"content": "patch body"})
kwargs = build_tool_call_item_kwargs(
call_id="c1",
name="apply_patch",
arguments_or_input=wrapped,
status="completed",
custom_tool_names={"apply_patch"},
)
assert kwargs["type"] == "custom_tool_call"
assert kwargs["input"] == "patch body"
assert "arguments" not in kwargs
def test_build_tool_call_item_kwargs_custom_in_progress(self):
"""An in-progress custom tool call seeds an empty input string."""
kwargs = build_tool_call_item_kwargs(
call_id="c2",
name="apply_patch",
arguments_or_input="ignored-until-completed",
status="in_progress",
custom_tool_names={"apply_patch"},
)
assert kwargs["input"] == ""
def test_build_tool_call_item_kwargs_regular_function(self):
"""A regular function call keeps raw arguments and uses function_call type."""
raw = json.dumps({"k": "v"})
kwargs = build_tool_call_item_kwargs(
call_id="c3",
name="get_weather",
arguments_or_input=raw,
status="completed",
custom_tool_names=set(),
)
assert kwargs["type"] == "function_call"
assert kwargs["arguments"] == raw
assert "input" not in kwargs
def test_unwrap_custom_tool_arguments_oversized_returns_raw(self):
"""Arguments larger than the safety cap are returned unchanged to avoid
OOM on JSON parsing a pathologically large string."""
oversized = "x" * (_MAX_ARGUMENTS_LEN + 1)
assert unwrap_custom_tool_arguments(oversized) == oversized
def test_unwrap_custom_tool_arguments_empty(self):
"""Empty arguments unwrap to an empty string, not the raw input."""
assert unwrap_custom_tool_arguments("") == ""
def test_convert_custom_tool_to_function_tool_with_format(self):
"""The grammar definition is embedded in the description so the model can
produce correctly-formatted output."""
tool = {
"type": "custom",
"name": "apply_patch",
"description": "Apply a patch",
"format": {
"type": "grammar",
"syntax": "lark",
"definition": "start: begin_patch",
},
}
result = convert_custom_tool_to_function_tool(tool)
assert result is not None
assert result["type"] == "function"
assert "begin_patch" in result["function"]["description"]
assert result["function"]["parameters"]["required"] == ["content"]
def test_convert_custom_tool_to_function_tool_non_custom_returns_none(self):
"""Non-custom tools are not convertible; the caller keeps them as-is."""
assert convert_custom_tool_to_function_tool({"type": "function"}) is None
class TestTransformationCustomTools:
"""Test custom tool handling in transformation logic."""
def test_transform_apply_patch_function_call_to_custom_tool_call(self):
"""Test that apply_patch function_call is converted to custom_tool_call."""
# Simulate a Chat Completion response with apply_patch function call
from litellm.types.utils import ModelResponse, Choices, Message, ChatCompletionMessageToolCall, Function
tool_call = ChatCompletionMessageToolCall(
id="call_abc123",
type="function",
function=Function(
name="apply_patch",
arguments=json.dumps({"content": "*** Begin Patch\n*** Add File: test.py\n+hello\n*** End Patch"}),
),
)
message = Message(role="assistant", content=None, tool_calls=[tool_call])
choices = [Choices(index=0, message=message, finish_reason="tool_calls")]
response = ModelResponse(
id="test_response", choices=choices, created=1234567890, model="gpt-4", object="chat.completion"
)
# Transform with custom tool names
responses_api_request = {
"tools": [{"type": "custom", "name": "apply_patch"}, {"type": "function", "name": "regular_tool"}]
}
result = LiteLLMCompletionResponsesConfig.transform_chat_completion_tools_to_responses_tools(
response, responses_api_request=responses_api_request
)
# Should return a CustomToolCallOutputItem object; ResponsesAPIResponse
# accepts it directly via its output item union.
assert len(result) == 1
item = result[0]
assert isinstance(item, CustomToolCallOutputItem)
assert item.type == "custom_tool_call"
assert item.call_id == "call_abc123"
assert item.name == "apply_patch"
assert item.input == "*** Begin Patch\n*** Add File: test.py\n+hello\n*** End Patch"
assert item.status == "completed"
def test_custom_tool_call_input_item_recovers_payload_from_input(self):
"""A custom_tool_call input item stores its payload in `input`; the
assistant tool call must carry it as a JSON content envelope whether
`arguments` is missing or an empty string."""
for arguments in (None, ""):
item = {
"type": "custom_tool_call",
"call_id": "call_1",
"name": "apply_patch",
"input": "*** Begin Patch\n+hello\n*** End Patch",
}
if arguments is not None:
item["arguments"] = arguments
messages = (
LiteLLMCompletionResponsesConfig._transform_responses_api_function_call_to_chat_completion_message(
function_call=item
)
)
tool_call = messages[0]["tool_calls"][0]
assert tool_call["function"]["arguments"] == json.dumps(
{"content": "*** Begin Patch\n+hello\n*** End Patch"}
)
def test_function_call_input_item_with_empty_arguments_keeps_them_empty(self):
"""A plain function_call input item with empty or missing `arguments`
must produce an empty arguments string, never a `{"content": ...}`
envelope (that recovery is reserved for custom_tool_call items) and
never the literal string "None"."""
for item in (
{
"type": "function_call",
"call_id": "call_2",
"name": "get_weather",
"arguments": "",
"input": "stray value",
},
{
"type": "function_call",
"call_id": "call_3",
"name": "get_weather",
},
):
messages = (
LiteLLMCompletionResponsesConfig._transform_responses_api_function_call_to_chat_completion_message(
function_call=item
)
)
tool_call = messages[0]["tool_calls"][0]
assert tool_call["function"]["arguments"] == ""
def test_transform_regular_function_call_unchanged(self):
"""Test that regular function calls remain as ResponseFunctionToolCall."""
from litellm.types.utils import ModelResponse, Choices, Message, ChatCompletionMessageToolCall, Function
tool_call = ChatCompletionMessageToolCall(
id="call_xyz789",
type="function",
function=Function(name="regular_tool", arguments=json.dumps({"param": "value"})),
)
message = Message(role="assistant", content=None, tool_calls=[tool_call])
choices = [Choices(index=0, message=message, finish_reason="tool_calls")]
response = ModelResponse(
id="test_response", choices=choices, created=1234567890, model="gpt-4", object="chat.completion"
)
# Transform with custom tool names (regular_tool is NOT custom)
responses_api_request = {
"tools": [{"type": "custom", "name": "apply_patch"}, {"type": "function", "name": "regular_tool"}]
}
result = LiteLLMCompletionResponsesConfig.transform_chat_completion_tools_to_responses_tools(
response, responses_api_request=responses_api_request
)
# Should return ResponseFunctionToolCall
assert len(result) == 1
item = result[0]
assert isinstance(item, ResponseFunctionToolCall)
assert item.type == "function_call"
assert item.name == "regular_tool"
assert item.arguments == json.dumps({"param": "value"})
def test_transform_mixed_tool_calls(self):
"""Test transformation with both custom and regular tool calls."""
from litellm.types.utils import ModelResponse, Choices, Message, ChatCompletionMessageToolCall, Function
custom_call = ChatCompletionMessageToolCall(
id="call_001",
type="function",
function=Function(name="apply_patch", arguments=json.dumps({"content": "patch content"})),
)
regular_call = ChatCompletionMessageToolCall(
id="call_002", type="function", function=Function(name="get_weather", arguments=json.dumps({"city": "SF"}))
)
message = Message(role="assistant", content=None, tool_calls=[custom_call, regular_call])
choices = [Choices(index=0, message=message, finish_reason="tool_calls")]
response = ModelResponse(
id="test_response", choices=choices, created=1234567890, model="gpt-4", object="chat.completion"
)
responses_api_request = {
"tools": [{"type": "custom", "name": "apply_patch"}, {"type": "function", "name": "get_weather"}]
}
result = LiteLLMCompletionResponsesConfig.transform_chat_completion_tools_to_responses_tools(
response, responses_api_request=responses_api_request
)
assert len(result) == 2
# First should be custom_tool_call object
first = result[0]
assert isinstance(first, CustomToolCallOutputItem)
assert first.type == "custom_tool_call"
assert first.name == "apply_patch"
assert first.input == "patch content"
# Second should be function_call
second = result[1]
assert isinstance(second, ResponseFunctionToolCall)
assert second.type == "function_call"
assert second.name == "get_weather"
if __name__ == "__main__":
pytest.main([__file__, "-v"])