scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-19 00:01:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

refactor: replace Any with precise types across 54 modules

Browse source 
Narrow or remove reportAny / reportExplicitAny sites in provider
transformations, caching, guardrails, proxy endpoints and enterprise
batch-cost polling. Public parameters widen to Mapping/Sequence rather
than dict/list so no caller signature breaks, and runtime behavior is
unchanged.
This commit is contained in:
mateo-berri 2026-09-14 11:13:39 +00:00
parent e46c816ec7
commit da7d5fe128
54 changed files with 485 additions and 336 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

127
enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py
View file

@ -2,10 +2,11 @@
Polls LiteLLM_ManagedObjectTable to check if the batch job is complete, and if the cost has been tracked.
"""
from collections.abc import Sequence
from dataclasses import replace as dataclasses_replace
from datetime import datetime, timedelta, timezone
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, List, Literal, Optional, Tuple, cast
from typing import TYPE_CHECKING, Final, List, Literal, Optional, Protocol, Tuple, cast
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
@ -18,8 +19,8 @@ if TYPE_CHECKING:
from prisma import models as prisma_models
from litellm.integrations.prometheus import PrometheusLogger
from litellm.proxy._types import LiteLLM_ManagedObjectTable
from litellm.proxy.utils import PrismaClient, ProxyLogging
from litellm.repositories.prisma_protocols import TableActions
from litellm.router import Router
from litellm.types.router import Deployment
from litellm.types.utils import LiteLLMBatch
@ -41,6 +42,48 @@ TERMINAL_MANAGED_OBJECT_STATUSES: Final[Tuple[str, ...]] = (
)
class _ManagedObjectRow(Protocol):
"""The managed-object row fields this poller reads off whatever the DB hands back."""
@property
def id(self) -> str: ...
@property
def unified_object_id(self) -> str: ...
@property
def created_by(self) -> str | None: ...
@property
def file_object(self) -> object: ...
def _managed_object_table(prisma_client: "PrismaClient") -> "TableActions[_ManagedObjectRow]":
"""The managed-object table's prisma actions, typed to the row fields this module reads."""
table: Final[TableActions[_ManagedObjectRow]] = prisma_client.db.litellm_managedobjecttable
return table
def _user_table(prisma_client: "PrismaClient") -> "TableActions[prisma_models.LiteLLM_UserTable]":
"""The user table's prisma actions."""
table: Final[TableActions[prisma_models.LiteLLM_UserTable]] = prisma_client.db.litellm_usertable
return table
def _token_table(prisma_client: "PrismaClient") -> "TableActions[prisma_models.LiteLLM_VerificationToken]":
"""The virtual-key table's prisma actions."""
table: Final[TableActions[prisma_models.LiteLLM_VerificationToken]] = (
prisma_client.db.litellm_verificationtoken
)
return table
def _team_table(prisma_client: "PrismaClient") -> "TableActions[prisma_models.LiteLLM_TeamTable]":
"""The team table's prisma actions."""
table: Final[TableActions[prisma_models.LiteLLM_TeamTable]] = prisma_client.db.litellm_teamtable
return table
class CheckBatchCost:
def __init__(
self,
@ -73,7 +116,7 @@ class CheckBatchCost:
inline for a batch the first poll cycle then accounts again.
"""
try:
await self.prisma_client.db.litellm_managedobjecttable.find_first(
await _managed_object_table(self.prisma_client).find_first(
where={"file_purpose": "batch", "batch_processed": False}
)
except Exception as probe_err:
@ -97,10 +140,8 @@ class CheckBatchCost:
if not user_id:
return {}
try:
user_row: prisma_models.LiteLLM_UserTable | None = (
await self.prisma_client.db.litellm_usertable.find_unique(
where={"user_id": user_id}
)
user_row: prisma_models.LiteLLM_UserTable | None = await _user_table(self.prisma_client).find_unique(
where={"user_id": user_id}
)
if user_row is None:
return {}
@ -117,11 +158,9 @@ class CheckBatchCost:
if not api_key:
return None
try:
key_row: prisma_models.LiteLLM_VerificationToken | None = (
await self.prisma_client.db.litellm_verificationtoken.find_unique(
where={"token": api_key}
)
)
key_row: prisma_models.LiteLLM_VerificationToken | None = await _token_table(
self.prisma_client
).find_unique(where={"token": api_key})
return getattr(key_row, "key_alias", None) if key_row is not None else None
except Exception as e:
verbose_proxy_logger.error(f"CheckBatchCost: could not look up key alias for batch {batch_id}: {e}")
@ -132,17 +171,15 @@ class CheckBatchCost:
if not team_id:
return None
try:
team_row: prisma_models.LiteLLM_TeamTable | None = (
await self.prisma_client.db.litellm_teamtable.find_unique(
where={"team_id": team_id}
)
team_row: prisma_models.LiteLLM_TeamTable | None = await _team_table(self.prisma_client).find_unique(
where={"team_id": team_id}
)
return getattr(team_row, "team_alias", None) if team_row is not None else None
except Exception as e:
verbose_proxy_logger.error(f"CheckBatchCost: could not look up team alias for team {team_id}: {e}")
return None
async def _get_org_id(self, job: "LiteLLM_ManagedObjectTable", batch_id: str) -> str | None:
async def _get_org_id(self, job: "_ManagedObjectRow", batch_id: str) -> str | None:
org_id = getattr(job, "org_id", None)
if org_id:
return org_id
@ -150,11 +187,9 @@ class CheckBatchCost:
team_id = getattr(job, "team_id", None)
if api_key:
try:
key_row: prisma_models.LiteLLM_VerificationToken | None = (
await self.prisma_client.db.litellm_verificationtoken.find_unique(
where={"token": api_key}
)
)
key_row: prisma_models.LiteLLM_VerificationToken | None = await _token_table(
self.prisma_client
).find_unique(where={"token": api_key})
key_org_id = getattr(key_row, "organization_id", None) if key_row is not None else None
if key_org_id:
return key_org_id
@ -166,10 +201,8 @@ class CheckBatchCost:
if not team_id:
return None
try:
team_row: prisma_models.LiteLLM_TeamTable | None = (
await self.prisma_client.db.litellm_teamtable.find_unique(
where={"team_id": team_id}
)
team_row: prisma_models.LiteLLM_TeamTable | None = await _team_table(self.prisma_client).find_unique(
where={"team_id": team_id}
)
return getattr(team_row, "organization_id", None) if team_row is not None else None
except Exception as e:
@ -177,7 +210,7 @@ class CheckBatchCost:
return None
async def _build_creator_attribution_metadata(
self, job: "LiteLLM_ManagedObjectTable", batch_id: str
self, job: "_ManagedObjectRow", batch_id: str
) -> dict[str, object]:
"""
Rebuild the spend-tracking metadata for the key, team, and tags that created the
@ -225,7 +258,7 @@ class CheckBatchCost:
should not be polled.
"""
cutoff: Final = datetime.now(timezone.utc) - timedelta(days=MANAGED_OBJECT_STALENESS_CUTOFF_DAYS)
result: Final = await self.prisma_client.db.litellm_managedobjecttable.update_many(
result: Final = await _managed_object_table(self.prisma_client).update_many(
where={
"file_purpose": "batch",
"status": {"not_in": list(TERMINAL_MANAGED_OBJECT_STATUSES)},
@ -244,7 +277,7 @@ class CheckBatchCost:
# A row already in a terminal status is never rewritten by the sweep above, so
# without this it keeps a poll-page slot forever and starves newer batches.
retired: Final = await self.prisma_client.db.litellm_managedobjecttable.update_many(
retired: Final = await _managed_object_table(self.prisma_client).update_many(
where={
"file_purpose": "batch",
"batch_processed": False,
@ -259,9 +292,9 @@ class CheckBatchCost:
f"{MANAGED_OBJECT_STALENESS_CUTOFF_DAYS} days that were never costed"
)
async def _fallback_find_jobs(self) -> list:
async def _fallback_find_jobs(self) -> "Sequence[_ManagedObjectRow]":
"""Query batch jobs without the batch_processed filter (for older schemas)."""
return await self.prisma_client.db.litellm_managedobjecttable.find_many(
return await _managed_object_table(self.prisma_client).find_many(
where={
"file_purpose": "batch",
"status": {
@ -279,7 +312,7 @@ class CheckBatchCost:
order={"created_at": "asc"},
)
async def _retire_job(self, job: "LiteLLM_ManagedObjectTable", reason: str) -> None:
async def _retire_job(self, job: "_ManagedObjectRow", reason: str) -> None:
"""
Take a row that can never be costed out of the poll page. Leaving it selectable
would burn one of the MAX_OBJECTS_PER_POLL_CYCLE slots on every future cycle, and
@ -292,7 +325,7 @@ class CheckBatchCost:
else {"status": "stale_expired"}
)
try:
await self.prisma_client.db.litellm_managedobjecttable.update(
await _managed_object_table(self.prisma_client).update(
where={"id": job.id},
data=data,
)
@ -306,7 +339,7 @@ class CheckBatchCost:
"so it will no longer be polled"
)
async def _claim_job_for_costing(self, job: "LiteLLM_ManagedObjectTable") -> bool:
async def _claim_job_for_costing(self, job: "_ManagedObjectRow") -> bool:
"""
Atomically flip batch_processed from false to true, returning whether this pod won
the row. Every pod and uvicorn worker schedules its own poller against the shared
@ -321,7 +354,7 @@ class CheckBatchCost:
if not self._has_batch_processed_column:
return True
try:
claimed: Final = await self.prisma_client.db.litellm_managedobjecttable.update_many(
claimed: Final = await _managed_object_table(self.prisma_client).update_many(
where={"id": job.id, "batch_processed": False},
data={"batch_processed": True},
)
@ -332,7 +365,7 @@ class CheckBatchCost:
return False
return claimed > 0
async def _release_job_claim(self, job: "LiteLLM_ManagedObjectTable") -> None:
async def _release_job_claim(self, job: "_ManagedObjectRow") -> None:
"""Give a claimed row back once billing it failed, so a later poll cycle retries it.
Safe to match on batch_processed=True: while this poller is active the retrieve
@ -342,7 +375,7 @@ class CheckBatchCost:
if not self._has_batch_processed_column:
return
try:
await self.prisma_client.db.litellm_managedobjecttable.update_many(
await _managed_object_table(self.prisma_client).update_many(
where={"id": job.id, "batch_processed": True},
data={"batch_processed": False},
)
@ -353,7 +386,7 @@ class CheckBatchCost:
)
@staticmethod
def _has_unified_id_without_model(job: "LiteLLM_ManagedObjectTable") -> bool:
def _has_unified_id_without_model(job: "_ManagedObjectRow") -> bool:
"""A unified id that decodes but carries no model_id can never be routed."""
from litellm.proxy.openai_files_endpoints.common_utils import (
convert_b64_uid_to_unified_uid,
@ -402,7 +435,7 @@ class CheckBatchCost:
return isinstance(error, (NotFoundError, openai.NotFoundError)) and output_file_id in str(error)
async def _finalize_unbilled_terminal_job(
self, job: "prisma_models.LiteLLM_ManagedObjectTable", response: "LiteLLMBatch"
self, job: "_ManagedObjectRow", response: "LiteLLMBatch"
) -> None:
"""Persist a terminal batch that has nothing billable, converting any raw
provider file ids to managed ids, and take it out of the poll page."""
@ -426,7 +459,7 @@ class CheckBatchCost:
"file_object": response.model_dump_json(),
**({"batch_processed": True} if self._has_batch_processed_column else {}),
}
await self.prisma_client.db.litellm_managedobjecttable.update(
await _managed_object_table(self.prisma_client).update(
where={"id": job.id},
data=update_data,
)
@ -447,7 +480,7 @@ class CheckBatchCost:
def _resolve_job_routing(
self,
job: "LiteLLM_ManagedObjectTable",
job: "_ManagedObjectRow",
prom_logger: Optional["PrometheusLogger"],
) -> Optional[Tuple[str, str]]:
"""
@ -524,7 +557,7 @@ class CheckBatchCost:
def _resolve_unmanaged_provider_routing(
self,
job: "LiteLLM_ManagedObjectTable",
job: "_ManagedObjectRow",
prom_logger: Optional["PrometheusLogger"],
llm_provider: str,
bare_model_name: str,
@ -620,7 +653,7 @@ class CheckBatchCost:
@classmethod
def _get_managed_file_model_name(
cls,
job: "LiteLLM_ManagedObjectTable",
job: "_ManagedObjectRow",
deployment_info: "Deployment",
) -> Optional[str]:
"""
@ -640,7 +673,7 @@ class CheckBatchCost:
)
@staticmethod
def _get_input_file_id(job: "LiteLLM_ManagedObjectTable") -> Optional[str]:
def _get_input_file_id(job: "_ManagedObjectRow") -> Optional[str]:
import json
from litellm.types.utils import LiteLLMBatch
@ -660,7 +693,7 @@ class CheckBatchCost:
async def _track_completed_batch_cost(
self,
job: "LiteLLM_ManagedObjectTable",
job: "_ManagedObjectRow",
response: "LiteLLMBatch",
model_id: str,
batch_id: str,
@ -936,7 +969,7 @@ class CheckBatchCost:
# endpoint may transition a batch to "complete" before
# CheckBatchCost runs. The batch_processed=False filter
# already prevents reprocessing finished batches.
jobs = await self.prisma_client.db.litellm_managedobjecttable.find_many(
jobs = await _managed_object_table(self.prisma_client).find_many(
where={
"file_purpose": "batch",
"batch_processed": False,
@ -1038,7 +1071,7 @@ class CheckBatchCost:
}
if self._has_batch_processed_column:
update_data["batch_processed"] = True
await self.prisma_client.db.litellm_managedobjecttable.update(
await _managed_object_table(self.prisma_client).update(
where={"id": job.id},
data=update_data,
)

31
enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py
View file

@ -6,7 +6,7 @@ same route are non-inference and free.
"""
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Dict, Optional, cast
from typing import TYPE_CHECKING, Dict, Final, Optional, Protocol, cast
import litellm
from litellm._logging import verbose_proxy_logger
@ -22,11 +22,34 @@ from litellm.types.utils import BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN
if TYPE_CHECKING:
from litellm.proxy.utils import PrismaClient, ProxyLogging
from litellm.repositories.prisma_protocols import TableActions
from litellm.router import Router
TERMINAL_RESPONSE_STATUSES = frozenset({"completed", "failed", "cancelled", "incomplete"})
class _ManagedObjectRow(Protocol):
"""The managed-object row fields this poller reads off whatever the DB hands back."""
@property
def id(self) -> str: ...
@property
def unified_object_id(self) -> str: ...
@property
def created_by(self) -> str | None: ...
@property
def file_object(self) -> object: ...
def _managed_object_table(prisma_client: "PrismaClient") -> "TableActions[_ManagedObjectRow]":
"""The managed-object table's prisma actions, typed to the row fields this poller reads."""
table: Final[TableActions[_ManagedObjectRow]] = prisma_client.db.litellm_managedobjecttable
return table
class CheckResponsesCost:
def __init__(
self,
@ -128,7 +151,7 @@ class CheckResponsesCost:
f"CheckResponsesCost: stale cleanup failed (poll will continue): {cleanup_err}"
)
jobs = await self.prisma_client.db.litellm_managedobjecttable.find_many(
jobs = await _managed_object_table(self.prisma_client).find_many(
where={
"status": {"in": ["queued", "in_progress"]},
"file_purpose": "response",
@ -138,7 +161,7 @@ class CheckResponsesCost:
)
verbose_proxy_logger.debug(f"Found {len(jobs)} response jobs to check")
completed_jobs = []
completed_jobs: Final[list[_ManagedObjectRow]] = []
for job in jobs:
unified_object_id = job.unified_object_id
@ -189,7 +212,7 @@ class CheckResponsesCost:
# Mark completed jobs in the database
if len(completed_jobs) > 0:
await self.prisma_client.db.litellm_managedobjecttable.update_many(
await _managed_object_table(self.prisma_client).update_many(
where={"id": {"in": [job.id for job in completed_jobs]}},
data={"status": "completed"},
)

22
enterprise/litellm_enterprise/proxy/hooks/managed_files.py
View file

@ -465,10 +465,8 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
"""
if self.prisma_client is None:
return
managed_object = (
await self.prisma_client.db.litellm_managedobjecttable.find_first(
where={"OR": [{"unified_object_id": object_id}, {"model_object_id": object_id}]}
)
managed_object = await _managed_object_table(self.prisma_client).find_first(
where={"OR": [{"unified_object_id": object_id}, {"model_object_id": object_id}]}
)
if managed_object is None:
return
@ -493,10 +491,8 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
"""
if self.prisma_client is None:
return
managed_file = (
await self.prisma_client.db.litellm_managedfiletable.find_first(
where={"OR": [{"unified_file_id": file_id}, {"flat_model_file_ids": {"has": file_id}}]}
)
managed_file = await _managed_file_table(self.prisma_client).find_first(
where={"OR": [{"unified_file_id": file_id}, {"flat_model_file_ids": {"has": file_id}}]}
)
if managed_file is None:
return
@ -519,8 +515,8 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
provider_file_ids = tuple(
file_id
for file_id in (
getattr(response, "output_file_id", None),
getattr(response, "error_file_id", None),
response.output_file_id,
response.error_file_id,
)
if file_id and not _is_base64_encoded_unified_file_id(file_id)
)
@ -528,10 +524,8 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
return
if self.prisma_client is None:
return
batch_row = (
await self.prisma_client.db.litellm_managedobjecttable.find_first(
where={"unified_object_id": response.id}
)
batch_row = await _managed_object_table(self.prisma_client).find_first(
where={"unified_object_id": response.id}
)
if batch_row is None or (
batch_row.created_by is None and batch_row.team_id is None

2
litellm/caching/caching.py
View file

@ -81,7 +81,7 @@ class Cache:
s3_aws_access_key_id: str | None = None,
s3_aws_secret_access_key: str | None = None,
s3_aws_session_token: str | None = None,
s3_config: Any | None = None,
s3_config: object | None = None,
s3_path: str | None = None,
gcs_bucket_name: str | None = None,
gcs_path_service_account: str | None = None,

10
litellm/caching/caching_handler.py
View file

@ -73,7 +73,7 @@ class CachingHandlerResponse(BaseModel):
For embeddings there can be a cache hit for some of the inputs in the list and a cache miss for others
"""
cached_result: Any | None = None
cached_result: object | None = None
final_embedding_cached_response: EmbeddingResponse | None = None
embedding_all_elements_cache_hit: bool = False # this is set to True when all elements in the list have a cache hit in the embedding cache, if true return the final_embedding_cached_response no need to make an API call
@ -707,7 +707,7 @@ class LLMCachingHandler:
async def _retrieve_from_cache(
self, call_type: str, kwargs: dict[str, object], args: tuple[object, ...]
) -> Any | None:
) -> object | None:
"""
Internal method to
- get cache key
@ -953,7 +953,7 @@ class LLMCachingHandler:
def _convert_cached_stream_response(
self,
cached_result: Any,
cached_result: dict[str, object],
call_type: str,
logging_obj: LiteLLMLoggingObj,
model: str,
@ -982,7 +982,7 @@ class LLMCachingHandler:
async def async_set_cache(
self,
result: Any,
result: object,
original_function: Callable,
kwargs: dict[str, Any],
args: tuple[object, ...] | None = None,
@ -1050,7 +1050,7 @@ class LLMCachingHandler:
def sync_set_cache(
self,
result: Any,
result: object,
kwargs: dict[str, object],
args: tuple[object, ...] | None = None,
):

4
litellm/caching/redis_cache.py
View file

@ -934,7 +934,7 @@ class RedisCache(BaseCache):
client: object = None,
) -> object:
async def execute() -> object:
executor: Callable[..., Awaitable[Any]] | None = litellm.in_memory_llm_clients_cache.get_cache(
executor: Callable[..., Awaitable[object]] | None = litellm.in_memory_llm_clients_cache.get_cache(
key=script_cache_key
)
if executor is None:
@ -946,7 +946,7 @@ class RedisCache(BaseCache):
return run_script
def _register_script_for_current_loop(self, script: str) -> Callable[..., Awaitable[Any]]:
def _register_script_for_current_loop(self, script: str) -> Callable[..., Awaitable[object]]:
"""
Register the script against the current event loop's Redis client.

6
litellm/completion_extras/litellm_responses_transformation/handler.py
View file

@ -2,7 +2,7 @@
Handler for transforming /chat/completions api requests to litellm.responses requests
"""
from collections.abc import Coroutine
from collections.abc import AsyncIterable, Coroutine, Iterable
from typing import TYPE_CHECKING, Any, Final, Union
from typing_extensions import TypedDict
@ -74,7 +74,7 @@ class ResponsesToCompletionBridgeHandler:
existing.setdefault(key, value)
return response
def _collect_response_from_stream(self, stream_iter: Any) -> "ResponsesAPIResponse":
def _collect_response_from_stream(self, stream_iter: Iterable[object]) -> "ResponsesAPIResponse":
for _ in stream_iter:
pass
@ -89,7 +89,7 @@ class ResponsesToCompletionBridgeHandler:
raise ValueError("Stream completed response is invalid")
return response
async def _collect_response_from_stream_async(self, stream_iter: Any) -> "ResponsesAPIResponse":
async def _collect_response_from_stream_async(self, stream_iter: AsyncIterable[object]) -> "ResponsesAPIResponse":
async for _ in stream_iter:
pass

30
litellm/completion_extras/litellm_responses_transformation/transformation.py
View file

@ -6,7 +6,7 @@ import json
import os
from collections.abc import AsyncIterator, Callable, Iterable, Iterator, Mapping, Sequence
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict, Union, cast, get_args
from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict, TypeVar, Union, cast, get_args
from openai.types.chat import ChatCompletion
from openai.types.responses import Response
@ -52,7 +52,7 @@ from litellm.types.llms.openai import (
from litellm.types.utils import GenericStreamingChunk, ModelResponseStream
if TYPE_CHECKING:
from openai.types.responses import ResponseInputImageParam
from openai.types.responses import ResponseInputImageParam, ResponseOutputItem
from openai.types.responses.response_text_config_param import (
ResponseTextConfigParam as ResponseText,
)
@ -197,6 +197,9 @@ def _as_chat_reasoning_items(
return cast(list[ChatCompletionReasoningItem], list(reasoning_items))
_ToolChoiceT = TypeVar("_ToolChoiceT")
def _map_incomplete_reason_to_finish_reason(incomplete_reason: str | None) -> Literal["length", "content_filter"]:
if incomplete_reason == "content_filter":
return "content_filter"
@ -291,7 +294,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
def __init__(self):
pass
def _normalize_tool_choice_for_responses_api(self, tool_choice: Any) -> Any:
def _normalize_tool_choice_for_responses_api(
self, tool_choice: _ToolChoiceT
) -> _ToolChoiceT | ToolChoiceFunctionParam | ToolChoiceCustomParam | Literal["auto", "none", "required"]:
"""Chat tool_choice nests the name under function/custom; Responses API expects top-level name."""
if not isinstance(tool_choice, dict):
return tool_choice
@ -497,7 +502,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
responses_api_request["max_output_tokens"] = value
elif key == "tools" and value is not None:
responses_api_request["tools"] = self._convert_tools_to_responses_format(
cast(list[dict[str, Any]], value)
cast(list[dict[str, object]], value)
)
elif key == "response_format":
text_format = self._transform_response_format_to_text_format(value)
@ -810,7 +815,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
response_output: Final = response_payload.get("output")
if not isinstance(response_output, list) or len(response_output) == 0:
return None
return cast(list[dict[str, Any]], response_output)
return cast(list[dict[str, object]], response_output)
@classmethod
def _recover_output_items_from_raw_sse(cls, raw_sse: str | None) -> list[dict[str, object]]:
@ -893,10 +898,12 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
output_items = raw_response.output
if len(output_items) == 0:
recovered_output_items: Final = self._recover_output_items_from_logging(logging_obj)
recovered_output_items: Final[list[ResponseOutputItem | dict[str, object]]] = [
*self._recover_output_items_from_logging(logging_obj)
]
if recovered_output_items:
output_items = cast(Any, recovered_output_items)
raw_response.output = cast(Any, recovered_output_items)
output_items = recovered_output_items
raw_response.output = recovered_output_items
verbose_logger.warning(
"Recovered empty Responses API output from raw SSE for model=%s",
model,
@ -1092,7 +1099,9 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
verbose_logger.debug("Chat provider: Other content type -> %s", result)
return result
def _convert_tools_to_responses_format(self, tools: list[dict[str, Any]]) -> list["ALL_RESPONSES_API_TOOL_PARAMS"]:
def _convert_tools_to_responses_format(
self, tools: list[dict[str, object]]
) -> list["ALL_RESPONSES_API_TOOL_PARAMS"]:
"""Convert chat completion tools to responses API tools format"""
responses_tools: Final[list[ALL_RESPONSES_API_TOOL_PARAMS]] = []
for tool in tools:
@ -1108,12 +1117,11 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
description=function_tool.get("description"),
)
)
elif tool.get("type") == "custom" and isinstance(tool.get("custom"), dict):
elif tool.get("type") == "custom" and isinstance(custom_payload := tool.get("custom"), dict):
from litellm.litellm_core_utils.prompt_templates.common_utils import (
convert_custom_tool_format_to_responses_shape,
)
custom_payload = tool["custom"]
flat_custom = CustomToolParam(type="custom", name=custom_payload.get("name", ""))
if custom_payload.get("description") is not None:
flat_custom["description"] = custom_payload["description"]

6
litellm/cost_calculator.py
View file

@ -351,7 +351,7 @@ def cost_per_token(
data_residency: str | None = None, # for OpenAI regional-processing uplift (e.g. "eu", "us")
### VERTEX LOCATION ###
vertex_location: str | None = None, # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global")
response: Any | None = None,
response: object | None = None,
### REQUEST MODEL ###
request_model: str | None = None, # original request model for router detection
custom_model_info: OCRPricing | None = None,
@ -607,7 +607,7 @@ def cost_per_token(
model=model,
custom_llm_provider=custom_llm_provider,
number_of_queries=number_of_queries or 1,
optional_params=(response._hidden_params if response and hasattr(response, "_hidden_params") else None),
optional_params=(getattr(response, "_hidden_params", None) if response else None),
)
elif custom_llm_provider == "vertex_ai":
cost_router: Final = google_cost_router(
@ -996,7 +996,7 @@ def _is_known_usage_objects(usage_obj):
)
def _infer_call_type(call_type: CallTypesLiteral | None, completion_response: Any) -> CallTypesLiteral | None:
def _infer_call_type(call_type: CallTypesLiteral | None, completion_response: object) -> CallTypesLiteral | None:
if call_type is not None:
return call_type

8
litellm/integrations/braintrust_logging.py
View file

@ -139,13 +139,13 @@ class BraintrustLogger(CustomLogger):
):
output = None
elif response_obj is not None and isinstance(response_obj, litellm.ModelResponse):
output = response_obj["choices"][0]["message"].json()
output = response_obj.choices[0].message.json()
choices = response_obj["choices"]
elif response_obj is not None and isinstance(response_obj, litellm.TextCompletionResponse):
output = response_obj.choices[0].text
choices = response_obj.choices
elif response_obj is not None and isinstance(response_obj, litellm.ImageResponse):
output = response_obj["data"]
output = response_obj.data
litellm_params: Final = kwargs.get("litellm_params", {}) or {}
dynamic_metadata: Final = litellm_params.get("metadata", {}) or {}
@ -264,13 +264,13 @@ class BraintrustLogger(CustomLogger):
):
output = None
elif response_obj is not None and isinstance(response_obj, litellm.ModelResponse):
output = response_obj["choices"][0]["message"].json()
output = response_obj.choices[0].message.json()
choices = response_obj["choices"]
elif response_obj is not None and isinstance(response_obj, litellm.TextCompletionResponse):
output = response_obj.choices[0].text
choices = response_obj.choices
elif response_obj is not None and isinstance(response_obj, litellm.ImageResponse):
output = response_obj["data"]
output = response_obj.data
litellm_params: Final = kwargs.get("litellm_params", {})
dynamic_metadata: Final = litellm_params.get("metadata", {}) or {}

2
litellm/integrations/custom_guardrail.py
View file

@ -155,7 +155,7 @@ class CustomGuardrail(CustomLogger):
def __init_subclass__(cls, **kwargs: object) -> None: # kwargs-ok: forwarded to cooperative __init_subclass__ hooks
super().__init_subclass__(**kwargs)
own_apply_guardrail: Final = cls.__dict__.get("apply_guardrail")
own_apply_guardrail: Final[object] = cls.__dict__.get("apply_guardrail")
if own_apply_guardrail is None or LOGS_GUARDRAIL_INFORMATION_MARKER in vars(own_apply_guardrail):
return
cls.apply_guardrail = log_guardrail_information(own_apply_guardrail)

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

@ -54,7 +54,7 @@ from litellm.types.utils import (
StandardLoggingPayloadErrorInformation,
)
_EMPTY_MAPPING: Final[Mapping[str, Any]] = MappingProxyType({})
_EMPTY_MAPPING: Final[Mapping[str, object]] = MappingProxyType({})
_EMPTY_MESSAGE: Final[Message] = {"role": "", "content": ""}
_MAX_PARSED_TOOL_ARGUMENT_CHARS: Final = 256 * 1024
_SAFE_REDACTED_MESSAGE_ROLES: Final = frozenset(
@ -154,7 +154,7 @@ def _guardrail_information_without_prompt_carriers(
return tuple(_guardrail_entry_without_prompt_carriers(entry) for entry in _guardrail_entries(guardrail_information))
def _metadata_without_prompt_carriers(standard_logging_metadata: Mapping[str, Any]) -> Mapping[str, Any]:
def _metadata_without_prompt_carriers(standard_logging_metadata: Mapping[str, object]) -> Mapping[str, object]:
"""The metadata minus the records that quote prompts, tool arguments, tool results, or retrieved text."""
return MappingProxyType(
{
@ -237,7 +237,7 @@ def _declared_cost_tags(span_tags: Sequence[str]) -> tuple[str, ...]:
return tuple(dimension for dimension in _COST_DIMENSIONS if dimension in present)
def _reasoning_output_tokens(usage_object: Mapping[str, Any] | None) -> float:
def _reasoning_output_tokens(usage_object: Mapping[str, object] | None) -> float:
"""The provider's reasoning-token count, from either the chat or the responses spelling."""
if usage_object is None:
return 0.0
@ -254,20 +254,20 @@ def _reasoning_output_tokens(usage_object: Mapping[str, Any] | None) -> float:
)
def _mapping_field(source: Mapping[str, Any], key: str) -> Mapping[str, Any]:
def _mapping_field(source: Mapping[str, object], key: str) -> Mapping[str, Any]:
"""The value at `key` when it is a mapping, else an empty one."""
value: Final = source.get(key)
return value if isinstance(value, dict) else _EMPTY_MAPPING
def _content_blocks(message: Mapping[str, Any]) -> tuple[Mapping[str, Any], ...]:
def _content_blocks(message: Mapping[str, object]) -> tuple[Mapping[str, Any], ...]:
content: Final = message.get("content")
if not isinstance(content, list):
return ()
return tuple(block for block in content if isinstance(block, dict))
def _to_dd_arguments(raw_arguments: object) -> dict[str, Any] | str:
def _to_dd_arguments(raw_arguments: object) -> dict[str, object] | str:
"""
Arguments as the object LLM Obs types them as, or the raw string when they are not one.
@ -282,7 +282,7 @@ def _to_dd_arguments(raw_arguments: object) -> dict[str, Any] | str:
return parsed if isinstance(parsed, dict) else raw_arguments
def _to_dd_tool_calls(message: Mapping[str, Any]) -> tuple[ToolCall, ...]:
def _to_dd_tool_calls(message: Mapping[str, object]) -> tuple[ToolCall, ...]:
"""
The tool calls a message carries, in LLM Obs' ToolCall schema, from either dialect.
@ -315,7 +315,7 @@ def _to_dd_tool_calls(message: Mapping[str, Any]) -> tuple[ToolCall, ...]:
return openai_calls + anthropic_calls
def _to_dd_tool_results(message: Mapping[str, Any], tool_call_names: Mapping[str, str]) -> tuple[ToolResult, ...]:
def _to_dd_tool_results(message: Mapping[str, object], tool_call_names: Mapping[str, str]) -> tuple[ToolResult, ...]:
"""
The tool results a message carries, linked back to the call each answers.
@ -400,7 +400,7 @@ def _to_dd_messages(messages: object) -> tuple[Message, ...]:
return tuple(_to_dd_message(message, tool_call_names) for message in messages)
def _to_dd_tool_definition(entry: Mapping[str, Any]) -> ToolDefinition | None:
def _to_dd_tool_definition(entry: Mapping[str, object]) -> ToolDefinition | None:
function: Final = entry.get("function")
declared: Final[Mapping[str, Any]] = function if isinstance(function, dict) else entry
name: Final = declared.get("name")
@ -683,7 +683,7 @@ class DataDogLLMObsLogger(CustomBatchLogger):
if callable(current_span_fn):
current_span: Final = current_span_fn()
if current_span is not None:
trace_id: Final = getattr(current_span, "trace_id", None)
trace_id: Final[object] = getattr(current_span, "trace_id", None)
if trace_id is not None:
return str(trace_id)
except Exception:
@ -716,7 +716,7 @@ class DataDogLLMObsLogger(CustomBatchLogger):
def redacts_messages_itself(self) -> bool:
return True
def _payload_logging_is_off(self, kwargs: Mapping[str, Any]) -> bool:
def _payload_logging_is_off(self, kwargs: Mapping[str, object]) -> bool:
return (
bool(self.turn_off_message_logging)
or self.message_logging is not True

11
litellm/integrations/galileo.py
View file

@ -396,12 +396,13 @@ class GalileoObserve(CustomLogger):
)
@staticmethod
def _log_v2_payload_validation(payload: dict[str, Any]) -> None:
def _log_v2_payload_validation(payload: dict[str, object]) -> None:
missing_fields: Final[list[str]] = []
traces: Final[Sequence[object]] = payload.get("traces", [])
if not traces:
traces_value: Final = payload.get("traces", [])
if not traces_value:
missing_fields.append("traces")
traces: Final[Sequence[object]] = traces_value if isinstance(traces_value, list) else []
for trace_index, trace in enumerate(traces):
if not isinstance(trace, dict):
continue
@ -425,8 +426,8 @@ class GalileoObserve(CustomLogger):
missing_fields,
)
def _log_flush_payload(self, url: str, payload: dict[str, Any]) -> None:
traces: Final[Sequence[object]] = payload.get("traces", [])
def _log_flush_payload(self, url: str, payload: dict[str, object]) -> None:
traces: Final = payload.get("traces")
verbose_logger.debug(
"Galileo Logger flush URL: %s trace_count=%s",
url,

8
litellm/integrations/langfuse/langfuse.py
View file

@ -4,7 +4,7 @@ import inspect
import os
import re
import traceback
from collections.abc import Callable, Iterable, Mapping
from collections.abc import Callable, Iterable, Mapping, Sequence
from datetime import datetime
from functools import lru_cache
from types import MappingProxyType
@ -447,7 +447,7 @@ class LangFuseLogger:
prompt: dict,
level: str,
status_message: str | None,
) -> tuple[dict | None, str | dict | list | None]:
) -> tuple[dict | None, str | dict | Sequence[object] | None]:
"""
Get the input and output content for Langfuse logging
@ -463,7 +463,7 @@ class LangFuseLogger:
output: The output content for Langfuse logging
"""
input = None
output: str | dict | list[Any] | None = None
output: str | dict | Sequence[object] | None = None
if level == "ERROR" and status_message is not None and isinstance(status_message, str):
input = prompt
output = status_message
@ -575,7 +575,7 @@ class LangFuseLogger:
user_id: str | None,
metadata: dict[str, object],
litellm_params: dict,
output: str | dict | list | None,
output: str | dict | Sequence[object] | None,
start_time: datetime | None,
end_time: datetime | None,
kwargs: dict,

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

@ -1,6 +1,6 @@
import datetime
from collections.abc import Mapping
from typing import Any, Final
from typing import Final
import httpx
@ -54,7 +54,7 @@ class ResponseMetadata:
Handles setting and managing `_hidden_params`, `response_time_ms`, and `litellm_overhead_time_ms` for LiteLLM responses
"""
def __init__(self, result: Any):
def __init__(self, result: object):
self.result = result
self._hidden_params: HiddenParams | dict = getattr(result, "_hidden_params", {}) or {}

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

@ -13,14 +13,6 @@ from pathlib import Path
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar, cast
from openai.types.chat.chat_completion_custom_tool_param import (
CustomFormatGrammar,
CustomFormatGrammarGrammar,
)
from openai.types.shared_params.custom_tool_input_format import (
Grammar as ResponsesGrammarFormat,
)
import litellm
from litellm import verbose_logger
from litellm.router_utils.batch_utils import InMemoryFile
@ -59,7 +51,7 @@ if TYPE_CHECKING:
def handle_any_messages_to_chat_completion_str_messages_conversion(
messages: Any,
messages: object,
) -> list[dict[str, str]]:
"""
Handles any messages to chat completion str messages conversion
@ -804,7 +796,7 @@ def extract_file_metadata(file_data: FileTypes) -> tuple[str | None, str | None]
"""
filename: str | None = None
content_type: str | None = None
file_content: Any = None
file_content: object = None
if isinstance(file_data, tuple):
if len(file_data) == 2:
@ -1002,7 +994,7 @@ def unpack_defs(
# Use iterative approach with queue to avoid recursion
# Each item in queue is (node, parent_container, key/index, active_defs, ref_chain)
queue: Final[deque[tuple[Any, dict | list | None, str | int | None, dict, set]]] = deque(
queue: Final[deque[tuple[object, dict | list | None, str | int | None, dict, set]]] = deque(
[(schema, None, None, root_defs, set())]
)
inlined_bytes = 0
@ -1624,7 +1616,10 @@ def is_function_call(optional_params: dict) -> bool:
return False
def convert_custom_tool_format_to_chat_shape(format_obj: Mapping[str, Any]) -> Mapping[str, Any]:
_CUSTOM_GRAMMAR_FIELDS: Final = ("definition", "syntax")
def convert_custom_tool_format_to_chat_shape(format_obj: Mapping[str, object]) -> Mapping[str, object]:
"""
Responses API grammar formats are flat ({"type": "grammar", "definition", "syntax"});
Chat Completions wraps the same fields in a "grammar" object. Text formats are
@ -1632,15 +1627,11 @@ def convert_custom_tool_format_to_chat_shape(format_obj: Mapping[str, Any]) -> M
"""
if format_obj.get("type") != "grammar" or "grammar" in format_obj:
return format_obj
grammar: Final = CustomFormatGrammarGrammar()
if "definition" in format_obj:
grammar["definition"] = format_obj["definition"]
if "syntax" in format_obj:
grammar["syntax"] = format_obj["syntax"]
return CustomFormatGrammar(type="grammar", grammar=grammar)
grammar: Final[Mapping[str, object]] = {key: format_obj[key] for key in _CUSTOM_GRAMMAR_FIELDS if key in format_obj}
return {"type": "grammar", "grammar": grammar}
def convert_custom_tool_format_to_responses_shape(format_obj: Mapping[str, Any]) -> Mapping[str, Any]:
def convert_custom_tool_format_to_responses_shape(format_obj: Mapping[str, object]) -> Mapping[str, object]:
"""
Inverse of convert_custom_tool_format_to_chat_shape: unwrap the Chat Completions
"grammar" object into the flat Responses API grammar shape.
@ -1648,12 +1639,10 @@ def convert_custom_tool_format_to_responses_shape(format_obj: Mapping[str, Any])
grammar: Final = format_obj.get("grammar")
if format_obj.get("type") != "grammar" or not isinstance(grammar, dict):
return format_obj
flat: Final = ResponsesGrammarFormat(type="grammar")
if "definition" in grammar:
flat["definition"] = grammar["definition"]
if "syntax" in grammar:
flat["syntax"] = grammar["syntax"]
return flat
return {
"type": "grammar",
**{key: grammar[key] for key in _CUSTOM_GRAMMAR_FIELDS if key in grammar},
}
def get_file_ids_from_messages(messages: list[AllMessageValues]) -> list[str]:

2
litellm/llms/a2a/chat/guardrail_translation/handler.py
View file

@ -125,7 +125,7 @@ class A2AGuardrailHandler(BaseTranslation):
litellm_logging_obj: Optional["LiteLLMLoggingObj"] = None,
user_api_key_dict: Optional["UserAPIKeyAuth"] = None,
request_data: dict | None = None,
) -> Any:
) -> object:
"""
Process A2A output response by applying guardrails to text content.

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

@ -6,7 +6,7 @@ from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, NoReturn, cast
import httpx
from pydantic import ValidationError
from pydantic import BaseModel, ValidationError
from typing_extensions import ReadOnly, TypedDict
import litellm
@ -150,7 +150,7 @@ class _AnthropicToolResultBlock(TypedDict, total=False):
content: ReadOnly[object]
_ENUM_TYPE_CHECKS: Final[Mapping[str, Callable[[object], bool]]] = MappingProxyType(
_ENUM_TYPE_CHECKS: Final[Mapping[object, Callable[[object], bool]]] = MappingProxyType(
{
"null": lambda v: v is None,
"boolean": lambda v: isinstance(v, bool),
@ -163,7 +163,7 @@ _ENUM_TYPE_CHECKS: Final[Mapping[str, Callable[[object], bool]]] = MappingProxyT
)
def _enum_conflicts_with_declared_type(schema: Mapping[str, Any]) -> bool:
def _enum_conflicts_with_declared_type(schema: Mapping[str, object]) -> bool:
"""Whether ``schema``'s ``enum`` cannot match its declared ``type``."""
enum_values: Final = schema.get("enum")
declared_type: Final = schema.get("type")
@ -658,7 +658,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
return result
def get_json_schema_from_pydantic_object(self, response_format: Any | dict | None) -> dict | None:
def get_json_schema_from_pydantic_object(self, response_format: type[BaseModel] | dict | None) -> dict | None:
return type_to_response_format_param(
response_format, ref_template="/$defs/{model}"
) # Relevant issue: https://github.com/BerriAI/litellm/issues/7755
@ -1061,7 +1061,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
@staticmethod
def _sanitize_tool_names_in_request(
optional_params: dict[str, Any],
optional_params: dict[str, object],
) -> tuple[dict[str, str], dict[str, str]]:
"""Sanitize ``optional_params['tools']`` and ``optional_params['tool_choice']``
in place so every name matches Anthropic's ``^[a-zA-Z0-9_-]{1,128}$``.
@ -1108,7 +1108,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
# so a caller reusing the same tool list/dicts across requests
# doesn't see its inputs permanently rewritten (which would also
# drop the original key from `forward` on the next request).
new_tools: Final[list[Any]] = []
new_tools: Final[list[object]] = []
for t in tools:
if (
isinstance(t, dict)
@ -1431,7 +1431,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
entry_type = entry.get("type")
if entry_type == "compaction":
anthropic_edit: dict[str, Any] = {"type": "compact_20260112"}
anthropic_edit: dict[str, object] = {"type": "compact_20260112"}
compact_threshold = entry.get("compact_threshold")
# Rewrite to 'trigger' with correct nesting if threshold exists
if compact_threshold is not None and isinstance(compact_threshold, (int, float)):
@ -2431,9 +2431,11 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
code_by_id: Final[dict[str, str]] = {}
for tc in tool_calls:
try:
args = json.loads(tc.get("function", {}).get("arguments", "{}"))
args: object = json.loads(tc.get("function", {}).get("arguments", "{}"))
if not isinstance(args, Mapping):
continue
call_id = tc.get("id")
command = args.get("command", "")
command: object = args.get("command", "")
if isinstance(call_id, str):
code_by_id[call_id] = command if isinstance(command, str) else ""
except Exception:
@ -2503,8 +2505,8 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
tool_results: Sequence[_AnthropicToolResultBlock] | None,
compaction_blocks: Sequence[object] | None,
tool_calls: list[ChatCompletionToolCallChunk],
) -> dict[str, Any]:
provider_specific_fields: Final[dict[str, Any]] = {
) -> dict[str, object]:
provider_specific_fields: Final[dict[str, object]] = {
"citations": citations,
"thinking_blocks": thinking_blocks,
}

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

@ -7,7 +7,7 @@ import re
from collections.abc import Mapping, MutableMapping, Sequence
from datetime import datetime, timezone
from types import MappingProxyType
from typing import Any, Final, Literal
from typing import Any, Final, Literal, TypeVar
import httpx
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
@ -39,6 +39,8 @@ from litellm.types.llms.anthropic import (
from litellm.types.llms.openai import AllMessageValues
from litellm.types.proxy.model_listing import ModelInfoResponse
_MessageT = TypeVar("_MessageT")
DROP_FORCED_TOOL_CHOICE_WARNING: Final = (
"Downgrading forced tool_choice to 'auto' for model=%s (drop_params=True): this model rejects tool_choice type "
"'any'/'tool' with a 400 because thinking is always on and a forced call would skip it."
@ -1074,7 +1076,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
return AnthropicTokenCounter()
def strip_advisor_blocks_from_messages(messages: list[Any], replace_with_text: bool = False) -> list[Any]:
def strip_advisor_blocks_from_messages(messages: list[_MessageT], replace_with_text: bool = False) -> list[_MessageT]:
"""
Remove (or replace) server_tool_use (name='advisor') and advisor_tool_result blocks
from assistant message content.
@ -1181,7 +1183,7 @@ def is_anthropic_invalid_thinking_block_error(error_text: str) -> bool:
return "must contain thinking" in lower
def strip_thinking_blocks_from_anthropic_messages(messages: list[Any]) -> list[Any]:
def strip_thinking_blocks_from_anthropic_messages(messages: Sequence[object]) -> list[object]:
"""
Return a new message list with thinking / redacted_thinking content blocks removed
from each message. Used to recover from invalid thinking signatures on retry.
@ -1189,7 +1191,7 @@ def strip_thinking_blocks_from_anthropic_messages(messages: list[Any]) -> list[A
Messages whose content is a list and becomes empty after stripping are omitted,
since Anthropic rejects empty content arrays.
"""
out: Final[list[Any]] = []
out: Final[list[object]] = []
for m in messages:
if not isinstance(m, dict):
out.append(m)

16
litellm/llms/anthropic/experimental_pass_through/messages/agentic_streaming_iterator.py
View file

@ -25,6 +25,9 @@ from litellm.constants import STREAM_SSE_KEEPALIVE_PING_BYTES
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.anthropic_messages.transformation import (
BaseAnthropicMessagesConfig,
)
HOLD_BACK_PING_INTERVAL_SECONDS: Final = 15.0
SERVER_FULFILLED_TOOL_LEAK_ERROR_SSE_BYTES: Final = (
@ -182,7 +185,7 @@ class AgenticAnthropicStreamingIterator:
http_handler: Any,
model: str,
messages: list[dict],
anthropic_messages_provider_config: Any,
anthropic_messages_provider_config: "BaseAnthropicMessagesConfig",
anthropic_messages_optional_request_params: dict,
logging_obj: "LiteLLMLoggingObj",
custom_llm_provider: str,
@ -402,7 +405,7 @@ class AgenticAnthropicStreamingIterator:
@staticmethod
def _rebuild_anthropic_response_from_sse(
raw_bytes: list[bytes],
) -> dict[str, Any] | None:
) -> dict[str, object] | None:
"""
Parse collected SSE bytes into an Anthropic Messages response dict.
@ -416,17 +419,18 @@ class AgenticAnthropicStreamingIterator:
"""
events: Final = _parse_sse_events(b"".join(raw_bytes))
response: Final[dict[str, Any]] = {
content: Final[list[dict[str, object]]] = []
response: Final[dict[str, object]] = {
"id": "",
"type": "message",
"role": "assistant",
"model": "",
"content": [],
"content": content,
"stop_reason": None,
"stop_sequence": None,
"usage": {"input_tokens": 0, "output_tokens": 0},
}
content_blocks: Final[dict[int, dict[str, Any]]] = {}
content_blocks: Final[dict[int, dict[str, object]]] = {}
saw_message_start = False
for event_type, data in events:
@ -448,6 +452,6 @@ class AgenticAnthropicStreamingIterator:
for idx in sorted(content_blocks.keys()):
block = content_blocks[idx]
block.pop("_partial_json", None)
response["content"].append(block)
content.append(block)
return response

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

@ -185,7 +185,11 @@ class AnthropicFilesHandler:
if not line.strip():
continue
anthropic_result = json.loads(line)
anthropic_result: object = json.loads(line)
if not isinstance(anthropic_result, dict):
raise TypeError(
f"Anthropic batch result line is not a JSON object: {type(anthropic_result).__name__}"
)
custom_id = anthropic_result.get("custom_id", "")
result = anthropic_result.get("result", {})
result_type = result.get("type", "")

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

@ -1037,7 +1037,7 @@ class AmazonConverseConfig(BaseConfig):
return optional_params
def _map_request_metadata_param(self, value: Any, optional_params: dict) -> None:
def _map_request_metadata_param(self, value: object, optional_params: dict) -> None:
if value is not None and isinstance(value, dict):
self._validate_request_metadata(value)
optional_params["requestMetadata"] = value

22
litellm/llms/chatgpt/responses/transformation.py
View file

@ -1,5 +1,8 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from litellm.exceptions import AuthenticationError
from litellm.litellm_core_utils.core_helpers import process_response_headers
from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import (
@ -13,6 +16,7 @@ from litellm.responses.sse_output_recovery import (
record_output_text_chunk,
)
from litellm.types.llms.openai import (
ResponseInputParam,
ResponsesAPIResponse,
ResponsesAPIStreamEvents,
)
@ -64,7 +68,7 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
def transform_responses_api_request(
self,
model: str,
input: Any,
input: str | ResponseInputParam,
response_api_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
@ -109,9 +113,9 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
def transform_response_api_response(
self,
model: str,
raw_response: Any,
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
):
) -> ResponsesAPIResponse:
body_text: Final = raw_response.text or ""
if not self._should_parse_as_sse(raw_response=raw_response, body_text=body_text):
return super().transform_response_api_response(
@ -135,7 +139,7 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
self._attach_response_headers(completed_response=completed_response, raw_response=raw_response)
return completed_response
def _should_parse_as_sse(self, raw_response: Any, body_text: str) -> bool:
def _should_parse_as_sse(self, raw_response: httpx.Response, body_text: str) -> bool:
content_type: Final = (raw_response.headers or {}).get("content-type", "")
if "text/event-stream" in content_type.lower():
return True
@ -150,8 +154,8 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
def _extract_completed_response_from_sse(self, body_text: str) -> tuple[ResponsesAPIResponse | None, str | None]:
completed_response = None
error_message = None
streamed_output_items: Final[dict[int, dict]] = {}
text_only_output_items: Final[dict[int, dict]] = {}
streamed_output_items: Final[dict[int, dict[str, object]]] = {}
text_only_output_items: Final[dict[int, dict[str, object]]] = {}
for chunk in body_text.splitlines():
parsed_chunk = parse_sse_json_chunk(chunk)
if parsed_chunk is None:
@ -178,7 +182,7 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
# output_index, but text-only items at indices without a
# matching OUTPUT_ITEM_DONE must still be preserved (e.g.
# providers that emit only OUTPUT_TEXT_DONE for some indices).
merged_items: dict[int, dict] = {**text_only_output_items}
merged_items: dict[int, dict[str, object]] = {**text_only_output_items}
merged_items.update(streamed_output_items)
completed_response = self._build_completed_response_from_chunk(
parsed_chunk=parsed_chunk,
@ -197,7 +201,7 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
return completed_response, error_message
def _build_completed_response_from_chunk(
self, parsed_chunk: dict[str, Any], streamed_output_items: dict[int, dict]
self, parsed_chunk: Mapping[str, object], streamed_output_items: Mapping[int, dict[str, object]]
) -> ResponsesAPIResponse | None:
response_payload = parsed_chunk.get("response")
if not isinstance(response_payload, dict):
@ -223,7 +227,7 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
def _attach_response_headers(
self,
completed_response: ResponsesAPIResponse,
raw_response: Any,
raw_response: httpx.Response,
) -> None:
raw_headers: Final = dict(raw_response.headers)
processed_headers: Final = process_response_headers(raw_headers)

2
litellm/llms/cohere/chat/transformation.py
View file

@ -110,7 +110,7 @@ class CohereChatConfig(BaseConfig):
tool_results: list | None = None,
seed: int | None = None,
) -> None:
locals_: Final = locals().copy()
locals_: Final[dict[str, object]] = locals().copy()
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)

10
litellm/llms/databricks/chat/transformation.py
View file

@ -3,7 +3,7 @@ Translates from OpenAI's `/v1/chat/completions` to Databricks' `/chat/completion
"""
import os
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, Literal, cast, overload
import httpx
@ -67,7 +67,7 @@ def _is_bare_assistant_message(message_dict: Mapping[str, object]) -> bool:
)
def _sanitize_empty_content(message_dict: dict[str, Any]) -> None:
def _sanitize_empty_content(message_dict: dict[str, object]) -> None:
"""
Remove or filter content so empty text blocks are not sent.
Databricks Model Serving uses Anthropic Messages API spec and rejects empty text blocks.
@ -430,7 +430,7 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
@overload
def _transform_messages(
self, messages: list[AllMessageValues], model: str, is_async: Literal[True]
) -> Coroutine[Any, Any, list[AllMessageValues]]: ...
) -> Coroutine[object, object, list[AllMessageValues]]: ...
@overload
def _transform_messages(
@ -442,7 +442,7 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
def _transform_messages(
self, messages: list[AllMessageValues], model: str, is_async: bool = False
) -> list[AllMessageValues] | Coroutine[Any, Any, list[AllMessageValues]]:
) -> list[AllMessageValues] | Coroutine[object, object, list[AllMessageValues]]:
"""
Databricks does not support:
- 'name' in user message.
@ -564,7 +564,7 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
@staticmethod
def extract_citations(
content: AllDatabricksContentValues | None,
) -> list[Any] | None:
) -> Sequence[Sequence[Mapping[str, object]]] | None:
if content is None:
return None
citations: Final = []

4
litellm/llms/fireworks_ai/chat/transformation.py
View file

@ -1,6 +1,6 @@
import json
from collections.abc import AsyncIterator, Iterator, Mapping
from typing import TYPE_CHECKING, Any, Final, Literal, cast
from typing import TYPE_CHECKING, Final, Literal, cast
import httpx
@ -751,7 +751,7 @@ class FireworksAIConfig(FireworksAIMixin, OpenAIGPTConfig):
streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
sync_stream: bool,
json_mode: bool | None = False,
) -> Any:
) -> "FireworksAIChatCompletionStreamingHandler":
return FireworksAIChatCompletionStreamingHandler(
streaming_response=streaming_response,
sync_stream=sync_stream,

2
litellm/llms/gemini/count_tokens/handler.py
View file

@ -84,7 +84,7 @@ class GoogleAIStudioTokenCounter:
api_key: str | None = None,
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
**kwargs: object,
) -> dict[str, Any]:
"""
Count tokens using Google Gen AI Studio countTokens endpoint.

13
litellm/llms/gemini/image_edit/transformation.py
View file

@ -1,4 +1,5 @@
import base64
from collections.abc import Mapping
from io import BufferedReader, BytesIO
from typing import TYPE_CHECKING, Any, Final, cast
@ -44,7 +45,7 @@ class GeminiImageEditConfig(BaseImageEditConfig):
image_edit_optional_params: ImageEditOptionalRequestParams,
model: str,
drop_params: bool,
) -> dict[str, Any]:
) -> dict[str, object]:
return map_openai_image_params_to_gemini(
params=image_edit_optional_params,
model=model,
@ -87,10 +88,10 @@ class GeminiImageEditConfig(BaseImageEditConfig):
model: str,
prompt: str | None,
image: FileTypes | None,
image_edit_optional_request_params: dict[str, Any],
image_edit_optional_request_params: Mapping[str, object],
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[dict[str, Any], RequestFiles | None]:
) -> tuple[dict[str, object], RequestFiles | None]:
inline_parts: Final = self._prepare_inline_image_parts(image) if image else []
if not inline_parts:
raise ValueError("Gemini image edit requires at least one image.")
@ -106,7 +107,7 @@ class GeminiImageEditConfig(BaseImageEditConfig):
}
]
request_body: Final[dict[str, Any]] = {"contents": contents}
request_body: Final[dict[str, object]] = {"contents": contents}
request_body["generationConfig"] = get_gemini_image_generation_config(
model=model,
@ -153,14 +154,14 @@ class GeminiImageEditConfig(BaseImageEditConfig):
model_response.usage = transform_gemini_image_usage(response_json["usageMetadata"])
return model_response
def _prepare_inline_image_parts(self, image: FileTypes | list[FileTypes]) -> list[dict[str, Any]]:
def _prepare_inline_image_parts(self, image: FileTypes | list[FileTypes]) -> list[dict[str, object]]:
images: list[FileTypes]
if isinstance(image, list):
images = image
else:
images = [image]
inline_parts: Final[list[dict[str, Any]]] = []
inline_parts: Final[list[dict[str, object]]] = []
for img in images:
if img is None:
continue

14
litellm/llms/gigachat/chat/transformation.py
View file

@ -81,9 +81,17 @@ class GigaChatConfig(BaseConfig):
repetition_penalty: float | None = None,
profanity_check: bool | None = None,
) -> None:
locals_: Final = locals().copy()
for key, value in locals_.items():
if key != "self" and value is not None:
config_params: Final[Mapping[str, float | int | bool | None]] = MappingProxyType(
{
"temperature": temperature,
"top_p": top_p,
"max_tokens": max_tokens,
"repetition_penalty": repetition_penalty,
"profanity_check": profanity_check,
}
)
for key, value in config_params.items():
if value is not None:
setattr(self.__class__, key, value)
# Instance variables for current request context
self._current_credentials: str | None = None

10
litellm/llms/huggingface/embedding/transformation.py
View file

@ -84,13 +84,13 @@ class HuggingFaceEmbeddingConfig(BaseConfig):
typical_p: float | None = None,
watermark: bool | None = None,
) -> None:
locals_: Final = locals().copy()
locals_: Final[dict[str, object]] = 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):
def get_config(cls) -> dict[str, object]:
return super().get_config()
def get_special_options_params(self):
@ -352,17 +352,17 @@ class HuggingFaceEmbeddingConfig(BaseConfig):
model: str,
data: dict,
api_key: str | None = None,
) -> list[dict[str, Any]]:
) -> list[dict[str, str]]:
streamed_response: Final = CustomStreamWrapper(
completion_stream=response.iter_lines(),
model=model,
custom_llm_provider="huggingface",
logging_obj=logging_obj,
)
content = ""
content: str = ""
for chunk in streamed_response:
content += chunk["choices"][0]["delta"]["content"]
completion_response: Final[list[dict[str, Any]]] = [{"generated_text": content}]
completion_response: Final[list[dict[str, str]]] = [{"generated_text": content}]
## LOGGING
logging_obj.post_call(
input=data,

20
litellm/llms/openai/chat/guardrail_translation/handler.py
View file

@ -17,7 +17,7 @@ This pattern can be replicated for other message formats (e.g., Anthropic).
import json
import time
import uuid
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, Union, cast
from typing_extensions import NotRequired, ReadOnly, TypedDict
@ -232,7 +232,7 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
def _extract_inputs(
self,
message: dict[str, Any],
message: Mapping[str, object],
msg_idx: int,
texts_to_check: list[str],
images_to_check: list[str],
@ -293,7 +293,7 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
async def _apply_guardrail_responses_to_input_texts(
self,
messages: list[dict[str, Any]],
messages: list[dict[str, object]],
responses: list[str],
task_mappings: list[tuple[int, int | None]],
) -> None:
@ -318,12 +318,12 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
elif isinstance(content, list) and content_idx_optional is not None:
# Replace specific text item in list content
messages[msg_idx]["content"][content_idx_optional]["text"] = guardrail_response
content[content_idx_optional]["text"] = guardrail_response
async def _apply_guardrail_responses_to_input_tool_calls(
self,
messages: list[dict[str, Any]],
tool_calls: list[dict[str, Any]],
messages: Sequence[Mapping[str, object]],
tool_calls: Sequence[Mapping[str, object]],
task_mappings: list[tuple[int, int]],
) -> None:
"""
@ -375,7 +375,7 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
texts_to_check: Final[list[str]] = []
images_to_check: Final[list[str]] = []
tool_calls_to_check: Final[list[dict[str, Any]]] = []
tool_calls_to_check: Final[list[dict[str, object]]] = []
text_task_mappings: Final[list[tuple[int, int | None]]] = []
tool_call_task_mappings: Final[list[tuple[int, int]]] = []
# text_task_mappings: Track (choice_index, content_index) for each text
@ -424,8 +424,8 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
guardrailed_texts: Final = guardrailed_inputs.get("texts", [])
returned_tool_calls: Final = guardrailed_inputs.get("tool_calls")
guardrailed_tool_calls: Final[list[dict[str, Any]]] = (
cast(list[dict[str, Any]], returned_tool_calls)
guardrailed_tool_calls: Final[list[dict[str, object]]] = (
cast(list[dict[str, object]], returned_tool_calls)
if isinstance(returned_tool_calls, list) and len(returned_tool_calls) == len(tool_calls_to_check)
else tool_calls_to_check
)
@ -864,7 +864,7 @@ class OpenAIChatCompletionsHandler(BaseTranslation):
choice_idx: int,
texts_to_check: list[str],
images_to_check: list[str],
tool_calls_to_check: list[dict[str, Any]],
tool_calls_to_check: list[dict[str, object]],
text_task_mappings: list[tuple[int, int | None]],
tool_call_task_mappings: list[tuple[int, int]],
) -> None:

6
litellm/llms/openai/videos/transformation.py
View file

@ -237,7 +237,7 @@ class OpenAIVideoConfig(BaseVideoConfig):
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
extra_body: dict[str, Any] | None = None,
extra_body: dict[str, object] | None = None,
) -> tuple[str, dict]:
"""
Transform the video remix request for OpenAI API.
@ -252,7 +252,7 @@ class OpenAIVideoConfig(BaseVideoConfig):
url: Final = f"{api_base.rstrip('/')}/{encoded_video_id}/remix"
# Prepare the request data
data: Final = {"prompt": prompt}
data: Final[dict[str, object]] = {"prompt": prompt}
# Add any extra body parameters
if extra_body:
@ -305,7 +305,7 @@ class OpenAIVideoConfig(BaseVideoConfig):
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: dict[str, Any] | None = None,
extra_query: dict[str, object] | None = None,
) -> tuple[str, dict]:
"""
Transform the video list request for OpenAI API.

11
litellm/llms/openrouter/image_edit/transformation.py
View file

@ -90,20 +90,21 @@ class OpenRouterImageEditConfig(BaseImageEditConfig):
drop_params: bool,
) -> dict:
supported_params: Final = self.get_supported_openai_params(model)
mapped_params: Final[dict[str, Any]] = {}
mapped_params: Final[dict[str, object]] = {}
image_config: Final[dict[str, str]] = {}
for key, value in image_edit_optional_params.items():
if key in supported_params:
if key == "size":
if "image_config" not in mapped_params:
mapped_params["image_config"] = {}
mapped_params["image_config"]["aspect_ratio"] = self._map_size_to_aspect_ratio(cast(str, value))
mapped_params["image_config"] = image_config
image_config["aspect_ratio"] = self._map_size_to_aspect_ratio(cast(str, value))
elif key == "quality":
image_size = self._map_quality_to_image_size(cast(str, value))
if image_size:
if "image_config" not in mapped_params:
mapped_params["image_config"] = {}
mapped_params["image_config"]["image_size"] = image_size
mapped_params["image_config"] = image_config
image_config["image_size"] = image_size
else:
mapped_params[key] = value

2
litellm/llms/perplexity/embedding/transformation.py
View file

@ -130,7 +130,7 @@ class PerplexityEmbeddingConfig(BaseEmbeddingConfig):
if isinstance(embedding_value, str):
raw_bytes: Final = base64.b64decode(embedding_value)
count: Final = len(raw_bytes)
int8_values: Final = struct.unpack(f"{count}b", raw_bytes)
int8_values: Final[tuple[int, ...]] = struct.unpack(f"{count}b", raw_bytes)
return [float(v) / 127.0 for v in int8_values]
return embedding_value

4
litellm/llms/vertex_ai/gemini/transformation.py
View file

@ -179,7 +179,7 @@ def _apply_gemini_metadata(
part: PartType,
model: str | None,
media_resolution_enum: dict[str, str] | None,
video_metadata: dict[str, Any] | None,
video_metadata: Mapping[str, object] | None,
) -> PartType:
"""
Apply media_resolution and video_metadata parameters to a Gemini part.
@ -480,7 +480,7 @@ def _process_gemini_media(
format: str | None = None,
media_resolution_enum: dict[str, str] | None = None,
model: str | None = None,
video_metadata: dict[str, Any] | None = None,
video_metadata: Mapping[str, object] | None = None,
vertex_project: str | None = None,
vertex_credentials: object = None,
) -> PartType:

2
litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
View file

@ -5574,7 +5574,7 @@ class MCPServerManager:
async def pre_call_tool_check(
self,
name: str,
arguments: dict[str, Any],
arguments: _ToolArguments,
server_name: str,
user_api_key_auth: UserAPIKeyAuth | None,
proxy_logging_obj: ProxyLogging | None,

26
litellm/proxy/auth/auth_utils.py
View file

@ -166,7 +166,7 @@ def check_regex_or_str_match(request_body_value: Any, regex_str: str) -> bool:
def _is_param_allowed(
param: str,
request_body_value: Any,
request_body_value: object,
configurable_clientside_auth_params: CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS,
) -> bool:
"""
@ -189,7 +189,7 @@ def _is_param_allowed(
def _allow_model_level_clientside_configurable_parameters(
model: str, param: str, request_body_value: Any, llm_router: Router | None
model: str, param: str, request_body_value: object, llm_router: Router | None
) -> bool:
"""
Check if model is allowed to use configurable client-side params
@ -532,7 +532,7 @@ def is_request_body_safe(request_body: dict, general_settings: dict, llm_router:
return True
def _coerce_metadata_to_dict(value: Any) -> dict[str, Any] | None:
def _coerce_metadata_to_dict(value: object) -> dict[str, object] | None:
"""Return ``value`` as a dict, parsing it from JSON if delivered as a string.
Multipart/form-data and ``extra_body`` callers send ``litellm_metadata``
@ -891,7 +891,7 @@ async def check_if_request_size_is_safe(request: Request) -> bool:
return True
async def check_response_size_is_safe(response: Any) -> bool:
async def check_response_size_is_safe(response: object) -> bool:
"""
Enterprise Only:
- Checks if the response size is within the limit
@ -1526,7 +1526,7 @@ def get_customer_user_header_from_mapping(user_id_mapping) -> list | None:
def _get_customer_id_from_standard_headers(
request_headers: dict | None,
request_headers: Mapping[str, object] | None,
) -> str | None:
"""
Check standard customer ID headers for a customer/end-user ID.
@ -1552,7 +1552,7 @@ def _get_customer_id_from_standard_headers(
return None
def _coerce_user_id_to_str(value: Any) -> str | None:
def _coerce_user_id_to_str(value: object) -> str | None:
"""Return a usable end-user identifier string, or None if the value isn't one.
Always drops non-string structured values (dict/list/tuple/set) because
@ -1579,7 +1579,7 @@ def _coerce_user_id_to_str(value: Any) -> str | None:
# behind the flag preserves backwards compatibility for deployments
# that intentionally pass JSON-encoded user identifiers.
if litellm.validate_end_user_id_in_db and stripped[:1] in ("{", "["):
parsed: Final = safe_json_loads(stripped)
parsed: Final[object] = safe_json_loads(stripped)
if isinstance(parsed, (dict, list)):
return None
return stripped
@ -1587,7 +1587,9 @@ def _coerce_user_id_to_str(value: Any) -> str | None:
return None
def get_end_user_id_from_request_body(request_body: dict, request_headers: dict | None = None) -> str | None:
def get_end_user_id_from_request_body(
request_body: Mapping[str, object], request_headers: Mapping[str, object] | None = None
) -> str | None:
# Import general_settings here to avoid potential circular import issues at module level
# and to ensure it's fetched at runtime.
from litellm.proxy.proxy_server import general_settings
@ -1636,7 +1638,7 @@ def get_end_user_id_from_request_body(request_body: dict, request_headers: dict
if user_id_str:
return user_id_str
def _as_dict(value: Any) -> dict:
def _as_dict(value: object) -> dict:
# metadata / litellm_metadata can arrive as JSON strings from
# multipart/form-data or extra_body; coerce so string-encoded
# payloads can't evade end-user attribution.
@ -1721,11 +1723,11 @@ _MODEL_ROUTING_ID_FIELDS: Final = (
)
def _append_model_candidates(candidates: list[str], value: Any) -> None:
def _append_model_candidates(candidates: list[str], value: object) -> None:
if value is None:
return
values: Final = value if isinstance(value, (list, tuple, set)) else [value]
values: Final[tuple[object, ...]] = tuple(value) if isinstance(value, (list, tuple, set)) else (value,)
for item in values:
if item is None:
continue
@ -1766,7 +1768,7 @@ def _route_uses_model_routing_sources(route: str) -> bool:
def _extract_models_from_managed_resource_id(
resource_id: Any,
resource_id: object,
resource_id_field: str | None = None,
llm_router: Router | None = None,
) -> list[str]:

60
litellm/proxy/client/cli/commands/configure.py
View file

@ -98,13 +98,15 @@ def _preflight(target: str) -> None:
raise click.ClickException(str(e)) from e
def _start(ctx: click.Context, api_key: str | None, target: str = _CLAUDE_TARGET) -> tuple[StaticToken, _Listing]:
def _start(
ctx: click.Context, base_url: str, api_key: str | None, target: str = _CLAUDE_TARGET
) -> tuple[StaticToken, _Listing]:
_preflight(target)
try:
credential: Final = resolve_credential(ctx, api_key)
except ClaudeSettingsError as e:
raise click.ClickException(str(e))
return credential, _listed_models(ctx.obj["base_url"], credential.token, target)
return credential, _listed_models(base_url, credential.token, target)
def _listing_error(base_url: str, error: PiSyncError, target: str) -> str:
@ -147,9 +149,7 @@ def _validated_model(model: str | None, listing: _Listing, base_url: str) -> str
return starting
def _apply_claude(ctx: click.Context, credential: StaticToken, listing: _Listing, model: str | None) -> None:
ctx_obj: Final[CliContextObj] = ctx.obj
base_url: Final = ctx_obj["base_url"]
def _apply_claude(base_url: str, credential: StaticToken, listing: _Listing, model: str | None) -> None:
listed: Final = listing.ids
starting: Final = _validated_model(model, listing, base_url)
settings_path: Final = claude_settings_path(os.environ)
@ -214,8 +214,7 @@ def _pick_codex_model(listed: Sequence[str]) -> str:
return str(inquirer.fuzzy(message="Model Codex starts on (type to filter):", choices=choices).execute())
def _apply_codex(ctx: click.Context, credential: StaticToken, listing: _Listing, model: str) -> None:
base_url: Final[str] = ctx.obj["base_url"]
def _apply_codex(base_url: str, credential: StaticToken, listing: _Listing, model: str) -> None:
_validated_model(model, listing, base_url)
settings_path: Final = codex_config_path(os.environ)
try:
@ -237,13 +236,12 @@ class _Setup:
def _choose_setup(
ctx: click.Context,
base_url: str,
target: str,
credential: StaticToken,
pick_model: Callable[[Sequence[str]], str | None],
pick_codex_model: Callable[[Sequence[str]], str],
) -> _Setup:
base_url: Final[str] = ctx.obj["base_url"]
listing: Final = _listed_models(base_url, credential.token, target)
model: Final = (
pick_model(tuple(item.source_model or item.id for item in listing.models))
@ -270,12 +268,15 @@ def interactive_configure(
credential: Final = resolve_credential(ctx, None)
except ClaudeSettingsError as e:
raise click.ClickException(str(e)) from e
setups: Final = tuple(_choose_setup(ctx, target, credential, pick_model, pick_codex_model) for target in targets)
base_url: Final[str] = ctx.obj["base_url"]
setups: Final = tuple(
_choose_setup(base_url, target, credential, pick_model, pick_codex_model) for target in targets
)
for setup in setups:
if setup.target == _CLAUDE_TARGET:
_apply_claude(ctx, credential, setup.listing, setup.model)
_apply_claude(base_url, credential, setup.listing, setup.model)
elif setup.model is not None:
_apply_codex(ctx, credential, setup.listing, setup.model)
_apply_codex(base_url, credential, setup.listing, setup.model)
class _ConnectionOptions(BaseModel):
@ -283,7 +284,8 @@ class _ConnectionOptions(BaseModel):
gateway_url: str | None = None
def _connection_context(ctx: click.Context, api_key: str | None, gateway_url: str | None) -> click.Context:
def _connection_settings(ctx: click.Context, api_key: str | None, gateway_url: str | None) -> CliContextObj:
"""The context object a subcommand runs with: its own --api-key / --gateway-url over the group's, over `lite`'s."""
ctx_obj: Final[CliContextObj] = ctx.obj
group: Final = (
_ConnectionOptions.model_validate(ctx.parent.params)
@ -300,7 +302,11 @@ def _connection_context(ctx: click.Context, api_key: str | None, gateway_url: st
"api_key": key if key is not None else ctx_obj.get("api_key"),
"api_key_from_token_file": False if key is not None else ctx_obj.get("api_key_from_token_file", False),
}
return click.Context(ctx.command, parent=ctx.parent, obj=connection)
return connection
def _connection_context(ctx: click.Context, settings: CliContextObj) -> click.Context:
return click.Context(ctx.command, parent=ctx.parent, obj=settings)
@click.group(name="configure", invoke_without_command=True)
@ -316,19 +322,19 @@ def configure_group(ctx: click.Context, api_key: str | None, gateway_url: str |
"""
if ctx.invoked_subcommand is not None:
return
connection: Final = _connection_context(ctx, api_key, gateway_url)
settings: Final = _connection_settings(ctx, api_key, gateway_url)
connection: Final = _connection_context(ctx, settings)
if not sys.stdin.isatty():
raise click.ClickException(
"`lite configure` asks questions, so it needs a terminal. Non-interactively, run "
"`lite configure claude --api-key <key> --model <model>` or "
"`lite configure codex --api-key <key> --model <model>`."
)
prompted: Final = (
connection
if connection.obj.get("base_url_explicit")
else _connection_context(connection, None, click.prompt("Gateway URL", default=connection.obj["base_url"]))
)
interactive_configure(prompted)
if settings.get("base_url_explicit"):
interactive_configure(connection)
return
prompted: Final = _connection_settings(connection, None, click.prompt("Gateway URL", default=settings["base_url"]))
interactive_configure(_connection_context(connection, prompted))
@click.group(name="unconfigure")
@ -356,9 +362,9 @@ def configure_claude(ctx: click.Context, api_key: str | None, model: str | None,
setting is kept, and what changed is recorded so `lite unconfigure claude` can put it back.
Assumes the proxy is already running.
"""
connection: Final = _connection_context(ctx, api_key, gateway_url)
credential, listing = _start(connection, api_key)
_apply_claude(connection, credential, listing, model)
settings: Final = _connection_settings(ctx, api_key, gateway_url)
credential, listing = _start(_connection_context(ctx, settings), settings["base_url"], api_key)
_apply_claude(settings["base_url"], credential, listing, model)
@configure_group.command(name="codex")
@ -368,9 +374,9 @@ def configure_claude(ctx: click.Context, api_key: str | None, model: str | None,
@click.pass_context
def configure_codex(ctx: click.Context, api_key: str | None, gateway_url: str | None, model: str) -> None:
"""Route plain `codex` through the gateway until `lite unconfigure codex`."""
connection: Final = _connection_context(ctx, api_key, gateway_url)
credential, listing = _start(connection, api_key, _CODEX_TARGET)
_apply_codex(connection, credential, listing, model)
settings: Final = _connection_settings(ctx, api_key, gateway_url)
credential, listing = _start(_connection_context(ctx, settings), settings["base_url"], api_key, _CODEX_TARGET)
_apply_codex(settings["base_url"], credential, listing, model)
@unconfigure_group.command(name="codex")

6
litellm/proxy/common_utils/callback_utils.py
View file

@ -713,7 +713,7 @@ def strip_callback_config(metadata: dict[str, object] | None) -> dict[str, objec
return {k: v for k, v in metadata.items() if k not in _CALLBACK_CONFIG_SLOTS}
def encrypt_callback_vars(metadata: Any) -> Any:
def encrypt_callback_vars(metadata: object) -> Any:
"""Return a deep copy of metadata with callback_vars values encrypted at rest.
Idempotent: a value that already decrypts cleanly is left unchanged so
@ -722,7 +722,7 @@ def encrypt_callback_vars(metadata: Any) -> Any:
return _transform_callback_vars(metadata, _encrypt_if_plaintext)
def decrypt_callback_vars(metadata: Any) -> Any:
def decrypt_callback_vars(metadata: object) -> Any:
"""Return a deep copy of metadata with callback_vars values decrypted.
Legacy plaintext rows pass through unchanged (decrypt failure → original).
@ -730,7 +730,7 @@ def decrypt_callback_vars(metadata: Any) -> Any:
return _transform_callback_vars(metadata, _decrypt_or_passthrough)
def _transform_callback_vars(metadata: object, transform: Callable[[str, Any], Any]) -> object:
def _transform_callback_vars(metadata: object, transform: Callable[[str, object], object]) -> object:
if not isinstance(metadata, dict):
return metadata
out: Final = copy.deepcopy(metadata)

20
litellm/proxy/db/db_spend_update_writer.py
View file

@ -15,7 +15,7 @@ import traceback
from collections.abc import Mapping, Sequence
from datetime import datetime, timedelta, timezone
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, cast, overload
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, TypeAlias, cast, overload
import litellm
from litellm._logging import verbose_proxy_logger
@ -115,6 +115,20 @@ class _SpendBatch(Protocol):
litellm_modelaccessgroupbudgettable: BatchTable
_EntitySpendTable: TypeAlias = Literal["litellm_tagtable", "litellm_agentstable", "litellm_modelaccessgroupbudgettable"]
def _entity_spend_table(batcher: _SpendBatch, table_accessor: _EntitySpendTable) -> BatchTable:
"""The batch table an entity type's spend increments are written to."""
match table_accessor:
case "litellm_tagtable":
return batcher.litellm_tagtable
case "litellm_agentstable":
return batcher.litellm_agentstable
case "litellm_modelaccessgroupbudgettable":
return batcher.litellm_modelaccessgroupbudgettable
class _SpendBatchManager(Protocol):
async def __aenter__(self) -> _SpendBatch: ...
@ -1750,7 +1764,7 @@ class DBSpendUpdateWriter:
async def _update_entity_spend_in_db(
entity_name: str,
transactions: dict[str, float] | None,
table_accessor: Literal["litellm_tagtable", "litellm_agentstable", "litellm_modelaccessgroupbudgettable"],
table_accessor: _EntitySpendTable,
where_field: str,
n_retry_times: int,
prisma_client: PrismaClient,
@ -1784,7 +1798,7 @@ class DBSpendUpdateWriter:
entity_id,
response_cost,
)
getattr(batcher, table_accessor).update_many(
_entity_spend_table(batcher, table_accessor).update_many(
where={where_field: entity_id},
data={"spend": {"increment": response_cost}},
)

36
litellm/proxy/guardrails/guardrail_hooks/akto/akto.py
View file

@ -11,10 +11,11 @@ import asyncio
import json
import os
from datetime import datetime
from typing import TYPE_CHECKING, Any, Final, Literal
from typing import TYPE_CHECKING, Final, Literal
import httpx
from fastapi import HTTPException
from typing_extensions import NotRequired, ReadOnly, TypedDict, Unpack
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import (
@ -25,12 +26,34 @@ from litellm.llms.custom_httpx.http_handler import (
get_async_httpx_client,
httpxSpecialProvider,
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.guardrails import GuardrailEventHooks, Mode
from litellm.types.utils import GenericGuardrailAPIInputs
if TYPE_CHECKING:
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
class _CustomGuardrailKwargs(TypedDict):
"""Keyword arguments forwarded verbatim to CustomGuardrail.__init__."""
guardrail_name: NotRequired[ReadOnly[str | None]]
event_hook: NotRequired[ReadOnly[GuardrailEventHooks | list[GuardrailEventHooks] | Mode | None]]
default_on: NotRequired[ReadOnly[bool]]
mask_request_content: NotRequired[ReadOnly[bool]]
mask_response_content: NotRequired[ReadOnly[bool]]
violation_message_template: NotRequired[ReadOnly[str | None]]
end_session_after_n_fails: NotRequired[ReadOnly[int | None]]
on_violation: NotRequired[ReadOnly[str | None]]
realtime_violation_message: NotRequired[ReadOnly[str | None]]
on_sensitive_data: NotRequired[ReadOnly[str | None]]
sensitive_data_route_to_model: NotRequired[ReadOnly[str | None]]
sticky_session_routing: NotRequired[ReadOnly[bool]]
run_in_parallel: NotRequired[ReadOnly[bool]]
scan_raw_request: NotRequired[ReadOnly[bool]]
only_scan_new_messages: NotRequired[ReadOnly[bool]]
supported_event_hooks: NotRequired[ReadOnly[list[GuardrailEventHooks]]]
HTTP_PROXY_PATH: Final = "/api/http-proxy"
AKTO_CONNECTOR_NAME: Final = "litellm"
DEFAULT_GUARDRAIL_TIMEOUT: Final = 5
@ -66,7 +89,7 @@ class AktoGuardrail(CustomGuardrail):
akto_vxlan_id: str | None = None,
unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
guardrail_timeout: int | None = None,
**kwargs: Any,
**kwargs: Unpack[_CustomGuardrailKwargs],
) -> None:
"""Initialize the Akto guardrail.
@ -96,8 +119,11 @@ class AktoGuardrail(CustomGuardrail):
self.akto_account_id = akto_account_id or os.environ.get("AKTO_ACCOUNT_ID", "1000000")
self.akto_vxlan_id = akto_vxlan_id or os.environ.get("AKTO_VXLAN_ID", "0")
kwargs["supported_event_hooks"] = list(self.get_supported_event_hooks())
super().__init__(**kwargs)
init_kwargs: Final[_CustomGuardrailKwargs] = {
**kwargs,
"supported_event_hooks": list(self.get_supported_event_hooks()),
}
super().__init__(**init_kwargs)
verbose_proxy_logger.debug(
"Akto guardrail initialized: base_url=%s fallback=%s",

9
litellm/proxy/guardrails/guardrail_hooks/custom_code/custom_code_guardrail.py
View file

@ -38,9 +38,10 @@ import asyncio
import threading
import time
from collections.abc import Callable, Mapping
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, cast
from typing import TYPE_CHECKING, Final, Literal, Optional, cast
from fastapi import HTTPException
from typing_extensions import TypedDict, Unpack
from litellm._logging import verbose_proxy_logger
from litellm.exceptions import ModifyResponseException
@ -74,6 +75,10 @@ class CustomCodeExecutionError(CustomCodeGuardrailError):
"""Raised when custom code fails during execution."""
class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
"""Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
class CustomCodeGuardrailConfigModel(GuardrailConfigModel):
"""Configuration parameters for the custom code guardrail."""
@ -109,7 +114,7 @@ class CustomCodeGuardrail(CustomGuardrail):
self,
custom_code: str,
guardrail_name: str | None = "custom_code",
**kwargs: Any,
**kwargs: Unpack[_CustomGuardrailOptions],
) -> None:
"""
Initialize the custom code guardrail.

12
litellm/proxy/guardrails/guardrail_hooks/lasso/lasso.py
View file

@ -121,7 +121,7 @@ class LassoGuardrail(CustomGuardrail):
super().__init__(**kwargs)
@staticmethod
def _get_field(obj: Any, field: str, default: object = None) -> Any:
def _get_field(obj: object, field: str, default: object = None) -> object:
"""Get a field from either a dict or a Pydantic object."""
if isinstance(obj, dict):
return obj.get(field, default)
@ -130,7 +130,7 @@ class LassoGuardrail(CustomGuardrail):
@staticmethod
def _extract_tool_call_fields(
call: object,
) -> tuple[str | None, str | None, dict[str, object] | None]:
) -> tuple[object, object, dict[str, object] | None]:
"""Extract (call_id, name, parsed_input) from a tool call.
Handles both dict-style and Pydantic object-style tool_calls.
@ -146,7 +146,7 @@ class LassoGuardrail(CustomGuardrail):
input_data: dict[str, object] | None = None
if args_str:
try:
parsed = json.loads(args_str)
parsed = json.loads(args_str) if isinstance(args_str, (str, bytes, bytearray)) else None
except (json.JSONDecodeError, TypeError):
parsed = None
if isinstance(parsed, dict):
@ -488,7 +488,7 @@ class LassoGuardrail(CustomGuardrail):
while preserving the original structure.
"""
# Index masked content by type so we can look up by id without caring about order.
masked_tool_use: Final[dict[str, dict[str, object]]] = {}
masked_tool_use: Final[dict[object, dict[str, object]]] = {}
masked_tool_result: Final[dict[str, str]] = {}
masked_text: Final[list[str]] = []
@ -565,7 +565,7 @@ class LassoGuardrail(CustomGuardrail):
def _update_tool_calls_from_masked(
self,
tool_calls: list[object],
masked_tool_use: dict[str, dict[str, object]],
masked_tool_use: Mapping[object, Mapping[str, object]],
) -> list[object]:
"""Replace tool_call arguments with masked values returned by Lasso."""
updated: Final = []
@ -922,7 +922,7 @@ class LassoGuardrail(CustomGuardrail):
) -> None:
"""Apply masking to the actual model response when mask=True and masked content is available."""
# Index masked tool_use blocks by id for O(1) lookup.
masked_tool_use: Final[dict[str, dict[str, object]]] = {}
masked_tool_use: Final[dict[object, dict[str, object]]] = {}
masked_text: Final[list[str]] = []
for masked_msg in masked_messages:
content = masked_msg.get("content")

23
litellm/proxy/guardrails/guardrail_hooks/vigil_guard/vigil_guard.py
View file

@ -3,7 +3,7 @@ from json import JSONDecodeError
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Protocol, TypeAlias, cast
import httpx
from typing_extensions import ReadOnly, TypedDict
from typing_extensions import ReadOnly, TypedDict, Unpack
from litellm._logging import verbose_proxy_logger
from litellm.exceptions import GuardrailRaisedException
@ -18,7 +18,8 @@ from litellm.llms.custom_httpx.http_handler import (
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import GenericGuardrailAPIInputs
from litellm.types.llms.openai import ChatCompletionToolCallChunk
from litellm.types.utils import ChatCompletionMessageToolCall, GenericGuardrailAPIInputs
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import (
@ -53,6 +54,7 @@ _METADATA_ALLOWLIST: Final = (
_FallbackMode: TypeAlias = Literal["fail_closed", "fail_open"]
_MetadataValue: TypeAlias = str | int | float | Sequence[str | int | float]
_ToolCalls: TypeAlias = list[ChatCompletionToolCallChunk] | list[ChatCompletionMessageToolCall]
class _AnalyzePayload(TypedDict):
@ -70,6 +72,12 @@ class _AnalysisView(TypedDict):
analysis: ReadOnly[Mapping[str, object]]
class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
"""Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
supported_event_hooks: ReadOnly[list[GuardrailEventHooks]]
class _AsyncPostHandler(Protocol):
def post(
self,
@ -93,7 +101,7 @@ class VigilGuardGuardrail(CustomGuardrail):
unreachable_fallback: str | None = None,
timeout: float | None = None,
async_handler: _AsyncPostHandler | None = None,
**kwargs: Any,
**kwargs: Unpack[_CustomGuardrailOptions],
) -> None:
resolved_base: Final = api_base or get_secret_str("VIGIL_GUARD_URL")
if not resolved_base:
@ -122,9 +130,12 @@ class VigilGuardGuardrail(CustomGuardrail):
llm_provider=httpxSpecialProvider.GuardrailCallback,
)
kwargs.setdefault("supported_event_hooks", list(self.get_supported_event_hooks()))
forwarded: Final[_CustomGuardrailOptions] = {
"supported_event_hooks": list(self.get_supported_event_hooks()),
**kwargs,
}
super().__init__(**kwargs)
super().__init__(**forwarded)
@staticmethod
def get_config_model() -> type["GuardrailConfigModel"] | None:
@ -264,7 +275,7 @@ class VigilGuardGuardrail(CustomGuardrail):
inputs: GenericGuardrailAPIInputs,
source: str,
final_texts: list[str],
final_tool_calls: Any,
final_tool_calls: _ToolCalls | None,
) -> GenericGuardrailAPIInputs:
if self.unreachable_fallback == "fail_open":
verbose_proxy_logger.error(

12
litellm/proxy/guardrails/guardrail_hooks/xecguard/xecguard.py
View file

@ -196,9 +196,9 @@ class XecGuardGuardrail(CustomGuardrail):
async def async_logging_hook(
self,
kwargs: dict,
result: Any,
result: object,
call_type: str,
) -> tuple[dict, Any]:
) -> tuple[dict, object]:
"""Observe-only scan for logging_only mode.
Never blocks, never raises - all errors are swallowed. Records a
@ -275,9 +275,9 @@ class XecGuardGuardrail(CustomGuardrail):
def logging_hook(
self,
kwargs: dict,
result: Any,
result: object,
call_type: str,
) -> tuple[dict, Any]:
) -> tuple[dict, object]:
"""Sync counterpart to ``async_logging_hook``.
Runs the async version on an available loop, swallowing every
@ -433,7 +433,7 @@ class XecGuardGuardrail(CustomGuardrail):
return {"role": role, "content": ""}
@staticmethod
def _synthesize_user_from_inputs(inputs: Any) -> dict | None:
def _synthesize_user_from_inputs(inputs: object) -> dict | None:
if not isinstance(inputs, dict):
return None
texts: Final = inputs.get("texts")
@ -490,7 +490,7 @@ class XecGuardGuardrail(CustomGuardrail):
return None
@staticmethod
def _content_to_text(content: Any) -> str | None:
def _content_to_text(content: object) -> str | None:
if isinstance(content, str) and content:
return content
if isinstance(content, list):

5
litellm/proxy/management_endpoints/model_management_endpoints.py
View file

@ -455,7 +455,6 @@ async def _auto_router_capability_slot(
ENFORCE_RPM_TPM_ON_MODEL_ADD_SETTING: Final = "enforce_rpm_tpm_on_model_add"
_REQUIRED_RATE_LIMIT_FIELDS: Final = ("rpm", "tpm")
def _raise_if_rate_limits_required_but_missing(*, litellm_params: GenericLiteLLMParams, enforced: bool) -> None:
@ -470,8 +469,8 @@ def _raise_if_rate_limits_required_but_missing(*, litellm_params: GenericLiteLLM
return
missing: Final = tuple(
field
for field in _REQUIRED_RATE_LIMIT_FIELDS
if (value := getattr(litellm_params, field)) is None or value <= 0
for field, value in (("rpm", litellm_params.rpm), ("tpm", litellm_params.tpm))
if value is None or value <= 0
)
if not missing:
return

3
litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_passthrough_logging_handler.py
View file

@ -28,6 +28,7 @@ from litellm.proxy.pass_through_endpoints.llm_provider_handlers.batch_attributio
optional_str,
request_tags_from_metadata,
)
from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
from litellm.types.utils import (
Choices,
EmbeddingResponse,
@ -47,8 +48,6 @@ else:
PassThroughEndpointLogging = Any
LiteLLMBatch = Any
EndpointType = Any
class VertexPassthroughLoggingHandler:
@staticmethod

2
litellm/proxy/response_api_endpoints/endpoints.py
View file

@ -1357,7 +1357,7 @@ async def _read_ws_model_from_first_frame(
return model, first_message
def _extract_model_from_first_ws_event(first_event: Any) -> str | None:
def _extract_model_from_first_ws_event(first_event: object) -> str | None:
"""Extract model from a response.create WS event, handling flat and nested formats.
Flat: {"type": "response.create", "model": "gpt-4o", ...}

28
litellm/proxy/video_endpoints/endpoints.py
View file

@ -89,7 +89,7 @@ async def video_generation(
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
generated: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -114,6 +114,8 @@ async def video_generation(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
else:
return generated
@router.get(
@ -174,7 +176,7 @@ async def video_list(
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
listed: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -199,6 +201,8 @@ async def video_list(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
else:
return listed
@router.get(
@ -272,7 +276,7 @@ async def video_status(
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
status: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -297,6 +301,8 @@ async def video_status(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
else:
return status
@router.get(
@ -478,7 +484,7 @@ async def video_remix(
# Process request using ProxyBaseLLMRequestProcessing
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
remixed: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -503,6 +509,8 @@ async def video_remix(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
else:
return remixed
@router.post(
@ -571,7 +579,7 @@ async def video_create_character(
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
response = await processor.base_process_llm_request(
response: object = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -678,7 +686,7 @@ async def video_get_character(
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
response = await processor.base_process_llm_request(
response: object = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -789,7 +797,7 @@ async def video_edit(
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
edited: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -814,6 +822,8 @@ async def video_edit(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
else:
return edited
@router.post(
@ -884,7 +894,7 @@ async def video_extension(
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
return await processor.base_process_llm_request(
extended: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -909,3 +919,5 @@ async def video_extension(
proxy_logging_obj=proxy_logging_obj,
version=version,
)
else:
return extended

10
litellm/responses/mcp/litellm_proxy_mcp_handler.py
View file

@ -1266,14 +1266,14 @@ class LiteLLM_Proxy_MCP_Handler:
return tool_execution_events
@staticmethod
def _prepare_initial_call_params(call_params: dict[str, Any], should_auto_execute: bool) -> dict[str, Any]:
def _prepare_initial_call_params(call_params: Mapping[str, object], should_auto_execute: bool) -> dict[str, Any]:
"""
Prepare call parameters for the initial LLM call.
For auto-execute scenarios, we need to disable streaming for the initial call
so we can process the tool calls before streaming the final response.
"""
initial_params: Final = call_params.copy()
initial_params: Final = dict(call_params)
if should_auto_execute:
# Disable streaming for initial call when auto-executing tools
@ -1282,14 +1282,16 @@ class LiteLLM_Proxy_MCP_Handler:
return initial_params
@staticmethod
def _prepare_follow_up_call_params(call_params: dict[str, Any], original_stream_setting: bool) -> dict[str, Any]:
def _prepare_follow_up_call_params(
call_params: Mapping[str, object], original_stream_setting: bool
) -> dict[str, Any]:
"""
Prepare call parameters for the follow-up LLM call after tool execution.
Restores the original streaming setting and removes tool_choice since
we're now providing tool results, not requesting tool calls.
"""
follow_up_params: Final = call_params.copy()
follow_up_params: Final = dict(call_params)
# Restore original streaming setting for follow-up call
follow_up_params["stream"] = original_stream_setting

2
litellm/responses/streaming_iterator.py
View file

@ -2353,7 +2353,7 @@ class ManagedResponsesWebSocketHandler:
await self.websocket.send_text(serialized)
@staticmethod
def _build_base_call_kwargs(msg_obj: _MutableJsonObject) -> dict[str, Any]:
def _build_base_call_kwargs(msg_obj: _MutableJsonObject) -> dict[str, object]:
"""
Extract Responses API params from the event, handling both wire formats:
Nested: {"type": "response.create", "response": {"input": [...], ...}}

32
litellm/router_strategy/complexity_router/complexity_router.py
View file

@ -363,7 +363,7 @@ def _append_custom_keywords(base_keywords: list[str], custom_keywords: list[str]
return [*base_keywords, *deduped_custom.values()]
def _parent_session_kwargs(request_kwargs: Mapping[str, Any] | None) -> Mapping[str, Any]:
def _parent_session_kwargs(request_kwargs: Mapping[str, object] | None) -> Mapping[str, Any]:
kwargs: Final = request_kwargs or {}
return {k: kwargs[k] for k in ("litellm_session_id", "litellm_trace_id") if kwargs.get(k) is not None}
@ -1165,7 +1165,7 @@ class ComplexityRouter(CustomLogger):
self,
model_name: str,
litellm_router_instance: Router,
complexity_router_config: dict[str, Any] | None = None,
complexity_router_config: Mapping[str, object] | None = None,
default_model: str | None = None,
derive_savings_baseline: bool = True,
):
@ -1736,7 +1736,7 @@ class ComplexityRouter(CustomLogger):
self,
prompt: str,
system_prompt: str | None,
request_kwargs: dict[str, Any] | None, # mutable-ok: handed to _classify_with_llm as-is
request_kwargs: dict[str, object] | None, # mutable-ok: handed to _classify_with_llm as-is
messages: Sequence[Mapping[str, object]] | None,
) -> ClassificationOutcome:
"""Score locally, and only pay for the classifier call when the scorer did not confidently
@ -1769,7 +1769,7 @@ class ComplexityRouter(CustomLogger):
self,
prompt: str,
system_prompt: str | None,
request_kwargs: dict[str, Any] | None, # mutable-ok: handed to _classify_with_llm as-is
request_kwargs: dict[str, object] | None, # mutable-ok: handed to _classify_with_llm as-is
messages: Sequence[Mapping[str, object]] | None,
) -> ClassificationOutcome:
"""Score locally, and only pay for the classifier when the score sits near a tier boundary.
@ -1824,7 +1824,7 @@ class ComplexityRouter(CustomLogger):
self,
prompt: str,
system_prompt: str | None,
request_kwargs: dict[str, Any] | None, # mutable-ok: handed to _classify_with_llm as-is
request_kwargs: dict[str, object] | None, # mutable-ok: handed to _classify_with_llm as-is
messages: Sequence[Mapping[str, object]] | None,
scored: ClassificationOutcome | None = None,
) -> ClassificationOutcome:
@ -1902,8 +1902,8 @@ class ComplexityRouter(CustomLogger):
self,
prompt: str,
system_prompt: str | None,
request_kwargs: dict[str, Any] | None, # mutable-ok: handed to resolve_structured_messages as-is
raw_messages: list[dict[str, Any]] | None, # mutable-ok: same shape _run_routing_plugins receives
request_kwargs: dict[str, object] | None, # mutable-ok: handed to resolve_structured_messages as-is
raw_messages: list[dict[str, object]] | None, # mutable-ok: same shape _run_routing_plugins receives
) -> ClassificationOutcome:
from litellm.litellm_core_utils.prompt_templates.factory import resolve_structured_messages
from litellm.types.router import RoutingContext
@ -2262,8 +2262,8 @@ class ComplexityRouter(CustomLogger):
async def _pick_model_for_tier(
self,
tier: ComplexityTier | str,
raw_messages: list[dict[str, Any]] | None,
resolved_messages: list[dict[str, Any]] | None,
raw_messages: list[dict[str, object]] | None,
resolved_messages: list[dict[str, object]] | None,
request_kwargs: dict,
allowed_models: tuple[str, ...] | None = None,
) -> str:
@ -2373,7 +2373,7 @@ class ComplexityRouter(CustomLogger):
self,
classified_tier: ComplexityTier | str,
user_message: str,
request_kwargs: dict[str, Any] | None = None,
request_kwargs: dict[str, object] | None = None,
hard_floor: ComplexityTier | str | None = None,
hard_ceiling: ComplexityTier | str | None = None,
fit_filter: frozenset[str] | None = None,
@ -2903,7 +2903,7 @@ class ComplexityRouter(CustomLogger):
async def _gate_response_modality(
self,
response: PreRoutingHookResponse,
messages: list[dict[str, Any]] | None, # mutable-ok: forwarded verbatim to the list-typed re-pick
messages: list[dict[str, object]] | None, # mutable-ok: forwarded verbatim to the list-typed re-pick
resolved_messages: Sequence[Mapping[str, object]] | None,
request_kwargs: dict, # mutable-ok: same shape the hook receives
context_fit: _RequestContextFit | None = None,
@ -3093,7 +3093,7 @@ class ComplexityRouter(CustomLogger):
async def _gate_response_health(
self,
response: PreRoutingHookResponse,
messages: list[dict[str, Any]] | None, # mutable-ok: forwarded verbatim to the list-typed re-pick
messages: list[dict[str, object]] | None, # mutable-ok: forwarded verbatim to the list-typed re-pick
input: str | list | None, # mutable-ok: mirrors the owner's own input parameter, which this forwards verbatim
resolved_messages: Sequence[Mapping[str, object]] | None,
request_kwargs: dict, # mutable-ok: same shape the hook receives
@ -3405,9 +3405,9 @@ class ComplexityRouter(CustomLogger):
def _resolve_messages(
self,
messages: list[dict[str, Any]] | None,
messages: list[dict[str, object]] | None,
request_kwargs: dict,
) -> list[dict[str, Any]] | None:
) -> list[dict[str, object]] | None:
"""
Resolve messages from the request, converting from other formats if needed.
@ -3422,7 +3422,7 @@ class ComplexityRouter(CustomLogger):
@staticmethod
def _extract_user_message_and_system_prompt(
messages: list[dict[str, Any]],
messages: Sequence[Mapping[str, object]],
) -> tuple[str | None, str | None]:
"""
Deprecated: use _extract_current_ask_and_system_prompt instead.
@ -3729,7 +3729,7 @@ class ComplexityRouter(CustomLogger):
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,
conversation_continuing: bool = True,

16
litellm/types/router.py
View file

@ -248,7 +248,7 @@ class ModelInfo(MirroredPricingParams):
# Custom .get() method to access attributes with a default value if the attribute doesn't exist
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key) -> object:
# Allow dictionary-style access to attributes
return getattr(self, key)
@ -358,7 +358,7 @@ class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams):
model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
merge_reasoning_content_in_choices: bool | None = False
model_info: dict | None = None
mock_response: str | ModelResponse | Exception | Any | None = None
mock_response: str | ModelResponse | Exception | object | None = None
# tag-based routing
tags: list[str] | None = None
@ -435,7 +435,7 @@ class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams):
# Custom .get() method to access attributes with a default value if the attribute doesn't exist
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key) -> object:
# Allow dictionary-style access to attributes
return getattr(self, key)
@ -460,7 +460,7 @@ class LiteLLM_Params(GenericLiteLLMParams):
# Custom .get() method to access attributes with a default value if the attribute doesn't exist
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key) -> object:
# Allow dictionary-style access to attributes
return getattr(self, key)
@ -1043,11 +1043,11 @@ class RoutingContext(BaseModel):
plugins that need the exact original payload can read `raw_messages`.
"""
raw_messages: list[dict[str, Any]]
structured_messages: list[dict[str, Any]]
raw_messages: list[dict[str, object]]
structured_messages: list[dict[str, object]]
candidate_models: list[str]
metadata: dict[str, Any] = Field(default_factory=dict)
signals: dict[str, Any] = Field(default_factory=dict)
metadata: dict[str, object] = Field(default_factory=dict)
signals: dict[str, object] = Field(default_factory=dict)
@runtime_checkable

5
litellm/vector_stores/vector_store_registry.py
View file

@ -112,9 +112,8 @@ class VectorStoreRegistry:
Dynamically extracts all parameters defined in VECTOR_STORE_OPENAI_PARAMS.
"""
# Get the list of supported param names from the Literal type
supported_params: Final = tuple(
param for param in get_args(VECTOR_STORE_OPENAI_PARAMS) if isinstance(param, str)
)
declared_params: Final[tuple[object, ...]] = get_args(VECTOR_STORE_OPENAI_PARAMS)
supported_params: Final = tuple(param for param in declared_params if isinstance(param, str))
# Extract only the params that exist in the tool
kwargs: Final = {param: tool.get(param) for param in supported_params if param in tool}