refactor(typing): replace Any with proven types in 65 backend files

Typing-only pass over backend modules that carried the most reportAny and
reportExplicitAny errors. Every new annotation is backed by a construction
site, a call site, or an isinstance narrowing that already existed; untyped
JSON boundaries were left alone rather than declared without validation.

Tree-wide basedpyright errors drop 138,481 to 138,007. reportAny drops 8,854
to 8,645 and reportExplicitAny drops 3,119 to 2,814.

litellm/a2a_protocol/providers/bedrock_agentcore/transformation.py

@@ -7,7 +7,7 @@ and signs requests via AmazonAgentCoreConfig (SigV4 or JWT).
 
 import json
 from collections.abc import AsyncIterator, Mapping
-from typing import Any, Final
+from typing import Any, Final, Protocol
 
 from litellm._logging import verbose_logger
 from litellm.llms.bedrock.chat.agentcore.transformation import AmazonAgentCoreConfig
@@ -35,6 +35,12 @@ _RESERVED_PREFIX_HEADERS: Final[tuple[str, ...]] = (
 )
 
 
+class _SSELineSource(Protocol):
+    """Minimal streaming-response surface used to read SSE lines."""
+
+    def aiter_lines(self) -> AsyncIterator[str]: ...
+
+
 def _filter_reserved_headers(
     agent_extra_headers: Mapping[str, str] | None,
 ) -> dict[str, str] | None:
@@ -77,7 +83,7 @@ class BedrockAgentCoreA2ATransformation:
     @staticmethod
     def get_url_and_signed_request(
         request_id: str,
-        params: dict[str, Any],
+        params: Mapping[str, object],
         litellm_params: dict[str, Any],
         method: str = "message/send",
         stream: bool = False,
@@ -170,7 +176,7 @@ class BedrockAgentCoreA2ATransformation:
         return url, signed_headers, signed_body
 
     @staticmethod
-    async def parse_sse_events(response: Any) -> AsyncIterator[dict[str, Any]]:
+    async def parse_sse_events(response: _SSELineSource) -> AsyncIterator[dict[str, Any]]:
         """
         Parse SSE events from an httpx streaming response.
 

litellm/caching/redis_semantic_cache.py

@@ -116,7 +116,7 @@ class RedisSemanticCache(BaseCache):
                 password = password or os.environ["REDIS_PASSWORD"]
             except KeyError as e:
                 # Raise a more informative exception if any of the required keys are missing
-                missing_var: Final = e.args[0]
+                missing_var: Final[object] = e.args[0]
                 raise ValueError(
                     f"Missing required Redis configuration: {missing_var}. Provide {missing_var} or redis_url."
                 ) from e
@@ -273,7 +273,7 @@ class RedisSemanticCache(BaseCache):
         return prompt or None
 
     @classmethod
-    def _collect_responses_input_text(cls, value: Any, prompt_parts: list[str]) -> None:
+    def _collect_responses_input_text(cls, value: object, prompt_parts: list[str]) -> None:
         value = cls._coerce_response_input_value(value)
         if value is None:
             return
@@ -334,7 +334,7 @@ class RedisSemanticCache(BaseCache):
             resolve_embedding_max_input_tokens(self.embedding_max_input_tokens, self.embedding_model, router),
         )
 
-    def _get_embedding(self, prompt: str, metadata: dict[str, Any] | None = None) -> list[float]:
+    def _get_embedding(self, prompt: str, metadata: dict[str, object] | None = None) -> list[float]:
         """
         Routes through the proxy Router when the embedding model is a Router
         deployment so per-deployment auth (e.g. Bedrock aws_role_name) applies,
@@ -425,7 +425,7 @@ class RedisSemanticCache(BaseCache):
 
             prompt_embedding: Final = self._get_embedding(prompt, metadata=kwargs.get("metadata"))
 
-            store_kwargs: Final[dict[str, Any]] = {
+            store_kwargs: Final[dict[str, object]] = {
                 "vector": prompt_embedding,
                 "filters": self._get_cache_filters(key),
             }
@@ -504,7 +504,7 @@ class RedisSemanticCache(BaseCache):
             print_verbose(f"Error retrieving from Redis semantic cache: {e}")
             kwargs.setdefault("metadata", {})["semantic-similarity"] = 0.0
 
-    async def _get_async_embedding(self, prompt: str, metadata: dict[str, Any] | None = None) -> list[float]:
+    async def _get_async_embedding(self, prompt: str, metadata: dict[str, object] | None = None) -> list[float]:
         """
         Asynchronously generate an embedding for the given prompt.
 
@@ -571,7 +571,7 @@ class RedisSemanticCache(BaseCache):
             # Generate embedding for the value (response) to cache
             prompt_embedding: Final = await self._get_async_embedding(prompt, metadata=kwargs.get("metadata"))
 
-            store_kwargs: Final[dict[str, Any]] = {
+            store_kwargs: Final[dict[str, object]] = {
                 "vector": prompt_embedding,
                 "filters": self._get_cache_filters(key),
             }
@@ -665,7 +665,7 @@ class RedisSemanticCache(BaseCache):
         aindex: Final = await self.llmcache._get_async_index()
         return await aindex.info()
 
-    async def async_set_cache_pipeline(self, cache_list: list[tuple[str, Any]], **kwargs: object) -> None:
+    async def async_set_cache_pipeline(self, cache_list: list[tuple[str, object]], **kwargs: object) -> None:
         """
         Asynchronously store multiple values in the semantic cache.
 

litellm/compression/scoring/embedding_scorer.py

@@ -5,6 +5,7 @@ Computes cosine similarity between the query embedding and each message embeddin
 """
 
 import math
+from collections.abc import Mapping
 from typing import Any, Final
 
 from litellm.caching.dual_cache import DualCache
@@ -49,7 +50,7 @@ def embedding_score_messages(
     messages: list[dict],
     model: str,
     cache: DualCache | None = None,
-    embedding_model_params: dict[str, Any] | None = None,
+    embedding_model_params: Mapping[str, object] | None = None,
 ) -> list[float]:
     """
     Score each message's semantic similarity to the query using embeddings.

litellm/experimental_mcp_client/client.py

@@ -5,18 +5,28 @@ LiteLLM Proxy uses this MCP Client to connnect to other MCP servers.
 import asyncio
 import base64
 import os
-from collections.abc import Awaitable, Callable, Generator
+from collections.abc import Awaitable, Callable, Generator, Sequence
+from contextlib import AbstractAsyncContextManager
 from datetime import timedelta
 from functools import partial
 from importlib import metadata
-from typing import Any, Final, TypeVar
+from typing import Any, Final, Protocol, TypeAlias, TypeVar
 
 import httpx
 from mcp import ClientSession, McpError, ReadResourceResult, Resource, StdioServerParameters
 from mcp.client.sse import sse_client
 from mcp.client.stdio import stdio_client
 
-streamable_http_client: Any | None = None
+_TransportContext: TypeAlias = AbstractAsyncContextManager[Sequence[Any]]
+
+
+class _StreamableHttpClientFactory(Protocol):
+    """The ``streamable_http_client`` entry point this module calls on the installed MCP SDK."""
+
+    def __call__(self, *, url: str, http_client: httpx.AsyncClient | None) -> _TransportContext: ...
+
+
+streamable_http_client: _StreamableHttpClientFactory | None = None
 try:
     import mcp.client.streamable_http as streamable_http_module
 
@@ -217,10 +227,12 @@ class MCPSigV4Auth(httpx.Auth):
         aws_region_name: str,
     ):
         """Call STS AssumeRole and return temporary credentials."""
+        import time
+
         import boto3
         from botocore.credentials import Credentials
 
-        session_name: Final = aws_session_name or f"litellm-mcp-{int(__import__('time').time())}"
+        session_name: Final = aws_session_name or f"litellm-mcp-{int(time.time())}"
         sts_kwargs: Final[dict] = {"region_name": aws_region_name}
         if aws_access_key_id and aws_secret_access_key:
             sts_kwargs["aws_access_key_id"] = aws_access_key_id
@@ -316,7 +328,7 @@ class MCPClient:
 
     def _create_transport_context(
         self,
-    ) -> tuple[Any, httpx.AsyncClient | None]:
+    ) -> tuple[_TransportContext, httpx.AsyncClient | None]:
         """
         Create the appropriate transport context based on transport type.
         Returns:
@@ -409,7 +421,7 @@ class MCPClient:
 
     async def _execute_session_operation(
         self,
-        transport_ctx: Any,
+        transport_ctx: _TransportContext,
         operation: Callable[[ClientSession], Awaitable[TSessionResult]],
     ) -> TSessionResult:
         """

litellm/files/main.py

@@ -431,7 +431,7 @@ async def afile_delete(
     extra_headers: dict[str, str] | None = None,
     extra_body: dict[str, str] | None = None,
     **kwargs,
-) -> Coroutine[Any, Any, FileObject]:
+) -> Coroutine[object, object, FileObject]:
     """
     Async: Delete file
 
@@ -1003,7 +1003,7 @@ def file_content_streaming(
     logging_obj: LiteLLMLoggingObj | None,
     _is_async: bool,
     client: Any | None,
-) -> FileContentStreamingResult | Coroutine[Any, Any, FileContentStreamingResult]:
+) -> FileContentStreamingResult | Coroutine[object, object, FileContentStreamingResult]:
     if logging_obj is not None:
         logging_obj.model = model or ""
         logging_obj.model_call_details["model"] = model or ""
@@ -1028,8 +1028,8 @@ def file_content_streaming(
             headers=response.headers,
         )
 
-    response: FileContentStreamingResult | Coroutine[Any, Any, FileContentStreamingResult] = FileContentStreamingResult(
-        stream_iterator=iter(()), headers={}
+    response: FileContentStreamingResult | Coroutine[object, object, FileContentStreamingResult] = (
+        FileContentStreamingResult(stream_iterator=iter(()), headers={})
     )
     if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS:
         openai_creds: Final = get_openai_credentials(

litellm/integrations/newrelic/newrelic.py

@@ -47,6 +47,8 @@ import os
 import threading
 import time
 import uuid
+from collections.abc import Mapping, Sequence
+from datetime import datetime
 from typing import Any, Final
 
 import litellm
@@ -408,8 +410,8 @@ class NewRelicLogger(CustomLogger):
     def _get_duration(
         self,
         kwargs: dict,
-        start_time: Any,
-        end_time: Any,
+        start_time: datetime | float | None,
+        end_time: datetime | float | None,
         standard_logging_object: StandardLoggingPayload | None = None,
     ) -> float | None:
         """
@@ -438,7 +440,7 @@ class NewRelicLogger(CustomLogger):
         self,
         kwargs: dict,
         standard_logging_object: StandardLoggingPayload | None = None,
-    ) -> dict[str, Any]:
+    ) -> dict[str, object]:
         """
         Extract request parameters like temperature and max_tokens, preferring
         StandardLoggingPayload.model_parameters.
@@ -450,7 +452,7 @@ class NewRelicLogger(CustomLogger):
         else:
             source_params = kwargs.get("optional_params") or {}
 
-        params: Final = {}
+        params: Final[dict[str, object]] = {}
 
         temperature: Final = source_params.get("temperature")
         if temperature is not None:
@@ -502,7 +504,7 @@ class NewRelicLogger(CustomLogger):
         response_model: str,
         vendor: str,
         standard_logging_object: StandardLoggingPayload | None = None,
-    ) -> list[dict[str, Any]]:
+    ) -> Sequence[Mapping[str, object]]:
         """
         Extract all messages (request + response) with sequence numbers and timestamps.
 
@@ -512,7 +514,7 @@ class NewRelicLogger(CustomLogger):
         Adds timestamps from StandardLoggingPayload (preferred) or kwargs if available
         (converted to epoch milliseconds).
         """
-        messages: Final = []
+        messages: Final[list[dict[str, object]]] = []
         sequence = 0
 
         # Extract timestamps, preferring StandardLoggingPayload
@@ -544,7 +546,7 @@ class NewRelicLogger(CustomLogger):
         else:
             request_messages = kwargs.get("messages") or []
         for msg in request_messages:
-            message_data = {
+            message_data: dict[str, object] = {
                 "role": msg.get("role") or "user",
                 "sequence": sequence,
                 "response.model": response_model,
@@ -599,11 +601,11 @@ class NewRelicLogger(CustomLogger):
         num_messages: int,
         usage: dict[str, int],
         duration: float | None = None,
-        request_params: dict[str, Any] | None = None,
+        request_params: Mapping[str, object] | None = None,
     ):
         """Record LlmChatCompletionSummary event to New Relic."""
         try:
-            event_data: Final = {
+            event_data: Final[dict[str, object]] = {
                 "id": request_id,
                 "request_id": request_id,
                 "request.model": request_model,
@@ -647,7 +649,7 @@ class NewRelicLogger(CustomLogger):
         request_id: str,
         llm_response_id: str,
         trace_id: str | None,
-        messages: list[dict[str, Any]],
+        messages: Sequence[Mapping[str, object]],
     ):
         """Record LlmChatCompletionMessage events to New Relic.
 
@@ -666,7 +668,7 @@ class NewRelicLogger(CustomLogger):
 
             for message in messages:
                 sequence = message["sequence"]
