scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-16 04:29:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 138

refactor(typing): clear 1143 basedpyright Any errors across 273 files (#46574)

Browse source 
* refactor(typing): narrow Any sources in proxy, guardrails, bedrock and responses code

Turns untyped json.loads, response.json(), header lookups and forwarded
kwargs into object or precise types at their source, so downstream code
type-checks without reportAny / reportExplicitAny. No runtime behavior
changes.

* refactor(typing): narrow Any sources in caching, MCP, providers and proxy request code

Decoded JSON, cache reads and header lookups now enter as object or
precise types, prisma tables used by the shadow-eval logger get typed
Protocol views, and a few SDK parameters move from dict[str, Any] to
Mapping[str, object]. No runtime behavior changes.

* refactor(typing): narrow Any sources in proxy server, router, logging and shared utils

* refactor(typing): narrow Any sources in provider transformations, integrations and proxy endpoints

* refactor(typing): narrow Any sources across providers, proxy and router

Deepens the Any cleanup on the hottest files: typed helpers in
core_helpers for attribute and item access, identity TypeAdapter(object)
wraps at untyped boundaries, explicit return annotations on shared
provider and proxy helpers, and Protocols in place of Any parameters.
No intended behavior change.
This commit is contained in:
Mateo Wang 2026-10-15 14:08:47 -07:00 • committed by GitHub
parent 46913be92b
commit 130699ca23
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
272 changed files with 2092 additions and 1160 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
enterprise/litellm_enterprise/proxy/auth/user_api_key_auth.py
View file

@ -1,14 +1,16 @@
from typing import Any, Optional
from collections.abc import Awaitable, Callable
from fastapi import Request
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import ProxyException, UserAPIKeyAuth
from litellm.proxy._types import ProxyException
async def enterprise_custom_auth(
request: Request, api_key: str, user_custom_auth: Optional[Any]
) -> Optional[UserAPIKeyAuth]:
request: Request,
api_key: str,
user_custom_auth: Callable[[Request, str], Awaitable[object]] | None,
) -> object:
from litellm_enterprise.proxy.proxy_server import custom_auth_settings
if user_custom_auth is None:

17
litellm/_logging.py
View file

@ -6,12 +6,14 @@ import logging
import os
import re
import sys
from collections.abc import Iterator
from collections.abc import Iterator, Mapping
from datetime import datetime
from logging import Formatter
from typing import Final, TextIO
from urllib.parse import unquote
from pydantic import TypeAdapter
import litellm
from litellm.constants import (
LITELLM_TRUNCATED_PAYLOAD_FIELD,
@ -84,6 +86,7 @@ _redact_string = redact_string
_REDACTED_RECORD_ATTR: Final = "litellm_redacted"
_REDACTED_STAMP: Final = object()
_DECODED_LOG_PAYLOAD: Final = TypeAdapter(object)
_UNREDACTED_SCALAR_TYPES: Final = (bool, int, float, type(None))
@ -91,6 +94,10 @@ def _is_redacted(record: logging.LogRecord) -> bool:
return getattr(record, _REDACTED_RECORD_ATTR, None) is _REDACTED_STAMP
def _record_attributes(record: logging.LogRecord) -> Mapping[str, object]:
return record.__dict__
def _scrubbing_changed_nothing(scrubbed: object, original: object) -> bool:
try:
return bool(scrubbed == original)
@ -438,7 +445,7 @@ def _process_record(record: logging.LogRecord, *, base64_limit: int, text_limit:
key, substituted_color if key == "color_message" and substituted_color is not None else value
),
)
for key, value in record.__dict__.items()
for key, value in _record_attributes(record).items()
if key not in _STANDARD_RECORD_ATTRS
and key != _REDACTED_RECORD_ATTR
and not isinstance(value, _UNREDACTED_SCALAR_TYPES)
@ -669,7 +676,7 @@ def _try_parse_json_message(message: str) -> dict[str, object] | None:
msg_stripped: Final = message.strip()
if not (msg_stripped.startswith("{") or msg_stripped.startswith("[")):
return None
parsed: Final[object] = safe_json_loads(message, default=None)
parsed: Final = _DECODED_LOG_PAYLOAD.validate_python(safe_json_loads(message, default=None))
if parsed is None or not isinstance(parsed, dict):
return None
return parsed
@ -699,7 +706,7 @@ def _try_parse_embedded_python_dict(message: str) -> dict[str, object] | None:
if depth == 0:
substr = message[start : j + 1]
try:
result: object = ast.literal_eval(substr)
result: object = _DECODED_LOG_PAYLOAD.validate_python(ast.literal_eval(substr))
if isinstance(result, dict) and len(result) > 0:
return result
except (ValueError, SyntaxError, TypeError):
@ -757,7 +764,7 @@ class JsonFormatter(Formatter):
json_record[key] = value
# Include extra attributes passed via logger.debug("msg", extra={...})
for key, value in record.__dict__.items():
for key, value in _record_attributes(record).items():
if key not in _NON_EXTRA_RECORD_ATTRS and key not in json_record:
json_record[key] = value

14
litellm/_redis.py
View file

@ -19,6 +19,7 @@ from urllib.parse import urlsplit, urlunsplit
import redis
import redis.asyncio as async_redis
from pydantic import TypeAdapter
from redis.credentials import CredentialProvider
from litellm import get_secret, get_secret_str
@ -40,6 +41,7 @@ from ._logging import verbose_logger
_generate_gcp_iam_access_token = generate_gcp_iam_access_token
AZURE_REDIS_SCOPE: Final = "https://redis.azure.com/.default"
_UNWRAPPED_INIT: Final = TypeAdapter(object)
_AWS_IAM_KWARG_NAMES: Final = (
"aws_iam_auth",
@ -64,7 +66,7 @@ def _unwrapped_init_args(cls: type) -> frozenset[str]:
``@deprecated_args`` too, which the same class of bug as ``_init_arg_names``
would otherwise silently empty this allowlist through (see its docstring).
"""
spec: Final = inspect.getfullargspec(inspect.unwrap(cls.__init__))
spec: Final = inspect.getfullargspec(_UNWRAPPED_INIT.validate_python(inspect.unwrap(cls.__init__)))
return frozenset(spec.args + spec.kwonlyargs)
@ -112,7 +114,7 @@ def _init_arg_names(cls: type) -> frozenset[str]:
name
for klass in inspect.getmro(cls)
if klass is not object
for spec in (inspect.getfullargspec(inspect.unwrap(klass.__init__)),)
for spec in (inspect.getfullargspec(_UNWRAPPED_INIT.validate_python(inspect.unwrap(klass.__init__))),)
for name in spec.args + spec.kwonlyargs
)
@ -935,6 +937,10 @@ def get_redis_connection_pool(
return async_redis.BlockingConnectionPool(timeout=REDIS_CONNECTION_POOL_TIMEOUT, **redis_kwargs)
def _mask_for_display(masker: SensitiveDataMasker, redis_kwargs: Mapping[str, object]) -> Mapping[str, object]:
return masker.mask_dict(redis_kwargs)
def _redis_kwargs_for_logging(redis_kwargs: Mapping[str, object]) -> Mapping[str, object]:
return {
key: "<credential provider>"
@ -966,7 +972,7 @@ def _pretty_print_redis_config(redis_kwargs: dict) -> None:
masker = SensitiveDataMasker()
# Mask sensitive data in redis_kwargs
masked_redis_kwargs = masker.mask_dict(redis_kwargs_for_logging)
masked_redis_kwargs = _mask_for_display(masker, redis_kwargs_for_logging)
# Create main panel title
title: Final = Text("Redis Configuration", style="bold blue")
@ -1029,7 +1035,7 @@ def _pretty_print_redis_config(redis_kwargs: dict) -> None:
except ImportError:
# Fallback to simple logging if rich is not available
masker = SensitiveDataMasker()
masked_redis_kwargs = masker.mask_dict(redis_kwargs_for_logging)
masked_redis_kwargs = _mask_for_display(masker, redis_kwargs_for_logging)
verbose_logger.info("Redis configuration: %s", masked_redis_kwargs)
except Exception as e:
verbose_logger.error("Error pretty printing Redis configuration: %s", e)

10
litellm/a2a_protocol/main.py
View file

@ -14,7 +14,9 @@ import datetime
import uuid
from collections.abc import AsyncIterator, Coroutine, Mapping
from types import MappingProxyType, ModuleType
from typing import TYPE_CHECKING, Any, Final, Optional, cast
from typing import TYPE_CHECKING, Final, Optional, cast
from typing_extensions import TypedDict, Unpack
import litellm
from litellm._logging import verbose_logger, verbose_proxy_logger
@ -89,6 +91,10 @@ from litellm.a2a_protocol.exceptions import A2ALocalhostURLError
A2ACardResolver: Final = LiteLLMA2ACardResolver
class _SendMessageOptions(TypedDict, total=False, extra_items=object):
pass
def _set_usage_on_logging_obj(
kwargs: Mapping[str, object],
prompt_tokens: int,
@ -563,7 +569,7 @@ async def asend_message(
def send_message(
a2a_client: "A2AClientType",
request: "SendMessageRequest",
**kwargs: Any,
**kwargs: Unpack[_SendMessageOptions],
) -> LiteLLMSendMessageResponse | Coroutine[object, object, LiteLLMSendMessageResponse]:
"""
Sync: Send a message to an A2A agent.

12
litellm/a2a_protocol/streaming_iterator.py
View file

@ -7,16 +7,21 @@ from collections.abc import AsyncIterator
from datetime import datetime
from typing import TYPE_CHECKING, Final
from pydantic import TypeAdapter
import litellm
from litellm._logging import verbose_logger
from litellm.a2a_protocol.cost_calculator import A2ACostCalculator
from litellm.a2a_protocol.utils import A2ARequestUtils
from litellm.litellm_core_utils.asyncify import asyncify
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
if TYPE_CHECKING:
from a2a.compat.v0_3.types import SendStreamingMessageRequest, SendStreamingMessageResponse
_OPAQUE_PAYLOAD: Final = TypeAdapter(object)
class A2AStreamingIterator:
"""
@ -160,8 +165,11 @@ class A2AStreamingIterator:
# Add final chunk result if available
if self.final_chunk:
try:
chunk_dict: Final = self.final_chunk.model_dump(mode="json", exclude_none=True)
result["result"] = chunk_dict.get("result", {})
chunk_dict: Final = as_str_mapping(
_OPAQUE_PAYLOAD.validate_python(self.final_chunk.model_dump(mode="json", exclude_none=True))
)
if chunk_dict is not None:
result["result"] = chunk_dict.get("result", {})
except Exception:
pass

5
litellm/anthropic_interface/messages/__init__.py
View file

@ -11,7 +11,6 @@ This is an __init__.py file to allow the following interface
"""
from collections.abc import AsyncIterator, Coroutine, Iterator
from typing import Any
from litellm.messages import (
anthropic_messages as _async_anthropic_messages,
@ -102,8 +101,8 @@ def create(
) -> (
AnthropicMessagesResponse
| Iterator[bytes]
| AsyncIterator[Any]
| Coroutine[Any, Any, AnthropicMessagesResponse | AsyncIterator[Any] | Iterator[bytes]]
| AsyncIterator[object]
| Coroutine[object, object, AnthropicMessagesResponse | AsyncIterator[object] | Iterator[bytes]]
):
"""
Async wrapper for Anthropic's messages API

5
litellm/batches/batch_utils.py
View file

@ -5,6 +5,8 @@ from dataclasses import replace as dataclasses_replace
from enum import Enum
from typing import Any, Final, Literal, cast
from pydantic import TypeAdapter
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.get_litellm_params import AWS_CREDENTIAL_KWARGS_KEYS
@ -35,6 +37,7 @@ class BatchCostUsageResult:
_COMPLETED_BATCH_STATUSES: Final = frozenset({"completed", "complete"})
_DECODED_BATCH_LINE: Final = TypeAdapter(object)
def _uses_native_vertex_output(
@ -624,7 +627,7 @@ def _iter_batch_output_entries(file_content: bytes) -> Iterator[dict[str, object
def parse_batch_output_line(line: bytes) -> dict[str, object] | None:
try:
parsed: Final[object] = json.loads(line)
parsed: Final = _DECODED_BATCH_LINE.validate_python(json.loads(line))
except ValueError as e:
verbose_logger.warning("skipping malformed batch output line: %s", str(e))
return None

12
litellm/batches/main.py
View file

@ -127,7 +127,7 @@ async def acreate_batch(
metadata: dict[str, str] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, str] | None = None,
output_expires_after: dict[str, Any] | None = None,
output_expires_after: FileExpiresAfter | None = None,
**kwargs,
) -> LiteLLMBatch:
"""
@ -191,9 +191,9 @@ def create_batch(
metadata: dict[str, str] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, str] | None = None,
output_expires_after: dict[str, Any] | None = None,
output_expires_after: FileExpiresAfter | None = None,
**kwargs,
) -> LiteLLMBatch | Coroutine[Any, Any, LiteLLMBatch]:
) -> LiteLLMBatch | Coroutine[object, None, LiteLLMBatch]:
"""
Creates and executes a batch from an uploaded file of request
@ -246,7 +246,7 @@ def create_batch(
extra_body=extra_body,
)
if output_expires_after is not None:
_create_batch_request["output_expires_after"] = cast(FileExpiresAfter, output_expires_after)
_create_batch_request["output_expires_after"] = output_expires_after
if model is not None:
provider_config = ProviderConfigManager.get_provider_batches_config(
model=model,
@ -568,7 +568,7 @@ def retrieve_batch(
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, str] | None = None,
**kwargs,
) -> LiteLLMBatch | Coroutine[Any, Any, LiteLLMBatch]:
) -> LiteLLMBatch | Coroutine[object, None, LiteLLMBatch]:
"""
Retrieves a batch.
@ -941,7 +941,7 @@ def cancel_batch(
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, str] | None = None,
**kwargs,
) -> LiteLLMBatch | Coroutine[Any, Any, LiteLLMBatch]:
) -> LiteLLMBatch | Coroutine[object, None, LiteLLMBatch]:
"""
Cancels a batch.

9
litellm/budget_manager.py
View file

@ -13,6 +13,8 @@ import threading
import time
from typing import Final, Literal
from pydantic import TypeAdapter
import litellm
from litellm.constants import (
DAYS_IN_A_MONTH,
@ -22,6 +24,8 @@ from litellm.constants import (
)
from litellm.utils import ModelResponse
_BUDGET_SERVER_REPLY: Final = TypeAdapter(object)
class BudgetManager:
def __init__(
@ -211,6 +215,5 @@ class BudgetManager:
elif self.client_type == "hosted":
url: Final = self.api_base + "/set_budget"
data: Final = {"project_name": self.project_name, "user_dict": self.user_dict}
response = litellm.module_level_client.post(url, headers=self.headers, json=data)
response = response.json()
return response
response: Final = litellm.module_level_client.post(url, headers=self.headers, json=data)
return _BUDGET_SERVER_REPLY.validate_python(response.json())

9
litellm/caching/caching.py
View file

@ -18,7 +18,7 @@ from contextlib import contextmanager
from enum import Enum
from typing import Any, Final, Literal, cast
from pydantic import BaseModel
from pydantic import BaseModel, TypeAdapter
import litellm
from litellm._internal_context import current_service_target, service_target
@ -42,6 +42,7 @@ from .redis_semantic_cache import RedisSemanticCache
from .s3_cache import S3Cache
RESPONSE_CACHE_TARGET: Final = "llm_response"
_DECODED_LITERAL: Final = TypeAdapter(object)
@contextmanager
@ -599,9 +600,11 @@ class Cache:
if isinstance(cached_response, dict):
pass
else:
cached_response = json.loads(cached_response) # Convert string to dictionary
cached_response = _DECODED_LITERAL.validate_python(
json.loads(cached_response)
) # Convert string to dictionary
except Exception:
cached_response = ast.literal_eval(cached_response)
cached_response = _DECODED_LITERAL.validate_python(ast.literal_eval(cached_response))
return cached_response
return cached_result

8
litellm/caching/gcs_cache.py
View file

@ -7,6 +7,8 @@ import json
from typing import Final
from urllib.parse import quote
from pydantic import TypeAdapter
from litellm._logging import print_verbose, verbose_logger
from litellm.integrations.gcs_bucket.gcs_bucket_base import GCSBucketBase
from litellm.llms.custom_httpx.http_handler import (
@ -17,6 +19,8 @@ from litellm.llms.custom_httpx.http_handler import (
from .base_cache import BaseCache
_DECODED_JSON: Final = TypeAdapter(object)
class GCSCache(BaseCache):
def __init__(
@ -70,7 +74,7 @@ class GCSCache(BaseCache):
url = f"https://storage.googleapis.com/storage/v1/b/{bucket_name}/o/{quote(object_name, safe='')}?alt=media"
response: Final = self.sync_client.get(url=url, headers=headers)
if response.status_code == 200:
cached_response: Final = json.loads(response.text)
cached_response: Final = _DECODED_JSON.validate_python(json.loads(response.text))
verbose_logger.debug(
"Got GCS Cache: key: %s, cached_response %s. Type Response %s",
key,
@ -90,7 +94,7 @@ class GCSCache(BaseCache):
url = f"https://storage.googleapis.com/storage/v1/b/{bucket_name}/o/{quote(object_name, safe='')}?alt=media"
response: Final = await self.async_client.get(url=url, headers=headers)
if response.status_code == 200:
return json.loads(response.text)
return _DECODED_JSON.validate_python(json.loads(response.text))
return None
except Exception as e:
verbose_logger.error("GCS Caching: async_get_cache() - Got exception from GCS: %s", e)

11
litellm/caching/qdrant_semantic_cache.py
View file

@ -15,6 +15,8 @@ import os
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Protocol, cast
from pydantic import TypeAdapter
import litellm
from litellm._logging import print_verbose
from litellm.constants import (
@ -38,6 +40,7 @@ from ._embedding_router import (
from .base_cache import BaseCache
_WAIT_FOR_INDEXING: Final = MappingProxyType({"wait": "true"})
_EMBEDDING_VECTOR: Final = TypeAdapter(object)
if TYPE_CHECKING:
from litellm.router import Router
@ -295,7 +298,7 @@ class QdrantSemanticCache(BaseCache):
)
# get the embedding
embedding: Final = embedding_response["data"][0]["embedding"]
embedding: Final = _EMBEDDING_VECTOR.validate_python(embedding_response["data"][0]["embedding"])
value = str(value)
assert isinstance(value, str)
@ -334,7 +337,7 @@ class QdrantSemanticCache(BaseCache):
)
# get the embedding
embedding: Final = embedding_response["data"][0]["embedding"]
embedding: Final = _EMBEDDING_VECTOR.validate_python(embedding_response["data"][0]["embedding"])
data: Final = {
"vector": embedding,
@ -404,7 +407,7 @@ class QdrantSemanticCache(BaseCache):
embedding_response: Final = await self._get_async_embedding(prompt, metadata=kwargs.get("metadata"))
# get the embedding
embedding: Final = embedding_response["data"][0]["embedding"]
embedding: Final = _EMBEDDING_VECTOR.validate_python(embedding_response["data"][0]["embedding"])
value = str(value)
assert isinstance(value, str)
@ -440,7 +443,7 @@ class QdrantSemanticCache(BaseCache):
embedding_response: Final = await self._get_async_embedding(prompt, metadata=kwargs.get("metadata"))
# get the embedding
embedding: Final = embedding_response["data"][0]["embedding"]
embedding: Final = _EMBEDDING_VECTOR.validate_python(embedding_response["data"][0]["embedding"])
data: Final = {
"vector": embedding,

8
litellm/caching/redis_cluster_cache.py
View file

@ -9,7 +9,7 @@ import asyncio
import contextvars
import os
import weakref
from collections.abc import Awaitable, Callable
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Final, TypeAlias, TypeVar
@ -24,7 +24,7 @@ if TYPE_CHECKING:
pipeline = Pipeline
async_redis_client = Redis
Span = _Span | Any
Span = _Span
else:
pipeline = Any
async_redis_client = Any
@ -126,13 +126,13 @@ class RedisClusterCache(RedisCache):
return call()
return self._connection_gate.run(call, self._get_async_client_cache_key())
def _run_redis_mget_operation(self, keys: list[str]) -> list[Any]:
def _run_redis_mget_operation(self, keys: list[str]) -> Sequence[bytes | str | None]:
"""
Overrides `_run_redis_mget_operation` in redis_cache.py
"""
return self.redis_client.mget_nonatomic(keys=keys)
async def _async_run_redis_mget_operation(self, keys: list[str]) -> list[Any]:
async def _async_run_redis_mget_operation(self, keys: list[str]) -> Sequence[bytes | str | None]:
"""
Overrides `_async_run_redis_mget_operation` in redis_cache.py
"""

9
litellm/caching/redis_semantic_cache.py
View file

@ -117,7 +117,8 @@ class RedisSemanticCache(BaseCache):
password = password or os.environ["REDIS_PASSWORD"]
except KeyError as e:
# Raise a more informative exception if any of the required keys are missing
missing_var: Final[object] = e.args[0]
key_error_args: Final[tuple[object, ...]] = e.args
missing_var: Final = key_error_args[0]
raise ValueError(
f"Missing required Redis configuration: {missing_var}. Provide {missing_var} or redis_url."
) from e
@ -310,9 +311,9 @@ class RedisSemanticCache(BaseCache):
return
return
content = getattr(value, "content", None)
if content is not None:
cls._collect_responses_input_text(content, prompt_parts)
attribute_content: Final[object] = getattr(value, "content", None)
if attribute_content is not None:
cls._collect_responses_input_text(attribute_content, prompt_parts)
return
for text_key in ("text", "output", "input_text", "output_text"):

5
litellm/caching/valkey_semantic_cache.py
View file

@ -21,6 +21,7 @@ from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Any, Final
from pydantic import TypeAdapter
from redis import Redis
from redis.asyncio import Redis as AsyncRedis
from redis.commands.search.field import TagField, VectorField
@ -34,6 +35,8 @@ from litellm.llms.valkey.common_utils import build_valkey_url, pack_vector
from ._embedding_router import resolve_embedding_timeout
from .redis_semantic_cache import RedisSemanticCache
_OPAQUE_SEARCH_FIELD: Final = TypeAdapter(object)
@dataclass(frozen=True, slots=True)
class _ValkeyCacheHit:
@ -226,7 +229,7 @@ class ValkeySemanticCache(RedisSemanticCache):
if not docs:
return None
doc: Final = docs[0]
response_field: Final[object] = getattr(doc, cls.RESPONSE_FIELD_NAME)
response_field: Final = _OPAQUE_SEARCH_FIELD.validate_python(getattr(doc, cls.RESPONSE_FIELD_NAME))
distance_field: Final[str | bytes | float] = getattr(doc, cls.DISTANCE_FIELD_NAME)
return _ValkeyCacheHit(
response=str(response_field),

3
litellm/containers/endpoint_factory.py
View file

@ -36,6 +36,7 @@ RESPONSE_TYPES: Final[dict[str, type]] = {
"DeleteContainerFileResponse": DeleteContainerFileResponse,
}
_ENDPOINTS_DOCUMENT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
_ENDPOINT_RESULT: Final = TypeAdapter(object)
def _load_endpoints_config() -> dict:
@ -173,7 +174,7 @@ def create_async_endpoint_function(
init_response: Final = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
response = _ENDPOINT_RESULT.validate_python(await init_response)
else:
response = init_response

3
litellm/containers/utils.py
View file

@ -1,3 +1,4 @@
from collections.abc import Mapping
from typing import Any, Final, TypeVar
from litellm.llms.base_llm.containers.transformation import BaseContainerConfig
@ -99,7 +100,7 @@ class ContainerRequestUtils:
response_obj: T,
custom_llm_provider: str | None,
litellm_metadata: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_body: Mapping[str, object] | None = None,
) -> T:
"""
Encode container_id in response object with provider/model metadata for routing.

