From 2f0cdb35bf35f77caa55faec321313af9c4615db Mon Sep 17 00:00:00 2001
From: Mateo Wang <277851410+mateo-berri@users.noreply.github.com>
Date: Mon, 6 Jul 2026 17:34:27 -0700
Subject: [PATCH] fix(responses-bridge): custom tool round-trip and allowlist
 preservation for Codex CLI (#32258)

* 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>
---
 .../guardrail_translation/handler.py          |   5 +-
 litellm/proxy/common_request_processing.py    |  35 +-
 .../custom_tools.py                           | 163 ++++++
 .../streaming_iterator.py                     | 112 ++--
 .../transformation.py                         | 539 +++++++++---------
 litellm/router.py                             |  15 +-
 litellm/types/llms/openai.py                  |   3 +
 litellm/types/responses/main.py               |  16 +
 .../proxy/test_common_request_processing.py   |  69 +++
 .../proxy/test_tools_allowlist_enforcement.py |  30 +
 .../test_litellm_completion_responses.py      | 283 ++++++++-
 .../responses/test_custom_tool_call.py        | 343 +++++++++++
 12 files changed, 1267 insertions(+), 346 deletions(-)
 create mode 100644 litellm/responses/litellm_completion_transformation/custom_tools.py
 create mode 100644 tests/test_litellm/responses/test_custom_tool_call.py

diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py
index d1323b1a2bf..093dffccac0 100644
--- a/litellm/llms/openai/responses/guardrail_translation/handler.py
+++ b/litellm/llms/openai/responses/guardrail_translation/handler.py
@@ -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"]))
diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py
index c4be2187292..abaf82fb661 100644
--- a/litellm/proxy/common_request_processing.py
+++ b/litellm/proxy/common_request_processing.py
@@ -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",
diff --git a/litellm/responses/litellm_completion_transformation/custom_tools.py b/litellm/responses/litellm_completion_transformation/custom_tools.py
new file mode 100644
index 00000000000..2417bf5cf2e
--- /dev/null
+++ b/litellm/responses/litellm_completion_transformation/custom_tools.py
@@ -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)
diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py
index b1198780bac..cf69654d15d 100644
--- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py
+++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py
@@ -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:
diff --git a/litellm/responses/litellm_completion_transformation/transformation.py b/litellm/responses/litellm_completion_transformation/transformation.py
index 866698b1f96..1aaa38cea14 100644
--- a/litellm/responses/litellm_completion_transformation/transformation.py
+++ b/litellm/responses/litellm_completion_transformation/transformation.py
@@ -2,15 +2,17 @@
 Handles transforming from Responses API -> LiteLLM completion  (Chat Completion API)
 """
 
+import json
 import re
 from collections.abc import Sequence
-from typing import Any, Dict, List, Literal, Optional, Set, Tuple, Union, cast
+from typing import Any, Literal, cast
 
 from openai.types.responses import ResponseFunctionToolCall
 from openai.types.responses.response_create_params import ResponseInputParam
 from openai.types.responses.tool_param import FunctionToolParam
 from typing_extensions import TypedDict
 
+from litellm._logging import verbose_logger
 from litellm.caching import InMemoryCache
 from litellm.litellm_core_utils.get_supported_openai_params import (
     get_supported_openai_params,
@@ -45,6 +47,7 @@ from litellm.types.llms.openai import (
     ValidChatCompletionMessageContentTypesLiteral,
 )
 from litellm.types.responses.main import (
+    CustomToolCallOutputItem,
     GenericResponseOutputItem,
     GenericResponseOutputItemContentAnnotation,
     OutputCodeInterpreterCall,
@@ -62,21 +65,26 @@ from litellm.types.utils import (
     Usage,
 )
 
+from .custom_tools import (
+    convert_custom_tool_to_function_tool,
+    extract_custom_tool_names,
+    is_custom_tool_call,
+    unwrap_custom_tool_arguments,
+)
+
 ########### Initialize Classes used for Responses API  ###########
 TOOL_CALLS_CACHE = InMemoryCache()
 
 
 class ChatCompletionSession(TypedDict, total=False):
-    messages: List[
-        Union[
-            AllMessageValues,
-            GenericChatCompletionMessage,
-            ChatCompletionMessageToolCall,
-            ChatCompletionResponseMessage,
-            Message,
-        ]
+    messages: list[
+        AllMessageValues
+        | GenericChatCompletionMessage
+        | ChatCompletionMessageToolCall
+        | ChatCompletionResponseMessage
+        | Message
     ]
-    litellm_session_id: Optional[str]
+    litellm_session_id: str | None
 
 
 ########### End of Initialize Classes used for Responses API  ###########
@@ -109,7 +117,7 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _transform_tool_choice(
         tool_choice: Any,
-    ) -> Optional[Union[str, Dict[str, Any]]]:
+    ) -> str | dict[str, Any] | None:
         """
         Transform tool_choice from various formats to OpenAI Chat Completion format.
 
@@ -159,7 +167,7 @@ class LiteLLMCompletionResponsesConfig:
         return tool_choice
 
     @staticmethod
