scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-04 07:27:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 195

refactor(responses): send bridged chat calls to the Responses executor
Some checks failed
LiteLLM Rust / rust-lint (push) Has been cancelled
Details
LiteLLM Rust / rust-test (push) Has been cancelled
Details
LiteLLM Rust / rust-wheel (push) Has been cancelled
Details

Browse source 
The chat-to-Responses bridge called litellm.responses() from inside the chat
call, so the @client pass and its deployment hooks ran twice per request. It now
hands the translated request to the native Responses executor, shared with the
/v1/messages bridge in litellm/responses/bridge_executor.py, and keeps
responses() for requests that need what only it adds
This commit is contained in:
narmstrong-berri 2026-10-26 03:24:33 +00:00
parent 6883772e3c
commit f9b0fb1c3a
10 changed files with 524 additions and 266 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
litellm/completion_extras/litellm_responses_transformation/handler.py
View file

@ -11,8 +11,10 @@ from typing import (
cast, # noqa: TID251 # optional_params reaches the bridge as an untyped dict
)
from pydantic import TypeAdapter, ValidationError
from typing_extensions import TypedDict
import litellm
from litellm.types.llms.openai import ResponsesAPIResponse
if TYPE_CHECKING:
@ -36,6 +38,48 @@ def _restore_routing_prefix(model: str, custom_llm_provider: str) -> str:
return f"{custom_llm_provider}/{model}"
_TOOLS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
_BRIDGED_REQUEST: Final = TypeAdapter(dict[str, object])
_TOOLS_ONLY_RESPONSES_SERVES: Final = frozenset({"mcp", "file_search"})
def _needs_responses_surface(request_data: Mapping[str, object]) -> bool:
"""`responses()` also checks the cache and dispatches MCP and emulated file_search tools, which the executor skips."""
if litellm.cache is not None or request_data.get("use_chat_completions_api"):
return True
model: Final = request_data.get("model")
if not isinstance(model, str) or model.startswith("openai/chat_completions/"):
return True
try:
tools: Final = _TOOLS.validate_python(request_data.get("tools") or ())
except ValidationError:
return True
return any(tool.get("type") in _TOOLS_ONLY_RESPONSES_SERVES for tool in tools)
def _executor_logging_obj(
request_data: Mapping[str, object], logging_obj: "LiteLLMLoggingObj"
) -> "LiteLLMLoggingObj | None":
from litellm.responses.bridge_executor import has_native_responses_config, native_logging_obj
model: Final = request_data.get("model")
if _needs_responses_surface(request_data) or not isinstance(model, str):
return None
if not has_native_responses_config(model, request_data):
return None
return native_logging_obj(logging_obj, request_data, request_data)
def _retagged_as_responses(
request_data: Mapping[str, object], logging_obj: "LiteLLMLoggingObj"
) -> Mapping[str, object]:
"""`responses()` logs the Responses object it returns, so the shared logging object takes its call type."""
from litellm.types.utils import CallTypes
setattr(logging_obj, "call_type", CallTypes.responses.value)
return request_data
class ResponsesToCompletionBridgeHandler:
def __init__(self):
from .transformation import LiteLLMResponsesTransformationHandler
@ -172,6 +216,7 @@ class ResponsesToCompletionBridgeHandler:
from litellm import responses
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.responses.bridge_executor import native_responses
validated_kwargs: Final = self.validate_input_kwargs(kwargs)
model: Final = validated_kwargs["model"]
@ -199,11 +244,16 @@ class ResponsesToCompletionBridgeHandler:
# `sanitized_litellm_params` already carries both, so passing them again
# would raise a duplicate-keyword TypeError.
request_data["custom_llm_provider"] = custom_llm_provider
request_data["model"] = _restore_routing_prefix(model, custom_llm_provider)
routed_model: Final = _restore_routing_prefix(model, custom_llm_provider)
request_data["model"] = routed_model
if kwargs.get("cache") is not None:
request_data["cache"] = kwargs["cache"]
result: Final = responses(
**request_data,
bridged_request: Final = _BRIDGED_REQUEST.validate_python(request_data)
executor_logging: Final = _executor_logging_obj(bridged_request, logging_obj)
result: Final = (
responses(**_retagged_as_responses(request_data, logging_obj))
if executor_logging is None
else native_responses(bridged_request, routed_model, bridged_request, executor_logging)
)
from litellm.types.utils import ModelResponse
@ -272,6 +322,7 @@ class ResponsesToCompletionBridgeHandler:
async def acompletion(self, *args, **kwargs) -> Union["ModelResponse", "CustomStreamWrapper"]:
from litellm import aresponses
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.responses.bridge_executor import anative_responses
validated_kwargs: Final = self.validate_input_kwargs(kwargs)
model: Final = validated_kwargs["model"]
@ -301,12 +352,16 @@ class ResponsesToCompletionBridgeHandler:
# `sanitized_litellm_params` already carries both, so passing them again
# would raise a duplicate-keyword TypeError.
request_data["custom_llm_provider"] = custom_llm_provider
request_data["model"] = _restore_routing_prefix(model, custom_llm_provider)
routed_model: Final = _restore_routing_prefix(model, custom_llm_provider)
request_data["model"] = routed_model
if kwargs.get("cache") is not None:
request_data["cache"] = kwargs["cache"]
result: Final = await aresponses(
**request_data,
aresponses=True,
bridged_request: Final = _BRIDGED_REQUEST.validate_python(request_data)
executor_logging: Final = _executor_logging_obj(bridged_request, logging_obj)
result: Final = (
await aresponses(**_retagged_as_responses(request_data, logging_obj), aresponses=True)
if executor_logging is None
else await anative_responses(bridged_request, routed_model, bridged_request, executor_logging)
)
from litellm.types.utils import ModelResponse

4
litellm/completion_extras/litellm_responses_transformation/transformation.py
View file

@ -1137,10 +1137,6 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
api_model: Final = model
from litellm.types.utils import CallTypes
setattr(litellm_logging_obj, "call_type", CallTypes.responses.value)
sanitized_litellm_params: Final = self._build_sanitized_litellm_params(litellm_params)
request_data: Final = {

236
litellm/llms/anthropic/pass_through/responses_adapters/handler.py
View file

@ -5,14 +5,11 @@ Used when the target model is an OpenAI or Azure model.
"""
from collections.abc import AsyncIterator, Callable, Coroutine, Mapping, Sequence
from typing import Final, NoReturn, TypeAlias
from typing import Final, TypeAlias
import httpx
from aiohttp import ClientSession
from pydantic import TypeAdapter, ValidationError
import litellm
from litellm.constants import request_timeout
from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
supports_openai_prompt_cache_breakpoint,
@ -20,20 +17,8 @@ from litellm.integrations.anthropic_cache_control_hook import (
from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.prompt_templates.common_utils import cache_control_as_prompt_cache_breakpoint
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.native.responses import execute_responses
from litellm.responses.main import (
apply_managed_file_id_mapping,
base_llm_http_handler,
reasoning_for_deployment,
requested_reasoning,
resolve_model_provider_for_responses,
resolve_responses_api_provider_config,
with_provider_affinity_header,
)
from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
from litellm.responses.utils import ResponsesAPIRequestUtils, auto_reasoning_summary
from litellm.router_utils.reasoning_effort_capability import deployment_reasoning_effort_levels
from litellm.responses.bridge_executor import anative_responses, native_logging_obj, native_responses
from litellm.responses.utils import auto_reasoning_summary
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
from litellm.types.llms.anthropic import (
AllAnthropicMessageValues,
@ -46,7 +31,6 @@ from litellm.types.llms.anthropic_messages.anthropic_response import (
AnthropicMessagesResponse,
)
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import CallTypes
from litellm.utils import ProviderConfigManager
@ -62,9 +46,6 @@ BlockRewrite: TypeAlias = Callable[[Mapping[str, object]], Mapping[str, object]]
_BLOCK_LIST: Final = TypeAdapter(list[dict[str, object]])
_ADAPTER: Final = LiteLLMAnthropicToResponsesAPIAdapter()
_STRING_KEYED_DICT: Final = TypeAdapter(dict[str, object])
_CLIENT_HEADERS: Final = TypeAdapter(dict[str, str])
_STRINGS: Final = TypeAdapter(tuple[str, ...])
_NEVER_FORWARDED: Final = CONVERTED_STREAM_KEYS | {"anthropic_messages"}
@ -258,156 +239,6 @@ def _build_responses_request(
return responses_request
def _optional_str(value: object) -> str | None:
return value if isinstance(value, str) else None
def _string_keyed_dict(value: object) -> dict[str, object] | None:
try:
return _STRING_KEYED_DICT.validate_python(value) if value is not None else None
except ValidationError:
return None
def _client_headers(value: object) -> Mapping[str, str] | None:
try:
return _CLIENT_HEADERS.validate_python(value) if isinstance(value, dict) else None
except ValidationError:
return None
def _shared_session(value: object) -> ClientSession | None:
return value if isinstance(value, ClientSession) else None
def _http_client(value: object) -> HTTPHandler | AsyncHTTPHandler | None:
return value if isinstance(value, (HTTPHandler, AsyncHTTPHandler)) else None
def _timeout(value: object) -> float | httpx.Timeout:
return value if isinstance(value, (float, int, httpx.Timeout)) and value else request_timeout
def _allowed_openai_params(value: object) -> Sequence[str] | None:
try:
return _STRINGS.validate_python(value) if value is not None else None
except ValidationError:
return None
def _native_responses_call(
request: Mapping[str, object],
model: str,
forwarded_kwargs: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
is_async: bool,
) -> (
ResponsesAPIResponse
| BaseResponsesAPIStreamingIterator
| Coroutine[object, object, ResponsesAPIResponse | BaseResponsesAPIStreamingIterator]
):
"""Resolve the deployment and hand the translated request to the native Responses executor."""
litellm_params: Final = GenericLiteLLMParams(**forwarded_kwargs)
provider_model, custom_llm_provider = resolve_model_provider_for_responses(
model=model,
custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")),
litellm_params=litellm_params,
local_vars={},
)
if custom_llm_provider is None:
raise ValueError(f"No provider resolved for {model}")
provider_config: Final = resolve_responses_api_provider_config(
provider_model, custom_llm_provider, forwarded_kwargs.get("model_info"), litellm_params.api_base
)
if provider_config is None:
raise ValueError(f"{custom_llm_provider} has no Responses API config for {provider_model}")
request_input, request_tools = apply_managed_file_id_mapping(
input=request["input"],
tools=request.get("tools"),
kwargs=dict(forwarded_kwargs),
local_vars={},
)
params: Final = {
**forwarded_kwargs,
**request,
"input": request_input,
"tools": request_tools,
"custom_llm_provider": custom_llm_provider,
}
deployment_model_info: Final = forwarded_kwargs.get("model_info")
reasoning: Final = reasoning_for_deployment(
requested_reasoning(params, provider_model, custom_llm_provider),
provider_model,
custom_llm_provider,
deployment_model_info,
forwarded_kwargs,
)
return execute_responses(
http_handler=base_llm_http_handler,
model=provider_model,
input=request_input,
provider_config=provider_config,
requested_params=ResponsesAPIRequestUtils.get_requested_response_api_optional_param(
{key: value for key, value in {**params, "reasoning": reasoning}.items() if key != "reasoning_effort"}
),
allowed_openai_params=_allowed_openai_params(forwarded_kwargs.get("allowed_openai_params")),
declared_reasoning_efforts=deployment_reasoning_effort_levels(deployment_model_info) or (),
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=logging_obj,
user=_optional_str(request.get("user")),
extra_headers=with_provider_affinity_header(
ResponsesAPIRequestUtils.merge_client_forwarded_headers(
extra_headers=_string_keyed_dict(forwarded_kwargs.get("extra_headers")),
client_headers=_client_headers(forwarded_kwargs.get("headers")),
),
litellm_params,
_string_keyed_dict(request.get("metadata")),
forwarded_kwargs,
provider_model,
custom_llm_provider,
),
extra_body=_string_keyed_dict(forwarded_kwargs.get("extra_body")),
timeout=_timeout(forwarded_kwargs.get("timeout")),
is_async=is_async,
stream=request.get("stream") is True,
client=_http_client(forwarded_kwargs.get("client")),
shared_session=_shared_session(forwarded_kwargs.get("shared_session")),
litellm_metadata=_string_keyed_dict(forwarded_kwargs.get("litellm_metadata")) or {},
request_kwargs=dict(forwarded_kwargs),
)
_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX: Final = "openai/chat_completions/"
def _requests_chat_completions_api(params: Mapping[str, object]) -> bool:
model: Final = params.get("model")
return bool(params.get("use_chat_completions_api")) or (
isinstance(model, str) and model.startswith(_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX)
)
def _native_logging_obj(
logging_obj: LiteLLMLoggingObj | None, forwarded_kwargs: Mapping[str, object], request: Mapping[str, object]
) -> LiteLLMLoggingObj | None:
"""The logging object when the native executor can serve the call, None when only `aresponses` can.
The executor skips the prompt management hooks `aresponses` runs, so a call that needs them re-enters it.
So does a deployment that asks for the chat completions API, which `aresponses` bridges to chat.
"""
if logging_obj is None:
return None
params: Final = {**forwarded_kwargs, **request}
if _requests_chat_completions_api(params):
return None
if logging_obj.should_run_prompt_management_hooks(
non_default_params=params, prompt_id=_optional_str(params.get("prompt_id"))
):
return None
return logging_obj
def _reentrant_kwargs(
forwarded_kwargs: Mapping[str, object], request: Mapping[str, object], logging_obj: LiteLLMLoggingObj | None
) -> dict[str, object]:
@ -416,55 +247,6 @@ def _reentrant_kwargs(
return {**forwarded_kwargs, **request}
def _raise_mapped(error: Exception, model: str, forwarded_kwargs: Mapping[str, object]) -> NoReturn:
raise litellm.exception_type(
model=model,
custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")),
original_exception=error,
extra_kwargs=dict(forwarded_kwargs),
) from error
def _priced(
response: ResponsesAPIResponse | BaseResponsesAPIStreamingIterator, logging_obj: LiteLLMLoggingObj
) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
"""The response, priced as `aresponses` would have priced it before the bridge translates it."""
if isinstance(response, ResponsesAPIResponse):
logging_obj.price_bridged_provider_response(response)
else:
logging_obj.price_bridged_provider_stream(response)
return response
def _native_responses(
request: Mapping[str, object],
model: str,
forwarded_kwargs: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
try:
response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=False)
except Exception as e: # noqa: BLE001 # every provider failure maps to the litellm exception aresponses would raise
_raise_mapped(e, model, forwarded_kwargs)
if isinstance(response, Coroutine):
raise TypeError("The sync Responses executor returned a coroutine")
return _priced(response, logging_obj)
async def _native_aresponses(
request: Mapping[str, object],
model: str,
forwarded_kwargs: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
try:
response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=True)
settled: Final = await response if isinstance(response, Coroutine) else response
except Exception as e: # noqa: BLE001 # every provider failure maps to the litellm exception aresponses would raise
_raise_mapped(e, model, forwarded_kwargs)
return _priced(settled, logging_obj)
class LiteLLMMessagesToResponsesAPIHandler:
"""
Handles Anthropic /v1/messages requests for OpenAI / Azure models by
@ -515,11 +297,11 @@ class LiteLLMMessagesToResponsesAPIHandler:
logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
safeguards_evaluator: Final = request_safeguards_evaluator(kwargs, messages)
native_logging_obj: Final = _native_logging_obj(logging_obj, forwarded_kwargs, request)
native_logging: Final = native_logging_obj(logging_obj, forwarded_kwargs, request)
result: Final = (
await litellm.aresponses(**_reentrant_kwargs(forwarded_kwargs, request, logging_obj))
if native_logging_obj is None
else await _native_aresponses(request, model, forwarded_kwargs, native_logging_obj)
if native_logging is None
else await anative_responses(request, model, forwarded_kwargs, native_logging)
)
if stream:
@ -605,11 +387,11 @@ class LiteLLMMessagesToResponsesAPIHandler:
)
forwarded_kwargs: Final = _forwarded_kwargs(kwargs)
logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
native_logging_obj: Final = _native_logging_obj(logging_obj, forwarded_kwargs, request)
native_logging: Final = native_logging_obj(logging_obj, forwarded_kwargs, request)
result: Final = (
litellm.responses(**_reentrant_kwargs(forwarded_kwargs, request, logging_obj))
if native_logging_obj is None
else _native_responses(request, model, forwarded_kwargs, native_logging_obj)
if native_logging is None
else native_responses(request, model, forwarded_kwargs, native_logging)
)
if stream:

246
litellm/responses/bridge_executor.py Normal file
View file

@ -0,0 +1,246 @@
"""Hands a bridged Responses request to the native Responses executor, so the bridge never re-enters `aresponses`."""
from collections.abc import Coroutine, Mapping, Sequence
from typing import Final, NoReturn
import httpx
from aiohttp import ClientSession
from pydantic import TypeAdapter, ValidationError
import litellm
from litellm.constants import request_timeout
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.native.responses import execute_responses
from litellm.responses.main import (
apply_managed_file_id_mapping,
base_llm_http_handler,
reasoning_for_deployment,
requested_reasoning,
resolve_model_provider_for_responses,
resolve_responses_api_provider_config,
with_provider_affinity_header,
)
from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.router_utils.reasoning_effort_capability import deployment_reasoning_effort_levels
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.router import GenericLiteLLMParams
__all__ = ("anative_responses", "has_native_responses_config", "native_logging_obj", "native_responses")
_STRING_KEYED_DICT: Final = TypeAdapter(dict[str, object])
_CLIENT_HEADERS: Final = TypeAdapter(dict[str, str])
_STRINGS: Final = TypeAdapter(tuple[str, ...])
def _optional_str(value: object) -> str | None:
return value if isinstance(value, str) else None
def _string_keyed_dict(value: object) -> dict[str, object] | None:
try:
return _STRING_KEYED_DICT.validate_python(value) if value is not None else None
except ValidationError:
return None
def _client_headers(value: object) -> Mapping[str, str] | None:
try:
return _CLIENT_HEADERS.validate_python(value) if isinstance(value, dict) else None
except ValidationError:
return None
def _shared_session(value: object) -> ClientSession | None:
return value if isinstance(value, ClientSession) else None
def _http_client(value: object) -> HTTPHandler | AsyncHTTPHandler | None:
return value if isinstance(value, (HTTPHandler, AsyncHTTPHandler)) else None
def _timeout(value: object) -> float | httpx.Timeout:
return value if isinstance(value, (float, int, httpx.Timeout)) and value else request_timeout
def _allowed_openai_params(value: object) -> Sequence[str] | None:
try:
return _STRINGS.validate_python(value) if value is not None else None
except ValidationError:
return None
def _responses_target(
model: str, forwarded_kwargs: Mapping[str, object], litellm_params: GenericLiteLLMParams
) -> tuple[str, str, BaseResponsesAPIConfig] | None:
provider_model, custom_llm_provider = resolve_model_provider_for_responses(
model=model,
custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")),
litellm_params=litellm_params,
local_vars={},
)
if custom_llm_provider is None:
return None
provider_config: Final = resolve_responses_api_provider_config(
provider_model, custom_llm_provider, forwarded_kwargs.get("model_info"), litellm_params.api_base
)
return None if provider_config is None else (provider_model, custom_llm_provider, provider_config)
def has_native_responses_config(model: str, forwarded_kwargs: Mapping[str, object]) -> bool:
return _responses_target(model, forwarded_kwargs, GenericLiteLLMParams(**forwarded_kwargs)) is not None
def _native_responses_call(
request: Mapping[str, object],
model: str,
forwarded_kwargs: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
is_async: bool,
) -> (
ResponsesAPIResponse
| BaseResponsesAPIStreamingIterator
| Coroutine[object, object, ResponsesAPIResponse | BaseResponsesAPIStreamingIterator]
):
"""Resolve the deployment and hand the translated request to the native Responses executor."""
litellm_params: Final = GenericLiteLLMParams(**forwarded_kwargs)
target: Final = _responses_target(model, forwarded_kwargs, litellm_params)
if target is None:
raise ValueError(f"No Responses API config for {model}")
provider_model, custom_llm_provider, provider_config = target
request_input, request_tools = apply_managed_file_id_mapping(
input=request["input"],
tools=request.get("tools"),
kwargs=dict(forwarded_kwargs),
local_vars={},
)
params: Final = {
**forwarded_kwargs,
**request,
"input": request_input,
"tools": request_tools,
"custom_llm_provider": custom_llm_provider,
}
deployment_model_info: Final = forwarded_kwargs.get("model_info")
reasoning: Final = reasoning_for_deployment(
requested_reasoning(params, provider_model, custom_llm_provider),
provider_model,
custom_llm_provider,
deployment_model_info,
forwarded_kwargs,
)
return execute_responses(
http_handler=base_llm_http_handler,
model=provider_model,
input=request_input,
provider_config=provider_config,
requested_params=ResponsesAPIRequestUtils.get_requested_response_api_optional_param(
{key: value for key, value in {**params, "reasoning": reasoning}.items() if key != "reasoning_effort"}
),
allowed_openai_params=_allowed_openai_params(forwarded_kwargs.get("allowed_openai_params")),
declared_reasoning_efforts=deployment_reasoning_effort_levels(deployment_model_info) or (),
custom_llm_provider=custom_llm_provider,
litellm_params=litellm_params,
logging_obj=logging_obj,
user=_optional_str(request.get("user")),
extra_headers=with_provider_affinity_header(
ResponsesAPIRequestUtils.merge_client_forwarded_headers(
extra_headers=_string_keyed_dict(forwarded_kwargs.get("extra_headers")),
client_headers=_client_headers(forwarded_kwargs.get("headers")),
),
litellm_params,
_string_keyed_dict(request.get("metadata")),
forwarded_kwargs,
provider_model,
custom_llm_provider,
),
extra_body=_string_keyed_dict(forwarded_kwargs.get("extra_body")),
timeout=_timeout(forwarded_kwargs.get("timeout")),
is_async=is_async,
stream=request.get("stream") is True,
client=_http_client(forwarded_kwargs.get("client")),
shared_session=_shared_session(forwarded_kwargs.get("shared_session")),
litellm_metadata=_string_keyed_dict(forwarded_kwargs.get("litellm_metadata")) or {},
request_kwargs=dict(forwarded_kwargs),
)
_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX: Final = "openai/chat_completions/"
def _requests_chat_completions_api(params: Mapping[str, object]) -> bool:
model: Final = params.get("model")
return bool(params.get("use_chat_completions_api")) or (
isinstance(model, str) and model.startswith(_OPENAI_CHAT_COMPLETIONS_MODEL_PREFIX)
)
def native_logging_obj(
logging_obj: LiteLLMLoggingObj | None, forwarded_kwargs: Mapping[str, object], request: Mapping[str, object]
) -> LiteLLMLoggingObj | None:
"""The logging object when the native executor can serve the call, None when only `aresponses` can.
The executor skips the prompt management hooks `aresponses` runs, so a call that needs them re-enters it.
So does a deployment that asks for the chat completions API, which `aresponses` bridges to chat.
"""
if logging_obj is None:
return None
params: Final = {**forwarded_kwargs, **request}
if _requests_chat_completions_api(params):
return None
if logging_obj.should_run_prompt_management_hooks(
non_default_params=params, prompt_id=_optional_str(params.get("prompt_id"))
):
return None
return logging_obj
def _raise_mapped(error: Exception, model: str, forwarded_kwargs: Mapping[str, object]) -> NoReturn:
raise litellm.exception_type(
model=model,
custom_llm_provider=_optional_str(forwarded_kwargs.get("custom_llm_provider")),
original_exception=error,
extra_kwargs=dict(forwarded_kwargs),
) from error
def _priced(
response: ResponsesAPIResponse | BaseResponsesAPIStreamingIterator, logging_obj: LiteLLMLoggingObj
) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
"""The response, priced as `aresponses` would have priced it before the bridge translates it."""
if isinstance(response, ResponsesAPIResponse):
logging_obj.price_bridged_provider_response(response)
else:
logging_obj.price_bridged_provider_stream(response)
return response
def native_responses(
request: Mapping[str, object],
model: str,
forwarded_kwargs: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
try:
response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=False)
except Exception as e: # noqa: BLE001 # every provider failure maps to the litellm exception aresponses would raise
_raise_mapped(e, model, forwarded_kwargs)
if isinstance(response, Coroutine):
raise TypeError("The sync Responses executor returned a coroutine")
return _priced(response, logging_obj)
async def anative_responses(
request: Mapping[str, object],
model: str,
forwarded_kwargs: Mapping[str, object],
logging_obj: LiteLLMLoggingObj,
) -> ResponsesAPIResponse | BaseResponsesAPIStreamingIterator:
try:
response: Final = _native_responses_call(request, model, forwarded_kwargs, logging_obj, is_async=True)
settled: Final = await response if isinstance(response, Coroutine) else response
except Exception as e: # noqa: BLE001 # every provider failure maps to the litellm exception aresponses would raise
_raise_mapped(e, model, forwarded_kwargs)
return _priced(settled, logging_obj)

121
tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py
View file

@ -1,15 +1,19 @@
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
import respx
import litellm
from litellm.caching.caching import Cache
from litellm.completion_extras.litellm_responses_transformation.handler import (
ResponsesToCompletionBridgeHandler,
)
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.responses import bridge_executor
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import ModelResponse
@ -41,6 +45,11 @@ def test_is_preformatted_cached_chat_stream_false_wrong_type():
)
@pytest.fixture
def _cache_on(monkeypatch):
monkeypatch.setattr(litellm, "cache", MagicMock(spec=Cache))
def _bridge_kwargs(stream: bool):
logging_obj = LiteLLMLogging(
litellm_call_id="test-call",
@ -63,7 +72,7 @@ def _bridge_kwargs(stream: bool):
}
def test_completion_returns_cached_model_response_directly():
def test_completion_returns_cached_model_response_directly(_cache_on):
"""Non-streaming bridge cache hit: responses() returns a ModelResponse -> bridge returns it as-is."""
cached = ModelResponse(id="chatcmpl-cached-nonstream", model="gpt-5.4")
bridge = ResponsesToCompletionBridgeHandler()
@ -82,7 +91,7 @@ def test_completion_returns_cached_model_response_directly():
@pytest.mark.asyncio
async def test_acompletion_returns_cached_model_response_directly():
async def test_acompletion_returns_cached_model_response_directly(_cache_on):
cached = ModelResponse(id="chatcmpl-cached-nonstream-async", model="gpt-5.4")
bridge = ResponsesToCompletionBridgeHandler()
@ -99,7 +108,7 @@ async def test_acompletion_returns_cached_model_response_directly():
assert result is cached
def test_completion_skips_rewrapping_preformatted_cached_chat_stream():
def test_completion_skips_rewrapping_preformatted_cached_chat_stream(_cache_on):
"""Streaming bridge cache hit returning CustomStreamWrapper(cached_response) -> bridge skips re-wrapping."""
stream = MagicMock(spec=CustomStreamWrapper)
stream.custom_llm_provider = "cached_response"
@ -124,7 +133,7 @@ def test_completion_skips_rewrapping_preformatted_cached_chat_stream():
assert result is stream
def test_completion_preserves_top_level_stream_flag_in_responses_request():
def test_completion_preserves_top_level_stream_flag_in_responses_request(_cache_on):
stream = MagicMock(spec=CustomStreamWrapper)
stream.custom_llm_provider = "cached_response"
bridge = ResponsesToCompletionBridgeHandler()
@ -152,7 +161,7 @@ def test_completion_preserves_top_level_stream_flag_in_responses_request():
@pytest.mark.asyncio
async def test_acompletion_skips_rewrapping_preformatted_cached_chat_stream():
async def test_acompletion_skips_rewrapping_preformatted_cached_chat_stream(_cache_on):
stream = MagicMock(spec=CustomStreamWrapper)
stream.custom_llm_provider = "cached_response"
bridge = ResponsesToCompletionBridgeHandler()
@ -177,7 +186,7 @@ async def test_acompletion_skips_rewrapping_preformatted_cached_chat_stream():
@pytest.mark.asyncio
async def test_acompletion_preserves_top_level_stream_flag_in_responses_request():
async def test_acompletion_preserves_top_level_stream_flag_in_responses_request(_cache_on):
stream = MagicMock(spec=CustomStreamWrapper)
stream.custom_llm_provider = "cached_response"
bridge = ResponsesToCompletionBridgeHandler()
@ -219,7 +228,7 @@ def _completed_chat_response() -> ModelResponse:
@pytest.mark.asyncio
async def test_acompletion_streams_completed_model_response():
async def test_acompletion_streams_completed_model_response(_cache_on):
"""A streaming request whose bridge call comes back already completed must still be
handed back as an async-iterable stream. Returning the bare ModelResponse crashed the
proxy's SSE generator with "'async for' requires an object with __aiter__ method".
@ -245,7 +254,7 @@ async def test_acompletion_streams_completed_model_response():
assert [c for c in chunks if c.choices[0].finish_reason], "stream never emitted a finish_reason"
def test_completion_streams_completed_model_response():
def test_completion_streams_completed_model_response(_cache_on):
completed = _completed_chat_response()
bridge = ResponsesToCompletionBridgeHandler()
@ -288,7 +297,7 @@ def _upstream_model_for(handed_model: str, custom_llm_provider: str) -> str:
def test_completion_keeps_provider_native_model_id_through_responses(
custom_llm_provider, bridge_model, expected_upstream_model
):
"""responses() resolves the provider itself, so the bridge must not hand it an already-stripped model."""
"""The Responses executor resolves the provider itself, so the bridge must not hand it an already-stripped model."""
cached = ModelResponse(id="chatcmpl-cached", model=bridge_model)
bridge = ResponsesToCompletionBridgeHandler()
kwargs = _bridge_kwargs(stream=False)
@ -301,11 +310,11 @@ def test_completion_keeps_provider_native_model_id_through_responses(
"transform_request",
return_value={"model": bridge_model, "input": "hi"},
),
patch("litellm.responses", return_value=cached) as responses_call,
patch.object(bridge_executor, "native_responses", return_value=cached) as responses_call,
):
bridge.completion(**kwargs)
handed_model = responses_call.call_args.kwargs["model"]
handed_model = responses_call.call_args.args[1]
assert _upstream_model_for(handed_model, custom_llm_provider) == expected_upstream_model
@ -329,9 +338,93 @@ async def test_acompletion_keeps_provider_native_model_id_through_responses(
"transform_request",
return_value={"model": bridge_model, "input": "hi"},
),
patch("litellm.aresponses", new=AsyncMock(return_value=cached)) as responses_call,
patch.object(bridge_executor, "anative_responses", new=AsyncMock(return_value=cached)) as responses_call,
):
await bridge.acompletion(**kwargs)
handed_model = responses_call.call_args.kwargs["model"]
handed_model = responses_call.call_args.args[1]
assert _upstream_model_for(handed_model, custom_llm_provider) == expected_upstream_model
class _DeploymentHookCounter(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.calls = 0
async def async_pre_call_deployment_hook(self, kwargs, call_type) -> None:
self.calls += 1
@pytest.mark.asyncio
async def test_bridged_chat_call_runs_deployment_hooks_once(monkeypatch):
counter = _DeploymentHookCounter()
monkeypatch.setattr(litellm, "callbacks", [counter])
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
litellm.in_memory_llm_clients_cache.flush_cache()
litellm.register_model(
{"openai/bridge-only-model": {"litellm_provider": "openai", "mode": "responses", "max_tokens": 100}}
)
with respx.mock(assert_all_called=False) as upstream:
responses_route = upstream.post("https://example.invalid/v1/responses").mock(
return_value=httpx.Response(
200,
json={
"id": "resp_bridge",
"object": "response",
"created_at": 1,
"status": "completed",
"model": "bridge-only-model",
"output": [
{
"type": "message",
"id": "msg_bridge",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "from upstream", "annotations": []}],
}
],
"usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
},
)
)
response = await litellm.acompletion(
model="openai/bridge-only-model",
api_base="https://example.invalid/v1",
api_key="x",
messages=[{"role": "user", "content": "hi"}],
)
assert response.choices[0].message.content == "from upstream"
assert responses_route.call_count == 1
assert counter.calls == 1
@pytest.mark.parametrize(
"bridge_model, tools",
[
("gpt-5.4", [{"type": "mcp", "server_label": "docs", "server_url": "litellm_proxy"}]),
("gpt-5.4", [{"type": "file_search", "vector_store_ids": ["vs_1"]}]),
("chat_completions/gpt-5.4", None),
],
)
def test_completion_keeps_responses_surface_for_what_only_it_serves(bridge_model, tools):
cached = ModelResponse(id="chatcmpl-surface", model="gpt-5.4")
bridge = ResponsesToCompletionBridgeHandler()
kwargs = _bridge_kwargs(stream=False)
kwargs["model"] = bridge_model
with (
patch.object(
bridge.transformation_handler,
"transform_request",
return_value={"model": bridge_model, "input": "hi", "tools": tools},
),
patch("litellm.responses", return_value=cached) as responses_call,
patch.object(bridge_executor, "native_responses") as executor_call,
):
result = bridge.completion(**kwargs)
assert result is cached
assert responses_call.call_count == 1
assert executor_call.call_count == 0

17
tests/unit/completion_extras/test_responses_bridge_provider_propagation.py
View file

@ -13,6 +13,7 @@ from litellm.completion_extras.litellm_responses_transformation.handler import (
ResponsesToCompletionBridgeHandler,
)
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.responses import bridge_executor
from litellm.types.utils import ModelResponse
MODEL = "openai.gpt-5.5"
@ -73,11 +74,11 @@ def test_completion_forwards_custom_llm_provider_to_responses():
bridge = ResponsesToCompletionBridgeHandler()
cached = ModelResponse(id="chatcmpl-cached", model="gpt-5.5")
with patch("litellm.responses", return_value=cached) as fake_responses:
with patch.object(bridge_executor, "native_responses", return_value=cached) as fake_responses:
result = bridge.completion(**_openai_kwargs())
assert result is cached
assert fake_responses.call_args.kwargs["custom_llm_provider"] == "openai"
assert fake_responses.call_args.args[2]["custom_llm_provider"] == "openai"
@pytest.mark.asyncio
@ -85,13 +86,13 @@ async def test_acompletion_forwards_custom_llm_provider_to_aresponses():
bridge = ResponsesToCompletionBridgeHandler()
cached = ModelResponse(id="chatcmpl-cached", model="gpt-5.5")
async def _fake_aresponses(**kwargs):
_fake_aresponses.kwargs = kwargs
async def _fake_aresponses(request, model, forwarded_kwargs, logging_obj):
_fake_aresponses.kwargs = forwarded_kwargs
return cached
_fake_aresponses.kwargs = {}
with patch("litellm.aresponses", _fake_aresponses):
with patch.object(bridge_executor, "anative_responses", _fake_aresponses):
result = await bridge.acompletion(**_openai_kwargs())
assert result is cached
@ -103,13 +104,13 @@ async def test_acompletion_forwards_aws_region_name_to_aresponses():
bridge = ResponsesToCompletionBridgeHandler()
cached = ModelResponse(id="chatcmpl-cached", model=MODEL)
async def _fake_aresponses(**kwargs):
_fake_aresponses.kwargs = kwargs
async def _fake_aresponses(request, model, forwarded_kwargs, logging_obj):
_fake_aresponses.kwargs = forwarded_kwargs
return cached
_fake_aresponses.kwargs = {}
with patch("litellm.aresponses", _fake_aresponses):
with patch.object(bridge_executor, "anative_responses", _fake_aresponses):
result = await bridge.acompletion(**_bedrock_mantle_kwargs())
assert result is cached

79
tests/unit/llms/anthropic/pass_through/messages/test_anthropic_experimental_pass_through_messages_handler.py
View file

@ -4,6 +4,7 @@ import json
import os
import unittest.mock
import uuid
from collections.abc import Mapping
from typing import Any, Dict, Final, List
from unittest.mock import AsyncMock, MagicMock, patch
@ -2328,3 +2329,81 @@ def test_native_messages_tool_config_point_marks_last_tool_and_drops_the_key(mon
assert "cache_control" not in captured["params"]["tools"][0]
assert captured["messages"][0]["content"][-1]["cache_control"] == {"type": "ephemeral"}
assert tools == tools_before
_REASONING_RESPONSE: Final = {
"id": "resp_thinking",
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-5.5",
"output": [
{"type": "reasoning", "id": "rs_1", "summary": [{"type": "summary_text", "text": "worked it out"}]},
{
"type": "message",
"id": "msg_1",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": "the answer", "annotations": []}],
},
],
"parallel_tool_calls": True,
"tool_choice": "auto",
"tools": [],
"usage": {"input_tokens": 3, "output_tokens": 4, "total_tokens": 7},
}
class _DeploymentHookCounter(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.calls = 0
async def async_pre_call_deployment_hook(self, kwargs: Mapping[str, object], call_type: object) -> None:
self.calls += 1
@pytest.mark.asyncio
async def test_forced_chat_sends_openai_thinking_to_responses_through_the_chat_bridge(
monkeypatch: pytest.MonkeyPatch,
):
counter: Final = _DeploymentHookCounter()
monkeypatch.setattr(litellm, "callbacks", [counter])
monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", True)
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
litellm.in_memory_llm_clients_cache.flush_cache()
with respx.mock(assert_all_called=False) as upstream:
responses_route: Final = upstream.post("http://openai.test/v1/responses").mock(
return_value=httpx.Response(200, json=_REASONING_RESPONSE)
)
chat_route: Final = upstream.post("http://openai.test/v1/chat/completions")
response: Final = await litellm.anthropic.messages.acreate(
model="openai/gpt-5.5",
api_base="http://openai.test/v1",
api_key="sk-test",
max_tokens=1024,
messages=[
{"role": "user", "content": "weather?"},
{
"role": "assistant",
"content": [
{"type": "text", "text": "let me check"},
{"type": "tool_use", "id": "toolu_1", "name": "get_w", "input": {"c": "a"}},
],
},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "sunny"}]},
],
tools=[{"name": "get_w", "description": "d", "input_schema": {"type": "object", "properties": {}}}],
thinking={"type": "enabled", "budget_tokens": 1024},
)
assert isinstance(response, dict)
assert [block["type"] for block in response["content"]] == ["thinking", "text"]
assert response["content"][1]["text"] == "the answer"
assert (responses_route.call_count, chat_route.call_count) == (1, 0)
assert counter.calls == 1
sent_input: Final = json.loads(responses_route.calls[0].request.content)["input"]
assert [item["type"] for item in sent_input] == ["message", "message", "function_call", "function_call_output"]
assert sent_input[1]["role"] == "assistant"
assert sent_input[3]["output"] == [{"type": "input_text", "text": "sunny"}]

8
tests/unit/llms/azure/test_azure.py
View file

@ -585,7 +585,9 @@ def test_azure_openai_responses_bridge():
litellm.turn_on_debug()
with patch.object(litellm, "responses") as mock_responses:
from litellm.responses import bridge_executor
with patch.object(bridge_executor, "native_responses") as mock_responses:
try:
response = completion(
model="azure/responses/test-azure-computer-use-preview",
@ -599,10 +601,10 @@ def test_azure_openai_responses_bridge():
mock_responses.assert_called_once()
assert (
mock_responses.call_args.kwargs["model"]
mock_responses.call_args.args[0]["model"]
== "azure/test-azure-computer-use-preview"
)
assert mock_responses.call_args.kwargs["custom_llm_provider"] == "azure"
assert mock_responses.call_args.args[0]["custom_llm_provider"] == "azure"
def test_azure_with_content_safety_error():

6
tests/unit/llms/openai/test_openai.py
View file

@ -1933,7 +1933,9 @@ def test_responses_gpt54_with_xhigh_reasoning():
Ensure chat->responses bridge sends the correct request payload for
openai/responses/gpt-5.4 with reasoning_effort="xhigh".
"""
with patch("litellm.responses") as mock_responses:
from litellm.responses import bridge_executor
with patch.object(bridge_executor, "native_responses") as mock_responses:
mock_responses.side_effect = RuntimeError("stop_after_request_build")
with pytest.raises(litellm.APIConnectionError):
@ -1945,7 +1947,7 @@ def test_responses_gpt54_with_xhigh_reasoning():
)
mock_responses.assert_called_once()
request_body = mock_responses.call_args.kwargs
request_body = mock_responses.call_args.args[0]
assert request_body["model"] == "openai/gpt-5.4"

4
tests/unit/test_router/test_router.py
View file

@ -1668,7 +1668,9 @@ def test_arouter_responses_api_bridge():
## CONFIRM BRIDGE IS CALLED
# an empty stream: the Router reads the first chunk before returning, and a stream of mocks never produces one
with patch.object(litellm, "responses", return_value=iter([])) as mock_responses:
from litellm.responses import bridge_executor
with patch.object(bridge_executor, "native_responses", return_value=iter([])) as mock_responses:
result = router.completion(
model="[IP-approved] o3-pro",
messages=[{"role": "user", "content": "Hello, world!"}],