refactor: clean up fresh tech debt in responses bridge check, chatgpt authenticator and image header adapters

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/images/main.py

@@ -68,6 +68,10 @@ from litellm.utils import (
 )
 
 _PROVIDER_ROUTING_HINTS: Final = TypeAdapter(ProviderRoutingHints)
+_OPTIONAL_OBJECT_DICT_ADAPTER: Final = TypeAdapter[dict[str, object] | None](dict[str, object] | None)
+_OPTIONAL_STRING_DICT_ADAPTER: Final = TypeAdapter[dict[str, str] | None](dict[str, str] | None)
+_OBJECT_DICT_ADAPTER: Final = TypeAdapter[dict[str, object]](dict[str, object])
+_STRING_DICT_ADAPTER: Final = TypeAdapter[dict[str, str]](dict[str, str])
 
 # Cache for ImageEditRequestUtils to avoid repeated __getattr__ calls
 _ImageEditRequestUtils_cache: type["ImageEditRequestUtils"] | None = None
@@ -89,8 +93,8 @@ def _chatgpt_image_params(kwargs: Mapping[str, object]) -> GenericLiteLLMParams:
     from litellm.responses.utils import ResponsesAPIRequestUtils
 
     headers: Final = ResponsesAPIRequestUtils.merge_client_forwarded_headers(
-        TypeAdapter[dict[str, object] | None](dict[str, object] | None).validate_python(kwargs.get("extra_headers")),
-        TypeAdapter[dict[str, str] | None](dict[str, str] | None).validate_python(kwargs.get("headers")),
+        _OPTIONAL_OBJECT_DICT_ADAPTER.validate_python(kwargs.get("extra_headers")),
+        _OPTIONAL_STRING_DICT_ADAPTER.validate_python(kwargs.get("headers")),
     )
     return GenericLiteLLMParams.model_validate({**kwargs, "extra_headers": headers})
 