-                event_data = {
+                event_data: dict[str, object] = {
                     "id": f"{llm_response_id}-{sequence}",
                     "request_id": request_id,
                     "completion_id": request_id,

litellm/integrations/opentelemetry.py

@@ -1,7 +1,7 @@
 import os
 import threading
 from collections import OrderedDict
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Iterable, Mapping
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -166,7 +166,7 @@ class OTELMetricAttributeFilter:
     exclude_list: list[str] | None = None
 
 
-def _build_metric_attribute_filter(value: Any) -> OTELMetricAttributeFilter:
+def _build_metric_attribute_filter(value: object) -> OTELMetricAttributeFilter:
     if isinstance(value, OTELMetricAttributeFilter):
         return value
     if not isinstance(value, dict):
@@ -205,7 +205,7 @@ def _resolve_metric_attribute_filter(
     )
 
 
-def _normalize_team_metadata_keys(value: Any) -> list[str]:
+def _normalize_team_metadata_keys(value: str | Iterable[object] | None) -> list[str]:
     """Coerce a team-metadata allowlist from a list or comma-separated string.
 
     config.yaml passes a YAML list; an env var passes a comma-separated string.
@@ -1569,7 +1569,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
             self.safe_set_attribute(span=span, key=RESPONSE_SERVICE_TIER_ATTRIBUTE, value=served_tier)
 
     @staticmethod
-    def _team_metadata_json(value: Any, allowed_keys: list[str]) -> str | None:
+    def _team_metadata_json(value: object, allowed_keys: list[str]) -> str | None:
         """JSON-serialize only the allowlisted sub-keys of a team's metadata.
 
         Returns ``None`` when nothing is allowlisted or no allowlisted key is
@@ -3524,7 +3524,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
             kwargs={"standard_logging_object": {"error_information": error_information}},
         )
 
-    def set_preprocessing_duration_attribute(self, span: Span | None, container: Any) -> None:
+    def set_preprocessing_duration_attribute(self, span: Span | None, container: object) -> None:
         """
         Set ``litellm.preprocessing.duration_ms`` (proxy-receive -> first
         provider handoff) on the proxy SERVER span. ``litellm_received_at``

litellm/integrations/prometheus.py

@@ -2607,7 +2607,7 @@ class PrometheusLogger(CustomLogger):
         for all successful requests (both streaming and non-streaming).
         """
 
-    def _safe_get(self, obj: Any, key: str, default: object = None) -> Any:
+    def _safe_get(self, obj: object, key: str, default: object = None) -> Any:
         """Get value from dict or Pydantic model."""
         if obj is None:
             return default
@@ -4215,8 +4215,8 @@ class PrometheusLogger(CustomLogger):
 
     def _safe_duration_seconds(
         self,
-        start_time: Any,
-        end_time: Any,
+        start_time: object,
+        end_time: object,
     ) -> float | None:
         """
         Compute the duration in seconds between two objects.

litellm/integrations/websearch_interception/tools.py

@@ -6,12 +6,13 @@ Native provider tools (like Anthropic's web_search_20250305) are converted
 to this format for consistent interception and execution.
 """
 
+from collections.abc import Mapping
 from typing import Any, Final
 
 from litellm.constants import LITELLM_WEB_SEARCH_TOOL_NAME
 
 
-def get_litellm_web_search_tool() -> dict[str, Any]:
+def get_litellm_web_search_tool() -> dict[str, object]:
     """
     Get the standard LiteLLM web search tool definition.
 
@@ -49,7 +50,7 @@ def get_litellm_web_search_tool() -> dict[str, Any]:
     }
 
 
-def get_litellm_web_search_tool_openai() -> dict[str, Any]:
+def get_litellm_web_search_tool_openai() -> dict[str, object]:
     """
     Get the standard LiteLLM web search tool definition in OpenAI format.
 
@@ -82,7 +83,7 @@ def get_litellm_web_search_tool_openai() -> dict[str, Any]:
     }
 
 
-def get_litellm_web_search_tool_responses() -> dict[str, Any]:
+def get_litellm_web_search_tool_responses() -> dict[str, object]:
     """
     Get the standard LiteLLM web search tool definition in Responses API format.
 
@@ -114,7 +115,7 @@ def get_litellm_web_search_tool_responses() -> dict[str, Any]:
     }
 
 
-def is_web_search_tool_responses(tool: dict[str, Any]) -> bool:
+def is_web_search_tool_responses(tool: Mapping[str, object]) -> bool:
     """
     Check if a tool is a web search tool for the Responses API.
 
@@ -195,7 +196,7 @@ def is_web_search_tool_chat_completion(tool: dict[str, Any]) -> bool:
     return False
 
 
-def is_anthropic_native_web_search_tool(tool: dict[str, Any]) -> bool:
+def is_anthropic_native_web_search_tool(tool: Mapping[str, object]) -> bool:
     """
     Check if a tool is an Anthropic-native ``web_search_*`` tool.
 

litellm/integrations/websearch_interception/transformation.py

@@ -24,7 +24,7 @@ class WebSearchTransformation:
 
     @staticmethod
     def transform_request(
-        response: Any,
+        response: object,
         stream: bool,
         response_format: str = "anthropic",
     ) -> tuple[bool, list[dict]]:
@@ -66,7 +66,7 @@ class WebSearchTransformation:
 
     @staticmethod
     def _detect_from_responses_response(
-        response: Any,
+        response: object,
     ) -> tuple[bool, list[dict]]:
         """Parse a Responses API response for ``litellm_web_search`` function calls.
 
@@ -399,7 +399,7 @@ class WebSearchTransformation:
     def build_web_search_tool_result_block(
         tool_use_id: str,
         search_response: SearchResponse | None,
-    ) -> dict[str, Any]:
+    ) -> dict[str, object]:
         """
         Build an Anthropic-native ``web_search_tool_result`` content block.
 
@@ -433,7 +433,7 @@ class WebSearchTransformation:
                 emitted with an empty result list (signals "search ran, no
                 results" rather than "search did not run").
         """
-        items: Final[list[dict[str, Any]]] = []
+        items: Final[list[dict[str, object]]] = []
         if search_response is not None:
             results: Final = getattr(search_response, "results", None) or []
             for r in results:

litellm/interactions/agents/http_handler.py

@@ -6,7 +6,7 @@ Extends InteractionsHTTPHandler so that the shared HTTP infrastructure
 duplicated. BaseAgentsAPIConfig stays as pure transform code.
 """
 
-from collections.abc import Coroutine
+from collections.abc import Coroutine, Mapping
 from typing import Any, Final
 
 import httpx
@@ -39,11 +39,11 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
         litellm_params: GenericLiteLLMParams,
         logging_obj: LiteLLMLoggingObj,
         extra_headers: dict[str, Any] | None = None,
-        extra_body: dict[str, Any] | None = None,
+        extra_body: Mapping[str, object] | None = None,
         timeout: float | httpx.Timeout | None = None,
         client: HTTPHandler | None = None,
         _is_async: bool = False,
-    ) -> AgentCreateResponse | Coroutine[Any, Any, AgentCreateResponse]:
+    ) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
         if _is_async:
             return self.async_create_agent(
                 agents_api_config=agents_api_config,
@@ -94,7 +94,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
         litellm_params: GenericLiteLLMParams,
         logging_obj: LiteLLMLoggingObj,
         extra_headers: dict[str, Any] | None = None,
-        extra_body: dict[str, Any] | None = None,
+        extra_body: Mapping[str, object] | None = None,
         timeout: float | httpx.Timeout | None = None,
         client: AsyncHTTPHandler | None = None,
     ) -> AgentCreateResponse:
@@ -145,7 +145,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
         timeout: float | httpx.Timeout | None = None,
         client: HTTPHandler | None = None,
         _is_async: bool = False,
-    ) -> AgentListResponse | Coroutine[Any, Any, AgentListResponse]:
+    ) -> AgentListResponse | Coroutine[object, object, AgentListResponse]:
         if _is_async:
             return self.async_list_agents(
                 agents_api_config=agents_api_config,
@@ -220,7 +220,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
         timeout: float | httpx.Timeout | None = None,
         client: HTTPHandler | None = None,
         _is_async: bool = False,
-    ) -> AgentCreateResponse | Coroutine[Any, Any, AgentCreateResponse]:
+    ) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
         if _is_async:
             return self.async_get_agent(
                 agents_api_config=agents_api_config,
@@ -299,7 +299,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
         timeout: float | httpx.Timeout | None = None,
         client: HTTPHandler | None = None,
         _is_async: bool = False,
-    ) -> AgentDeleteResult | Coroutine[Any, Any, AgentDeleteResult]:
+    ) -> AgentDeleteResult | Coroutine[object, object, AgentDeleteResult]:
         if _is_async:
             return self.async_delete_agent(
                 agents_api_config=agents_api_config,
@@ -378,7 +378,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
         timeout: float | httpx.Timeout | None = None,
         client: HTTPHandler | None = None,
         _is_async: bool = False,
-    ) -> AgentVersionsResponse | Coroutine[Any, Any, AgentVersionsResponse]:
+    ) -> AgentVersionsResponse | Coroutine[object, object, AgentVersionsResponse]:
         if _is_async:
             return self.async_list_agent_versions(
                 agents_api_config=agents_api_config,

litellm/interactions/agents/main.py

@@ -30,7 +30,7 @@ Usage:
 
 import asyncio
 import contextvars
-from collections.abc import Coroutine
+from collections.abc import Coroutine, Mapping
 from functools import partial
 from typing import Any, Final
 
@@ -75,7 +75,7 @@ def _make_logging_obj(
     model: str,
     custom_llm_provider: str,
     call_type: str,
-    optional_params: dict[str, Any],
+    optional_params: dict[str, object],
 ) -> LiteLLMLoggingObj:
     litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
     litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
@@ -102,7 +102,7 @@ async def acreate(
     base_environment: InteractionEnvironment | None = None,
     custom_llm_provider: str | None = None,
     extra_headers: dict[str, Any] | None = None,
-    extra_body: dict[str, Any] | None = None,
+    extra_body: Mapping[str, object] | None = None,
     timeout: float | httpx.Timeout | None = None,
     **kwargs,
 ) -> AgentCreateResponse:
@@ -146,10 +146,10 @@ def create(
     base_environment: InteractionEnvironment | None = None,
     custom_llm_provider: str | None = None,
     extra_headers: dict[str, Any] | None = None,
-    extra_body: dict[str, Any] | None = None,
+    extra_body: Mapping[str, object] | None = None,
     timeout: float | httpx.Timeout | None = None,
     **kwargs,
-) -> AgentCreateResponse | Coroutine[Any, Any, AgentCreateResponse]:
+) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
     """
     Sync: Create a managed agent on the provider side.
 
@@ -244,7 +244,7 @@ def list(
     extra_headers: dict[str, Any] | None = None,
     timeout: float | httpx.Timeout | None = None,
     **kwargs,
-) -> AgentListResponse | Coroutine[Any, Any, AgentListResponse]:
+) -> AgentListResponse | Coroutine[object, object, AgentListResponse]:
     """Sync: List all agents on the provider side."""
     local_vars: Final = locals()
     custom_llm_provider = custom_llm_provider or kwargs.get("custom_llm_provider") or "gemini"
@@ -320,7 +320,7 @@ def get(
     extra_headers: dict[str, Any] | None = None,
     timeout: float | httpx.Timeout | None = None,
     **kwargs,
-) -> AgentCreateResponse | Coroutine[Any, Any, AgentCreateResponse]:
+) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
     """Sync: Get a specific agent by name."""
     local_vars: Final = locals()
     custom_llm_provider = custom_llm_provider or kwargs.get("custom_llm_provider") or "gemini"
@@ -397,7 +397,7 @@ def delete(
     extra_headers: dict[str, Any] | None = None,
     timeout: float | httpx.Timeout | None = None,
     **kwargs,
-) -> AgentDeleteResult | Coroutine[Any, Any, AgentDeleteResult]:
+) -> AgentDeleteResult | Coroutine[object, object, AgentDeleteResult]:
     """Sync: Delete a specific agent by name."""
     local_vars: Final = locals()
     custom_llm_provider = custom_llm_provider or kwargs.get("custom_llm_provider") or "gemini"
@@ -474,7 +474,7 @@ def list_versions(
     extra_headers: dict[str, Any] | None = None,
     timeout: float | httpx.Timeout | None = None,
     **kwargs,
-) -> AgentVersionsResponse | Coroutine[Any, Any, AgentVersionsResponse]:
+) -> AgentVersionsResponse | Coroutine[object, object, AgentVersionsResponse]:
     """Sync: List versions of a specific agent."""
     local_vars: Final = locals()
     custom_llm_provider = custom_llm_provider or kwargs.get("custom_llm_provider") or "gemini"

litellm/interactions/litellm_responses_transformation/transformation.py

@@ -34,8 +34,8 @@ class LiteLLMResponsesInteractionsConfig:
         model: str,
         input: InteractionInput | None,
         optional_params: InteractionsAPIOptionalRequestParams,
-        **kwargs,
-    ) -> dict[str, Any]:
+        **kwargs: object,
+    ) -> dict[str, object]:
         """
         Transform an Interactions API request to a Responses API request.
 
@@ -45,7 +45,7 @@ class LiteLLMResponsesInteractionsConfig:
         - tools -> tools (similar format)
         - generation_config -> temperature, top_p, etc.
         """
