From da7d5fe1284da59502cb6738d853acc8c431af40 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Mon, 14 Sep 2026 11:13:39 +0000 Subject: [PATCH] refactor: replace Any with precise types across 54 modules 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. --- .../proxy/common_utils/check_batch_cost.py | 127 +++++++++++------- .../common_utils/check_responses_cost.py | 31 ++++- .../proxy/hooks/managed_files.py | 22 ++- litellm/caching/caching.py | 2 +- litellm/caching/caching_handler.py | 10 +- litellm/caching/redis_cache.py | 4 +- .../handler.py | 6 +- .../transformation.py | 30 +++-- litellm/cost_calculator.py | 6 +- litellm/integrations/braintrust_logging.py | 8 +- litellm/integrations/custom_guardrail.py | 2 +- .../integrations/datadog/datadog_llm_obs.py | 22 +-- litellm/integrations/galileo.py | 11 +- litellm/integrations/langfuse/langfuse.py | 8 +- .../llm_response_utils/response_metadata.py | 4 +- .../prompt_templates/common_utils.py | 39 ++---- .../a2a/chat/guardrail_translation/handler.py | 2 +- litellm/llms/anthropic/chat/transformation.py | 24 ++-- litellm/llms/anthropic/common_utils.py | 10 +- .../messages/agentic_streaming_iterator.py | 16 ++- litellm/llms/anthropic/files/handler.py | 6 +- .../bedrock/chat/converse_transformation.py | 2 +- .../llms/chatgpt/responses/transformation.py | 22 +-- litellm/llms/cohere/chat/transformation.py | 2 +- .../llms/databricks/chat/transformation.py | 10 +- .../llms/fireworks_ai/chat/transformation.py | 4 +- litellm/llms/gemini/count_tokens/handler.py | 2 +- .../llms/gemini/image_edit/transformation.py | 13 +- litellm/llms/gigachat/chat/transformation.py | 14 +- .../huggingface/embedding/transformation.py | 10 +- .../chat/guardrail_translation/handler.py | 20 +-- litellm/llms/openai/videos/transformation.py | 6 +- .../openrouter/image_edit/transformation.py | 11 +- .../perplexity/embedding/transformation.py | 2 +- .../llms/vertex_ai/gemini/transformation.py | 4 +- .../mcp_server/mcp_server_manager.py | 2 +- litellm/proxy/auth/auth_utils.py | 26 ++-- .../proxy/client/cli/commands/configure.py | 60 +++++---- litellm/proxy/common_utils/callback_utils.py | 6 +- litellm/proxy/db/db_spend_update_writer.py | 20 ++- .../guardrails/guardrail_hooks/akto/akto.py | 36 ++++- .../custom_code/custom_code_guardrail.py | 9 +- .../guardrails/guardrail_hooks/lasso/lasso.py | 12 +- .../vigil_guard/vigil_guard.py | 23 +++- .../guardrail_hooks/xecguard/xecguard.py | 12 +- .../model_management_endpoints.py | 5 +- .../vertex_passthrough_logging_handler.py | 3 +- .../proxy/response_api_endpoints/endpoints.py | 2 +- litellm/proxy/video_endpoints/endpoints.py | 28 ++-- .../mcp/litellm_proxy_mcp_handler.py | 10 +- litellm/responses/streaming_iterator.py | 2 +- .../complexity_router/complexity_router.py | 32 ++--- litellm/types/router.py | 16 +-- .../vector_stores/vector_store_registry.py | 5 +- 54 files changed, 485 insertions(+), 336 deletions(-) diff --git a/enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py b/enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py index 13e9e5093a8..3ea9b7d9bfd 100644 --- a/enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py +++ b/enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py @@ -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, ) diff --git a/enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py b/enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py index 06cf5fcf82f..1bc41f2aa5b 100644 --- a/enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py +++ b/enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py @@ -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"}, ) diff --git a/enterprise/litellm_enterprise/proxy/hooks/managed_files.py b/enterprise/litellm_enterprise/proxy/hooks/managed_files.py index 4899b87da7a..5204894bee6 100644 --- a/enterprise/litellm_enterprise/proxy/hooks/managed_files.py +++ b/enterprise/litellm_enterprise/proxy/hooks/managed_files.py @@ -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 diff --git a/litellm/caching/caching.py b/litellm/caching/caching.py index d6dd2a073af..813ee655a1d 100644 --- a/litellm/caching/caching.py +++ b/litellm/caching/caching.py @@ -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, diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index 139dcf058d2..2c4f80b8708 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -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, ): diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index d0cefcb6086..e2b48159e01 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -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. diff --git a/litellm/completion_extras/litellm_responses_transformation/handler.py b/litellm/completion_extras/litellm_responses_transformation/handler.py index f494d6610a1..642a78789b2 100644 --- a/litellm/completion_extras/litellm_responses_transformation/handler.py +++ b/litellm/completion_extras/litellm_responses_transformation/handler.py @@ -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 diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 5a6debc4af5..a5d67273768 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -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"] diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 440d97d13be..d93bf1e8769 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -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 diff --git a/litellm/integrations/braintrust_logging.py b/litellm/integrations/braintrust_logging.py index aaf72a0bc4e..501f5749ea4 100644 --- a/litellm/integrations/braintrust_logging.py +++ b/litellm/integrations/braintrust_logging.py @@ -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 {} diff --git a/litellm/integrations/custom_guardrail.py b/litellm/integrations/custom_guardrail.py index 39adea30828..eaee84fae93 100644 --- a/litellm/integrations/custom_guardrail.py +++ b/litellm/integrations/custom_guardrail.py @@ -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) diff --git a/litellm/integrations/datadog/datadog_llm_obs.py b/litellm/integrations/datadog/datadog_llm_obs.py index c64a12c6d75..3afe38ea075 100644 --- a/litellm/integrations/datadog/datadog_llm_obs.py +++ b/litellm/integrations/datadog/datadog_llm_obs.py @@ -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 diff --git a/litellm/integrations/galileo.py b/litellm/integrations/galileo.py index b27618993a3..010f8ad8ef2 100644 --- a/litellm/integrations/galileo.py +++ b/litellm/integrations/galileo.py @@ -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, diff --git a/litellm/integrations/langfuse/langfuse.py b/litellm/integrations/langfuse/langfuse.py index b75369965de..c89506facbd 100644 --- a/litellm/integrations/langfuse/langfuse.py +++ b/litellm/integrations/langfuse/langfuse.py @@ -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, diff --git a/litellm/litellm_core_utils/llm_response_utils/response_metadata.py b/litellm/litellm_core_utils/llm_response_utils/response_metadata.py index c83c266a17e..d279fb9e259 100644 --- a/litellm/litellm_core_utils/llm_response_utils/response_metadata.py +++ b/litellm/litellm_core_utils/llm_response_utils/response_metadata.py @@ -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 {} diff --git a/litellm/litellm_core_utils/prompt_templates/common_utils.py b/litellm/litellm_core_utils/prompt_templates/common_utils.py index 2485896184e..f2b03e72497 100644 --- a/litellm/litellm_core_utils/prompt_templates/common_utils.py +++ b/litellm/litellm_core_utils/prompt_templates/common_utils.py @@ -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]: diff --git a/litellm/llms/a2a/chat/guardrail_translation/handler.py b/litellm/llms/a2a/chat/guardrail_translation/handler.py index 5c30ff4747a..92dc49ea9c1 100644 --- a/litellm/llms/a2a/chat/guardrail_translation/handler.py +++ b/litellm/llms/a2a/chat/guardrail_translation/handler.py @@ -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. diff --git a/litellm/llms/anthropic/chat/transformation.py b/litellm/llms/anthropic/chat/transformation.py index 0f99441a115..6899c334618 100644 --- a/litellm/llms/anthropic/chat/transformation.py +++ b/litellm/llms/anthropic/chat/transformation.py @@ -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, } diff --git a/litellm/llms/anthropic/common_utils.py b/litellm/llms/anthropic/common_utils.py index 87c4ec8938e..7ec15ebd1f4 100644 --- a/litellm/llms/anthropic/common_utils.py +++ b/litellm/llms/anthropic/common_utils.py @@ -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) diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/agentic_streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/messages/agentic_streaming_iterator.py index 171f5156594..306041d9949 100644 --- a/litellm/llms/anthropic/experimental_pass_through/messages/agentic_streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/messages/agentic_streaming_iterator.py @@ -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 diff --git a/litellm/llms/anthropic/files/handler.py b/litellm/llms/anthropic/files/handler.py index dfd62ca575b..e4c75a704ec 100644 --- a/litellm/llms/anthropic/files/handler.py +++ b/litellm/llms/anthropic/files/handler.py @@ -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", "") diff --git a/litellm/llms/bedrock/chat/converse_transformation.py b/litellm/llms/bedrock/chat/converse_transformation.py index fa18361e44c..aabd4f1afdc 100644 --- a/litellm/llms/bedrock/chat/converse_transformation.py +++ b/litellm/llms/bedrock/chat/converse_transformation.py @@ -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 diff --git a/litellm/llms/chatgpt/responses/transformation.py b/litellm/llms/chatgpt/responses/transformation.py index b96e06be3d8..9774b762396 100644 --- a/litellm/llms/chatgpt/responses/transformation.py +++ b/litellm/llms/chatgpt/responses/transformation.py @@ -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) diff --git a/litellm/llms/cohere/chat/transformation.py b/litellm/llms/cohere/chat/transformation.py index 319603b0dad..fa46bd7f6cf 100644 --- a/litellm/llms/cohere/chat/transformation.py +++ b/litellm/llms/cohere/chat/transformation.py @@ -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) diff --git a/litellm/llms/databricks/chat/transformation.py b/litellm/llms/databricks/chat/transformation.py index 82c3b5d91d3..dd257cd68b0 100644 --- a/litellm/llms/databricks/chat/transformation.py +++ b/litellm/llms/databricks/chat/transformation.py @@ -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 = [] diff --git a/litellm/llms/fireworks_ai/chat/transformation.py b/litellm/llms/fireworks_ai/chat/transformation.py index 05160d83c12..1e9082a1ef1 100644 --- a/litellm/llms/fireworks_ai/chat/transformation.py +++ b/litellm/llms/fireworks_ai/chat/transformation.py @@ -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, diff --git a/litellm/llms/gemini/count_tokens/handler.py b/litellm/llms/gemini/count_tokens/handler.py index cb2be2c860e..c2f0ef473ae 100644 --- a/litellm/llms/gemini/count_tokens/handler.py +++ b/litellm/llms/gemini/count_tokens/handler.py @@ -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. diff --git a/litellm/llms/gemini/image_edit/transformation.py b/litellm/llms/gemini/image_edit/transformation.py index e6c22dc60b4..31d3963c70c 100644 --- a/litellm/llms/gemini/image_edit/transformation.py +++ b/litellm/llms/gemini/image_edit/transformation.py @@ -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 diff --git a/litellm/llms/gigachat/chat/transformation.py b/litellm/llms/gigachat/chat/transformation.py index 89920ebd27b..d9250ea8836 100644 --- a/litellm/llms/gigachat/chat/transformation.py +++ b/litellm/llms/gigachat/chat/transformation.py @@ -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 diff --git a/litellm/llms/huggingface/embedding/transformation.py b/litellm/llms/huggingface/embedding/transformation.py index f6fe7f2fa10..33b0e21e326 100644 --- a/litellm/llms/huggingface/embedding/transformation.py +++ b/litellm/llms/huggingface/embedding/transformation.py @@ -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, diff --git a/litellm/llms/openai/chat/guardrail_translation/handler.py b/litellm/llms/openai/chat/guardrail_translation/handler.py index 58ff03e6a0d..665215303d9 100644 --- a/litellm/llms/openai/chat/guardrail_translation/handler.py +++ b/litellm/llms/openai/chat/guardrail_translation/handler.py @@ -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: diff --git a/litellm/llms/openai/videos/transformation.py b/litellm/llms/openai/videos/transformation.py index 94dc30f41e5..9a4b030993f 100644 --- a/litellm/llms/openai/videos/transformation.py +++ b/litellm/llms/openai/videos/transformation.py @@ -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. diff --git a/litellm/llms/openrouter/image_edit/transformation.py b/litellm/llms/openrouter/image_edit/transformation.py index b01c25aad0c..3d46277a69e 100644 --- a/litellm/llms/openrouter/image_edit/transformation.py +++ b/litellm/llms/openrouter/image_edit/transformation.py @@ -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 diff --git a/litellm/llms/perplexity/embedding/transformation.py b/litellm/llms/perplexity/embedding/transformation.py index a911fa62719..c93206db2bb 100644 --- a/litellm/llms/perplexity/embedding/transformation.py +++ b/litellm/llms/perplexity/embedding/transformation.py @@ -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 diff --git a/litellm/llms/vertex_ai/gemini/transformation.py b/litellm/llms/vertex_ai/gemini/transformation.py index 13e2238fdf6..e3cc3bbb2dc 100644 --- a/litellm/llms/vertex_ai/gemini/transformation.py +++ b/litellm/llms/vertex_ai/gemini/transformation.py @@ -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: diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py index fb0c623473a..f640db7e5dc 100644 --- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py +++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py @@ -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, diff --git a/litellm/proxy/auth/auth_utils.py b/litellm/proxy/auth/auth_utils.py index be65c3b39ec..bd248da52ef 100644 --- a/litellm/proxy/auth/auth_utils.py +++ b/litellm/proxy/auth/auth_utils.py @@ -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]: diff --git a/litellm/proxy/client/cli/commands/configure.py b/litellm/proxy/client/cli/commands/configure.py index 7988f8aef3c..539aa2581f5 100644 --- a/litellm/proxy/client/cli/commands/configure.py +++ b/litellm/proxy/client/cli/commands/configure.py @@ -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 --model ` or " "`lite configure codex --api-key --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") diff --git a/litellm/proxy/common_utils/callback_utils.py b/litellm/proxy/common_utils/callback_utils.py index 561a53409f4..eafbb1ef95f 100644 --- a/litellm/proxy/common_utils/callback_utils.py +++ b/litellm/proxy/common_utils/callback_utils.py @@ -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) diff --git a/litellm/proxy/db/db_spend_update_writer.py b/litellm/proxy/db/db_spend_update_writer.py index eaa03c5d7f7..19af995932a 100644 --- a/litellm/proxy/db/db_spend_update_writer.py +++ b/litellm/proxy/db/db_spend_update_writer.py @@ -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}}, ) diff --git a/litellm/proxy/guardrails/guardrail_hooks/akto/akto.py b/litellm/proxy/guardrails/guardrail_hooks/akto/akto.py index 2c27531cea1..a7f45a37ae6 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/akto/akto.py +++ b/litellm/proxy/guardrails/guardrail_hooks/akto/akto.py @@ -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", diff --git a/litellm/proxy/guardrails/guardrail_hooks/custom_code/custom_code_guardrail.py b/litellm/proxy/guardrails/guardrail_hooks/custom_code/custom_code_guardrail.py index d5ef1e949b8..252b94b76c8 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/custom_code/custom_code_guardrail.py +++ b/litellm/proxy/guardrails/guardrail_hooks/custom_code/custom_code_guardrail.py @@ -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. diff --git a/litellm/proxy/guardrails/guardrail_hooks/lasso/lasso.py b/litellm/proxy/guardrails/guardrail_hooks/lasso/lasso.py index cf5da27e9ca..63821428c62 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/lasso/lasso.py +++ b/litellm/proxy/guardrails/guardrail_hooks/lasso/lasso.py @@ -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") diff --git a/litellm/proxy/guardrails/guardrail_hooks/vigil_guard/vigil_guard.py b/litellm/proxy/guardrails/guardrail_hooks/vigil_guard/vigil_guard.py index a5945a39589..e807da7079e 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/vigil_guard/vigil_guard.py +++ b/litellm/proxy/guardrails/guardrail_hooks/vigil_guard/vigil_guard.py @@ -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( diff --git a/litellm/proxy/guardrails/guardrail_hooks/xecguard/xecguard.py b/litellm/proxy/guardrails/guardrail_hooks/xecguard/xecguard.py index 831df43692b..f4330ad6aa9 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/xecguard/xecguard.py +++ b/litellm/proxy/guardrails/guardrail_hooks/xecguard/xecguard.py @@ -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): diff --git a/litellm/proxy/management_endpoints/model_management_endpoints.py b/litellm/proxy/management_endpoints/model_management_endpoints.py index 2234e825090..50a498cf9bf 100644 --- a/litellm/proxy/management_endpoints/model_management_endpoints.py +++ b/litellm/proxy/management_endpoints/model_management_endpoints.py @@ -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 diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_passthrough_logging_handler.py index 119a53c2411..ce44596c19c 100644 --- a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_passthrough_logging_handler.py +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_passthrough_logging_handler.py @@ -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 diff --git a/litellm/proxy/response_api_endpoints/endpoints.py b/litellm/proxy/response_api_endpoints/endpoints.py index 5907ffc64eb..7f618526f11 100644 --- a/litellm/proxy/response_api_endpoints/endpoints.py +++ b/litellm/proxy/response_api_endpoints/endpoints.py @@ -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", ...} diff --git a/litellm/proxy/video_endpoints/endpoints.py b/litellm/proxy/video_endpoints/endpoints.py index 66071c05b4f..fe966c2e31a 100644 --- a/litellm/proxy/video_endpoints/endpoints.py +++ b/litellm/proxy/video_endpoints/endpoints.py @@ -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 diff --git a/litellm/responses/mcp/litellm_proxy_mcp_handler.py b/litellm/responses/mcp/litellm_proxy_mcp_handler.py index a5021e2f777..c4bbf8c8b25 100644 --- a/litellm/responses/mcp/litellm_proxy_mcp_handler.py +++ b/litellm/responses/mcp/litellm_proxy_mcp_handler.py @@ -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 diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 40ff88fc557..e1cf7847972 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -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": [...], ...}} diff --git a/litellm/router_strategy/complexity_router/complexity_router.py b/litellm/router_strategy/complexity_router/complexity_router.py index 9deccc9a468..eea0b2ec564 100644 --- a/litellm/router_strategy/complexity_router/complexity_router.py +++ b/litellm/router_strategy/complexity_router/complexity_router.py @@ -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, diff --git a/litellm/types/router.py b/litellm/types/router.py index c7363502017..f0d2405c88f 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -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 diff --git a/litellm/vector_stores/vector_store_registry.py b/litellm/vector_stores/vector_store_registry.py index b71d6784873..c7aed77286c 100644 --- a/litellm/vector_stores/vector_store_registry.py +++ b/litellm/vector_stores/vector_store_registry.py @@ -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}