@@ -163,7 +167,7 @@ async def aimage_generation(*args, **kwargs) -> ImageResponse:
             custom_llm_provider=provider_hints.get("custom_llm_provider"),
             api_base=provider_hints.get("api_base"),
             litellm_params=(
-                _chatgpt_image_params(TypeAdapter(dict[str, object]).validate_python(kwargs))
+                _chatgpt_image_params(_OBJECT_DICT_ADAPTER.validate_python(kwargs))
                 if provider_hints.get("custom_llm_provider") == "chatgpt"
                 or (isinstance(model, str) and model.startswith("chatgpt/"))
                 else None
@@ -288,7 +292,7 @@ def image_generation(
             headers.update(extra_headers)
         model_response: ImageResponse = litellm.utils.ImageResponse()
         chatgpt_params: Final = (
-            _chatgpt_image_params(TypeAdapter(dict[str, object]).validate_python({**kwargs, "api_key": api_key}))
+            _chatgpt_image_params(_OBJECT_DICT_ADAPTER.validate_python({**kwargs, "api_key": api_key}))
             if custom_llm_provider == "chatgpt" or (model is not None and model.startswith("chatgpt/"))
             else None
         )
@@ -567,7 +571,7 @@ def image_generation(
                         _chatgpt_image_headers(
                             model,
                             chatgpt_params.model_copy(update={"api_key": dynamic_api_key or api_key}),
-                            TypeAdapter(dict[str, str]).validate_python(headers),
+                            _STRING_DICT_ADAPTER.validate_python(headers),
                         )
                     )
                     if chatgpt_params is not None

litellm/litellm_core_utils/get_provider_specific_headers.py

@@ -12,10 +12,18 @@ _PROVIDER_HEADERS: Final[TypeAdapter[ProviderSpecificHeader | Sequence[ProviderS
 
 class ProviderSpecificHeaderUtils:
     @staticmethod
-    def forwarded_authorization(provider_specific_header: object, custom_llm_provider: str | None) -> str | None:
-        headers: Final = ProviderSpecificHeaderUtils.get_provider_specific_headers(
+    def validated_provider_specific_headers(
+        provider_specific_header: object, custom_llm_provider: str | None
+    ) -> Mapping[str, str]:
+        return ProviderSpecificHeaderUtils.get_provider_specific_headers(
             _PROVIDER_HEADERS.validate_python(provider_specific_header), custom_llm_provider
         )
+
+    @staticmethod
+    def forwarded_authorization(provider_specific_header: object, custom_llm_provider: str | None) -> str | None:
+        headers: Final = ProviderSpecificHeaderUtils.validated_provider_specific_headers(
+            provider_specific_header, custom_llm_provider
+        )
         return next((value for key, value in headers.items() if key.lower() == "authorization"), None)
 
     @staticmethod

litellm/llms/chatgpt/authenticator.py

@@ -66,6 +66,13 @@ def get_cached_authenticator(auth_file: str) -> "Authenticator":
     return Authenticator(auth_file=auth_file)
 
 
+def resolve_chatgpt_authenticator(
+    litellm_params: "Mapping[str, object] | GenericLiteLLMParams | None", default: "Authenticator"
+) -> "Authenticator":
+    auth_file: Final = get_chatgpt_auth_file(litellm_params)
+    return get_cached_authenticator(auth_file) if auth_file else default
+
+
 def _token_refresh_timeout() -> httpx.Timeout:
     configured: Final = get_configured_request_timeout()
     seconds: Final = (

litellm/llms/chatgpt/chat/transformation.py

@@ -6,7 +6,7 @@ from litellm.llms.openai.openai import OpenAIConfig
 from litellm.types.llms.openai import AllMessageValues
 from litellm.types.router import GenericLiteLLMParams
 
-from ..authenticator import Authenticator, get_cached_authenticator, get_chatgpt_auth_file
+from ..authenticator import Authenticator, get_chatgpt_auth_file, resolve_chatgpt_authenticator
 from ..common_utils import (
     GetAccessTokenError,
     chatgpt_client_api_base,
@@ -30,12 +30,6 @@ class ChatGPTConfig(OpenAIConfig):
         super().__init__()
         self.authenticator = authenticator if authenticator is not None else Authenticator()
 
-    def _resolve_authenticator(
-        self, litellm_params: Mapping[str, object] | GenericLiteLLMParams | None
-    ) -> Authenticator:
-        auth_file: Final = get_chatgpt_auth_file(litellm_params)
-        return get_cached_authenticator(auth_file) if auth_file else self.authenticator
-
     @staticmethod
     def _get_access_token_or_raise(authenticator: Authenticator, model: str, llm_provider: str) -> str:
         try:
@@ -63,7 +57,7 @@ class ChatGPTConfig(OpenAIConfig):
             return chatgpt_client_api_base(api_base), client_credential.access_token, custom_llm_provider
         if api_key:
             return api_base or self.api_base_without_login(), api_key, custom_llm_provider
-        authenticator: Final = self._resolve_authenticator(litellm_params)
+        authenticator: Final = resolve_chatgpt_authenticator(litellm_params, self.authenticator)
         dynamic_api_base: Final = authenticator.get_api_base()
         dynamic_api_key: Final = self._get_access_token_or_raise(authenticator, model, custom_llm_provider)
         return dynamic_api_base, dynamic_api_key, custom_llm_provider
@@ -98,7 +92,7 @@ class ChatGPTConfig(OpenAIConfig):
                 ),
                 **strip_chatgpt_client_credential_headers(headers),
             )
-        authenticator: Final = self._resolve_authenticator(litellm_params)
+        authenticator: Final = resolve_chatgpt_authenticator(litellm_params, self.authenticator)
         resolved_api_key: Final = (
             self._get_access_token_or_raise(authenticator, model, "chatgpt")
             if get_chatgpt_auth_file(litellm_params)

litellm/llms/chatgpt/common_utils.py

@@ -6,7 +6,7 @@ import base64
 import json
 import os
 import platform
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping
 from dataclasses import dataclass
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Final, TypeAlias
@@ -18,14 +18,10 @@ from pydantic import JsonValue, TypeAdapter
 
 from litellm.litellm_core_utils.get_provider_specific_headers import ProviderSpecificHeaderUtils
 from litellm.llms.base_llm.chat.transformation import BaseLLMException
-from litellm.types.utils import ProviderSpecificHeader
 
 if TYPE_CHECKING:
     from litellm.types.router import GenericLiteLLMParams
 
-_SCOPED_HEADERS: Final[TypeAdapter[ProviderSpecificHeader | Sequence[ProviderSpecificHeader] | None]] = TypeAdapter(
-    ProviderSpecificHeader | Sequence[ProviderSpecificHeader] | None
-)
 _STRING_HEADERS: Final = TypeAdapter(Mapping[str, str])
 _PARAMS: Final = TypeAdapter(Mapping[str, object])
 
@@ -355,8 +351,8 @@ def get_chatgpt_request_credential(
     normalized: Final = _normalize_litellm_params(params)
     scoped: Final = normalized.get("provider_specific_header")
     extra: Final = normalized.get("extra_headers")
-    headers: Final = ProviderSpecificHeaderUtils.get_provider_specific_headers(
-        provider_specific_header=_SCOPED_HEADERS.validate_python(scoped), custom_llm_provider="chatgpt"
+    headers: Final = ProviderSpecificHeaderUtils.validated_provider_specific_headers(
+        provider_specific_header=scoped, custom_llm_provider="chatgpt"
     )
     return get_chatgpt_client_credential(headers) or get_chatgpt_client_credential(
         _STRING_HEADERS.validate_python(extra) if extra is not None else MappingProxyType({})

litellm/llms/chatgpt/responses/transformation.py

@@ -25,7 +25,7 @@ from litellm.types.llms.openai import (
 from litellm.types.router import GenericLiteLLMParams
 from litellm.types.utils import LlmProviders
 
-from ..authenticator import Authenticator, get_cached_authenticator, get_chatgpt_auth_file
+from ..authenticator import Authenticator, get_chatgpt_auth_file, resolve_chatgpt_authenticator
 from ..common_utils import (
     CHATGPT_API_BASE,
     GetAccessTokenError,
@@ -51,10 +51,6 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
     def custom_llm_provider(self) -> LlmProviders:
         return LlmProviders.CHATGPT
 
-    def _resolve_authenticator(self, litellm_params: GenericLiteLLMParams | None) -> Authenticator:
-        auth_file: Final = get_chatgpt_auth_file(litellm_params)
-        return get_cached_authenticator(auth_file) if auth_file else self.authenticator
-
     def validate_environment(
         self,
         headers: dict[str, str],
@@ -68,7 +64,7 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
                 **get_chatgpt_default_headers(client_credential.access_token, client_credential.account_id, session_id),
                 **strip_chatgpt_client_credential_headers(headers),
             }
-        authenticator: Final = self._resolve_authenticator(litellm_params)
+        authenticator: Final = resolve_chatgpt_authenticator(litellm_params, self.authenticator)
         safe_headers: Final = strip_chatgpt_client_credential_headers(headers)
         configured_api_key: Final = litellm_params.api_key if litellm_params is not None else None
         if configured_api_key and not get_chatgpt_auth_file(litellm_params):

litellm/responses/main.py

@@ -705,9 +705,7 @@ def _api_base_kwarg(kwargs: Mapping[str, object]) -> str | None:
     return api_base if isinstance(api_base, str) else None
 
 
-def _dispatched_model_name(
-    model: str, custom_llm_provider: str, api_base: str | None, litellm_params: GenericLiteLLMParams | None = None
-) -> str:
+def _dispatched_model_name(model: str, custom_llm_provider: str) -> str:
     from litellm.litellm_core_utils.get_llm_provider_logic import resolve_model_and_provider_for_metadata
 
     provider_model, _ = resolve_model_and_provider_for_metadata(model=model, custom_llm_provider=custom_llm_provider)
@@ -720,7 +718,6 @@ def _will_bridge_to_chat_completions(
     use_chat_completions_api: bool,
     model_info: object,
     api_base: str | None,
-    litellm_params: GenericLiteLLMParams | None = None,
 ) -> bool:
     """``_bridges_to_chat_completions`` for callers running before the provider config is resolved.
 
@@ -737,7 +734,7 @@ def _will_bridge_to_chat_completions(
         return True
     return _bridges_to_chat_completions(
         _resolve_responses_api_provider_config(
-            _dispatched_model_name(normalized_model[0], custom_llm_provider, api_base, litellm_params),
+            _dispatched_model_name(normalized_model[0], custom_llm_provider),
             custom_llm_provider,
             model_info,
             api_base,
@@ -761,7 +758,6 @@ def provider_stores_the_conversation(model: str, custom_llm_provider: str | None
         bool(kwargs.get("use_chat_completions_api")),
         kwargs.get("model_info"),
         api_base,
-        litellm_params,
     )
 
 
@@ -943,7 +939,6 @@ async def aresponses(
                     bool(kwargs.get("use_chat_completions_api")),
                     kwargs.get("model_info"),
                     _api_base_kwarg(kwargs),
-                    GenericLiteLLMParams(**kwargs),
                 ),
             ):
                 _stamp_injection_points_for_the_hook(kwargs, model, custom_llm_provider, input, tools)
@@ -1119,7 +1114,6 @@ def _apply_prompt_management_to_responses_call(
                 use_chat_completions_api,
                 kwargs.get("model_info"),
                 _api_base_kwarg(kwargs),
-                GenericLiteLLMParams(**kwargs),
             ),
         ):
             _stamp_injection_points_for_the_hook(kwargs, model, custom_llm_provider, input, tools)