-    def _should_drop_derived_web_search_options(model: str, custom_llm_provider: Optional[str]) -> bool:
+    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
@@ -169,7 +177,7 @@ class LiteLLMCompletionResponsesConfig:
         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: Optional[List[str]] = get_supported_openai_params(
+        supported_params: 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
@@ -177,11 +185,11 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def transform_responses_api_request_to_chat_completion_request(
         model: str,
-        input: Union[str, ResponseInputParam],
+        input: str | ResponseInputParam,
         responses_api_request: ResponsesAPIOptionalRequestParams,
-        custom_llm_provider: Optional[str] = None,
-        stream: Optional[bool] = None,
-        extra_headers: Optional[Dict[str, Any]] = None,
+        custom_llm_provider: str | None = None,
+        stream: bool | None = None,
+        extra_headers: dict[str, Any] | None = None,
         **kwargs,
     ) -> dict:
         """
@@ -205,7 +213,7 @@ class LiteLLMCompletionResponsesConfig:
             response_format = LiteLLMCompletionResponsesConfig._transform_text_format_to_response_format(text_param)
 
         # Extract reasoning_effort from reasoning parameter
-        reasoning_effort: Optional[Union[Reasoning, str]] = None
+        reasoning_effort: Reasoning | str | None = None
         reasoning_param = responses_api_request.get("reasoning")
         if reasoning_param:
             if isinstance(reasoning_param, dict):
@@ -255,7 +263,7 @@ class LiteLLMCompletionResponsesConfig:
                 "include_usage": True,
             }
             litellm_completion_request["stream_options"] = stream_options
-            litellm_logging_obj: Optional[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
+            litellm_logging_obj: LiteLLMLoggingObj | None = kwargs.get("litellm_logging_obj")
             if litellm_logging_obj:
                 litellm_logging_obj.stream_options = stream_options
 
@@ -265,28 +273,24 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def transform_responses_api_input_to_messages(
-        input: Union[str, ResponseInputParam],
-        responses_api_request: Union[ResponsesAPIOptionalRequestParams, dict],
-    ) -> List[
-        Union[
-            AllMessageValues,
-            GenericChatCompletionMessage,
-            ChatCompletionMessageToolCall,
-            ChatCompletionResponseMessage,
-            Message,
-        ]
+        input: str | ResponseInputParam,
+        responses_api_request: ResponsesAPIOptionalRequestParams | dict,
+    ) -> list[
+        AllMessageValues
+        | GenericChatCompletionMessage
+        | ChatCompletionMessageToolCall
+        | ChatCompletionResponseMessage
+        | Message
     ]:
         """
         Transform a Responses API input into a list of messages
         """
-        messages: List[
-            Union[
-                AllMessageValues,
-                GenericChatCompletionMessage,
-                ChatCompletionMessageToolCall,
-                ChatCompletionResponseMessage,
-                Message,
-            ]
+        messages: list[
+            AllMessageValues
+            | GenericChatCompletionMessage
+            | ChatCompletionMessageToolCall
+            | ChatCompletionResponseMessage
+            | Message
         ] = []
         if responses_api_request.get("instructions"):
             messages.append(
@@ -373,31 +377,24 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _transform_response_input_param_to_chat_completion_message(
-        input: Union[str, ResponseInputParam],
-    ) -> List[
-        Union[
-            AllMessageValues,
-            GenericChatCompletionMessage,
-            ChatCompletionMessageToolCall,
-            ChatCompletionResponseMessage,
-        ]
+        input: str | ResponseInputParam,
+    ) -> list[
+        AllMessageValues | GenericChatCompletionMessage | ChatCompletionMessageToolCall | ChatCompletionResponseMessage
     ]:
         """
         Transform a ResponseInputParam into a Chat Completion message
         """
-        messages: List[
-            Union[
-                AllMessageValues,
-                GenericChatCompletionMessage,
-                ChatCompletionMessageToolCall,
-                ChatCompletionResponseMessage,
-            ]
+        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: Set[str] = set()
+            existing_tool_call_ids: set[str] = set()
             for _input in input:
                 chat_completion_messages = (
                     LiteLLMCompletionResponsesConfig._transform_responses_api_input_item_to_chat_completion_message(
@@ -449,7 +446,7 @@ class LiteLLMCompletionResponsesConfig:
                     if not chat_completion_messages:
                         continue
 
-                    deduped_in_place: List[Any] = []
+                    deduped_in_place: list[Any] = []
                     for m in chat_completion_messages:
                         role = ""
                         if isinstance(m, dict):
@@ -491,36 +488,27 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _deduplicate_tool_call_output_messages(
-        tool_call_output_messages: List[
-            Union[
-                AllMessageValues,
-                GenericChatCompletionMessage,
-                ChatCompletionMessageToolCall,
-                ChatCompletionResponseMessage,
-            ]
+        tool_call_output_messages: list[
+            AllMessageValues
+            | GenericChatCompletionMessage
+            | ChatCompletionMessageToolCall
+            | ChatCompletionResponseMessage
         ],
-        existing_tool_call_ids: Set[str],
-    ) -> List[
-        Union[
-            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[
-            Union[
-                AllMessageValues,
-                GenericChatCompletionMessage,
-                ChatCompletionMessageToolCall,
-                ChatCompletionResponseMessage,
-            ]
+        filtered_messages: list[
+            AllMessageValues
+            | GenericChatCompletionMessage
+            | ChatCompletionMessageToolCall
+            | ChatCompletionResponseMessage
         ] = []
-        seen_tool_call_ids: Set[str] = set(existing_tool_call_ids)
+        seen_tool_call_ids: set[str] = set(existing_tool_call_ids)
 
         for tool_call_message in tool_call_output_messages:
             if isinstance(tool_call_message, dict):
@@ -563,7 +551,7 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _ensure_tool_call_output_has_corresponding_tool_call(
-        messages: List[Union[AllMessageValues, GenericChatCompletionMessage]],
+        messages: list[AllMessageValues | GenericChatCompletionMessage],
     ) -> bool:
         """
         If any tool call output is present, ensure there is a corresponding tool call/tool_use block
@@ -574,7 +562,7 @@ class LiteLLMCompletionResponsesConfig:
         return False
 
     @staticmethod
-    def _find_previous_assistant_idx(messages: List[Any], current_idx: int) -> Optional[int]:
+    def _find_previous_assistant_idx(messages: list[Any], 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":
@@ -600,7 +588,7 @@ class LiteLLMCompletionResponsesConfig:
         return ""
 
     @staticmethod
-    def _get_tool_calls_list(assistant_message: Any) -> List[Any]:
+    def _get_tool_calls_list(assistant_message: Any) -> list[Any]:
         """Extract tool_calls as a list from assistant message."""
         tool_calls_raw = (
             assistant_message.get("tool_calls")
@@ -616,10 +604,10 @@ class LiteLLMCompletionResponsesConfig:
         return []
 
     @staticmethod
-    def _check_tool_call_exists(tool_calls: List[Any], tool_call_id: str) -> bool:
+    def _check_tool_call_exists(tool_calls: list[Any], 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: Optional[str] = None
+            tool_call_id_to_check: str | None = None
             if isinstance(tool_call, dict):
                 tool_call_id_to_check = tool_call.get("id")
             elif hasattr(tool_call, "id"):
@@ -629,7 +617,7 @@ class LiteLLMCompletionResponsesConfig:
         return False
 
     @staticmethod
-    def _reconstruct_tool_call_from_tools(tool_call_id: str, tools: List[Any]) -> Optional[Dict[str, Any]]:
+    def _reconstruct_tool_call_from_tools(tool_call_id: str, tools: list[Any]) -> dict[str, Any] | None:
         """Reconstruct a minimal tool_call definition from tools list."""
         for tool in tools:
             if isinstance(tool, dict):
@@ -668,13 +656,13 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _create_tool_call_chunk(
-        tool_use_definition: Dict[str, Any], tool_call_id: str, index: int
+        tool_use_definition: dict[str, Any], 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 = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(function_raw, "name")
         function_arguments_raw = LiteLLMCompletionResponsesConfig._get_mapping_or_attr_value(function_raw, "arguments")
-        function: Dict[str, Any] = {
+        function: dict[str, Any] = {
             "name": function_name_raw or "",
             "arguments": function_arguments_raw or "{}",
         }
@@ -693,7 +681,7 @@ class LiteLLMCompletionResponsesConfig:
         )
 
     @staticmethod
-    def _normalize_tool_use_definition(tool_use_definition: Any, tool_call_id: str) -> Optional[Dict[str, Any]]:
+    def _normalize_tool_use_definition(tool_use_definition: Any, tool_call_id: str) -> dict[str, Any] | None:
         """
         Normalize cached tool_call definitions to a dict-like shape consumed by _create_tool_call_chunk.
         """
@@ -701,7 +689,7 @@ class LiteLLMCompletionResponsesConfig:
             return None
 
         if isinstance(tool_use_definition, dict):
-            normalized_definition: Dict[str, Any] = dict(tool_use_definition)
+            normalized_definition: dict[str, Any] = dict(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")
@@ -737,7 +725,7 @@ class LiteLLMCompletionResponsesConfig:
     def _add_tool_call_to_assistant(assistant_message: Any, tool_call_chunk: ChatCompletionToolCallChunk) -> None:
         """Add a tool_call to an assistant message."""
         if isinstance(assistant_message, dict):
-            prev_assistant_dict = cast(Dict[str, Any], assistant_message)
+            prev_assistant_dict = cast(dict[str, Any], assistant_message)
             if "tool_calls" not in prev_assistant_dict:
                 prev_assistant_dict["tool_calls"] = []
             tool_calls_list = prev_assistant_dict["tool_calls"]
@@ -752,23 +740,19 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _ensure_tool_results_have_corresponding_tool_calls(
         messages: Sequence[
-            Union[
-                AllMessageValues,
-                GenericChatCompletionMessage,
-                ChatCompletionResponseMessage,
-                ChatCompletionMessageToolCall,
-                Message,
-            ]
+            AllMessageValues
+            | GenericChatCompletionMessage
+            | ChatCompletionResponseMessage
+            | ChatCompletionMessageToolCall
+            | Message
         ],
-        tools: Optional[List[Any]] = None,
-    ) -> List[
-        Union[
-            AllMessageValues,
-            GenericChatCompletionMessage,
-            ChatCompletionResponseMessage,
-            ChatCompletionMessageToolCall,
-            Message,
-        ]
+        tools: list[Any] | None = None,
+    ) -> list[
+        AllMessageValues
+        | GenericChatCompletionMessage
+        | ChatCompletionResponseMessage
+        | ChatCompletionMessageToolCall
+        | Message
     ]:
         """
         Ensure that tool_result messages have corresponding tool_calls in the previous assistant message.
@@ -789,14 +773,12 @@ class LiteLLMCompletionResponsesConfig:
         # Create a deep copy to avoid modifying the original (use list() so we can mutate and return List)
         import copy
 
-        fixed_messages: List[
-            Union[
-                AllMessageValues,
-                GenericChatCompletionMessage,
-                ChatCompletionResponseMessage,
-                ChatCompletionMessageToolCall,
-                Message,
-            ]
+        fixed_messages: list[
+            AllMessageValues
+            | GenericChatCompletionMessage
+            | ChatCompletionResponseMessage
+            | ChatCompletionMessageToolCall
+            | Message
         ] = list(copy.deepcopy(messages))
         messages_to_remove = []
 
@@ -828,7 +810,7 @@ class LiteLLMCompletionResponsesConfig:
                     # 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, Any], message)
+                        message_dict = cast(dict[str, Any], message)
                         message_dict["tool_call_id"] = tool_call_id
                     elif hasattr(message, "tool_call_id"):
                         setattr(message, "tool_call_id", tool_call_id)
@@ -881,13 +863,7 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _transform_responses_api_input_item_to_chat_completion_message(
         input_item: Any,
-    ) -> List[
-        Union[
-            AllMessageValues,
-            GenericChatCompletionMessage,
-            ChatCompletionResponseMessage,
-        ]
-    ]:
+    ) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage]:
         """
         Transform a Responses API input item into a Chat Completion message
 
@@ -937,6 +913,7 @@ class LiteLLMCompletionResponsesConfig:
         """
         return input_item.get("type") in [
             "function_call_output",
+            "custom_tool_call_output",
             "web_search_call",
             "computer_call_output",
             "tool_result",  # Anthropic/MCP format
@@ -945,20 +922,16 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _is_input_item_function_call(input_item: Any) -> bool:
         """
-        Check if the input item is a function call
+        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") == "function_call"
+        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: Dict[str, Any],
-    ) -> List[
-        Union[
-            AllMessageValues,
-            GenericChatCompletionMessage,
-            ChatCompletionResponseMessage,
-        ]
-    ]:
+        tool_call_output: dict[str, Any],
+    ) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage]:
         """
         ChatCompletionToolMessage is used to indicate the output from a tool call
         """
@@ -992,8 +965,8 @@ class LiteLLMCompletionResponsesConfig:
 
             # Some adapters represent tool output as a list of "input_*" parts
             if isinstance(output, list):
-                normalized_blocks: List[Dict[str, Any]] = []
-                text_acc: List[str] = []
+                normalized_blocks: list[dict[str, Any]] = []
+                text_acc: list[str] = []
                 for part in output:
                     if not isinstance(part, dict):
                         continue
@@ -1093,14 +1066,8 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _transform_responses_api_function_call_to_chat_completion_message(
-        function_call: Dict[str, Any],
-    ) -> List[
-        Union[
-            AllMessageValues,
-            GenericChatCompletionMessage,
-            ChatCompletionResponseMessage,
-        ]
-    ]:
+        function_call: dict[str, Any],
+    ) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage]:
         """
         Transform a Responses API function_call into a Chat Completion message with tool calls
 
@@ -1117,13 +1084,19 @@ class LiteLLMCompletionResponsesConfig:
         }
         ```
         """
-        # Create a tool call for the function call
+        # 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 not raw_arguments and function_call.get("type") == "custom_tool_call":
+            raw_input = function_call.get("input") or ""
+            raw_arguments = json.dumps({"content": raw_input}) if raw_input else ""
         tool_call = ChatCompletionToolCallChunk(
             id=function_call.get("call_id") or function_call.get("id") or "",
             type="function",
             function=ChatCompletionToolCallFunctionChunk(
                 name=function_call.get("name") or "",
-                arguments=str(function_call.get("arguments") or ""),
+                arguments=str(raw_arguments or ""),
             ),
             index=0,
         )
@@ -1138,7 +1111,7 @@ class LiteLLMCompletionResponsesConfig:
         return [chat_completion_response_message]
 
     @staticmethod
-    def _resolve_file_id(item: Dict[str, Any]) -> Optional[str]:
+    def _resolve_file_id(item: dict[str, Any]) -> str | None:
         """
         Return the effective file_id for a Responses API input_file item.
         Explicit file_id takes precedence; file_url is used as fallback so
@@ -1147,7 +1120,7 @@ class LiteLLMCompletionResponsesConfig:
         return item.get("file_id") or item.get("file_url") or None
 
     @staticmethod
-    def _transform_input_file_item_to_file_item(item: Dict[str, Any]) -> Dict[str, Any]:
+    def _transform_input_file_item_to_file_item(item: dict[str, Any]) -> dict[str, Any]:
         """
         Transform a Responses API input_file item to a Chat Completion file item
 
@@ -1157,21 +1130,21 @@ class LiteLLMCompletionResponsesConfig:
         Returns:
             Dictionary with transformed file structure for Chat Completion
         """
-        file_dict: Dict[str, Any] = {}
+        file_dict: dict[str, Any] = {}
         file_id = 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: Dict[str, Any] = {"type": "file", "file": file_dict}
+        new_item: dict[str, Any] = {"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: Dict[str, Any],
+        item: dict[str, Any],
     ) -> ChatCompletionImageObject:
         """
         Transform a Responses API input_image item to a Chat Completion image item
@@ -1185,7 +1158,7 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _transform_responses_api_content_to_chat_completion_content(
         content: Any,
-    ) -> Union[str, List[Union[str, Dict[str, Any]]]]:
+    ) -> str | list[str | dict[str, Any]]:
         """
         Transform a Responses API content into a Chat Completion content
 
@@ -1199,7 +1172,7 @@ class LiteLLMCompletionResponsesConfig:
         elif isinstance(content, str):
             return content
         elif isinstance(content, list):
-            content_list: List[Union[str, Dict[str, Any]]] = []
+            content_list: list[str | dict[str, Any]] = []
             for item in content:
                 if isinstance(item, str):
                     content_list.append(item)
@@ -1220,7 +1193,7 @@ class LiteLLMCompletionResponsesConfig:
                         text_value = item.get("text")
                         if text_value is None:
                             continue
-                        content_block: Dict[str, Any] = {
+                        content_block: dict[str, Any] = {
                             "type": LiteLLMCompletionResponsesConfig._get_chat_completion_request_content_type(
                                 item.get("type") or "text"
                             ),
@@ -1268,7 +1241,7 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def transform_instructions_to_system_message(
-        instructions: Optional[str],
+        instructions: str | None,
     ) -> ChatCompletionSystemMessage:
         """
         Transform a Instructions into a system message
