From c2e03ade87cb47adc72d8e12827ffd7c85436386 Mon Sep 17 00:00:00 2001 From: narmstrong-berri Date: Thu, 15 Oct 2026 17:46:46 +0000 Subject: [PATCH] refactor: extract execute_messages and execute_responses native executors The provider-facing tail of anthropic_messages_handler and responses() moves into litellm/native, so the bridges can call a native executor instead of re-entering a public entry point. No behavior change --- .../pass_through/messages/handler.py | 35 ++---- litellm/native/messages.py | 60 +++++++++ litellm/native/responses.py | 115 ++++++++++++++++++ litellm/responses/main.py | 78 ++---------- litellm/responses/utils.py | 16 +-- 5 files changed, 209 insertions(+), 95 deletions(-) create mode 100644 litellm/native/messages.py create mode 100644 litellm/native/responses.py diff --git a/litellm/llms/anthropic/pass_through/messages/handler.py b/litellm/llms/anthropic/pass_through/messages/handler.py index 73030aae623..1a44d3099e6 100644 --- a/litellm/llms/anthropic/pass_through/messages/handler.py +++ b/litellm/llms/anthropic/pass_through/messages/handler.py @@ -28,6 +28,7 @@ from litellm.llms.base_llm.anthropic_messages.transformation import ( from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler +from litellm.native.messages import execute_messages from litellm.native.route_plan import NativeChat, NativeResponses, native_messages_config, plan_messages_route from litellm.types.llms.anthropic import AllAnthropicToolsValues from litellm.types.llms.anthropic_messages.anthropic_request import AnthropicMetadata @@ -47,7 +48,7 @@ from ..safeguards import ( with_safeguard_results, ) from .interceptors import get_messages_interceptors -from .utils import AnthropicMessagesRequestUtils, mock_response, prepare_native_messages +from .utils import mock_response, prepare_native_messages __all__ = ("anthropic_messages", "anthropic_messages_handler") @@ -610,7 +611,6 @@ def anthropic_messages_handler( **kwargs, ) - anthropic_messages_provider_config: Final = route.config if custom_llm_provider is None: raise ValueError( f"custom_llm_provider is required for Anthropic messages, passed in model={model}, custom_llm_provider={custom_llm_provider}" @@ -623,23 +623,11 @@ def anthropic_messages_handler( if marked_tools is not None: AnthropicCacheControlHook.record_gateway_injection(request_kwargs, 1) local_vars.update({**request_kwargs, **({"tools": list(marked_tools)} if marked_tools is not None else {})}) - anthropic_messages_optional_request_params: Final = ( - AnthropicMessagesRequestUtils.get_requested_anthropic_messages_optional_param( - params=local_vars, - model=model, - drop_params=litellm_params.get("drop_params") is True, - custom_llm_provider=custom_llm_provider, - ) - ) - resolved_api_base: Final = ( - dynamic_api_base - if dynamic_api_base is not None and anthropic_messages_provider_config.uses_get_llm_provider_api_base() - else api_base - ) safeguard_kwargs: Final = ( _SAFEGUARD_KWARGS.validate_python(kwargs) if is_async is True and "safeguards" in kwargs else None ) - native_response: Final = base_llm_http_handler.anthropic_messages_handler( + native_response: Final = execute_messages( + http_handler=base_llm_http_handler, model=model, messages=cast( # cast-ok: anthropic messages are dicts at runtime "list[dict[str, object]]", @@ -647,22 +635,23 @@ def anthropic_messages_handler( cast(Sequence[object], messages) # cast-ok: anthropic payloads arrive as untyped dicts ), ), - anthropic_messages_provider_config=anthropic_messages_provider_config, - anthropic_messages_optional_request_params=dict(anthropic_messages_optional_request_params), - _is_async=is_async, - client=client, + provider_config=route.config, + request_params=local_vars, custom_llm_provider=custom_llm_provider, litellm_params=litellm_params, logging_obj=litellm_logging_obj, api_key=api_key, - api_base=resolved_api_base, + api_base=api_base, + dynamic_api_base=dynamic_api_base, + client=client, + is_async=is_async, stream=stream, - kwargs=kwargs, + request_kwargs=kwargs, ) if safeguard_kwargs is None or not isinstance(native_response, Coroutine): return native_response return _with_native_safeguard_results( native_response, - partial(_proxy_classifies, model, safeguard_kwargs.get("safeguards"), anthropic_messages_provider_config), + partial(_proxy_classifies, model, safeguard_kwargs.get("safeguards"), route.config), partial(_native_safeguards_evaluator, safeguard_kwargs, messages), ) diff --git a/litellm/native/messages.py b/litellm/native/messages.py new file mode 100644 index 00000000000..a7faa4e216a --- /dev/null +++ b/litellm/native/messages.py @@ -0,0 +1,60 @@ +"""Sends an Anthropic Messages request to a provider that speaks Messages natively.""" + +from collections.abc import AsyncIterator, Coroutine, Mapping +from typing import Final + +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.anthropic.pass_through.messages.utils import AnthropicMessagesRequestUtils +from litellm.llms.base_llm.anthropic_messages.transformation import BaseAnthropicMessagesConfig +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler +from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler +from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse +from litellm.types.router import GenericLiteLLMParams + +__all__ = ("execute_messages",) + + +def execute_messages( + *, + http_handler: BaseLLMHTTPHandler, + model: str, + messages: list[dict[str, object]], + provider_config: BaseAnthropicMessagesConfig, + request_params: Mapping[str, object], + custom_llm_provider: str, + litellm_params: GenericLiteLLMParams, + logging_obj: LiteLLMLoggingObj, + api_key: str | None, + api_base: str | None, + dynamic_api_base: str | None, + client: AsyncHTTPHandler | None, + is_async: bool, + stream: bool | None, + request_kwargs: dict[str, object], +) -> AnthropicMessagesResponse | Coroutine[object, object, AnthropicMessagesResponse | AsyncIterator[bytes]]: + optional_params: Final = AnthropicMessagesRequestUtils.get_requested_anthropic_messages_optional_param( + params=request_params, + model=model, + drop_params=litellm_params.get("drop_params") is True, + custom_llm_provider=custom_llm_provider, + ) + resolved_api_base: Final = ( + dynamic_api_base + if dynamic_api_base is not None and provider_config.uses_get_llm_provider_api_base() + else api_base + ) + return http_handler.anthropic_messages_handler( + model=model, + messages=messages, + anthropic_messages_provider_config=provider_config, + anthropic_messages_optional_request_params=dict(optional_params), + _is_async=is_async, + client=client, + custom_llm_provider=custom_llm_provider, + litellm_params=litellm_params, + logging_obj=logging_obj, + api_key=api_key, + api_base=resolved_api_base, + stream=stream, + kwargs=request_kwargs, + ) diff --git a/litellm/native/responses.py b/litellm/native/responses.py new file mode 100644 index 00000000000..9b359ec8e1d --- /dev/null +++ b/litellm/native/responses.py @@ -0,0 +1,115 @@ +"""Sends a Responses API request to a provider that speaks Responses natively.""" + +from collections.abc import Coroutine, Mapping, Sequence +from typing import TYPE_CHECKING, Final + +import httpx + +from litellm.litellm_core_utils.completion_timeout import CompletionTimeout, StreamKind +from litellm.litellm_core_utils.core_helpers import normalize_drop_params +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.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler +from litellm.llms.openai.data_residency import infer_openai_data_residency +from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator +from litellm.responses.utils import ResponsesAPIRequestUtils +from litellm.types.llms.openai import ResponseInputParam, ResponsesAPIOptionalRequestParams, ResponsesAPIResponse +from litellm.types.router import GenericLiteLLMParams + +if TYPE_CHECKING: + from aiohttp import ClientSession + +__all__ = ("execute_responses",) + + +def execute_responses( + *, + http_handler: BaseLLMHTTPHandler, + model: str, + input: str | ResponseInputParam, + provider_config: BaseResponsesAPIConfig, + requested_params: ResponsesAPIOptionalRequestParams, + allowed_openai_params: Sequence[str] | None, + custom_llm_provider: str, + litellm_params: GenericLiteLLMParams, + logging_obj: LiteLLMLoggingObj, + user: str | None, + extra_headers: dict[str, object] | None, + extra_body: Mapping[str, object] | None, + timeout: float | httpx.Timeout | None, + is_async: bool, + stream: bool | None, + client: HTTPHandler | AsyncHTTPHandler | None, + shared_session: "ClientSession | None", + litellm_metadata: dict[str, object] | None, + request_kwargs: dict[str, object], +) -> ( + ResponsesAPIResponse + | BaseResponsesAPIStreamingIterator + | Coroutine[object, object, ResponsesAPIResponse | BaseResponsesAPIStreamingIterator] +): + request_params: Final = ResponsesAPIRequestUtils.get_optional_params_responses_api( + model=model, + responses_api_provider_config=provider_config, + response_api_optional_params=requested_params, + allowed_openai_params=allowed_openai_params, + drop_params=normalize_drop_params(request_kwargs.get("drop_params")), + ) + + logging_obj.update_from_kwargs( + kwargs=request_kwargs, + model=model, + user=user, + optional_params=dict(request_params), + litellm_params={ + **request_params, + "aresponses": is_async, + "litellm_call_id": request_kwargs.get("litellm_call_id"), + "model_info": request_kwargs.get("model_info"), + "data_residency": infer_openai_data_residency(custom_llm_provider, litellm_params.api_base), + "metadata": ( + request_kwargs["litellm_metadata"] + if "litellm_metadata" in request_kwargs + else request_kwargs.get("metadata") + ), + "provider_affinity_header": litellm_params.provider_affinity_header, + }, + custom_llm_provider=custom_llm_provider, + ) + + fake_stream: Final = provider_config.should_fake_stream( + model=model, stream=stream, custom_llm_provider=custom_llm_provider + ) + applied_timeout: Final = CompletionTimeout.for_stream( + (StreamKind.FAKED if fake_stream else StreamKind.LIVE) if stream is True else StreamKind.NOT_STREAMED, + CompletionTimeout.as_limit(timeout), + None if stream is not True or litellm_params.stream_timeout is None else float(litellm_params.stream_timeout), + ) + logging_obj.record_applied_timeout(applied_timeout) + response: Final = http_handler.response_api_handler( + model=model, + input=ResponsesAPIRequestUtils.restore_encrypted_content_item_ids_in_input(input), + responses_api_provider_config=provider_config, + response_api_optional_request_params=request_params, + custom_llm_provider=custom_llm_provider, + litellm_params=litellm_params, + logging_obj=logging_obj, + extra_headers=extra_headers, + extra_body=extra_body, + timeout=applied_timeout, + _is_async=is_async, + client=client, + fake_stream=fake_stream, + litellm_metadata=litellm_metadata, + shared_session=shared_session, + ) + if not isinstance(response, ResponsesAPIResponse): + return response + stamped: Final = ResponsesAPIRequestUtils.update_responses_api_response_id_with_model_id( + responses_api_response=response, + litellm_metadata=litellm_metadata, + custom_llm_provider=custom_llm_provider, + ) + stamped.hidden_params["custom_llm_provider"] = custom_llm_provider + return stamped diff --git a/litellm/responses/main.py b/litellm/responses/main.py index 26e9fb5e39f..d420d7a885b 100644 --- a/litellm/responses/main.py +++ b/litellm/responses/main.py @@ -28,7 +28,6 @@ from litellm.integrations.anthropic_cache_control_hook import ( configured_injection_points, ) from litellm.litellm_core_utils.asyncify import run_async_function -from litellm.litellm_core_utils.completion_timeout import CompletionTimeout, StreamKind from litellm.litellm_core_utils.core_helpers import normalize_drop_params from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup @@ -40,6 +39,7 @@ from litellm.litellm_core_utils.provider_affinity import add_provider_affinity_h from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler from litellm.llms.openai_like.responses.transformation import OpenAILikeResponsesConfig +from litellm.native.responses import execute_responses from litellm.native.route_plan import NativeChat, plan_responses_route from litellm.responses.litellm_completion_transformation.handler import ( LiteLLMCompletionTransformationHandler, @@ -1398,7 +1398,6 @@ def responses( try: litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj") - litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None) _is_async: Final = kwargs.pop("aresponses", False) is True skip_mcp_handler: Final = kwargs.pop("_skip_mcp_handler", False) require_encrypted_task_support: Final = kwargs.pop("_require_encrypted_task_support", False) is True @@ -1618,81 +1617,30 @@ def responses( **bridge_kwargs, ) - # Get optional parameters for the responses API - request_drop_params: Final = kwargs.get("drop_params") - responses_api_request_params: Final[dict] = ResponsesAPIRequestUtils.get_optional_params_responses_api( - model=model, - responses_api_provider_config=responses_api_provider_config, - response_api_optional_params=response_api_optional_params, - allowed_openai_params=allowed_openai_params, - drop_params=normalize_drop_params(request_drop_params), - ) - - litellm_logging_obj.update_from_kwargs( - kwargs=kwargs, - model=model, - user=user, - optional_params=dict(responses_api_request_params), - litellm_params={ - **responses_api_request_params, - "aresponses": _is_async, - "litellm_call_id": litellm_call_id, - "model_info": kwargs.get("model_info"), - "data_residency": infer_openai_data_residency(custom_llm_provider, litellm_params.api_base), - "metadata": (kwargs["litellm_metadata"] if "litellm_metadata" in kwargs else kwargs.get("metadata")), - "provider_affinity_header": litellm_params.provider_affinity_header, - }, - custom_llm_provider=custom_llm_provider, - ) - - # Decode any litellm-encoded encrypted-content item IDs back to their original IDs - input = ResponsesAPIRequestUtils.restore_encrypted_content_item_ids_in_input(input) - - # Call the handler with _is_async flag instead of directly calling the async handler - if custom_llm_provider is None: + if custom_llm_provider is None or responses_api_provider_config is None: raise ValueError("custom_llm_provider is required but passed as None") - fake_stream: Final = responses_api_provider_config.should_fake_stream( - model=model, stream=stream, custom_llm_provider=custom_llm_provider - ) - applied_timeout: Final = CompletionTimeout.for_stream( - (StreamKind.FAKED if fake_stream else StreamKind.LIVE) if stream is True else StreamKind.NOT_STREAMED, - CompletionTimeout.as_limit(timeout or request_timeout), - None - if stream is not True or litellm_params.stream_timeout is None - else float(litellm_params.stream_timeout), - ) - litellm_logging_obj.record_applied_timeout(applied_timeout) - response = base_llm_http_handler.response_api_handler( + return execute_responses( + http_handler=base_llm_http_handler, model=model, input=input, - responses_api_provider_config=responses_api_provider_config, - response_api_optional_request_params=responses_api_request_params, + provider_config=responses_api_provider_config, + requested_params=response_api_optional_params, + allowed_openai_params=allowed_openai_params, custom_llm_provider=custom_llm_provider, litellm_params=litellm_params, logging_obj=litellm_logging_obj, + user=user, extra_headers=effective_extra_headers, extra_body=extra_body, - timeout=applied_timeout, - _is_async=_is_async, + timeout=timeout or request_timeout, + is_async=_is_async, + stream=stream, client=kwargs.get("client"), - fake_stream=fake_stream, - litellm_metadata=kwargs.get("litellm_metadata", {}), shared_session=kwargs.get("shared_session"), + litellm_metadata=kwargs.get("litellm_metadata", {}), + request_kwargs=kwargs, ) - - # Update the responses_api_response_id with the model_id - if isinstance(response, ResponsesAPIResponse): - response = ResponsesAPIRequestUtils.update_responses_api_response_id_with_model_id( - responses_api_response=response, - litellm_metadata=kwargs.get("litellm_metadata", {}), - custom_llm_provider=custom_llm_provider, - ) - # Stamp custom_llm_provider so callbacks can identify the provider - # (mirrors litellm/main.py:1371 for chat completions) - response.hidden_params["custom_llm_provider"] = custom_llm_provider - - return response except Exception as e: raise litellm.exception_type( model=model, diff --git a/litellm/responses/utils.py b/litellm/responses/utils.py index 8e5e1c1cfee..7719aa84ec2 100644 --- a/litellm/responses/utils.py +++ b/litellm/responses/utils.py @@ -31,7 +31,9 @@ from litellm.types.utils import ( ) -def _apply_nested_drop_params(params: dict[str, object], additional_drop_params: list[str] | None) -> dict[str, object]: +def _apply_nested_drop_params( + params: dict[str, object], additional_drop_params: Sequence[str] | None +) -> dict[str, object]: nested_paths: Final = tuple(path for path in additional_drop_params or () if is_nested_path(path)) return reduce(lambda acc, path: delete_nested_value(acc, path), nested_paths, params) @@ -153,7 +155,7 @@ class ResponsesAPIRequestUtils: @staticmethod def merge_client_forwarded_headers( extra_headers: dict[str, object] | None, - client_headers: dict[str, str] | None, + client_headers: Mapping[str, str] | None, ) -> dict[str, object] | None: """ Merge headers forwarded by the proxy (`headers` kwarg, set when @@ -174,8 +176,8 @@ class ResponsesAPIRequestUtils: @staticmethod def _check_valid_arg( - supported_params: list[str] | None, - non_default_params: dict, + supported_params: Sequence[str] | None, + non_default_params: Mapping[str, object], drop_params: bool | None, custom_llm_provider: str | None, model: str, @@ -200,7 +202,7 @@ class ResponsesAPIRequestUtils: model: str, responses_api_provider_config: BaseResponsesAPIConfig, response_api_optional_params: ResponsesAPIOptionalRequestParams, - allowed_openai_params: list[str] | None = None, + allowed_openai_params: Sequence[str] | None = None, drop_params: bool | None = None, ) -> dict: """ @@ -225,7 +227,7 @@ class ResponsesAPIRequestUtils: non_default_params: Final = cast(dict, response_api_optional_params) # Check for unsupported parameters ResponsesAPIRequestUtils._check_valid_arg( - supported_params=supported_params + (allowed_openai_params or []), + supported_params=[*supported_params, *(allowed_openai_params or ())], non_default_params=non_default_params, drop_params=should_drop_params, custom_llm_provider=responses_api_provider_config.custom_llm_provider, @@ -252,7 +254,7 @@ class ResponsesAPIRequestUtils: return apply_openai_param_overrides( optional_params=params_with_normalized_stream_options, non_default_params=non_default_params, - allowed_openai_params=allowed_openai_params or [], + allowed_openai_params=list(allowed_openai_params or ()), ) @staticmethod