diff --git a/litellm/litellm_core_utils/clientside_credential_handler.py b/litellm/litellm_core_utils/clientside_credential_handler.py index db506e8a8e5..ca774f3b9c8 100644 --- a/litellm/litellm_core_utils/clientside_credential_handler.py +++ b/litellm/litellm_core_utils/clientside_credential_handler.py @@ -82,6 +82,7 @@ def _admin_config_fields_to_clear_on_base_override() -> list[str]: _ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE: Final = _admin_config_fields_to_clear_on_base_override() +UPSTREAM_BOUND_PARAMS: Final = frozenset((*clientside_credential_keys, *_ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE)) def is_clientside_credential(request_kwargs: dict) -> bool: diff --git a/litellm/litellm_core_utils/fallback_utils.py b/litellm/litellm_core_utils/fallback_utils.py index 87888311508..753f576cc7f 100644 --- a/litellm/litellm_core_utils/fallback_utils.py +++ b/litellm/litellm_core_utils/fallback_utils.py @@ -1,85 +1,241 @@ -from typing import Final +import functools +import inspect +from collections.abc import Awaitable, Callable, Generator, Iterator, Mapping +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from types import MappingProxyType +from typing import Final, ParamSpec, TypeAlias, TypeVar + +from pydantic import TypeAdapter import litellm from litellm._logging import verbose_logger from litellm._uuid import uuid +from litellm.constants import INTERNAL_KWARG_PREFIX from litellm.litellm_core_utils.add_retry_fallback_headers import ( add_fallback_headers_to_response, ) +from litellm.litellm_core_utils.asyncify import run_async_function +from litellm.litellm_core_utils.clientside_credential_handler import UPSTREAM_BOUND_PARAMS from litellm.litellm_core_utils.core_helpers import ( filter_internal_params, safe_deep_copy, ) +from litellm.types.utils import CallTypes, CustomPricingLiteLLMParams -from .asyncify import run_async_function +FALLBACK_ATTEMPT_KWARG: Final = f"{INTERNAL_KWARG_PREFIX}fallback_attempt" + +FallbackChain: TypeAlias = tuple[str | dict[str, object], ...] +_FALLBACK_CHAIN: Final = TypeAdapter(FallbackChain) +_REQUEST_KWARGS: Final = TypeAdapter(dict[str, object]) +_NO_OVERRIDES: Final[Mapping[str, object]] = MappingProxyType({}) +_RUNNER_PARAMS: Final = frozenset({"kwargs", "fallbacks", "acompletion", "model", "litellm_logging_obj"}) +_PROVIDER_BOUND_PARAMS: Final = UPSTREAM_BOUND_PARAMS | { + "custom_llm_provider", + "client", + "headers", + "litellm_credential_name", +} +_MODEL_BOUND_PARAMS: Final = frozenset(CustomPricingLiteLLMParams.model_fields) | {"base_model"} +_GLOBAL_FALLBACKS_OFF: Final[ContextVar[bool]] = ContextVar("litellm_global_model_fallbacks_off", default=False) +_P = ParamSpec("_P") +_R = TypeVar("_R") -async def async_completion_with_fallbacks(**kwargs): - """ - Asynchronously attempts completion with fallback models if the primary model fails. +@contextmanager +def without_global_fallbacks() -> Generator[None]: + token: Final = _GLOBAL_FALLBACKS_OFF.set(True) + try: + yield + finally: + _GLOBAL_FALLBACKS_OFF.reset(token) - Args: - **kwargs: Keyword arguments for completion, including: - - model (str): Primary model to use - - fallbacks (List[Union[str, dict]]): List of fallback models/configs - - Other completion parameters - Returns: - ModelResponse: The completion response from the first successful model +async def _awaited(pending: Awaitable[object]) -> object: + return await pending - Raises: - Exception: If all models fail and no response is generated - """ - # Extract and prepare parameters - nested_kwargs: Final = kwargs.pop("kwargs", {}) - original_model: Final = kwargs["model"] - model = original_model - fallbacks: Final = [original_model] + nested_kwargs.pop("fallbacks", []) - kwargs.pop("acompletion", None) # Remove to prevent keyword conflicts - litellm_call_id: Final = str(uuid.uuid4()) - base_kwargs: Final = {**kwargs, **nested_kwargs, "litellm_call_id": litellm_call_id} - # fields to remove - base_kwargs.pop("model", None) # Remove model as it will be set per fallback - litellm_logging_obj: Final = base_kwargs.pop("litellm_logging_obj", None) +def without_global_fallbacks_around(endpoint: Callable[_P, _R]) -> Callable[_P, _R]: + if not inspect.iscoroutinefunction(endpoint): - # Try each fallback model - most_recent_exception_str: str | None = None - for attempted_fallbacks, fallback in enumerate(fallbacks): - try: - completion_kwargs = safe_deep_copy(base_kwargs) - # Handle dictionary fallback configurations - if isinstance(fallback, dict): - fallback_config = safe_deep_copy(dict(fallback)) - model = fallback_config.pop("model", original_model) - completion_kwargs.update(fallback_config) - else: - model = fallback + @functools.wraps(endpoint) + def call(*args: _P.args, **kwargs: _P.kwargs) -> _R: # kwargs-ok: endpoint payload + with without_global_fallbacks(): + return endpoint(*args, **kwargs) - # Filter out internal parameters that shouldn't be sent to provider APIs - completion_kwargs = filter_internal_params(completion_kwargs) + return call - response = await litellm.acompletion( - **completion_kwargs, - model=model, - litellm_logging_obj=litellm_logging_obj, - ) + @functools.wraps(endpoint) + async def acall(*args: _P.args, **kwargs: _P.kwargs) -> object: # kwargs-ok: endpoint payload + with without_global_fallbacks(): + result: Final = endpoint(*args, **kwargs) + return await _awaited(result) if inspect.isawaitable(result) else result - if response is not None: - return add_fallback_headers_to_response( - response=response, - attempted_fallbacks=attempted_fallbacks, - ) + return acall # pyright: ignore[reportReturnType] # a coroutine endpoint's _R is the coroutine acall returns - except Exception as e: - verbose_logger.exception("Fallback attempt failed for model %s: %s", model, e) - most_recent_exception_str = str(e) - continue - raise Exception( - f"{most_recent_exception_str}. All fallback attempts failed. Enable verbose logging with `litellm.set_verbose=True` for details." +def _configured_fallbacks(request_kwargs: Mapping[str, object]) -> object: + if request_kwargs.get(FALLBACK_ATTEMPT_KWARG) is True: + return None + per_call: Final = request_kwargs.get("fallbacks") + if per_call is not None: + return per_call + return None if _GLOBAL_FALLBACKS_OFF.get() else litellm.model_fallbacks + + +def uses_fallbacks(request_kwargs: Mapping[str, object]) -> bool: + return bool(_configured_fallbacks(request_kwargs)) + + +def resolve_fallbacks(request_kwargs: Mapping[str, object]) -> FallbackChain | None: + configured: Final = _configured_fallbacks(request_kwargs) + return _FALLBACK_CHAIN.validate_python(configured) if configured else None + + +@dataclass(frozen=True, slots=True) +class _FallbackAttempt: + model: str + kwargs: Mapping[str, object] + + +@dataclass(frozen=True, slots=True) +class _FallbackPlan: + original_model: str + base_kwargs: Mapping[str, object] + fallbacks: FallbackChain + logging_obj: object + + +def _provider_of(model: str, custom_llm_provider: object) -> str | None: + from litellm.litellm_core_utils.get_llm_provider_logic import resolve_model_and_provider_for_metadata + + try: + return resolve_model_and_provider_for_metadata( + model, custom_llm_provider if isinstance(custom_llm_provider, str) else None + )[1] + except litellm.exceptions.BadRequestError: + return None + + +def _inherited_kwargs( + base_kwargs: Mapping[str, object], original_model: str, model: str, custom_llm_provider: object +) -> Mapping[str, object]: + if model == original_model: + return base_kwargs + requested_provider: Final = base_kwargs.get("custom_llm_provider") + original_provider: Final = _provider_of(original_model, requested_provider) + provider: Final = _provider_of(model, custom_llm_provider) or _provider_of(model, requested_provider) + switches_provider: Final = original_provider is None or original_provider != provider + dropped: Final = (_MODEL_BOUND_PARAMS | _PROVIDER_BOUND_PARAMS) if switches_provider else _MODEL_BOUND_PARAMS + return {key: value for key, value in base_kwargs.items() if key not in dropped} + + +def _fallback_attempt(plan: _FallbackPlan, fallback: str | Mapping[str, object]) -> _FallbackAttempt: + overrides: Final = _NO_OVERRIDES if isinstance(fallback, str) else fallback + model: Final = fallback if isinstance(fallback, str) else str(overrides.get("model", plan.original_model)) + inherited: Final = _inherited_kwargs( + plan.base_kwargs, plan.original_model, model, overrides.get("custom_llm_provider") + ) + request: Final = filter_internal_params(safe_deep_copy({**inherited, **overrides})) + return _FallbackAttempt( + model=model, + kwargs={**request, "model": model, "litellm_logging_obj": plan.logging_obj, FALLBACK_ATTEMPT_KWARG: True}, ) -def completion_with_fallbacks(**kwargs): - return run_async_function(async_function=async_completion_with_fallbacks, **kwargs) +def _fallback_plan(runner_kwargs: Mapping[str, object]) -> _FallbackPlan: + nested_kwargs: Final = _REQUEST_KWARGS.validate_python(runner_kwargs.get("kwargs") or {}) + merged: Final = {**runner_kwargs, **nested_kwargs} + return _FallbackPlan( + original_model=str(runner_kwargs["model"]), + base_kwargs={ + **{key: value for key, value in merged.items() if key not in _RUNNER_PARAMS}, + "litellm_call_id": merged.get("litellm_call_id") or str(uuid.uuid4()), + }, + fallbacks=_FALLBACK_CHAIN.validate_python(nested_kwargs.get("fallbacks") or ()), + logging_obj=merged.get("litellm_logging_obj"), + ) + + +def _fallback_attempts(plan: _FallbackPlan) -> Iterator[_FallbackAttempt]: + return (_fallback_attempt(plan, fallback) for fallback in plan.fallbacks) + + +def _all_attempts_failed(last_error: Exception) -> Exception: + return Exception( + f"{last_error}. All fallback attempts failed. Enable verbose logging with `litellm.set_verbose=True` for details." + ) + + +async def async_completion_with_fallbacks(**kwargs: object) -> object: + plan: Final = _fallback_plan(kwargs) + primary: Final = _fallback_attempt(plan, plan.original_model) + try: + return add_fallback_headers_to_response( + response=await litellm.acompletion(**primary.kwargs), attempted_fallbacks=0 + ) + except Exception as primary_error: # noqa: BLE001 # every failed attempt moves on to the next fallback + verbose_logger.exception("Fallback attempt failed for model %s: %s", primary.model, primary_error) + last_error: Exception = primary_error # rebind-ok: the newest failure names the final error + for attempted_fallbacks, attempt in enumerate(_fallback_attempts(plan), start=1): + try: + return add_fallback_headers_to_response( + response=await litellm.acompletion(**attempt.kwargs), attempted_fallbacks=attempted_fallbacks + ) + except Exception as fallback_error: # noqa: BLE001 # any failed attempt moves on to the next fallback + verbose_logger.exception("Fallback attempt failed for model %s: %s", attempt.model, fallback_error) + last_error = fallback_error + raise _all_attempts_failed(last_error) from last_error + + +def _is_fallback_attempt(request_kwargs: Mapping[str, object], call_type: str) -> bool: + return call_type == CallTypes.completion.value and request_kwargs.get(FALLBACK_ATTEMPT_KWARG) is True + + +def checked_fallback_attempt(request_kwargs: Mapping[str, object], call_type: str, response: object) -> object: + if not _is_fallback_attempt(request_kwargs, call_type): + return response + from litellm.utils import async_post_call_success_deployment_hook + + return run_async_function( + async_post_call_success_deployment_hook, request_kwargs, response, CallTypes.completion, include_v2=True + ) + + +def report_failed_sync_fallback_attempt(request_kwargs: Mapping[str, object], call_type: str, error: Exception) -> None: + if not _is_fallback_attempt(request_kwargs, call_type) or request_kwargs.get("acompletion") is True: + return + from litellm.utils import async_post_call_failure_deployment_hook + + run_async_function(async_post_call_failure_deployment_hook, request_kwargs, error, call_type, include_v2=True) + + +def _completion_with_pre_call_hooks(request_kwargs: Mapping[str, object]) -> object: + from litellm.utils import async_pre_call_deployment_hook + + request: Final = _REQUEST_KWARGS.validate_python( + run_async_function(async_pre_call_deployment_hook, request_kwargs, CallTypes.completion.value, include_v2=True) + ) + return litellm.completion(**request) + + +def completion_with_fallbacks(**kwargs: object) -> object: + plan: Final = _fallback_plan(kwargs) + primary: Final = _fallback_attempt(plan, plan.original_model) + try: + return add_fallback_headers_to_response( + response=_completion_with_pre_call_hooks(primary.kwargs), attempted_fallbacks=0 + ) + except Exception as primary_error: # noqa: BLE001 # every failed attempt moves on to the next fallback + verbose_logger.exception("Fallback attempt failed for model %s: %s", primary.model, primary_error) + last_error: Exception = primary_error # rebind-ok: the newest failure names the final error + for attempted_fallbacks, attempt in enumerate(_fallback_attempts(plan), start=1): + try: + return add_fallback_headers_to_response( + response=_completion_with_pre_call_hooks(attempt.kwargs), attempted_fallbacks=attempted_fallbacks + ) + except Exception as fallback_error: # noqa: BLE001 # any failed attempt moves on to the next fallback + verbose_logger.exception("Fallback attempt failed for model %s: %s", attempt.model, fallback_error) + last_error = fallback_error + raise _all_attempts_failed(last_error) from last_error diff --git a/litellm/main.py b/litellm/main.py index 83ad140e504..57790ce0fbd 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -243,6 +243,8 @@ from .litellm_core_utils.core_helpers import ( from .litellm_core_utils.fallback_utils import ( async_completion_with_fallbacks, completion_with_fallbacks, + resolve_fallbacks, + without_global_fallbacks, ) from .litellm_core_utils.prompt_templates.common_utils import ( add_system_prompt_to_messages, @@ -748,7 +750,6 @@ async def acompletion( - The `completion` function is called using `run_in_executor` to execute synchronously in the event loop. - If `stream` is True, the function returns an async generator that yields completion lines. """ - fallbacks: Final = kwargs.get("fallbacks", None) or litellm.model_fallbacks mock_timeout = kwargs.get("mock_timeout", None) if mock_timeout is True: @@ -765,6 +766,7 @@ async def acompletion( custom_llm_provider = kwargs.get("custom_llm_provider", None) _ = _resolve_control_options(kwargs, model) caller_kwargs: Final = cast("Mapping[str, object]", {**kwargs}) # cast-ok: acompletion takes untyped kwargs + fallbacks: Final = resolve_fallbacks(caller_kwargs) ## PROMPT MANAGEMENT HOOKS ## ######################################################### @@ -922,7 +924,7 @@ async def acompletion( if fallbacks is not None: response = await async_completion_with_fallbacks( - **completion_kwargs, kwargs={"fallbacks": fallbacks, **caller_kwargs} + **completion_kwargs, kwargs={**caller_kwargs, "fallbacks": fallbacks} ) if response is None: raise Exception( @@ -5784,7 +5786,6 @@ def completion( metadata: Final = kwargs.get("metadata", None) model_info: Final = kwargs.get("model_info", None) proxy_server_request: Final = kwargs.get("proxy_server_request", None) - fallbacks = kwargs.get("fallbacks", None) provider_specific_header: Final = cast( ProviderSpecificHeader | Sequence[ProviderSpecificHeader] | None, kwargs.get("provider_specific_header", None), @@ -5917,10 +5918,14 @@ def completion( elif num_retries is not None: max_retries = num_retries logging: Final[LiteLLMLoggingObj] = cast(LiteLLMLoggingObj, litellm_logging_obj) - fallbacks = fallbacks or litellm.model_fallbacks + routing_kwargs: Final = cast("Mapping[str, object]", kwargs) # cast-ok: kwargs is the untyped request dict + fallbacks: Final = resolve_fallbacks(routing_kwargs) if fallbacks is not None: - return completion_with_fallbacks( # pyright: ignore[reportReturnType] # fallback runner is untyped; resolves to ModelResponse|CustomStreamWrapper at runtime - **args + fallback_runner: Final = ( + async_completion_with_fallbacks if acompletion is True else completion_with_fallbacks + ) + return fallback_runner( # pyright: ignore[reportReturnType] # fallback runner is untyped; resolves to ModelResponse|CustomStreamWrapper (or its coroutine) at runtime + **{**args, "kwargs": {**kwargs, "fallbacks": fallbacks}} ) if model_list is not None: deployments: Final = [m["litellm_params"] for m in model_list if m["model_name"] == model] @@ -5940,7 +5945,6 @@ def completion( if deployment_id is not None: # azure llms model = deployment_id custom_llm_provider = "azure" - routing_kwargs: Final = cast("Mapping[str, object]", kwargs) # cast-ok: kwargs is the untyped request dict requested_model: Final = model model, custom_llm_provider, dynamic_api_key, api_base = get_llm_provider( model=model, @@ -9619,12 +9623,13 @@ async def ahealth_check( "batch": ("alist_batches" if custom_llm_provider in LIST_BATCHES_SUPPORTED_PROVIDERS else "acompletion"), }.get("decisions" if is_decisions_model_mode(mode) else mode, "acompletion") if "*" in model: - return await HealthCheckHelpers.ahealth_check_wildcard_models( - model=model, - custom_llm_provider=custom_llm_provider, - model_params=model_params, - litellm_logging_obj=litellm_logging_obj, - ) + with without_global_fallbacks(): + return await HealthCheckHelpers.ahealth_check_wildcard_models( + model=model, + custom_llm_provider=custom_llm_provider, + model_params=model_params, + litellm_logging_obj=litellm_logging_obj, + ) mode_handlers: Final = HealthCheckHelpers.get_mode_handlers( model=model, @@ -9636,7 +9641,8 @@ async def ahealth_check( handler_mode: Final = "decisions" if is_decisions_model_mode(mode) else mode if handler_mode in mode_handlers: - _response: Final = await mode_handlers[handler_mode]() + with without_global_fallbacks(): + _response: Final = await mode_handlers[handler_mode]() _response_headers: Final = cast( # cast-ok: provider headers are stored as a string-keyed mapping Mapping[str, object], (get_hidden_params(_response) or {}).get("headers", {}) or {} ) diff --git a/litellm/proxy/auth/fallback_model_access.py b/litellm/proxy/auth/fallback_model_access.py index 6be863b24a2..db53ad0a871 100644 --- a/litellm/proxy/auth/fallback_model_access.py +++ b/litellm/proxy/auth/fallback_model_access.py @@ -9,6 +9,7 @@ checks for each fallback target before it is attempted. Enabled by default. Opt key, team and user search tool grants instead, the same check /search runs on the requested tool. """ +import inspect from collections.abc import Callable, Mapping from dataclasses import dataclass from typing import Final @@ -67,7 +68,8 @@ async def is_search_tool_authorized_for_token(*, search_tool_name: str, valid_to def _is_search_request(request_kwargs: Mapping[str, object]) -> bool: - return request_kwargs.get("original_generic_function") is asearch + handler: Final = request_kwargs.get("original_generic_function") + return callable(handler) and inspect.unwrap(handler, stop=lambda function: function is asearch) is asearch def _token_in_metadata(metadata: object) -> UserAPIKeyAuth | None: diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index e5bb5011b15..6c766eaafed 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -7428,6 +7428,10 @@ class ProxyConfig: litellm.drop_params = drop_params_flag(value, "litellm_settings.drop_params", verbose_proxy_logger) elif key == "suppress_startup_output": continue + elif key == "model_fallbacks": + verbose_proxy_logger.warning( + "litellm_settings.model_fallbacks has no effect on the proxy, set router_settings.fallbacks instead" + ) else: verbose_proxy_logger.debug( "%s setting litellm.%s=%s%s", diff --git a/litellm/proxy/route_llm_request.py b/litellm/proxy/route_llm_request.py index 5810a2f71ab..2575bdc68ef 100644 --- a/litellm/proxy/route_llm_request.py +++ b/litellm/proxy/route_llm_request.py @@ -15,6 +15,7 @@ from pydantic import BeforeValidator, TypeAdapter, ValidationError import litellm from litellm.litellm_core_utils.core_helpers import as_str_mapping, normalize_drop_params +from litellm.litellm_core_utils.fallback_utils import without_global_fallbacks_around from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.routing_common_utils import is_proxy_admin_request from litellm.litellm_core_utils.total_deadline import enforce_total_timeout_scope @@ -898,6 +899,9 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr } ) + def direct_litellm_endpoint(): # noqa: ANN202 # the endpoint is looked up by name, so its call stays as untyped as before + return without_global_fallbacks_around(getattr(litellm, f"{route_type}")) + team_id: Final = get_team_id_from_data(data) router_model_names: Final = llm_router.registry().names(kind="group_set") if llm_router is not None else [] is_proxy_admin_without_team: Final = team_id is None and is_proxy_admin_request(data) @@ -911,7 +915,7 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr if llm_router is not None: return getattr(llm_router, f"{route_type}")(**data) else: - return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs()) + return direct_litellm_endpoint()(**direct_litellm_kwargs()) elif ( route_type == "acompletion" @@ -966,7 +970,7 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr # If we can't get deployment creds, continue without them pass - return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs()) + return direct_litellm_endpoint()(**direct_litellm_kwargs()) # Skip model-based routing for container operations if route_type in [ "acreate_container", @@ -1006,7 +1010,7 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr else await _authorized_team_file_deployment(llm_router, model_less_team_file_call) ) if team_file_deployment is not None: - return getattr(litellm, f"{route_type}")( + return direct_litellm_endpoint()( **_with_team_deployment_attribution( {**direct_litellm_kwargs(), **team_file_deployment.credentials}, team_file_deployment, @@ -1034,7 +1038,7 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr "aingest", ] and (data.get("model") is None or data.get("model") == ""): # These endpoints don't need a model, use custom_llm_provider directly - return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs()) + return direct_litellm_endpoint()(**direct_litellm_kwargs()) if "model" not in data: raise ProxyMissingRequiredParamError( @@ -1075,7 +1079,7 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr # Check wildcards before checking deployment_names # Priority: 1. Exact model_name match, 2. Wildcard match, 3. deployment_names match if llm_router.settings().router_general_settings.pass_through_all_models: - return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs()) + return direct_litellm_endpoint()(**direct_litellm_kwargs()) elif ( llm_router.settings().default_deployment is not None or len(llm_router.registry().patterns().patterns) > 0 @@ -1130,7 +1134,7 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr return getattr(llm_router, f"{route_type}")(**data) except Exception: # If router fails (e.g., model not found in router), fall back to direct call - return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs()) + return direct_litellm_endpoint()(**direct_litellm_kwargs()) elif _is_a2a_agent_model(data.get("model", "")): from litellm.proxy.agent_endpoints.a2a_routing import ( route_a2a_agent_request, @@ -1142,7 +1146,7 @@ async def _route_request_single_attempt( # noqa: ANN202 # returns unawaited pr # Fall through to raise exception below if result is None elif user_model is not None or route_type == "allm_passthrough_route": - return getattr(litellm, f"{route_type}")(**direct_litellm_kwargs()) + return direct_litellm_endpoint()(**direct_litellm_kwargs()) # if no route found then it's a bad request route_name: Final = ROUTE_ENDPOINT_MAPPING.get(route_type, route_type) diff --git a/litellm/router_python_host.py b/litellm/router_python_host.py index 425d38d6233..25cb23238e9 100644 --- a/litellm/router_python_host.py +++ b/litellm/router_python_host.py @@ -23,6 +23,7 @@ from pydantic import TypeAdapter from typing_extensions import overload import litellm +from litellm.litellm_core_utils.fallback_utils import without_global_fallbacks_around from litellm.router_config import RouterConfig, RouterDynamicConfig, live_dynamic_config if TYPE_CHECKING: @@ -111,7 +112,7 @@ class PythonHost: raise AttributeError(f"litellm has no endpoint {op!r}") if not callable(endpoint): raise TypeError(f"litellm.{op} is not an endpoint") - return endpoint + return without_global_fallbacks_around(endpoint) def sleep(self, seconds: float) -> Awaitable[None]: return asyncio.sleep(seconds) diff --git a/litellm/utils.py b/litellm/utils.py index 1d6a8c34185..2c0dc03563c 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -106,6 +106,12 @@ from litellm.litellm_core_utils.fallback_generalizations import ( match_capability_generalizations, match_fill_missing_generalizations, ) +from litellm.litellm_core_utils.fallback_utils import ( + checked_fallback_attempt, + report_failed_sync_fallback_attempt, + uses_fallbacks, + without_global_fallbacks, +) from litellm.litellm_core_utils.get_model_cost_map import ( GetModelCostMap, fetch_registration_document, @@ -2360,7 +2366,11 @@ def client(original_function): clamp_max_tokens(model, args, kwargs, call_type) # MODEL CALL - result = original_function(*args, **kwargs) + try: + result = original_function(*args, **kwargs) + except Exception as deployment_error: + report_failed_sync_fallback_attempt(kwargs, call_type, deployment_error) + raise end_time = datetime.datetime.now() if _returns_requested_stream(kwargs, call_type, result) or _is_converted_stream_result(result): if "complete_response" in kwargs and kwargs["complete_response"] is True: @@ -2414,17 +2424,18 @@ def client(original_function): original_function=original_function, rules_obj=rules_obj, ) + checked_result: Final = checked_fallback_attempt(kwargs, call_type, result) # [OPTIONAL] ADD TO CACHE _llm_caching_handler.sync_set_cache( - result=result, + result=checked_result, args=args, kwargs=kwargs, ) _update_response_metadata: Final[_ResponseMetadataUpdater] = litellm_utils.update_response_metadata _update_response_metadata( - result=result, + result=checked_result, logging_obj=logging_obj, model=model, kwargs=kwargs, @@ -2434,7 +2445,8 @@ def client(original_function): # LOG SUCCESS - handle streaming success logging in the _next_ object, remove `handle_success` once it's deprecated verbose_logger.info("Wrapper: Completed Call, calling success_handler") - if not is_internal_call.get(): + logged_by_fallback_attempt: Final = call_type == CallTypes.completion.value and uses_fallbacks(kwargs) + if not is_internal_call.get() and not logged_by_fallback_attempt: # Copy the current context to propagate it to the background thread # This is essential for OpenTelemetry span context propagation ctx: Final = contextvars.copy_context() @@ -2442,12 +2454,12 @@ def client(original_function): executor.submit( ctx.run, logging_obj.success_handler, - result, + checked_result, start_time, end_time, ) # RETURN RESULT - return result + return checked_result except Exception as e: if call_type == CallTypes.completion.value: num_retries = kwargs.get("num_retries", None) or litellm.num_retries or None @@ -2546,7 +2558,7 @@ def client(original_function): kwargs["litellm_call_id"] = str(uuid.uuid4()) model: Final[str | None] = args[0] if len(args) > 0 else kwargs.get("model", None) - is_completion_with_fallbacks: Final = kwargs.get("fallbacks") is not None + is_completion_with_fallbacks: Final = bool(kwargs.get("fallbacks")) kwargs.pop("_is_litellm_internal_call", None) # discard if injected _is_litellm_internal_call: Final = is_internal_call.get() _deployment_call_end_time: datetime.datetime | None = None @@ -8501,7 +8513,8 @@ def check_valid_key(model: str, api_key: str): """ messages: Final = [{"role": "user", "content": "Hey, how's it going?"}] try: - litellm.completion(model=model, messages=messages, api_key=api_key, max_tokens=10) + with without_global_fallbacks(): + litellm.completion(model=model, messages=messages, api_key=api_key, max_tokens=10) return True except AuthenticationError: return False diff --git a/tests/unit/litellm_core_utils/test_fallback_utils.py b/tests/unit/litellm_core_utils/test_fallback_utils.py index c0a344cc6c3..96b5676b5c9 100644 --- a/tests/unit/litellm_core_utils/test_fallback_utils.py +++ b/tests/unit/litellm_core_utils/test_fallback_utils.py @@ -1,23 +1,39 @@ """Tests for litellm.litellm_core_utils.fallback_utils.""" +import asyncio +import contextlib import json import pathlib -from typing import Final +import threading +from collections.abc import Mapping +from concurrent.futures import ThreadPoolExecutor +from typing import Final, NoReturn import httpx import pytest import respx +from pydantic import TypeAdapter import litellm +from litellm.caching.caching import Cache, DualCache +from litellm.integrations.custom_guardrail import CustomGuardrail +from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils import litellm_logging from litellm.litellm_core_utils.core_helpers import process_response_headers from litellm.litellm_core_utils.fallback_utils import ( async_completion_with_fallbacks, ) +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER +from litellm.llms.chatgpt.common_utils import CHATGPT_DEVICE_CODE_URL from litellm.main import acompletion as python_acompletion from litellm.main import completion as python_completion +from litellm.types.guardrails import GuardrailEventHooks +from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth +from litellm.types.utils import CallTypes, CallTypesLiteral, Choices, CredentialItem, LLMResponseTypes, Message, ModelResponse from litellm.vector_stores.vector_store_registry import LiteLLM_ManagedVectorStore, VectorStoreRegistry +LITELLM_PARAMS: Final = TypeAdapter(dict[str, object]) + @pytest.mark.asyncio async def test_fallback_dict_not_mutated(monkeypatch): @@ -179,9 +195,7 @@ def test_process_response_headers_ignores_preserve_flag_for_httpx_headers(): @pytest.mark.asyncio -async def test_acompletion_fallbacks_bad_models( - respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch -) -> None: +async def test_acompletion_fallbacks_bad_models(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( side_effect=( @@ -210,7 +224,7 @@ async def test_acompletion_fallbacks_bad_models( ) ) - with pytest.raises(Exception, match="All fallback attempts failed"): + with pytest.raises(Exception, match=r"fallback unavailable\. All fallback attempts failed"): await python_acompletion( model="openai/first", messages=[{"role": "user", "content": "hello"}], @@ -222,9 +236,7 @@ async def test_acompletion_fallbacks_bad_models( @pytest.mark.asyncio -async def test_acompletion_fallbacks_basic( - respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch -) -> None: +async def test_acompletion_fallbacks_basic(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( side_effect=( @@ -271,9 +283,7 @@ async def test_acompletion_fallbacks_basic( @pytest.mark.asyncio -async def test_acompletion_fallbacks_empty_list( - respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch -) -> None: +async def test_acompletion_fallbacks_empty_list(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( return_value=httpx.Response( @@ -401,9 +411,7 @@ async def test_acompletion_fallbacks_with_dict_config( ) -def test_completion_fallbacks_sync( - respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_completion_fallbacks_sync(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( side_effect=( @@ -449,6 +457,27 @@ def test_completion_fallbacks_sync( assert tuple(json.loads(call.request.content)["model"] for call in route.calls) == ("first", "second") +@pytest.mark.timeout(30, method="thread") +def test_completion_fallbacks_name_the_last_error_when_every_model_fails( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + fallback_unavailable: Final = { + "error": {"message": "fallback unavailable", "type": "invalid_request_error", "param": None, "code": None} + } + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=(_model_unavailable(), httpx.Response(404, json=fallback_unavailable)) + ) + + with pytest.raises(Exception, match=r"fallback unavailable\. All fallback attempts failed"): + python_completion( + model="openai/first", + messages=[{"role": "user", "content": "hello"}], + api_key="test-key", + fallbacks=["openai/second"], + ) + + CACHE_POINTS: Final = ({"location": "message", "role": "system"},) CACHED_MESSAGES: Final = ( {"role": "system", "content": "You are a careful assistant."}, @@ -631,3 +660,763 @@ async def test_acompletion_fallback_sends_the_prompt_the_primary_model_prefix_te direct_body, fallback_body = (json.loads(call.request.content) for call in anthropic_route.calls) assert "What color is the sky?" in json.dumps(direct_body) assert fallback_body == direct_body + + +def _model_unavailable() -> httpx.Response: + return httpx.Response( + 404, + json={ + "error": { + "message": "model unavailable", + "type": "invalid_request_error", + "param": None, + "code": "model_not_found", + } + }, + ) + + +def _replies(contents: tuple[str | None, ...]) -> tuple[httpx.Response, ...]: + return tuple( + _model_unavailable() + if content is None + else httpx.Response( + 200, + json={ + **OPENAI_ANSWER, + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop"} + ], + }, + ) + for content in contents + ) + + +def _requested_models(route: respx.Route) -> tuple[str, ...]: + return tuple(json.loads(call.request.content)["model"] for call in route.calls) + + +GLOBAL_CHAIN_OUTCOMES: Final = ( + pytest.param(("first answer",), ("first",), id="primary-answers"), + pytest.param((None, None, "third answer"), ("first", "second", "third"), id="falls-through-in-order"), +) + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.parametrize(("contents", "expected_models"), GLOBAL_CHAIN_OUTCOMES) +@pytest.mark.asyncio +async def test_acompletion_tries_the_global_model_fallbacks_once_each_in_order( + respx_mock: respx.MockRouter, + monkeypatch: pytest.MonkeyPatch, + contents: tuple[str | None, ...], + expected_models: tuple[str, ...], +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second", "openai/third"]) + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies(contents)) + + response: Final = await python_acompletion( + model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key" + ) + + assert response.choices[0].message.content == contents[-1] + assert _requested_models(route) == expected_models + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.parametrize(("contents", "expected_models"), GLOBAL_CHAIN_OUTCOMES) +def test_completion_tries_the_global_model_fallbacks_once_each_in_order( + respx_mock: respx.MockRouter, + monkeypatch: pytest.MonkeyPatch, + contents: tuple[str | None, ...], + expected_models: tuple[str, ...], +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second", "openai/third"]) + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies(contents)) + + response: Final = python_completion( + model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key" + ) + + assert response.choices[0].message.content == contents[-1] + assert _requested_models(route) == expected_models + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.asyncio +async def test_acompletion_per_call_fallbacks_win_over_the_global_model_fallbacks( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/global-fallback"]) + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies((None, "per-call answer")) + ) + + response: Final = await python_acompletion( + model="openai/first", + messages=[{"role": "user", "content": "hello"}], + api_key="test-key", + fallbacks=["openai/second"], + ) + + assert response.choices[0].message.content == "per-call answer" + assert _requested_models(route) == ("first", "second") + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.asyncio +async def test_acompletion_empty_per_call_fallbacks_opt_out_of_the_global_model_fallbacks( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"]) + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies((None, "fallback answer")) + ) + + with pytest.raises(litellm.NotFoundError, match="model unavailable"): + await python_acompletion( + model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key", fallbacks=[] + ) + + assert _requested_models(route) == ("first",) + + +async def _flush_success_logs() -> None: + await asyncio.sleep(0) + GLOBAL_LOGGING_WORKER.start() + await asyncio.wait_for(GLOBAL_LOGGING_WORKER.flush(), timeout=10.0) + + +class _SuccessRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.models: list[object] = [] + self.call_ids: list[object] = [] + + async def async_log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + self.models.append(kwargs.get("model")) + self.call_ids.append(LITELLM_PARAMS.validate_python(kwargs.get("litellm_params")).get("litellm_call_id")) + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.asyncio +async def test_acompletion_through_the_global_model_fallbacks_logs_one_success( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + recorder: Final = _SuccessRecorder() + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"]) + monkeypatch.setattr(litellm, "callbacks", [recorder]) + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies((None, "fallback answer")) + ) + + await python_acompletion( + model="openai/first", + messages=[{"role": "user", "content": "hello"}], + api_key="test-key", + litellm_call_id="caller-call-id", + ) + await _flush_success_logs() + + assert recorder.models == ["second"] + assert recorder.call_ids == ["caller-call-id"] + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.parametrize( + "global_fallbacks", + (pytest.param(["openai/second"], id="global-set"), pytest.param(None, id="global-unset")), +) +@pytest.mark.asyncio +async def test_acompletion_with_empty_per_call_fallbacks_logs_one_success( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, global_fallbacks: list[str] | None +) -> None: + recorder: Final = _SuccessRecorder() + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", global_fallbacks) + monkeypatch.setattr(litellm, "callbacks", [recorder]) + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies(("first answer",))) + + await python_acompletion( + model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key", fallbacks=[] + ) + await _flush_success_logs() + + assert recorder.models == ["first"] + + +class _EverySuccessRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.events: list[tuple[str, object]] = [] + + def log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + self.events.append(("sync", kwargs.get("model"))) + + async def async_log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + self.events.append(("async", kwargs.get("model"))) + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.asyncio +async def test_atext_completion_through_the_global_model_fallbacks_logs_one_success( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + recorder: Final = _EverySuccessRecorder() + logging_executor: Final = ThreadPoolExecutor(max_workers=2) + monkeypatch.setattr(litellm.utils, "executor", logging_executor) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"]) + monkeypatch.setattr(litellm, "callbacks", [recorder]) + monkeypatch.setenv("OPENAI_API_KEY", "test-key") + chat_route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies(("fallback answer",)) + ) + text_route: Final = respx_mock.post("https://api.openai.com/v1/completions").mock( + side_effect=(_model_unavailable(),) + ) + + response: Final = await litellm.atext_completion(model="openai/first", prompt="hello") + await _flush_success_logs() + logging_executor.shutdown(wait=True) + + assert response.choices[0].text == "fallback answer" + assert _requested_models(text_route) == ("first",) + assert _requested_models(chat_route) == ("second",) + assert len(recorder.events) == 1 + + +class _DeploymentAttemptRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.attempts: list[tuple[str, object]] = [] + + async def async_post_call_failure_deployment_hook( + self, + request_data: Mapping[str, object], + exception: Exception, + call_type: CallTypes | None, + fallback_depth: int | None = None, + ) -> None: + self.attempts.append(("failure", request_data.get("model"))) + + async def async_post_call_success_deployment_hook( + self, request_data: dict[str, object], response: LLMResponseTypes, call_type: CallTypes | None + ) -> LLMResponseTypes | None: + self.attempts.append(("success", request_data.get("model"))) + return response + + +@pytest.mark.timeout(30, method="thread") +def test_completion_with_fallbacks_reports_every_attempt_to_the_deployment_hooks_in_order( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + recorder: Final = _DeploymentAttemptRecorder() + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "callbacks", [recorder]) + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies((None, None, "third answer")) + ) + + response: Final = python_completion( + model="openai/first", + messages=[{"role": "user", "content": "hello"}], + api_key="test-key", + fallbacks=["openai/second", "openai/third"], + ) + + assert response.choices[0].message.content == "third answer" + assert recorder.attempts == [("failure", "openai/first"), ("failure", "openai/second"), ("success", "openai/third")] + + +CREDENTIAL_ROUTES: Final = ( + pytest.param( + "openai/second", + "https://gateway.example/v1/chat/completions", + ("authorization", "Bearer caller-key"), + id="same-provider-keeps-the-callers-credentials", + ), + pytest.param( + "anthropic/claude-haiku-5-5", + "https://api.anthropic.com/v1/messages", + ("x-api-key", "anthropic-env-key"), + id="other-provider-uses-its-own-credentials", + ), +) + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.parametrize(("fallback", "fallback_url", "credential"), CREDENTIAL_ROUTES) +@pytest.mark.asyncio +async def test_acompletion_fallback_sends_credentials_only_to_the_provider_they_belong_to( + respx_mock: respx.MockRouter, + monkeypatch: pytest.MonkeyPatch, + fallback: str, + fallback_url: str, + credential: tuple[str, str], +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", [fallback]) + monkeypatch.setenv("ANTHROPIC_API_KEY", "anthropic-env-key") + _ = respx_mock.post("https://gateway.example/v1/chat/completions").mock( + side_effect=(_model_unavailable(), httpx.Response(200, json=OPENAI_ANSWER)) + ) + _ = respx_mock.post("https://api.anthropic.com/v1/messages").mock( + return_value=httpx.Response(200, json=ANTHROPIC_ANSWER) + ) + + response: Final = await python_acompletion( + model="openai/first", + messages=[{"role": "user", "content": "hello"}], + api_key="caller-key", + api_base="https://gateway.example/v1", + ) + + header, value = credential + assert response.choices[0].message.content == "ok" + assert str(respx_mock.calls.last.request.url) == fallback_url + assert respx_mock.calls.last.request.headers[header] == value + + +GATEWAY_CREDENTIAL_CARRIERS: Final = ( + pytest.param("openai/first", {"litellm_credential_name": "gateway"}, id="named-credential"), + pytest.param( + "openai/first", + { + "api_key": "gateway-key", + "api_base": "https://gateway.example/v1", + "headers": {"authorization": "Bearer gateway-key"}, + }, + id="request-headers", + ), + pytest.param( + "first-custom-model", + {"custom_llm_provider": "openai", "api_key": "gateway-key", "api_base": "https://gateway.example/v1"}, + id="explicit-provider", + ), +) + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.parametrize(("model", "gateway_credentials"), GATEWAY_CREDENTIAL_CARRIERS) +@pytest.mark.asyncio +async def test_acompletion_fallback_to_another_provider_never_reuses_the_first_providers_credentials( + respx_mock: respx.MockRouter, + monkeypatch: pytest.MonkeyPatch, + model: str, + gateway_credentials: dict[str, object], +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr( + litellm, + "credential_list", + [ + CredentialItem( + credential_name="gateway", + credential_values={"api_key": "gateway-key", "api_base": "https://gateway.example/v1"}, + credential_info={}, + ) + ], + ) + monkeypatch.setenv("ANTHROPIC_API_KEY", "anthropic-env-key") + _ = respx_mock.post("https://gateway.example/v1/chat/completions").mock(return_value=_model_unavailable()) + _ = respx_mock.post("https://api.anthropic.com/v1/messages").mock( + return_value=httpx.Response(200, json=ANTHROPIC_ANSWER) + ) + + response: Final = await python_acompletion( + model=model, + messages=[{"role": "user", "content": "hello"}], + fallbacks=["anthropic/claude-haiku-5-5"], + **gateway_credentials, + ) + + fallback_request: Final = respx_mock.calls.last.request + assert response.choices[0].message.content == "ok" + assert str(fallback_request.url) == "https://api.anthropic.com/v1/messages" + assert fallback_request.headers["x-api-key"] == "anthropic-env-key" + assert not any("gateway-key" in value for value in fallback_request.headers.values()) + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.asyncio +async def test_acompletion_fallback_to_an_unprefixed_model_keeps_the_requests_provider_and_credentials( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + route: Final = respx_mock.post("https://models.example/v1/chat/completions").mock( + side_effect=_replies((None, "fallback answer")) + ) + + response: Final = await python_acompletion( + model="first-custom-model", + custom_llm_provider="openai", + api_base="https://models.example/v1", + api_key="models-key", + messages=[{"role": "user", "content": "hello"}], + fallbacks=["second-custom-model"], + ) + + assert response.choices[0].message.content == "fallback answer" + assert _requested_models(route) == ("first-custom-model", "second-custom-model") + assert route.calls.last.request.headers["authorization"] == "Bearer models-key" + + +@pytest.mark.timeout(30, method="thread") +def test_completion_fallback_from_a_model_without_a_provider_never_reuses_its_credentials( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setenv("ANTHROPIC_API_KEY", "anthropic-env-key") + route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock( + return_value=httpx.Response(200, json=ANTHROPIC_ANSWER) + ) + + response: Final = python_completion( + model="first-without-a-provider", + messages=[{"role": "user", "content": "hello"}], + api_key="first-key", + fallbacks=["anthropic/claude-haiku-5-5"], + ) + + assert response.choices[0].message.content == "ok" + assert route.calls.last.request.headers["x-api-key"] == "anthropic-env-key" + + +async def _router_completion(router: litellm.Router, *, is_async: bool) -> object: + messages: Final = [{"role": "user", "content": "hello"}] + if is_async: + return await router.acompletion(model="first", messages=messages) + return router.completion(model="first", messages=messages) + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.parametrize("is_async", [True, False]) +@pytest.mark.asyncio +async def test_router_leaves_fallbacks_to_the_router_when_global_model_fallbacks_are_set( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, is_async: bool +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"]) + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies((None, "global fallback answer")) + ) + router: Final = litellm.Router( + model_list=[{"model_name": "first", "litellm_params": {"model": "openai/first", "api_key": "test-key"}}], + num_retries=0, + ) + + with pytest.raises(litellm.NotFoundError, match="model unavailable"): + await _router_completion(router, is_async=is_async) + + assert _requested_models(route) == ("first",) + + +STREAMED_ANSWER: Final = ( + "".join( + f"data: {json.dumps(chunk)}\n\n" + for chunk in ( + { + "id": "chatcmpl-1", + "object": "chat.completion.chunk", + "created": 1, + "model": "first", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "streamed answer"}}], + }, + { + "id": "chatcmpl-1", + "object": "chat.completion.chunk", + "created": 1, + "model": "first", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + }, + ) + ) + + "data: [DONE]\n\n" +) + + +@pytest.mark.timeout(30, method="thread") +def test_completion_streams_through_the_global_model_fallbacks( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"]) + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + return_value=httpx.Response(200, text=STREAMED_ANSWER, headers={"content-type": "text/event-stream"}) + ) + + stream: Final = python_completion( + model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key", stream=True + ) + + assert "".join(chunk.choices[0].delta.content or "" for chunk in stream) == "streamed answer" + + +class _ConcurrentSyncSuccessRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.models: list[object] = [] + self._pair_concurrent_logs: Final = threading.Barrier(2, timeout=0.5) + + def logging_hook( + self, kwargs: dict[str, object], result: object, call_type: str + ) -> tuple[dict[str, object], object]: + with contextlib.suppress(threading.BrokenBarrierError): + _ = self._pair_concurrent_logs.wait() + return kwargs, result + + def log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + self.models.append(kwargs.get("model")) + + +@pytest.mark.timeout(30, method="thread") +def test_completion_through_the_global_model_fallbacks_logs_one_success( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + recorder: Final = _ConcurrentSyncSuccessRecorder() + logging_executor: Final = ThreadPoolExecutor(max_workers=2) + monkeypatch.setattr(litellm.utils, "executor", logging_executor) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"]) + monkeypatch.setattr(litellm, "callbacks", [recorder]) + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies((None, "fallback answer")) + ) + + python_completion(model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key") + logging_executor.shutdown(wait=True) + + assert recorder.models == ["second"] + + +class _RedactingGuardrail(CustomGuardrail): + def __init__(self) -> None: + super().__init__( + guardrail_name="redact", event_hook=[GuardrailEventHooks.pre_call, GuardrailEventHooks.post_call] + ) + + async def async_pre_call_hook( + self, user_api_key_dict: UserAPIKeyAuth, cache: DualCache, data: dict, call_type: CallTypesLiteral + ) -> dict[str, object]: + return {**data, "messages": [{"role": "user", "content": "[REDACTED]"}]} + + async def async_post_call_success_hook( + self, data: dict, user_api_key_dict: UserAPIKeyAuth, response: LLMResponseTypes + ) -> ModelResponse: + return ModelResponse(model="checked", choices=[Choices(message=Message(content="[CHECKED]"))]) + + +@pytest.mark.timeout(30, method="thread") +def test_completion_with_fallbacks_runs_the_requested_guardrail_before_every_attempt( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "callbacks", [_RedactingGuardrail()]) + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies((None, "fallback answer")) + ) + + response: Final = python_completion( + model="openai/first", + messages=[{"role": "user", "content": "my card is 4111 1111 1111 1111"}], + api_key="test-key", + fallbacks=["openai/second"], + guardrails=["redact"], + ) + + assert response.choices[0].message.content == "[CHECKED]" + assert [json.loads(call.request.content)["messages"] for call in route.calls] == [ + [{"role": "user", "content": "[REDACTED]"}] + ] * 2 + + +class _RejectingGuardrail(CustomGuardrail): + def __init__(self) -> None: + super().__init__(guardrail_name="reject", event_hook=GuardrailEventHooks.post_call) + + async def async_post_call_success_hook( + self, data: dict, user_api_key_dict: UserAPIKeyAuth, response: LLMResponseTypes + ) -> NoReturn: + raise ValueError("answer rejected") + + +def _completion_checked_by_the_rejecting_guardrail() -> object: + return python_completion( + model="openai/first", + messages=[{"role": "user", "content": "hello"}], + api_key="test-key", + fallbacks=["openai/second"], + guardrails=["reject"], + ) + + +@pytest.mark.timeout(30, method="thread") +def test_completion_with_fallbacks_never_caches_an_answer_its_guardrail_rejected( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "cache", Cache(type="local")) + monkeypatch.setattr(litellm, "callbacks", [_RejectingGuardrail()]) + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies(("rejected answer",) * 4) + ) + + with pytest.raises(Exception, match="answer rejected"): + _ = _completion_checked_by_the_rejecting_guardrail() + with pytest.raises(Exception, match="answer rejected"): + _ = _completion_checked_by_the_rejecting_guardrail() + + assert _requested_models(route) == ("first", "second", "first", "second") + + +class _SyncAnswerRecorder(CustomLogger): + def __init__(self) -> None: + super().__init__() + self.answers: list[object] = [] + + def log_success_event( + self, kwargs: dict[str, object], response_obj: object, start_time: object, end_time: object + ) -> None: + self.answers.append( + response_obj.choices[0].message.content if isinstance(response_obj, ModelResponse) else None + ) + + +@pytest.mark.parametrize( + ("guardrail_class", "logged_answers"), + [(_RedactingGuardrail, ["[CHECKED]"]), (_RejectingGuardrail, [])], + ids=["rewritten", "rejected"], +) +@pytest.mark.timeout(30, method="thread") +def test_completion_with_fallbacks_logs_only_the_answer_its_guardrail_returned( + respx_mock: respx.MockRouter, + monkeypatch: pytest.MonkeyPatch, + guardrail_class: type[CustomGuardrail], + logged_answers: list[str], +) -> None: + guardrail: Final = guardrail_class() + recorder: Final = _SyncAnswerRecorder() + logging_executor: Final = ThreadPoolExecutor(max_workers=2) + monkeypatch.setattr(litellm.utils, "executor", logging_executor) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "callbacks", [guardrail, recorder]) + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies((None, "raw answer"))) + + with contextlib.suppress(Exception): + _ = python_completion( + model="openai/first", + messages=[{"role": "user", "content": "hello"}], + api_key="test-key", + fallbacks=["openai/second"], + guardrails=[guardrail.guardrail_name], + ) + logging_executor.shutdown(wait=True) + + assert recorder.answers == logged_answers + + +@pytest.mark.timeout(30, method="thread") +def test_completion_streams_through_fallbacks_with_the_requested_guardrail( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "callbacks", [_RedactingGuardrail()]) + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + return_value=httpx.Response(200, text=STREAMED_ANSWER, headers={"content-type": "text/event-stream"}) + ) + + stream: Final = python_completion( + model="openai/first", + messages=[{"role": "user", "content": "my card is 4111 1111 1111 1111"}], + api_key="test-key", + stream=True, + fallbacks=["openai/second"], + guardrails=["redact"], + ) + + assert "".join(chunk.choices[0].delta.content or "" for chunk in stream) == "streamed answer" + assert json.loads(route.calls.last.request.content)["messages"] == [{"role": "user", "content": "[REDACTED]"}] + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.asyncio +@pytest.mark.parametrize( + "primary_price", + ({"input_cost_per_token": 1.0, "output_cost_per_token": 1.0}, {"base_model": "gpt-5.4"}), +) +async def test_acompletion_fallback_to_another_model_is_priced_like_a_direct_call_to_it( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, primary_price: dict[str, object] +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=( + _model_unavailable(), + httpx.Response(200, json=OPENAI_ANSWER), + httpx.Response(200, json=OPENAI_ANSWER), + ) + ) + through_fallback: Final = await python_acompletion( + model="openai/first", + messages=[{"role": "user", "content": "hello"}], + api_key="test-key", + fallbacks=["openai/gpt-5.4-nano"], + **primary_price, + ) + direct: Final = await python_acompletion( + model="openai/gpt-5.4-nano", messages=[{"role": "user", "content": "hello"}], api_key="test-key" + ) + + assert through_fallback._hidden_params["response_cost"] == direct._hidden_params["response_cost"] + + +@pytest.mark.timeout(30, method="thread") +def test_completion_never_signs_in_to_a_chatgpt_fallback_the_call_does_not_use( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["chatgpt/gpt-5.4"]) + monkeypatch.setenv("CHATGPT_TOKEN_DIR", str(tmp_path)) + device_login: Final = respx_mock.post(CHATGPT_DEVICE_CODE_URL).mock(return_value=httpx.Response(503)) + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_replies(("first answer",))) + + response: Final = python_completion( + model="openai/first", messages=[{"role": "user", "content": "hello"}], api_key="test-key" + ) + + assert response.choices[0].message.content == "first answer" + assert not device_login.called + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.asyncio +async def test_acompletion_falls_back_from_a_chatgpt_model_that_is_not_signed_in( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/second"]) + monkeypatch.setenv("CHATGPT_TOKEN_DIR", str(tmp_path)) + monkeypatch.setenv("OPENAI_API_KEY", "openai-env-key") + route: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + side_effect=_replies(("fallback answer",)) + ) + + response: Final = await python_acompletion( + model="gpt-5.4", custom_llm_provider="chatgpt", messages=[{"role": "user", "content": "hello"}] + ) + + assert response.choices[0].message.content == "fallback answer" + assert _requested_models(route) == ("second",) diff --git a/tests/unit/litellm_core_utils/test_health_check_helpers.py b/tests/unit/litellm_core_utils/test_health_check_helpers.py index 950a91beb81..79b6e07906a 100644 --- a/tests/unit/litellm_core_utils/test_health_check_helpers.py +++ b/tests/unit/litellm_core_utils/test_health_check_helpers.py @@ -1987,3 +1987,77 @@ async def test_wildcard_health_check_with_pinned_provider_calls_a_model_without_ assert "error" not in result, result assert json.loads(together_route.calls.last.request.content)["model"] == "fake-together-chat" assert not other_hosts.called + + +HEALTH_CHECK_ROUTES: Final = ( + pytest.param( + "chat", + {"model": "openai/missing-model", "api_base": "https://deployment.example/v1"}, + "https://deployment.example/v1/chat/completions", + id="chat", + ), + pytest.param( + "responses", + {"model": "anthropic/missing-model", "api_base": "https://deployment.example"}, + "https://deployment.example/v1/messages", + id="responses-bridged-to-chat", + ), +) + + +def _mock_failing_deployment_with_a_healthy_openai_fallback( + respx_mock: respx.MockRouter, deployment_url: str +) -> respx.Route: + _ = respx_mock.post("https://api.openai.com/v1/chat/completions").mock( + return_value=httpx.Response( + 200, + json={ + "id": "chatcmpl-1", + "object": "chat.completion", + "created": 1, + "model": "healthy-fallback", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + }, + ) + ) + return respx_mock.post(deployment_url).mock( + return_value=httpx.Response(404, json={"error": {"message": "model unavailable", "type": "not_found"}}) + ) + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.parametrize(("mode", "deployment_params", "deployment_url"), HEALTH_CHECK_ROUTES) +@pytest.mark.asyncio +async def test_ahealth_check_reports_a_failing_deployment_whatever_the_global_model_fallbacks( + respx_mock: respx.MockRouter, + monkeypatch: pytest.MonkeyPatch, + mode: Literal["chat", "responses"], + deployment_params: dict[str, str], + deployment_url: str, +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/healthy-fallback"]) + deployment: Final = _mock_failing_deployment_with_a_healthy_openai_fallback(respx_mock, deployment_url) + + result: Final = await ahealth_check({**deployment_params, "api_key": "sk-test"}, mode=mode) + + assert "error" in result + assert len(deployment.calls) == 1 + + +@pytest.mark.timeout(30, method="thread") +@pytest.mark.asyncio +async def test_calls_after_a_health_check_still_use_the_global_model_fallbacks( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/healthy-fallback"]) + monkeypatch.setenv("OPENAI_API_KEY", "openai-env-key") + deployment: Final = {"model": "anthropic/missing-model", "api_base": "https://deployment.example", "api_key": "sk-test"} + _ = _mock_failing_deployment_with_a_healthy_openai_fallback(respx_mock, "https://deployment.example/v1/messages") + + _ = await ahealth_check(deployment, mode="chat") + response: Final = await litellm.acompletion(**deployment, messages=[{"role": "user", "content": "hi"}]) + + assert response.choices[0].message.content == "ok" diff --git a/tests/unit/proxy/test_proxy_config_unit_test.py b/tests/unit/proxy/test_proxy_config_unit_test.py index 28e017bb050..3fcd74b52e8 100644 --- a/tests/unit/proxy/test_proxy_config_unit_test.py +++ b/tests/unit/proxy/test_proxy_config_unit_test.py @@ -18,6 +18,9 @@ import io import asyncio import logging +import yaml + +import litellm from litellm.proxy.proxy_server import ProxyConfig INVALID_FILES = ["config_with_missing_include.yaml"] @@ -355,3 +358,23 @@ async def test_load_config_rejects_unknown_mcp_disabled_capability(tmp_path: Pat with pytest.raises(ValidationError): await ProxyConfig().load_config(router=None, config_file_path=str(config_path)) + + +@pytest.mark.asyncio +async def test_proxy_config_model_fallbacks_never_becomes_the_sdk_global_fallback_list( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "model_fallbacks", None) + config_file: Final = tmp_path / "config.yaml" + _ = config_file.write_text( + yaml.safe_dump( + { + "model_list": [{"model_name": "gpt-5.4-nano", "litellm_params": {"model": "openai/gpt-5.4-nano"}}], + "litellm_settings": {"model_fallbacks": ["anthropic/claude-haiku-5-5"]}, + } + ) + ) + + _ = await ProxyConfig().load_config(router=None, config_file_path=str(config_file)) + + assert litellm.model_fallbacks is None diff --git a/tests/unit/proxy/test_route_llm_request.py b/tests/unit/proxy/test_route_llm_request.py index 5927f906530..7fb034022d0 100644 --- a/tests/unit/proxy/test_route_llm_request.py +++ b/tests/unit/proxy/test_route_llm_request.py @@ -398,7 +398,7 @@ async def test_route_request_no_model_required_with_router_settings_and_no_route } with patch.object(litellm, "acompletion", return_value="fake_response") as mock_completion: - await route_request(data, None, "gpt-3.5-turbo", "acompletion") + assert await (await route_request(data, None, "gpt-3.5-turbo", "acompletion")) == "fake_response" mock_completion.assert_called_once_with(**data) @@ -1480,6 +1480,40 @@ async def test_route_request_keeps_key_fallbacks_out_of_pass_through_provider_bo assert provider_bodies and all("gpt-4" not in body for body in provider_bodies), provider_bodies +@pytest.mark.asyncio +async def test_route_request_pass_through_never_falls_back_to_the_global_model_fallbacks( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + monkeypatch.setattr(litellm, "model_fallbacks", ["openai/gpt-not-granted"]) + monkeypatch.setenv("OPENAI_API_KEY", "sk-pass-through") + monkeypatch.delenv("OPENAI_BASE_URL", raising=False) + monkeypatch.delenv("OPENAI_API_BASE", raising=False) + router: Final = litellm.Router( + model_list=[ + { + "model_name": "gpt-4", + "litellm_params": {"model": "openai/gpt-4", "api_key": "sk-proxy", "api_base": _PROXY_DEPLOYMENT_BASE}, + } + ], + router_general_settings=RouterGeneralSettings(pass_through_all_models=True), + ) + data: Final = {"model": "openai/gpt-pass-through", "messages": list(_HELLO_MESSAGES)} + + with respx.mock(assert_all_called=False) as respx_mock: + requested: Final = respx_mock.post( + "https://api.openai.com/v1/chat/completions", json__model="gpt-pass-through" + ).mock(return_value=httpx.Response(404, json={"error": {"message": "no such model", "code": "model_not_found"}})) + not_granted: Final = respx_mock.post( + "https://api.openai.com/v1/chat/completions", json__model="gpt-not-granted" + ).mock(return_value=_chat_completion_response("gpt-not-granted")) + with pytest.raises(litellm.NotFoundError): + await (await route_request(data, router, None, "acompletion")) + + assert requested.called + assert not not_granted.called + + @pytest.mark.parametrize( "route_type, param, route", [ diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 5a21de5c382..eba411e27ac 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -11107,6 +11107,36 @@ def test_check_valid_key_returns_false_for_http_unauthorized() -> None: assert upstream.calls.last.request.headers["Authorization"] == "Bearer invalid-test-key" +def test_check_valid_key_never_counts_a_global_fallback_answer_as_a_valid_key( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(litellm, "model_fallbacks", ["anthropic/claude-haiku-5-5"]) + monkeypatch.setenv("ANTHROPIC_API_KEY", "anthropic-env-key") + upstream: Final[respx.MockRouter] + + with respx.mock(assert_all_called=False) as upstream: + _ = upstream.post("https://api.openai.com/v1/chat/completions").respond( + 401, json={"error": {"message": "Invalid API key", "type": "invalid_request_error", "code": "invalid_api_key"}} + ) + fallback_route: Final = upstream.post("https://api.anthropic.com/v1/messages").respond( + 200, + json={ + "id": "msg_test", + "type": "message", + "role": "assistant", + "model": "claude-haiku-5-5", + "content": [{"type": "text", "text": "ok"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 1, "output_tokens": 1}, + }, + ) + + valid_key: Final = check_valid_key(model="gpt-5-mini", api_key="invalid-test-key") + + assert valid_key is False + assert not fallback_route.called + + def test_check_valid_key_returns_true_for_successful_completion() -> None: response_body: Final = { "id": "chatcmpl-test",