@@ -1277,18 +1250,18 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def transform_responses_api_tools_to_chat_completion_tools(
-        tools: Optional[List[Union[FunctionToolParam, OpenAIMcpServerTool]]],
-    ) -> Tuple[
-        List[Union[ChatCompletionToolParam, OpenAIMcpServerTool]],
-        Optional[OpenAIWebSearchOptions],
+        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
-        chat_completion_tools: List[Union[ChatCompletionToolParam, OpenAIMcpServerTool]] = []
-        web_search_options: Optional[OpenAIWebSearchOptions] = None
+        chat_completion_tools: 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))
@@ -1296,8 +1269,8 @@ class LiteLLMCompletionResponsesConfig:
                 _search_context_size: Literal["low", "medium", "high"] = cast(
                     Literal["low", "medium", "high"], tool.get("search_context_size")
                 )
-                _user_location: Optional[OpenAIWebSearchUserLocation] = cast(
-                    Optional[OpenAIWebSearchUserLocation],
+                _user_location: OpenAIWebSearchUserLocation | None = cast(
+                    OpenAIWebSearchUserLocation | None,
                     tool.get("user_location") or None,
                 )
                 web_search_options = OpenAIWebSearchOptions(
@@ -1310,7 +1283,7 @@ class LiteLLMCompletionResponsesConfig:
                 parameters = dict(typed_tool.get("parameters", {}) or {})
                 if not parameters or "type" not in parameters:
                     parameters["type"] = "object"
-                chat_completion_tool: Dict[str, Any] = {
+                chat_completion_tool: dict[str, Any] = {
                     "type": "function",
                     "function": {
                         "name": typed_tool.get("name") or "",
@@ -1328,14 +1301,30 @@ class LiteLLMCompletionResponsesConfig:
                 if tool.get("input_examples"):
                     chat_completion_tool["input_examples"] = tool.get("input_examples")  # type: ignore
                 chat_completion_tools.append(cast(ChatCompletionToolParam, chat_completion_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:
-                chat_completion_tools.append(cast(Union[ChatCompletionToolParam, OpenAIMcpServerTool], tool))
+                _tool_type = tool.get("type")
+                if _tool_type in ("computer_use", "image_generation", "namespace", "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: Optional[List[Union[ChatCompletionToolParam, OpenAIMcpServerTool]]],
-    ) -> List[Dict[str, Any]]:
+        chat_completion_tools: list[ChatCompletionToolParam | OpenAIMcpServerTool] | None,
+    ) -> list[dict[str, Any]]:
         """
         Transform Chat Completion tool params (e.g. from guardrail output) back to
         Responses API request tool format. Inverse of
@@ -1343,17 +1332,17 @@ class LiteLLMCompletionResponsesConfig:
         """
         if chat_completion_tools is None or not chat_completion_tools:
             return []
-        result: List[Dict[str, Any]] = []
+        result: list[dict[str, Any]] = []
         for tool in chat_completion_tools:
             if not isinstance(tool, dict):
                 result.append(tool)  # type: ignore
                 continue
             if tool.get("type") == "function":
-                fn = cast(Dict[str, Any], tool.get("function") or {})
+                fn = cast(dict[str, Any], 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, Any] = {
+                responses_tool: dict[str, Any] = {
                     "type": "function",
                     "name": fn.get("name") or "",
                     "description": fn.get("description") or "",
@@ -1377,11 +1366,16 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def transform_chat_completion_tools_to_responses_tools(
         chat_completion_response: ModelResponse,
-    ) -> List[ResponseFunctionToolCall]:
+        responses_api_request: ResponsesAPIOptionalRequestParams | None = None,
+    ) -> list[ResponseFunctionToolCall | CustomToolCallOutputItem]:
         """
-        Transform a Chat Completion tools into a Responses API tools
+        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: List[ChatCompletionMessageToolCall] = []
+        all_chat_completion_tools: list[ChatCompletionMessageToolCall] = []
         for choice in chat_completion_response.choices:
             if isinstance(choice, Choices):
                 if choice.message.tool_calls:
@@ -1392,53 +1386,77 @@ class LiteLLMCompletionResponsesConfig:
                             value=tool_call,
                         )
 
-        responses_tools: List[ResponseFunctionToolCall] = []
+        # Extract custom tool names from the original request
+        custom_tool_names: set[str] = set()
+        if responses_api_request and "tools" in responses_api_request:
+            custom_tool_names = extract_custom_tool_names(responses_api_request["tools"])
+
+        responses_tools: list[ResponseFunctionToolCall | CustomToolCallOutputItem] = []
         for tool in all_chat_completion_tools:
             if tool.type == "function":
                 function_definition = tool.function
-                provider_specific_fields: Optional[Dict] = 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)  # type: ignore
-                            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)  # type: ignore
-                            if hasattr(provider_specific_fields, "__dict__")
-                            else {}
-                        )
+                tool_name = function_definition.name or ""
+                tool_id = tool.id or ""
+                tool_arguments = function_definition.get("arguments") or ""
 
-                output_tool_call: ResponseFunctionToolCall = ResponseFunctionToolCall(
-                    name=function_definition.name or "",
-                    arguments=function_definition.get("arguments") or "",
-                    call_id=tool.id or "",
-                    id=tool.id or "",
-                    type="function_call",  # critical this is "function_call" to work with tools like openai codex
-                    status=function_definition.get("status") or "completed",
-                )
+                # 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
+                    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)  # type: ignore
+                                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)  # type: ignore
+                                if hasattr(provider_specific_fields, "__dict__")
+                                else {}
+                            )
 
-                # Pass through provider_specific_fields as-is if present
-                if provider_specific_fields:
-                    setattr(
-                        output_tool_call,
-                        "provider_specific_fields",
-                        provider_specific_fields,
-                    )  # type: ignore
+                    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",
+                    )
 
-                responses_tools.append(output_tool_call)
+                    # Pass through provider_specific_fields as-is if present
+                    if provider_specific_fields:
+                        setattr(
+                            output_tool_call,
+                            "provider_specific_fields",
+                            provider_specific_fields,
+                        )  # type: ignore
+
+                    responses_tools.append(output_tool_call)
         return responses_tools
 
     @staticmethod
     def _map_chat_completion_finish_reason_to_responses_status(
-        finish_reason: Optional[str],
+        finish_reason: str | None,
     ) -> ResponsesAPIStatus:
         """
         Map chat completion finish_reason to responses API status.
@@ -1465,7 +1483,7 @@ class LiteLLMCompletionResponsesConfig:
             return "completed"
 
     @staticmethod
-    def _tool_call_id_from_responses_item(item_id: Optional[str], call_id: Optional[str]) -> str:
+    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
@@ -1480,7 +1498,7 @@ class LiteLLMCompletionResponsesConfig:
     def convert_response_function_tool_call_to_chat_completion_tool_call(
         tool_call_item: Any,
         index: int = 0,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Convert ResponseFunctionToolCall to ChatCompletionToolCallChunk format.
 
@@ -1510,7 +1528,7 @@ class LiteLLMCompletionResponsesConfig:
                     )
                 )
 
-        function_dict: Dict[str, Any] = {
+        function_dict: dict[str, Any] = {
             "name": tool_call_item.name,
             "arguments": tool_call_item.arguments,
         }
@@ -1518,7 +1536,7 @@ class LiteLLMCompletionResponsesConfig:
         if provider_specific_fields:
             function_dict["provider_specific_fields"] = provider_specific_fields
 
-        tool_call_dict: Dict[str, Any] = {
+        tool_call_dict: dict[str, Any] = {
             "id": LiteLLMCompletionResponsesConfig._tool_call_id_from_responses_item(
                 getattr(tool_call_item, "id", None),
                 getattr(tool_call_item, "call_id", None),
@@ -1537,7 +1555,7 @@ class LiteLLMCompletionResponsesConfig:
     def convert_apply_patch_tool_call_to_chat_completion_tool_call(
         tool_call_item: Any,
         index: int = 0,
-    ) -> Dict[str, Any]:
+    ) -> dict[str, Any]:
         """
         Convert ResponseApplyPatchToolCall to ChatCompletionToolCallChunk format.
 
@@ -1555,7 +1573,7 @@ class LiteLLMCompletionResponsesConfig:
         import json
 
         operation_dict = tool_call_item.operation.model_dump()
-        tool_call_dict: Dict[str, Any] = {
+        tool_call_dict: dict[str, Any] = {
             "id": tool_call_item.call_id,
             "function": {
                 "name": "apply_patch",
@@ -1568,9 +1586,9 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def transform_chat_completion_response_to_responses_api_response(
-        request_input: Union[str, ResponseInputParam],
+        request_input: str | ResponseInputParam,
         responses_api_request: ResponsesAPIOptionalRequestParams,
-        chat_completion_response: Union[ModelResponse, dict],
+        chat_completion_response: ModelResponse | dict,
     ) -> ResponsesAPIResponse:
         """
         Transform a Chat Completion response into a Responses API response
@@ -1578,8 +1596,8 @@ class LiteLLMCompletionResponsesConfig:
         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: Optional[str] = None
-        choices: List[Choices] = getattr(chat_completion_response, "choices", [])
+        finish_reason: str | None = None
+        choices: list[Choices] = getattr(chat_completion_response, "choices", [])
         if choices and len(choices) > 0:
             finish_reason = choices[0].finish_reason
 
@@ -1595,6 +1613,7 @@ class LiteLLMCompletionResponsesConfig:
             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),
@@ -1626,24 +1645,23 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _transform_chat_completion_choices_to_responses_output(
         chat_completion_response: ModelResponse,
-        choices: List[Choices],
-    ) -> List[
-        Union[
-            GenericResponseOutputItem,
-            OutputCodeInterpreterCall,
-            OutputFunctionToolCall,
-            OutputImageGenerationCall,
-            ResponseFunctionToolCall,
-        ]
+        choices: list[Choices],
+        responses_api_request: ResponsesAPIOptionalRequestParams | None = None,
+    ) -> list[
+        GenericResponseOutputItem
+        | OutputCodeInterpreterCall
+        | OutputFunctionToolCall
+        | OutputImageGenerationCall
+        | ResponseFunctionToolCall
+        | CustomToolCallOutputItem
     ]:
-        responses_output: List[
-            Union[
-                GenericResponseOutputItem,
-                OutputCodeInterpreterCall,
-                OutputFunctionToolCall,
-                OutputImageGenerationCall,
-                ResponseFunctionToolCall,
-            ]
+        responses_output: list[
+            GenericResponseOutputItem
+            | OutputCodeInterpreterCall
+            | OutputFunctionToolCall
+            | OutputImageGenerationCall
+            | ResponseFunctionToolCall
+            | CustomToolCallOutputItem
         ] = []
 
         responses_output.extend(
@@ -1654,7 +1672,8 @@ class LiteLLMCompletionResponsesConfig:
         )
         responses_output.extend(
             LiteLLMCompletionResponsesConfig.transform_chat_completion_tools_to_responses_tools(
-                chat_completion_response=chat_completion_response
+                chat_completion_response=chat_completion_response,
+                responses_api_request=responses_api_request,
             )
         )
 
@@ -1713,8 +1732,8 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _extract_reasoning_output_items(
         chat_completion_response: ModelResponse,
-        choices: List[Choices],
-    ) -> List[GenericResponseOutputItem]:
+        choices: list[Choices],
+    ) -> list[GenericResponseOutputItem]:
         for choice in choices:
             if hasattr(choice, "message") and choice.message:
                 message = choice.message
@@ -1743,7 +1762,7 @@ class LiteLLMCompletionResponsesConfig:
     def _extract_image_generation_output_items(
         chat_completion_response: ModelResponse,
         choice: Choices,
-    ) -> List[OutputImageGenerationCall]:
+    ) -> list[OutputImageGenerationCall]:
         """
         Extract image generation outputs from a choice that contains images.
 
@@ -1762,7 +1781,7 @@ class LiteLLMCompletionResponsesConfig:
             'result': 'iVBORw0...'  # Pure base64 without data: prefix
         }
         """
-        image_generation_items: List[OutputImageGenerationCall] = []
+        image_generation_items: list[OutputImageGenerationCall] = []
 
         images = getattr(choice.message, "images", [])
         if not images:
@@ -1789,7 +1808,7 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _map_finish_reason_to_image_generation_status(
-        finish_reason: Optional[str],
+        finish_reason: str | None,
     ) -> Literal["in_progress", "completed", "incomplete", "failed"]:
         """
         Map finish_reason to image generation status.
@@ -1808,7 +1827,7 @@ class LiteLLMCompletionResponsesConfig:
             return "completed"
 
     @staticmethod
-    def _extract_base64_from_data_url(data_url: str) -> Optional[str]:
+    def _extract_base64_from_data_url(data_url: str) -> str | None:
         """
         Extract pure base64 string from a data URL.
 
@@ -1834,9 +1853,9 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _extract_message_output_items(
         chat_completion_response: ModelResponse,
-        choices: List[Choices],
-    ) -> List[Union[GenericResponseOutputItem, OutputImageGenerationCall]]:
-        message_output_items: List[Union[GenericResponseOutputItem, OutputImageGenerationCall]] = []
+        choices: list[Choices],
+    ) -> list[GenericResponseOutputItem | OutputImageGenerationCall]:
+        message_output_items: list[GenericResponseOutputItem | OutputImageGenerationCall] = []
         for choice in choices:
             # Check if message has images (image generation)
             if hasattr(choice.message, "images") and choice.message.images:
@@ -1868,20 +1887,8 @@ class LiteLLMCompletionResponsesConfig:
     @staticmethod
     def _transform_responses_api_outputs_to_chat_completion_messages(
         responses_api_output: ResponsesAPIResponse,
-    ) -> List[
-        Union[
-            AllMessageValues,
-            GenericChatCompletionMessage,
-            ChatCompletionMessageToolCall,
-        ]
-    ]:
-        messages: List[
-            Union[
-                AllMessageValues,
-                GenericChatCompletionMessage,
-                ChatCompletionMessageToolCall,
-            ]
-        ] = []
+    ) -> list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionMessageToolCall]:
+        messages: list[AllMessageValues | GenericChatCompletionMessage | ChatCompletionMessageToolCall] = []
         output_items = responses_api_output.output
         for _output_item in output_items:
             output_item: dict = dict(_output_item)
@@ -1939,9 +1946,9 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _transform_chat_completion_annotations_to_response_output_annotations(
-        annotations: Optional[List[ChatCompletionAnnotation]],
-    ) -> List[GenericResponseOutputItemContentAnnotation]:
-        response_output_annotations: List[GenericResponseOutputItemContentAnnotation] = []
+        annotations: list[ChatCompletionAnnotation] | None,
+    ) -> list[GenericResponseOutputItemContentAnnotation]:
+        response_output_annotations: list[GenericResponseOutputItemContentAnnotation] = []
 
         if annotations is None:
             return response_output_annotations
@@ -1965,10 +1972,10 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _transform_chat_completion_usage_to_responses_usage(
-        chat_completion_response: Union[ModelResponse, Usage],
+        chat_completion_response: ModelResponse | Usage,
     ) -> ResponseAPIUsage:
         if isinstance(chat_completion_response, ModelResponse):
-            usage: Optional[Usage] = getattr(chat_completion_response, "usage", None)
+            usage: Usage | None = getattr(chat_completion_response, "usage", None)
         else:
             usage = chat_completion_response
         if usage is None:
@@ -1991,7 +1998,7 @@ class LiteLLMCompletionResponsesConfig:
         # Translate prompt_tokens_details to input_tokens_details
         if hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details is not None:
             prompt_details = usage.prompt_tokens_details
-            input_details_dict: Dict[str, int] = {}
+            input_details_dict: 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
@@ -2010,7 +2017,7 @@ class LiteLLMCompletionResponsesConfig:
         # Translate completion_tokens_details to output_tokens_details
         if hasattr(usage, "completion_tokens_details") and usage.completion_tokens_details is not None:
             completion_details = usage.completion_tokens_details
-            output_details_dict: Dict[str, int] = {}
+            output_details_dict: dict[str, int] = {}
             if hasattr(completion_details, "reasoning_tokens") and completion_details.reasoning_tokens is not None:
                 output_details_dict["reasoning_tokens"] = completion_details.reasoning_tokens
             else:
@@ -2029,8 +2036,8 @@ class LiteLLMCompletionResponsesConfig:
 
     @staticmethod
     def _transform_text_format_to_response_format(
-        text_param: Union[Dict[str, Any], Any],
-    ) -> Optional[Dict[str, Any]]:
+        text_param: dict[str, Any] | Any,
+    ) -> dict[str, Any] | None:
         """
         Transform Responses API text.format parameter to Chat Completion response_format parameter.
 
diff --git a/litellm/router.py b/litellm/router.py
index 64b90172c04..12b96430334 100644
--- a/litellm/router.py
+++ b/litellm/router.py
@@ -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,
diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py
index 4c656b32081..3ab5a7b736e 100644
--- a/litellm/types/llms/openai.py
+++ b/litellm/types/llms/openai.py
@@ -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,
             ]
         ],
     ]
diff --git a/litellm/types/responses/main.py b/litellm/types/responses/main.py
index ebd2ad5b5a8..32e07f9e52f 100644
--- a/litellm/types/responses/main.py
+++ b/litellm/types/responses/main.py
@@ -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
diff --git a/tests/test_litellm/proxy/test_common_request_processing.py b/tests/test_litellm/proxy/test_common_request_processing.py
index c185b694e68..c61fdec370c 100644
--- a/tests/test_litellm/proxy/test_common_request_processing.py
+++ b/tests/test_litellm/proxy/test_common_request_processing.py
@@ -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
diff --git a/tests/test_litellm/proxy/test_tools_allowlist_enforcement.py b/tests/test_litellm/proxy/test_tools_allowlist_enforcement.py
index 8196cc97f50..31f5fbf606b 100644
--- a/tests/test_litellm/proxy/test_tools_allowlist_enforcement.py
+++ b/tests/test_litellm/proxy/test_tools_allowlist_enforcement.py
@@ -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(
diff --git a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py
index 426c73645c1..9cf60ea5f91 100644
--- a/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py
+++ b/tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py
@@ -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."""
diff --git a/tests/test_litellm/responses/test_custom_tool_call.py b/tests/test_litellm/responses/test_custom_tool_call.py
new file mode 100644
index 00000000000..c605ef24934
--- /dev/null
+++ b/tests/test_litellm/responses/test_custom_tool_call.py
@@ -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"])