scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Merge branch 'litellm_internal_staging' into litellm_/testing-strategy-audit-c39e33

Browse source
This commit is contained in:
yuneng-jiang 2026-08-26 11:24:55 -07:00 • committed by GitHub
commit 309da5e70e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
76 changed files with 2125 additions and 379 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
basedpyright-code-budget.json
View file

@ -1,6 +1,6 @@
{
"reportAny": {
"limit": 18505
"limit": 18483
},
"reportArgumentType": {
"limit": 2564
@ -24,7 +24,7 @@
"limit": 19
},
"reportExplicitAny": {
"limit": 5976
"limit": 5960
},
"reportFunctionMemberAccess": {
"limit": 7
@ -57,7 +57,7 @@
"limit": 5659
},
"reportMissingTypeArgument": {
"limit": 15504
"limit": 15484
},
"reportMissingTypeStubs": {
"limit": 40
@ -105,13 +105,13 @@
"limit": 109
},
"reportUnknownMemberType": {
"limit": 38828
"limit": 38808
},
"reportUnknownParameterType": {
"limit": 19847
"limit": 19829
},
"reportUnknownVariableType": {
"limit": 30386
"limit": 30356
},
"reportUnnecessaryCast": {
"limit": 117

4
litellm/batches/batch_utils.py
View file

@ -551,7 +551,7 @@ def _get_batch_job_usage_from_response_body(
return usage
def _get_anthropic_result_from_batch_results_line(batch_results_line: Mapping[str, Any]) -> dict:
def _get_anthropic_result_from_batch_results_line(batch_results_line: Mapping[str, Any]) -> Mapping[str, Any]:
"""
Get the ``result`` object from a line of an Anthropic message batch results JSONL file.
@ -563,7 +563,7 @@ def _get_anthropic_result_from_batch_results_line(batch_results_line: Mapping[st
def _get_response_from_batch_job_output_file(
batch_job_output_file: Mapping[str, Any], custom_llm_provider: str = "openai"
) -> Any:
) -> Mapping[str, Any]:
"""
Get the response from the batch job output file
"""

2
litellm/integrations/custom_logger.py
View file

@ -295,7 +295,7 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
async def async_post_call_failure_deployment_hook(
self,
request_data: Mapping[str, Any],
request_data: Mapping[str, object],
exception: Exception,
call_type: CallTypes | None,
fallback_depth: int | None = None,

5
litellm/integrations/otel/plumbing/routing.py
View file

@ -15,7 +15,7 @@ from collections import OrderedDict
from collections.abc import Mapping
from dataclasses import dataclass
from types import MappingProxyType
from typing import Any, Final, TypeAlias
from typing import Final, TypeAlias
from urllib.parse import quote
from opentelemetry.sdk.trace import TracerProvider
@ -32,6 +32,7 @@ from litellm.integrations.otel.presets import (
dynamic_otlp_headers,
project_routing_headers,
)
from litellm.types.utils import StandardCallbackDynamicParams
# Exporter kinds that ignore headers — never rewritten with dynamic credentials.
_NON_OTLP_KINDS: Final = ("console", "in_memory", "inmemory", "memory")
@ -166,7 +167,7 @@ class TenantTracerCache:
def route_for(
self,
default: Tracer,
dynamic_params: Any,
dynamic_params: StandardCallbackDynamicParams | None,
auth_metadata: Mapping[str, str] | None = None,
) -> TenantRoute:
"""Return the tracer (and trace-detachment flag) for this request.

16
litellm/integrations/prometheus.py
View file

@ -2495,12 +2495,12 @@ class PrometheusLogger(CustomLogger):
return None
def _get_user_email() -> str | None:
val = _metadata.get("user_api_key_user_email")
if val is not None:
return val
val = _litellm_params_metadata.get("user_api_key_user_email")
if val is not None:
return val
from_metadata: Final = _metadata.get("user_api_key_user_email")
if from_metadata is not None:
return from_metadata
from_params: Final = _litellm_params_metadata.get("user_api_key_user_email")
if from_params is not None:
return from_params
if user_api_key_auth is not None:
return self._safe_get(user_api_key_auth, "user_email")
return None
@ -3576,7 +3576,9 @@ class PrometheusLogger(CustomLogger):
except Exception as e:
verbose_logger.exception("Error initializing user/team count metrics: %s", e)
async def _set_key_list_budget_metrics(self, keys: list[str | UserAPIKeyAuth | LiteLLM_DeletedVerificationToken]):
async def _set_key_list_budget_metrics(
self, keys: list[str | UserAPIKeyAuth | LiteLLM_DeletedVerificationToken]
) -> None:
"""Helper function to set budget metrics for a list of keys"""
for key in keys:
if isinstance(key, UserAPIKeyAuth):

138
litellm/litellm_core_utils/exception_mapping_utils.py
View file

@ -550,6 +550,13 @@ def _map_anthropic_exception(
llm_provider="anthropic",
model=model,
)
elif original_exception.status_code == 403:
raise PermissionDeniedError(
message=f"AnthropicException - {error_str}",
llm_provider="anthropic",
model=model,
response=original_exception.response,
)
elif original_exception.status_code == 400 or original_exception.status_code == 413:
raise BadRequestError(
message=f"AnthropicException - {error_str}",
@ -755,12 +762,19 @@ def _map_openai_like_exception(
llm_provider=custom_llm_provider,
model=model,
)
elif original_exception.status_code == 401 or original_exception.status_code == 403:
elif original_exception.status_code == 401:
raise AuthenticationError(
message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}",
llm_provider=custom_llm_provider,
model=model,
)
elif original_exception.status_code == 403:
raise PermissionDeniedError(
message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}",
llm_provider=custom_llm_provider,
model=model,
response=_response_or_stub(original_exception, status_code=403),
)
elif original_exception.status_code == 400:
raise BadRequestError(
message=f"{custom_llm_provider.capitalize()}Exception - {original_exception.message}",
@ -2187,6 +2201,120 @@ def _map_openrouter_exception(
)
def _response_or_stub(original_exception: _ProviderHTTPException, status_code: int) -> httpx.Response:
response: Final = original_exception.response if hasattr(original_exception, "response") else None
if response is not None:
return response
return httpx.Response(
status_code=status_code, request=httpx.Request(method="POST", url="https://docs.litellm.ai/docs")
)
def _map_exception_by_status(
*,
model: str,
original_exception: _ProviderHTTPException,
custom_llm_provider: str,
error_str: str,
exception_provider: str,
extra_information: str,
) -> None:
status_code: Final = original_exception.status_code if hasattr(original_exception, "status_code") else None
if not isinstance(status_code, int) or status_code < 400:
return
message: Final = f"{exception_provider} - {error_str}"
response: Final = original_exception.response if hasattr(original_exception, "response") else None
match status_code:
case 401:
raise AuthenticationError(
message=message,
llm_provider=custom_llm_provider,
model=model,
response=response,
litellm_debug_info=extra_information,
)
case 403:
raise PermissionDeniedError(
message=message,
llm_provider=custom_llm_provider,
model=model,
response=_response_or_stub(original_exception, status_code=status_code),
litellm_debug_info=extra_information,
)
case 404:
raise NotFoundError(
message=message,
model=model,
llm_provider=custom_llm_provider,
response=response,
litellm_debug_info=extra_information,
)
case 408:
raise Timeout(
message=message,
model=model,
llm_provider=custom_llm_provider,
litellm_debug_info=extra_information,
)
case 429:
raise RateLimitError(
message=message,
model=model,
llm_provider=custom_llm_provider,
response=response,
litellm_debug_info=extra_information,
)
case 500:
raise InternalServerError(
message=message,
llm_provider=custom_llm_provider,
model=model,
response=response,
litellm_debug_info=extra_information,
)
case 502:
raise BadGatewayError(
message=message,
llm_provider=custom_llm_provider,
model=model,
response=response,
litellm_debug_info=extra_information,
)
case 503:
raise ServiceUnavailableError(
message=message,
llm_provider=custom_llm_provider,
model=model,
response=response,
litellm_debug_info=extra_information,
)
case 504:
raise Timeout(
message=message,
model=model,
llm_provider=custom_llm_provider,
litellm_debug_info=extra_information,
exception_status_code=status_code,
)
case _ if status_code < 500:
raise BadRequestError(
message=message,
model=model,
llm_provider=custom_llm_provider,
response=response,
litellm_debug_info=extra_information,
)
case _:
raise APIError(
status_code=status_code,
message=message,
llm_provider=custom_llm_provider,
model=model,
request=original_exception.request if hasattr(original_exception, "request") else None,
litellm_debug_info=extra_information,
)
def exception_type(
model,
original_exception,
@ -2501,6 +2629,14 @@ def exception_type(
For unmapped exceptions - raise the exception with traceback - https://github.com/BerriAI/litellm/issues/4201
"""
exception_mapping_worked = True
_map_exception_by_status(
model=model,
original_exception=mappable_exception,
custom_llm_provider=custom_llm_provider,
error_str=error_str,
exception_provider=exception_provider,
extra_information=extra_information,
)
if hasattr(original_exception, "request"):
raise APIConnectionError(
message=f"{exception_provider} - {error_str}",

37
litellm/litellm_core_utils/llm_cost_calc/usage_object_transformation.py
View file

@ -1,6 +1,6 @@
from collections.abc import Mapping, Sequence
from types import MappingProxyType
from typing import Any
from typing import Any, Final
from litellm.types.utils import (
CompletionTokensDetailsWrapper,
@ -39,7 +39,7 @@ class TranscriptionUsageObjectTransformation:
return None
_INTERACTIONS_MODALITY_FIELDS: Mapping[str, str] = MappingProxyType(
_INTERACTIONS_MODALITY_FIELDS: Final[Mapping[str, str]] = MappingProxyType(
{
"text": "text_tokens",
"audio": "audio_tokens",
@ -59,7 +59,7 @@ def _token_count(value: object) -> int:
def _modality_token_sums(entries: Sequence[Mapping[str, Any]]) -> Mapping[str, int]:
fields = frozenset(field for entry in entries if (field := _modality_field(entry)) is not None)
fields: Final = frozenset(field for entry in entries if (field := _modality_field(entry)) is not None)
return MappingProxyType(
{
field: sum(_token_count(entry.get("tokens")) for entry in entries if _modality_field(entry) == field)
@ -69,10 +69,13 @@ def _modality_token_sums(entries: Sequence[Mapping[str, Any]]) -> Mapping[str, i
def _google_search_query_count(usage_object: Mapping[str, Any]) -> int:
entries: Final = usage_object.get("grounding_tool_count")
if not isinstance(entries, Sequence):
return 0
return sum(
_token_count(entry.get("count"))
for entry in tuple(usage_object.get("grounding_tool_count") or ())
if isinstance(entry, Mapping) and entry.get("type") == "google_search" # pyright: ignore[reportUnnecessaryIsInstance] # provider JSON, not the empty tuple inferred from `or ()`
for entry in entries
if isinstance(entry, Mapping) and entry.get("type") == "google_search"
)
@ -112,30 +115,30 @@ class InteractionsUsageObjectTransformation:
@staticmethod
def transform_interactions_usage_object(usage_object: Mapping[str, Any]) -> Usage:
input_entries = tuple(usage_object.get("input_tokens_by_modality") or ()) + tuple(
input_entries: Final = tuple(usage_object.get("input_tokens_by_modality") or ()) + tuple(
usage_object.get("tool_use_tokens_by_modality") or ()
)
cached_sums = _modality_token_sums(tuple(usage_object.get("cached_tokens_by_modality") or ()))
output_sums = _modality_token_sums(tuple(usage_object.get("output_tokens_by_modality") or ()))
cached_sums: Final = _modality_token_sums(tuple(usage_object.get("cached_tokens_by_modality") or ()))
output_sums: Final = _modality_token_sums(tuple(usage_object.get("output_tokens_by_modality") or ()))
total_cached_tokens = _token_count(usage_object.get("total_cached_tokens"))
input_sums = _subtract_cached_from_input(
total_cached_tokens: Final = _token_count(usage_object.get("total_cached_tokens"))
input_sums: Final = _subtract_cached_from_input(
input_sums=_modality_token_sums(input_entries),
cached_sums=cached_sums,
total_cached_tokens=total_cached_tokens,
)
reasoning_tokens = _token_count(usage_object.get("total_reasoning_tokens")) or _token_count(
reasoning_tokens: Final = _token_count(usage_object.get("total_reasoning_tokens")) or _token_count(
usage_object.get("total_thought_tokens")
)
prompt_tokens = _token_count(usage_object.get("total_input_tokens")) + _token_count(
prompt_tokens: Final = _token_count(usage_object.get("total_input_tokens")) + _token_count(
usage_object.get("total_tool_use_tokens")
)
completion_tokens = _token_count(usage_object.get("total_output_tokens")) + reasoning_tokens
total_tokens = _token_count(usage_object.get("total_tokens")) or (prompt_tokens + completion_tokens)
completion_tokens: Final = _token_count(usage_object.get("total_output_tokens")) + reasoning_tokens
total_tokens: Final = _token_count(usage_object.get("total_tokens")) or (prompt_tokens + completion_tokens)
web_search_requests = _google_search_query_count(usage_object)
prompt_tokens_details = (
web_search_requests: Final = _google_search_query_count(usage_object)
prompt_tokens_details: Final = (
PromptTokensDetailsWrapper(
cached_tokens=total_cached_tokens or None,
web_search_requests=web_search_requests or None,
@ -144,7 +147,7 @@ class InteractionsUsageObjectTransformation:
if input_sums or total_cached_tokens or web_search_requests
else None
)
completion_tokens_details = (
completion_tokens_details: Final = (
CompletionTokensDetailsWrapper(
reasoning_tokens=reasoning_tokens or None,
**output_sums,

6
litellm/litellm_core_utils/prompt_templates/common_utils.py
View file

@ -511,9 +511,6 @@ def update_messages_with_model_file_ids(
if "llm_output_file_id," in unified_file_id:
provider_file_id = unified_file_id.split("llm_output_file_id,")[1].split(";")[0]
if not provider_file_id and is_model_embedded_id(file_id):
# `litellm:<raw_id>;model,<m>` encoding from the
# x-litellm-model upload path. Strip the wrapper
# so the provider sees its own ID.
provider_file_id = get_original_file_id(file_id)
file_object_file_field["file_id"] = provider_file_id or file_id
if format:
@ -588,9 +585,6 @@ def update_responses_input_with_model_file_ids(
updated_content_item["file_id"] = provider_file_id
updated_content.append(updated_content_item)
elif is_model_embedded_id(file_id):
# `litellm:<raw_id>;model,<m>` encoding from the
# x-litellm-model upload path. Strip the wrapper
# so the provider sees its own ID.
updated_content_item = content_item.copy()
updated_content_item["file_id"] = get_original_file_id(file_id)
updated_content.append(updated_content_item)

3
litellm/litellm_core_utils/redact_messages.py
View file

@ -10,6 +10,7 @@
import asyncio
import copy
import inspect
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import litellm
@ -191,7 +192,7 @@ def _redact_standard_logging_object(model_call_details: dict):
standard_logging_object["response"] = {"text": redacted_str}
def _redact_tool_calls_dict(message: dict) -> None:
def _redact_tool_calls_dict(message: Mapping[str, object]) -> None:
"""Redact tool call / function_call arguments in a dict-form message or delta."""
tool_calls: Final = message.get("tool_calls")
if isinstance(tool_calls, list):

4
litellm/llms/anthropic/common_utils.py
View file

@ -4,7 +4,7 @@ This file contains common utils for anthropic calls.
import copy
import re
from collections.abc import Mapping, Sequence
from collections.abc import Mapping, MutableMapping, Sequence
from datetime import datetime, timezone
from types import MappingProxyType
from typing import Any, Final, Literal
@ -468,7 +468,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
@staticmethod
def maybe_drop_disabled_thinking(
model: str,
optional_params: dict, # mutable-ok: in-place out-param, same contract as AnthropicConfig._maybe_drop_speed_param
optional_params: MutableMapping[str, object], # mutable-ok: in-place out-param, as in _maybe_drop_speed_param
custom_llm_provider: str,
) -> None:
"""Omit ``thinking={'type': 'disabled'}`` for always-on-thinking models

4
litellm/llms/anthropic/experimental_pass_through/context_management/editors/compact.py
View file

@ -352,8 +352,8 @@ async def _check_summary_model_budget(
)
return False
user_model_max_budget: Final = getattr(user_api_key_auth, "user_model_max_budget", None)
user_id: Final = getattr(user_api_key_auth, "user_id", None)
user_model_max_budget: Final = user_api_key_auth.user_model_max_budget
user_id: Final = user_api_key_auth.user_id
if isinstance(user_model_max_budget, dict) and user_model_max_budget and user_id is not None:
try:
await model_max_budget_limiter.is_user_within_model_budget(

21
litellm/llms/anthropic/experimental_pass_through/messages/handler.py
View file

@ -12,6 +12,7 @@ from functools import partial
from typing import Any, Final, cast
import litellm
from litellm.litellm_core_utils.exception_mapping_utils import exception_type
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.anthropic.common_utils import (
flatten_unencrypted_web_search_results_in_anthropic_messages,
@ -21,6 +22,7 @@ from litellm.llms.anthropic.common_utils import (
from litellm.llms.base_llm.anthropic_messages.transformation import (
BaseAnthropicMessagesConfig,
)
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.types.llms.anthropic_messages.anthropic_request import AnthropicMetadata
@ -382,13 +384,18 @@ async def anthropic_messages(
)
ctx: Final = contextvars.copy_context()
func_with_context: Final = partial(ctx.run, func)
init_response: Final = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
else:
response = init_response
return response
try:
init_response: Final = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
return await init_response
return init_response
except BaseLLMException as e:
raise exception_type(
model=model,
custom_llm_provider=custom_llm_provider,
original_exception=e,
extra_kwargs=kwargs,
)
def validate_anthropic_api_metadata(metadata: dict | None = None) -> dict | None:

2
litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py
View file

@ -582,7 +582,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
"type": "json_schema",
"name": "structured_output",
"schema": schema,
"strict": True,
"strict": output_format.get("strict", False),
}
}

8
litellm/llms/bedrock_mantle/responses/transformation.py
View file

@ -243,7 +243,7 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
return remaining_input, cls._filter_unsupported_tools(hoisted_tools)
@staticmethod
def _agent_message_text(item: "Mapping[str, Any]") -> str:
def _agent_message_text(item: "Mapping[str, object]") -> str:
content: Final = item.get("content")
if not isinstance(content, list):
return ""
@ -254,7 +254,7 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
)
@classmethod
def _normalize_agent_message_item(cls, item: "Mapping[str, Any]") -> "_RewrittenAssistantMessageItem | None":
def _normalize_agent_message_item(cls, item: "Mapping[str, object]") -> "_RewrittenAssistantMessageItem | None":
text: Final = cls._agent_message_text(item)
if not text:
return None
@ -266,7 +266,7 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
return rewritten
@staticmethod
def _normalize_context_compaction_item(item: "Mapping[str, Any]") -> "_RewrittenCompactionItem | None":
def _normalize_context_compaction_item(item: "Mapping[str, object]") -> "_RewrittenCompactionItem | None":
encrypted_content: Final = item.get("encrypted_content")
if not isinstance(encrypted_content, str) or not encrypted_content:
return None
@ -274,7 +274,7 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
return rewritten
@staticmethod
def _normalize_local_shell_call_item(item: "Mapping[str, Any]") -> "_RewrittenFunctionCallItem | None":
def _normalize_local_shell_call_item(item: "Mapping[str, object]") -> "_RewrittenFunctionCallItem | None":
call_id: Final = item.get("call_id")
if not isinstance(call_id, str) or not call_id:
return None

6
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -5620,10 +5620,9 @@ class BaseLLMHTTPHandler:
kwargs=hook_kwargs,
)
except Exception as e:
_call_id = getattr(logging_obj, "litellm_call_id", "unknown")
verbose_logger.exception(
"LiteLLM.AgenticHookError: Exception in async_should_run_agentic_loop [call_id=%s model=%s]: %s",
_call_id,
logging_obj.litellm_call_id,
model,
str(e),
)
@ -5645,10 +5644,9 @@ class BaseLLMHTTPHandler:
except AgenticLoopSafetyError as e:
if not self._can_replace_turn_with_terminal_response(stream, api_surface):
raise
_call_id = getattr(logging_obj, "litellm_call_id", "unknown")
verbose_logger.warning(
"LiteLLM.AgenticLoopRefused: ending turn [call_id=%s model=%s]: %s",
_call_id,
logging_obj.litellm_call_id,
model,
str(e),
)

81
litellm/llms/together_ai/chat/transformation.py
View file

@ -4,8 +4,7 @@ Translates from OpenAI's `/v1/chat/completions` to Together AI's `/v1/chat/compl
Docs: https://docs.together.ai/docs/chat-overview
"""
from collections.abc import Container, Coroutine
from types import MappingProxyType
from collections.abc import Callable, Container, Coroutine
from typing import (
Final,
Literal,
@ -17,26 +16,42 @@ import litellm
from litellm._logging import verbose_logger
from litellm.exceptions import UnsupportedParamsError
from litellm.types.llms.openai import AllMessageValues
from litellm.utils import supports_function_calling
from litellm.utils import supports_function_calling, supports_response_schema
from ...openai.chat.gpt_transformation import OpenAIGPTConfig
TOOL_CALLING_PARAMS: Final = ("tools", "tool_choice", "function_call")
LITELLM_INTERNAL_ASSISTANT_FIELDS: Final = frozenset({"thinking_blocks", "provider_specific_fields"})
PLAIN_TEXT_RESPONSE_FORMAT: Final = MappingProxyType({"type": "text"})
FUNCTION_CALLING_DOCS_URL: Final = "https://docs.together.ai/docs/function-calling"
STRUCTURED_OUTPUTS_DOCS_URL: Final = "https://docs.together.ai/docs/inference/chat/structured-outputs"
def _registry_verdict(model: str, flag: str, check: Callable[[str], bool]) -> bool | None:
try:
if check(model):
return True
except Exception as e:
verbose_logger.debug("Error checking together_ai %s for %s: %s", flag, model, e)
registry_entry: Final = litellm.model_cost.get(f"together_ai/{model}")
if isinstance(registry_entry, dict) and registry_entry.get(flag) is False:
return False
return None
def _function_calling_verdict(model: str) -> bool | None:
try:
if supports_function_calling(model, custom_llm_provider="together_ai"):
return True
except Exception as e:
verbose_logger.debug("Error checking together_ai function calling support for %s: %s", model, e)
registry_entry: Final = litellm.model_cost.get(f"together_ai/{model}")
if isinstance(registry_entry, dict) and registry_entry.get("supports_function_calling") is False:
return False
return None
return _registry_verdict(
model,
"supports_function_calling",
lambda checked_model: supports_function_calling(checked_model, custom_llm_provider="together_ai"),
)
def _response_schema_verdict(model: str) -> bool | None:
return _registry_verdict(
model,
"supports_response_schema",
lambda checked_model: supports_response_schema(checked_model, custom_llm_provider="together_ai"),
)
def _tool_params_to_drop(passed_params: Container[str], model: str, drop_params: bool) -> tuple[str, ...]:
@ -68,6 +83,32 @@ def _tool_params_to_drop(passed_params: Container[str], model: str, drop_params:
)
def _drop_response_format(passed_params: Container[str], model: str, drop_params: bool) -> bool:
if "response_format" not in passed_params:
return False
verdict: Final = _response_schema_verdict(model)
if verdict is True:
return False
if verdict is None:
verbose_logger.warning(
"together_ai model %s has no structured outputs entry in the model registry; passing response_format through for Together to validate. Docs - %s",
model,
STRUCTURED_OUTPUTS_DOCS_URL,
)
return False
if drop_params or litellm.drop_params:
verbose_logger.warning(
"together_ai model %s does not support structured outputs per the model registry; dropping response_format. Docs - %s",
model,
STRUCTURED_OUTPUTS_DOCS_URL,
)
return True
raise UnsupportedParamsError(
status_code=500,
message=f"together_ai does not support parameters: response_format, for model={model}. To drop it from the call, set `litellm.drop_params = True`.",
)
def _without_litellm_internal_fields(message: AllMessageValues) -> AllMessageValues:
if message["role"] != "assistant" or LITELLM_INTERNAL_ASSISTANT_FIELDS.isdisjoint(message):
return message
@ -112,18 +153,6 @@ class TogetherAIChatConfig(OpenAIGPTConfig):
return super()._transform_messages(stripped, model, is_async=True)
return super()._transform_messages(stripped, model, is_async=False)
def get_supported_openai_params(self, model: str) -> list:
supports_fc: Final = _function_calling_verdict(model)
supported_params: Final = super().get_supported_openai_params(model)
if supports_fc is True:
return supported_params
verbose_logger.debug(
"Only some together models support response_format. Docs - https://docs.together.ai/docs/function-calling"
)
return [ # mutable-ok: the inherited contract returns a plain list; building fresh avoids mutating the base class's value
param for param in supported_params if param != "response_format"
]
def map_openai_params(
self,
non_default_params: dict,
@ -134,6 +163,6 @@ class TogetherAIChatConfig(OpenAIGPTConfig):
mapped_openai_params: Final = super().map_openai_params(non_default_params, optional_params, model, drop_params)
for param in _tool_params_to_drop(mapped_openai_params, model, drop_params):
mapped_openai_params.pop(param)
if mapped_openai_params.get("response_format") == PLAIN_TEXT_RESPONSE_FORMAT:
if _drop_response_format(mapped_openai_params, model, drop_params):
mapped_openai_params.pop("response_format")
return mapped_openai_params

4
litellm/model_prices_and_context_window_backup.json
View file

@ -12282,7 +12282,7 @@
},
"claude-3-haiku-20240307": {
"cache_creation_input_token_cost": 3e-07,
"cache_creation_input_token_cost_above_1hr": 6e-06,
"cache_creation_input_token_cost_above_1hr": 5e-07,
"cache_read_input_token_cost": 3e-08,
"deprecation_date": "2026-04-20",
"input_cost_per_token": 2.5e-07,
@ -12301,7 +12301,7 @@
},
"claude-3-opus-20240229": {
"cache_creation_input_token_cost": 1.875e-05,
"cache_creation_input_token_cost_above_1hr": 6e-06,
"cache_creation_input_token_cost_above_1hr": 3e-05,
"cache_read_input_token_cost": 1.5e-06,
"deprecation_date": "2026-01-05",
"input_cost_per_token": 1.5e-05,

6
litellm/proxy/_types.py
View file

@ -2819,7 +2819,7 @@ class UserAPIKeyAuth(LiteLLM_VerificationTokenView): # the expected response ob
# Values stay `object` rather than BudgetConfig: this is the raw JSON column,
# and validating it here would make one malformed row fail auth outright.
# resolve_model_budget validates the single entry a request actually needs.
user_model_max_budget: dict[str, object] | None = None
user_model_max_budget: Mapping[str, object] | None = None
request_route: str | None = None
is_session_token: bool = False
# Server-only marker set exclusively by the MCP gateway admission path
@ -2997,8 +2997,8 @@ class UserInfoV2Response(LiteLLMPydanticObjectBase):
sso_user_id: str | None = None
teams: list[str] = [] # Just team IDs, not full team objects
object_permission: LiteLLM_ObjectPermissionTable | None = None
model_max_budget: dict | None = None
model_max_budget_usage: dict | None = None
model_max_budget: Mapping[str, object] | None = None
model_max_budget_usage: Mapping[str, Mapping[str, object]] | None = None
from litellm.models.config import LiteLLM_Config as LiteLLM_Config # noqa: E402

11
litellm/proxy/auth/user_api_key_auth.py
View file

@ -212,9 +212,9 @@ async def _read_user_model_max_budget(
user_id: str | None,
prisma_client: PrismaClient | None,
user_api_key_cache: UserApiKeyCache,
parent_otel_span: object,
parent_otel_span: Span | None,
proxy_logging_obj: ProxyLogging,
) -> dict | None:
) -> Mapping[str, object] | None:
"""The user row's `model_max_budget`, or None when the row cannot be read.
A user whose row is missing must not be refused: this is a budget lookup,
@ -228,13 +228,13 @@ async def _read_user_model_max_budget(
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
user_id_upsert=False,
parent_otel_span=parent_otel_span, # pyright: ignore[reportArgumentType] # Span is a runtime union, not usable in an annotation here
parent_otel_span=parent_otel_span,
proxy_logging_obj=proxy_logging_obj,
)
except Exception as e: # noqa: BLE001 # mirrors the main path's tolerance
verbose_logger.debug("Unable to read user for the per-model budget check: %s", e)
return None
return getattr(user_obj, "model_max_budget", None)
return user_obj.model_max_budget if user_obj is not None else None
async def _check_user_model_budget(
@ -3267,8 +3267,7 @@ async def _run_post_custom_auth_checks(
# loaded the user row yet. The attach is unconditional because the post-call
# spend hook reads this field off the token: gating it on the same condition
# as enforcement would leave the user's counter uncharged whenever this
# request was not itself enforceable, which is the untracked-spend bug this
# PR exists to fix.
# request was not itself enforceable, so its spend would go untracked.
user_budget: Final = await _read_user_model_max_budget(
user_id=valid_token.user_id,
prisma_client=prisma_client,

9
litellm/proxy/common_request_processing.py
View file

@ -4,7 +4,7 @@ import json
import logging
import math
import traceback
from collections.abc import AsyncGenerator, Awaitable, Callable, Mapping
from collections.abc import AsyncGenerator, Awaitable, Callable, Mapping, Sequence
from datetime import datetime
from functools import lru_cache
from types import MappingProxyType
@ -284,7 +284,7 @@ def _deferred_stream_logging_is_armed(request_data: dict) -> bool:
)
def _assembled_model_came_from_a_later_chunk(chunks: list, assembled_model: object) -> bool:
def _assembled_model_came_from_a_later_chunk(chunks: Sequence[object], assembled_model: object) -> bool:
"""Report whether stream_chunk_builder picked a model the first chunk did not carry.
Azure Model Router puts the routed model on the chunks after the first one, and the
@ -306,7 +306,10 @@ def _assembled_model_came_from_a_later_chunk(chunks: list, assembled_model: obje
)
def _assembled_model_is_the_name_the_client_asked_for(request_data: dict, assembled_model: object) -> bool:
def _assembled_model_is_the_name_the_client_asked_for(
request_data: Mapping[str, object],
assembled_model: object,
) -> bool:
"""Report whether the assembled model is the public name the proxy stamps onto chunks.
That stamp is what leaves an unpriced alias on the partial response, so the deployment's

2
litellm/proxy/common_utils/reset_budget_job.py
View file

@ -1182,7 +1182,7 @@ class ResetBudgetJob:
if not raw:
continue
row_id: str = row[source.id_column]
windows: list = raw if isinstance(raw, list) else json.loads(raw)
windows: list[dict[str, object]] = raw if isinstance(raw, list) else json.loads(raw)
changed = False
for window in windows:
counter_key = f"{source.counter_prefix}:{row_id}:window:{window['budget_duration']}"

11
litellm/proxy/common_utils/sse_keepalive.py
View file

@ -91,6 +91,17 @@ def is_sse_content_type(content_type: str | None) -> bool:
return content_type is not None and content_type.split(";", 1)[0].strip().lower() == _SSE_MEDIA_TYPE
def split_complete_sse_frames(pending: bytes) -> tuple[bytes, bytes]:
"""Split buffered SSE bytes into ``(complete_frames, unterminated_tail)``."""
boundary_end: Final = max(
(pending.rfind(delimiter) + len(delimiter) for delimiter in _SSE_FRAME_DELIMITERS if delimiter in pending),
default=0,
)
if boundary_end == 0:
return b"", pending
return pending[:boundary_end], pending[boundary_end:]
def wrap_passthrough_sse_bytes_with_keepalive_pings(
stream: AsyncGenerator[bytes, None],
ping_interval_seconds: float | str | None,

17
litellm/proxy/db/db_url_settings.py
View file

@ -60,6 +60,11 @@ AzureTokenAuthFlag = Annotated[
bool, BeforeValidator(partial(token_auth_flag_enabled, env_var=AZURE_POSTGRESQL_AUTH_ENV_VAR))
]
DISABLE_PREPARED_STATEMENTS_ENV_VAR: Final = "DATABASE_DISABLE_PREPARED_STATEMENTS"
DisablePreparedStatementsFlag = Annotated[
bool, BeforeValidator(partial(token_auth_flag_enabled, env_var=DISABLE_PREPARED_STATEMENTS_ENV_VAR))
]
# schema.prisma pins `provider = "postgresql"`, so these are the only schemes
# Prisma can actually connect with.
SUPPORTED_DB_SCHEMES: Final[frozenset[str]] = frozenset({"postgresql", "postgres"})
@ -153,6 +158,9 @@ class DatabaseURLSettings(BaseSettings):
iam_token_db_auth: IamTokenAuthFlag = Field(default=False, validation_alias=IAM_TOKEN_DB_AUTH_ENV_VAR)
azure_postgresql_auth: AzureTokenAuthFlag = Field(default=False, validation_alias=AZURE_POSTGRESQL_AUTH_ENV_VAR)
disable_prepared_statements: DisablePreparedStatementsFlag = Field(
default=False, validation_alias=DISABLE_PREPARED_STATEMENTS_ENV_VAR
)
# Writer
database_url: str | None = Field(default=None, validation_alias="DATABASE_URL")
@ -375,6 +383,15 @@ class DatabaseURLSettings(BaseSettings):
self._raise_for_unsupported_scheme()
wrote_writer: Final = self.apply_writer_url_to_env()
# DATABASE_DISABLE_PREPARED_STATEMENTS maps to Prisma's `pgbouncer=true`
# URL param, same as the CLI's `database_disable_prepared_statements`
# config key. An explicit `pgbouncer` value already on the URL wins.
if self.disable_prepared_statements:
for env_var in ("DATABASE_URL", "DIRECT_URL"):
url = os.environ.get(env_var)
if url:
os.environ[env_var] = add_missing_query_params(url, MappingProxyType({"pgbouncer": "true"}))
# The reader inherits the writer's connection params (pool size, timeouts,
# pgbouncer mode). Without this the reader pool ignores the configured cap
# and falls back to Prisma's `num_physical_cpus * 2 + 1` default.

2
litellm/proxy/db/token_auth.py
View file

@ -62,7 +62,7 @@ def token_auth_flag_enabled(value: str | bool | None, *, env_var: str) -> bool:
return False
raise ValueError(
f"{env_var}={value!r} is not a recognized boolean. Set it to one of "
f"{', '.join(sorted(TRUTHY_TOKEN_AUTH_VALUES))} to turn token auth on, or to one of "
f"{', '.join(sorted(TRUTHY_TOKEN_AUTH_VALUES))} to turn it on, or to one of "
f"{', '.join(sorted(v for v in FALSY_TOKEN_AUTH_VALUES if v))} to turn it off."
)

2
litellm/proxy/health_check.py
View file

@ -183,7 +183,7 @@ async def run_with_timeout(task, timeout):
return {"error": "Timeout exceeded", "exception": timeout_exception}
def _is_strategy_router_deployment(litellm_params: dict) -> bool:
def _is_strategy_router_deployment(litellm_params: Mapping[str, object]) -> bool:
"""True for strategy-router deployments."""
model: Final[object] = litellm_params.get("model", "")
return isinstance(model, str) and classify_strategy_router_model(model) is not None

18
litellm/proxy/hooks/responses_id_security.py
View file

@ -28,6 +28,21 @@ if TYPE_CHECKING:
from litellm.proxy._types import UserAPIKeyAuth
_RESPONSES_API_PROVIDER_PREFIX: Final = "/openai"
_RESPONSES_API_CREATE_ROUTES: Final = frozenset({"/v1/responses", "/responses"})
def _is_responses_api_create_route(request_route: str | None) -> bool:
if request_route is None:
return False
canonical: Final = (
request_route[len(_RESPONSES_API_PROVIDER_PREFIX) :]
if request_route.startswith(_RESPONSES_API_PROVIDER_PREFIX + "/")
else request_route
)
return canonical in _RESPONSES_API_CREATE_ROUTES
class ResponsesIDSecurity(CustomLogger):
def __init__(self):
pass
@ -267,8 +282,7 @@ class ResponsesIDSecurity(CustomLogger):
async for chunk in response:
if (
isinstance(chunk, BaseLiteLLMOpenAIResponseObject)
and user_api_key_dict.request_route
== "/v1/responses" # only encrypt the response id for the responses api
and _is_responses_api_create_route(user_api_key_dict.request_route)
and not general_settings.get("disable_responses_id_security", False)
):
chunk = self._encrypt_response_id(chunk, user_api_key_dict, request_encryption_cache)

4
litellm/proxy/litellm_pre_call_utils.py
View file

@ -4,7 +4,7 @@ import json
import re
import time
from collections import OrderedDict
from collections.abc import Mapping
from collections.abc import Mapping, MutableMapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final
@ -1629,7 +1629,7 @@ class LiteLLMProxyRequestSetup:
def refresh_proxy_server_request_body_snapshot(
data: dict, # mutable-ok: mutates proxy_server_request.body in place on the shared request dict
data: MutableMapping[str, object],
) -> None:
"""
Re-snapshot ``data["proxy_server_request"]["body"]`` from the current state of ``data``.

3
litellm/proxy/management_endpoints/key_management_endpoints.py
View file

@ -2294,10 +2294,9 @@ async def _process_single_key_update(
prisma_client=prisma_client,
)
_existing_row_metadata: Final = getattr(existing_key_row, "metadata", None)
enforce_batch_enqueued_token_limit_is_admin_only(
data=update_key_request,
existing_metadata=_existing_row_metadata if isinstance(_existing_row_metadata, dict) else None,
existing_metadata=existing_key_row.metadata,
user_api_key_dict=user_api_key_dict,
entity="key",
)

9
litellm/proxy/openai_files_endpoints/common_utils.py
View file

@ -1354,11 +1354,12 @@ def _completed_batch_safe_to_retire(response: "LiteLLMBatch") -> bool:
reports no successful request lines. When counts are unknown, stay eligible so
the next poller pass revisits it. (#37713)
"""
if getattr(response, "output_file_id", None) is not None:
if response.output_file_id is not None:
return True
request_counts = getattr(response, "request_counts", None)
completed = getattr(request_counts, "completed", None)
return completed == 0
request_counts = response.request_counts
if request_counts is None:
return False
return request_counts.completed == 0
async def update_batch_in_database(

120
litellm/proxy/pass_through_endpoints/managed_id_rewriter.py
View file

@ -32,7 +32,7 @@ from __future__ import annotations
import json
import re
from collections.abc import Callable, Mapping, Sequence
from collections.abc import AsyncGenerator, Callable, Mapping, Sequence
from typing import (
TYPE_CHECKING,
Final,
@ -43,7 +43,7 @@ from typing import (
from urllib.parse import quote, unquote
from fastapi import HTTPException
from pydantic import JsonValue
from pydantic import JsonValue, TypeAdapter, ValidationError
from litellm._logging import verbose_proxy_logger
from litellm.llms.base_llm.managed_resources.isolation import (
@ -52,6 +52,7 @@ from litellm.llms.base_llm.managed_resources.isolation import (
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.batches_endpoints.common_utils import validate_batch_list_limit
from litellm.proxy.common_utils.sse_keepalive import split_complete_sse_frames
from litellm.repositories.table_repositories import (
ManagedFileRepository,
ManagedObjectRepository,
@ -820,6 +821,121 @@ async def rewrite_response_ids(
return mutated if changed else body
_RESPONSE_ID_PREFIX: Final = "resp_"
_STREAMED_RESPONSE_ID_SPEC: Final[_FieldSpec] = ("id", _RESPONSE_ID_PREFIX)
_SSE_DATA_PREFIX: Final = "data:"
_SSE_EVENT_ADAPTER: Final = TypeAdapter(Mapping[str, JsonValue])
def _first_streamed_response(frames: bytes) -> tuple[str, Mapping[str, JsonValue]] | None:
for line in frames.decode("utf-8", errors="replace").splitlines():
if not line.startswith(_SSE_DATA_PREFIX):
continue
try:
event = _SSE_EVENT_ADAPTER.validate_json(line[len(_SSE_DATA_PREFIX) :])
except ValidationError:
continue
response = event.get("response")
if not isinstance(response, dict):
continue
raw_id = response.get("id")
if isinstance(raw_id, str) and raw_id.startswith(_RESPONSE_ID_PREFIX):
return raw_id, response
return None
class _StreamedResponseIdRewriter:
__slots__ = ("_is_create_route", "_pending", "_prisma_client", "_provider", "_replacement", "_user_api_key_dict")
def __init__(
self,
provider: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: PrismaClient,
is_create_route: bool,
) -> None:
self._provider: Final = provider
self._user_api_key_dict: Final = user_api_key_dict
self._prisma_client: Final = prisma_client
self._is_create_route: Final = is_create_route
self._pending = b""
self._replacement: tuple[bytes, bytes] | None = None
async def feed(self, chunk: bytes) -> bytes:
complete_frames, self._pending = split_complete_sse_frames(self._pending + chunk)
if not complete_frames:
return b""
if self._replacement is None:
self._replacement = await self._mint(complete_frames)
return self._rewrite(complete_frames)
def flush(self) -> bytes:
tail: Final = self._pending
self._pending = b""
return self._rewrite(tail)
async def _mint(self, frames: bytes) -> tuple[bytes, bytes] | None:
first: Final = _first_streamed_response(frames)
if first is None:
return None
raw_id, snapshot = first
managed_id: Final = await _mint_or_reuse_object(
raw_id,
self._provider,
"response",
snapshot,
self._user_api_key_dict,
self._prisma_client,
self._is_create_route,
)
return raw_id.encode(), managed_id.encode()
def _rewrite(self, frames: bytes) -> bytes:
if self._replacement is None:
return frames
raw_id, managed_id = self._replacement
return frames.replace(raw_id, managed_id)
async def rewrite_streamed_response_ids(
stream: AsyncGenerator[bytes, None],
provider: str,
method: str,
route: str,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: PrismaClient,
) -> AsyncGenerator[bytes, None]:
"""
Record ownership of the response object streamed back by a Responses API
passthrough and swap its managed id into every SSE frame, so a streamed
response is owned and resolved exactly like a non-streamed one.
Streams for any other ``(provider, method, route)`` are relayed untouched.
"""
from litellm.proxy.auth.auth_utils import normalize_request_route
canonical: Final = normalize_request_route(_canonical_path(route))
field_specs: Final = BUILTIN_OUTPUT_ID_FIELD_MAP.get((provider, method, canonical), ())
if _STREAMED_RESPONSE_ID_SPEC not in field_specs:
async for chunk in stream:
yield chunk
return
rewriter: Final = _StreamedResponseIdRewriter(
provider=provider,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
is_create_route="{" not in canonical,
)
async for chunk in stream:
rewritten_frames = await rewriter.feed(chunk)
if rewritten_frames:
yield rewritten_frames
tail: Final = rewriter.flush()
if tail:
yield tail
# ---------------------------------------------------------------------------
# List-route interception — serve listing entirely from DB
# ---------------------------------------------------------------------------

72
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -1209,14 +1209,19 @@ async def pass_through_request(
return StreamingResponse(
wrap_passthrough_sse_bytes_with_keepalive_pings(
stream=PassThroughStreamingHandler.chunk_processor(
response=response,
request_body=_parsed_body,
litellm_logging_obj=logging_obj,
endpoint_type=endpoint_type,
start_time=start_time,
passthrough_success_handler_obj=pass_through_endpoint_logging,
url_route=str(url),
stream=_own_streamed_managed_ids(
stream=PassThroughStreamingHandler.chunk_processor(
response=response,
request_body=_parsed_body,
litellm_logging_obj=logging_obj,
endpoint_type=endpoint_type,
start_time=start_time,
passthrough_success_handler_obj=pass_through_endpoint_logging,
url_route=str(url),
),
managed_id_provider=_managed_id_provider,
request=request,
user_api_key_dict=user_api_key_dict,
),
ping_interval_seconds=litellm.sse_keepalive_ping_interval_seconds,
upstream_headers=response.headers,
@ -1285,14 +1290,19 @@ async def pass_through_request(
return StreamingResponse(
wrap_passthrough_sse_bytes_with_keepalive_pings(
stream=PassThroughStreamingHandler.chunk_processor(
response=response,
request_body=_parsed_body,
litellm_logging_obj=logging_obj,
endpoint_type=endpoint_type,
start_time=start_time,
passthrough_success_handler_obj=pass_through_endpoint_logging,
url_route=str(url),
stream=_own_streamed_managed_ids(
stream=PassThroughStreamingHandler.chunk_processor(
response=response,
request_body=_parsed_body,
litellm_logging_obj=logging_obj,
endpoint_type=endpoint_type,
start_time=start_time,
passthrough_success_handler_obj=pass_through_endpoint_logging,
url_route=str(url),
),
managed_id_provider=_managed_id_provider,
request=request,
user_api_key_dict=user_api_key_dict,
),
ping_interval_seconds=litellm.sse_keepalive_ping_interval_seconds,
upstream_headers=response.headers,
@ -2441,6 +2451,36 @@ def _is_streaming_response(response: httpx.Response) -> bool:
return False
def _own_streamed_managed_ids(
stream: AsyncGenerator[bytes, None],
managed_id_provider: str | None,
request: Request,
user_api_key_dict: UserAPIKeyAuth,
) -> AsyncGenerator[bytes, None]:
from litellm.proxy.proxy_server import general_settings, prisma_client, proxy_logging_obj
if (
managed_id_provider is None
or not general_settings.get("passthrough_managed_object_ids", False)
or prisma_client is None
or proxy_logging_obj.get_proxy_hook("managed_files") is None
):
return stream
from litellm.proxy.auth.auth_utils import get_request_route
from litellm.proxy.pass_through_endpoints.managed_id_rewriter import (
rewrite_streamed_response_ids,
)
return rewrite_streamed_response_ids(
stream=stream,
provider=managed_id_provider,
method=request.method,
route=get_request_route(request),
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
def _should_buffer_passthrough_response(response: httpx.Response) -> bool:
"""
Decide from the response headers whether the body must be read into memory.

14
litellm/proxy/pass_through_endpoints/streaming_handler.py
View file

@ -10,6 +10,7 @@ from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.proxy._types import PassThroughEndpointLoggingResultValues
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
from litellm.proxy.common_utils.sse_keepalive import split_complete_sse_frames
from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
from litellm.types.utils import StandardPassThroughResponseObject
@ -101,7 +102,7 @@ class PassThroughStreamingHandler:
async for chunk in response.aiter_bytes():
raw_bytes.append(chunk)
PassThroughStreamingHandler._stamp_first_chunk_if_needed(litellm_logging_obj)
complete_frames, pending = PassThroughStreamingHandler._split_complete_sse_frames(
complete_frames, pending = split_complete_sse_frames(
pending + chunk
) # rebind-ok: SSE frame reassembly buffer across transport chunks
if complete_frames:
@ -139,17 +140,6 @@ class PassThroughStreamingHandler:
except Exception as e:
verbose_proxy_logger.error("Error scheduling chunk_processor logging: %s", e)
@staticmethod
def _split_complete_sse_frames(pending: bytes) -> tuple[bytes, bytes]:
lf_boundary_end: Final = pending.rfind(b"\n\n") + 2
crlf_boundary_end: Final = pending.rfind(b"\r\n\r\n") + 4
boundary_end: Final = max(
lf_boundary_end if lf_boundary_end >= 2 else 0, crlf_boundary_end if crlf_boundary_end >= 4 else 0
)
if boundary_end == 0:
return b"", pending
return pending[:boundary_end], pending[boundary_end:]
@staticmethod
async def _route_streaming_logging_to_handler(
litellm_logging_obj: LiteLLMLoggingObj,

6
litellm/proxy/proxy_server.py
View file

@ -4101,7 +4101,7 @@ def resolve_complexity_router_plugins(
complexity_router_config["classifier_plugin"] = resolved_classifier # rebind-ok: out-param, resolved in place
def validate_deployment_max_agentic_loops(model: Mapping[str, Any]) -> None:
def validate_deployment_max_agentic_loops(model: Mapping[str, object]) -> None:
"""
Reject a per-deployment `max_agentic_loops` the agentic loop cannot honor.
@ -4111,7 +4111,9 @@ def validate_deployment_max_agentic_loops(model: Mapping[str, Any]) -> None:
start. Left unchecked entirely, a `0` used to read as the default ceiling
of 3 and a non-integer failed every request to that model instead.
"""
litellm_params: Final = model.get("litellm_params") or {}
litellm_params: Final = model.get("litellm_params")
if not isinstance(litellm_params, Mapping):
return
if "max_agentic_loops" not in litellm_params:
return

2
litellm/proxy/spend_tracking/budget_reservation.py
View file

@ -1267,7 +1267,7 @@ def _count_input_tokens_for_models(
_INPUT_SIZE_FIELDS: Final = ("messages", "prompt", "input", "query", "documents", "tools", "tool_choice")
def _approximate_input_size(request_body: dict) -> int:
def _approximate_input_size(request_body: Mapping[str, object]) -> int:
"""Length of the request's input text, a cheap stand-in for tokenizing cost.
Every field _count_input_tokens hands the tokenizer is sized here, and

2
litellm/repositories/model_repository.py
View file

@ -32,7 +32,7 @@ class _PrismaClientView(Protocol):
class ModelRepository(BaseRepository[LiteLLM_ProxyModelTable]):
"""Repository for proxy model database operations with encryption support."""
def __init__(self, prisma_client: object, encryption_key: str | None = None):
def __init__(self, prisma_client: object, encryption_key: str | None = None) -> None:
super().__init__(prisma_client)
self._encryption_key = encryption_key

30
litellm/responses/litellm_completion_transformation/transformation.py
View file

@ -1300,16 +1300,14 @@ class LiteLLMCompletionResponsesConfig:
if isinstance(content, str) and content.strip():
return content
if isinstance(content, list):
text_parts: Final[list[str]] = [] # mutable-ok: text accumulator
for block in content:
if not isinstance(block, Mapping):
continue
block_type = block.get("type")
if block_type in ("encrypted_content", "redacted_thinking"):
continue
text = block.get("text")
if isinstance(text, str) and text.strip():
text_parts.append(text.strip())
text_parts: Final = tuple(
text.strip()
for block in content
if isinstance(block, Mapping)
and block.get("type") not in ("encrypted_content", "redacted_thinking")
and isinstance(text := block.get("text"), str)
and text.strip()
)
if text_parts:
return "\n".join(text_parts)
return None
@ -1325,13 +1323,11 @@ class LiteLLMCompletionResponsesConfig:
summary: Final[object] = input_item.get("summary")
if not isinstance(summary, list):
return None
text_parts: Final[list[str]] = [] # mutable-ok: text accumulator
for block in summary:
if not isinstance(block, Mapping):
continue
text = block.get("text")
if isinstance(text, str) and text.strip():
text_parts.append(text.strip())
text_parts: Final = tuple(
text.strip()
for block in summary
if isinstance(block, Mapping) and isinstance(text := block.get("text"), str) and text.strip()
)
return "\n".join(text_parts) if text_parts else None
@staticmethod

7
litellm/router.py
View file

@ -7344,7 +7344,7 @@ class Router:
):
raise error # then raise the error
if isinstance(error, openai.AuthenticationError):
if isinstance(error, (openai.AuthenticationError, openai.PermissionDeniedError)):
"""
- if other deployments available -> retry
- else -> raise error
@ -10697,10 +10697,9 @@ class Router:
_router_model_name: str = model_value
elif isinstance(model_value, dict):
_model_value = RouterModelGroupAliasItem(**model_value)
if _model_value["hidden"] is True:
if _model_value["hidden"] is True and model_name is None:
continue
else:
_router_model_name = _model_value["model"]
_router_model_name = _model_value["model"]
else:
continue

16
litellm/types/integrations/prometheus.py
View file

@ -3,7 +3,7 @@ from collections.abc import Mapping, Sequence
from dataclasses import MISSING, dataclass, field, fields
from enum import Enum
from types import MappingProxyType
from typing import Any, ClassVar, Final, Literal
from typing import Any, ClassVar, Final, Literal, cast
import litellm
@ -326,21 +326,25 @@ def validate_prometheus_deployment_and_latency_caller_identity() -> str:
)
def validate_caller_identity_settings(litellm_settings: Mapping[str, Any]) -> None:
def validate_caller_identity_settings(litellm_settings: Mapping[str, object]) -> None:
"""Store the caller-identity mode from litellm_settings and validate it together
with prometheus_metrics_config, raising on an invalid value or on include_labels
that request a label the selected mode removes."""
if "prometheus_deployment_and_latency_caller_identity" not in litellm_settings:
return
litellm.prometheus_deployment_and_latency_caller_identity = litellm_settings[
"prometheus_deployment_and_latency_caller_identity"
]
litellm.prometheus_deployment_and_latency_caller_identity = (
cast( # cast-ok: validated on the next line, which raises on an invalid value
'Literal["api_key_alias", "user_email", "both"]',
litellm_settings["prometheus_deployment_and_latency_caller_identity"],
)
)
caller_identity_mode: Final = validate_prometheus_deployment_and_latency_caller_identity()
if caller_identity_mode != "user_email":
return
raw_metrics_config: Final = litellm_settings.get("prometheus_metrics_config")
conflicting_metrics: Final = tuple(
metric_name
for metric_config in (litellm_settings.get("prometheus_metrics_config") or ())
for metric_config in (raw_metrics_config if isinstance(raw_metrics_config, list) else ())
if isinstance(metric_config, dict) and "api_key_alias" in (metric_config.get("include_labels") or ())
for metric_name in (metric_config.get("metrics") or ())
if metric_name in PROMETHEUS_DEPLOYMENT_AND_LATENCY_CALLER_IDENTITY_METRICS

1
litellm/types/llms/anthropic.py
View file

@ -36,6 +36,7 @@ AnthropicInputSchema = TypedDict(
class AnthropicOutputSchema(TypedDict, total=False):
type: Required[Literal["json_schema"]]
schema: Required[dict]
strict: ReadOnly[bool]
class AnthropicOutputConfig(TypedDict, total=False):

2
litellm/utils.py
View file

@ -1292,7 +1292,7 @@ async def async_post_call_success_deployment_hook(
async def async_post_call_failure_deployment_hook(
request_data: Mapping[str, Any], exception: Exception, call_type: str
request_data: Mapping[str, object], exception: Exception, call_type: str
) -> None:
"""
Notify CustomLogger callbacks that a deployment attempt failed.

4
model_prices_and_context_window.json
View file

@ -12282,7 +12282,7 @@
},
"claude-3-haiku-20240307": {
"cache_creation_input_token_cost": 3e-07,
"cache_creation_input_token_cost_above_1hr": 6e-06,
"cache_creation_input_token_cost_above_1hr": 5e-07,
"cache_read_input_token_cost": 3e-08,
"deprecation_date": "2026-04-20",
"input_cost_per_token": 2.5e-07,
@ -12301,7 +12301,7 @@
},
"claude-3-opus-20240229": {
"cache_creation_input_token_cost": 1.875e-05,
"cache_creation_input_token_cost_above_1hr": 6e-06,
"cache_creation_input_token_cost_above_1hr": 3e-05,
"cache_read_input_token_cost": 1.5e-06,
"deprecation_date": "2026-01-05",
"input_cost_per_token": 1.5e-05,

6
ruff-strict-budget.json
View file

@ -12,10 +12,10 @@
"limit": 2012
},
"ANN202": {
"limit": 852
"limit": 847
},
"ANN204": {
"limit": 711
"limit": 706
},
"ANN205": {
"limit": 112
@ -24,7 +24,7 @@
"limit": 133
},
"ANN401": {
"limit": 1157
"limit": 1153
},
"ASYNC230": {
"limit": 11

18
tests/e2e/coverage_registry/README.md
View file

@ -77,6 +77,24 @@ Strict mode exits non-zero on `@pytest.mark.covers(...)` ids that are not checke
the registry. Add `--fail-on-collection-errors` when the job should also fail on pytest
collection errors.
## Provider x feature matrix: customer-run Bedrock combinations
The provider and feature combinations customers actually run get explicit cells, expanded
here as incidents surface new ones. The current Bedrock set, seeded from a customer's
production shape (regional `us.anthropic.*` inference-profile ids over both chat routes,
provider response headers for AWS-side correlation, and the Test Connection probe for a
responses-mode Bedrock Mantle deployment):
| Cell | Feature | Covering test |
|------|---------|---------------|
| `llm.chat_completions.bedrock_converse.basic.nonstream.works` | regional `us.` id, Converse | `llm_translation/test_chat_completions_regression_e2e.py` |
| `llm.chat_completions.bedrock_converse.basic.stream.works` | regional `us.` id, Converse stream | `llm_translation/test_chat_completions_regression_e2e.py` |
| `llm.chat_completions.bedrock_invoke.basic.nonstream.works` | regional `us.` id, Invoke | `llm_translation/test_bedrock_provider_matrix_e2e.py` |
| `llm.chat_completions.bedrock_invoke.basic.stream.works` | regional `us.` id, Invoke stream | `llm_translation/test_bedrock_provider_matrix_e2e.py` |
| `llm.chat_completions.bedrock_converse.response_headers.nonstream.works` | `llm_provider-*` headers | `llm_translation/test_bedrock_provider_matrix_e2e.py` |
| `llm.chat_completions.bedrock_converse.response_headers.stream.works` | `llm_provider-*` headers, stream | `llm_translation/test_bedrock_provider_matrix_e2e.py` |
| `mgmt.model.test_connection.happy_path` | Test Connection, Bedrock Mantle | `management/test_model_test_connection_e2e.py` |
## Status: this is a draft for review
The cells were enumerated from the codebase and the tiers are a first proposal. Known

4
tests/e2e/coverage_registry/llm_conversational.yaml
View file

@ -29,6 +29,10 @@
- {id: llm.chat_completions.bedrock_converse.vision.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: bedrock_converse, capability: vision, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Bedrock vision (Anthropic/Nova)"}
- {id: llm.chat_completions.bedrock_converse.prompt_cache_5m.nonstream.works, module: llm, tier: P1, subject_endpoint: chat_completions, route: bedrock_converse, capability: prompt_cache_5m, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Anthropic-on-Bedrock caching"}
- {id: llm.chat_completions.bedrock_converse.thinking.nonstream.works, module: llm, tier: P1, subject_endpoint: chat_completions, route: bedrock_converse, capability: thinking, streaming: nonstream, assertions: [works], source: "model_prices json", rationale: "Anthropic thinking on Bedrock"}
- {id: llm.chat_completions.bedrock_converse.response_headers.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: bedrock_converse, capability: response_headers, streaming: nonstream, assertions: [works], source: "llms/bedrock/chat/converse_handler.py:248", rationale: "Bedrock request ids must surface as llm_provider-* response headers on /chat/completions so callers can correlate calls with AWS-side logs (#37003)", fail_before_fix: proven}
- {id: llm.chat_completions.bedrock_converse.response_headers.stream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: bedrock_converse, capability: response_headers, streaming: stream, assertions: [works], source: "llms/bedrock/chat/converse_handler.py:154", rationale: "The llm_provider-* headers must also surface on streaming /chat/completions, where CustomStreamWrapper carries them instead of the nonstream setter"}
- {id: llm.chat_completions.bedrock_invoke.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: bedrock_invoke, capability: basic, streaming: nonstream, assertions: [works], source: "proxy_server.py:8455", rationale: "Regional inference-profile ids (us.anthropic.*) over the invoke route, the deployment shape behind a customer timeout report on v1.90.0"}
- {id: llm.chat_completions.bedrock_invoke.basic.stream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: bedrock_invoke, capability: basic, streaming: stream, assertions: [works], source: "proxy_server.py:8455", rationale: "Streaming with regional inference-profile ids over the invoke route"}
- {id: llm.chat_completions.vertex.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: vertex, capability: basic, streaming: nonstream, assertions: [works], source: "proxy_server.py:8455", rationale: "P0 route; Vertex AI"}
- {id: llm.chat_completions.gemini.basic.nonstream.works, module: llm, tier: P0, subject_endpoint: chat_completions, route: gemini, capability: basic, streaming: nonstream, assertions: [works], source: "test_chat_completions_regression_e2e.py", rationale: "Gemini OpenAI-compatible chat translation"}
- {id: llm.chat_completions.gemini.basic.nonstream.cost_logged, module: llm, tier: P0, subject_endpoint: chat_completions, route: gemini, capability: basic, streaming: nonstream, assertions: [works, cost_logged], source: "test_chat_completions_regression_e2e.py", rationale: "Gemini chat cost lands in SpendLogs"}

1
tests/e2e/coverage_registry/mgmt.yaml
View file

@ -75,3 +75,4 @@
- {id: mgmt.workflow.list.happy_path, module: mgmt, tier: P2, surface: api, assertions: [happy_path], source: "workflow_management_endpoints.py", rationale: "Workflow tracking (smoke)"}
- {id: mgmt.credential_migration.check.happy_path, module: mgmt, tier: P2, surface: api, assertions: [happy_path], source: "key_management_endpoints.py:4252", rationale: "Encryption migration (smoke)"}
- {id: mgmt.credential.new.serves_request, module: mgmt, tier: P1, surface: api, assertions: [serves_request], source: "credential_endpoints/endpoints.py:42", rationale: "Stored credential resolves into a deployment and serves a live /messages request"}
- {id: mgmt.model.test_connection.happy_path, module: mgmt, tier: P0, surface: api, assertions: [happy_path], source: "_health_endpoints.py:1785", rationale: "Test Connection for a responses-mode Bedrock Mantle deployment reaches the live provider and reports success; this exact shape 500ed on an acompletion partial before v1.91.0", fail_before_fix: proven}

1
tests/e2e/coverage_registry/schema.py
View file

@ -71,6 +71,7 @@ LlmCapability = Literal[
"pdf_input",
"prompt_cache_1h",
"prompt_cache_5m",
"response_headers",
"service_tier",
"structured_output",
"thinking",

157
tests/e2e/llm_translation/test_bedrock_provider_matrix_e2e.py Normal file
View file

@ -0,0 +1,157 @@
"""Live e2e for the Bedrock cells of the provider-feature matrix: provider
response headers on /chat/completions and regional inference-profile model ids
(us.anthropic.*) over the invoke route.
Header forwarding is the #37003 contract: the proxy surfaces Bedrock's response
headers prefixed llm_provider- (llm_provider-x-amzn-requestid above all) so a
caller can hand AWS support the request id behind a completion. Regional
inference-profile ids are the deployment shape most Bedrock customers run; a
v1.90.0 regression timed them out, and the Converse route keeps them covered in
test_chat_completions_regression_e2e.py, so the invoke route carries its own
rows here.
"""
from __future__ import annotations
import pytest
from pydantic import BaseModel
from e2e_config import unique_marker
from e2e_http import StreamingResponse, unwrap
from lifecycle import ResourceManager
from models import ChatBody, ChatMessage, ChatResponse, LiteLLMParamsBody
from passthrough_client import PassthroughClient
pytestmark = pytest.mark.e2e
CONVERSE_REGIONAL_BACKEND = "bedrock/us.anthropic.claude-haiku-4-5-20251001-v1:0"
INVOKE_REGIONAL_BACKEND = "bedrock/invoke/us.anthropic.claude-haiku-4-5-20251001-v1:0"
PROVIDER_HEADER_PREFIX = "llm_provider-"
BEDROCK_REQUEST_ID_HEADER = "llm_provider-x-amzn-requestid"
class _StreamDelta(BaseModel):
content: str | None = None
class _StreamChoice(BaseModel):
delta: _StreamDelta = _StreamDelta()
class _StreamChunk(BaseModel):
choices: list[_StreamChoice] = []
def _streamed_text(events: list[str]) -> str:
chunks = [_StreamChunk.model_validate_json(event) for event in events]
return "".join(choice.delta.content or "" for chunk in chunks for choice in chunk.choices)
def _assert_streamed_completion(result: StreamingResponse) -> None:
assert result.ok and result.is_streaming, f"stream was not established: {result}"
assert result.stream_error is None, f"stream carried an error event: {result.stream_error}"
assert len(result.stream_events) > 1, f"stream did not deliver multiple data events: {result}"
assert _streamed_text(result.stream_events).strip(), (
f"stream completed with no content deltas: {result.stream_events[:3]}"
)
def _assert_request_id_header(result: StreamingResponse) -> None:
forwarded = [name for name in result.headers if name.startswith(PROVIDER_HEADER_PREFIX)]
assert result.headers.get(BEDROCK_REQUEST_ID_HEADER), (
f"missing {BEDROCK_REQUEST_ID_HEADER}; forwarded provider headers: {forwarded}"
)
def _assert_completion(response: ChatResponse) -> None:
assert response.choices, f"completion returned no choices: {response}"
message = response.choices[0].message
content = (message.content if message else None) or ""
assert content.strip(), f"completion carried no content: {response}"
def _register_bedrock_model(
client: PassthroughClient, resources: ResourceManager, prefix: str, backend: str
) -> str:
model = f"{prefix}-{unique_marker()}"
model_id = client.proxy.create_model(
model,
LiteLLMParamsBody(
model=backend,
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
aws_region_name="os.environ/AWS_REGION",
),
)
resources.defer(lambda: client.proxy.delete_model(model_id))
return model
def _prompt() -> list[ChatMessage]:
return [ChatMessage(role="user", content="reply with one word")]
class TestBedrockResponseHeaders:
@pytest.mark.covers(
"llm.chat_completions.bedrock_converse.response_headers.nonstream.works",
exercised_on=[],
)
def test_bedrock_request_id_header_surfaces(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
model = _register_bedrock_model(client, resources, "e2e-bedrock-headers", CONVERSE_REGIONAL_BACKEND)
key = resources.key()
result = client.proxy.transport.send(
"/chat/completions",
headers=client.proxy.transport.bearer(key),
json=ChatBody(model=model, messages=_prompt(), max_tokens=64),
)
assert result.ok, f"chat call failed: {result.status_code} {result.body[:300]}"
_assert_request_id_header(result)
@pytest.mark.covers(
"llm.chat_completions.bedrock_converse.response_headers.stream.works",
exercised_on=[],
)
def test_bedrock_request_id_header_surfaces_on_stream(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
model = _register_bedrock_model(
client, resources, "e2e-bedrock-headers-stream", CONVERSE_REGIONAL_BACKEND
)
key = resources.key()
result = client.proxy.chat_stream(
key, ChatBody(model=model, messages=_prompt(), stream=True, max_tokens=64)
)
_assert_streamed_completion(result)
_assert_request_id_header(result)
class TestBedrockInvokeRegionalModelIds:
@pytest.mark.covers("llm.chat_completions.bedrock_invoke.basic.nonstream.works", exercised_on=[])
def test_invoke_regional_id_completes(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
model = _register_bedrock_model(client, resources, "e2e-bedrock-invoke", INVOKE_REGIONAL_BACKEND)
key = resources.key()
response = unwrap(client.proxy.chat(key, ChatBody(model=model, messages=_prompt(), max_tokens=64)))
_assert_completion(response)
@pytest.mark.covers("llm.chat_completions.bedrock_invoke.basic.stream.works", exercised_on=[])
def test_invoke_regional_id_streams(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
model = _register_bedrock_model(client, resources, "e2e-bedrock-invoke-stream", INVOKE_REGIONAL_BACKEND)
key = resources.key()
result = client.proxy.chat_stream(
key, ChatBody(model=model, messages=_prompt(), stream=True, max_tokens=64)
)
_assert_streamed_completion(result)

52
tests/e2e/llm_translation/test_together_ai_e2e.py
View file

@ -5,14 +5,19 @@ in the proxy's own cost map that carries both capability flags. Two backends are
pinned because the registry has no flag for what they prove: ``enable_thinking`` is a
Qwen chat-template contract, and MiniMax-M3 is the serverless model whose template
renders a replayed ``reasoning_content`` back into the prompt (Qwen and DeepSeek
silently drop it). Requires TOGETHER_API_KEY on the proxy; no skip gate.
silently drop it). MiniMax-M3 honors that replayed field on nearly every call, not
every call (one miss in dozens of otherwise identical calls), so the replay case asks
up to ``REPLAY_ATTEMPTS`` times and fails only when no answer carries the secret, which
a proxy that strips the field guarantees. Requires TOGETHER_API_KEY on the proxy; no
skip gate.
"""
from __future__ import annotations
from collections.abc import Mapping
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from datetime import date
from typing import Final
import pytest
from e2e_config import unique_marker
@ -51,6 +56,7 @@ REASONING_REPLAY_BACKEND = "together_ai/MiniMaxAI/MiniMax-M3"
SECRET_PROMPT = "Remember this for later and reply with just OK."
SECRET_REASONING = "The user told me their favorite color is chartreuse. I must remember it."
SECRET_QUESTION = "What is my favorite color? Answer with one word."
REPLAY_ATTEMPTS: Final = 3
ARITHMETIC_PROMPT = "What is 17 + 26? Answer with just the number."
WEATHER_PROMPT = "What is the weather in Paris? Use the tool."
@ -179,6 +185,18 @@ def _message(response: ChatResponse) -> OutMessage:
return message
def _carries_secret(answer: OutMessage) -> bool:
return answer.content is not None and "chartreuse" in answer.content.lower()
def _answers_until_secret(client: PassthroughClient, key: str, body: ChatBody) -> Iterator[OutMessage]:
answers: Final = (_message(unwrap(client.proxy.chat(key, body))) for _ in range(REPLAY_ATTEMPTS))
for answer in answers:
yield answer
if _carries_secret(answer):
return
def _deltas(result: StreamingResponse) -> list[_StreamDelta]:
require_successful_call(result)
assert result.is_streaming, f"response was not streamed: {result.headers}"
@ -376,25 +394,19 @@ class TestTogetherChatCompletions:
self, client: PassthroughClient, resources: ResourceManager
) -> None:
model, key = _register(client, resources, REASONING_REPLAY_BACKEND)
answer = _message(
unwrap(
client.proxy.chat(
key,
ChatBody(
model=model,
messages=[
ChatMessage(role="user", content=SECRET_PROMPT),
ChatAssistantTurn(content="OK.", reasoning_content=SECRET_REASONING),
ChatMessage(role="user", content=SECRET_QUESTION),
],
max_tokens=512,
),
)
)
body: Final = ChatBody(
model=model,
messages=[
ChatMessage(role="user", content=SECRET_PROMPT),
ChatAssistantTurn(content="OK.", reasoning_content=SECRET_REASONING),
ChatMessage(role="user", content=SECRET_QUESTION),
],
max_tokens=512,
)
assert answer.content and "chartreuse" in answer.content.lower(), (
f"the replayed reasoning_content never reached Together: {answer}"
answers: Final = tuple(_answers_until_secret(client, key, body))
assert any(_carries_secret(answer) for answer in answers), (
f"the replayed reasoning_content never reached Together in {len(answers)} attempts: {answers}"
)
@pytest.mark.covers("llm.chat_completions.together_ai.basic.nonstream.cost_logged")

58
tests/e2e/logging/test_otel_trace_e2e.py
View file

@ -743,3 +743,61 @@ class TestOtelTraceCompleteness:
)
genai = next(span for span in hits[0].spans if span.operation_name == genai_span)
_assert_error_span_contract(genai)
@pytest.mark.covers("logging.otel.failure.exports_metric", exercised_on=["messages"])
def test_failed_messages_error_span_attributes(
self, client: LoggingClient, otel_reader: OtelReader, resources: ResourceManager
) -> None:
"""A failed `/v1/messages` request must carry the same error-span
contract as a failed `/chat/completions` request (LIT-6164). The
async messages entrypoint used to surface the provider handler's raw
BaseLLMException to the failure logger, so the model-call span came
out with error.type=BaseLLMException and no
litellm.provider.error.llm_provider attribute.
Same setup as the chat sibling: a deployment with an invalid upstream
API key passes proxy auth and fails at the provider with a real 401,
and failed requests are not billed, so no cost-write span."""
route = "/v1/messages"
_assert_otel_destination_configured(client)
model_name = f"otel-err-{unique_marker()}"
model_id = client.create_model(
model_name,
LiteLLMParamsBody(model="anthropic/claude-haiku-4-5", api_key=INVALID_UPSTREAM_API_KEY),
)
resources.defer(lambda: client.delete_model(model_id))
key = client.key_with_alias(f"otel-err-{unique_marker()}", models=[model_name])
resources.defer(lambda: client.delete_key(key))
deadline = time.monotonic() + client.proxy.poll_timeout
while True:
outcome = client.messages_raw(key, model_name, "trigger an upstream auth failure", max_tokens=16)
assert not outcome.ok, "the call must fail; the deployment's upstream key is invalid"
if "AnthropicException" in outcome.body or time.monotonic() >= deadline:
break
time.sleep(client.proxy.poll_interval)
assert "AnthropicException" in outcome.body, (
"never saw the mapped upstream provider failure before the deadline; either the key is "
"still propagating or the messages route surfaced the raw unmapped provider error - "
f"last outcome {outcome.status_code}: {outcome.body[:200]}"
)
assert outcome.status_code == 401, (
f"an upstream auth failure must map to 401, got {outcome.status_code}: {outcome.body[:200]}"
)
assert outcome.call_id is not None, "failed responses must still carry x-litellm-call-id"
genai_span = f"chat {model_name}"
hits = otel_reader.poll_traces_for_call(
call_id=outcome.call_id,
settled_names=_settled_names(route=route, genai_span=genai_span, require_cost_span=False),
settled_prefixes={DB_SPAN_PREFIX},
)
_assert_complete_trace(hits, route=route, genai_span=genai_span, require_cost_span=False)
root = next(span for span in hits[0].spans if not span.references)
assert str(_tag(root, "http.status_code")) == "401", (
f"the SERVER span must record the 401 the client received, got {_tag(root, 'http.status_code')!r}"
)
genai = next(span for span in hits[0].spans if span.operation_name == genai_span)
_assert_error_span_contract(genai)

13
tests/e2e/management/management_client.py
View file

@ -14,6 +14,8 @@ from e2e_http import NoBody, ProbeResult, Result, StreamingResponse, Success, Un
from models import (
ChatBody,
ChatMessage,
ConnectionTestBody,
ConnectionTestResponse,
CustomerDeleteBody,
CustomerInfoParams,
CustomerNewBody,
@ -118,6 +120,17 @@ class ManagementClient:
)
)
def connection_test(self, body: ConnectionTestBody) -> Result[ConnectionTestResponse]:
"""POST /health/test_connection, the call behind the Admin UI's Test
Connection button, probing the live provider with the supplied params."""
return self.proxy.transport.post(
"/health/test_connection",
headers=self.proxy.transport.master,
json=body,
response_type=ConnectionTestResponse,
timeout=120.0,
)
def block_key(self, key: str) -> None:
_ = unwrap(
self.proxy.transport.post(

67
tests/e2e/management/test_model_test_connection_e2e.py Normal file
View file

@ -0,0 +1,67 @@
"""Live e2e for POST /health/test_connection, the API behind the Admin UI's
Test Connection button on the add-model form.
The covered cell is a responses-mode Bedrock Mantle deployment: exactly this
shape 500ed on a functools.partial acompletion conflict before v1.91.0 while
every chat-mode probe stayed green, so the happy path asserts a real success
verdict from the live provider rather than just a 200 envelope. The region is a
literal because the endpoint rejects request-supplied os.environ/ references;
credentials fall through to the proxy's own environment (bearer token locally,
pod identity in CI).
The endpoint caps every probe at HEALTH_CHECK_TIMEOUT_SECONDS and answers a
timed-out probe with HTTP 200 and an in-body "Timeout exceeded", which the
harness's status-code retry policy cannot see. A Mantle probe can hit that cap
transiently while the rest of the suite saturates the same AWS account, so only
that exact error is retried here; any other error verdict fails immediately.
"""
from __future__ import annotations
import time
import pytest
from e2e_http import unwrap
from management_client import ManagementClient
from models import ConnectionTestBody, ConnectionTestResponse, LiteLLMParamsBody
pytestmark = pytest.mark.e2e
MANTLE_RESPONSES_BACKEND = "bedrock_mantle/openai.gpt-5.6-luna"
MANTLE_REGION = "us-east-1"
PROBE_TIMEOUT_ERROR = "Timeout exceeded"
PROBE_ATTEMPTS = 3
PROBE_RETRY_SLEEP_SECONDS = 30
def _probe_mantle(client: ManagementClient) -> ConnectionTestResponse:
return unwrap(
client.connection_test(
ConnectionTestBody(
litellm_params=LiteLLMParamsBody(
model=MANTLE_RESPONSES_BACKEND, aws_region_name=MANTLE_REGION
),
mode="responses",
)
)
)
class TestModelTestConnection:
@pytest.mark.covers("mgmt.model.test_connection.happy_path")
def test_bedrock_mantle_responses_connection_succeeds(self, client: ManagementClient) -> None:
for attempt in range(1, PROBE_ATTEMPTS + 1):
response = _probe_mantle(client)
if response.status == "success":
return
error = response.result.error if response.result else None
assert error == PROBE_TIMEOUT_ERROR, f"test_connection reported an error: {error}"
if attempt < PROBE_ATTEMPTS:
print(
f"test_connection probe timed out; retry {attempt}/{PROBE_ATTEMPTS - 1}"
f" in {PROBE_RETRY_SLEEP_SECONDS}s",
flush=True,
)
time.sleep(PROBE_RETRY_SLEEP_SECONDS)
pytest.fail(f"test_connection timed out on all {PROBE_ATTEMPTS} attempts")

20
tests/e2e/models.py
View file

@ -857,6 +857,26 @@ class ModelDeleteBody(BaseModel):
id: str
class ConnectionTestBody(BaseModel):
"""POST /health/test_connection body, the API behind the Admin UI's Test
Connection button: the deployment params as typed into the add-model form and
the health-check mode picking which endpoint the probe calls. The endpoint
rejects `os.environ/` references, so credentials are either literal values or
omitted to fall through to the proxy's own environment."""
litellm_params: LiteLLMParamsBody
mode: Literal["chat", "completion", "embedding", "responses"]
class ConnectionTestResult(BaseModel):
error: str | None = None
class ConnectionTestResponse(BaseModel):
status: Literal["success", "error"]
result: ConnectionTestResult | None = None
class CredentialCreateBody(BaseModel):
credential_name: str
credential_values: dict[str, str]

130
tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py
View file

@ -716,6 +716,136 @@ def test_generic_cost_per_token_tier_without_an_output_rate_bills_the_model_rate
litellm.model_cost.pop(model, None)
def test_generic_cost_per_token_tier_without_cache_rates_bills_cache_at_the_tier_input_rate():
model = "litellm-test-tiered-no-cache-rates"
custom_llm_provider = "openrouter"
litellm.register_model(
{
model: {
"litellm_provider": custom_llm_provider,
"mode": "chat",
"cache_read_input_token_cost": 9e-09,
"cache_creation_input_token_cost": 9e-06,
"tiered_pricing": [
{
"range": [0, 32000],
"input_cost_per_token": 4.6e-07,
"output_cost_per_token": 2.3e-06,
},
{
"range": [32000, 128000],
"input_cost_per_token": 7e-07,
"output_cost_per_token": 3.5e-06,
},
],
}
}
)
try:
uncached = Usage(prompt_tokens=40000, completion_tokens=100, total_tokens=40100)
cached = Usage(
prompt_tokens=40000,
completion_tokens=100,
total_tokens=40100,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=5000, cache_creation_tokens=15000
),
)
uncached_prompt_cost, _ = generic_cost_per_token(
model=model,
usage=uncached,
custom_llm_provider=custom_llm_provider,
)
cached_prompt_cost, cached_completion_cost = generic_cost_per_token(
model=model,
usage=cached,
custom_llm_provider=custom_llm_provider,
)
tier_input_rate = 7e-07
assert round(cached_prompt_cost, 12) == round(40000 * tier_input_rate, 12)
assert round(cached_prompt_cost, 12) == round(uncached_prompt_cost, 12)
assert round(cached_completion_cost, 12) == round(100 * 3.5e-06, 12)
finally:
litellm.model_cost.pop(model, None)
def test_generic_cost_per_token_tier_without_a_1hr_cache_rate_bills_the_tier_cache_creation_rate():
model = "litellm-test-tiered-no-1hr-cache-rate"
custom_llm_provider = "openrouter"
litellm.register_model(
{
model: {
"litellm_provider": custom_llm_provider,
"mode": "chat",
"cache_creation_input_token_cost_above_1hr": 9e-05,
"tiered_pricing": [
{
"range": [0, 128000],
"input_cost_per_token": 7e-07,
"output_cost_per_token": 3.5e-06,
"cache_creation_input_token_cost": 8.75e-07,
}
],
}
}
)
try:
usage = Usage(
prompt_tokens=1000,
completion_tokens=10,
total_tokens=1010,
prompt_tokens_details=PromptTokensDetailsWrapper(
cache_creation_tokens=800,
cache_creation_token_details=CacheCreationTokenDetails(
ephemeral_5m_input_tokens=300, ephemeral_1h_input_tokens=500
),
),
)
prompt_cost, completion_cost = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
tier_cache_creation_rate = 8.75e-07
expected_prompt = (200 * 7e-07) + (800 * tier_cache_creation_rate)
assert round(prompt_cost, 12) == round(expected_prompt, 12)
assert round(completion_cost, 12) == round(10 * 3.5e-06, 12)
finally:
litellm.model_cost.pop(model, None)
def test_generic_cost_per_token_tier_without_an_input_rate_is_not_a_priced_tier():
model = "litellm-test-tiered-no-input-rate"
custom_llm_provider = "openrouter"
litellm.register_model(
{
model: {
"litellm_provider": custom_llm_provider,
"mode": "chat",
"input_cost_per_token": 1e-06,
"output_cost_per_token": 2e-06,
"tiered_pricing": [{"range": [0, 128000], "output_cost_per_token": 3.5e-06}],
}
}
)
try:
usage = Usage(prompt_tokens=1000, completion_tokens=100, total_tokens=1100)
prompt_cost, completion_cost = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
assert round(prompt_cost, 12) == round(1000 * 1e-06, 12)
assert round(completion_cost, 12) == round(100 * 2e-06, 12)
finally:
litellm.model_cost.pop(model, None)
def test_router_deployment_with_input_only_tiers_bills_completions_at_the_backend_rate():
"""Regression: the router registers a deployment's custom pricing as a standalone
model_cost entry holding only the supplied fields, so an input-only tier table left

124
tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py
View file

@ -13,6 +13,7 @@ from litellm.litellm_core_utils.exception_mapping_utils import (
extract_and_raise_litellm_exception,
)
from litellm.llms.openai.common_utils import OpenAIError
from litellm.types.utils import LlmProviders
# Test cases for is_error_str_context_window_exceeded
# Tuple format: (error_message, expected_result)
@ -785,33 +786,24 @@ OPENAI_SHAPED = {
503: (litellm.ServiceUnavailableError, 503),
}
UPSTREAM_STATUS_DISCARDED = (litellm.APIConnectionError, 500)
PERMISSION_DENIED = (litellm.PermissionDeniedError, 403)
PROVIDERS_THAT_DISCARD_THE_UPSTREAM_STATUS = ("cloudflare", "ollama", "vllm")
STATUS_KEYED = {**OPENAI_SHAPED, 403: PERMISSION_DENIED}
DEVIATIONS_FROM_THE_OPENAI_SHAPE = {
"anthropic": {403: UPSTREAM_STATUS_DISCARDED, 422: UPSTREAM_STATUS_DISCARDED},
"anthropic": {403: PERMISSION_DENIED},
"azure": {500: (litellm.APIError, 500)},
"bedrock": {
403: UPSTREAM_STATUS_DISCARDED,
403: PERMISSION_DENIED,
500: (litellm.ServiceUnavailableError, 503),
},
"cohere": {
401: UPSTREAM_STATUS_DISCARDED,
403: UPSTREAM_STATUS_DISCARDED,
404: UPSTREAM_STATUS_DISCARDED,
422: UPSTREAM_STATUS_DISCARDED,
429: UPSTREAM_STATUS_DISCARDED,
503: UPSTREAM_STATUS_DISCARDED,
},
"cloudflare": {403: PERMISSION_DENIED},
"cohere": {403: PERMISSION_DENIED},
"databricks": {
403: (litellm.AuthenticationError, 401),
403: PERMISSION_DENIED,
422: (litellm.BadRequestError, 400),
},
"gemini": {
403: (litellm.PermissionDeniedError, 403),
422: UPSTREAM_STATUS_DISCARDED,
},
"gemini": {403: PERMISSION_DENIED},
"huggingface": {
404: (litellm.APIError, 404),
422: (litellm.APIError, 422),
@ -824,6 +816,7 @@ DEVIATIONS_FROM_THE_OPENAI_SHAPE = {
500: (litellm.APIError, 500),
503: (litellm.APIError, 503),
},
"ollama": {403: PERMISSION_DENIED},
"openrouter": {500: (litellm.APIError, 500)},
"replicate": {
403: (litellm.APIError, 500),
@ -833,17 +826,11 @@ DEVIATIONS_FROM_THE_OPENAI_SHAPE = {
503: (litellm.APIError, 500),
},
"sagemaker": {
403: UPSTREAM_STATUS_DISCARDED,
403: PERMISSION_DENIED,
500: (litellm.ServiceUnavailableError, 503),
},
"vertex_ai": {
403: (litellm.PermissionDeniedError, 403),
422: UPSTREAM_STATUS_DISCARDED,
},
**{
provider: dict.fromkeys(UPSTREAM_STATUS_CODES, UPSTREAM_STATUS_DISCARDED)
for provider in PROVIDERS_THAT_DISCARD_THE_UPSTREAM_STATUS
},
"vertex_ai": {403: PERMISSION_DENIED},
"vllm": {403: PERMISSION_DENIED},
}
PROVIDERS_WITH_A_HANDLER = (
@ -875,6 +862,38 @@ PROVIDERS_WITH_A_HANDLER = (
"xai",
)
PROVIDER_ALIASES_WITH_A_HANDLER = (
"aleph_alpha",
"anthropic_text",
"azure_text",
"bedrock_mantle",
"cohere_chat",
"custom_openai",
"lemonade",
"litellm_proxy",
"ollama_chat",
"predibase",
"sagemaker_chat",
"text-completion-openai",
"vertex_ai_beta",
"watsonx",
)
PROVIDERS_WITHOUT_A_HANDLER = tuple(
sorted(
frozenset(provider.value for provider in LlmProviders)
- frozenset(PROVIDERS_WITH_A_HANDLER)
- frozenset(PROVIDER_ALIASES_WITH_A_HANDLER)
- frozenset(litellm.openai_compatible_providers)
)
)
MINIMAX_401_BODY = (
'{"type":"error","error":{"type":"authorized_error","message":"login fail: Please carry the API secret key '
"in the 'Authorization' field of the request header (1004)\",\"http_code\":\"401\"},"
'"request_id":"06ddc9ba97ee6340e38f10e09787f547"}'
)
def _expected_for(provider: str, status_code: int) -> tuple[type[Exception], int]:
return DEVIATIONS_FROM_THE_OPENAI_SHAPE.get(provider, {}).get(
@ -938,6 +957,51 @@ def test_an_already_mapped_litellm_exception_passes_through_untouched(
assert returned is already_mapped
@pytest.mark.parametrize("status_code", UPSTREAM_STATUS_CODES)
@pytest.mark.parametrize("provider", PROVIDERS_WITHOUT_A_HANDLER)
def test_a_provider_without_a_handler_maps_by_the_upstream_status(
provider, status_code, quiet_exception_mapping
):
expected_class, expected_status = STATUS_KEYED[status_code]
with pytest.raises(openai.APIError) as raised:
exception_type(
model="test-model",
original_exception=_UpstreamHTTPError(status_code=status_code),
custom_llm_provider=provider,
)
assert type(raised.value) is expected_class
assert raised.value.status_code == expected_status
assert raised.value.llm_provider == provider
assert raised.value.model == "test-model"
def test_a_minimax_bad_key_is_an_authentication_error(quiet_exception_mapping):
from litellm.llms.base_llm.chat.transformation import BaseLLMException
with pytest.raises(litellm.AuthenticationError) as raised:
exception_type(
model="MiniMax-M2.5",
original_exception=BaseLLMException(status_code=401, message=MINIMAX_401_BODY),
custom_llm_provider="minimax",
)
assert raised.value.status_code == 401
assert raised.value.llm_provider == "minimax"
assert raised.value.message.startswith("litellm.AuthenticationError: MinimaxException - ")
assert "login fail" in raised.value.message
def test_an_exception_without_a_status_is_still_a_connection_error(quiet_exception_mapping):
with pytest.raises(litellm.APIConnectionError):
exception_type(
model="MiniMax-M2.5",
original_exception=RuntimeError("socket hung up"),
custom_llm_provider="minimax",
)
CONTEXT_WINDOW_MESSAGE = "This model's maximum context length is 4096 tokens."
CONTENT_POLICY_MESSAGE = (
'{"error": {"type": "invalid_request_error", "code": "content_policy_violation"}}'
@ -993,9 +1057,7 @@ class _UpstreamErrorWithMessage(_UpstreamHTTPError):
def test_a_full_context_window_reaches_the_caller_as_the_router_needs_it(
provider, quiet_exception_mapping
):
if provider in PROVIDERS_THAT_DISCARD_THE_UPSTREAM_STATUS:
expected_class, expected_status = UPSTREAM_STATUS_DISCARDED
elif provider in PROVIDERS_THAT_RECOGNISE_A_FULL_CONTEXT_WINDOW:
if provider in PROVIDERS_THAT_RECOGNISE_A_FULL_CONTEXT_WINDOW:
expected_class, expected_status = litellm.ContextWindowExceededError, 400
else:
expected_class, expected_status = litellm.BadRequestError, 400
@ -1015,9 +1077,7 @@ def test_a_full_context_window_reaches_the_caller_as_the_router_needs_it(
def test_a_content_policy_block_reaches_the_caller_as_the_router_needs_it(
provider, quiet_exception_mapping
):
if provider in PROVIDERS_THAT_DISCARD_THE_UPSTREAM_STATUS:
expected_class, expected_status = UPSTREAM_STATUS_DISCARDED
elif provider in PROVIDERS_THAT_RECOGNISE_A_CONTENT_POLICY_BLOCK:
if provider in PROVIDERS_THAT_RECOGNISE_A_CONTENT_POLICY_BLOCK:
expected_class, expected_status = litellm.ContentPolicyViolationError, 400
else:
expected_class, expected_status = litellm.BadRequestError, 400

2
tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py
View file

@ -679,7 +679,7 @@ def test_translate_anthropic_to_openai_orders_top_level_and_midturn_system():
def _translate_with_metadata(
model: str, metadata: dict[str, Any], custom_llm_provider: str | None
model: str, metadata: dict[str, str], custom_llm_provider: str | None
) -> dict[str, Any]:
openai_request, _ = LiteLLMAnthropicMessagesAdapter().translate_anthropic_to_openai(
anthropic_message_request={

94
tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_anthropic_experimental_pass_through_messages_handler.py
View file

@ -7,6 +7,7 @@ from typing import Any, Dict, List
import httpx
import pytest
from fastapi.testclient import TestClient
from pydantic import ValidationError
from unittest.mock import AsyncMock, MagicMock, patch
@ -1286,3 +1287,96 @@ class TestMessagesStreamingSuccessLogging:
assert payload["call_type"] == "acompletion"
assert payload["total_tokens"] > 0
assert payload["response_cost"] > 0
class _FailureCapture(CustomLogger):
def __init__(self):
super().__init__()
self.error_information: List[Dict[str, Any]] = []
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
payload = kwargs.get("standard_logging_object") or {}
self.error_information.append(payload.get("error_information") or {})
@pytest.mark.asyncio
@pytest.mark.parametrize(
"upstream_status, upstream_error_type, expected_exception",
[
(401, "authentication_error", litellm.AuthenticationError),
(403, "permission_error", litellm.PermissionDeniedError),
],
)
async def test_anthropic_messages_maps_provider_exception_before_failure_logging(
monkeypatch, upstream_status, upstream_error_type, expected_exception
):
"""Regression test for LIT-6164. The async /v1/messages entrypoint awaited the
provider handler without exception_type mapping, so the @client failure
handler (and every logger behind it, e.g. OTel error spans) saw the raw
BaseLLMException: error.type=BaseLLMException and no llm_provider.
The 403 row pins the upstream status on the way through the mapper: Anthropic's
documented permission_error must reach the caller as a 403, never as the mapper's
APIConnectionError 500 fallthrough."""
from litellm.llms.anthropic.experimental_pass_through.messages import handler
capture = _FailureCapture()
monkeypatch.setattr(litellm, "callbacks", [capture])
def upstream_rejects_the_request(request: httpx.Request) -> httpx.Response:
return httpx.Response(
upstream_status,
json={"type": "error", "error": {"type": upstream_error_type, "message": "rejected upstream"}},
request=request,
)
upstream = AsyncHTTPHandler()
upstream.client = httpx.AsyncClient(transport=httpx.MockTransport(upstream_rejects_the_request))
with pytest.raises(expected_exception) as excinfo:
await handler.anthropic_messages(
max_tokens=16,
messages=[{"role": "user", "content": "hi"}],
model="anthropic/claude-haiku-4-5",
custom_llm_provider="anthropic",
api_key="sk-invalid",
client=upstream,
)
assert excinfo.value.status_code == upstream_status
assert excinfo.value.llm_provider == "anthropic"
assert "AnthropicException" in excinfo.value.message
assert f'"{upstream_error_type}"' in excinfo.value.message
assert capture.error_information, "the failure handler must have logged the mapped exception"
error_information = capture.error_information[0]
assert error_information.get("error_class") == expected_exception.__name__
assert error_information.get("llm_provider") == "anthropic"
assert error_information.get("error_code") == str(upstream_status)
@pytest.mark.asyncio
async def test_anthropic_messages_leaves_non_provider_failures_unmapped():
"""The mapping boundary is for provider failures only. A request rejected before
the provider call (here invalid metadata) must surface as the original exception,
not as the mapper's APIConnectionError, whose message embeds a server traceback."""
from litellm.llms.anthropic.experimental_pass_through.messages import handler
def upstream_must_not_be_called(request: httpx.Request) -> httpx.Response:
raise AssertionError("the provider must not be called for a request rejected locally")
upstream = AsyncHTTPHandler()
upstream.client = httpx.AsyncClient(transport=httpx.MockTransport(upstream_must_not_be_called))
with pytest.raises(ValidationError) as excinfo:
await handler.anthropic_messages(
max_tokens=16,
messages=[{"role": "user", "content": "hi"}],
model="anthropic/claude-haiku-4-5",
custom_llm_provider="anthropic",
api_key="sk-invalid",
client=upstream,
metadata={"user_id": 123},
)
assert "Traceback" not in str(excinfo.value)

64
tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py
View file

@ -135,16 +135,24 @@ class TestOutputConfigStructuredOutput:
}
def test_output_config_format_json_schema_converted(self):
"""output_config.format.json_schema is converted to OpenAI text.format."""
"""output_config.format.json_schema is converted to OpenAI text.format, defaulting strict to False."""
req = _make_request(output_config={"format": {"type": "json_schema", "schema": self._SCHEMA}})
kwargs = _ADAPTER.translate_request(req)
assert "text" in kwargs
fmt = kwargs["text"]["format"]
assert fmt["type"] == "json_schema"
assert fmt["schema"] == self._SCHEMA
assert fmt["strict"] is True
assert fmt["strict"] is False
assert fmt["name"] == "structured_output"
def test_output_config_format_explicit_strict_true_is_preserved(self):
"""Nested output_config.format with explicit strict=True is preserved."""
req = _make_request(
output_config={"format": {"type": "json_schema", "schema": self._SCHEMA, "strict": True}}
)
kwargs = _ADAPTER.translate_request(req)
assert kwargs["text"]["format"]["strict"] is True
def test_output_config_without_format_does_not_set_text(self):
"""output_config with only non-format keys doesn't produce text.format."""
req = _make_request(output_config={"effort": "high"})
@ -152,21 +160,65 @@ class TestOutputConfigStructuredOutput:
assert "text" not in kwargs
def test_output_format_still_works(self):
"""The original output_format field still takes precedence when present."""
"""The original output_format field still takes precedence when present, defaulting strict to False."""
req = _make_request(output_format={"type": "json_schema", "schema": self._SCHEMA})
kwargs = _ADAPTER.translate_request(req)
assert "text" in kwargs
assert kwargs["text"]["format"]["type"] == "json_schema"
assert kwargs["text"]["format"]["strict"] is False
def test_output_format_explicit_strict_false_is_preserved(self):
"""output_format with an explicit strict=False is preserved as False."""
req = _make_request(output_format={"type": "json_schema", "schema": self._SCHEMA, "strict": False})
kwargs = _ADAPTER.translate_request(req)
assert kwargs["text"]["format"]["strict"] is False
def test_output_format_explicit_strict_true_is_preserved(self):
"""output_format with an explicit strict=True is preserved as True."""
req = _make_request(output_format={"type": "json_schema", "schema": self._SCHEMA, "strict": True})
kwargs = _ADAPTER.translate_request(req)
assert kwargs["text"]["format"]["strict"] is True
def test_output_format_takes_precedence_over_output_config(self):
"""output_format takes precedence over output_config.format."""
"""output_format takes precedence over output_config.format, for both schema and strict."""
other_schema = {"type": "object", "properties": {"id": {"type": "integer"}}}
req = _make_request(
output_format={"type": "json_schema", "schema": self._SCHEMA},
output_config={"format": {"type": "json_schema", "schema": other_schema}},
output_format={"type": "json_schema", "schema": self._SCHEMA, "strict": False},
output_config={"format": {"type": "json_schema", "schema": other_schema, "strict": True}},
)
kwargs = _ADAPTER.translate_request(req)
assert kwargs["text"]["format"]["schema"] == self._SCHEMA
assert kwargs["text"]["format"]["strict"] is False
def test_optional_property_stays_out_of_required_list(self):
"""A property absent from required must stay absent from required in the translated schema."""
schema = {
"type": "object",
"properties": {
"name": {"type": "string"},
"nickname": {"type": "string"},
},
"required": ["name"],
"additionalProperties": False,
}
req = _make_request(output_format={"type": "json_schema", "schema": schema})
kwargs = _ADAPTER.translate_request(req)
fmt_schema = kwargs["text"]["format"]["schema"]
assert fmt_schema["required"] == ["name"]
assert "nickname" not in fmt_schema["required"]
assert fmt_schema["additionalProperties"] is False
def test_translate_request_does_not_mutate_input_schema(self):
"""translate_request must not mutate the caller's output_format or schema dicts."""
schema = {"type": "object", "properties": {"x": {"type": "number"}}, "required": ["x"]}
output_format = {"type": "json_schema", "schema": schema, "strict": False}
req = _make_request(output_format=output_format)
snapshot = json.loads(json.dumps(output_format))
_ADAPTER.translate_request(req)
assert output_format == snapshot
assert req["output_format"] == snapshot
# ---------------------------------------------------------------------------

2
tests/test_litellm/llms/base_llm/search/test_base_search_transformation.py
View file

@ -322,7 +322,7 @@ async def test_query_param_key_not_leaked_with_dummy_caller_key(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.get",
fake_get,
):
with pytest.raises(litellm.APIConnectionError):
with pytest.raises(litellm.InternalServerError):
await litellm.asearch(
query="secrets",
search_provider=provider,

2
tests/test_litellm/llms/compactifai/test_compactifai.py
View file

@ -172,7 +172,7 @@ def test_compactifai_authentication_error(respx_mock):
json=mock_error, status_code=401
)
with pytest.raises(litellm.APIConnectionError) as exc_info:
with pytest.raises(litellm.AuthenticationError) as exc_info:
litellm.completion(
model="compactifai/cai-llama-3-1-8b-slim",
messages=[{"role": "user", "content": "test"}],

2
tests/test_litellm/llms/langflow/chat/test_langflow_chat_transformation.py
View file

@ -233,7 +233,7 @@ def test_langflow_extra_body_cannot_inject_tweaks_into_run_payload():
return resp
with patch.object(HTTPHandler, "post", side_effect=fake_post):
with pytest.raises(litellm.APIConnectionError):
with pytest.raises(litellm.BadRequestError):
litellm.completion(
model="langflow/my-flow",
messages=[{"role": "user", "content": "hello"}],

199
tests/test_litellm/llms/together_ai/chat/test_together_ai_chat_transformation.py
View file

@ -20,11 +20,24 @@ TOOL_CALLING_MODEL = "openai/gpt-oss-20b"
REASONING_MODEL = "deepseek-ai/DeepSeek-V3.1"
UNMAPPED_MODEL = "example-org/brand-new-model"
NO_TOOLS_MODEL = "example-org/no-tools-model"
NO_SCHEMA_MODEL = "example-org/no-schema-model"
TOOL_PARAMS = ("tools", "tool_choice", "function_call")
WEATHER_TOOLS = [{"type": "function", "function": {"name": "get_weather", "parameters": {}}}]
VOICE_NOTE_SCHEMA = {
"type": "object",
"properties": {"title": {"type": "string"}, "summary": {"type": "string"}},
"required": ["title", "summary"],
"additionalProperties": False,
}
JSON_SCHEMA_RESPONSE_FORMAT = {
"type": "json_schema",
"json_schema": {"name": "voice_note", "schema": VOICE_NOTE_SCHEMA, "strict": True},
}
REGEX_RESPONSE_FORMAT = {"type": "regex", "pattern": "(positive|neutral|negative)"}
@pytest.fixture(autouse=True)
def force_local_model_cost(monkeypatch):
@ -48,6 +61,15 @@ def registry_disables_function_calling(monkeypatch):
)
@pytest.fixture
def registry_disables_response_schema(monkeypatch):
monkeypatch.setitem(
litellm.model_cost,
f"together_ai/{NO_SCHEMA_MODEL}",
{"litellm_provider": "together_ai", "mode": "chat", "supports_response_schema": False},
)
@pytest.fixture
def together_warning_log(caplog):
from litellm._logging import verbose_logger
@ -70,7 +92,7 @@ def test_supported_params_unmapped_model_keeps_tool_params():
for param in TOOL_PARAMS:
assert param in supported
assert "response_format" not in supported
assert "response_format" in supported
assert "stream" in supported
assert "temperature" in supported
@ -80,7 +102,7 @@ def test_supported_params_no_tools_model_keeps_tool_params(registry_disables_fun
for param in TOOL_PARAMS:
assert param in supported
assert "response_format" not in supported
assert "response_format" in supported
def test_map_openai_params_tool_calling_model_passes_tools():
@ -148,21 +170,17 @@ def test_map_openai_params_reasoning_model_passes_sampling_params():
assert mapped["max_tokens"] == 512
def test_map_openai_params_drops_text_response_format():
mapped = TogetherAIChatConfig().map_openai_params(
non_default_params={"response_format": {"type": "text"}, "temperature": 0.5},
optional_params={},
model=REASONING_MODEL,
drop_params=False,
)
assert "response_format" not in mapped
assert mapped["temperature"] == 0.5
def test_map_openai_params_keeps_json_response_format():
response_format = {"type": "json_object"}
@pytest.mark.parametrize(
"response_format",
[
{"type": "text"},
{"type": "json_object"},
{"type": "json_object", "schema": VOICE_NOTE_SCHEMA},
JSON_SCHEMA_RESPONSE_FORMAT,
REGEX_RESPONSE_FORMAT,
],
)
def test_map_openai_params_schema_model_passes_response_format_through(response_format):
mapped = TogetherAIChatConfig().map_openai_params(
non_default_params={"response_format": response_format},
optional_params={},
@ -173,6 +191,46 @@ def test_map_openai_params_keeps_json_response_format():
assert mapped["response_format"] == response_format
@pytest.mark.parametrize("drop_params", [False, True])
def test_map_openai_params_unmapped_model_passes_response_format_through(drop_params, together_warning_log):
mapped = TogetherAIChatConfig().map_openai_params(
non_default_params={"response_format": JSON_SCHEMA_RESPONSE_FORMAT},
optional_params={},
model=UNMAPPED_MODEL,
drop_params=drop_params,
)
assert mapped["response_format"] == JSON_SCHEMA_RESPONSE_FORMAT
assert UNMAPPED_MODEL in together_warning_log.text
assert "passing response_format through" in together_warning_log.text
def test_map_openai_params_no_schema_model_drops_response_format_with_warning(
registry_disables_response_schema, together_warning_log
):
mapped = TogetherAIChatConfig().map_openai_params(
non_default_params={"response_format": JSON_SCHEMA_RESPONSE_FORMAT, "temperature": 0.5},
optional_params={},
model=NO_SCHEMA_MODEL,
drop_params=True,
)
assert "response_format" not in mapped
assert mapped["temperature"] == 0.5
assert NO_SCHEMA_MODEL in together_warning_log.text
assert "dropping response_format" in together_warning_log.text
def test_map_openai_params_no_schema_model_raises_without_drop_params(registry_disables_response_schema):
with pytest.raises(UnsupportedParamsError, match="response_format"):
TogetherAIChatConfig().map_openai_params(
non_default_params={"response_format": JSON_SCHEMA_RESPONSE_FORMAT},
optional_params={},
model=NO_SCHEMA_MODEL,
drop_params=False,
)
def _transform_response(message: dict) -> ModelResponse:
raw_response_json = {
"id": "chatcmpl-test",
@ -206,26 +264,20 @@ def _transform_response(message: dict) -> ModelResponse:
def test_transform_response_maps_reasoning_to_reasoning_content():
result = _transform_response(
{"role": "assistant", "content": "4", "reasoning": "2+2 equals 4"}
)
result = _transform_response({"role": "assistant", "content": "4", "reasoning": "2+2 equals 4"})
assert result.choices[0].message.content == "4"
assert result.choices[0].message.reasoning_content == "2+2 equals 4"
def test_transform_response_preserves_reasoning_content_field():
result = _transform_response(
{"role": "assistant", "content": "4", "reasoning_content": "adding 2 and 2"}
)
result = _transform_response({"role": "assistant", "content": "4", "reasoning_content": "adding 2 and 2"})
assert result.choices[0].message.reasoning_content == "adding 2 and 2"
def test_streaming_chunk_maps_delta_reasoning_to_reasoning_content():
iterator = TogetherAIChatConfig().get_model_response_iterator(
streaming_response=iter(()), sync_stream=True
)
iterator = TogetherAIChatConfig().get_model_response_iterator(streaming_response=iter(()), sync_stream=True)
assert isinstance(iterator, OpenAIChatCompletionStreamingHandler)
parsed = iterator.chunk_parser(
@ -241,9 +293,7 @@ def test_streaming_chunk_maps_delta_reasoning_to_reasoning_content():
def test_streaming_chunk_preserves_tool_call_index_and_id():
iterator = TogetherAIChatConfig().get_model_response_iterator(
streaming_response=iter(()), sync_stream=True
)
iterator = TogetherAIChatConfig().get_model_response_iterator(streaming_response=iter(()), sync_stream=True)
def parse_tool_call_chunk(tool_call: dict):
parsed = iterator.chunk_parser(
@ -374,9 +424,7 @@ def test_together_ai_config_alias_points_at_chat_config():
def test_provider_config_manager_returns_together_chat_config():
from litellm.utils import ProviderConfigManager
config = ProviderConfigManager.get_provider_chat_config(
model=REASONING_MODEL, provider=LlmProviders.TOGETHER_AI
)
config = ProviderConfigManager.get_provider_chat_config(model=REASONING_MODEL, provider=LlmProviders.TOGETHER_AI)
assert isinstance(config, TogetherAIChatConfig)
@ -484,6 +532,72 @@ def test_completion_unmapped_model_sends_tools_to_together():
assert json.loads(tool_call.function.arguments) == {"city": "San Francisco"}
def _capture_completion_request(model: str, **completion_kwargs) -> dict:
from litellm.llms.custom_httpx.http_handler import HTTPHandler
captured_requests = []
def respond(request: httpx.Request) -> httpx.Response:
captured_requests.append(request)
return httpx.Response(
200,
json={
"id": "chatcmpl-together-structured",
"object": "chat.completion",
"created": 1234567890,
"model": model,
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": '{"title": "t", "summary": "s"}'},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
},
)
client = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond)))
litellm.completion(
model=f"together_ai/{model}",
messages=[{"role": "user", "content": "Summarize with a title and summary."}],
api_key="fake-key",
client=client,
**completion_kwargs,
)
return json.loads(captured_requests[0].content)
def test_completion_unmapped_model_sends_json_schema_to_together():
request_body = _capture_completion_request(
UNMAPPED_MODEL, response_format=JSON_SCHEMA_RESPONSE_FORMAT, drop_params=True
)
assert request_body["response_format"] == JSON_SCHEMA_RESPONSE_FORMAT
def test_completion_pydantic_response_format_sends_json_schema_to_together():
from pydantic import BaseModel
class VoiceNote(BaseModel):
title: str
summary: str
request_body = _capture_completion_request(TOOL_CALLING_MODEL, response_format=VoiceNote)
sent = request_body["response_format"]
assert sent["type"] == "json_schema"
assert sent["json_schema"]["name"] == "VoiceNote"
assert sent["json_schema"]["strict"] is True
assert sent["json_schema"]["schema"]["required"] == ["title", "summary"]
def test_completion_regex_response_format_sends_pattern_to_together():
request_body = _capture_completion_request(TOOL_CALLING_MODEL, response_format=REGEX_RESPONSE_FORMAT)
assert request_body["response_format"] == REGEX_RESPONSE_FORMAT
TOGETHER_CHAT_URL = "https://api.together.ai/v1/chat/completions"
WEATHER_AND_TIME_TOOLS = [
@ -552,14 +666,23 @@ PARALLEL_TOOL_CALL_STREAM = (
_chunk(
{
"tool_calls": [
{"index": 0, "id": "call_weather", "type": "function", "function": {"name": "get_weather", "arguments": ""}}
{
"index": 0,
"id": "call_weather",
"type": "function",
"function": {"name": "get_weather", "arguments": ""},
}
]
}
),
_chunk({"tool_calls": [{"index": 0, "function": {"arguments": '{"city": "San'}}]}),
_chunk({"tool_calls": [{"index": 0, "function": {"arguments": ' Francisco"}'}}]}),
_chunk(
{"tool_calls": [{"index": 1, "id": "call_time", "type": "function", "function": {"name": "get_time", "arguments": ""}}]}
{
"tool_calls": [
{"index": 1, "id": "call_time", "type": "function", "function": {"name": "get_time", "arguments": ""}}
]
}
),
_chunk({"tool_calls": [{"index": 1, "function": {"arguments": '{"tz": "PST"}'}}]}, finish_reason="tool_calls"),
)
@ -802,10 +925,14 @@ def test_anthropic_messages_streams_together_tool_call_as_input_json_delta():
if event["type"] == "content_block_start" and event["content_block"]["type"] == "tool_use"
}
input_json_deltas = [
event for event in events if event["type"] == "content_block_delta" and event["delta"]["type"] == "input_json_delta"
event
for event in events
if event["type"] == "content_block_delta" and event["delta"]["type"] == "input_json_delta"
]
tool_inputs = {
block["name"]: json.loads("".join(delta["delta"]["partial_json"] for delta in input_json_deltas if delta["index"] == index))
block["name"]: json.loads(
"".join(delta["delta"]["partial_json"] for delta in input_json_deltas if delta["index"] == index)
)
for index, block in tool_starts.items()
}
assert {block["id"] for block in tool_starts.values()} == {"call_weather", "call_time"}

6
tests/test_litellm/llms/vertex_ai/vertex_gemma_models/test_vertex_gemma_transformation.py
View file

@ -238,7 +238,7 @@ class TestVertexGemmaCompletion:
Expected: Proper error handling when 'predictions' field is missing
"""
from litellm.exceptions import APIConnectionError
from litellm.exceptions import BadRequestError
# Invalid response without predictions field
invalid_response = {
@ -260,8 +260,8 @@ class TestVertexGemmaCompletion:
mock_client.post = AsyncMock(return_value=mock_response)
mock_get_client.return_value = mock_client
# Should raise exception (wrapped as APIConnectionError by LiteLLM)
with pytest.raises(APIConnectionError) as exc_info:
# Should raise exception (wrapped as BadRequestError by LiteLLM)
with pytest.raises(BadRequestError) as exc_info:
await litellm.acompletion(
model="vertex_ai/gemma/gemma-3-12b-it",
messages=[{"role": "user", "content": "Test"}],

4
tests/test_litellm/proxy/common_utils/test_reset_budget_job.py
View file

@ -1948,14 +1948,14 @@ class FakePodLockManager:
if self.redis_cache is not None:
self.redis_cache.async_get_cache = AsyncMock(return_value="another-pod" if held_by_other else None)
self._acquired = acquired
self.acquire_calls: List[Dict[str, Any]] = []
self.acquire_calls: List[Dict[str, str | int | None]] = []
self.release_calls: List[str] = []
@staticmethod
def get_redis_lock_key(cronjob_id: str) -> str:
return f"cronjob_lock:{cronjob_id}"
async def acquire_lock(self, cronjob_id: str, ttl: Any = None) -> bool:
async def acquire_lock(self, cronjob_id: str, ttl: int | None = None) -> bool:
self.acquire_calls.append({"cronjob_id": cronjob_id, "ttl": ttl})
return self._acquired

14
tests/test_litellm/proxy/common_utils/test_sse_keepalive.py
View file

@ -10,6 +10,7 @@ from litellm.proxy.common_utils.sse_keepalive import (
ANTHROPIC_PING_SSE_CHUNK,
SSE_COMMENT_PING_BYTES,
resolve_ttft_keepalive_interval,
split_complete_sse_frames,
wrap_passthrough_sse_bytes_with_keepalive_pings,
wrap_sse_stream_with_keepalive_pings,
)
@ -18,6 +19,19 @@ MESSAGE_START_CHUNK: Final = 'data: {"type": "message_start"}\n\n'
TEXT_DELTA_CHUNK: Final = 'data: {"type": "content_block_delta"}\n\n'
@pytest.mark.parametrize("delimiter", [b"\n\n", b"\r\n\r\n", b"\r\r"])
def test_split_complete_sse_frames_recognizes_every_sse_frame_delimiter(delimiter: bytes):
newline: Final = delimiter[: len(delimiter) // 2]
frame: Final = b"event: response.created" + newline + b"data: {}" + delimiter
tail: Final = b"data: partial"
assert split_complete_sse_frames(frame + tail) == (frame, tail)
def test_split_complete_sse_frames_holds_bytes_with_no_complete_frame():
assert split_complete_sse_frames(b"data: unterminated") == (b"", b"data: unterminated")
@pytest.mark.asyncio
async def test_pings_fill_mid_stream_silence_and_preserve_chunk_order():
async def gappy_stream() -> AsyncGenerator[str, None]:

78
tests/test_litellm/proxy/db/test_db_url_settings.py
View file

@ -34,6 +34,7 @@ def _apply() -> bool:
_MANAGED_DB_ENV_VARS = (
"IAM_TOKEN_DB_AUTH",
"AZURE_POSTGRESQL_AUTH",
"DATABASE_DISABLE_PREPARED_STATEMENTS",
"DATABASE_URL",
"DIRECT_URL",
"DATABASE_URL_READ_REPLICA",
@ -656,6 +657,83 @@ def test_reader_url_left_alone_when_writer_has_no_params(monkeypatch):
)
# ---------------------------------------------------------------------------
# DATABASE_DISABLE_PREPARED_STATEMENTS
# ---------------------------------------------------------------------------
def test_disable_prepared_statements_appends_pgbouncer_to_assembled_writer(monkeypatch):
monkeypatch.setenv("DATABASE_DISABLE_PREPARED_STATEMENTS", "true")
monkeypatch.setenv("DATABASE_HOST", "writer.example.com")
monkeypatch.setenv("DATABASE_USER", "litellm")
monkeypatch.setenv("DATABASE_NAME", "litellm_db")
monkeypatch.setenv("DATABASE_PASSWORD", "s3cr3t")
assert _apply() is True
assert os.environ["DATABASE_URL"] == (
"postgresql://litellm:s3cr3t@writer.example.com:5432/litellm_db?pgbouncer=true"
)
assert "DIRECT_URL" not in os.environ
def test_disable_prepared_statements_appends_pgbouncer_to_pinned_writer(monkeypatch):
"""The componentized entrypoints (gateway / backend / migrations) receive a
pinned DATABASE_URL and call apply_to_env; without the pgbouncer param Prisma
keeps named prepared statements and 42P05 collisions surface behind a
transaction-pooling pgbouncer."""
monkeypatch.setenv("DATABASE_DISABLE_PREPARED_STATEMENTS", "true")
monkeypatch.setenv("DATABASE_URL", "postgresql://u:p@db.example.com:5432/litellm_db")
assert _apply() is False
assert os.environ["DATABASE_URL"] == "postgresql://u:p@db.example.com:5432/litellm_db?pgbouncer=true"
def test_disable_prepared_statements_respects_a_pinned_pgbouncer_value(monkeypatch):
monkeypatch.setenv("DATABASE_DISABLE_PREPARED_STATEMENTS", "true")
monkeypatch.setenv("DATABASE_URL", "postgresql://u:p@db.example.com:5432/litellm_db?pgbouncer=false")
_apply()
assert os.environ["DATABASE_URL"] == "postgresql://u:p@db.example.com:5432/litellm_db?pgbouncer=false"
def test_disable_prepared_statements_applies_to_direct_url(monkeypatch):
monkeypatch.setenv("DATABASE_DISABLE_PREPARED_STATEMENTS", "true")
monkeypatch.setenv("DATABASE_URL", "postgresql://u:p@db.example.com:5432/litellm_db")
monkeypatch.setenv("DIRECT_URL", "postgresql://u:p@direct.example.com:5432/litellm_db")
_apply()
assert os.environ["DIRECT_URL"] == "postgresql://u:p@direct.example.com:5432/litellm_db?pgbouncer=true"
def test_reader_inherits_pgbouncer_from_disable_prepared_statements(monkeypatch):
monkeypatch.setenv("DATABASE_DISABLE_PREPARED_STATEMENTS", "true")
monkeypatch.setenv("DATABASE_URL", "postgresql://u:p@writer.example.com:5432/db")
monkeypatch.setenv("DATABASE_URL_READ_REPLICA", "postgresql://u:p@reader.example.com:5432/db")
_apply()
query = urllib.parse.parse_qs(urllib.parse.urlsplit(os.environ["DATABASE_URL_READ_REPLICA"]).query)
assert query["pgbouncer"] == ["true"]
def test_disable_prepared_statements_off_leaves_urls_alone(monkeypatch):
monkeypatch.setenv("DATABASE_DISABLE_PREPARED_STATEMENTS", "false")
monkeypatch.setenv("DATABASE_URL", "postgresql://u:p@db.example.com:5432/litellm_db")
_apply()
assert os.environ["DATABASE_URL"] == "postgresql://u:p@db.example.com:5432/litellm_db"
def test_disable_prepared_statements_rejects_an_unreadable_value(monkeypatch):
monkeypatch.setenv("DATABASE_DISABLE_PREPARED_STATEMENTS", "enabled")
with pytest.raises(ValidationError, match="DATABASE_DISABLE_PREPARED_STATEMENTS"):
DatabaseURLSettings.from_env()
def test_unsupported_db_scheme_message_names_var_and_scheme():
msg = unsupported_db_scheme_message("DIRECT_URL", "sqlite")
assert "DIRECT_URL" in msg

134
tests/test_litellm/proxy/pass_through_endpoints/test_managed_id_rewriter.py
View file

@ -1,12 +1,15 @@
import datetime
import json
from collections.abc import AsyncIterator, Iterable
from unittest.mock import AsyncMock, MagicMock
import pytest
from litellm.proxy._types import ProxyException, UserAPIKeyAuth
from litellm.proxy.pass_through_endpoints.managed_id_codec import new_managed_id
from litellm.proxy.pass_through_endpoints.managed_id_codec import decode, new_managed_id
from litellm.proxy.pass_through_endpoints.managed_id_rewriter import (
list_passthrough_ids_from_db,
rewrite_streamed_response_ids,
)
@ -27,9 +30,39 @@ def _prisma_client(file_rows=None, batch_rows=None) -> MagicMock:
pc.db.litellm_managedobjecttable.find_many = AsyncMock(
side_effect=lambda *args, take=None, **kwargs: list(batch_rows or [])[:take]
)
pc.db.litellm_managedobjecttable.upsert = AsyncMock(return_value=None)
return pc
RAW_RESPONSE_ID = "resp_0123456789abcdef"
def _response_stream_bytes(raw_id: str = RAW_RESPONSE_ID) -> bytes:
events = (
("response.created", {"type": "response.created", "response": {"id": raw_id, "status": "in_progress"}}),
("response.output_text.delta", {"type": "response.output_text.delta", "delta": "mango"}),
("response.completed", {"type": "response.completed", "response": {"id": raw_id, "status": "completed"}}),
)
return b"".join(f"event: {name}\ndata: {json.dumps(payload)}\n\n".encode() for name, payload in events)
async def _chunks(payload: bytes, size: int) -> AsyncIterator[bytes]:
for start in range(0, len(payload), size):
yield payload[start : start + size]
async def _collect(stream: AsyncIterator[bytes]) -> bytes:
return b"".join([chunk async for chunk in stream])
def _response_ids(sse: bytes) -> Iterable[str]:
for line in sse.decode().splitlines():
if line.startswith("data:"):
event = json.loads(line[len("data:") :])
if "response" in event:
yield event["response"]["id"]
def _file_row(unified_id: str) -> MagicMock:
row = MagicMock()
row.unified_file_id = unified_id
@ -67,9 +100,7 @@ def _batch_row(unified_id: str) -> MagicMock:
),
],
)
async def test_list_batches_out_of_range_limit_raises_400(
limit, expected_message, expected_openai_code
):
async def test_list_batches_out_of_range_limit_raises_400(limit, expected_message, expected_openai_code):
pc = _prisma_client(batch_rows=[_batch_row(new_managed_id("openai", "batch_abc"))])
with pytest.raises(ProxyException) as exc:
@ -147,3 +178,98 @@ async def test_list_files_drops_batch_guardrail_key_persisted_by_an_older_proxy(
assert result is not None
assert "litellm_batch_guardrail" not in result["data"][0]
assert result["data"][0]["filename"] == "test.jsonl"
@pytest.mark.asyncio
@pytest.mark.parametrize("chunk_size", [1, 7, 4096])
async def test_streamed_response_is_owned_and_rewritten_across_chunk_boundaries(chunk_size: int):
"""A streamed POST /v1/responses records the caller as owner once and returns
the minted id in every event, no matter how the transport splits the SSE bytes."""
pc = _prisma_client()
output = await _collect(
rewrite_streamed_response_ids(
stream=_chunks(_response_stream_bytes(), chunk_size),
provider="openai",
method="POST",
route="/openai_passthrough/v1/responses",
user_api_key_dict=_user(),
prisma_client=pc,
)
)
pc.db.litellm_managedobjecttable.upsert.assert_awaited_once()
created = pc.db.litellm_managedobjecttable.upsert.await_args.kwargs["data"]["create"]
assert created["created_by"] == "user-1"
assert created["team_id"] == "team-1"
assert created["file_purpose"] == "response"
assert created["model_object_id"] == f"passthrough:openai:{RAW_RESPONSE_ID}"
managed_id = created["unified_object_id"]
assert decode(managed_id).raw_provider_id == RAW_RESPONSE_ID
assert list(_response_ids(output)) == [managed_id, managed_id]
assert RAW_RESPONSE_ID.encode() not in output
assert output == _response_stream_bytes(managed_id)
@pytest.mark.asyncio
async def test_streamed_response_with_cr_only_frame_delimiters_is_still_owned_and_rewritten():
"""SSE also terminates lines with a lone CR; those frames must mint and rewrite too."""
pc = _prisma_client()
payload = _response_stream_bytes().replace(b"\n", b"\r")
output = await _collect(
rewrite_streamed_response_ids(
stream=_chunks(payload, 7),
provider="openai",
method="POST",
route="/openai_passthrough/v1/responses",
user_api_key_dict=_user(),
prisma_client=pc,
)
)
pc.db.litellm_managedobjecttable.upsert.assert_awaited_once()
managed_id = pc.db.litellm_managedobjecttable.upsert.await_args.kwargs["data"]["create"]["unified_object_id"]
assert RAW_RESPONSE_ID.encode() not in output
assert output == _response_stream_bytes(managed_id).replace(b"\n", b"\r")
@pytest.mark.asyncio
async def test_streamed_bytes_untouched_on_routes_without_a_response_id():
pc = _prisma_client()
payload = _response_stream_bytes()
output = await _collect(
rewrite_streamed_response_ids(
stream=_chunks(payload, 5),
provider="openai",
method="POST",
route="/openai_passthrough/v1/chat/completions",
user_api_key_dict=_user(),
prisma_client=pc,
)
)
assert output == payload
pc.db.litellm_managedobjecttable.upsert.assert_not_awaited()
@pytest.mark.asyncio
async def test_streamed_response_stays_raw_and_intact_when_the_row_cannot_be_persisted():
pc = _prisma_client()
pc.db.litellm_managedobjecttable.upsert = AsyncMock(side_effect=RuntimeError("db down"))
payload = _response_stream_bytes()
output = await _collect(
rewrite_streamed_response_ids(
stream=_chunks(payload, 3),
provider="openai",
method="POST",
route="/openai_passthrough/v1/responses",
user_api_key_dict=_user(),
prisma_client=pc,
)
)
assert output == payload
pc.db.litellm_managedobjecttable.upsert.assert_awaited_once()

80
tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py
View file

@ -1493,6 +1493,86 @@ async def test_pass_through_request_sse_response_marks_logging_obj_as_stream():
assert logging_obj.model_call_details["stream"] is True
@pytest.mark.asyncio
async def test_pass_through_request_streamed_response_is_owned_by_the_caller():
"""
Regression: with passthrough_managed_object_ids on, a streamed
POST /openai_passthrough/v1/responses left the raw resp_ id in the stream and
recorded no owner, so any other key could read, continue, and delete it.
"""
import litellm
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.types.llms.custom_http import httpxSpecialProvider
raw_id = "resp_0123456789abcdef"
upstream_body = (
b'event: response.created\ndata: {"type": "response.created", "response": {"id": "%s"}}\n\n'
b'event: response.completed\ndata: {"type": "response.completed", "response": {"id": "%s"}}\n\n'
) % (raw_id.encode(), raw_id.encode())
prisma_client = MagicMock()
prisma_client.db.litellm_managedobjecttable.find_first = AsyncMock(return_value=None)
prisma_client.db.litellm_managedobjecttable.upsert = AsyncMock(return_value=None)
prisma_client.db.litellm_managedfiletable.find_first = AsyncMock(return_value=None)
def transport_handler(upstream_request: httpx.Request) -> httpx.Response:
return httpx.Response(200, content=upstream_body, headers={"content-type": "text/event-stream"})
real_handler = get_async_httpx_client(
llm_provider=httpxSpecialProvider.PassThroughEndpoint,
params={"timeout": resolve_pass_through_request_timeout(None)},
)
cache_dict = litellm.in_memory_llm_clients_cache.cache_dict
cache_key = next(key for key, cached in cache_dict.items() if cached is real_handler)
cache_dict[cache_key] = SimpleNamespace(client=httpx.AsyncClient(transport=httpx.MockTransport(transport_handler)))
mock_proxy_logging = MagicMock()
mock_proxy_logging.pre_call_hook = AsyncMock(side_effect=lambda user_api_key_dict, data, call_type: data)
mock_proxy_logging.post_call_failure_hook = AsyncMock()
mock_proxy_logging.post_call_response_headers_hook = AsyncMock(return_value={})
mock_proxy_logging.get_proxy_hook = MagicMock(return_value=MagicMock())
mock_request = MagicMock(spec=Request)
mock_request.method = "POST"
mock_request.scope = {"path": "/openai_passthrough/v1/responses"}
mock_request.url = MagicMock()
mock_request.url.path = "/openai_passthrough/v1/responses"
mock_request.body = AsyncMock(return_value=b'{"model": "gpt-5.1", "input": "hi", "stream": true}')
mock_request.headers = Headers({"content-type": "application/json"})
mock_request.query_params = QueryParams({})
flag_on = {"passthrough_managed_object_ids": True}
proxy_server_globals = (
patch("litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging), # test-quality-ok: read at call time
patch("litellm.proxy.proxy_server.general_settings", flag_on), # test-quality-ok: read at call time
patch("litellm.proxy.proxy_server.prisma_client", prisma_client), # test-quality-ok: read at call time
)
try:
with ExitStack() as stack:
for patched_global in proxy_server_globals:
stack.enter_context(patched_global)
response = await pass_through_request(
request=mock_request,
target="https://api.openai.com/v1/responses",
custom_headers={},
user_api_key_dict=UserAPIKeyAuth(user_id="user-a", team_id="team-a"),
custom_llm_provider="openai",
)
streamed = b"".join([chunk async for chunk in response.body_iterator])
finally:
cache_dict[cache_key] = real_handler
assert response.status_code == 200
prisma_client.db.litellm_managedobjecttable.upsert.assert_awaited_once()
created = prisma_client.db.litellm_managedobjecttable.upsert.await_args.kwargs["data"]["create"]
assert created["created_by"] == "user-a"
assert created["team_id"] == "team-a"
assert created["model_object_id"] == f"passthrough:openai:{raw_id}"
managed_id = created["unified_object_id"]
assert raw_id.encode() not in streamed
assert streamed == upstream_body.replace(raw_id.encode(), managed_id.encode())
@pytest.mark.asyncio
async def test_create_pass_through_endpoint():
"""

59
tests/test_litellm/test_cost_calculator.py
View file

@ -1,4 +1,7 @@
import json
from pathlib import Path
import pytest
@ -14,7 +17,13 @@ from litellm.cost_calculator import (
response_cost_calculator,
)
from litellm.types.llms.openai import OpenAIRealtimeStreamList
from litellm.types.utils import ModelInfo, ModelResponse, PromptTokensDetailsWrapper, Usage
from litellm.types.utils import (
CacheCreationTokenDetails,
ModelInfo,
ModelResponse,
PromptTokensDetailsWrapper,
Usage,
)
from litellm.utils import TranscriptionResponse
@ -3781,3 +3790,51 @@ def test_completion_cost_prices_anthropic_shaped_cache_read_tokens(_local_model_
)
assert cost == pytest.approx(3 * 4e-6 + 4014 * 4e-7 + 5 * 2e-5, rel=1e-9)
@pytest.mark.parametrize(
("model", "expected_1hr_rate"),
[("claude-3-haiku-20240307", 5e-07), ("claude-3-opus-20240229", 3e-05)],
)
def test_claude_3_one_hour_cache_writes_bill_at_double_input(
_local_model_cost_map, model: str, expected_1hr_rate: float
):
"""Regression: both models carried the Sonnet 1h cache-write rate (6e-06) instead of
2x their own input price, overbilling haiku 12x and underbilling opus 5x."""
usage = Usage(
prompt_tokens=1000,
completion_tokens=0,
total_tokens=1000,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=0,
cache_creation_tokens=1000,
cache_creation_token_details=CacheCreationTokenDetails(
ephemeral_5m_input_tokens=0, ephemeral_1h_input_tokens=1000
),
),
)
prompt_cost, _ = cost_per_token(model=model, usage_object=usage, custom_llm_provider="anthropic")
assert prompt_cost == pytest.approx(1000 * expected_1hr_rate, rel=1e-9)
def test_every_one_hour_cache_write_rate_is_double_its_input_rate():
"""Guard against pasting one model's 1h cache-write price onto another: every provider
LiteLLM tracks (Anthropic, Bedrock, Vertex, Azure) publishes the 1h write at 2x input."""
cost_map = json.loads(
(Path(__file__).parents[2] / "model_prices_and_context_window.json").read_text()
)
one_hour_prefix = "cache_creation_input_token_cost_above_1hr"
deviations = {
(name, key): (entry["input_cost_per_token" + key[len(one_hour_prefix) :]], entry[key])
for name, entry in cost_map.items()
if isinstance(entry, dict)
for key in entry
if key.startswith(one_hour_prefix)
and entry[key] != pytest.approx(2 * entry["input_cost_per_token" + key[len(one_hour_prefix) :]], rel=1e-9)
}
assert deviations == {}

120
tests/test_litellm/test_responses_id_security.py
View file

@ -9,8 +9,15 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi import HTTPException
from litellm.proxy.hooks.responses_id_security import ResponsesIDSecurity
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.proxy.hooks.responses_id_security import (
ResponsesIDSecurity,
_is_responses_api_create_route,
)
from litellm.types.llms.openai import (
ResponseCompletedEvent,
ResponsesAPIResponse,
ResponsesAPIStreamEvents,
)
from litellm.types.utils import SpecialEnums
@ -575,6 +582,115 @@ class TestAsyncPreCallHook:
assert "team" in exc_info.value.detail.lower()
class TestIsResponsesApiCreateRoute:
"""Test the route gate that decides whether a streamed response id is encrypted."""
@pytest.mark.parametrize(
"route",
[
"/v1/responses",
"/responses",
"/openai/v1/responses",
],
)
def test_create_routes_match(self, route):
assert _is_responses_api_create_route(route) is True
@pytest.mark.parametrize(
"route",
[
None,
"/chat/completions",
"/openai/v1/chat/completions",
"/v1/responses/{response_id}",
"/openai/v1/responses/{response_id}",
"/v1/responsesX",
"/responsesX",
],
)
def test_non_create_routes_do_not_match(self, route):
assert _is_responses_api_create_route(route) is False
class TestAsyncPostCallStreamingIteratorHook:
"""Regression test for LIT-6167: streamed responses on /openai/v1/responses and
/responses must have their ids security-encrypted, not just on the exact
/v1/responses path. A streamed create emits ResponseCompletedEvent, whose
client-visible id lives on event.response.id, so the test drives that production
event shape (not a top-level id) and uses real encryption, asserting the id
round-trips back to the raw provider id plus the caller's user/team, which is the
access-control wrapper the aliases were leaking without."""
@staticmethod
async def _agen(chunks):
for chunk in chunks:
yield chunk
@staticmethod
def _completed_event(response_id):
return ResponseCompletedEvent(
type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
response=ResponsesAPIResponse(
id=response_id,
created_at=0,
model="gpt-5.1",
object="response",
output=[],
parallel_tool_calls=False,
tool_choice="auto",
tools=[],
),
)
async def _drain_streamed_id(self, responses_id_security, route, monkeypatch):
monkeypatch.setenv("LITELLM_SALT_KEY", "sk-test-salt-key-abcdefghij")
event = self._completed_event("resp_rawprovider123")
mock_auth = MagicMock()
mock_auth.user_id = "user-a"
mock_auth.team_id = "team-a"
mock_auth.request_route = route
collected = [
out
async for out in responses_id_security.async_post_call_streaming_iterator_hook(
user_api_key_dict=mock_auth,
response=self._agen([event]),
request_data={},
)
]
return collected[0].response.id
@pytest.mark.asyncio
@pytest.mark.parametrize(
"route",
["/v1/responses", "/responses", "/openai/v1/responses"],
)
async def test_streamed_id_encrypted_on_all_responses_routes(
self, responses_id_security, route, monkeypatch
):
streamed_id = await self._drain_streamed_id(responses_id_security, route, monkeypatch)
assert streamed_id != "resp_rawprovider123"
assert responses_id_security._is_encrypted_response_id(streamed_id)
assert responses_id_security._decrypt_response_id(streamed_id) == (
"resp_rawprovider123",
"user-a",
"team-a",
)
@pytest.mark.asyncio
async def test_streamed_id_untouched_on_non_responses_route(
self, responses_id_security, monkeypatch
):
streamed_id = await self._drain_streamed_id(
responses_id_security, "/chat/completions", monkeypatch
)
assert streamed_id == "resp_rawprovider123"
assert not responses_id_security._is_encrypted_response_id(streamed_id)
class TestAsyncPostCallSuccessHook:
"""Test async_post_call_success_hook function"""

42
tests/test_litellm/test_router.py
View file

@ -10607,3 +10607,45 @@ async def test_async_function_with_fallbacks_scrubs_spoofed_values_from_sibling_
assert litellm_metadata["client_key"] == "client_value"
assert metadata["attempted_fallbacks"] == 0
assert metadata["original_model_group"] == "gpt-3.5-turbo"
def _permission_denied_error() -> litellm.PermissionDeniedError:
return litellm.PermissionDeniedError(
message="OpenrouterException - this key has no access to the model",
llm_provider="openrouter",
model="openrouter/openai/gpt-4o",
response=httpx.Response(status_code=403, request=httpx.Request(method="POST", url="https://openrouter.ai")),
)
def test_permission_denied_error_is_not_retried_against_a_single_deployment():
router = litellm.Router(
model_list=[
{"model_name": "gpt-4o", "litellm_params": {"model": "openrouter/openai/gpt-4o", "api_key": "sk-test"}},
]
)
with pytest.raises(litellm.PermissionDeniedError):
router.should_retry_this_error(
error=_permission_denied_error(),
healthy_deployments=router.model_list,
all_deployments=router.model_list,
)
def test_permission_denied_error_is_retried_when_other_deployments_exist():
router = litellm.Router(
model_list=[
{"model_name": "gpt-4o", "litellm_params": {"model": "openrouter/openai/gpt-4o", "api_key": "sk-test"}},
{"model_name": "gpt-4o", "litellm_params": {"model": "openai/gpt-4o", "api_key": "sk-test"}},
]
)
assert (
router.should_retry_this_error(
error=_permission_denied_error(),
healthy_deployments=router.model_list,
all_deployments=router.model_list,
)
is True
)

39
tests/test_litellm/test_router_order_fallback.py
View file

@ -367,6 +367,45 @@ async def test_router_order_fallback_with_wildcard_model_group():
assert response._hidden_params["model_id"] == "2"
@pytest.mark.asyncio
async def test_router_order_fallback_with_hidden_model_group_alias():
router = Router(
model_list=[
{
"model_name": "canonical-model",
"litellm_params": {
"model": "gpt-4o",
"api_key": "bad",
"mock_response": Exception("fail order 1"),
"order": 1,
},
"model_info": {"id": "1"},
},
{
"model_name": "canonical-model",
"litellm_params": {
"model": "gpt-4o",
"api_key": "good",
"mock_response": "success from order 2",
"order": 2,
},
"model_info": {"id": "2"},
},
],
model_group_alias={"hidden-alias": {"model": "canonical-model", "hidden": True}},
num_retries=0,
)
assert "hidden-alias" not in {deployment["model_name"] for deployment in router.get_model_list() or []}
response = await router.acompletion(
model="hidden-alias",
messages=[{"role": "user", "content": "hi"}],
)
assert response._hidden_params["model_id"] == "2"
def test_check_non_standard_fallback_format():
from litellm.router_utils.fallback_event_handlers import (
_check_non_standard_fallback_format,

6
type-discipline-budget.json
View file

@ -1,9 +1,9 @@
{
"LIT001": {
"limit": 22749
"limit": 22733
},
"LIT002": {
"limit": 26866
"limit": 26864
},
"LIT003": {
"limit": 269
@ -27,7 +27,7 @@
"limit": 0
},
"LIT010": {
"limit": 16655
"limit": 16621
},
"LIT011": {
"limit": 5585

3
ui/litellm-dashboard/eslint-suppressions.json
View file

@ -1465,9 +1465,6 @@
"src/components/add_model/conditional_public_model_name.tsx": {
"local/filename-pascal-case": {
"count": 1
},
"local/no-complex-jsx-arrow": {
"count": 1
}
},
"src/components/add_model/handle_add_auto_router_submit.tsx": {

21
ui/litellm-dashboard/src/components/add_model/conditional_public_model_name.test.tsx
View file

@ -1,4 +1,5 @@
import { render, screen } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import React, { useEffect, useRef } from "react";
import { useFormContext, useWatch } from "react-hook-form";
import { describe, expect, it } from "vitest";
@ -69,4 +70,24 @@ describe("ConditionalPublicModelName", () => {
expect(screen.getByText("my-custom-model")).toBeInTheDocument();
expect(screen.queryByDisplayValue("custom")).not.toBeInTheDocument();
});
it("keeps the public name input focused across keystrokes", async () => {
const user = userEvent.setup();
render(
<MountedFormHost
defaultValues={{
model: ["gpt-4"],
model_mappings: [{ public_name: "gpt-4", litellm_model: "gpt-4" }],
}}
>
<ConditionalPublicModelName />
</MountedFormHost>,
);
const input = screen.getByDisplayValue("gpt-4");
await user.type(input, "-prod");
expect(input).toHaveValue("gpt-4-prod");
expect(input).toHaveFocus();
});
});

155
ui/litellm-dashboard/src/components/add_model/conditional_public_model_name.tsx
View file

@ -36,6 +36,82 @@ const modelMappingsRule = {
const tooltipCodeClassName = "rounded-sm bg-background/20 px-1 py-0.5 font-mono text-xs";
const ANTHROPIC_1M_HEADERS = JSON.stringify({ extra_headers: { "anthropic-beta": "context-1m-2025-08-07" } }, null, 2);
const publicNameTooltipContent = (
<div className="flex flex-col gap-2 text-left font-normal">
<div>The name you specify in your API calls to LiteLLM Proxy</div>
<div>
<strong>Example:</strong> If you name your public model <code className={tooltipCodeClassName}>example-name</code>
, and choose <code className={tooltipCodeClassName}>openai/qwen-plus-latest</code> as the LiteLLM model
</div>
<div>
<strong>Usage:</strong> You make an API call to the LiteLLM proxy with{" "}
<code className={tooltipCodeClassName}>model = &quot;example-name&quot;</code>
</div>
<div>
<strong>Result:</strong> LiteLLM sends <code className={tooltipCodeClassName}>qwen-plus-latest</code> to the
provider
</div>
</div>
);
const PublicNameInput: React.FC<{ readonly index: number; readonly value: string }> = ({ index, value }) => {
const form = useFormContext<MountedFormValues>();
const selectedProvider = useWatch({ control: form.control, name: "custom_llm_provider" });
const handleChange = (event: React.ChangeEvent<HTMLInputElement>) => {
const typed = event.target.value;
const litellmParams = form.getValues("litellm_extra_params") as string | undefined;
const wantsAnthropic1m =
selectedProvider === Providers.Anthropic && typed.endsWith("-1m") && (litellmParams ?? "").trim() === "";
if (wantsAnthropic1m) {
form.setValue("litellm_extra_params", ANTHROPIC_1M_HEADERS);
}
const publicName = wantsAnthropic1m ? typed.slice(0, -"-1m".length) : typed;
const current = (form.getValues("model_mappings") as ModelMapping[]) ?? [];
form.setValue(
"model_mappings",
current.map((mapping, mappingIndex) =>
mappingIndex === index ? { ...mapping, public_name: publicName } : mapping,
),
);
};
return <Input value={value} onChange={handleChange} />;
};
/**
* Module-level so the header and cell renderers keep a stable identity: React treats a renderer
* declared inside the component as a new element type on every render and remounts the input,
* which drops focus after each keystroke.
*/
const columns: ColumnDef<ModelMapping>[] = [
{
id: "public_name",
accessorKey: "public_name",
header: () => (
<span className="flex items-center">
Public Model Name
<SimpleTooltip content={publicNameTooltipContent} width="500px" />
</span>
),
cell: ({ row }) => <PublicNameInput index={row.index} value={row.original.public_name} />,
},
{
id: "litellm_model",
accessorKey: "litellm_model",
header: () => (
<span className="flex items-center">
LiteLLM Model Name
<SimpleTooltip content={<div>The model name LiteLLM will send to the LLM API</div>} width="360px" />
</span>
),
},
];
const ConditionalPublicModelName: React.FC = () => {
const form = useFormContext<MountedFormValues>();
@ -124,85 +200,6 @@ const ConditionalPublicModelName: React.FC = () => {
if (!showPublicModelName) return null;
const publicNameTooltipContent = (
<div className="flex flex-col gap-2 text-left font-normal">
<div>The name you specify in your API calls to LiteLLM Proxy</div>
<div>
<strong>Example:</strong> If you name your public model{" "}
<code className={tooltipCodeClassName}>example-name</code>, and choose{" "}
<code className={tooltipCodeClassName}>openai/qwen-plus-latest</code> as the LiteLLM model
</div>
<div>
<strong>Usage:</strong> You make an API call to the LiteLLM proxy with{" "}
<code className={tooltipCodeClassName}>model = &quot;example-name&quot;</code>
</div>
<div>
<strong>Result:</strong> LiteLLM sends <code className={tooltipCodeClassName}>qwen-plus-latest</code> to the
provider
</div>
</div>
);
const liteLLMModelTooltipContent = <div>The model name LiteLLM will send to the LLM API</div>;
const columns: ColumnDef<ModelMapping>[] = [
{
id: "public_name",
accessorKey: "public_name",
header: () => (
<span className="flex items-center">
Public Model Name
<SimpleTooltip content={publicNameTooltipContent} width="500px" />
</span>
),
cell: ({ row }) => {
return (
<Input
value={row.original.public_name}
onChange={(e) => {
const newValue = e.target.value;
const newMappings = [...((form.getValues("model_mappings") as ModelMapping[]) ?? [])];
// Check conditions for Anthropic -1m suffix handling
const isAnthropic = selectedProvider === Providers.Anthropic;
const endsWith1m = newValue.endsWith("-1m");
const litellmParams = form.getValues("litellm_extra_params") as string | undefined;
const isLitellmParamsEmpty = !litellmParams || litellmParams.trim() === "";
let finalPublicName = newValue;
if (isAnthropic && endsWith1m && isLitellmParamsEmpty) {
// Set litellm params with extra_headers
const litellmParamsValue = JSON.stringify(
{ extra_headers: { "anthropic-beta": "context-1m-2025-08-07" } },
null,
2,
);
form.setValue("litellm_extra_params", litellmParamsValue);
// Remove -1m suffix from public_name
finalPublicName = newValue.slice(0, -3); // Remove "-1m" (3 characters)
}
newMappings[row.index].public_name = finalPublicName;
form.setValue("model_mappings", newMappings);
}}
/>
);
},
},
{
id: "litellm_model",
accessorKey: "litellm_model",
header: () => (
<span className="flex items-center">
LiteLLM Model Name
<SimpleTooltip content={liteLLMModelTooltipContent} width="360px" />
</span>
),
},
];
return (
<MountedFormField
name="model_mappings"

4
ui/litellm-dashboard/src/lib/http/schema.d.ts generated vendored
View file

@ -36262,7 +36262,9 @@ export interface components {
} | null;
/** Model Max Budget Usage */
model_max_budget_usage?: {
[key: string]: unknown;
[key: string]: {
[key: string]: unknown;
};
} | null;
/**
* Models