-        responses_request: Final[dict[str, Any]] = {
+        responses_request: Final[dict[str, object]] = {
             "model": model,
         }
 
@@ -201,15 +201,15 @@ class LiteLLMResponsesInteractionsConfig:
         - Extract usage
         """
         # Extract text from outputs and build both `outputs` (legacy) and `steps` (new schema).
-        outputs: Final[list[dict[str, Any]]] = []
-        steps: Final[list[dict[str, Any]]] = []
+        outputs: Final[list[dict[str, object]]] = []
+        steps: Final[list[dict[str, object]]] = []
         if hasattr(responses_response, "output") and responses_response.output:
             for output_item in responses_response.output:
                 # Use getattr with None default to safely access content
                 content = getattr(output_item, "content", None)
                 if content is not None:
                     content_items = content if isinstance(content, list) else [content]
-                    model_output_contents: list[dict[str, Any]] = []
+                    model_output_contents: list[dict[str, object]] = []
                     for content_item in content_items:
                         # Check if content_item has text attribute
                         text = getattr(content_item, "text", None)
@@ -264,7 +264,7 @@ class LiteLLMResponsesInteractionsConfig:
         # Add usage if available
         # Map Responses API usage (input_tokens, output_tokens) to Interactions API spec format
         # (total_input_tokens, total_output_tokens)
-        usage: Final = getattr(responses_response, "usage", None)
+        usage: Final[object] = getattr(responses_response, "usage", None)
         if usage:
             interactions_response_dict["usage"] = {
                 "total_input_tokens": getattr(usage, "input_tokens", 0),

litellm/interactions/main.py

@@ -229,7 +229,7 @@ def create(
 ) -> (
     InteractionsAPIResponse
     | Iterator[InteractionsAPIStreamingResponse]
-    | Coroutine[Any, Any, InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]]
+    | Coroutine[object, object, InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]]
 ):
     """
     Sync: Create a new interaction using Google's Interactions API.
@@ -406,7 +406,7 @@ def get(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> InteractionsAPIResponse | Coroutine[Any, Any, InteractionsAPIResponse]:
+) -> InteractionsAPIResponse | Coroutine[object, object, InteractionsAPIResponse]:
     """Sync: Get an interaction by its ID."""
     local_vars: Final = locals()
     custom_llm_provider = custom_llm_provider or "gemini"
@@ -510,7 +510,7 @@ def delete(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> DeleteInteractionResult | Coroutine[Any, Any, DeleteInteractionResult]:
+) -> DeleteInteractionResult | Coroutine[object, object, DeleteInteractionResult]:
     """Sync: Delete an interaction by its ID."""
     local_vars: Final = locals()
     custom_llm_provider = custom_llm_provider or "gemini"
@@ -612,7 +612,7 @@ def cancel(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> CancelInteractionResult | Coroutine[Any, Any, CancelInteractionResult]:
+) -> CancelInteractionResult | Coroutine[object, object, CancelInteractionResult]:
     """Sync: Cancel an interaction by its ID."""
     local_vars: Final = locals()
     custom_llm_provider = custom_llm_provider or "gemini"

litellm/litellm_core_utils/core_helpers.py

@@ -419,7 +419,7 @@ def safe_deep_copy(data):
     if litellm.safe_memory_mode is True:
         return data
 
-    litellm_parent_otel_span: Any | None = None
+    litellm_parent_otel_span: object | None = None
     # Step 1: Remove the litellm_parent_otel_span
     litellm_parent_otel_span = None
     if isinstance(data, dict):
@@ -510,7 +510,7 @@ def independent_snapshot(
     }
 
 
-def filter_exceptions_from_params(data: Any, max_depth: int = 20) -> Any:
+def filter_exceptions_from_params(data: object, max_depth: int = 20) -> Any:
     """
     Recursively filter out Exception objects and callable objects from dicts/lists.
 
@@ -542,7 +542,7 @@ def filter_exceptions_from_params(data: Any, max_depth: int = 20) -> Any:
         return None
 
     if isinstance(data, dict):
-        result: Final[dict[str, Any]] = {}
+        result: Final[dict[str, object]] = {}
         for k, v in data.items():
             # Skip exception and callable values
             if isinstance(v, Exception) or (callable(v) and not isinstance(v, type)):
@@ -556,7 +556,7 @@ def filter_exceptions_from_params(data: Any, max_depth: int = 20) -> Any:
                 continue
         return result
     elif isinstance(data, list):
-        result_list: Final[list[Any]] = []
+        result_list: Final[list[object]] = []
         for item in data:
             # Skip exception and callable items
             if isinstance(item, Exception) or (callable(item) and not isinstance(item, type)):
@@ -624,7 +624,7 @@ def redact_nested_match_and_regex_keys(
     # Iterative traversal; `seen` guards against cyclic refs preserved by deepcopy.
     try:
         seen: Final[set] = set()
-        stack: Final[list[Any]] = [redacted]
+        stack: Final[list[object]] = [redacted]
         while stack:
             node = stack.pop()
             node_id = id(node)

litellm/litellm_core_utils/llm_response_utils/response_metadata.py

@@ -168,7 +168,7 @@ class ResponseMetadata:
 
 
 def update_response_metadata(
-    result: Any,
+    result: object,
     logging_obj: LiteLLMLoggingObject,
     model: str | None,
     kwargs: dict,

litellm/litellm_core_utils/prompt_templates/factory.py

@@ -1708,8 +1708,8 @@ def _find_server_tool_result(
 
 def convert_to_anthropic_tool_invoke(
     tool_calls: list[ChatCompletionAssistantToolCall],
-    web_search_results: list[Any] | None = None,
-    tool_results: list[Any] | None = None,
+    web_search_results: Sequence[object] | None = None,
+    tool_results: Sequence[object] | None = None,
 ) -> list[AnthropicMessagesToolUseParam | dict[str, Any]]:
     """
     OpenAI tool invokes:
@@ -5349,7 +5349,7 @@ class NormalizedToolCall(TypedDict):
     arguments: dict[str, object]
 
 
-def _parse_tool_call_arguments(raw: Any, tool_name: str | None, context: str) -> dict[str, object]:
+def _parse_tool_call_arguments(raw: object, tool_name: str | None, context: str) -> dict[str, object]:
     # Anthropic's tool_use blocks already carry a parsed dict in "input";
     # chat completions and the Responses API carry a JSON string that may be
     # truncated by the model, so route those through the repair-aware parser.

litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py

@@ -4,7 +4,7 @@ import copy
 import json
 import traceback
 from collections import deque
-from collections.abc import AsyncIterator, Iterator, Sequence
+from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -418,7 +418,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
         augmented["usage"] = augmented_usage
         return augmented
 
-    def _next_compaction_event(self) -> dict[str, Any] | None:
+    def _next_compaction_event(self) -> dict[str, object] | None:
         """Return the next compaction content-block SSE event, or ``None``.
 
         Anthropic delivers compaction as a single delta (no token-by-token
@@ -457,7 +457,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
                 "delta": {"type": "compaction_delta", "content": summary_content},
             }
 
-        stop_event: Final = {
+        stop_event: Final[dict[str, object]] = {
             "type": "content_block_stop",
             "index": compaction_index,
         }
@@ -989,7 +989,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
         self.current_content_block_index += 1
 
     @staticmethod
-    def _delta_has_content(processed_chunk: dict[str, Any]) -> bool:
+    def _delta_has_content(processed_chunk: Mapping[str, object]) -> bool:
         """Return True if a translated chunk carries a non-empty
         ``content_block_delta`` payload.
 

litellm/llms/anthropic/experimental_pass_through/messages/transformation.py

@@ -87,7 +87,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
         Processes both `system` and `messages` content blocks.
         """
 
-        def _sanitize(cache_control: Any) -> None:
+        def _sanitize(cache_control: object) -> None:
             if isinstance(cache_control, dict):
                 cache_control.pop("scope", None)
 
@@ -152,7 +152,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
             return system_param
 
     @staticmethod
-    def _as_system_content_blocks(value: Any) -> list:
+    def _as_system_content_blocks(value: object) -> list:
         if value is None:
             return []
         if isinstance(value, list):
@@ -162,7 +162,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
         return [value]
 
     @staticmethod
-    def _is_system_role_message(message: Any) -> bool:
+    def _is_system_role_message(message: object) -> bool:
         return isinstance(message, dict) and message.get("role") == "system"
 
     _CONVERTED_SYSTEM_NOTE: Final = (

litellm/llms/azure_ai/vector_stores/transformation.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, Final
 
 import httpx
@@ -114,8 +115,8 @@ class AzureAIVectorStoreConfig(BaseVectorStoreConfig, BaseAzureLLM):
         api_base: str,
         litellm_logging_obj: LiteLLMLoggingObj,
         litellm_params: dict,
-        extra_body: dict[str, Any] | None = None,
-    ) -> tuple[str, dict[str, Any]]:
+        extra_body: Mapping[str, object] | None = None,
+    ) -> tuple[str, dict[str, object]]:
         """
         Transform search request for Azure AI Search API
 
@@ -162,7 +163,7 @@ class AzureAIVectorStoreConfig(BaseVectorStoreConfig, BaseAzureLLM):
         url: Final = f"{api_base}/indexes/{index_name}/docs/search?api-version=2024-07-01"
 
         # Build the request body for Azure AI Search with vector search
-        request_body: Final = {
+        request_body: Final[dict[str, object]] = {
             "search": "*",  # Get all documents (filtered by vector similarity)
             "vectorQueries": [
                 {

litellm/llms/base_llm/guardrail_translation/base_translation.py

@@ -39,7 +39,7 @@ class BaseTranslation(ABC):
     @staticmethod
     def transform_user_api_key_dict_to_metadata(
         user_api_key_dict: Any | None,
-    ) -> dict[str, Any]:
+    ) -> dict[str, object]:
         """
         Transform user_api_key_dict to a metadata dict with prefixed keys.
 
@@ -62,7 +62,7 @@ class BaseTranslation(ABC):
             return {}
 
         # Transform keys to be prefixed with 'user_api_key_'
-        transformed: Final = {}
+        transformed: Final[dict[str, object]] = {}
         for key, value in user_dict.items():
             # Skip None values and internal fields
             if value is None or key.startswith("_"):
@@ -155,7 +155,7 @@ class BaseTranslation(ABC):
         self,
         exc: "ModifyResponseException",
         stream_started: bool = False,
-        responses_so_far: Sequence[Any] | None = None,
+        responses_so_far: Sequence[object] | None = None,
     ) -> Sequence[bytes] | None:
         """
         Build the streaming chunks that deliver a guardrail block message and
@@ -178,8 +178,8 @@ class BaseTranslation(ABC):
     def build_stream_error_items(
         self,
         exc: "HTTPException",
-        responses_so_far: Sequence[Any] | None = None,
-    ) -> Sequence[Any] | None:
+        responses_so_far: Sequence[object] | None = None,
+    ) -> Sequence[object] | None:
         """
         Build the stream items that surface a guardrail HTTPException (a block
         with the default exception-on-block config, or a failed scan) after the

litellm/llms/bedrock/common_utils.py

@@ -52,8 +52,8 @@ _BEDROCK_AWS_AUTH_PARAMETER_KEYS: Final[tuple[str, ...]] = (
 
 
 def merge_bedrock_aws_request_params(
-    litellm_params: Mapping[str, Any],
-    optional_params: Mapping[str, Any],
+    litellm_params: Mapping[str, object],
+    optional_params: Mapping[str, object],
 ) -> dict[str, Any]:
     """Merge deployment and request parameters without allowing auth escalation.
 
@@ -303,7 +303,7 @@ def normalize_json_schema_custom_types_to_object(schema: dict) -> None:
 
     Uses an explicit stack (not recursion) to satisfy recursive-function guards in CI.
     """
-    stack: Final[list[Any]] = [schema]
+    stack: Final[list[object]] = [schema]
     seen: Final[set[int]] = set()
     while stack:
         node = stack.pop()
@@ -901,7 +901,7 @@ def _get_bedrock_converse_strict_tools_flag(base_model: str) -> bool | None:
     return None
 
 
-def normalize_bedrock_opus_output_config_effort(model: str, output_config: Any) -> None:
+def normalize_bedrock_opus_output_config_effort(model: str, output_config: object) -> None:
     """
     Normalize Anthropic ``output_config.effort`` values for Bedrock Opus ids.
 
@@ -1424,6 +1424,11 @@ class BedrockEventStreamDecoderBase:
             return chunk.decode()
 
 
+def _decoded_json_value(raw: str) -> object:
+    """Decode a JSON document into an opaque value for isinstance narrowing."""
+    return json.loads(raw)
+
+
 def get_anthropic_beta_from_headers(headers: dict) -> list[str]:
     """
     Extract anthropic-beta header values and convert them to a list.
@@ -1451,7 +1456,7 @@ def get_anthropic_beta_from_headers(headers: dict) -> list[str]:
         anthropic_beta_header = anthropic_beta_header.strip()
         if anthropic_beta_header.startswith("[") and anthropic_beta_header.endswith("]"):
             try:
-                parsed: Final = json.loads(anthropic_beta_header)
+                parsed: Final = _decoded_json_value(anthropic_beta_header)
                 if isinstance(parsed, list):
                     return [str(beta).strip() for beta in parsed]
             except json.JSONDecodeError:
@@ -1464,8 +1469,8 @@ def get_anthropic_beta_from_headers(headers: dict) -> list[str]:
 
 
 def resolve_s3_encryption_key_id(
-    litellm_params: Mapping[str, Any],
-    optional_params: Mapping[str, Any] | None = None,
+    litellm_params: Mapping[str, object],
+    optional_params: Mapping[str, object] | None = None,
 ) -> str | None:
     """
     Resolve the SSE-KMS key configured for Bedrock batch/file S3 objects.

litellm/llms/bedrock/image_edit/amazon_nova_canvas_image_edit_transformation.py

@@ -47,7 +47,7 @@ def _nova_canvas_task_body(
     task_type: str | None,
     mask_prompt: str | None,
     out_painting_mode: str | None,
-) -> dict[str, Any]:
+) -> dict[str, object]:
     """Build InvokeModel body task section (without imageGenerationConfig)."""
     if task_type == "BACKGROUND_REMOVAL":
         return {
@@ -60,7 +60,7 @@ def _nova_canvas_task_body(
                 "OUTPAINTING requires either a mask image or a mask prompt. "
                 "Pass mask=<file> or maskPrompt=<str> in the request."
             )
-        out_params: Final[dict[str, Any]] = {
+        out_params: Final[dict[str, object]] = {
             "image": image_b64,
             "text": text,
         }
@@ -79,7 +79,7 @@ def _nova_canvas_task_body(
     # Honour explicit IMAGE_VARIATION even when a mask is present (mask is ignored
     # for this task type; callers use INPAINTING when they want mask semantics).
     if task_type == "IMAGE_VARIATION":
-        var_params_explicit: Final[dict[str, Any]] = {
+        var_params_explicit: Final[dict[str, object]] = {
             "images": [image_b64],
             "text": text,
         }
@@ -100,7 +100,7 @@ def _nova_canvas_task_body(
                 "or omit taskType for automatic routing (mask → INPAINTING, else IMAGE_VARIATION)."
             )
     if mask_b64 is not None or mask_prompt is not None or task_type == "INPAINTING":
-        in_params: Final[dict[str, Any]] = {"image": image_b64, "text": text}
+        in_params: Final[dict[str, object]] = {"image": image_b64, "text": text}
         if mask_prompt is not None:
             in_params["maskPrompt"] = mask_prompt
         elif mask_b64 is not None:
@@ -114,7 +114,7 @@ def _nova_canvas_task_body(
                 "See https://docs.aws.amazon.com/nova/latest/userguide/image-gen-req-resp-structure.html"
             )
         return {"taskType": "INPAINTING", "inPaintingParams": in_params}
-    var_params: Final[dict[str, Any]] = {
+    var_params: Final[dict[str, object]] = {
         "images": [image_b64],
         "text": text,
     }
@@ -250,9 +250,9 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
         image_edit_optional_params: ImageEditOptionalRequestParams,
         model: str,
         drop_params: bool,
-    ) -> dict[str, Any]:
+    ) -> dict[str, object]:
         supported: Final = set(self.get_supported_openai_params(model))
-        mapped: Final[dict[str, Any]] = dict(image_edit_optional_params)
+        mapped: Final[dict[str, object]] = dict(image_edit_optional_params)
         _size: Final = mapped.pop("size", None)
         if _size is not None and isinstance(_size, str) and "x" in _size:
             w, h = _size.split("x", 1)
@@ -327,7 +327,7 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
         cfg_scale: Final = op.pop("cfgScale", None)
         seed: Final = op.pop("seed", None)
 
-        image_generation_config: Final[dict[str, Any]] = {}
+        image_generation_config: Final[dict[str, object]] = {}
         nested_igc: Final = op.pop("imageGenerationConfig", None)
         if isinstance(nested_igc, dict):
             image_generation_config.update(nested_igc)

litellm/llms/bedrock/vector_stores/transformation.py

@@ -203,7 +203,7 @@ class BedrockVectorStoreConfig(BaseVectorStoreConfig, BaseAWSLLM):
         encoded_vector_store_id: Final = encode_url_path_segment(vector_store_id, field_name="vector_store_id")
         url: Final = f"{api_base}/{encoded_vector_store_id}/retrieve"
 
-        request_body: Final[dict[str, Any]] = {
+        request_body: Final[dict[str, object]] = {
             "retrievalQuery": BedrockKBRetrievalQuery(text=query),
         }
 
@@ -288,7 +288,7 @@ class BedrockVectorStoreConfig(BaseVectorStoreConfig, BaseAWSLLM):
         data_source_id: Final = metadata.get("x-amz-bedrock-kb-data-source-id", "unknown") if metadata else "unknown"
         return f"bedrock-kb-document-{data_source_id}"
 
-    def _get_attributes_from_metadata(self, metadata: dict[str, Any]) -> dict[str, Any]:
+    def _get_attributes_from_metadata(self, metadata: dict[str, object]) -> dict[str, object]:
         """
         Extract all attributes from Bedrock KB metadata.
         Returns a copy of the metadata dictionary.

litellm/llms/black_forest_labs/image_edit/transformation.py

@@ -84,7 +84,7 @@ class BlackForestLabsImageEditConfig(BaseImageEditConfig):
 
         BFL-specific params are passed through directly.
         """
-        optional_params: Final[dict[str, Any]] = {}
+        optional_params: Final[dict[str, object]] = {}
 
         # Pass through BFL-specific params
         bfl_params: Final = [
@@ -246,7 +246,7 @@ class BlackForestLabsImageEditConfig(BaseImageEditConfig):
         b64_image: Final = base64.b64encode(image_bytes).decode("utf-8")
 
         # Build request body
-        request_body: Final[dict[str, Any]] = {
+        request_body: Final[dict[str, object]] = {
             "prompt": prompt,
             "input_image": b64_image,
         }

litellm/llms/gemini/agents/transformation.py

@@ -9,6 +9,7 @@ Proxies the Gemini v1beta Agents API:
   GET    /v1beta/agents/{name}/versions         list versions
 """
 
+from collections.abc import Mapping
 from typing import Any, Final
 
 import httpx
@@ -87,7 +88,7 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
     def get_complete_url(
         self,
         api_base: str | None,
-        litellm_params: dict[str, Any],
+        litellm_params: Mapping[str, object],
     ) -> str:
         return f"{self._base_url(api_base)}/agents"
 
@@ -132,9 +133,9 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
     def transform_create_request(
         self,
         name: str,
-        litellm_params: dict[str, Any],
-    ) -> dict[str, Any]:
-        body: Final[dict[str, Any]] = {"name": name}
+        litellm_params: Mapping[str, object],
+    ) -> dict[str, object]:
+        body: Final[dict[str, object]] = {"name": name}
         for key in _GEMINI_AGENT_BODY_KEYS:
             value = litellm_params.get(key)
             if value is not None:
@@ -174,10 +175,10 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
     def transform_list_request(
         self,
         api_base: str | None,
-        litellm_params: dict[str, Any],
-    ) -> tuple[str, dict[str, Any]]:
+        litellm_params: Mapping[str, object],
+    ) -> tuple[str, dict[str, object]]:
         url: Final = f"{self._base_url(api_base)}/agents"
-        params: Final[dict[str, Any]] = {}
+        params: Final[dict[str, object]] = {}
         if litellm_params.get("page_size"):
             params["pageSize"] = litellm_params["page_size"]
         if litellm_params.get("page_token"):
@@ -207,8 +208,8 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
         self,
         name: str,
         api_base: str | None,
-        litellm_params: dict[str, Any],
-    ) -> tuple[str, dict[str, Any]]:
+        litellm_params: Mapping[str, object],
+    ) -> tuple[str, dict[str, object]]:
         url: Final = f"{self._base_url(api_base)}/agents/{name}"
         return url, {}
 
@@ -236,7 +237,7 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
         self,
         name: str,
         api_base: str | None,
-        litellm_params: dict[str, Any],
+        litellm_params: Mapping[str, object],
     ) -> str:
         return f"{self._base_url(api_base)}/agents/{name}"
 
@@ -262,10 +263,10 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
         self,
         name: str,
         api_base: str | None,
-        litellm_params: dict[str, Any],
-    ) -> tuple[str, dict[str, Any]]:
+        litellm_params: Mapping[str, object],
+    ) -> tuple[str, dict[str, object]]:
         url: Final = f"{self._base_url(api_base)}/agents/{name}/versions"
-        params: Final[dict[str, Any]] = {}
+        params: Final[dict[str, object]] = {}
         if litellm_params.get("page_size"):
             params["pageSize"] = litellm_params["page_size"]
         if litellm_params.get("page_token"):

litellm/llms/milvus/vector_stores/transformation.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, Final
 
 import httpx
@@ -122,8 +123,8 @@ class MilvusVectorStoreConfig(BaseVectorStoreConfig):
         api_base: str,
         litellm_logging_obj: LiteLLMLoggingObj,
         litellm_params: dict,
-        extra_body: dict[str, Any] | None = None,
-    ) -> tuple[str, dict[str, Any]]:
+        extra_body: Mapping[str, object] | None = None,
+    ) -> tuple[str, dict[str, object]]:
         """
         Transform search request for Azure AI Search API
 
@@ -165,7 +166,7 @@ class MilvusVectorStoreConfig(BaseVectorStoreConfig):
         url: Final = f"{api_base}/v2/vectordb/entities/search"
 
         # Build the request body for Azure AI Search with vector search
-        request_body: Final[dict[str, Any]] = {
+        request_body: Final[dict[str, object]] = {
             "collectionName": index_name,
             "data": [query_vector],
             "annsField": "book_intro_vector",

litellm/llms/minimax/text_to_speech/transformation.py

@@ -5,6 +5,7 @@ Maps OpenAI TTS spec to MiniMax TTS API (WebSocket-based HTTP API)
 Reference: https://platform.minimax.io/docs
 """
 
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, Final
 
 import httpx
@@ -86,8 +87,8 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
 
     def _resolve_voice_id(
         self,
-        voice: str | dict[str, Any] | None,
-        params: dict[str, Any],
+        voice: str | Mapping[str, object] | None,
+        params: dict[str, object],
     ) -> str:
         """
         Determine the MiniMax voice_id based on provided voice input or parameters.
@@ -127,7 +128,7 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
         """
         Map OpenAI parameters to MiniMax TTS parameters
         """
-        mapped_params: Final[dict[str, Any]] = {}
+        mapped_params: Final[dict[str, object]] = {}
 
         # Work on a copy so we don't mutate the caller's dictionary
         params: Final = dict(optional_params) if optional_params else {}
@@ -242,7 +243,7 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
         # Output format: 'url' or 'hex' (default is 'hex')
         output_format: Final = params.pop("output_format", "hex")
 
-        request_body: Final[dict[str, Any]] = {
+        request_body: Final[dict[str, object]] = {
             "model": model,
             "text": input,
             "stream": False,  # HTTP endpoint doesn't support streaming

litellm/llms/openai/responses/count_tokens/transformation.py

@@ -117,16 +117,16 @@ class OpenAICountTokensConfig:
     def transform_request_to_count_tokens(
         self,
         model: str,
-        input: str | list[Any],
+        input: str | Sequence[object],
         tools: list[dict[str, Any]] | None = None,
         instructions: str | None = None,
-    ) -> dict[str, Any]:
+    ) -> dict[str, object]:
         """
         Transform request to OpenAI Responses API token counting format.
 
         The Responses API uses `input` (not `messages`) and `instructions` (not `system`).
         """
-        request: Final[dict[str, Any]] = {
+        request: Final[dict[str, object]] = {
             "model": model,
             "input": input,
         }
@@ -145,7 +145,7 @@ class OpenAICountTokensConfig:
             "Authorization": f"Bearer {api_key}",
         }
 
-    def validate_request(self, model: str, input: str | list[Any]) -> None:
+    def validate_request(self, model: str, input: str | Sequence[object]) -> None:
         if not model:
             raise ValueError("model parameter is required")
 
@@ -155,18 +155,18 @@ class OpenAICountTokensConfig:
     @staticmethod
     def _transform_tools_for_responses_api(
         tools: list[dict[str, Any]],
-    ) -> list[dict[str, Any]]:
+    ) -> list[dict[str, object]]:
         """
         Transform OpenAI chat tools format to Responses API tools format.
 
         Chat format:  {"type": "function", "function": {"name": "...", "parameters": {...}}}
         Responses format: {"type": "function", "name": "...", "parameters": {...}}
         """
-        transformed: Final = []
+        transformed: Final[list[dict[str, object]]] = []
         for tool in tools:
             if tool.get("type") == "function" and "function" in tool:
                 func = tool["function"]
-                item: dict[str, Any] = {
+                item: dict[str, object] = {
                     "type": "function",
                     "name": func.get("name", ""),
                     "description": func.get("description", ""),
@@ -191,7 +191,7 @@ class OpenAICountTokensConfig:
             (input_items, instructions) tuple where instructions is extracted
             from system/developer messages.
         """
-        input_items: Final[list[dict[str, Any]]] = []
+        input_items: Final[list[dict[str, object]]] = []
         instructions_parts: Final[list[str]] = []
 
         for msg in messages:

litellm/llms/openai/responses/guardrail_translation/handler.py

@@ -110,7 +110,7 @@ class ResponsesStreamChunk(TypedDict, total=False):
     content_index: ReadOnly[int]
 
 
-def _next_stream_sequence_number(responses_so_far: Sequence[Any] | None) -> int:
+def _next_stream_sequence_number(responses_so_far: Sequence[object] | None) -> int:
     sequence_numbers: Final = (
         item.get("sequence_number") if isinstance(item, dict) else getattr(item, "sequence_number", None)
         for item in reversed(responses_so_far or ())
@@ -337,7 +337,7 @@ class OpenAIResponsesHandler(BaseTranslation):
 
     def _extract_input_text_and_images(
         self,
-        message: Any,
+        message: Mapping[str, object],
         msg_idx: int,
         texts_to_check: list[str],
         images_to_check: list[str],
@@ -661,8 +661,8 @@ class OpenAIResponsesHandler(BaseTranslation):
     def build_stream_error_items(
         self,
         exc: "HTTPException",
-        responses_so_far: Sequence[Any] | None = None,
-    ) -> Sequence[Any] | None:
+        responses_so_far: Sequence[object] | None = None,
+    ) -> Sequence[object] | None:
         from litellm.proxy.common_request_processing import (
             serialize_http_exception_detail,
         )

litellm/llms/openai/videos/transformation.py

@@ -571,8 +571,8 @@ class OpenAIVideoConfig(BaseVideoConfig):
 
     def _add_image_to_files(
         self,
-        files_list: list[tuple[str, Any]],
-        image: Any,
+        files_list: list[tuple[str, FileTypes]],
+        image: FileContent,
         field_name: str,
     ) -> None:
         """Add an image to the files list with appropriate content type"""

litellm/llms/openrouter/image_edit/transformation.py

@@ -152,7 +152,7 @@ class OpenRouterImageEditConfig(BaseImageEditConfig):
         litellm_params: GenericLiteLLMParams,
         headers: dict,
     ) -> tuple[dict, RequestFiles]:
-        content_parts: Final[list[dict[str, Any]]] = []
+        content_parts: Final[list[dict[str, object]]] = []
 
         # Add source image(s) as base64 data URLs
         if image is not None:
@@ -174,7 +174,7 @@ class OpenRouterImageEditConfig(BaseImageEditConfig):
         if prompt:
             content_parts.append({"type": "text", "text": prompt})
 
-        request_body: Final[dict[str, Any]] = {
+        request_body: Final[dict[str, object]] = {
             "model": model,
             "messages": [
                 {

litellm/llms/pass_through/guardrail_translation/handler.py

@@ -127,7 +127,7 @@ class PassThroughEndpointHandler(BaseTranslation):
 
     async def process_output_response(
         self,
-        response: Any,
+        response: object,
         guardrail_to_apply: "CustomGuardrail",
         litellm_logging_obj: Optional["LiteLLMLoggingObj"] = None,
         user_api_key_dict: Any | None = None,
@@ -236,7 +236,7 @@ class LlmPassthroughRouteHandler(BaseTranslation):
 
     async def process_output_response(
         self,
-        response: Any,
+        response: object,
         guardrail_to_apply: "CustomGuardrail",
         litellm_logging_obj: Optional["LiteLLMLoggingObj"] = None,
         user_api_key_dict: Any | None = None,

litellm/llms/perplexity/embedding/transformation.py

@@ -13,7 +13,7 @@ This module decodes them into float arrays for OpenAI-compatible responses.
 
 import base64
 import struct
-from typing import Any, Final
+from typing import Final
 
 import httpx
 
@@ -117,7 +117,7 @@ class PerplexityEmbeddingConfig(BaseEmbeddingConfig):
         }
 
     @staticmethod
-    def _decode_base64_embedding(embedding_value: Any) -> list[float]:
+    def _decode_base64_embedding(embedding_value: object) -> object:
         """
         Decode a Perplexity embedding into a list of floats.
 
@@ -154,7 +154,7 @@ class PerplexityEmbeddingConfig(BaseEmbeddingConfig):
         model_response.object = raw_response_json.get("object", "list")
 
         raw_data: Final = raw_response_json.get("data", [])
-        decoded_data: Final[list[dict[str, Any]]] = []
+        decoded_data: Final[list[dict[str, object]]] = []
         for item in raw_data:
             decoded_item = dict(item)
             decoded_item["embedding"] = self._decode_base64_embedding(item.get("embedding"))

litellm/llms/ragflow/vector_stores/transformation.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, Final
 
 import httpx
@@ -91,7 +92,7 @@ class RAGFlowVectorStoreConfig(BaseVectorStoreConfig):
         api_base: str,
         litellm_logging_obj: LiteLLMLoggingObj,
         litellm_params: dict,
-        extra_body: dict[str, Any] | None = None,
+        extra_body: Mapping[str, object] | None = None,
     ) -> tuple[str, dict]:
         """RAGFlow vector stores are management-only, search is not supported."""
         raise NotImplementedError("RAGFlow vector stores support dataset management only, not search/retrieval")
@@ -121,7 +122,7 @@ class RAGFlowVectorStoreConfig(BaseVectorStoreConfig):
             raise ValueError("name is required for RAGFlow dataset creation")
 
         # Build request body
-        request_body: Final[dict[str, Any]] = {
+        request_body: Final[dict[str, object]] = {
             "name": name,
         }
 

litellm/llms/vertex_ai/fine_tuning/handler.py

@@ -2,7 +2,7 @@ import json
 import traceback
 from collections.abc import Coroutine
 from datetime import datetime
-from typing import Any, Final, Literal
+from typing import Final, Literal
 
 import httpx
 
@@ -207,7 +207,7 @@ class VertexFineTuningAPI(VertexLLM):
         timeout: float | httpx.Timeout,
         kwargs: dict | None = None,
         original_hyperparameters: dict | None = {},
-    ) -> LiteLLMFineTuningJob | Coroutine[Any, Any, LiteLLMFineTuningJob]:
+    ) -> LiteLLMFineTuningJob | Coroutine[object, object, LiteLLMFineTuningJob]:
         verbose_logger.debug("creating fine tuning job, args= %s", create_fine_tuning_job_data)
         _auth_header, vertex_project = self._ensure_access_token(
             credentials=vertex_credentials,

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -430,7 +430,7 @@ def _clear_oauth_state_cookie(response: Response, request: Request, state: str)
     )
 
 
-def _get_validated_client_redirect_uri(request: Request, state_data: dict[str, Any]) -> str:
+def _get_validated_client_redirect_uri(request: Request, state_data: Mapping[str, object]) -> str:
     """Return a trusted (same-origin, loopback, or ops-allowlisted)
     client redirect URI from OAuth state.
     """
@@ -469,7 +469,7 @@ def _resolve_oauth2_server_for_root_endpoints(
     return None
 
 
-def _normalize_for_token_comparison(value: Any) -> str:
+def _normalize_for_token_comparison(value: object) -> str:
     """Stringify ``value`` for token-rule comparison.
 
     Booleans are lower-cased so Python's ``True`` / ``False`` line up with
@@ -481,8 +481,8 @@ def _normalize_for_token_comparison(value: Any) -> str:
 
 
 def _validate_token_response(
-    token_response: dict[str, Any],
-    validation_rules: dict[str, Any],
+    token_response: Mapping[str, object],
+    validation_rules: Mapping[str, object],
     server_id: str,
 ) -> None:
     """Raise HTTPException 403 if any validation rule doesn't match the token response.
@@ -496,10 +496,10 @@ def _validate_token_response(
     responses of ``{"verified": true}``.
     """
     for key, expected in validation_rules.items():
-        actual: Any = token_response.get(key)
+        actual: object | None = token_response.get(key)
         # Try dot-notation traversal when top-level lookup returns None
         if actual is None and "." in key:
-            obj: Any = token_response
+            obj: object = token_response
             for part in key.split("."):
                 if isinstance(obj, dict):
                     obj = obj.get(part)

litellm/proxy/_experimental/mcp_server/elicitation_handler.py

@@ -9,7 +9,7 @@ MCP Spec Reference:
     https://modelcontextprotocol.io/specification/2025-11-25/client/elicitation
 """
 
-from typing import TYPE_CHECKING, Any, Final, Union
+from typing import TYPE_CHECKING, Final, Protocol, Union
 
 from litellm._logging import verbose_logger
 
@@ -37,11 +37,21 @@ except ImportError:
     MCP_ELICITATION_AVAILABLE = False
 
 
+class _DownstreamElicitSession(Protocol):
+    """The downstream MCP client session methods this module relays elicitation requests through."""
+
+    async def elicit_url(self, message: str, url: str, elicitation_id: str) -> "ElicitResult": ...
+
+    async def elicit_form(self, message: str, requestedSchema: dict[str, object]) -> "ElicitResult": ...
+
+    async def elicit(self, message: str, requestedSchema: dict[str, object]) -> "ElicitResult": ...
+
+
 async def handle_elicitation_request(
-    context: Any,
+    context: object,
     params: "ElicitRequestParams",
-    downstream_session: Any | None = None,
-    downstream_capabilities: Any | None = None,
+    downstream_session: _DownstreamElicitSession | None = None,
+    downstream_capabilities: object = None,
 ) -> Union["ElicitResult", "ErrorData"]:
     """
     Handle an MCP elicitation/create request from an upstream MCP server.
@@ -94,8 +104,8 @@ async def handle_elicitation_request(
 
 async def _relay_elicitation_to_downstream(
     params: "ElicitRequestParams",
-    downstream_session: Any,
-    downstream_capabilities: Any | None = None,
+    downstream_session: _DownstreamElicitSession,
+    downstream_capabilities: object = None,
 ) -> Union["ElicitResult", "ErrorData"]:
     """
     Relay an elicitation request to the downstream MCP client.
@@ -111,17 +121,17 @@ async def _relay_elicitation_to_downstream(
     mode: Final = getattr(params, "mode", "form")
     # Check if the downstream client supports the requested mode
     if downstream_capabilities is not None:
-        elicit_caps: Final = getattr(downstream_capabilities, "elicitation", None)
+        elicit_caps: Final[object] = getattr(downstream_capabilities, "elicitation", None)
         if elicit_caps is None:
             verbose_logger.info("MCP elicitation: downstream client does not support elicitation")
             return ElicitResult(action="decline")
         if mode == "url":
-            url_cap: Final = getattr(elicit_caps, "url", None)
+            url_cap: Final[object] = getattr(elicit_caps, "url", None)
             if url_cap is None:
                 verbose_logger.info("MCP elicitation: downstream client does not support URL mode")
                 return ElicitResult(action="decline")
         if mode == "form":
-            form_cap: Final = getattr(elicit_caps, "form", None)
+            form_cap: Final[object] = getattr(elicit_caps, "form", None)
             if form_cap is None:
                 verbose_logger.info("MCP elicitation: downstream client does not support form mode")
                 return ElicitResult(action="decline")
@@ -135,14 +145,14 @@ async def _relay_elicitation_to_downstream(
             result = await downstream_session.elicit_url(
                 message=params.message,
                 url=params.url,
-                elicitation_id=getattr(params, "elicitationId", None),
+                elicitation_id=params.elicitationId,
             )
         elif isinstance(params, ElicitRequestFormParams):
             # Form mode: relay structured form to client
             verbose_logger.info("MCP elicitation: relaying form mode to downstream")
             result = await downstream_session.elicit_form(
                 message=params.message,
-                requestedSchema=getattr(params, "requestedSchema", None),
+                requestedSchema=params.requestedSchema,
             )
         else:
             # Fallback for generic ElicitRequestParams — pass an empty schema

litellm/proxy/_experimental/mcp_server/server.py

@@ -3477,7 +3477,7 @@ if MCP_AVAILABLE:
         is best-effort in that mode.
         """
 
-        def _bytes_for_hash(value: Any) -> bytes | None:
+        def _bytes_for_hash(value: object) -> bytes | None:
             """Only hash str/bytes secrets; skip mocks and other unexpected types."""
             if value is None:
                 return None

litellm/proxy/a2a/version_convert.py

@@ -25,7 +25,6 @@ The two wire shapes:
 """
 
 from collections.abc import Callable
-from types import ModuleType
 from typing import Final, Literal
 
 from pydantic import BaseModel
@@ -181,7 +180,7 @@ def _send_result_to(result: JsonDict, target: A2AVersion, request_id: RequestId)
     )
 
     if target == "1.0":
-        compat_result: Final = _validate_message_or_task(result, types_v03)
+        compat_result: Final = _validate_message_or_task(result)
         response: Final = types_v03.SendMessageResponse(
             root=types_v03.SendMessageSuccessResponse(
                 id=str(request_id) if request_id is not None else "",
@@ -285,7 +284,7 @@ def _stream_result_to(result: JsonDict, target: A2AVersion, request_id: RequestI
     )
 
     if target == "1.0":
-        event: Final = _validate_stream_event(result, types_v03)
+        event: Final = _validate_stream_event(result)
         wrapper: Final = types_v03.SendStreamingMessageSuccessResponse(
             id=str(request_id) if request_id is not None else "",
             result=event,  # pyright: ignore[reportArgumentType]
@@ -318,13 +317,17 @@ def _convert_agent_card(card: JsonDict, target: A2AVersion) -> JsonDict:
     return MessageToDict(core, preserving_proto_field_name=False)
 
 
-def _validate_message_or_task(result: JsonDict, types_v03: ModuleType) -> BaseModel:
+def _validate_message_or_task(result: JsonDict) -> BaseModel:
+    from a2a.compat.v0_3.conversions import types_v03
+
     if result.get("kind") == "task":
         return types_v03.Task.model_validate(result)
     return types_v03.Message.model_validate(result)
 
 
-def _validate_stream_event(result: JsonDict, types_v03: ModuleType) -> BaseModel:
+def _validate_stream_event(result: JsonDict) -> BaseModel:
+    from a2a.compat.v0_3.conversions import types_v03
+
     kind: Final = result.get("kind")
     if kind == "task":
         return types_v03.Task.model_validate(result)

litellm/proxy/client/cli/commands/models.py

@@ -17,8 +17,8 @@ from ... import Client
 @dataclass
 class ModelYamlInfo:
     model_name: str
-    model_params: dict[str, Any]
-    model_info: dict[str, Any]
+    model_params: dict[str, object]
+    model_info: dict[str, object]
     model_id: str
     access_groups: list[str]
     provider: str

litellm/proxy/common_utils/cache_coordinator.py

@@ -13,14 +13,14 @@ pattern: global spend, feature flags, config, or other shared read-through data.
 import asyncio
 import time
 from collections.abc import Awaitable, Callable
-from typing import Any, Final, Protocol, TypeVar
+from typing import Final, Protocol, TypeVar
 
 from litellm._logging import verbose_proxy_logger
 
 T = TypeVar("T")
 
 
-class AsyncCacheProtocol(Protocol):
+class AsyncCacheProtocol(Protocol[T]):
     """Protocol for cache backends used by EventDrivenCacheCoordinator.
 
     Matches ``DualCache`` / ``UserApiKeyCache`` call shapes (explicit optional params
@@ -30,18 +30,18 @@ class AsyncCacheProtocol(Protocol):
     async def async_get_cache(
         self,
         key: str,
-        parent_otel_span: Any = None,
+        parent_otel_span: object = None,
         local_only: bool = False,
-        **kwargs: Any,
-    ) -> Any: ...
+        **kwargs: object,
+    ) -> T | None: ...
 
     async def async_set_cache(
         self,
         key: str,
-        value: Any,
+        value: T,
         local_only: bool = False,
-        **kwargs: Any,
-    ) -> Any: ...
+        **kwargs: object,
+    ) -> object: ...
 
 
 class EventDrivenCacheCoordinator:
@@ -64,11 +64,11 @@ class EventDrivenCacheCoordinator:
         self._query_in_progress = False
         self._log_prefix = log_prefix
 
-    async def _get_cached(self, cache_key: str, cache: AsyncCacheProtocol) -> Any | None:
+    async def _get_cached(self, cache_key: str, cache: AsyncCacheProtocol[T]) -> T | None:
         """Return value from cache if present, else None."""
         return await cache.async_get_cache(key=cache_key)
 
-    def _log_cache_hit(self, value: T) -> None:
+    def _log_cache_hit(self, value: object) -> None:
         if self._log_prefix:
             verbose_proxy_logger.debug("%s Cache hit, value: %s", self._log_prefix, value)
 
@@ -98,7 +98,7 @@ class EventDrivenCacheCoordinator:
         self,
         event: asyncio.Event,
         cache_key: str,
-        cache: AsyncCacheProtocol,
+        cache: AsyncCacheProtocol[T],
     ) -> T | None:
         """Wait for loader to finish, then read from cache."""
         await event.wait()
@@ -118,7 +118,7 @@ class EventDrivenCacheCoordinator:
     async def _load_and_cache(
         self,
         cache_key: str,
-        cache: AsyncCacheProtocol,
+        cache: AsyncCacheProtocol[T],
         load_fn: Callable[[], Awaitable[T]],
     ) -> T | None:
         """Double-check cache, run load_fn, set cache, return value. Caller must call _signal_done in finally."""
@@ -163,7 +163,7 @@ class EventDrivenCacheCoordinator:
     async def get_or_load(
         self,
         cache_key: str,
-        cache: AsyncCacheProtocol,
+        cache: AsyncCacheProtocol[T],
         load_fn: Callable[[], Awaitable[T]],
     ) -> T | None:
         """

litellm/proxy/common_utils/http_parsing_utils.py

@@ -1,6 +1,6 @@
 import json
 import re
-from collections.abc import Collection
+from collections.abc import Collection, Mapping
 from typing import Any, Final
 
 import orjson
@@ -186,7 +186,7 @@ def _safe_get_request_headers(request: Request | None) -> dict:
     if request is None:
         return {}
     state: Final = getattr(request, "state", None)
-    cached: Final = getattr(state, "_cached_headers", None)
+    cached: Final[object] = getattr(state, "_cached_headers", None)
     if isinstance(cached, dict):
         return cached
     if cached is not None:
@@ -344,7 +344,9 @@ async def get_request_body(request: Request) -> dict[str, Any]:
     return {}
 
 
-def extract_nested_form_metadata(form_data: dict[str, Any], prefix: str = "litellm_metadata[") -> dict[str, Any]:
+def extract_nested_form_metadata(
+    form_data: Mapping[str, object], prefix: str = "litellm_metadata["
+) -> dict[str, object]:
     """
     Extract nested metadata from form data with bracket notation.
 
@@ -382,7 +384,7 @@ def extract_nested_form_metadata(form_data: dict[str, Any], prefix: str = "litel
     if not form_data:
         return {}
 
-    metadata: Final[dict[str, Any]] = {}
+    metadata: Final[dict[str, object]] = {}
 
     for key, value in form_data.items():
         # Skip keys that don't start with the prefix
@@ -430,7 +432,7 @@ def extract_nested_form_metadata(form_data: dict[str, Any], prefix: str = "litel
     return metadata
 
 
-def get_tags_from_request_body(request_body: dict) -> list[str]:
+def get_tags_from_request_body(request_body: Mapping[str, object]) -> list[str]:
     """
     Extract tags from request body metadata.
 
@@ -447,12 +449,12 @@ def get_tags_from_request_body(request_body: dict) -> list[str]:
     if isinstance(metadata, str):
         from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
 
-        parsed: Final = safe_json_loads(metadata)
+        parsed: Final[object] = safe_json_loads(metadata)
         metadata = parsed if isinstance(parsed, dict) else {}
     elif not isinstance(metadata, dict):
         metadata = {}
-    tags_in_metadata: Final[Any] = metadata.get("tags", [])
-    tags_in_request_body: Final[Any] = request_body.get("tags", [])
+    tags_in_metadata: Final[object] = metadata.get("tags", [])
+    tags_in_request_body: Final[object] = request_body.get("tags", [])
     combined_tags: Final[list[str]] = []
 
     ######################################

litellm/proxy/container_endpoints/handler_factory.py

@@ -7,7 +7,7 @@ FastAPI route handlers for ALL container file endpoints.
 
 import json
 from pathlib import Path
-from typing import Any, Final
+from typing import Final
 
 from fastapi import APIRouter, Depends, Request, Response
 from fastapi.responses import ORJSONResponse
@@ -194,7 +194,7 @@ async def _process_binary_request(
         user_api_key_dict=user_api_key_dict,
         custom_llm_provider=custom_llm_provider,
     )
-    data: Final[dict[str, Any]] = {
+    data: Final[dict[str, object]] = {
         "file_id": file_id,
         **(
             await get_container_forwarding_params(
@@ -374,7 +374,7 @@ async def _process_request(
     )
 
     query_params: Final = dict(request.query_params)
-    data: Final[dict[str, Any]] = {
+    data: Final[dict[str, object]] = {
         "query_params": query_params,
         **path_params,
     }

litellm/proxy/db/prisma_client.py

@@ -503,7 +503,7 @@ class PrismaWrapper:
     async def recreate_prisma_client(
         self,
         new_db_url: str,
-        http_client: Any | None = None,
+        http_client: object | None = None,
         *,
         expected_generation: int | None = None,
     ) -> bool:
@@ -541,7 +541,7 @@ class PrismaWrapper:
     async def _recreate_prisma_client_locked(
         self,
         new_db_url: str,
-        http_client: Any | None = None,
+        http_client: object | None = None,
         *,
         expected_generation: int | None = None,
     ) -> bool:

litellm/proxy/guardrails/_content_utils.py

@@ -54,7 +54,7 @@ def _part_text(part: Mapping[str, object]) -> str | None:
     return None
 
 
-def _iter_text_parts_in_content(content: Any) -> Iterator[str]:
+def _iter_text_parts_in_content(content: object) -> Iterator[str]:
     """Yield text fragments from a ``message.content`` value (string or
     multimodal list). Non-text parts (images, audio, …) are skipped."""
     if isinstance(content, str):
@@ -75,13 +75,13 @@ def _iter_text_parts_in_content(content: Any) -> Iterator[str]:
                 yield text
 
 
-def _coerce_input_to_messages(input_value: Any) -> list[dict[str, Any]]:
+def _coerce_input_to_messages(input_value: object) -> list[dict[str, object]]:
     """Coerce a Responses-API ``data["input"]`` value into chat-style messages."""
     if isinstance(input_value, str):
         return [{"role": "user", "content": input_value}]
     if not isinstance(input_value, list):
         return []
-    messages: Final[list[dict[str, Any]]] = []
+    messages: Final[list[dict[str, object]]] = []
     for item in input_value:
         if isinstance(item, str):
             messages.append({"role": "user", "content": item})
@@ -110,7 +110,7 @@ def _coerce_input_to_messages(input_value: Any) -> list[dict[str, Any]]:
     return messages
 
 
-def _iter_inspection_messages(data: dict[str, Any]) -> Iterator[dict[str, Any]]:
+def _iter_inspection_messages(data: Mapping[str, object]) -> Iterator[object]:
     """Yield every message-like dict, walking ``messages`` AND ``input``."""
     messages: Final = data.get("messages")
     if isinstance(messages, list):
@@ -118,7 +118,7 @@ def _iter_inspection_messages(data: dict[str, Any]) -> Iterator[dict[str, Any]]:
     yield from _coerce_input_to_messages(data.get("input"))
 
 
-def iter_message_text(data: dict[str, Any]) -> Iterator[str]:
+def iter_message_text(data: Mapping[str, object]) -> Iterator[str]:
     """Yield every text fragment from ``messages`` AND ``input``.
 
     Walks every role (user, assistant, system, …) — guardrails inspect
@@ -139,7 +139,7 @@ def walk_user_text(data: dict[str, Any], visit: Callable[[str], str]) -> int:
     """
     visited = 0
 
-    def _rewrite_content(content: Any) -> Any:
+    def _rewrite_content(content: object) -> object:
         nonlocal visited
         if isinstance(content, str):
             if content:
@@ -147,7 +147,7 @@ def walk_user_text(data: dict[str, Any], visit: Callable[[str], str]) -> int:
                 return visit(content)
             return content
         if isinstance(content, list):
-            new_parts: Final[list[Any]] = []
+            new_parts: Final[list[object]] = []
             for part in content:
                 if isinstance(part, str) and part:
                     visited += 1
@@ -218,7 +218,7 @@ def apply_redacted_messages_back(data: dict[str, Any], redacted_messages: list[d
         data["input"] = "\n".join(text_parts)
 
 
-def has_non_string_content(data: dict[str, Any]) -> bool:
+def has_non_string_content(data: Mapping[str, object]) -> bool:
     """Return True if any inspected content is not a plain string.
 
     Used by hooks whose mask/redact path operates on string offsets and

litellm/proxy/guardrails/guardrail_hooks/qualifire/qualifire.py

@@ -139,7 +139,7 @@ class QualifireGuardrail(CustomGuardrail):
             ]
         )
 
-    def _convert_messages_to_api_format(self, messages: list[AllMessageValues]) -> list[dict[str, Any]]:
+    def _convert_messages_to_api_format(self, messages: list[AllMessageValues]) -> list[dict[str, object]]:
         """
         Convert LiteLLM messages to Qualifire API format.
         Supports tool calls for tool_selection_quality_check.
@@ -167,7 +167,7 @@ class QualifireGuardrail(CustomGuardrail):
                         text_parts.append(part)
                 content = "\n".join(text_parts)
 
-            api_message: dict[str, Any] = {
+            api_message: dict[str, object] = {
                 "role": role,
                 "content": content if isinstance(content, str) else str(content),
             }
@@ -205,7 +205,7 @@ class QualifireGuardrail(CustomGuardrail):
 
         return api_messages
 
-    def _convert_tools_to_api_format(self, tools: list[Any] | None) -> list[dict[str, Any]] | None:
+    def _convert_tools_to_api_format(self, tools: list[object] | None) -> list[dict[str, object]] | None:
         """
         Convert OpenAI-format tools to Qualifire API format.
 
@@ -264,13 +264,13 @@ class QualifireGuardrail(CustomGuardrail):
 
     def _build_evaluate_payload(
         self,
-        api_messages: list[dict[str, Any]],
+        api_messages: list[dict[str, object]],
         output: str | None,
         assertions: list[str] | None,
-        available_tools: list[dict[str, Any]] | None,
-    ) -> dict[str, Any]:
+        available_tools: list[dict[str, object]] | None,
+    ) -> dict[str, object]:
         """Build payload dictionary for the /api/evaluation/evaluate endpoint."""
-        payload: Final[dict[str, Any]] = {"messages": api_messages}
+        payload: Final[dict[str, object]] = {"messages": api_messages}
 
         if output is not None:
             payload["output"] = output
@@ -305,7 +305,7 @@ class QualifireGuardrail(CustomGuardrail):
         messages: list[AllMessageValues],
         output: str | None,
         dynamic_params: dict[str, Any],
-        available_tools: list[Any] | None = None,
+        available_tools: list[object] | None = None,
     ) -> None:
         """
         Core Qualifire check logic - shared between hooks.

litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py

@@ -97,7 +97,7 @@ class SingulrGuardrail(CustomGuardrail):
         request_data: dict[str, Any],
         inputs: GenericGuardrailAPIInputs,
         input_type: str,
-    ) -> dict[str, Any]:
+    ) -> dict[str, object]:
         if not request_data:
             texts: Final = inputs.get("texts", [])
 
@@ -138,7 +138,7 @@ class SingulrGuardrail(CustomGuardrail):
             if value
         )
 
-    async def _call_api(self, payload: dict[str, Any]) -> SingulrGuardrailResponse | None:
+    async def _call_api(self, payload: dict[str, object]) -> SingulrGuardrailResponse | None:
         endpoint: Final = f"{self.singulr_api_base}{_GUARD_ENDPOINT}"
         verbose_proxy_logger.debug("Singulr: %s", endpoint)
 

litellm/proxy/guardrails/guardrail_hooks/unified_guardrail/unified_guardrail.py

@@ -878,7 +878,7 @@ class UnifiedLLMGuardrails(CustomLogger):
     async def async_post_call_streaming_iterator_hook(
         self,
         user_api_key_dict: UserAPIKeyAuth,
-        response: Any,
+        response: AsyncIterable[object],
         request_data: dict,
         guardrail_to_apply: CustomGuardrail | None = None,
         buffer_until_moderated_default: bool = False,

litellm/proxy/management_endpoints/team_callback_endpoints.py

@@ -9,7 +9,7 @@ import copy
 import json
 import traceback
 from datetime import datetime, timezone
-from typing import Annotated, Any, Final
+from typing import Annotated, Final
 
 from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
 
@@ -62,7 +62,7 @@ def _validate_team_callback(data: "AddTeamCallback") -> None:
         raise _callback_config_error(error)
 
 
-def _redact_callback_secrets(metadata: Any) -> Any:
+def _redact_callback_secrets(metadata: object) -> object:
     """Strip secret values out of a team-metadata snapshot before audit logging.
 
     Both ``team_metadata["logging"]`` (list of ``AddTeamCallback`` dicts) and
@@ -176,8 +176,8 @@ def _log_audit_task_exception(task: "asyncio.Task[None]") -> None:
 async def _emit_team_callback_audit_log(
     *,
     team_id: str,
-    before_metadata: Any,
-    after_metadata: Any,
+    before_metadata: object,
+    after_metadata: object,
     user_api_key_dict: UserAPIKeyAuth,
     litellm_changed_by: str | None,
 ) -> None:

litellm/proxy/openai_evals_endpoints/endpoints.py

@@ -35,7 +35,7 @@ async def create_eval(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Create a new evaluation.
 
@@ -131,7 +131,7 @@ async def list_evals(
     order_by: str | None = None,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     List evaluations with pagination.
 
@@ -228,7 +228,7 @@ async def get_eval(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Get a specific evaluation by ID.
 
@@ -316,7 +316,7 @@ async def update_eval(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Update an evaluation.
 
@@ -406,7 +406,7 @@ async def delete_eval(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Delete an evaluation.
 
@@ -494,7 +494,7 @@ async def cancel_eval(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Cancel a running evaluation.
 
@@ -587,7 +587,7 @@ async def create_run(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Create a new run for an evaluation.
 
@@ -690,7 +690,7 @@ async def list_runs(
     order: str | None = None,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     List all runs for an evaluation with pagination.
 
@@ -780,7 +780,7 @@ async def get_run(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Get a specific run by ID.
 
@@ -867,7 +867,7 @@ async def cancel_run(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Cancel a running run.
 
@@ -956,7 +956,7 @@ async def delete_run(
     request: Request,
     custom_llm_provider: str | None = "openai",
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
-):
+) -> object:
     """
     Delete a run.
 

litellm/proxy/policy_engine/init_policies.py

@@ -6,6 +6,7 @@ Configuration structure:
 - policy_attachments: Define WHERE policies apply (teams, keys, models)
 """
 
+from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Final, Optional
 
 from litellm._logging import verbose_proxy_logger
@@ -25,8 +26,8 @@ _reset_color_code: Final = "\033[0m"
 
 
 def _print_policies_on_startup(
-    policies_config: dict[str, Any],
-    policy_attachments_config: list[dict[str, Any]] | None = None,
+    policies_config: Mapping[str, Mapping[str, object]],
+    policy_attachments_config: Sequence[Mapping[str, object]] | None = None,
 ) -> None:
     """
     Print loaded policies to console on startup (similar to model list).

litellm/proxy/vector_store_endpoints/management_endpoints.py

@@ -87,7 +87,7 @@ def _get_embedding_config_cache() -> InMemoryCache:
     return _embedding_config_cache
 
 
-def _redact_sensitive_litellm_params(litellm_params: Any, _depth: int = 0) -> Any:
+def _redact_sensitive_litellm_params(litellm_params: object, _depth: int = 0) -> Any:
     """
     Replace credential-bearing values in ``litellm_params`` with
     ``REDACTED_BY_LITELM`` while preserving non-secret keys (``api_base``,
@@ -119,7 +119,7 @@ def _redact_sensitive_litellm_params(litellm_params: Any, _depth: int = 0) -> An
         return json.dumps(_redact_sensitive_litellm_params(parsed, _depth + 1))
     if not isinstance(litellm_params, dict):
         return litellm_params
-    out: Final[dict[str, Any]] = {}
+    out: Final[dict[str, object]] = {}
     for k, v in litellm_params.items():
         if _LITELLM_PARAMS_MASKER.is_sensitive_key(k):
             out[k] = REDACTED_BY_LITELM_STRING

litellm/rag/ingestion/gemini_ingestion.py

@@ -7,7 +7,7 @@ so this implementation skips the embedding step and directly uploads files.
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Final, cast
+from typing import TYPE_CHECKING, Final, cast
 
 from litellm._logging import verbose_logger
 from litellm.llms.custom_httpx.http_handler import (
@@ -83,7 +83,7 @@ class GeminiRAGIngestion(BaseRAGIngestion):
         """
         vector_store_id = self.vector_store_config.get("vector_store_id")
 
-        vector_store_config: Final = cast(dict[str, Any], self.vector_store_config)
+        vector_store_config: Final = self.vector_store_config
 
         # Get API credentials
         api_key: Final = cast(str | None, vector_store_config.get("api_key")) or GeminiModelInfo.get_api_key()
@@ -228,7 +228,7 @@ class GeminiRAGIngestion(BaseRAGIngestion):
         url: Final = f"{api_base}/upload/v1beta/{vector_store_id}:uploadToFileSearchStore"
 
         # Build request body with chunking config and metadata if provided
-        request_body: Final[dict[str, Any]] = {"displayName": filename}
+        request_body: Final[dict[str, object]] = {"displayName": filename}
 
         # Add chunking configuration if provided
         chunking_strategy: Final = self.chunking_strategy
@@ -244,7 +244,7 @@ class GeminiRAGIngestion(BaseRAGIngestion):
 
         # Add custom metadata if provided in vector_store_config
         custom_metadata: Final = cast(
-            list[dict[str, Any]] | None,
+            list[dict[str, object]] | None,
             self.vector_store_config.get("custom_metadata"),
         )
         if custom_metadata:

litellm/realtime_api/main.py

@@ -4,7 +4,7 @@ import asyncio
 import os
 from collections.abc import Mapping
 from types import MappingProxyType
-from typing import Any, Final, Literal, cast
+from typing import TYPE_CHECKING, Any, Final, Literal, cast
 
 import litellm
 from litellm.constants import (
@@ -41,6 +41,9 @@ from ..llms.vertex_ai.vertex_llm_base import VertexBase
 from ..llms.xai.realtime.handler import XAIRealtime
 from ..utils import client as wrapper_client
 
+if TYPE_CHECKING:
+    from litellm.llms.base_llm.realtime.http_transformation import BaseRealtimeHTTPConfig
+
 azure_realtime: Final = AzureOpenAIRealtime()
 openai_realtime: Final = OpenAIRealtime()
 bedrock_realtime: Final = BedrockRealtime()
@@ -50,7 +53,7 @@ base_llm_http_handler = BaseLLMHTTPHandler()
 _EMPTY_MODEL_PARAMS: Final[Mapping[str, Any]] = MappingProxyType({})
 
 
-def _with_resolved_session_model(session: dict[str, Any], model_name: str) -> dict[str, Any]:
+def _with_resolved_session_model(session: dict[str, object], model_name: str) -> dict[str, object]:
     if "model" not in session:
         return session
     return {**session, "model": model_name}
@@ -70,7 +73,7 @@ def _get_realtime_http_provider_config(
     dynamic_api_base: str | None,
     dynamic_api_key: str | None,
     litellm_params: GenericLiteLLMParams,
-) -> tuple[Any, str, str]:
+) -> tuple["BaseRealtimeHTTPConfig | None", str, str]:
     """
     Return (provider_config, resolved_api_base, resolved_api_key) for the
     realtime HTTP endpoints (client_secrets / realtime_calls).

litellm/repositories/config_repository.py

@@ -17,6 +17,11 @@ from litellm._logging import verbose_proxy_logger
 from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value_helper
 
 
+def _decoded_json(raw: str) -> object:
+    """Decode a JSON-encoded config row value into an opaque object."""
+    return json.loads(raw)
+
+
 class _ConfigRow(Protocol):
     @property
     def param_name(self) -> str: ...
@@ -48,7 +53,7 @@ class _PrismaHandle(Protocol):
 class ConfigParam:
     """Simple wrapper for config parameter from DB."""
 
-    def __init__(self, param_name: str, param_value: Any):
+    def __init__(self, param_name: str, param_value: object):
         self.param_name = param_name
         self.param_value = param_value
 
@@ -85,12 +90,12 @@ class ConfigRepository:
         record: Final = await self._config_table.find_unique(where={"param_name": param_name})
         if record is None:
             return None
-        param_value = record.param_value
+        param_value: object = record.param_value
         if isinstance(param_value, str):
-            param_value = json.loads(param_value)
+            param_value = _decoded_json(param_value)
         return ConfigParam(param_name=param_name, param_value=param_value)
 
-    async def set_param(self, param_name: str, param_value: Any) -> ConfigParam:
+    async def set_param(self, param_name: str, param_value: object) -> ConfigParam:
         """Set a config parameter in the database."""
         value_json: Final = json.dumps(param_value) if not isinstance(param_value, str) else param_value
         await self._config_table.upsert(
@@ -115,9 +120,9 @@ class ConfigRepository:
         records: Final = await self._config_table.find_many()
         result: Final[dict[str, object]] = {}
         for record in records:
-            param_value = record.param_value
+            param_value: object = record.param_value
             if isinstance(param_value, str):
-                param_value = json.loads(param_value)
+                param_value = _decoded_json(param_value)
             result[record.param_name] = param_value
         return result
 

litellm/router_strategy/adaptive_router/hooks.py

@@ -13,7 +13,8 @@ from __future__ import annotations
 
 import hashlib
 import json
-from typing import Any, Final
+from collections.abc import Mapping, Sequence
+from typing import TYPE_CHECKING, Any, Final
 
 from litellm._logging import verbose_router_logger
 from litellm.integrations.custom_logger import CustomLogger
@@ -26,6 +27,9 @@ from litellm.router_strategy.adaptive_router.config import (
 )
 from litellm.router_strategy.adaptive_router.signals import Turn
 
+if TYPE_CHECKING:
+    from litellm.proxy._types import UserAPIKeyAuth
+
 # Identity fields hashed into a derived session key so the same conversation
 # from the same caller produces a stable key, while different keys/teams/users
 # stay segregated even if they happen to send identical first messages.
@@ -100,8 +104,8 @@ def _last_user_content(messages: list[dict[str, Any]] | None) -> str | None:
 
 
 def _recent_tool_results(
-    messages: list[dict[str, Any]] | None,
-) -> list[dict[str, Any]]:
+    messages: Sequence[Mapping[str, object]] | None,
+) -> list[dict[str, object]]:
     """Extract the current turn's tool result payloads from the request messages.
 
     Tool results are `role == "tool"` messages that sit at the tail of the
@@ -115,7 +119,7 @@ def _recent_tool_results(
     """
     if not messages:
         return []
-    results: Final[list[dict[str, Any]]] = []
+    results: Final[list[dict[str, object]]] = []
     for msg in reversed(messages):
         if not isinstance(msg, dict):
             break
@@ -154,7 +158,7 @@ def _assistant_content_and_tool_calls(response_obj: Any) -> tuple:
     raw_tool_calls = getattr(msg, "tool_calls", None)
     if raw_tool_calls is None and isinstance(msg, dict):
         raw_tool_calls = msg.get("tool_calls")
-    tool_calls: Final[list[dict[str, Any]]] = []
+    tool_calls: Final[list[dict[str, object]]] = []
     for tc in raw_tool_calls or []:
         if isinstance(tc, dict):
             tool_calls.append(tc)
@@ -174,11 +178,11 @@ class AdaptiveRouterPostCallHook(CustomLogger):
 
     async def async_post_call_response_headers_hook(
         self,
-        data: dict[str, Any],
-        user_api_key_dict: Any,
-        response: Any,
+        data: Mapping[str, object],
+        user_api_key_dict: UserAPIKeyAuth,
+        response: object,
         request_headers: dict[str, str] | None = None,
-        litellm_call_info: dict[str, Any] | None = None,
+        litellm_call_info: dict[str, object] | None = None,
     ) -> dict[str, str] | None:
         """
         Surface the chosen logical model as the `x-litellm-adaptive-router-model`
@@ -209,7 +213,7 @@ class AdaptiveRouterPostCallHook(CustomLogger):
     async def _record(
         self,
         kwargs: dict[str, Any],
-        response_obj: Any,
+        response_obj: object,
         response_status: int,
     ) -> None:
         try:

litellm/router_strategy/quality_router/quality_router.py

@@ -16,6 +16,7 @@ then cheapest `model_info.input_cost_per_token`).
 """
 
 import math
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, Final, Optional
 
 from litellm._logging import verbose_router_logger
@@ -98,7 +99,7 @@ class QualityRouter(CustomLogger):
             self._tier_to_models_cache = self._build_tier_index()
         return self._tier_to_models_cache
 
-    def _get_routing_preferences(self, deployment: Any) -> dict[str, Any] | None:
+    def _get_routing_preferences(self, deployment: object) -> dict[str, Any] | None:
         """
         Extract litellm_routing_preferences from a deployment, handling both
         dict-shaped and Pydantic-object-shaped deployments.
@@ -119,7 +120,7 @@ class QualityRouter(CustomLogger):
             return model_info.get("litellm_routing_preferences")
         return getattr(model_info, "litellm_routing_preferences", None)
 
-    def _get_deployment_input_cost(self, deployment: Any) -> float | None:
+    def _get_deployment_input_cost(self, deployment: object) -> float | None:
         """
         Extract `input_cost_per_token` from a deployment's model_info.
 
@@ -144,7 +145,7 @@ class QualityRouter(CustomLogger):
         except (TypeError, ValueError):
             return None
 
-    def _get_deployment_model_name(self, deployment: Any) -> str | None:
+    def _get_deployment_model_name(self, deployment: object) -> str | None:
         """Extract `model_name` from a dict- or object-shaped deployment."""
         if isinstance(deployment, dict):
             return deployment.get("model_name")
@@ -304,8 +305,8 @@ class QualityRouter(CustomLogger):
 
     def _stash_decision(
         self,
-        request_kwargs: dict[str, Any] | None,
-        decision: dict[str, Any],
+        request_kwargs: dict[str, object] | None,
+        decision: Mapping[str, object],
     ) -> None:
         """
         Stash the routing decision in request_kwargs.metadata so the Router can

litellm/router_utils/fallback_event_handlers.py

@@ -1,6 +1,6 @@
 import hashlib
 import json
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Final
@@ -39,7 +39,7 @@ _REQUEST_SCOPED_STATUS_CODES: Final = frozenset((404,))
 
 def _trigger_cooldown_for_failed_deployment(
     litellm_router: LitellmRouter,
-    kwargs: Mapping[str, Any],
+    kwargs: Mapping[str, object],
     exception: Exception,
 ) -> None:
     """
@@ -218,7 +218,7 @@ PRE_ROUTING_SELECTED_MODEL_KEY: Final = "pre_routing_selected_model"
 _ROUTER_METADATA_BUCKETS: Final = ("metadata", "litellm_metadata")
 
 
-def record_pre_routing_selection(request_kwargs: Mapping[str, Any] | None, selected_model: str) -> None:
+def record_pre_routing_selection(request_kwargs: Mapping[str, object] | None, selected_model: str) -> None:
     """
     Remember which model a pre-routing hook picked, so fallback lookup can key off it.
 
@@ -257,14 +257,14 @@ def clear_pre_routing_selection(request_kwargs: Mapping[str, object] | None) ->
             del bucket[PRE_ROUTING_SELECTED_MODEL_KEY]
 
 
-def get_pre_routing_selection(kwargs: Mapping[str, Any]) -> str | None:
+def get_pre_routing_selection(kwargs: Mapping[str, object]) -> str | None:
     """The model a pre-routing hook selected for this request, if one did."""
     buckets: Final = (kwargs.get(name) for name in _ROUTER_METADATA_BUCKETS)
     selections: Final = (bucket.get(PRE_ROUTING_SELECTED_MODEL_KEY) for bucket in buckets if isinstance(bucket, dict))
     return next((selected for selected in selections if isinstance(selected, str) and selected), None)
 
 
-def fallback_lookup_groups(kwargs: Mapping[str, Any], model_group: str | None) -> tuple[str, ...]:
+def fallback_lookup_groups(kwargs: Mapping[str, object], model_group: str | None) -> tuple[str, ...]:
     """
     Ordered keys for resolving a fallback chain: the tier a pre-routing hook selected wins,
     and the requested group still resolves when no tier-keyed chain exists, so configs keyed
@@ -413,7 +413,7 @@ def creates_provider_scoped_resource(kwargs: Mapping[str, object]) -> bool:
 
 
 async def run_async_fallback(
-    *args: tuple[Any],
+    *args: object,
     litellm_router: LitellmRouter,
     fallback_model_group: list[str],
     original_model_group: str,
@@ -630,5 +630,5 @@ def _check_non_standard_fallback_format(fallbacks: list[Any] | None) -> bool:
     return False
 
 
-def run_non_standard_fallback_format(fallbacks: list[str] | list[dict[str, Any]], model_group: str):
+def run_non_standard_fallback_format(fallbacks: Sequence[str] | Sequence[Mapping[str, object]], model_group: str):
     pass

litellm/router_utils/pre_call_checks/io_token_rate_limit_check.py

@@ -526,8 +526,8 @@ async def async_io_token_pre_call_check(
 
 def io_token_reconcile_success(
     dual_cache: DualCache,
-    kwargs: Any,
-    response_obj: Any,
+    kwargs: Mapping[str, object] | None,
+    response_obj: object,
 ) -> None:
     request_kwargs: Final[Mapping[str, object] | None] = kwargs
     response: Final[object] = response_obj
@@ -577,8 +577,8 @@ def io_token_reconcile_success(
 
 async def async_io_token_reconcile_success(
     dual_cache: DualCache,
-    kwargs: Any,
-    response_obj: Any,
+    kwargs: Mapping[str, object] | None,
+    response_obj: object,
     *,
     parent_otel_span: Span | None = None,
 ) -> None:
@@ -638,7 +638,7 @@ async def async_io_token_reconcile_success(
 
 def io_token_refund_failure(
     dual_cache: DualCache,
-    kwargs: Any,
+    kwargs: Mapping[str, object] | None,
 ) -> None:
     request_kwargs: Final[Mapping[str, object] | None] = kwargs
     itpm_reserved, otpm_reserved, itpm_key, otpm_key = _read_reservation_from_kwargs(request_kwargs)
@@ -689,7 +689,7 @@ def refund_stale_reservation_before_retry(dual_cache: DualCache, kwargs: Mapping
 
 async def async_io_token_refund_failure(
     dual_cache: DualCache,
-    kwargs: Any,
+    kwargs: Mapping[str, object] | None,
     *,
     parent_otel_span: Span | None = None,
 ) -> None:

litellm/router_utils/search_api_router.py

@@ -10,10 +10,19 @@ import traceback
 from collections.abc import Callable
 from functools import partial
 from types import MappingProxyType
-from typing import Any, Final
+from typing import TYPE_CHECKING, Any, Final, Protocol
 
 from litellm._logging import verbose_router_logger
 
+if TYPE_CHECKING:
+    from litellm.types.router import SearchToolTypedDict
+
+
+class _SearchToolsRouter(Protocol):
+    """The one router attribute the search-tool helpers read and replace."""
+
+    search_tools: "list[SearchToolTypedDict]"
+
 
 class SearchAPIRouter:
     """
@@ -45,7 +54,7 @@ class SearchAPIRouter:
         return resolved_api_key, resolved_api_base
 
     @staticmethod
-    async def update_router_search_tools(router_instance: Any, search_tools: list):
+    async def update_router_search_tools(router_instance: _SearchToolsRouter, search_tools: list):
         """
         Update the router with search tools from the database.
 
@@ -83,7 +92,7 @@ class SearchAPIRouter:
 
     @staticmethod
     def get_matching_search_tools(
-        router_instance: Any,
+        router_instance: _SearchToolsRouter,
         search_tool_name: str,
     ) -> list:
         """
@@ -175,7 +184,7 @@ class SearchAPIRouter:
 
     @staticmethod
     async def async_search_with_fallbacks_helper(
-        router_instance: Any,
+        router_instance: _SearchToolsRouter,
         model: str,
         original_generic_function: Callable,
         **kwargs,

litellm/secret_managers/aws_secret_manager_v2.py

@@ -266,7 +266,7 @@ class AWSSecretsManagerV2(BaseAWSLLM, BaseSecretManager):
         """
         from litellm._uuid import uuid
 
-        data: Final[dict[str, Any]] = {
+        data: Final[dict[str, object]] = {
             "Name": secret_name,
             "SecretString": secret_value,
             "ClientRequestToken": str(uuid.uuid4()),
@@ -415,7 +415,7 @@ class AWSSecretsManagerV2(BaseAWSLLM, BaseSecretManager):
         """
         from litellm._uuid import uuid
 
-        data: Final[dict[str, Any]] = {
+        data: Final[dict[str, object]] = {
             "SecretId": secret_name,
             "SecretString": secret_value,
             "ClientRequestToken": str(uuid.uuid4()),

litellm/skills/main.py

@@ -5,7 +5,7 @@ Provides create, list, get, and delete operations for skills
 
 import asyncio
 import contextvars
-from collections.abc import Coroutine
+from collections.abc import Coroutine, Mapping
 from functools import partial
 from typing import Any, Final
 
@@ -35,7 +35,7 @@ DEFAULT_ANTHROPIC_API_BASE: Final = "https://api.anthropic.com/v1"
 _litellm_skills_handler = None
 
 
-def _get_user_api_key_auth_from_kwargs(kwargs: dict[str, Any]) -> Any | None:
+def _get_user_api_key_auth_from_kwargs(kwargs: Mapping[str, object]) -> Any | None:
     for metadata_key in ("metadata", "litellm_metadata"):
         metadata = kwargs.get(metadata_key)
         if isinstance(metadata, dict) and metadata.get("user_api_key_auth") is not None:
@@ -44,7 +44,7 @@ def _get_user_api_key_auth_from_kwargs(kwargs: dict[str, Any]) -> Any | None:
 
 
 def _get_skill_request_metadata(
-    kwargs: dict[str, Any],
+    kwargs: Mapping[str, object],
     extra_body: dict[str, Any] | None,
 ) -> dict[str, Any] | None:
     if extra_body and isinstance(extra_body.get("metadata"), dict):
@@ -73,7 +73,7 @@ async def acreate_skill(
     files: list[Any] | None = None,
     display_title: str | None = None,
     extra_headers: dict[str, Any] | None = None,
-    extra_query: dict[str, Any] | None = None,
+    extra_query: Mapping[str, object] | None = None,
     extra_body: dict[str, Any] | None = None,
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
@@ -136,12 +136,12 @@ def create_skill(
     files: list[Any] | None = None,
     display_title: str | None = None,
     extra_headers: dict[str, Any] | None = None,
-    extra_query: dict[str, Any] | None = None,
+    extra_query: Mapping[str, object] | None = None,
     extra_body: dict[str, Any] | None = None,
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> Skill | Coroutine[Any, Any, Skill]:
+) -> Skill | Coroutine[object, object, Skill]:
     """
     Create a new skill
 
@@ -330,7 +330,7 @@ def list_skills(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> ListSkillsResponse | Coroutine[Any, Any, ListSkillsResponse]:
+) -> ListSkillsResponse | Coroutine[object, object, ListSkillsResponse]:
     """
     List all skills
 
@@ -444,7 +444,7 @@ def list_skills(
 async def aget_skill(
     skill_id: str,
     extra_headers: dict[str, Any] | None = None,
-    extra_query: dict[str, Any] | None = None,
+    extra_query: Mapping[str, object] | None = None,
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
@@ -501,11 +501,11 @@ async def aget_skill(
 def get_skill(
     skill_id: str,
     extra_headers: dict[str, Any] | None = None,
-    extra_query: dict[str, Any] | None = None,
+    extra_query: Mapping[str, object] | None = None,
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> Skill | Coroutine[Any, Any, Skill]:
+) -> Skill | Coroutine[object, object, Skill]:
     """
     Get a skill by ID
 
@@ -608,7 +608,7 @@ def get_skill(
 async def adelete_skill(
     skill_id: str,
     extra_headers: dict[str, Any] | None = None,
-    extra_query: dict[str, Any] | None = None,
+    extra_query: Mapping[str, object] | None = None,
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
@@ -665,11 +665,11 @@ async def adelete_skill(
 def delete_skill(
     skill_id: str,
     extra_headers: dict[str, Any] | None = None,
-    extra_query: dict[str, Any] | None = None,
+    extra_query: Mapping[str, object] | None = None,
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> DeleteSkillResponse | Coroutine[Any, Any, DeleteSkillResponse]:
+) -> DeleteSkillResponse | Coroutine[object, object, DeleteSkillResponse]:
     """
     Delete a skill by ID
 

litellm/types/vector_stores.py

@@ -1,3 +1,4 @@
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
@@ -128,31 +129,31 @@ class VertexSearchDataStoreExtraBody(TypedDict, total=False):
     pageToken: str
     offset: int
     oneBoxPageSize: int
-    pageCategories: list[str]
-    imageQuery: dict[str, Any]
+    pageCategories: Sequence[str]
+    imageQuery: Mapping[str, object]
     filter: str
     canonicalFilter: str
     orderBy: str
-    userInfo: dict[str, Any]
+    userInfo: Mapping[str, object]
     languageCode: str
-    facetSpecs: list[dict[str, Any]]
-    boostSpec: dict[str, Any]
-    params: dict[str, Any]
-    queryExpansionSpec: dict[str, Any]
-    spellCorrectionSpec: dict[str, Any]
+    facetSpecs: Sequence[Mapping[str, object]]
+    boostSpec: Mapping[str, object]
+    params: Mapping[str, object]
+    queryExpansionSpec: Mapping[str, object]
+    spellCorrectionSpec: Mapping[str, object]
     userPseudoId: str
-    contentSearchSpec: dict[str, Any]
+    contentSearchSpec: Mapping[str, object]
     rankingExpression: str
     rankingExpressionBackend: str
     safeSearch: bool
-    userLabels: dict[str, str]
-    naturalLanguageQueryUnderstandingSpec: dict[str, Any]
-    searchAsYouTypeSpec: dict[str, Any]
-    displaySpec: dict[str, Any]
-    crowdingSpecs: list[dict[str, Any]]
+    userLabels: Mapping[str, str]
+    naturalLanguageQueryUnderstandingSpec: Mapping[str, object]
+    searchAsYouTypeSpec: Mapping[str, object]
+    displaySpec: Mapping[str, object]
+    crowdingSpecs: Sequence[Mapping[str, object]]
     relevanceThreshold: str
-    relevanceScoreSpec: dict[str, Any]
-    customRankingParams: dict[str, Any]
+    relevanceScoreSpec: Mapping[str, object]
+    customRankingParams: Mapping[str, object]
 
 
 class VertexSearchEngineExtraBody(VertexSearchDataStoreExtraBody, total=False):
@@ -166,7 +167,7 @@ class VertexSearchEngineExtraBody(VertexSearchDataStoreExtraBody, total=False):
     (per-store scoping/filtering) and ``numResultsPerDataStore``.
     """
 
-    dataStoreSpecs: list[dict[str, Any]]
+    dataStoreSpecs: Sequence[Mapping[str, object]]
     numResultsPerDataStore: int
 
 
@@ -256,7 +257,7 @@ class IndexCreateLiteLLMParams(BaseModel):
 class IndexCreateRequest(BaseModel):
     index_name: str
     litellm_params: IndexCreateLiteLLMParams
-    index_info: dict[str, Any] | None = None
+    index_info: dict[str, object] | None = None
 
 
 class BaseVectorStoreAuthCredentials(TypedDict, total=False):
@@ -270,7 +271,7 @@ class LiteLLM_ManagedVectorStoreIndex(BaseModel):
     id: str
     index_name: str
     litellm_params: IndexCreateLiteLLMParams
-    index_info: dict[str, Any] | None = None
+    index_info: dict[str, object] | None = None
     created_at: datetime | None = None
     created_by: str | None = None
     updated_at: datetime | None = None

litellm/vector_store_files/main.py

@@ -39,7 +39,7 @@ def _ensure_provider(custom_llm_provider: str | None) -> str:
 def _prepare_registry_credentials(
     *,
     vector_store_id: str,
-    kwargs: dict[str, Any],
+    kwargs: dict[str, object],
 ) -> None:
     if litellm.vector_store_registry is None:
         return
@@ -116,7 +116,7 @@ def create(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> VectorStoreFileObject | Coroutine[Any, Any, VectorStoreFileObject]:
+) -> VectorStoreFileObject | Coroutine[object, object, VectorStoreFileObject]:
     local_vars: Final = locals()
     try:
         litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
@@ -245,7 +245,7 @@ def list(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> VectorStoreFileListResponse | Coroutine[Any, Any, VectorStoreFileListResponse]:
+) -> VectorStoreFileListResponse | Coroutine[object, object, VectorStoreFileListResponse]:
     local_vars: Final = locals()
     try:
         litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
@@ -355,7 +355,7 @@ def retrieve(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> VectorStoreFileObject | Coroutine[Any, Any, VectorStoreFileObject]:
+) -> VectorStoreFileObject | Coroutine[object, object, VectorStoreFileObject]:
     local_vars: Final = locals()
     try:
         litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
@@ -463,7 +463,7 @@ def retrieve_content(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> VectorStoreFileContentResponse | Coroutine[Any, Any, VectorStoreFileContentResponse]:
+) -> VectorStoreFileContentResponse | Coroutine[object, object, VectorStoreFileContentResponse]:
     local_vars: Final = locals()
     try:
         litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
@@ -577,7 +577,7 @@ def update(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> VectorStoreFileObject | Coroutine[Any, Any, VectorStoreFileObject]:
+) -> VectorStoreFileObject | Coroutine[object, object, VectorStoreFileObject]:
     local_vars: Final = locals()
     try:
         litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
@@ -692,7 +692,7 @@ def delete(
     timeout: float | httpx.Timeout | None = None,
     custom_llm_provider: str | None = None,
     **kwargs,
-) -> VectorStoreFileDeleteResponse | Coroutine[Any, Any, VectorStoreFileDeleteResponse]:
+) -> VectorStoreFileDeleteResponse | Coroutine[object, object, VectorStoreFileDeleteResponse]:
     local_vars: Final = locals()
     try:
         litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")