120
litellm/google_genai/main.py
View file

@ -1,11 +1,10 @@
import asyncio
import contextvars
from collections.abc import Iterator
from functools import partial
from typing import TYPE_CHECKING, Any, ClassVar, Final
import httpx
from pydantic import ConfigDict
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.constants import request_timeout
@ -39,6 +38,7 @@ else:
####### ENVIRONMENT VARIABLES ###################
# Initialize any necessary instances or variables here
base_llm_http_handler = BaseLLMHTTPHandler()
_GENERATE_CONTENT_RESULT: Final = TypeAdapter(object)
#################################################
@ -248,7 +248,7 @@ async def agenerate_content(
# LiteLLM specific params,
custom_llm_provider: str | None = None,
**kwargs,
) -> Any:
) -> object:
"""
Async: Generate content using Google GenAI
"""
@ -286,7 +286,7 @@ async def agenerate_content(
init_response: Final = await loop.run_in_executor(None, func_with_context)
if asyncio.iscoroutine(init_response):
response = await init_response
response = _GENERATE_CONTENT_RESULT.validate_python(await init_response)
else:
response = init_response
@ -316,7 +316,7 @@ def generate_content(
# LiteLLM specific params,
custom_llm_provider: str | None = None,
**kwargs,
) -> Any:
) -> object:
"""
Generate content using Google GenAI
"""
@ -362,22 +362,24 @@ def generate_content(
)
# Call the standard handler
response: Final = base_llm_http_handler.generate_content_handler(
model=setup_result.model,
contents=contents,
tools=tools,
generate_content_provider_config=setup_result.generate_content_provider_config,
generate_content_config_dict=setup_result.generate_content_config_dict,
custom_llm_provider=setup_result.custom_llm_provider,
litellm_params=setup_result.litellm_params,
logging_obj=setup_result.litellm_logging_obj,
extra_headers=extra_headers,
extra_body=_merge_native_request_fields(setup_result.native_request_fields, extra_body),
timeout=timeout or request_timeout,
_is_async=_is_async,
client=kwargs.get("client"),
litellm_metadata=kwargs.get("litellm_metadata", {}),
system_instruction=system_instruction,
response: Final = _GENERATE_CONTENT_RESULT.validate_python(
base_llm_http_handler.generate_content_handler(
model=setup_result.model,
contents=contents,
tools=tools,
generate_content_provider_config=setup_result.generate_content_provider_config,
generate_content_config_dict=setup_result.generate_content_config_dict,
custom_llm_provider=setup_result.custom_llm_provider,
litellm_params=setup_result.litellm_params,
logging_obj=setup_result.litellm_logging_obj,
extra_headers=extra_headers,
extra_body=_merge_native_request_fields(setup_result.native_request_fields, extra_body),
timeout=timeout or request_timeout,
_is_async=_is_async,
client=kwargs.get("client"),
litellm_metadata=kwargs.get("litellm_metadata", {}),
system_instruction=system_instruction,
)
)
return response
@ -406,7 +408,7 @@ async def agenerate_content_stream(
# LiteLLM specific params,
custom_llm_provider: str | None = None,
**kwargs,
) -> Any:
) -> object:
"""
Async: Generate content using Google GenAI with streaming response
"""
@ -457,23 +459,25 @@ async def agenerate_content_stream(
# Call the handler with async enabled and streaming
# Return the coroutine directly for the router to handle
return await base_llm_http_handler.generate_content_handler(
model=setup_result.model,
contents=contents,
generate_content_provider_config=setup_result.generate_content_provider_config,
generate_content_config_dict=setup_result.generate_content_config_dict,
tools=tools,
custom_llm_provider=setup_result.custom_llm_provider,
litellm_params=setup_result.litellm_params,
logging_obj=setup_result.litellm_logging_obj,
extra_headers=extra_headers,
extra_body=_merge_native_request_fields(setup_result.native_request_fields, extra_body),
timeout=timeout or request_timeout,
_is_async=True,
client=kwargs.get("client"),
stream=True,
litellm_metadata=kwargs.get("litellm_metadata", {}),
system_instruction=system_instruction,
return _GENERATE_CONTENT_RESULT.validate_python(
await base_llm_http_handler.generate_content_handler(
model=setup_result.model,
contents=contents,
generate_content_provider_config=setup_result.generate_content_provider_config,
generate_content_config_dict=setup_result.generate_content_config_dict,
tools=tools,
custom_llm_provider=setup_result.custom_llm_provider,
litellm_params=setup_result.litellm_params,
logging_obj=setup_result.litellm_logging_obj,
extra_headers=extra_headers,
extra_body=_merge_native_request_fields(setup_result.native_request_fields, extra_body),
timeout=timeout or request_timeout,
_is_async=True,
client=kwargs.get("client"),
stream=True,
litellm_metadata=kwargs.get("litellm_metadata", {}),
system_instruction=system_instruction,
)
)
except Exception as e:
@ -501,7 +505,7 @@ def generate_content_stream(
# LiteLLM specific params,
custom_llm_provider: str | None = None,
**kwargs,
) -> Iterator[Any]:
) -> object:
"""
Generate content using Google GenAI with streaming response
"""
@ -546,23 +550,25 @@ def generate_content_stream(
)
# Call the handler with streaming enabled (sync version)
return base_llm_http_handler.generate_content_handler(
model=setup_result.model,
contents=contents,
generate_content_provider_config=setup_result.generate_content_provider_config,
generate_content_config_dict=setup_result.generate_content_config_dict,
tools=tools,
custom_llm_provider=setup_result.custom_llm_provider,
litellm_params=setup_result.litellm_params,
logging_obj=setup_result.litellm_logging_obj,
extra_headers=extra_headers,
extra_body=_merge_native_request_fields(setup_result.native_request_fields, extra_body),
timeout=timeout or request_timeout,
_is_async=_is_async,
client=kwargs.get("client"),
stream=True,
litellm_metadata=kwargs.get("litellm_metadata", {}),
system_instruction=system_instruction,
return _GENERATE_CONTENT_RESULT.validate_python(
base_llm_http_handler.generate_content_handler(
model=setup_result.model,
contents=contents,
generate_content_provider_config=setup_result.generate_content_provider_config,
generate_content_config_dict=setup_result.generate_content_config_dict,
tools=tools,
custom_llm_provider=setup_result.custom_llm_provider,
litellm_params=setup_result.litellm_params,
logging_obj=setup_result.litellm_logging_obj,
extra_headers=extra_headers,
extra_body=_merge_native_request_fields(setup_result.native_request_fields, extra_body),
timeout=timeout or request_timeout,
_is_async=_is_async,
client=kwargs.get("client"),
stream=True,
litellm_metadata=kwargs.get("litellm_metadata", {}),
system_instruction=system_instruction,
)
)
except Exception as e:

6
litellm/google_genai/streaming_iterator.py
View file

@ -67,7 +67,7 @@ class BaseGoogleGenAIGenerateContentStreamingIterator:
request_body: dict,
model: str,
custom_llm_provider: str,
hidden_params: dict[str, Any] | None = None,
hidden_params: dict[str, object] | None = None,
):
self.litellm_logging_obj = litellm_logging_obj
self.request_body = request_body
@ -118,7 +118,7 @@ class GoogleGenAIGenerateContentStreamingIterator(BaseGoogleGenAIGenerateContent
litellm_metadata: dict,
custom_llm_provider: str,
request_body: dict | None = None,
hidden_params: dict[str, Any] | None = None,
hidden_params: dict[str, object] | None = None,
):
super().__init__(
litellm_logging_obj=logging_obj,
@ -169,7 +169,7 @@ class AsyncGoogleGenAIGenerateContentStreamingIterator(BaseGoogleGenAIGenerateCo
litellm_metadata: dict,
custom_llm_provider: str,
request_body: dict | None = None,
hidden_params: dict[str, Any] | None = None,
hidden_params: dict[str, object] | None = None,
):
super().__init__(
litellm_logging_obj=logging_obj,

15
litellm/harness/handlers/deepagents_handler.py
View file

@ -13,8 +13,8 @@ import asyncio
import importlib
from collections.abc import AsyncIterator, Callable, Mapping
from dataclasses import dataclass
from types import MappingProxyType, ModuleType
from typing import TYPE_CHECKING, Any
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Protocol
from litellm.harness.context import SessionContext
from litellm.harness.errors import HarnessError, HarnessInstallFailed
@ -51,10 +51,17 @@ if TYPE_CHECKING:
from langgraph.types import Command
from litellm.llms.base_llm.harness.transformation import BaseHarnessConfig
from litellm.llms.deepagents.harness.sandbox_backend import SandboxBackend, ToolFilterMiddleware, UsageCallback
_MODEL_NODE = "model"
class _SandboxBackendModule(Protocol):
SandboxBackend: type[SandboxBackend]
UsageCallback: type[UsageCallback]
ToolFilterMiddleware: type[ToolFilterMiddleware]
@dataclass(frozen=True)
class DeepAgentsDeps:
"""The optional-dependency entrypoints this handler uses."""
@ -65,7 +72,7 @@ class DeepAgentsDeps:
command_cls: Callable[..., Command]
subagent_defaults: Mapping[str, object]
convert_to_openai_messages: Any
backend: ModuleType
backend: _SandboxBackendModule
def load_deps() -> DeepAgentsDeps:
@ -77,7 +84,7 @@ def load_deps() -> DeepAgentsDeps:
memory = importlib.import_module("langgraph.checkpoint.memory")
lg_types = importlib.import_module("langgraph.types")
messages = importlib.import_module("langchain_core.messages")
backend = importlib.import_module("litellm.llms.deepagents.harness.sandbox_backend")
from litellm.llms.deepagents.harness import sandbox_backend as backend
except ImportError as e:
raise HarnessInstallFailed(f"{INSTALL_HINT} ({e})") from e
return DeepAgentsDeps(

7
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -75,6 +75,7 @@ OPENAI_API_BASE_ENV_VARS: Final = ("OPENAI_BASE_URL", "OPENAI_API_BASE")
_OBJECT_MAPPING_ADAPTER: Final = TypeAdapter(dict[object, object])
_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
_CONTENT_PART_ADAPTER: Final = TypeAdapter(dict[str, object])
_ATTRIBUTE_VALUE: Final = TypeAdapter(object)
AllToolParamValues = ChatCompletionToolParam | AllAnthropicToolsValues
@ -216,7 +217,7 @@ def _carries_cache_breakpoint(block: object) -> bool:
def _attribute_or_key(value: object, key: str) -> object | None:
if hasattr(value, key):
return getattr(value, key)
return _ATTRIBUTE_VALUE.validate_python(getattr(value, key))
if isinstance(value, Mapping):
return value.get(key)
return None
@ -983,7 +984,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
@staticmethod
def maybe_seed_default_injection_points(
non_default_params: dict[str, Any],
non_default_params: dict[str, object],
messages: list[AllMessageValues],
model: str,
custom_llm_provider: str | None,
@ -1109,7 +1110,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
def maybe_inject_cache_control(
messages: list[dict],
system: str | list | None,
kwargs: dict[str, Any],
kwargs: dict[str, object],
model: str | None = None,
custom_llm_provider: str | None = None,
tools: list[dict] | None = None,

10
litellm/integrations/arize/arize_phoenix.py
View file

@ -1,6 +1,7 @@
import os
import threading
from collections import OrderedDict
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from litellm._logging import verbose_logger
@ -22,7 +23,7 @@ if TYPE_CHECKING:
Protocol = _Protocol
OpenTelemetryConfig = _OpenTelemetryConfig
Span = _Span | Any
Span = _Span
OpenTelemetry = _OpenTelemetry
LITELLM_TRACER_NAME: str
else:
@ -100,7 +101,7 @@ class ArizePhoenixLogger(OpenTelemetry):
if getattr(self, "_use_injected_tracer_provider", False):
return
shared_processor: Final = getattr(self, "_shared_span_processor", None)
shared_processor: Final[SpanProcessor | None] = getattr(self, "_shared_span_processor", None)
if shared_processor is not None:
try:
shared_processor.force_flush()
@ -111,7 +112,8 @@ class ArizePhoenixLogger(OpenTelemetry):
)
with getattr(self, "_project_providers_lock", threading.Lock()):
providers: Final = list(getattr(self, "_project_providers", {}).values())
project_providers: Final[Mapping[str, TracerProvider]] = getattr(self, "_project_providers", {})
providers: Final = list(project_providers.values())
for provider in providers:
try:
@ -193,7 +195,7 @@ class ArizePhoenixLogger(OpenTelemetry):
primary ``otel`` callback which handles proxy-level parent spans.
"""
def set_attributes(self, span: Span, kwargs, response_obj: Any | None):
def set_attributes(self, span: Span, kwargs, response_obj: object):
ArizePhoenixLogger.set_arize_phoenix_attributes(span, kwargs, response_obj)
@staticmethod

13
litellm/integrations/arize/arize_phoenix_client.py
View file

@ -68,7 +68,7 @@ class ArizePhoenixClient:
try:
# Use the underlying httpx client directly to avoid query param extraction
response = self.http_handler.get(url, headers=self.headers)
response: Final = self.http_handler.get(url, headers=self.headers)
response.raise_for_status()
data: Final = response.json()
@ -76,15 +76,16 @@ class ArizePhoenixClient:
except Exception as e:
# Check if it's an HTTP error
response = getattr(e, "response", None)
if response is not None and hasattr(response, "status_code"):
if response.status_code == 404:
error_response: Final[object] = getattr(e, "response", None)
if error_response is not None and hasattr(error_response, "status_code"):
status_code: Final[object] = getattr(error_response, "status_code", None)
if status_code == 404:
return None
elif response.status_code == 403:
elif status_code == 403:
raise Exception(
f"Access denied to prompt version '{prompt_version_id}'. Check your Arize Phoenix permissions."
)
elif response.status_code == 401:
elif status_code == 401:
raise Exception("Authentication failed. Check your Arize Phoenix API key and permissions.")
else:
raise Exception(f"Failed to fetch prompt version '{prompt_version_id}': {e}")

6
litellm/integrations/bitbucket/bitbucket_client.py
View file

@ -47,6 +47,10 @@ def _sanitize_file_path(file_path: str) -> str:
return "/".join(urllib.parse.quote(part, safe="") for part in parts)
def _content_type(headers: Mapping[str, str]) -> str:
return headers.get("content-type", "")
class BitBucketClient:
"""
Client for interacting with BitBucket API to fetch .prompt files.
@ -119,7 +123,7 @@ class BitBucketClient:
response.raise_for_status()
# BitBucket returns file content as base64 encoded
if response.headers.get("content-type", "").startswith("text/"):
if _content_type(response.headers).startswith("text/"):
return response.text
else:
# For binary files or when content-type is not text, try to decode as base64

20
litellm/integrations/custom_guardrail.py
View file

@ -145,6 +145,10 @@ def _unified_hook_fields(guardrail: "CustomGuardrail", request_data: Mapping[str
}
def _instance_attributes(value: object) -> Mapping[str, object]:
return vars(value)
def _strict_guardrail_modes_enabled() -> bool:
"""Whether guardrail-mode validation raises (default) or logs a warning.
@ -396,7 +400,7 @@ class CustomGuardrail(CustomLogger):
def inject_advisory_message(
self,
data: dict[str, Any], # mutable-ok: caller's dict is mutated in place, matching mark_pre_call_hook_ran
data: dict[str, object], # mutable-ok: caller's dict is mutated in place, matching mark_pre_call_hook_ran
message: str,
) -> bool:
"""
@ -473,8 +477,8 @@ class CustomGuardrail(CustomLogger):
def raise_sensitive_data_route_exception(
self,
route_to_model: str,
request_data: dict[str, Any],
detection_info: dict[str, Any] | None = None,
request_data: dict[str, object],
detection_info: dict[str, object] | None = None,
) -> None:
"""
Raise an exception to reroute the request to a different model.
@ -620,8 +624,8 @@ class CustomGuardrail(CustomLogger):
def handle_sensitive_data_detection(
self,
request_data: dict[str, Any],
detection_info: dict[str, Any] | None = None,
request_data: dict[str, object],
detection_info: dict[str, object] | None = None,
) -> None:
"""
Handle sensitive data detection based on guardrail configuration.
@ -888,7 +892,9 @@ class CustomGuardrail(CustomLogger):
) from e
return unified_guardrail
async def async_pre_call_deployment_hook(self, kwargs: dict[str, Any], call_type: CallTypes | None) -> dict | None:
async def async_pre_call_deployment_hook(
self, kwargs: dict[str, object], call_type: CallTypes | None
) -> dict | None:
# should run guardrail
litellm_guardrails: Final = kwargs.get("guardrails")
if litellm_guardrails is None or not isinstance(litellm_guardrails, list):
@ -1579,7 +1585,7 @@ class CustomGuardrail(CustomLogger):
"""
Update the guardrails litellm params in memory
"""
for key, value in vars(litellm_params).items():
for key, value in _instance_attributes(litellm_params).items():
setattr(self, key, value)
def get_guardrails_messages_for_call_type(

2
litellm/integrations/custom_logger.py
View file

@ -256,7 +256,7 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
self,
model: str,
request_kwargs: dict,
messages: list[dict[str, Any]] | None = None,
messages: list[dict[str, object]] | None = None,
input: str | list | None = None,
specific_deployment: bool | None = False,
) -> PreRoutingHookResponse | None:

10
litellm/integrations/datadog/datadog_cost_management.py
View file

@ -3,7 +3,7 @@ import os
import time
from collections.abc import Mapping
from datetime import datetime
from typing import Any, Final, cast
from typing import Final
from litellm._logging import verbose_logger
from litellm.integrations.custom_batch_logger import CustomBatchLogger
@ -45,6 +45,10 @@ _RESERVED_TAG_KEYS: Final[frozenset] = frozenset(
)
def _as_mapping(metadata: Mapping[str, object]) -> Mapping[str, object]:
return metadata
class DatadogCostManagementLogger(CustomBatchLogger):
def __init__(self, cost_tag_keys: list[str] | None = None, **kwargs):
self.cost_tag_keys: list[str] = list(cost_tag_keys) if cost_tag_keys else []
@ -180,9 +184,7 @@ class DatadogCostManagementLogger(CustomBatchLogger):
self._add_tag(tags, "model", log.get("model"))
self._add_tag(tags, "model_id", log.get("model_id"))
# cast because StandardLoggingMetadata is a TypedDict; we iterate it
# as a generic mapping below.
metadata: Final[Mapping[str, object]] = cast(dict[str, Any], log.get("metadata") or {})
metadata: Final = _as_mapping(log.get("metadata") or {})
# Backwards-compat: team/user/model_group preserved regardless of allowlist.
if metadata.get("user_api_key_alias"):

5
litellm/integrations/datadog/datadog_llm_obs.py
View file

@ -62,6 +62,7 @@ _EMPTY_MAPPING: Final[Mapping[str, object]] = MappingProxyType({})
_EMPTY_MESSAGE: Final[Message] = {"role": "", "content": ""}
_MAX_PARSED_TOOL_ARGUMENT_CHARS: Final = 256 * 1024
_JSON_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
_DECODED_RESPONSE: Final = TypeAdapter(object)
_SAFE_REDACTED_MESSAGE_ROLES: Final = frozenset(
{"agent", "assistant", "developer", "function", "model", "system", "tool", "user"}
)
@ -815,11 +816,11 @@ class DataDogLLMObsLogger(CustomBatchLogger):
try:
import ast
response_obj = ast.literal_eval(response_obj)
response_obj = _DECODED_RESPONSE.validate_python(ast.literal_eval(response_obj))
except (ValueError, SyntaxError):
try:
# fallback to json parsing
response_obj = json.loads(str(response_obj))
response_obj = _DECODED_RESPONSE.validate_python(json.loads(str(response_obj)))
except json.JSONDecodeError:
return ()

7
litellm/integrations/gcs_bucket/gcs_bucket_base.py
View file

@ -2,6 +2,8 @@ import json
import os
from typing import TYPE_CHECKING, Any, Final
from pydantic import TypeAdapter
from litellm._logging import verbose_logger
from litellm.integrations.custom_batch_logger import CustomBatchLogger
from litellm.integrations.gcs_bucket.gcs_bucket_mock_client import (
@ -25,6 +27,7 @@ if TYPE_CHECKING:
else:
VertexBase = Any
IAM_AUTH_KEY: Final = "IAM_AUTH"
_UPLOAD_RESPONSE: Final = TypeAdapter(object)
class GCSBucketBase(CustomBatchLogger):
@ -297,7 +300,7 @@ class GCSBucketBase(CustomBatchLogger):
bucket_name: str,
object_name: str,
logging_payload: StandardLoggingPayload | str,
):
) -> object:
"""
Helper function to make POST request to GCS Bucket in the specified bucket.
"""
@ -325,4 +328,4 @@ class GCSBucketBase(CustomBatchLogger):
verbose_logger.debug("GCS Bucket status code %s", response.status_code)
verbose_logger.debug("GCS Bucket response.text %s", response.text)
return response.json()
return _UPLOAD_RESPONSE.validate_python(response.json())

2
litellm/integrations/generic_api/generic_api_callback.py
View file

@ -252,7 +252,7 @@ class GenericAPILogger(CustomBatchLogger):
if self.retry_delay <= 0:
return
delay: Final = self.retry_delay * (2**attempt)
delay: Final = self.retry_delay * (2.0**attempt)
await asyncio.sleep(delay)
async def _post_with_retries(self, data: str) -> httpx.Response:

4
litellm/integrations/gitlab/gitlab_prompt_manager.py
View file

@ -297,7 +297,7 @@ class GitLabPromptManager(CustomPromptManagement):
def __init__(
self,
gitlab_config: dict[str, Any],
gitlab_config: dict[str, object],
prompt_id: str | None = None,
ref: str | None = None, # tag/branch/SHA override
gitlab_client: GitLabClient | None = None,
@ -662,7 +662,7 @@ class GitLabPromptCache:
def __init__(
self,
gitlab_config: dict[str, Any],
gitlab_config: dict[str, object],
*,
ref: str | None = None,
gitlab_client: GitLabClient | None = None,

5
litellm/integrations/lago.py
View file

@ -147,9 +147,10 @@ class LagoLogger(CustomLogger):
response.raise_for_status()
except Exception as e:
error_response: Final = getattr(e, "response", None)
error_response: Final[object] = getattr(e, "response", None)
if error_response is not None and hasattr(error_response, "text"):
verbose_logger.debug("\nError Message: %s", error_response.text)
error_text: Final[object] = getattr(error_response, "text", None)
verbose_logger.debug("\nError Message: %s", error_text)
raise e
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):

14
litellm/integrations/langfuse/langfuse_otel.py
View file

@ -1,6 +1,7 @@
import base64
import json
import os
from collections.abc import Sequence
from datetime import datetime
from typing import TYPE_CHECKING, Any, Final, Optional
@ -19,7 +20,7 @@ from litellm.types.utils import StandardCallbackDynamicParams
if TYPE_CHECKING:
from opentelemetry.trace import Span as _Span
Span = _Span | Any
Span = _Span
else:
Span = Any
@ -190,18 +191,19 @@ class LangfuseOtelLogger(OpenTelemetry):
}
)
elif item_type == "message":
item_content: Sequence[object] = getattr(item, "content", [{}])
output_items_data.append(
{
"role": getattr(item, "role", "assistant"),
"content": getattr(getattr(item, "content", [{}])[0], "text", ""),
"content": getattr(item_content[0], "text", ""),
}
)
elif item_type == "function_call":
arguments_str = getattr(item, "arguments", "{}")
item_arguments: object = getattr(item, "arguments", "{}")
arguments_obj = (
safe_json_loads(arguments_str, default={})
if isinstance(arguments_str, str)
else arguments_str
safe_json_loads(item_arguments, default={})
if isinstance(item_arguments, str)
else item_arguments
)
langfuse_tool_call = {
"id": getattr(item, "id", ""),

8
litellm/integrations/langsmith.py
View file

@ -10,7 +10,7 @@ from datetime import datetime, timezone
from typing import Any, Final
import httpx
from pydantic import BaseModel
from pydantic import BaseModel, TypeAdapter
import litellm
from litellm._logging import verbose_logger
@ -28,6 +28,8 @@ from litellm.llms.custom_httpx.http_handler import (
from litellm.types.integrations.langsmith import *
from litellm.types.utils import StandardCallbackDynamicParams, StandardLoggingPayload
_RUN_RESPONSE: Final = TypeAdapter(object)
def is_serializable(value):
non_serializable_types: Final = (
@ -524,7 +526,7 @@ class LangsmithLogger(CustomBatchLogger):
# If we can't get an event loop, create a new one
asyncio.run(self.async_send_batch())
def get_run_by_id(self, run_id):
def get_run_by_id(self, run_id) -> object:
langsmith_api_key: Final = self.default_credentials["LANGSMITH_API_KEY"]
langsmith_api_base: Final = self.default_credentials["LANGSMITH_BASE_URL"]
langsmith_tenant_id: Final = self.default_credentials.get("LANGSMITH_TENANT_ID")
@ -538,7 +540,7 @@ class LangsmithLogger(CustomBatchLogger):
headers=headers,
)
return response.json()
return _RUN_RESPONSE.validate_python(response.json())
def make_dot_order(self, run_id: str):
st: Final = datetime.now(timezone.utc)

4
litellm/integrations/literal_ai.py
View file

@ -2,6 +2,7 @@
# This file contains the LiteralAILogger class which is used to log steps to the LiteralAI observability platform.
import asyncio
import os
from collections.abc import Mapping
from typing import Final
import httpx
@ -181,7 +182,8 @@ class LiteralAILogger(CustomBatchLogger):
if messages and isinstance(messages, list) and isinstance(messages[0], dict):
for message in messages:
if literal_prompt := getattr(message, "__literal_prompt__", None):
literal_prompt: Mapping[str, object] | None = getattr(message, "__literal_prompt__", None)
if literal_prompt:
prompt_id = literal_prompt.get("prompt_id")
variables = literal_prompt.get("variables")
message["uuid"] = literal_prompt.get("uuid")

11
litellm/integrations/lunary.py
View file

@ -3,10 +3,15 @@
import importlib
import traceback
from datetime import datetime, timezone
from typing import Final
from typing import TYPE_CHECKING, Final
import packaging
if TYPE_CHECKING:
from openai.types.chat.chat_completion_message_custom_tool_call import Custom
from litellm.types.utils import ChatCompletionCustomToolCallPayload, ChatCompletionDeltaCustomToolCallPayload
# convert to {completion: xx, tokens: xx}
def parse_usage(usage):
@ -21,7 +26,9 @@ def parse_tool_calls(tool_calls):
return None
def clean_tool_call(tool_call):
custom: Final = getattr(tool_call, "custom", None)
custom: Final[
ChatCompletionCustomToolCallPayload | ChatCompletionDeltaCustomToolCallPayload | Custom | None
] = getattr(tool_call, "custom", None)
if custom is not None:
name, arguments = custom.name, custom.input
else:

10
litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py
View file

@ -24,6 +24,8 @@ from collections.abc import Callable
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any, Final
from typing_extensions import TypedDict, Unpack
import litellm
from litellm._logging import verbose_proxy_logger
from litellm.constants import MAVVRIK_FOCUS_EXPORT_JOB_NAME
@ -82,10 +84,16 @@ def _is_empty_metrics_marker(marker: object | None) -> bool:
return False
class _CustomLoggerOptions(TypedDict, total=False, extra_items=object):
pass
class MavvrikFocusLogger(FocusLogger):
"""FOCUS-based export logger that routes to the Mavvrik destination."""
def __init__(self, *, clock: Callable[[], datetime] = get_utc_datetime, **kwargs: Any) -> None:
def __init__(
self, *, clock: Callable[[], datetime] = get_utc_datetime, **kwargs: Unpack[_CustomLoggerOptions]
) -> None:
self._clock: Final = clock
frequency: Final = os.getenv("MAVVRIK_FOCUS_FREQUENCY", "daily").lower()
if frequency != "daily":

3
litellm/integrations/opentelemetry_utils/gen_ai_semconv.py
View file

@ -40,6 +40,7 @@ from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
if TYPE_CHECKING:
from opentelemetry._logs import SeverityNumber as _SeverityNumber
from opentelemetry.trace import Span as _Span
from opentelemetry.trace import SpanContext
from litellm.integrations.opentelemetry import OpenTelemetryConfig
@ -231,7 +232,7 @@ class OTELGenAISemconvMixin:
response_obj: dict,
provider: str | None,
otel_logger,
parent_ctx,
parent_ctx: "SpanContext",
) -> None:
"""Emit the consolidated ``gen_ai.client.inference.operation.details``
log event, correlated to the request span via ``parent_ctx``.

6
litellm/integrations/otel/logger.py
View file

@ -976,12 +976,16 @@ def _excluded_db_systems(logger: "OpenTelemetryV2") -> frozenset[str]:
callback folds into the preset, whose config is env-only.
"""
otel_settings: Final = (litellm.callback_settings or {}).get("otel")
configured: Final = otel_settings.get("excluded_services") if isinstance(otel_settings, dict) else None
configured: Final = _excluded_services_setting(otel_settings) if isinstance(otel_settings, dict) else None
if configured is None:
return logger.config.excluded_services
return excluded_db_systems_from(configured)
def _excluded_services_setting(otel_settings: Mapping[str, object]) -> object:
return otel_settings.get("excluded_services")
def _v2_configs(in_memory_loggers: Sequence[object], logger: "OpenTelemetryV2") -> tuple[OpenTelemetryV2Config, ...]:
"""Every v2 logger's config, the published logger's first.

5
litellm/integrations/otel/model/metadata.py
View file

@ -42,6 +42,8 @@ from datetime import datetime
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, cast, get_args
from pydantic import TypeAdapter
from litellm.constants import (
INTERNAL_CALL_ORIGIN_METADATA_KEY,
LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL,
@ -56,6 +58,7 @@ if TYPE_CHECKING:
from litellm.types.utils import StandardLoggingPayload
_INTERNAL_CALL_ORIGINS: Final[frozenset[str]] = frozenset(get_args(InternalCallOrigin))
_AUTH_ATTRIBUTE: Final = TypeAdapter(object)
REQUESTER_METADATA_KEY: Final = "requester_metadata"
REQUESTER_METADATA_PATH: Final = f"{REQUESTER_METADATA_KEY}."
@ -129,7 +132,7 @@ class RequestIdentity:
snapshot) is flattened to dotted keys so ``requester_metadata.<key>``
resolves too.
"""
get: Final[Callable[[str], object]] = lambda name: getattr(auth, name, None) # noqa: E731
get: Final[Callable[[str], object]] = lambda name: _AUTH_ATTRIBUTE.validate_python(getattr(auth, name, None)) # noqa: E731
auth_meta: Final = tuple(
(meta_key, str(value))
for meta_key, attr in (

2
litellm/integrations/prometheus.py
View file

@ -3098,7 +3098,7 @@ class PrometheusLogger(CustomLogger):
if exception is not None:
exception_status = str(getattr(exception, "status_code", None))
if exception_status == "None" or not exception_status:
code: Final = getattr(exception, "code", None)
code: Final[object] = getattr(exception, "code", None)
if code is not None:
exception_status = str(code)

60
litellm/integrations/shadow_eval_logger.py
View file

@ -20,9 +20,10 @@ from datetime import datetime, timezone
from itertools import groupby
from operator import itemgetter
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Literal
from typing import TYPE_CHECKING, Final, Literal, Protocol
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError, field_validator, model_validator
from typing_extensions import ReadOnly, TypedDict
from litellm._logging import verbose_logger
from litellm.caching.in_memory_cache import InMemoryCache
@ -801,6 +802,52 @@ def _scope_admits(router: "Router | None", job: "ActiveShadowEvalJob", model_gro
return not job.models or any(_canonical_group(router, name) == model_group for name in job.models)
class _ShadowEvalJobRow(Protocol):
@property
def id(self) -> str: ...
@property
def target_type(self) -> str: ...
@property
def target_id(self) -> str: ...
class _AttemptCount(TypedDict):
_all: ReadOnly[int]
class _AttemptGroup(TypedDict):
job_id: ReadOnly[str]
_count: ReadOnly[_AttemptCount]
_sum: ReadOnly[Mapping[str, object] | None]
class _ShadowEvalJobTable(Protocol):
async def find_many(self, *, where: Mapping[str, object]) -> Sequence[_ShadowEvalJobRow]: ...
class _ShadowEvalAttemptTable(Protocol):
async def group_by(
self,
*,
by: Sequence[str],
count: bool,
sum: Mapping[str, bool],
where: Mapping[str, object],
) -> Sequence[_AttemptGroup]: ...
async def create(self, *, data: Mapping[str, object]) -> object: ...
def _shadow_eval_job_table(prisma: "PrismaClient") -> _ShadowEvalJobTable:
return prisma.db.litellm_shadowevaljob
def _shadow_eval_attempt_table(prisma: "PrismaClient") -> _ShadowEvalAttemptTable:
return prisma.db.litellm_shadowevalattempt
def _as_active_job(record: object, attempts: int, spend: float) -> ActiveShadowEvalJob | None:
"""The sampling path's view of one job row, or None for a row it cannot sample: an
unknown direction, or a reverse job with no baseline model to duplicate against.
@ -860,14 +907,14 @@ class ShadowEvalLogger(CustomLogger):
if prisma is None:
return _EMPTY_JOBS
try:
records: Final = await prisma.db.litellm_shadowevaljob.find_many(
records: Final = await _shadow_eval_job_table(prisma).find_many(
where={
"stopped_at": None,
"ends_at": {"gt": datetime.now(timezone.utc)},
},
)
grouped: Final = (
await prisma.db.litellm_shadowevalattempt.group_by(
await _shadow_eval_attempt_table(prisma).group_by(
by=["job_id"],
count=True,
sum={"judge_cost": True, "shadow_cost": True, "shadow_classifier_cost": True},
@ -894,7 +941,10 @@ class ShadowEvalLogger(CustomLogger):
)
)
jobs: Final = MappingProxyType(
{target: tuple(job for _, job in group) for target, group in groupby(by_target, key=itemgetter(0))}
{
target: tuple(job for _, job in group)
for target, group in groupby(by_target, key=lambda entry: entry[0])
}
)
await self._jobs_cache.async_set_cache(_JOBS_CACHE_KEY, jobs)
self._job_starts = {}
@ -1242,7 +1292,7 @@ class ShadowEvalLogger(CustomLogger):
if prisma is None:
return
try:
await prisma.db.litellm_shadowevalattempt.create(
await _shadow_eval_attempt_table(prisma).create(
data={
"job_id": job.id,
"request_id": request_id,

8
litellm/integrations/vantage/vantage_logger.py
View file

@ -9,6 +9,8 @@ from __future__ import annotations
import os
from typing import TYPE_CHECKING, Any, Final, cast
from typing_extensions import TypedDict, Unpack
import litellm
from litellm._logging import verbose_logger
from litellm.integrations.custom_logger import CustomLogger
@ -24,6 +26,10 @@ else:
VANTAGE_USAGE_DATA_JOB_NAME: Final = "vantage_export_usage_data"
class _CustomLoggerOptions(TypedDict, total=False, extra_items=object):
pass
class VantageLogger(FocusLogger):
"""FocusLogger pre-configured for Vantage (CSV format, Vantage API destination).
@ -43,7 +49,7 @@ class VantageLogger(FocusLogger):
base_url: str | None = None,
frequency: str | None = None,
interval_seconds: int | None = None,
**kwargs: Any,
**kwargs: Unpack[_CustomLoggerOptions],
) -> None:
resolved_api_key: Final = api_key or os.getenv("VANTAGE_API_KEY")
resolved_token: Final = integration_token or os.getenv("VANTAGE_INTEGRATION_TOKEN")

21
litellm/integrations/websearch_interception/transformation.py
View file

@ -6,8 +6,9 @@ Transforms between Anthropic/OpenAI tool_use format and LiteLLM search format.
import json
from collections.abc import Sequence
from typing import Any, Final
from typing import Final
from pydantic import TypeAdapter
from typing_extensions import assert_never
from litellm._logging import verbose_logger
@ -21,6 +22,8 @@ from litellm.types.integrations.websearch_interception import (
WebSearchToolResultErrorCode,
)
_DECODED_JSON: Final = TypeAdapter(object)
class WebSearchTransformation:
"""
@ -138,7 +141,7 @@ class WebSearchTransformation:
@staticmethod
def _detect_from_non_streaming_response(
response: Any,
response: object,
) -> tuple[bool, list[dict]]:
"""Parse non-streaming response for WebSearch tool_use"""
@ -149,7 +152,7 @@ class WebSearchTransformation:
if not hasattr(response, "content"):
verbose_logger.debug("WebSearchInterception: Response has no content attribute")
return False, []
content = response.content or []
content = getattr(response, "content", None) or []
if not content:
verbose_logger.debug("WebSearchInterception: Response has empty content")
@ -194,7 +197,7 @@ class WebSearchTransformation:
@staticmethod
def _detect_from_openai_response(
response: Any,
response: object,
) -> tuple[bool, list[dict]]:
"""Parse OpenAI-style response for WebSearch tool_calls"""
@ -205,7 +208,7 @@ class WebSearchTransformation:
if not hasattr(response, "choices"):
verbose_logger.debug("WebSearchInterception: Response has no choices attribute")
return False, []
choices = response.choices or []
choices = getattr(response, "choices", None) or []
if not choices:
verbose_logger.debug("WebSearchInterception: Response has empty choices")
@ -247,9 +250,9 @@ class WebSearchTransformation:
else:
tool_id = getattr(tool_call, "id", None)
tool_type = getattr(tool_call, "type", None)
function = getattr(tool_call, "function", None)
function_name = getattr(function, "name", None) if function else None
function_arguments = getattr(function, "arguments", None) if function else None
function_obj: object = getattr(tool_call, "function", None)
function_name = getattr(function_obj, "name", None) if function_obj else None
function_arguments = getattr(function_obj, "arguments", None) if function_obj else None
# Detect function-style web search tool_calls. ``WebSearch`` is
# intentionally omitted — see is_web_search_tool for the Cowork
@ -262,7 +265,7 @@ class WebSearchTransformation:
# Parse arguments (might be JSON string)
if isinstance(function_arguments, str):
try:
arguments = json.loads(function_arguments)
arguments = _DECODED_JSON.validate_python(json.loads(function_arguments))
except json.JSONDecodeError:
verbose_logger.warning(
"WebSearchInterception: Failed to parse function arguments: %s", function_arguments

20
litellm/interactions/agents/main.py
View file

@ -101,7 +101,7 @@ async def acreate(
instructions: str | None = None,
base_environment: InteractionEnvironment | None = None,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
@ -145,7 +145,7 @@ def create(
instructions: str | None = None,
base_environment: InteractionEnvironment | None = None,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
@ -207,7 +207,7 @@ def create(
@client
async def alist(
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentListResponse:
@ -241,7 +241,7 @@ async def alist(
@client
def list(
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentListResponse | Coroutine[object, object, AgentListResponse]:
@ -281,7 +281,7 @@ def list(
async def aget(
name: str,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentCreateResponse:
@ -317,7 +317,7 @@ async def aget(
def get(
name: str,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
@ -358,7 +358,7 @@ def get(
async def adelete(
name: str,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentDeleteResult:
@ -394,7 +394,7 @@ async def adelete(
def delete(
name: str,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentDeleteResult | Coroutine[object, object, AgentDeleteResult]:
@ -435,7 +435,7 @@ def delete(
async def alist_versions(
name: str,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentVersionsResponse:
@ -471,7 +471,7 @@ async def alist_versions(
def list_versions(
name: str,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentVersionsResponse | Coroutine[object, object, AgentVersionsResponse]:

13
litellm/interactions/http_handler.py
View file

@ -4,8 +4,8 @@ HTTP Handler for Interactions API requests.
This module handles the HTTP communication for the Google Interactions API.
"""
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping
from typing import Any, Final
from collections.abc import AsyncIterator, Callable, Coroutine, Iterator, Mapping
from typing import Final, Protocol
import httpx
@ -34,6 +34,11 @@ from litellm.types.interactions import (
from litellm.types.router import GenericLiteLLMParams
class _ErrorClassSource(Protocol):
@property
def get_error_class(self) -> Callable[..., Exception]: ...
class _BaseHTTPHandler:
"""
Shared HTTP infrastructure for LiteLLM handler classes.
@ -43,7 +48,7 @@ class _BaseHTTPHandler:
not duplicate this boilerplate.
"""
def _handle_error(self, e: Exception, provider_config: Any) -> Exception:
def _handle_error(self, e: Exception, provider_config: _ErrorClassSource) -> Exception:
if isinstance(e, httpx.HTTPStatusError):
return provider_config.get_error_class(
error_message=e.response.text,
@ -77,7 +82,7 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
HTTP handler for Interactions API requests.
"""
# _handle_error is inherited from _BaseHTTPHandler (accepts Any provider_config).
# _handle_error is inherited from _BaseHTTPHandler (accepts any config exposing get_error_class).
# AgentsHTTPHandler also extends this class and passes BaseAgentsAPIConfig, which
# is structurally compatible but a different type — keeping the override here with
# BaseInteractionsAPIConfig would cause type errors in the subclass.

32
litellm/interactions/main.py
View file

@ -33,9 +33,9 @@ Usage:
import asyncio
import contextvars
from collections.abc import AsyncIterator, Coroutine, Iterator
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping
from functools import partial
from typing import Any, Final
from typing import Final
import httpx
@ -79,7 +79,7 @@ async def acreate(
# System instruction
system_instruction: str | None = None,
# Generation config
generation_config: dict[str, Any] | None = None,
generation_config: Mapping[str, object] | None = None,
# Streaming
stream: bool | None = None,
# Storage
@ -90,13 +90,13 @@ async def acreate(
environment: InteractionEnvironment | None = None,
# Response format
response_modalities: list[str] | None = None,
response_format: dict[str, Any] | None = None,
response_format: Mapping[str, object] | None = None,
response_mime_type: str | None = None,
# Continuation
previous_interaction_id: str | None = None,
# Extra params
extra_headers: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
# LiteLLM params
custom_llm_provider: str | None = None,
@ -204,7 +204,7 @@ def create(
# System instruction
system_instruction: str | None = None,
# Generation config
generation_config: dict[str, Any] | None = None,
generation_config: Mapping[str, object] | None = None,
# Streaming
stream: bool | None = None,
# Storage
@ -215,13 +215,13 @@ def create(
environment: InteractionEnvironment | None = None,
# Response format
response_modalities: list[str] | None = None,
response_format: dict[str, Any] | None = None,
response_format: Mapping[str, object] | None = None,
response_mime_type: str | None = None,
# Continuation
previous_interaction_id: str | None = None,
# Extra params
extra_headers: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
# LiteLLM params
custom_llm_provider: str | None = None,
@ -359,7 +359,7 @@ def create(
@client
async def aget(
interaction_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -402,7 +402,7 @@ async def aget(
@client
def get(
interaction_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -461,7 +461,7 @@ def get(
@client
async def adelete(
interaction_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs: object,
@ -506,7 +506,7 @@ async def adelete(
@client
def delete(
interaction_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -565,7 +565,7 @@ def delete(
@client
async def acancel(
interaction_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -608,7 +608,7 @@ async def acancel(
@client
def cancel(
interaction_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,

39
litellm/litellm_core_utils/core_helpers.py
View file

@ -3,9 +3,9 @@
import copy
import logging
import re
from collections.abc import Collection, Iterable, Mapping
from collections.abc import Collection, Iterable, Iterator, Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, cast
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, cast, runtime_checkable
import httpx
from pydantic import TypeAdapter, ValidationError
@ -56,6 +56,33 @@ def as_str_mapping(value: object) -> Mapping[str, object] | None:
return value if isinstance(value, Mapping) else None
@runtime_checkable
class _SupportsIter(Protocol):
def __iter__(self) -> Iterator[object]: ...
@runtime_checkable
class _SupportsGetItem(Protocol):
def __getitem__(self, index: int, /) -> object: ...
def attribute_or_none(source: object, name: str) -> object:
return getattr(source, name, None)
def iter_items(value: object) -> Iterable[object]:
if isinstance(value, _SupportsIter):
return value
if isinstance(value, _SupportsGetItem):
return iter(value)
raise TypeError(f"'{type(value).__name__}' object is not iterable")
def attribute_items(source: object, name: str) -> Iterable[object]:
value: Final = attribute_or_none(source, name)
return iter_items(value) if value else ()
def normalize_drop_params(value: object) -> bool | None:
if value is None or isinstance(value, bool):
return value
@ -275,7 +302,13 @@ def remove_index_from_tool_calls(
tool_call.pop("index", None)
def remove_items_at_indices(items: list[Any] | None, indices: Iterable[int]) -> None:
class _PoppableByIndex(Protocol):
def __len__(self) -> int: ...
def pop(self, index: int, /) -> object: ...
def remove_items_at_indices(items: _PoppableByIndex | None, indices: Iterable[int]) -> None:
"""Remove items from a list in-place by index"""
if items is None:
return

15
litellm/litellm_core_utils/exception_mapping_utils.py
View file

@ -16,6 +16,7 @@ from litellm.litellm_core_utils.bug_report import (
build_bug_report,
should_report_bug,
)
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.secret_redaction import redact_string as redact_secret_string
from litellm.types.utils import LlmProviders
@ -185,12 +186,20 @@ def get_error_message(error_obj) -> str | None:
####### EXCEPTION MAPPING ################
_DECODED_ERROR_BODY: Final = TypeAdapter(object)
def _get_body_error_code(error_str: str) -> int | None:
"""Return error.code from a JSON error body, or None if not parseable."""
try:
body: Final = json.loads(error_str)
code: Final = body.get("error", {}).get("code")
return int(code) if code is not None else None
body: Final = as_str_mapping(_DECODED_ERROR_BODY.validate_python(json.loads(error_str)))
if body is None:
return None
error: Final = as_str_mapping(body.get("error", {}))
if error is None:
return None
code: Final = error.get("code")
return int(code) if isinstance(code, (str, int, float)) else None
except Exception:
return None

7
litellm/litellm_core_utils/get_model_cost_map.py
View file

@ -241,8 +241,9 @@ class _FetchAttemptRetryable:
def _parse_retry_after_seconds(response: httpx.Response) -> float | None:
header: Final[str | None] = response.headers.get("Retry-After")
if header is None:
try:
header: Final = response.headers["Retry-After"]
except KeyError:
return None
try:
seconds: Final = float(header)
@ -254,7 +255,7 @@ def _parse_retry_after_seconds(response: httpx.Response) -> float | None:
def _retry_wait_seconds(outcome: _FetchAttemptRetryable, attempt: int, rng: random.Random) -> float:
if outcome.retry_after_seconds is not None:
return min(outcome.retry_after_seconds, MODEL_COST_MAP_FETCH_MAX_WAIT_SECONDS)
return min(float(2**attempt) + rng.uniform(0.0, 1.0), MODEL_COST_MAP_FETCH_MAX_WAIT_SECONDS)
return min(2.0**attempt + rng.uniform(0.0, 1.0), MODEL_COST_MAP_FETCH_MAX_WAIT_SECONDS)
class _AsyncGetClient(Protocol):

6
litellm/litellm_core_utils/get_supported_openai_params.py
View file

@ -282,7 +282,11 @@ def get_supported_openai_params(
return ElevenLabsAudioTranscriptionConfig().get_supported_openai_params(model=model)
elif custom_llm_provider == "soniox":
if request_type == "transcription":
return litellm.SonioxAudioTranscriptionConfig().get_supported_openai_params(model=model)
from litellm.llms.soniox.audio_transcription.transformation import (
SonioxAudioTranscriptionConfig,
)
return SonioxAudioTranscriptionConfig().get_supported_openai_params(model=model)
elif custom_llm_provider in litellm._custom_providers:
if request_type == "chat_completion":
provider_config = litellm.ProviderConfigManager.get_provider_chat_config(

6
litellm/litellm_core_utils/json_validation_rule.py
View file

@ -1,8 +1,12 @@
import json
from typing import Any, Final
from pydantic import TypeAdapter
from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
_DECODED_RESPONSE: Final = TypeAdapter(object)
def normalize_json_schema_types(
schema: object,
@ -109,7 +113,7 @@ def validate_schema(schema: dict, response: str):
from litellm import JSONSchemaValidationError
try:
response_dict: Final = json.loads(response)
response_dict: Final = _DECODED_RESPONSE.validate_python(json.loads(response))
except json.JSONDecodeError:
raise JSONSchemaValidationError(model="", llm_provider="", raw_response=response, schema=json.dumps(schema))

78
litellm/litellm_core_utils/litellm_logging.py
View file

@ -609,6 +609,10 @@ class Logging(LiteLLMLoggingBaseClass):
def _on_detached_stream_failure(self, value: Callable[[Exception], Awaitable[None]] | None) -> None:
self.on_detached_stream_failure = value
@property
def _model_call_details_view(self) -> Mapping[str, object]:
return self.model_call_details
def __init__(
self,
model: str,
@ -1467,13 +1471,13 @@ class Logging(LiteLLMLoggingBaseClass):
try:
if callback == "supabase" and supabaseClient is not None:
verbose_logger.debug("reaches supabase for logging!")
model = self.model_call_details["model"]
messages = self.model_call_details["input"]
model = self._model_call_details_view["model"]
messages = self._model_call_details_view["input"]
verbose_logger.debug("supabaseClient: %s", supabaseClient)
supabaseClient.input_log_event(
model=model,
messages=messages,
end_user=self.model_call_details.get("user", "default"),
end_user=self._model_call_details_view.get("user", "default"),
litellm_call_id=self.litellm_params["litellm_call_id"],
print_verbose=print_verbose,
)
@ -1617,7 +1621,7 @@ class Logging(LiteLLMLoggingBaseClass):
callattr: Final = verbose_logger.warning if attr == "warning" else verbose_logger.debug
callattr(
"RAW RESPONSE:\n{}\n\n".format(
self.model_call_details.get("original_response", self.model_call_details)
self._model_call_details_view.get("original_response", self._model_call_details_view)
)
)
if getattr(self, "logger_fn", None) and callable(self.logger_fn):
@ -1755,8 +1759,8 @@ class Logging(LiteLLMLoggingBaseClass):
def get_response_ms(self) -> float:
now: Final = datetime.datetime.now()
start_seconds: Final = _timestamp_seconds(self.model_call_details.get("start_time", now))
end_seconds: Final = _timestamp_seconds(self.model_call_details.get("end_time", now))
start_seconds: Final = _timestamp_seconds(self._model_call_details_view.get("start_time", now))
end_seconds: Final = _timestamp_seconds(self._model_call_details_view.get("end_time", now))
if start_seconds is None or end_seconds is None:
return 0.0
return (end_seconds - start_seconds) * 1000
@ -2522,6 +2526,11 @@ class Logging(LiteLLMLoggingBaseClass):
self.callback_duration_ms += (time.time() - _start) * 1000
return payload
def _emit_stored_standard_logging_payload(self) -> None:
stored_payload: Final[StandardLoggingPayload | None] = self.model_call_details.get("standard_logging_object")
if stored_payload is not None:
emit_standard_logging_payload(stored_payload)
def _transform_usage_objects(self, result: object) -> object:
if isinstance(result, ResponsesAPIResponse):
result = result.model_copy()
@ -2956,10 +2965,10 @@ class Logging(LiteLLMLoggingBaseClass):
continue
else:
print_verbose("reaches supabase for streaming logging!")
result = kwargs["complete_streaming_response"]
result = self._model_call_details_view["complete_streaming_response"]
model = kwargs["model"]
messages = kwargs["messages"]
model = self._model_call_details_view["model"]
messages = self._model_call_details_view["messages"]
optional_params = kwargs.get("optional_params", {})
litellm_params = kwargs.get("litellm_params", {})
supabaseClient.log_event(
@ -2988,7 +2997,7 @@ class Logging(LiteLLMLoggingBaseClass):
if callback == "logfire" and logfireLogger is not None:
verbose_logger.debug("reaches logfire for success logging!")
kwargs = {}
for k, v in self.model_call_details.items():
for k, v in self._model_call_details_view.items():
if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine
kwargs[k] = v
@ -2998,7 +3007,7 @@ class Logging(LiteLLMLoggingBaseClass):
continue
else:
print_verbose("reaches logfire for streaming logging!")
result = kwargs["complete_streaming_response"]
result = self._model_call_details_view["complete_streaming_response"]
logfireLogger.log_event(
kwargs=self.model_call_details,
@ -3023,7 +3032,7 @@ class Logging(LiteLLMLoggingBaseClass):
if "complete_streaming_response" not in kwargs:
continue
else:
result = kwargs["complete_streaming_response"]
result = self._model_call_details_view["complete_streaming_response"]
lunaryLogger.log_event(
type=type,
@ -3033,7 +3042,7 @@ class Logging(LiteLLMLoggingBaseClass):
input=input,
user_id=kwargs.get("user", None),
# user_props=self.model_call_details.get("user_props", None),
extra=kwargs.get("optional_params", {}),
extra=self._model_call_details_view.get("optional_params", {}),
response_obj=result,
start_time=start_time,
end_time=end_time,
@ -3043,7 +3052,7 @@ class Logging(LiteLLMLoggingBaseClass):
if callback == "helicone" and heliconeLogger is not None:
print_verbose("reaches helicone for logging!")
model = self.model
messages = self.model_call_details["input"]
messages = self._model_call_details_view["input"]
kwargs = self.model_call_details
# this only logs streaming once, complete_streaming_response exists i.e when stream ends
@ -3052,7 +3061,7 @@ class Logging(LiteLLMLoggingBaseClass):
continue
else:
print_verbose("reaches helicone for streaming logging!")
result = kwargs["complete_streaming_response"]
result = self._model_call_details_view["complete_streaming_response"]
heliconeLogger.log_success(
model=model,
@ -3067,7 +3076,7 @@ class Logging(LiteLLMLoggingBaseClass):
global langFuseLogger
print_verbose("reaches langfuse for success logging!")
kwargs = {}
for k, v in self.model_call_details.items():
for k, v in self._model_call_details_view.items():
if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine
kwargs[k] = v
# this only logs streaming once, complete_streaming_response exists i.e when stream ends
@ -3080,7 +3089,7 @@ class Logging(LiteLLMLoggingBaseClass):
continue
else:
print_verbose("reaches langfuse for streaming logging!")
result = kwargs["complete_streaming_response"]
result = self._model_call_details_view["complete_streaming_response"]
langfuse_logger_to_use = LangFuseHandler.get_langfuse_logger_for_request(
globalLangfuseLogger=langFuseLogger,
@ -3105,7 +3114,7 @@ class Logging(LiteLLMLoggingBaseClass):
)
if callback == "greenscale" and greenscaleLogger is not None:
kwargs = {}
for k, v in self.model_call_details.items():
for k, v in self._model_call_details_view.items():
if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine
kwargs[k] = v
# this only logs streaming once, complete_streaming_response exists i.e when stream ends
@ -3118,7 +3127,7 @@ class Logging(LiteLLMLoggingBaseClass):
continue
else:
print_verbose("reaches greenscale for streaming logging!")
result = kwargs["complete_streaming_response"]
result = self._model_call_details_view["complete_streaming_response"]
greenscaleLogger.log_event(
kwargs=kwargs,
@ -3129,7 +3138,7 @@ class Logging(LiteLLMLoggingBaseClass):
)
if callback == "athina" and athinaLogger is not None:
deep_copy = {}
for k, v in self.model_call_details.items():
for k, v in self._model_call_details_view.items():
deep_copy[k] = v
athinaLogger.log_event(
kwargs=deep_copy,
@ -3140,7 +3149,7 @@ class Logging(LiteLLMLoggingBaseClass):
)
if callback == "traceloop":
deep_copy = {}
for k, v in self.model_call_details.items():
for k, v in self._model_call_details_view.items():
if k != "original_response":
deep_copy[k] = v
traceloopLogger.log_event(
@ -3193,7 +3202,7 @@ class Logging(LiteLLMLoggingBaseClass):
self.model_call_details["complete_response"] = self.model_call_details.get(
"complete_streaming_response", {}
)
result = self.model_call_details["complete_response"]
result = self._model_call_details_view["complete_response"]
openMeterLogger.log_success_event(
kwargs=self.model_call_details,
response_obj=result,
@ -3218,7 +3227,7 @@ class Logging(LiteLLMLoggingBaseClass):
self.model_call_details["complete_response"] = self.model_call_details.get(
"complete_streaming_response", {}
)
result = self.model_call_details["complete_response"]
result = self._model_call_details_view["complete_response"]
callback.log_success_event(
kwargs=self.model_call_details,
@ -3369,8 +3378,7 @@ class Logging(LiteLLMLoggingBaseClass):
self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload(
result, start_time, end_time
)
if (prepared_payload := self.model_call_details.get("standard_logging_object")) is not None:
emit_standard_logging_payload(prepared_payload)
self._emit_stored_standard_logging_payload()
## BUILD COMPLETE STREAMED RESPONSE
if "async_complete_streaming_response" in self.model_call_details:
@ -3434,8 +3442,7 @@ class Logging(LiteLLMLoggingBaseClass):
)
# print standard logging payload
if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None:
emit_standard_logging_payload(standard_logging_payload)
self._emit_stored_standard_logging_payload()
elif self.call_type == "pass_through_endpoint":
print_verbose("Async success callbacks: Got a pass-through endpoint response")
@ -3457,8 +3464,7 @@ class Logging(LiteLLMLoggingBaseClass):
)
# print standard logging payload
if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None:
emit_standard_logging_payload(standard_logging_payload)
self._emit_stored_standard_logging_payload()
callbacks: Final = self.get_combined_callback_list(
dynamic_success_callbacks=self.dynamic_async_success_callbacks,
global_callbacks=litellm._async_success_callback,
@ -3815,7 +3821,7 @@ class Logging(LiteLLMLoggingBaseClass):
model = self.model
input = self.model_call_details["input"]
input = self._model_call_details_view["input"]
_type = "embed" if self.call_type == CallTypes.embedding.value else "llm"
@ -3823,7 +3829,7 @@ class Logging(LiteLLMLoggingBaseClass):
kwargs=self.model_call_details,
type=_type,
event="error",
user_id=self.model_call_details.get("user", "default"),
user_id=self._model_call_details_view.get("user", "default"),
model=model,
input=input,
error=traceback_exception,
@ -3844,11 +3850,11 @@ class Logging(LiteLLMLoggingBaseClass):
supabaseClient.log_event(
model=self.model if hasattr(self, "model") else "",
messages=self.messages,
end_user=self.model_call_details.get("user", "default"),
end_user=self._model_call_details_view.get("user", "default"),
response_obj=result,
start_time=start_time,
end_time=end_time,
litellm_call_id=self.model_call_details["litellm_call_id"],
litellm_call_id=self._model_call_details_view["litellm_call_id"],
print_verbose=print_verbose,
)
if callable(callback): # custom logger functions
@ -3878,7 +3884,7 @@ class Logging(LiteLLMLoggingBaseClass):
global langFuseLogger
verbose_logger.debug("reaches langfuse for logging failure")
kwargs = {}
for k, v in self.model_call_details.items():
for k, v in self._model_call_details_view.items():
if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine
kwargs[k] = v
# this only logs streaming once, complete_streaming_response exists i.e when stream ends
@ -3918,7 +3924,7 @@ class Logging(LiteLLMLoggingBaseClass):
if callback == "logfire" and logfireLogger is not None:
verbose_logger.debug("reaches logfire for failure logging!")
kwargs = {}
for k, v in self.model_call_details.items():
for k, v in self._model_call_details_view.items():
if k != "original_response": # copy.deepcopy raises errors as this could be a coroutine
kwargs[k] = v
kwargs["exception"] = exception
@ -4379,7 +4385,7 @@ class Logging(LiteLLMLoggingBaseClass):
)
model_response: Final = litellm.ModelResponse(id=served_id)
model_response.model = self.model
usage: Final = getattr(result, "usage", None)
usage: Final = result.usage
if usage is not None and ResponseAPILoggingUtils.is_response_api_usage(usage):
setattr(
model_response,

3
litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py
View file

@ -405,6 +405,7 @@ class _ParallelToolUseArgs(TypedDict):
_PARALLEL_TOOL_USE_ARGS: Final = TypeAdapter(_ParallelToolUseArgs)
_DECODED_TOOL_ARGUMENTS: Final = TypeAdapter(object)
class _CachedMessageProviderFields(TypedDict, total=False):
@ -453,7 +454,7 @@ def handle_invalid_parallel_tool_calls(
if isinstance(tool_call, ChatCompletionMessageCustomToolCall):
continue
current_function = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
function_args = _DECODED_TOOL_ARGUMENTS.validate_python(json.loads(tool_call.function.arguments))
if current_function == "multi_tool_use.parallel":
verbose_logger.debug("OpenAI did a weird pseudo-multi-tool-use call, fixing call structure..")
for _fake_i, _fake_tool_use in enumerate(

2
litellm/litellm_core_utils/llm_response_utils/response_metadata.py
View file

@ -204,7 +204,7 @@ class ResponseMetadata:
#########################################################
# 3. Detailed per-phase timing (opt-in via env var)
#########################################################
llm_api_duration_ms: Final = logging_obj.model_call_details.get("llm_api_duration_ms")
llm_api_duration_ms: Final[float | None] = logging_obj.model_call_details.get("llm_api_duration_ms")
if LITELLM_DETAILED_TIMING and llm_api_duration_ms is not None:
detailed: Final[dict[str, float]] = {
"timing_llm_api_ms": round(llm_api_duration_ms, 4),

2
litellm/litellm_core_utils/logging_callback_manager.py
View file

@ -29,7 +29,7 @@ class LoggingCallbackManager:
# healthy maximum number of callbacks - unlikely someone needs more than 20
MAX_CALLBACKS = 30
def _is_async_callable(self, callback) -> bool:
def _is_async_callable(self, callback: object) -> bool:
"""Check if a callback is async. Used to auto-route callbacks to the correct list."""
try:
from litellm.litellm_core_utils.coroutine_checker import coroutine_checker

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

@ -2539,6 +2539,9 @@ def merge_consecutive_system_messages(
]
_DECODED_JSON: Final = TypeAdapter(object)
def _attempt_json_repair(s: str) -> object | None:
"""
Attempt to repair truncated JSON produced by LLM tool calls.
@ -2590,7 +2593,7 @@ def _attempt_json_repair(s: str) -> object | None:
candidate += "".join(reversed(opener_stack))
try:
return json.loads(candidate)
return _DECODED_JSON.validate_python(json.loads(candidate))
except json.JSONDecodeError:
pass
@ -2629,7 +2632,7 @@ def parse_tool_call_arguments(
return {}
try:
return json.loads(arguments)
return _DECODED_JSON.validate_python(json.loads(arguments))
except json.JSONDecodeError as original_error:
repaired: Final = _attempt_json_repair(arguments)
if repaired is not None:

15
litellm/litellm_core_utils/prompt_templates/factory.py
View file

@ -12,7 +12,7 @@ from types import MappingProxyType
from typing import Any, Final, TypeAlias, TypedDict, cast, overload
from jinja2.sandbox import ImmutableSandboxedEnvironment
from pydantic import BaseModel, JsonValue
from pydantic import BaseModel, JsonValue, TypeAdapter
import litellm
import litellm.types
@ -28,7 +28,7 @@ from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import
parts_of,
)
from litellm.litellm_core_utils.url_utils import async_safe_get, safe_get
from litellm.llms.custom_httpx.http_handler import HTTPHandler, get_async_httpx_client
from litellm.llms.custom_httpx.http_handler import HTTPHandler, get_async_httpx_client, header_value
from litellm.types.files import get_file_extension_from_mime_type
from litellm.types.llms.anthropic import *
from litellm.types.llms.bedrock import CachePointBlock
@ -216,6 +216,7 @@ def _handle_ollama_system_message(messages: list, prompt: str, msg_i: int) -> tu
_OLLAMA_USER_ROLES: Final = frozenset({"user", "tool", "function"})
_DECODED_JSON: Final = TypeAdapter(object)
def _ollama_bad_message(model: str, message: AllMessageValues, msg_i: int, detail: str) -> "litellm.BadRequestError":
@ -1291,7 +1292,7 @@ def get_thought_signature_from_tool(tool: Mapping[str, object]) -> str | None:
if signature_from_function_fields:
return cast(str, signature_from_function_fields) # cast-ok: untyped provider response field
elif hasattr(function, "provider_specific_fields") and getattr(function, "provider_specific_fields"):
function_provider_fields: Final[object] = getattr(function, "provider_specific_fields")
function_provider_fields: Final[object] = getattr(function, "provider_specific_fields", None)
if isinstance(function_provider_fields, dict):
typed_function_provider_fields: Final = cast( # cast-ok: provider fields are dynamic
dict[str, object], function_provider_fields
@ -3032,7 +3033,7 @@ def parse_xml_params(xml_content, json_schema: dict | None = None):
for value in _element:
try:
if value.text is not None:
_value = json.loads(value.text)
_value = _DECODED_JSON.validate_python(json.loads(value.text))
else:
continue
except json.JSONDecodeError:
@ -3042,7 +3043,7 @@ def parse_xml_params(xml_content, json_schema: dict | None = None):
# If property is not an array, append the value directly
elif _element is not None and _element.text is not None:
try:
_value = json.loads(_element.text)
_value = _DECODED_JSON.validate_python(json.loads(_element.text))
except json.JSONDecodeError:
_value = _element.text
params[prop] = _value
@ -3402,7 +3403,7 @@ def _load_image_from_url(image_url):
response.raise_for_status() # Raise an exception for HTTP errors
# Check the response's content type to ensure it is an image
content_type: Final = response.headers.get("content-type")
content_type: Final = header_value(response.headers, "content-type")
if not content_type or "image" not in content_type:
raise ValueError(f"URL does not point to a valid image (content-type: {content_type})")
@ -3604,7 +3605,7 @@ class BedrockImageProcessor:
@staticmethod
def _post_call_image_processing(response: httpx.Response, image_url: str = "") -> tuple[str, str]:
# Check the response's content type to ensure it is an image
content_type: str | None = response.headers.get("content-type")
content_type: str | None = header_value(response.headers, "content-type")
# Use helper function to infer content type with fallback logic
content_type = infer_content_type_from_url_and_content(

7
litellm/litellm_core_utils/prompt_templates/huggingface_template_handler.py
View file

@ -2,6 +2,7 @@ import json
from datetime import datetime
from typing import Any, Final, Literal
from pydantic import TypeAdapter
from typing_extensions import NotRequired, ReadOnly, TypedDict
from litellm.llms.custom_httpx.http_handler import (
@ -10,6 +11,8 @@ from litellm.llms.custom_httpx.http_handler import (
)
from litellm.types.llms.custom_http import httpxSpecialProvider
_DECODED_TOKENIZER_CONFIG: Final = TypeAdapter(object)
class _TokenizerConfigResult(TypedDict):
"""Outcome of a tokenizer_config.json fetch, carrying the parsed document when the fetch succeeded."""
@ -55,7 +58,7 @@ def get_tokenizer_config(hf_model_name: str) -> _TokenizerConfigResult:
except Exception as e:
raise e
if response.status_code == 200:
tokenizer_config: Final = json.loads(response.content)
tokenizer_config: Final = _DECODED_TOKENIZER_CONFIG.validate_python(json.loads(response.content))
return {"status": "success", "tokenizer": tokenizer_config}
else:
return {"status": "failure"}
@ -83,7 +86,7 @@ async def aget_tokenizer_config(hf_model_name: str) -> _TokenizerConfigResult:
except Exception as e:
raise e
if response.status_code == 200:
tokenizer_config: Final = json.loads(response.content)
tokenizer_config: Final = _DECODED_TOKENIZER_CONFIG.validate_python(json.loads(response.content))
return {"status": "success", "tokenizer": tokenizer_config}
else:
return {"status": "failure"}

11
litellm/litellm_core_utils/prompt_templates/image_handling.py
View file

@ -29,6 +29,13 @@ _FETCH_ATTEMPTS: Final = 3
in_memory_cache: Final = InMemoryCache(max_size_in_memory=MAX_IMGS_IN_MEMORY)
def _header_value(response: Response, name: str) -> str | None:
try:
return response.headers[name]
except KeyError:
return None
def _process_image_response(response: Response, url: str) -> str:
if response.status_code != 200:
raise litellm.ImageFetchError(
@ -36,7 +43,7 @@ def _process_image_response(response: Response, url: str) -> str:
)
# Check size before downloading if Content-Length header is present
content_length: Final = response.headers.get("Content-Length")
content_length: Final = _header_value(response, "Content-Length")
if content_length is not None:
size_mb = int(content_length) / (1024 * 1024)
if size_mb > MAX_IMAGE_URL_DOWNLOAD_SIZE_MB:
@ -64,7 +71,7 @@ def _process_image_response(response: Response, url: str) -> str:
img_type: Final = infer_content_type_from_url_and_content(
url=url,
content=bytes(image_bytes),
current_content_type=response.headers.get("Content-Type"),
current_content_type=_header_value(response, "Content-Type"),
)
except ValueError as e:
raise litellm.ImageFetchError(

10
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -6,6 +6,7 @@ from dataclasses import dataclass
from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol, TypedDict, cast
from pydantic import TypeAdapter
from typing_extensions import ReadOnly
import litellm
@ -107,6 +108,9 @@ class _LoggingWorker(Protocol):
def ensure_initialized_and_enqueue(self, async_coroutine: Coroutine[object, object, None]) -> None: ...
_DECODED_JSON: Final = TypeAdapter(object)
def _decode_json_object(payload: str) -> Mapping[str, object]:
"""Decode a realtime frame into its top-level field mapping."""
return json.loads(payload)
@ -293,11 +297,13 @@ class RealTimeStreaming:
"""Extract user text content from client WebSocket events for spend logging."""
try:
if isinstance(message, str):
msg_obj = json.loads(message)
msg_obj = _DECODED_JSON.validate_python(json.loads(message))
elif isinstance(message, dict):
msg_obj = message
else:
return
if not isinstance(msg_obj, dict):
return
msg_type: Final = msg_obj.get("type", "")
@ -773,7 +779,7 @@ class RealTimeStreaming:
if not self._has_audio_transcription_guardrails():
return setup_message
try:
obj: Final = json.loads(setup_message)
obj: Final = _DECODED_JSON.validate_python(json.loads(setup_message))
except (json.JSONDecodeError, TypeError):
return setup_message
setup: Final = obj.get("setup") if isinstance(obj, dict) else None

8
litellm/litellm_core_utils/safe_json_loads.py
View file

@ -3,7 +3,11 @@ Helper for safe JSON loading in LiteLLM.
"""
import json
from typing import Any
from typing import Any, Final
from pydantic import TypeAdapter
_DECODED_JSON: Final = TypeAdapter(object)
def safe_json_loads(data: str, default: object = None) -> Any:
@ -11,6 +15,6 @@ def safe_json_loads(data: str, default: object = None) -> Any:
Safely parse a JSON string. If parsing fails, return the default value (None by default).
"""
try:
return json.loads(data)
return _DECODED_JSON.validate_python(json.loads(data))
except Exception:
return default

2
litellm/litellm_core_utils/specialty_caches/dynamic_logging_cache.py
View file

@ -69,6 +69,6 @@ class DynamicLoggingCache:
response: Final = self.cache.get_cache(key=key_name)
return response
def set_cache(self, credentials: dict, service_name: str, logging_obj: Any) -> None:
def set_cache(self, credentials: dict, service_name: str, logging_obj: object) -> None:
key_name: Final = self.get_cache_key(args={**credentials, "service_name": service_name})
self.cache.set_cache(key=key_name, value=logging_obj)

11
litellm/litellm_core_utils/streaming_handler.py
View file

@ -81,6 +81,11 @@ def _stamp_served_service_tier(response: ModelResponseStream, complete_streaming
setattr(response, "service_tier", served_tier) # noqa: B010 # pydantic extra, not a declared field
def _apply_provider_fields(response: ModelResponseStream, fields: Mapping[str, object]) -> None:
for key, value in fields.items():
setattr(response, key, value)
def is_async_iterable(obj: object) -> bool:
"""
Check if an object is an async iterable (can be used with 'async for').
@ -857,8 +862,7 @@ class CustomStreamWrapper:
provider_specific_fields: Final = getattr(original_chunk, "provider_specific_fields", None)
if provider_specific_fields is not None:
model_response.provider_specific_fields = provider_specific_fields
for k, v in provider_specific_fields.items():
setattr(model_response, k, v)
_apply_provider_fields(model_response, provider_specific_fields)
return model_response
def is_chunk_non_empty(
@ -1211,8 +1215,7 @@ class CustomStreamWrapper:
"provider_specific_fields" in anthropic_response_obj
and anthropic_response_obj["provider_specific_fields"] is not None
):
for key, value in anthropic_response_obj["provider_specific_fields"].items():
setattr(model_response, key, value)
_apply_provider_fields(model_response, anthropic_response_obj["provider_specific_fields"])
response_obj = cast(dict[str, object], anthropic_response_obj)
elif self.model == "replicate" or self.custom_llm_provider == "replicate":

13
litellm/llms/anthropic/chat/guardrail_translation/handler.py
View file

@ -24,6 +24,7 @@ from pydantic import TypeAdapter, ValidationError
from typing_extensions import ReadOnly, TypedDict, assert_never
from litellm._logging import verbose_proxy_logger
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.anthropic.pass_through.adapters.transformation import (
LiteLLMAnthropicMessagesAdapter,
@ -225,6 +226,7 @@ def _write_back_message_text(message: _WritableMessage, target: MessageTextTarge
_TOOL_USE_INPUT_ADAPTER: Final = TypeAdapter(dict[str, object])
_DECODED_SSE_DATA: Final = TypeAdapter(object)
_RELEASED_TOOL_USE_STOP: Final = (
b"event: message_delta\n"
b'data: {"type": "message_delta", "delta": {"stop_reason": "tool_use", "stop_sequence": null}, '
@ -358,8 +360,8 @@ class AnthropicMessagesHandler(BaseTranslation):
def build_stream_error_items(
self,
exc: "HTTPException",
responses_so_far: Sequence[Any] | None = None,
) -> Sequence[Any] | None:
responses_so_far: Sequence[object] | None = None,
) -> Sequence[object] | None:
from litellm.proxy.common_request_processing import (
serialize_http_exception_detail,
)
@ -473,12 +475,13 @@ class AnthropicMessagesHandler(BaseTranslation):
if not line.startswith("data:"):
return ()
try:
parsed: Final[object] = json.loads(line[len("data:") :].strip())
parsed: Final = _DECODED_SSE_DATA.validate_python(json.loads(line[len("data:") :].strip()))
except json.JSONDecodeError:
return ()
if not isinstance(parsed, dict):
event: Final = as_str_mapping(parsed)
if event is None:
return ()
return (_as_str_mapping(parsed),)
return (event,)
@staticmethod
def _iter_sse_events(item: object) -> Sequence[Mapping[str, object]]:

6
litellm/llms/anthropic/chat/handler.py
View file

@ -11,6 +11,7 @@ from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Union, cast
import httpx
from pydantic import TypeAdapter
import litellm
import litellm.litellm_core_utils
@ -81,6 +82,9 @@ def _loads_stream_chunk(payload: str) -> dict[str, object]:
return json.loads(payload)
_DECODED_JSON: Final = TypeAdapter(object)
_EnvelopeVerdict = Literal["undecided", "pass_through", "hold"]
_FIRST_OBJECT_KEY: Final = re.compile(r'^\s*\{\s*"((?:[^"\\]|\\.)*)"\s*:')
_OBJECT_KEY_PREFIX: Final = re.compile(r'^\s*(?:\{\s*(?:"(?:[^"\\]|\\.)*\\?(?:"\s*)?)?)?$')
@ -164,7 +168,7 @@ def _prefix_verdict(buffered_json: str, schema: ResponseFormatSchema, tool_name:
def _held_answer(held: str, schema: ResponseFormatSchema) -> str:
try:
payload: Final[object] = json.loads(held)
payload: Final = _DECODED_JSON.validate_python(json.loads(held))
except (json.JSONDecodeError, RecursionError):
return held
contents: Final = response_format_envelope_contents(payload, schema, RESPONSE_FORMAT_TOOL_NAME)

13
litellm/llms/anthropic/chat/transformation.py
View file

@ -523,7 +523,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
)
# Include caller information if present (for programmatic tool calling)
if "caller" in anthropic_tool_content:
tool_call["caller"] = cast(dict[str, Any], anthropic_tool_content["caller"])
tool_call["caller"] = cast(dict[str, object], anthropic_tool_content["caller"])
return tool_call
@staticmethod
@ -1707,8 +1707,8 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
@staticmethod
def map_openai_context_management_to_anthropic(
context_management: list[dict[str, Any]] | dict[str, Any],
) -> dict[str, Any] | None:
context_management: Sequence[Mapping[str, object]] | dict[str, object],
) -> dict[str, object] | None:
"""
OpenAI format: [{"type": "compaction", "compact_threshold": 200000}]
Anthropic format: {
@ -1732,7 +1732,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
# If in OpenAI format (list), transform to Anthropic format
if isinstance(context_management, list):
anthropic_edits: Final = []
anthropic_edits: Final[list[dict[str, object]]] = []
for entry in context_management:
if not isinstance(entry, dict):
continue
@ -2771,7 +2771,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
code_by_id: Final[dict[str, str]] = {}
for tc in tool_calls:
try:
args: object = json.loads(tc.get("function", {}).get("arguments", "{}"))
args: object = _DECODED_JSON.validate_python(json.loads(tc.get("function", {}).get("arguments", "{}")))
if not isinstance(args, Mapping):
continue
call_id = tc.get("id")
@ -3096,7 +3096,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if split_answer is not None:
return litellm.Message(content=json.dumps(split_answer))
try:
payload: Final[object] = json.loads(json_mode_content_str)
payload: Final = _DECODED_JSON.validate_python(json.loads(json_mode_content_str))
except (json.JSONDecodeError, RecursionError):
# json decode error does occur, return the original tool response str
return litellm.Message(content=json_mode_content_str)
@ -3128,6 +3128,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
)
_DECODED_JSON: Final = TypeAdapter(object)
_REQUEST_TOOLS: Final = TypeAdapter(tuple[Mapping[str, object], ...])

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

@ -111,6 +111,7 @@ _DOTTED_VERSION_RE: Final = re.compile(r"(\d)\.(\d)")
_CLAUDE_CODE_BILLING_HEADER_PREFIX: Final = "x-anthropic-billing-header:"
_OBJECT_MAPPING_ADAPTER: Final = TypeAdapter(dict[object, object])
_CLAUDE_CODE_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
_DECODED_JSON: Final = TypeAdapter(object)
_CLAUDE_CODE_USER_AGENT_PREFIXES: Final = ("claude-cli/", "claude-code/")
@ -415,7 +416,7 @@ def _sanitized_anthropic_error(response: httpx.Response, detail: str | None = No
if detail is not None:
return f"HTTP {response.status_code}: {detail}"
try:
body: Final = response.json()
body: Final = _DECODED_JSON.validate_python(response.json())
except ValueError:
return f"HTTP {response.status_code}"
error: Final = body.get("error") if isinstance(body, dict) else None
@ -1700,7 +1701,7 @@ def strip_encrypted_reasoning_blocks_from_anthropic_messages(
def strip_thinking_blocks_from_anthropic_messages_request_dict(
data: dict[str, Any],
data: dict[str, object],
) -> None:
"""
Mutate an Anthropic Messages-style request dict: strip thinking blocks from

4
litellm/llms/anthropic/pass_through/adapters/transformation.py
View file

@ -1352,12 +1352,12 @@ class LiteLLMAnthropicMessagesAdapter:
choices: list[Choices],
tool_name_mapping: dict[str, str] | None = None,
) -> list[dict[str, object]]:
new_content: Final[list[dict[str, Any]]] = []
new_content: Final[list[dict[str, object]]] = []
for choice, compaction_blocks in (
(choice, _compaction_blocks(_optional_attr(choice.message, "provider_specific_fields")))
for choice in choices
):
new_content.extend(_COMPACTION_BLOCK.validate_python(block) for block in compaction_blocks)
new_content.extend(dict(_COMPACTION_BLOCK.validate_python(block)) for block in compaction_blocks)
# Handle thinking blocks first
if hasattr(choice.message, "thinking_blocks") and choice.message.thinking_blocks:
for thinking_block in choice.message.thinking_blocks:

8
litellm/llms/anthropic/pass_through/messages/handler.py
View file

@ -9,7 +9,7 @@ import asyncio
import contextvars
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
from functools import partial
from typing import Any, Final, cast
from typing import Final, cast
from pydantic import TypeAdapter
@ -312,7 +312,7 @@ async def anthropic_messages(
client: AsyncHTTPHandler | None = None,
custom_llm_provider: str | None = None,
**kwargs,
) -> AnthropicMessagesResponse | Iterator[bytes] | AsyncIterator[Any]:
) -> AnthropicMessagesResponse | Iterator[bytes] | AsyncIterator[object]:
"""
Async: Make llm api request in Anthropic /messages API spec.
@ -489,8 +489,8 @@ def anthropic_messages_handler(
) -> (
AnthropicMessagesResponse
| Iterator[bytes]
| AsyncIterator[Any]
| Coroutine[Any, Any, AnthropicMessagesResponse | AsyncIterator[Any] | Iterator[bytes]]
| AsyncIterator[object]
| Coroutine[object, None, AnthropicMessagesResponse | AsyncIterator[object] | Iterator[bytes]]
):
"""
Makes Anthropic `/v1/messages` API calls In the Anthropic API Spec

5
litellm/llms/anthropic/pass_through/messages/streaming_iterator.py
View file

@ -91,11 +91,14 @@ def is_anthropic_content_delta_chunk(chunk: object) -> bool:
return False
_DECODED_SSE_DATA: Final = TypeAdapter(object)
def _decoded_sse_data_line(line: bytes) -> object | None:
if not line.startswith(b"data:"):
return None
try:
return json.loads(line[len(b"data:") :].strip())
return _DECODED_SSE_DATA.validate_python(json.loads(line[len(b"data:") :].strip()))
except (ValueError, TypeError):
return None

6
litellm/llms/anthropic/pass_through/messages/transformation.py
View file

@ -1,5 +1,5 @@
from collections.abc import AsyncIterator, Mapping, Sequence
from typing import Any, ClassVar, Final
from typing import ClassVar, Final
import httpx
@ -265,7 +265,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
self,
headers: dict,
model: str,
messages: list[Any],
messages: list[object],
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,
@ -293,7 +293,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
self,
headers: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
model: str,
messages: list[Any], # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
messages: list[object], # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
optional_params: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
litellm_params: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
api_key: str | None = None,

16
litellm/llms/anthropic/pass_through/messages/utils.py
View file

@ -1,10 +1,11 @@
from collections.abc import Iterable, Mapping, Sequence
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Final, cast, get_type_hints
from typing import TYPE_CHECKING, Final, cast, get_type_hints
from pydantic import JsonValue
from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.llms.anthropic.common_utils import (
flatten_unencrypted_web_search_results_in_anthropic_messages,
sanitize_tool_use_ids_in_anthropic_messages,
@ -30,7 +31,7 @@ INCOMPLETE_STREAM_ERROR_MESSAGE: Final = (
)
def get_safeguard_refusal_stop_details(response: object) -> Mapping[str, Any] | None:
def get_safeguard_refusal_stop_details(response: object) -> Mapping[str, object] | None:
"""
Return the ``stop_details`` of an Anthropic Messages response refused by a
safeguard (``stop_reason: "refusal"`` carrying ``stop_details``:
@ -175,7 +176,7 @@ def anthropic_messages_optional_param_keys() -> frozenset[str]:
class AnthropicMessagesRequestUtils:
@staticmethod
def get_requested_anthropic_messages_optional_param(
params: dict[str, Any],
params: Mapping[str, object],
*,
model: str | None = None,
drop_params: bool = False,
@ -222,10 +223,15 @@ class AnthropicMessagesRequestUtils:
)
if is_reasoning_auto_summary_enabled():
thinking_param: Final = filtered_params.get("thinking")
if isinstance(thinking_param, dict) and thinking_param.get("type") != "disabled":
thinking_mapping: Final = as_str_mapping(thinking_param)
if (
isinstance(thinking_param, dict)
and thinking_mapping is not None
and thinking_mapping.get("type") != "disabled"
):
return cast(
AnthropicMessagesRequestOptionalParams,
{**filtered_params, "thinking": {**thinking_param, "display": "summarized"}},
{**filtered_params, "thinking": {**thinking_mapping, "display": "summarized"}},
)
return cast(AnthropicMessagesRequestOptionalParams, filtered_params)

12
litellm/llms/azure/realtime/handler.py
View file

@ -6,7 +6,7 @@ This requires websockets, and is currently only supported on LiteLLM Proxy.
from collections.abc import Mapping
from types import MappingProxyType
from typing import Any, Final, Protocol, cast
from typing import Final, Protocol, cast
from litellm._logging import verbose_proxy_logger
from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
@ -28,7 +28,15 @@ from ..azure import AzureChatCompletion
# BACKEND_WS_URL = "ws://localhost:8080/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
async def forward_messages(client_ws: Any, backend_ws: Any):
class _RealtimeTextSource(Protocol):
async def recv(self) -> str: ...
class _RealtimeTextSink(Protocol):
async def send_text(self, data: str) -> None: ...
async def forward_messages(client_ws: _RealtimeTextSink, backend_ws: _RealtimeTextSource):
import websockets
try:

5
litellm/llms/azure/responses/transformation.py
View file

@ -3,6 +3,7 @@ from typing import TYPE_CHECKING, Any, Final, Literal
import httpx
from openai.types.responses import ResponseReasoningItem
from pydantic import TypeAdapter
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
@ -21,6 +22,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_DECODED_JSON: Final = TypeAdapter(object)
class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
# Parameters not supported by Azure Responses API
@ -345,7 +348,7 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
Transform the cancel response API response into a ResponsesAPIResponse
"""
try:
raw_response_json: Final = raw_response.json()
raw_response_json: Final = _DECODED_JSON.validate_python(raw_response.json())
except Exception:
from litellm.llms.azure.chat.gpt_transformation import AzureOpenAIError

3
litellm/llms/azure_ai/anthropic/messages_transformation.py
View file

@ -2,6 +2,7 @@
Azure Anthropic messages transformation config - extends AnthropicMessagesConfig with Azure authentication
"""
from collections.abc import Sequence
from typing import Any, Final
from litellm.llms.anthropic.pass_through.messages.transformation import (
@ -29,7 +30,7 @@ class AzureAnthropicMessagesConfig(AnthropicMessagesConfig):
self,
headers: dict,
model: str,
messages: list[Any],
messages: Sequence[object],
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,

13
litellm/llms/base_llm/base_utils.py
View file

@ -14,6 +14,7 @@ from openai.lib import _parsing, _pydantic # pyright: ignore[reportPrivateUsage
from pydantic import BaseModel, JsonValue, TypeAdapter, ValidationError
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolCallChunk
from litellm.types.utils import Message, ProviderSpecificModelInfo, TokenCountResponse
@ -24,8 +25,8 @@ class BaseTokenCounter(ABC):
self,
model_to_use: str,
messages: list[dict[str, Any]] | None,
contents: list[dict[str, Any]] | None,
deployment: dict[str, Any] | None = None,
contents: list[dict[str, object]] | None,
deployment: dict[str, object] | None = None,
request_model: str = "",
tools: list[dict[str, Any]] | None = None,
system: Any | None = None,
@ -141,8 +142,9 @@ def convert_tool_response_to_message(
json_mode_content_str: Final[str | None] = tool_calls[0]["function"].get("arguments")
try:
if json_mode_content_str is not None:
args: Final = json.loads(json_mode_content_str)
if isinstance(args, dict) and (values := args.get("values")) is not None:
args: Final = _DECODED_JSON.validate_python(json.loads(json_mode_content_str))
args_object: Final = as_str_mapping(args)
if args_object is not None and (values := args_object.get("values")) is not None:
_message = Message(content=json.dumps(values))
return _message
else:
@ -170,6 +172,7 @@ class ResponseFormatSchema:
return self.allows_undeclared_keys and not self.required_names
_DECODED_JSON: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, JsonValue])
_NAMES: Final = TypeAdapter(frozenset[str])
@ -224,7 +227,7 @@ def _json_object_or_encoded_object(value: JsonValue) -> Mapping[str, JsonValue]
if not isinstance(value, str):
return as_json_object(value)
try:
return as_json_object(json.loads(value))
return as_json_object(_DECODED_JSON.validate_python(json.loads(value)))
except (json.JSONDecodeError, RecursionError):
return None

6
litellm/llms/base_llm/batches/transformation.py
View file

@ -1,6 +1,7 @@
import types
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from httpx import Headers
@ -42,9 +43,10 @@ class BaseBatchesConfig(ABC):
@classmethod
def get_config(cls):
"""Get configuration dictionary for this class."""
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("__")
and not k.startswith("_abc")
and not isinstance(

8
litellm/llms/base_llm/containers/transformation.py
View file

@ -2,6 +2,7 @@ from __future__ import annotations
import types
from abc import ABC, abstractmethod
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -45,9 +46,10 @@ class BaseContainerConfig(ABC):
@classmethod
def get_config(cls):
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("__")
and not k.startswith("_abc")
and not isinstance(
@ -130,7 +132,7 @@ class BaseContainerConfig(ABC):
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: dict[str, Any] | None = None,
extra_query: Mapping[str, object] | None = None,
) -> tuple[str, dict]:
"""Transform the container list request into a URL and params.
@ -206,7 +208,7 @@ class BaseContainerConfig(ABC):
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: dict[str, Any] | None = None,
extra_query: Mapping[str, object] | None = None,
) -> tuple[str, dict]:
"""Transform the container file list request into a URL and params.

3
litellm/llms/base_llm/files/transformation.py
View file

@ -24,10 +24,11 @@ if TYPE_CHECKING:
from litellm.litellm_core_utils.tokenizer import Encoding as Tokenizer
from litellm.router import Router as _Router
from litellm.types.llms.openai import HttpxBinaryResponseContent
from litellm.types.proxy.auth.user_api_key_auth import Span as _Span
from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth as _UserAPIKeyAuth
LiteLLMLoggingObj = _LiteLLMLoggingObj
Span = Any
Span = _Span
Router = _Router
UserAPIKeyAuth = _UserAPIKeyAuth
else:

10
litellm/llms/base_llm/google_genai/transformation.py
View file

@ -1,6 +1,7 @@
import types
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -30,9 +31,10 @@ class BaseGoogleGenAIGenerateContentConfig(ABC):
@classmethod
def get_config(cls):
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("__")
and not k.startswith("_abc")
and not isinstance(
@ -78,7 +80,7 @@ class BaseGoogleGenAIGenerateContentConfig(ABC):
self,
generate_content_config_dict: GenerateContentConfigDict,
model: str,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Map Google GenAI parameters to provider-specific format.
@ -161,7 +163,7 @@ class BaseGoogleGenAIGenerateContentConfig(ABC):
contents: GenerateContentContentListUnionDict,
tools: ToolConfigDict | None,
generate_content_config_dict: dict,
system_instruction: Any | None = None,
system_instruction: object | None = None,
) -> dict:
"""
Transform the request parameters for the generate content API.

7
litellm/llms/base_llm/managed_resources/utils.py
View file

@ -7,13 +7,13 @@ different managed resource types (files, vector stores, etc.).
import base64
import re
from typing import Any, Final, Literal
from typing import Final, Literal
PASSTHROUGH_MANAGED_ID_AZURE_PROVIDERS: Final = ("azure", "azure_ai")
def resolve_passthrough_managed_id_provider(
custom_llm_provider: Any,
custom_llm_provider: object,
) -> str | None:
"""Map a pass-through ``custom_llm_provider`` to the provider scope that
namespaces passthrough managed object IDs, or ``None`` when the route is not
@ -29,7 +29,8 @@ def resolve_passthrough_managed_id_provider(
Splitting them would make a managed ID minted on ``azure`` fail to resolve
when replayed on ``azure_ai`` and vice versa.
"""
provider: Final = str(getattr(custom_llm_provider, "value", custom_llm_provider) or "").lower()
provider_value: Final[object] = getattr(custom_llm_provider, "value", custom_llm_provider)
provider: Final = str(provider_value or "").lower()
if not provider:
return None
if provider in PASSTHROUGH_MANAGED_ID_AZURE_PROVIDERS or provider.endswith((".azure", ".azure_ai")):

3
litellm/llms/base_llm/responses/transformation.py
View file

@ -38,9 +38,10 @@ class BaseResponsesAPIConfig(ABC):
@classmethod
def get_config(cls):
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("__")
and not k.startswith("_abc")
and not isinstance(

6
litellm/llms/base_llm/vector_store_files/transformation.py
View file

@ -53,7 +53,7 @@ class BaseVectorStoreFilesConfig(ABC):
return optional_params
@abstractmethod
def get_auth_credentials(self, litellm_params: dict[str, Any]) -> VectorStoreFileAuthCredentials: ...
def get_auth_credentials(self, litellm_params: Mapping[str, object]) -> VectorStoreFileAuthCredentials: ...
@abstractmethod
def get_vector_store_file_endpoints_by_type(
@ -75,7 +75,7 @@ class BaseVectorStoreFilesConfig(ABC):
*,
api_base: str | None,
vector_store_id: str,
litellm_params: dict[str, Any],
litellm_params: Mapping[str, object],
) -> str:
if api_base is None:
raise ValueError("api_base is required")
@ -183,7 +183,7 @@ class BaseVectorStoreFilesConfig(ABC):
*,
error_message: str,
status_code: int,
headers: dict[str, Any] | httpx.Headers,
headers: dict[str, str] | httpx.Headers,
) -> BaseLLMException:
from ..chat.transformation import BaseLLMException

5
litellm/llms/base_llm/videos/transformation.py
View file

@ -1,7 +1,7 @@
import types
from abc import ABC, abstractmethod
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Final
import httpx
from httpx._types import FileContent, RequestFiles
@ -35,9 +35,10 @@ class BaseVideoConfig(ABC):
@classmethod
def get_config(cls):
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("__")
and not k.startswith("_abc")
and not isinstance(

12
litellm/llms/bedrock/base_aws_llm.py
View file

@ -613,7 +613,10 @@ class BaseAWSLLM(SignsRequestsWithAWS):
if provider is not None:
return provider
for provider in get_args(BEDROCK_INVOKE_PROVIDERS_LITERAL):
invoke_providers: Final[tuple[BEDROCK_INVOKE_PROVIDERS_LITERAL, ...]] = get_args(
BEDROCK_INVOKE_PROVIDERS_LITERAL
)
for provider in invoke_providers:
if provider in model:
return provider
return None
@ -726,9 +729,12 @@ class BaseAWSLLM(SignsRequestsWithAWS):
return cast(BEDROCK_EMBEDDING_PROVIDERS_LITERAL, potential_provider)
# Fallback: check if any provider name appears in the model string
for provider in get_args(BEDROCK_EMBEDDING_PROVIDERS_LITERAL):
embedding_providers: Final[tuple[BEDROCK_EMBEDDING_PROVIDERS_LITERAL, ...]] = get_args(
BEDROCK_EMBEDDING_PROVIDERS_LITERAL
)
for provider in embedding_providers:
if provider in model:
return cast(BEDROCK_EMBEDDING_PROVIDERS_LITERAL, provider)
return provider
return None

9
litellm/llms/bedrock/chat/converse_handler.py
View file

@ -1,5 +1,6 @@
import json
from typing import Any, Final
from collections.abc import Iterator
from typing import Final
import httpx
@ -35,7 +36,7 @@ def make_sync_call(
fake_stream: bool = False,
stream_chunk_size: int | None = None,
timeout: float | httpx.Timeout | None = None,
) -> tuple[Any, httpx.Headers]:
) -> tuple[MockResponseIterator | Iterator[object], httpx.Headers]:
if client is None:
client = get_httpx_client() # Create a new client if none provided
@ -69,7 +70,9 @@ def make_sync_call(
messages=messages,
encoding=litellm.encoding,
)
completion_stream: Any = MockResponseIterator(model_response=model_response, json_mode=json_mode)
completion_stream: MockResponseIterator | Iterator[object] = MockResponseIterator(
model_response=model_response, json_mode=json_mode
)
else:
decoder: Final = AWSEventStreamDecoder(model=model, json_mode=json_mode)
completion_stream = decoder.iter_bytes(

16
litellm/llms/bedrock/chat/converse_transformation.py
View file

@ -145,6 +145,8 @@ UNSUPPORTED_BEDROCK_CONVERSE_BETA_PATTERNS: Final = [
_TOOLS_AS_SENT: Final = TypeAdapter(tuple[Mapping[str, object], ...])
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
_TOOLS_AS_RECEIVED: Final = TypeAdapter(list[object])
CLAUDE_ON_BEDROCK_INVOKE_REMEDY: Final = " This model runs tool search on Bedrock through `bedrock/invoke/<model>`."
_DEPLOYMENT_PARAMS: Final = TypeAdapter(Mapping[str, object])
@ -327,9 +329,10 @@ class AmazonConverseConfig(BaseConfig):
@classmethod
def get_config(cls):
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("_")
and not isinstance(
v,
@ -2586,12 +2589,13 @@ class AmazonConverseConfig(BaseConfig):
Unwrapped JSON string or original if unwrapping not needed
"""
try:
response_data = json.loads(json_str)
if isinstance(response_data, dict) and "properties" in response_data and len(response_data) == 1:
response_data = response_data["properties"]
return json.dumps(response_data)
response_data: Final = _JSON_VALUE.validate_python(json.loads(json_str))
except json.JSONDecodeError:
pass
return json_str
if isinstance(response_data, dict):
json_object: Final = _JSON_OBJECT.validate_python(response_data)
if "properties" in json_object and len(json_object) == 1:
return json.dumps(json_object["properties"])
return json_str
@staticmethod

7
litellm/llms/bedrock/chat/invoke_handler.py
View file

@ -116,16 +116,17 @@ class AmazonCohereChatConfig:
stop_sequences: str | None = None,
raw_prompting: bool | None = None,
) -> None:
locals_: Final = locals().copy()
locals_: Final[Mapping[str, object]] = types.MappingProxyType(locals().copy())
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)
@classmethod
def get_config(cls):
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("__")
and not isinstance(
v,
@ -830,7 +831,7 @@ class AWSEventStreamDecoder:
def iter_bytes(
self, iterator: Iterator[bytes], *, response_headers: Mapping[str, str] | None = None
) -> Iterator[GChunk | ModelResponseStream | dict]:
) -> Iterator[GChunk | ModelResponseStream | dict[str, object]]:
"""Given an iterator that yields lines, iterate over it & yield every event encountered"""
from botocore.eventstream import EventStreamBuffer

4
litellm/llms/bedrock/chat/invoke_transformations/amazon_deepseek_transformation.py
View file

@ -1,4 +1,4 @@
from typing import TYPE_CHECKING, Any, Final, cast
from typing import TYPE_CHECKING, Final, cast
from httpx import Response
@ -80,7 +80,7 @@ class AmazonDeepSeekR1Config(AmazonLlamaConfig):
class AmazonDeepseekR1ResponseIterator(BaseModelResponseIterator):
def __init__(self, streaming_response: Any, sync_stream: bool) -> None:
def __init__(self, streaming_response: object, sync_stream: bool) -> None:
super().__init__(streaming_response=streaming_response, sync_stream=sync_stream)
self.has_finished_thinking = False

5
litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py
View file

@ -20,6 +20,7 @@ from litellm.llms.bedrock.chat.invoke_transformations.base_invoke_transformation
AmazonInvokeConfig,
)
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.custom_httpx.http_handler import header_value
from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import ModelResponse, Usage
from litellm.utils import get_base64_str
@ -245,8 +246,8 @@ class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig):
)
# Calculate usage from headers
bedrock_input_tokens: Final = raw_response.headers.get("x-amzn-bedrock-input-token-count", None)
bedrock_output_tokens: Final = raw_response.headers.get("x-amzn-bedrock-output-token-count", None)
bedrock_input_tokens: Final = header_value(raw_response.headers, "x-amzn-bedrock-input-token-count")
bedrock_output_tokens: Final = header_value(raw_response.headers, "x-amzn-bedrock-output-token-count")
prompt_tokens: Final = int(bedrock_input_tokens or litellm.token_counter(messages=messages))

8
litellm/llms/bedrock/chat/invoke_transformations/base_invoke_transformation.py
View file

@ -56,6 +56,10 @@ _INVOKE_PROVIDERS: Final = TypeAdapter(tuple[litellm.BEDROCK_INVOKE_PROVIDERS_LI
)
def _response_header(headers: Mapping[str, str], name: str) -> str | None:
return headers.get(name, None)
def _bedrock_invoke_guardrail_headers(raw_guardrail_config: object) -> "dict[str, str]":
try:
guardrail_config: Final = _GUARDRAIL_CONFIG_VALIDATOR.validate_python(raw_guardrail_config)
@ -398,8 +402,8 @@ class AmazonInvokeConfig(BaseConfig, BaseAWSLLM):
)
## CALCULATING USAGE - bedrock returns usage in the headers
bedrock_input_tokens: Final = raw_response.headers.get("x-amzn-bedrock-input-token-count", None)
bedrock_output_tokens: Final = raw_response.headers.get("x-amzn-bedrock-output-token-count", None)
bedrock_input_tokens: Final = _response_header(raw_response.headers, "x-amzn-bedrock-input-token-count")
bedrock_output_tokens: Final = _response_header(raw_response.headers, "x-amzn-bedrock-output-token-count")
prompt_tokens: Final = int(bedrock_input_tokens or litellm.token_counter(messages=messages))

9
litellm/llms/bedrock/claude_platform/transformation.py
View file

@ -1,5 +1,5 @@
from collections.abc import AsyncIterator, Iterator, Mapping
from typing import Any, Final
from typing import TYPE_CHECKING, Final
import litellm
from litellm.llms.anthropic.chat.transformation import AnthropicConfig, JsonModeRequest
@ -13,6 +13,9 @@ from .common_utils import (
resolve_unsupported_override,
)
if TYPE_CHECKING:
from litellm.llms.anthropic.chat.handler import ModelResponseIterator
class BedrockClaudePlatformConfig(BedrockClaudePlatformMixin, AnthropicConfig):
"""
@ -96,7 +99,7 @@ class BedrockClaudePlatformConfig(BedrockClaudePlatformMixin, AnthropicConfig):
streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
sync_stream: bool,
json_mode: bool | None = False,
) -> Any:
) -> "ModelResponseIterator":
return self.get_model_response_iterator_for_request(
streaming_response=streaming_response, sync_stream=sync_stream, json_mode=json_mode, optional_params={}
)
@ -107,7 +110,7 @@ class BedrockClaudePlatformConfig(BedrockClaudePlatformMixin, AnthropicConfig):
sync_stream: bool,
json_mode: bool | None,
optional_params: Mapping[str, object],
) -> object:
) -> "ModelResponseIterator":
from litellm.llms.anthropic.chat.handler import ModelResponseIterator
return ModelResponseIterator(

3
litellm/llms/bedrock/image_edit/amazon_nova_canvas_image_edit_transformation.py
View file

@ -17,6 +17,7 @@ import os
from typing import TYPE_CHECKING, Any, Final, cast
import httpx
from httpx._types import RequestFiles
from litellm._logging import verbose_logger
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
@ -319,7 +320,7 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
image_edit_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[dict, Any]:
) -> tuple[dict, RequestFiles]:
op: Final = dict(image_edit_optional_request_params)
image_b64: Final = _file_types_to_b64(image)

4
litellm/llms/bedrock/image_generation/amazon_nova_canvas_transformation.py
View file

@ -1,4 +1,5 @@
import types
from collections.abc import Mapping
from typing import Any, Final
from openai.types.image import Image
@ -26,9 +27,10 @@ class AmazonNovaCanvasConfig:
@classmethod
def get_config(cls):
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("__")
and not isinstance(
v,

6
litellm/llms/bedrock/image_generation/amazon_stability1_transformation.py
View file

@ -1,6 +1,7 @@
import copy
import os
import types
from collections.abc import Mapping
from typing import Final
from openai.types.image import Image
@ -52,16 +53,17 @@ class AmazonStabilityConfig:
width: int | None = None,
height: int | None = None,
) -> None:
locals_: Final = locals().copy()
locals_: Final[Mapping[str, object]] = types.MappingProxyType(locals().copy())
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)
@classmethod
def get_config(cls):
class_attributes: Final[Mapping[str, object]] = types.MappingProxyType(cls.__dict__)
return {
k: v
for k, v in cls.__dict__.items()
for k, v in class_attributes.items()
if not k.startswith("__")
and not isinstance(
v,

4
litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py
View file

@ -103,7 +103,7 @@ class AmazonAnthropicClaudeMessagesConfig(
self,
headers: dict,
model: str,
messages: list[Any],
messages: list[object],
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,
@ -878,7 +878,7 @@ class AmazonAnthropicClaudeMessagesConfig(
chunk_type = chunk.get("type")
if chunk_type == "message_start":
msg: dict[str, object] = cast(dict[str, Any], chunk.get("message") or {})
msg: dict[str, object] = cast(dict[str, object], chunk.get("message") or {})
u = msg.get("usage")
if isinstance(u, dict):
start_usage_snapshot = dict(u)

4
litellm/llms/bedrock/passthrough/transformation.py
View file

@ -1,5 +1,6 @@
import json
from collections.abc import Callable, Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Optional, cast
import httpx
@ -39,7 +40,8 @@ def _plain_text_delta(chunk: ModelResponseStream) -> str | None:
choice: Final = chunk.choices[0]
if choice.finish_reason or choice.logprobs is not None:
return None
populated: Final = frozenset(key for key, value in choice.delta.model_dump().items() if value is not None)
delta_fields: Final[Mapping[str, object]] = MappingProxyType(choice.delta.model_dump())
populated: Final = frozenset(key for key, value in delta_fields.items() if value is not None)
if not populated <= _TEXT_ONLY_DELTA_FIELDS:
return None
content: Final = choice.delta.get("content")

9
litellm/llms/bedrock/realtime/transformation.py
View file

@ -7,9 +7,11 @@ Transforms between OpenAI Realtime API format and Bedrock Nova Sonic format.
import base64
import json
import uuid as uuid_lib
from collections.abc import Mapping
from typing import Final, cast
import httpx
from pydantic import TypeAdapter
from litellm._logging import verbose_logger
from litellm._uuid import uuid
@ -44,6 +46,9 @@ from litellm.types.realtime import (
RealtimeResponseTypedDict,
)
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
class BedrockContentEnd(LiteLLMBaseModel):
stopReason: str | None = None
@ -839,10 +844,10 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
if not isinstance(additional_model_fields, str):
return None
try:
parsed: Final = json.loads(additional_model_fields)
parsed: Final = _JSON_VALUE.validate_python(json.loads(additional_model_fields))
except json.JSONDecodeError:
return None
stage: Final = parsed.get("generationStage") if isinstance(parsed, dict) else None
stage: Final = _JSON_OBJECT.validate_python(parsed).get("generationStage") if isinstance(parsed, dict) else None
return stage if isinstance(stage, str) else None
def _current_user_item_id(self, new_utterance: bool = False) -> str:

12
litellm/llms/bedrock_mantle/chat/transformation.py
View file

@ -11,7 +11,7 @@ Auth: Bearer token (litellm_params.api_key, BEDROCK_MANTLE_API_KEY, or the
"""
from collections.abc import AsyncIterator, Iterator
from typing import Any, Final
from typing import TYPE_CHECKING, Final
import httpx
@ -26,12 +26,18 @@ from litellm.llms.openai.chat.gpt_5_transformation import is_gpt_reasoning_serie
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllMessageValues
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import ModelResponse
from ...base_llm.chat.transformation import BaseLLMException
from ...bedrock.common_utils import BedrockError
from ...openai_like.chat.transformation import OpenAILikeChatConfig
from ..common_utils import mantle_base_segment, split_mantle_region_prefix
if TYPE_CHECKING:
from litellm.llms.openai.chat.gpt_transformation import (
OpenAIChatCompletionStreamingHandler,
)
class BedrockMantleChatConfig(BedrockMantleAuthMixin, OpenAILikeChatConfig):
"""
@ -161,10 +167,10 @@ class BedrockMantleChatConfig(BedrockMantleAuthMixin, OpenAILikeChatConfig):
def get_model_response_iterator(
self,
streaming_response: Iterator[str] | AsyncIterator[str] | Any,
streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
sync_stream: bool,
json_mode: bool | None = False,
) -> Any:
) -> "OpenAIChatCompletionStreamingHandler":
from litellm.llms.openai.chat.gpt_transformation import (
OpenAIChatCompletionStreamingHandler,
)

Some files were not shown because too many files have changed in this diff Show more