diff --git a/litellm/a2a_protocol/providers/pydantic_ai_agents/config.py b/litellm/a2a_protocol/providers/pydantic_ai_agents/config.py
index b7546e1a2a1..20404e3702b 100644
--- a/litellm/a2a_protocol/providers/pydantic_ai_agents/config.py
+++ b/litellm/a2a_protocol/providers/pydantic_ai_agents/config.py
@@ -23,7 +23,7 @@ class PydanticAIProviderConfig(BaseA2AProviderConfig):
         params: dict[str, Any],
         api_base: str | None = None,
         **kwargs: Any,
-    ) -> dict[str, Any]:
+    ) -> dict[str, object]:
         """Handle non-streaming request to Pydantic AI agent."""
         if api_base is None:
             raise ValueError("api_base is required for PydanticAIProviderConfig")
@@ -41,7 +41,7 @@ class PydanticAIProviderConfig(BaseA2AProviderConfig):
         params: dict[str, Any],
         api_base: str | None = None,
         **kwargs,
-    ) -> AsyncIterator[dict[str, Any]]:
+    ) -> AsyncIterator[dict[str, object]]:
         """Handle streaming request with fake streaming."""
         if not api_base:
             raise ValueError("api_base is required for Pydantic AI agents")
diff --git a/litellm/integrations/argilla.py b/litellm/integrations/argilla.py
index 9a87a94cf0b..664ef8efda1 100644
--- a/litellm/integrations/argilla.py
+++ b/litellm/integrations/argilla.py
@@ -7,7 +7,8 @@ import json
 import os
 import random
 import types
-from typing import Any, Final
+from collections.abc import Mapping
+from typing import Final
 
 import httpx
 from pydantic import BaseModel
@@ -69,7 +70,7 @@ class ArgillaLogger(CustomBatchLogger):
         self.flush_lock = asyncio.Lock()
         super().__init__(**kwargs, flush_lock=self.flush_lock)
 
-    def validate_argilla_transformation_object(self, argilla_transformation_object: dict[str, Any]):
+    def validate_argilla_transformation_object(self, argilla_transformation_object: Mapping[str, object]):
         if not isinstance(argilla_transformation_object, dict):
             raise Exception("'argilla_transformation_object' must be a dictionary, to log your payload to Argilla.")
 
@@ -115,7 +116,7 @@ class ArgillaLogger(CustomBatchLogger):
             ARGILLA_DATASET_NAME=_credentials_dataset_name,
         )
 
-    def get_chat_messages(self, payload: StandardLoggingPayload) -> list[dict[str, Any]]:
+    def get_chat_messages(self, payload: StandardLoggingPayload) -> list[dict[str, object]]:
         payload_messages: Final = payload.get("messages", None)
 
         if payload_messages is None:
diff --git a/litellm/integrations/dynamodb.py b/litellm/integrations/dynamodb.py
index 38f5924a233..3401ced4efb 100644
--- a/litellm/integrations/dynamodb.py
+++ b/litellm/integrations/dynamodb.py
@@ -3,12 +3,25 @@
 
 import os
 import traceback
-from typing import Any, Final
+from collections.abc import Mapping
+from typing import Final, Protocol
 
 import litellm
 from litellm._uuid import uuid
 
 
+class _DynamoTable(Protocol):
+    """The one boto3 DynamoDB table call this logger makes."""
+
+    def put_item(self, *, Item: Mapping[str, object]) -> object: ...
+
+
+class _DynamoResource(Protocol):
+    """The one boto3 DynamoDB resource call this logger makes."""
+
+    def Table(self, name: str) -> _DynamoTable: ...
+
+
 class DyanmoDBLogger:
     # Class variables or attributes
 
@@ -16,7 +29,7 @@ class DyanmoDBLogger:
         # Instance variables
         import boto3
 
-        self.dynamodb: Any = boto3.resource("dynamodb", region_name=os.environ["AWS_REGION_NAME"])
+        self.dynamodb: Final[_DynamoResource] = boto3.resource("dynamodb", region_name=os.environ["AWS_REGION_NAME"])
         if litellm.dynamodb_table_name is None:
             raise ValueError(
                 "LiteLLM Error, trying to use DynamoDB but not table name passed. Create a table and set `litellm.dynamodb_table_name=<your-table>`"
@@ -41,7 +54,7 @@ class DyanmoDBLogger:
             id: Final = response_obj.get("id", str(uuid.uuid4()))
 
             # Build the initial payload
-            payload: Final = {
+            payload: Final[dict[str, object]] = {
                 "id": id,
                 "call_type": call_type,
                 "startTime": start_time,
diff --git a/litellm/integrations/focus/database.py b/litellm/integrations/focus/database.py
index 657c7e0d264..891318f1c54 100644
--- a/litellm/integrations/focus/database.py
+++ b/litellm/integrations/focus/database.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Any, Final
+from typing import Final
 
 import polars as pl
 
@@ -32,7 +32,7 @@ class FocusLiteLLMDatabase:
         client: Final = self._ensure_prisma_client()
 
         where_clauses: Final[list[str]] = []
-        query_params: Final[list[Any]] = []
+        query_params: Final[list[datetime | int]] = []
         placeholder_index = 1
         if start_time_utc:
             where_clauses.append(f"dus.updated_at >= ${placeholder_index}::timestamptz")
@@ -112,7 +112,7 @@ class FocusLiteLLMDatabase:
         except Exception as exc:
             raise RuntimeError(f"Error retrieving usage data: {exc}") from exc
 
-    async def get_table_info(self) -> dict[str, Any]:
+    async def get_table_info(self) -> dict[str, object]:
         """Return metadata about the spend table for diagnostics."""
         client: Final = self._ensure_prisma_client()
 
diff --git a/litellm/integrations/focus/destinations/vantage_destination.py b/litellm/integrations/focus/destinations/vantage_destination.py
index 132f27779c2..68b0d399975 100644
--- a/litellm/integrations/focus/destinations/vantage_destination.py
+++ b/litellm/integrations/focus/destinations/vantage_destination.py
@@ -4,7 +4,8 @@ from __future__ import annotations
 
 import csv
 import io
-from typing import Any, Final
+from collections.abc import Mapping
+from typing import Final
 
 import httpx  # noqa: F401 - used at runtime (AsyncClient, HTTPStatusError)
 
@@ -94,7 +95,7 @@ class FocusVantageDestination(FocusDestination):
         self,
         *,
         prefix: str,
-        config: dict[str, Any] | None = None,
+        config: Mapping[str, object] | None = None,
     ) -> None:
         config = config or {}
         api_key: Final = config.get("api_key")
diff --git a/litellm/integrations/langfuse/langfuse_otel_attributes.py b/litellm/integrations/langfuse/langfuse_otel_attributes.py
index 70fea1abb3b..1eb3ce1a9c2 100644
--- a/litellm/integrations/langfuse/langfuse_otel_attributes.py
+++ b/litellm/integrations/langfuse/langfuse_otel_attributes.py
@@ -5,6 +5,7 @@ Relevant Issue: https://github.com/BerriAI/litellm/issues/13764
 """
 
 import json
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, Final
 
 from pydantic import BaseModel
@@ -40,7 +41,7 @@ def get_output_content_by_type(
     | HttpxBinaryResponseContent
     | ResponsesAPIResponse
     | list,
-    kwargs: dict[str, Any] | None = None,
+    kwargs: Mapping[str, object] | None = None,
 ) -> str:
     """
     Extract output content from response objects based on their type.
diff --git a/litellm/integrations/langsmith.py b/litellm/integrations/langsmith.py
index 9607eccef52..2c3837b12ec 100644
--- a/litellm/integrations/langsmith.py
+++ b/litellm/integrations/langsmith.py
@@ -75,9 +75,9 @@ class LangsmithLogger(CustomBatchLogger):
         if _batch_size:
             self.batch_size = int(_batch_size)
         self.log_queue: list[LangsmithQueueObject] = []
-        self._flush_task: asyncio.Task[Any] | None = self._start_periodic_flush_task()
+        self._flush_task: asyncio.Task[None] | None = self._start_periodic_flush_task()
 
-    def _start_periodic_flush_task(self) -> asyncio.Task[Any] | None:
+    def _start_periodic_flush_task(self) -> asyncio.Task[None] | None:
         """Start the periodic flush task only when an event loop is already running."""
         try:
             loop: Final = asyncio.get_running_loop()
@@ -152,9 +152,9 @@ class LangsmithLogger(CustomBatchLogger):
 
         return self._redact_metadata(extra_metadata)
 
-    def _build_outputs_with_usage(self, payload: StandardLoggingPayload) -> dict[str, Any]:
+    def _build_outputs_with_usage(self, payload: StandardLoggingPayload) -> dict[str, object]:
         response: Final = payload["response"]
-        outputs: dict[str, Any]
+        outputs: dict[str, object]
         if isinstance(response, dict):
             outputs = {**response}
         else:
diff --git a/litellm/integrations/otel/model/metadata.py b/litellm/integrations/otel/model/metadata.py
index ee116aca46b..9eab3021d58 100644
--- a/litellm/integrations/otel/model/metadata.py
+++ b/litellm/integrations/otel/model/metadata.py
@@ -36,7 +36,7 @@ model. They coincide on the SDK path, which is correct.
 
 from __future__ import annotations
 
-from collections.abc import Iterator, Mapping
+from collections.abc import Callable, Iterator, Mapping
 from dataclasses import dataclass, field
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, cast
@@ -57,7 +57,7 @@ class RequestIdentity:
     # The team's free-form metadata, carried raw (empty/missing -> None) and
     # filtered to an operator allowlist only at Baggage-promotion time, so an
     # unconfigured deployment never promotes any of it.
-    team_metadata: Mapping[str, Any] | None = None
+    team_metadata: Mapping[str, object] | None = None
     key_hash: str | None = None
     end_user: str | None = None
     # The model litellm dispatched to the provider. Only known once the call
@@ -103,7 +103,7 @@ class RequestIdentity:
         ``user_api_key_*`` names that ``baggage.DEFAULT_BAGGAGE_METADATA_KEYS``
         promotes.
         """
-        get: Final = lambda name: getattr(auth, name, None)  # noqa: E731
+        get: Final[Callable[[str], object]] = lambda name: getattr(auth, name, None)  # noqa: E731
         metadata: Final = {
             meta_key: str(value)
             for meta_key, attr in (
@@ -217,7 +217,7 @@ class LLMCallEvent:
     time_to_first_chunk_seconds: float | None
 
     @classmethod
-    def from_dict(cls, kwargs: Mapping[str, Any]) -> LLMCallEvent:
+    def from_dict(cls, kwargs: Mapping[str, object]) -> LLMCallEvent:
         raw_payload: Final = kwargs.get("standard_logging_object")
         payload: Final = cast("StandardLoggingPayload", raw_payload) if raw_payload else None
         operation: Final = resolve_operation(as_str(kwargs.get("call_type")))
@@ -239,7 +239,7 @@ def time_to_first_chunk_seconds(kwargs: Mapping[str, Any]) -> float | None:
     to the first streamed chunk (``completion_start_time``); ``None`` for
     non-streaming calls, where ``completion_start_time`` is backfilled with the
     end time and would not measure first-chunk latency."""
-    optional_params: Final = cast(Mapping[str, Any], kwargs.get("optional_params") or {})
+    optional_params: Final = cast(Mapping[str, object], kwargs.get("optional_params") or {})
     if not optional_params.get("stream"):
         return None
     api_call_start: Final = to_seconds(kwargs.get("api_call_start_time"))
@@ -307,7 +307,7 @@ def _metadata_dicts(
     )
 
 
-def _call_id(payload: StandardLoggingPayload | None, kwargs: Mapping[str, Any]) -> str | None:
+def _call_id(payload: StandardLoggingPayload | None, kwargs: Mapping[str, object]) -> str | None:
     """The call id from the payload (when closed) or the bare kwargs (at pre_call)."""
     if payload is not None:
         call_id: Final = as_str(payload.get("litellm_call_id")) or as_str(payload.get("id"))
@@ -351,7 +351,7 @@ def _model_info_id(model_info: object) -> str | None:
     return None
 
 
-def _team_metadata_dict(value: object) -> Mapping[str, Any] | None:
+def _team_metadata_dict(value: object) -> Mapping[str, object] | None:
     """The team's free-form metadata as a raw mapping, or ``None`` when missing
     or empty.
 
diff --git a/litellm/integrations/otel/mount.py b/litellm/integrations/otel/mount.py
index ac647c2c4f6..776d8722d14 100644
--- a/litellm/integrations/otel/mount.py
+++ b/litellm/integrations/otel/mount.py
@@ -12,11 +12,14 @@ when the feature gate is off.
 """
 
 import os
-from typing import Any, Final
+from typing import TYPE_CHECKING, Final, Protocol
 
 from litellm._logging import verbose_logger
 from litellm.integrations.otel.model.config import is_otel_v2_enabled
 
+if TYPE_CHECKING:
+    from fastapi import FastAPI
+
 # Routes excluded from server-span tracing by default: high-frequency pollers and
 # static UI/docs assets, none of which are LLM traffic. Entries are substring-matched
 # against the request path (unanchored, so they survive a ``server_root_path`` prefix
@@ -65,7 +68,17 @@ PASSTHROUGH_PREFIXES: Final = frozenset(
 )
 
 
-def _passthrough_span_name_hook(span: Any, scope: dict) -> None:
+class _RenameableSpan(Protocol):
+    """The span surface the passthrough naming hook drives."""
+
+    def is_recording(self) -> bool: ...
+
+    def update_name(self, name: str) -> None: ...
+
+    def set_attribute(self, key: str, value: str) -> None: ...
+
+
+def _passthrough_span_name_hook(span: "_RenameableSpan | None", scope: dict) -> None:
     """FastAPI ``server_request_hook``: give passthrough server spans a useful name.
 
     The instrumentation matches the route at span creation, so both the span name
@@ -88,7 +101,7 @@ def _passthrough_span_name_hook(span: Any, scope: dict) -> None:
         pass
 
 
-def instrument_fastapi_app(app: Any) -> None:
+def instrument_fastapi_app(app: "FastAPI") -> None:
     """Attach OTel server-span instrumentation to the proxy FastAPI app.
 
     Safe no-op when the V2 gate is off or ``opentelemetry-instrumentation-fastapi``
diff --git a/litellm/litellm_core_utils/coroutine_checker.py b/litellm/litellm_core_utils/coroutine_checker.py
index 52fc44ba8dc..a5d4f8b968f 100644
--- a/litellm/litellm_core_utils/coroutine_checker.py
+++ b/litellm/litellm_core_utils/coroutine_checker.py
@@ -16,7 +16,7 @@ class CoroutineChecker:
     """
 
     def __init__(self):
-        self._cache = WeakKeyDictionary()
+        self._cache: WeakKeyDictionary[object, bool] = WeakKeyDictionary()
         self._max_size = COROUTINE_CHECKER_MAX_SIZE_IN_MEMORY
 
     def is_async_callable(self, callback: Any) -> bool:
@@ -33,10 +33,10 @@ class CoroutineChecker:
             pass
 
         # Determine target - optimized path for common cases
-        target = callback
+        target: object = callback
         if not inspect.isfunction(target) and not inspect.ismethod(target):
             try:
-                call_attr: Final = getattr(target, "__call__", None)
+                call_attr: Final[object] = getattr(target, "__call__", None)
                 if call_attr is not None:
                     target = call_attr
             except Exception:
diff --git a/litellm/litellm_core_utils/exception_mapping_utils.py b/litellm/litellm_core_utils/exception_mapping_utils.py
index 82708d412c9..92f07057da5 100644
--- a/litellm/litellm_core_utils/exception_mapping_utils.py
+++ b/litellm/litellm_core_utils/exception_mapping_utils.py
@@ -1,7 +1,7 @@
 import json
 import re
 import traceback
-from typing import Any, Final, Protocol, cast
+from typing import Final, Protocol, cast
 
 import httpx
 
@@ -191,7 +191,7 @@ def _get_response_headers(original_exception: Exception) -> httpx.Headers | None
     _response_headers: httpx.Headers | None = None
     try:
         _response_headers = getattr(original_exception, "headers", None)
-        error_response: Final = getattr(original_exception, "response", None)
+        error_response: Final[object] = getattr(original_exception, "response", None)
         if not _response_headers and error_response:
             _response_headers = getattr(error_response, "headers", None)
         if not _response_headers:
@@ -203,7 +203,7 @@ def _get_response_headers(original_exception: Exception) -> httpx.Headers | None
 
 
 def extract_and_raise_litellm_exception(
-    response: Any | None,
+    response: object | None,
     error_str: str,
     model: str,
     custom_llm_provider: str,
diff --git a/litellm/litellm_core_utils/prompt_templates/huggingface_template_handler.py b/litellm/litellm_core_utils/prompt_templates/huggingface_template_handler.py
index 3878b36cd91..8f8228d6dfd 100644
--- a/litellm/litellm_core_utils/prompt_templates/huggingface_template_handler.py
+++ b/litellm/litellm_core_utils/prompt_templates/huggingface_template_handler.py
@@ -1,6 +1,8 @@
 import json
 from datetime import datetime
-from typing import Any, Final
+from typing import Any, Final, Literal
+
+from typing_extensions import NotRequired, ReadOnly, TypedDict
 
 from litellm.llms.custom_httpx.http_handler import (
     _get_httpx_client,
@@ -9,6 +11,20 @@ from litellm.llms.custom_httpx.http_handler import (
 from litellm.types.llms.custom_http import httpxSpecialProvider
 
 
+class _TokenizerConfigResult(TypedDict):
+    """Outcome of a tokenizer_config.json fetch, carrying the parsed document when the fetch succeeded."""
+
+    status: ReadOnly[Literal["success", "failure"]]
+    tokenizer: NotRequired[ReadOnly[object]]
+
+
+class _ChatTemplateFileResult(TypedDict):
+    """Outcome of a chat template file fetch, carrying the template body when the fetch succeeded."""
+
+    status: ReadOnly[Literal["success", "failure"]]
+    chat_template: NotRequired[ReadOnly[str]]
+
+
 def strftime_now(fmt: str) -> str:
     """
     Custom function for templates that need current date/time formatting (e.g., gpt-oss)
@@ -22,7 +38,7 @@ def strftime_now(fmt: str) -> str:
     return datetime.now().strftime(fmt)
 
 
-def _get_tokenizer_config(hf_model_name: str) -> dict[str, Any]:
+def _get_tokenizer_config(hf_model_name: str) -> _TokenizerConfigResult:
     """
     Fetch tokenizer_config.json from HuggingFace (sync)
 
@@ -45,7 +61,7 @@ def _get_tokenizer_config(hf_model_name: str) -> dict[str, Any]:
         return {"status": "failure"}
 
 
-async def _aget_tokenizer_config(hf_model_name: str) -> dict[str, Any]:
+async def _aget_tokenizer_config(hf_model_name: str) -> _TokenizerConfigResult:
     """
     Fetch tokenizer_config.json from HuggingFace (async)
 
@@ -70,7 +86,7 @@ async def _aget_tokenizer_config(hf_model_name: str) -> dict[str, Any]:
         return {"status": "failure"}
 
 
-def _get_chat_template_file(hf_model_name: str) -> dict[str, Any]:
+def _get_chat_template_file(hf_model_name: str) -> _ChatTemplateFileResult:
     """
     Fetch chat template from separate .jinja file (sync)
 
@@ -98,7 +114,7 @@ def _get_chat_template_file(hf_model_name: str) -> dict[str, Any]:
     return {"status": "failure"}
 
 
-async def _aget_chat_template_file(hf_model_name: str) -> dict[str, Any]:
+async def _aget_chat_template_file(hf_model_name: str) -> _ChatTemplateFileResult:
     """
     Fetch chat template from separate .jinja file (async)
 
diff --git a/litellm/llms/azure_ai/anthropic/count_tokens/handler.py b/litellm/llms/azure_ai/anthropic/count_tokens/handler.py
index 3cc90823af9..36d5a56db0d 100644
--- a/litellm/llms/azure_ai/anthropic/count_tokens/handler.py
+++ b/litellm/llms/azure_ai/anthropic/count_tokens/handler.py
@@ -33,7 +33,7 @@ class AzureAIAnthropicCountTokensHandler(AzureAIAnthropicCountTokensConfig):
         litellm_params: dict[str, Any] | None = None,
         timeout: float | httpx.Timeout | None = None,
         tools: list[dict[str, Any]] | None = None,
-        system: Any | None = None,
+        system: object = None,
     ) -> dict[str, Any]:
         """
         Handle a CountTokens request using httpx with Azure authentication.
diff --git a/litellm/llms/base_llm/videos/transformation.py b/litellm/llms/base_llm/videos/transformation.py
index f725b295d0f..88cdcd61dd6 100644
--- a/litellm/llms/base_llm/videos/transformation.py
+++ b/litellm/llms/base_llm/videos/transformation.py
@@ -180,7 +180,7 @@ class BaseVideoConfig(ABC):
         api_base: str,
         litellm_params: GenericLiteLLMParams,
         headers: dict,
-        extra_body: dict[str, Any] | None = None,
+        extra_body: dict[str, object] | None = None,
     ) -> tuple[str, dict]:
         """
         Transform the video remix request into a URL and data
@@ -207,7 +207,7 @@ class BaseVideoConfig(ABC):
         after: str | None = None,
         limit: int | None = None,
         order: str | None = None,
-        extra_query: dict[str, Any] | None = None,
+        extra_query: dict[str, object] | None = None,
     ) -> tuple[str, dict]:
         """
         Transform the video list request into a URL and params
@@ -342,8 +342,8 @@ class BaseVideoConfig(ABC):
         litellm_params: GenericLiteLLMParams,
         headers: dict,
         video_file: FileContent | None = None,
-        extra_body: dict[str, Any] | None = None,
-        prefetched_source_data: dict[str, Any] | None = None,
+        extra_body: dict[str, object] | None = None,
+        prefetched_source_data: dict[str, object] | None = None,
     ) -> tuple[str, Mapping[str, object], RequestFiles | None]:
         """
         Transform the video edit request into a URL plus either JSON data or
@@ -373,7 +373,7 @@ class BaseVideoConfig(ABC):
         api_base: str,
         litellm_params: GenericLiteLLMParams,
         headers: dict,
-        extra_body: dict[str, Any] | None = None,
+        extra_body: dict[str, object] | None = None,
     ) -> tuple[str, dict]:
         """
         Transform the video extension request into a URL and JSON data.
diff --git a/litellm/llms/bedrock/count_tokens/bedrock_token_counter.py b/litellm/llms/bedrock/count_tokens/bedrock_token_counter.py
index 9c5211ed072..2d02b152c61 100644
--- a/litellm/llms/bedrock/count_tokens/bedrock_token_counter.py
+++ b/litellm/llms/bedrock/count_tokens/bedrock_token_counter.py
@@ -2,6 +2,7 @@
 Bedrock Token Counter implementation using the CountTokens API.
 """
 
+from collections.abc import Mapping, Sequence
 from typing import Any, Final
 
 from litellm._logging import verbose_logger
@@ -26,12 +27,12 @@ class BedrockTokenCounter(BaseTokenCounter):
     async def count_tokens(
         self,
         model_to_use: str,
-        messages: list[dict[str, Any]] | None,
-        contents: list[dict[str, Any]] | None,
+        messages: Sequence[Mapping[str, object]] | None,
+        contents: Sequence[Mapping[str, object]] | None,
         deployment: dict[str, Any] | None = None,
         request_model: str = "",
-        tools: list[dict[str, Any]] | None = None,
-        system: Any | None = None,
+        tools: Sequence[Mapping[str, object]] | None = None,
+        system: object | None = None,
     ) -> TokenCountResponse | None:
         """
         Count tokens using AWS Bedrock's CountTokens API.
@@ -56,7 +57,7 @@ class BedrockTokenCounter(BaseTokenCounter):
         litellm_params: Final = deployment.get("litellm_params", {})
 
         # Build request data in the format expected by BedrockCountTokensHandler
-        request_data: Final[dict[str, Any]] = {
+        request_data: Final[dict[str, object]] = {
             "model": model_to_use,
             "messages": messages,
         }
diff --git a/litellm/llms/bedrock/files/transformation.py b/litellm/llms/bedrock/files/transformation.py
index 33b27943ad8..f50d63c9f6f 100644
--- a/litellm/llms/bedrock/files/transformation.py
+++ b/litellm/llms/bedrock/files/transformation.py
@@ -247,7 +247,7 @@ def _validate_file_id_against_configured_buckets(
     return validate_against(configured_bucket_names[-1])
 
 
-def _uploaded_object_size(litellm_params: Mapping[str, object], raw_response: Response) -> int:
+def _uploaded_object_size(litellm_params: Mapping[str, object], response_headers: Mapping[str, str]) -> int:
     """
     S3 answers PutObject with an empty body, so the stored object size comes from the
     signed request recorded by `transform_create_file_request`, not the response headers.
@@ -255,7 +255,7 @@ def _uploaded_object_size(litellm_params: Mapping[str, object], raw_response: Re
     uploaded_size: Final = litellm_params.get(UPLOAD_CONTENT_LENGTH_PARAM)
     if isinstance(uploaded_size, int):
         return uploaded_size
-    response_content_length: Final = raw_response.headers.get("Content-Length", "0")
+    response_content_length: Final = response_headers.get("Content-Length", "0")
     return int(response_content_length) if response_content_length.isdigit() else 0
 
 
@@ -1161,7 +1161,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
             filename=filename,
             created_at=int(time.time()),  # Current timestamp
             status="uploaded",
-            bytes=_uploaded_object_size(litellm_params=litellm_params, raw_response=raw_response),
+            bytes=_uploaded_object_size(litellm_params=litellm_params, response_headers=raw_response.headers),
             object="file",
         )
 
diff --git a/litellm/llms/bedrock/passthrough/guardrail_translation/handler.py b/litellm/llms/bedrock/passthrough/guardrail_translation/handler.py
index b87f6196e51..eac8afd767c 100644
--- a/litellm/llms/bedrock/passthrough/guardrail_translation/handler.py
+++ b/litellm/llms/bedrock/passthrough/guardrail_translation/handler.py
@@ -14,6 +14,9 @@ from litellm.types.utils import GenericGuardrailAPIInputs
 if TYPE_CHECKING:
     from litellm.integrations.custom_guardrail import CustomGuardrail
     from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
+    from litellm.llms.pass_through.guardrail_translation.handler import (
+        PassThroughEndpointHandler,
+    )
     from litellm.proxy._types import UserAPIKeyAuth
     from litellm.proxy.utils import ProxyLogging
 
@@ -27,7 +30,7 @@ def _is_converse_endpoint(endpoint: str) -> bool:
     return bool(parts) and parts[-1] in _CONVERSE_ACTIONS
 
 
-def _generic_passthrough_handler() -> BaseTranslation:
+def _generic_passthrough_handler() -> "PassThroughEndpointHandler":
     """
     Fallback for non-Converse Bedrock routes (e.g. invoke). The generic
     handler scans the full request/response payload so blocking guardrails
diff --git a/litellm/llms/chatgpt/common_utils.py b/litellm/llms/chatgpt/common_utils.py
index 35e32e4172f..fe33219f110 100644
--- a/litellm/llms/chatgpt/common_utils.py
+++ b/litellm/llms/chatgpt/common_utils.py
@@ -268,7 +268,7 @@ def _normalize_litellm_params(litellm_params: Any | None) -> dict:
     return {}
 
 
-def get_chatgpt_session_id(litellm_params: Any | None) -> str | None:
+def get_chatgpt_session_id(litellm_params: object) -> str | None:
     params: Final = _normalize_litellm_params(litellm_params)
     for key in ("litellm_session_id", "session_id"):
         value = params.get(key)
@@ -286,5 +286,5 @@ def get_chatgpt_session_id(litellm_params: Any | None) -> str | None:
     return None
 
 
-def ensure_chatgpt_session_id(litellm_params: Any | None) -> str:
+def ensure_chatgpt_session_id(litellm_params: object) -> str:
     return get_chatgpt_session_id(litellm_params) or str(uuid4())
diff --git a/litellm/llms/github_copilot/responses/transformation.py b/litellm/llms/github_copilot/responses/transformation.py
index 5a4bb798851..8b85b668cba 100644
--- a/litellm/llms/github_copilot/responses/transformation.py
+++ b/litellm/llms/github_copilot/responses/transformation.py
@@ -19,6 +19,7 @@ from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfi
 from litellm.types.llms.openai import (
     ResponseInputParam,
     ResponsesAPIOptionalRequestParams,
+    ResponsesAPIStreamingResponse,
 )
 from litellm.types.router import GenericLiteLLMParams
 from litellm.types.utils import LlmProviders
@@ -129,7 +130,7 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
         model: str,
         parsed_chunk: dict,
         logging_obj: LiteLLMLoggingObj,
-    ) -> Any:
+    ) -> ResponsesAPIStreamingResponse:
         parsed_chunk = self._normalize_stream_item_id(parsed_chunk)
         return super().transform_streaming_response(
             model=model,
@@ -262,7 +263,7 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
         # Return the responses endpoint
         return f"{effective_api_base}/responses"
 
-    def _handle_reasoning_item(self, item: dict[str, Any]) -> dict[str, Any]:
+    def _handle_reasoning_item(self, item: dict[str, object]) -> dict[str, object]:
         """
         Handle reasoning items for GitHub Copilot, preserving encrypted_content.
 
@@ -280,7 +281,7 @@ class GithubCopilotResponsesAPIConfig(OpenAIResponsesAPIConfig):
 
             # Filter out None values for known problematic fields,
             # but preserve encrypted_content even if it exists
-            filtered_item: Final[dict[str, Any]] = {}
+            filtered_item: Final[dict[str, object]] = {}
             for k, v in item.items():
                 # Always include encrypted_content if present (even if None)
                 if k == "encrypted_content":
diff --git a/litellm/llms/hosted_vllm/chat/transformation.py b/litellm/llms/hosted_vllm/chat/transformation.py
index 29dc485732f..92bc857e385 100644
--- a/litellm/llms/hosted_vllm/chat/transformation.py
+++ b/litellm/llms/hosted_vllm/chat/transformation.py
@@ -28,12 +28,12 @@ from ...openai.chat.gpt_transformation import OpenAIGPTConfig
 
 
 class HostedVLLMChatConfig(OpenAIGPTConfig):
-    def _convert_custom_tools_to_function_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    def _convert_custom_tools_to_function_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, object]]:
         """
         vLLM chat completions currently accepts only OpenAI function tools.
         Convert custom tools into function tools so request validation does not fail.
         """
-        converted_tools: Final[list[dict[str, Any]]] = []
+        converted_tools: Final[list[dict[str, object]]] = []
         for idx, tool in enumerate(tools):
             if not isinstance(tool, dict):
                 converted_tools.append(tool)
@@ -63,17 +63,14 @@ class HostedVLLMChatConfig(OpenAIGPTConfig):
                     "required": ["input"],
                 }
 
-            function_tool: dict[str, Any] = {
-                "type": "function",
-                "function": {
-                    "name": str(tool_name),
-                    "parameters": tool_parameters,
-                },
+            function_definition: dict[str, object] = {
+                "name": str(tool_name),
+                "parameters": tool_parameters,
             }
             if isinstance(tool_description, str):
-                function_tool["function"]["description"] = tool_description
+                function_definition["description"] = tool_description
 
-            converted_tools.append(function_tool)
+            converted_tools.append({"type": "function", "function": function_definition})
 
         return converted_tools
 
@@ -148,7 +145,7 @@ class HostedVLLMChatConfig(OpenAIGPTConfig):
     @overload
     def _transform_messages(
         self, messages: list[AllMessageValues], model: str, is_async: Literal[True]
-    ) -> Coroutine[Any, Any, list[AllMessageValues]]: ...
+    ) -> Coroutine[object, object, list[AllMessageValues]]: ...
 
     @overload
     def _transform_messages(
@@ -160,7 +157,7 @@ class HostedVLLMChatConfig(OpenAIGPTConfig):
 
     def _transform_messages(
         self, messages: list[AllMessageValues], model: str, is_async: bool = False
-    ) -> list[AllMessageValues] | Coroutine[Any, Any, list[AllMessageValues]]:
+    ) -> list[AllMessageValues] | Coroutine[object, object, list[AllMessageValues]]:
         """
         Support translating:
         - video files from file_id or file_data to video_url
diff --git a/litellm/llms/openai/common_utils.py b/litellm/llms/openai/common_utils.py
index 2db6d78a218..e60b6145b44 100644
--- a/litellm/llms/openai/common_utils.py
+++ b/litellm/llms/openai/common_utils.py
@@ -10,7 +10,7 @@ import ssl
 import time
 import uuid
 from collections.abc import AsyncIterator, Iterator, Mapping
-from typing import TYPE_CHECKING, Any, Final, Literal, NamedTuple, Optional
+from typing import TYPE_CHECKING, Final, Literal, NamedTuple, Optional
 from urllib.parse import urlsplit
 
 import httpx
@@ -87,8 +87,8 @@ class OpenAIError(BaseLLMException):
 ###################################################################
 def drop_params_from_unprocessable_entity_error(
     e: openai.UnprocessableEntityError | httpx.HTTPStatusError,
-    data: dict[str, Any],
-) -> dict[str, Any]:
+    data: Mapping[str, object],
+) -> dict[str, object]:
     """
     Helper function to read OpenAI UnprocessableEntityError and drop the params that raised an error from the error message.
 
diff --git a/litellm/llms/snowflake/chat/transformation.py b/litellm/llms/snowflake/chat/transformation.py
index f65b0876202..aeff902f655 100644
--- a/litellm/llms/snowflake/chat/transformation.py
+++ b/litellm/llms/snowflake/chat/transformation.py
@@ -315,7 +315,7 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
         for msg in messages:
             if isinstance(msg, dict):
                 role = msg.get("role", "")
-                content: Any = msg.get("content", "")
+                content: object = msg.get("content", "")
                 msg_cache_control: object = msg.get("cache_control")
             else:
                 role = getattr(msg, "role", "")
@@ -463,7 +463,7 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
 
         return body
 
-    def _transform_tool_choice_to_anthropic(self, tool_choice: Any) -> dict[str, Any]:
+    def _transform_tool_choice_to_anthropic(self, tool_choice: object) -> Mapping[str, object]:
         """
         Convert tool_choice from OpenAI format to Anthropic format.
 
diff --git a/litellm/llms/stability/image_edit/transformations.py b/litellm/llms/stability/image_edit/transformations.py
index 0b6052ad593..94711d21b50 100644
--- a/litellm/llms/stability/image_edit/transformations.py
+++ b/litellm/llms/stability/image_edit/transformations.py
@@ -74,7 +74,7 @@ class StabilityImageEditConfig(BaseImageEditConfig):
         }
 
         # Create a copy to not mutate original - convert TypedDict to regular dict
-        mapped_params: Final[dict[str, Any]] = dict(image_edit_optional_params)
+        mapped_params: Final[dict[str, object]] = dict(image_edit_optional_params)
 
         for k, v in image_edit_optional_params.items():
             if k in param_mapping:
@@ -182,7 +182,7 @@ class StabilityImageEditConfig(BaseImageEditConfig):
         # Build Stability request
         # Populate multipart form-data as separate text fields (data) and files.
         # Stability expects prompt/output_format/etc. as normal form fields, not file parts.
-        data: Final[dict[str, Any]] = {
+        data: Final[dict[str, object]] = {
             "output_format": "png",  # Default to PNG
         }
 
diff --git a/litellm/llms/triton/completion/transformation.py b/litellm/llms/triton/completion/transformation.py
index 98a68ba2c36..3c868b3a96f 100644
--- a/litellm/llms/triton/completion/transformation.py
+++ b/litellm/llms/triton/completion/transformation.py
@@ -4,7 +4,7 @@ Translates from OpenAI's `/v1/chat/completions` endpoint to Triton's `/generate`
 
 import json
 from collections.abc import AsyncIterator, Iterator
-from typing import TYPE_CHECKING, Any, Final, Literal
+from typing import TYPE_CHECKING, Final, Literal
 
 from httpx import Headers, Response
 
@@ -172,7 +172,7 @@ class TritonConfig(BaseConfig):
         streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
         sync_stream: bool,
         json_mode: bool | None = False,
-    ) -> Any:
+    ) -> "TritonResponseIterator":
         return TritonResponseIterator(
             streaming_response=streaming_response,
             sync_stream=sync_stream,
@@ -195,14 +195,14 @@ class TritonGenerateConfig(TritonConfig):
     ) -> dict:
         inference_params: Final = optional_params.copy()
         stream: Final = inference_params.pop("stream", False)
-        data_for_triton: Final[dict[str, Any]] = {
+        data_for_triton: Final[dict[str, object]] = {
             "text_input": prompt_factory(model=model, messages=messages),
             "parameters": {
                 "max_tokens": int(optional_params.get("max_tokens", DEFAULT_MAX_TOKENS_FOR_TRITON)),
+                **inference_params,
             },
             "stream": bool(stream),
         }
-        data_for_triton["parameters"].update(inference_params)
         return data_for_triton
 
     def transform_response(
diff --git a/litellm/llms/vertex_ai/rerank/transformation.py b/litellm/llms/vertex_ai/rerank/transformation.py
index 2ec4f2da79b..2b1ed3a29a8 100644
--- a/litellm/llms/vertex_ai/rerank/transformation.py
+++ b/litellm/llms/vertex_ai/rerank/transformation.py
@@ -5,7 +5,7 @@ Why separate file? Make it easy to see how transformation works
 """
 
 from collections.abc import Mapping
-from typing import Any, Final
+from typing import Final
 
 import httpx
 
@@ -227,7 +227,7 @@ class VertexAIRerankConfig(BaseRerankConfig, VertexBase):
         model: str,
         drop_params: bool,
         query: str,
-        documents: list[str | dict[str, Any]],
+        documents: list[str | dict[str, object]],
         custom_llm_provider: str | None = None,
         top_n: int | None = None,
         rank_fields: list[str] | None = None,
diff --git a/litellm/llms/volcengine/embedding/transformation.py b/litellm/llms/volcengine/embedding/transformation.py
index 091b9dfd334..7c626c66e9d 100644
--- a/litellm/llms/volcengine/embedding/transformation.py
+++ b/litellm/llms/volcengine/embedding/transformation.py
@@ -3,7 +3,8 @@ Volcengine Embedding Transformation
 Transforms OpenAI embedding requests to Volcengine format
 """
 
-from typing import Any, Final
+from collections.abc import Mapping
+from typing import Final
 
 import httpx
 
@@ -83,11 +84,11 @@ class VolcEngineEmbeddingConfig(BaseEmbeddingConfig):
 
     def map_openai_params(
         self,
-        non_default_params: dict[str, Any],
-        optional_params: dict[str, Any],
+        non_default_params: Mapping[str, object],
+        optional_params: dict[str, object],
         model: str,
         drop_params: bool,
-    ) -> dict[str, Any]:
+    ) -> dict[str, object]:
         """
         Map OpenAI embedding parameters to Volcengine format.
 
diff --git a/litellm/llms/watsonx/rerank/transformation.py b/litellm/llms/watsonx/rerank/transformation.py
index 293880b188d..0c81d50a5fe 100644
--- a/litellm/llms/watsonx/rerank/transformation.py
+++ b/litellm/llms/watsonx/rerank/transformation.py
@@ -5,8 +5,8 @@ Docs - https://cloud.ibm.com/apidocs/watsonx-ai#text-rerank
 """
 
 import uuid
-from collections.abc import Mapping
-from typing import Any, Final, cast
+from collections.abc import Mapping, Sequence
+from typing import Final, cast
 
 import httpx
 
@@ -96,7 +96,7 @@ class IBMWatsonXRerankConfig(IBMWatsonXMixin, BaseRerankConfig):
         model: str,
         drop_params: bool,
         query: str,
-        documents: list[str | dict[str, Any]],
+        documents: Sequence[str | Mapping[str, object]],
         custom_llm_provider: str | None = None,
         top_n: int | None = None,
         rank_fields: list[str] | None = None,
@@ -178,7 +178,7 @@ class IBMWatsonXRerankConfig(IBMWatsonXMixin, BaseRerankConfig):
         transformed_results: Final = []
 
         for result in _results:
-            transformed_result: dict[str, Any] = {
+            transformed_result: dict[str, object] = {
                 "index": result["index"],
                 "relevance_score": result["score"],
             }
diff --git a/litellm/llms/xai/realtime/transformation.py b/litellm/llms/xai/realtime/transformation.py
index e9d16daad7c..5efe125ee60 100644
--- a/litellm/llms/xai/realtime/transformation.py
+++ b/litellm/llms/xai/realtime/transformation.py
@@ -16,7 +16,7 @@ construction time (see ``handler.py``) so all normalization is isolated here
 and ``RealTimeStreaming`` stays provider-agnostic.
 """
 
-from typing import Any, Final
+from typing import Final
 
 
 class XAIRealtimeNormalizer:
@@ -58,7 +58,7 @@ class XAIRealtimeNormalizer:
         # Cache content-part objects keyed by (response_id, item_id, content_index)
         # so that ``response.content_part.done`` events missing ``part`` can be
         # back-filled from earlier ``content_part.added`` / delta-done events.
-        self._content_part_by_key: dict[tuple, dict[str, Any]] = {}
+        self._content_part_by_key: dict[tuple, dict[str, object]] = {}
 
     # ---------------------------------------------------------------------------
     # Public interface consumed by RealTimeStreaming
@@ -140,7 +140,7 @@ class XAIRealtimeNormalizer:
             }
         self._content_part_by_key[key] = updated
 
-    def _resolve_content_part(self, event: dict) -> dict[str, Any]:
+    def _resolve_content_part(self, event: dict) -> dict[str, object]:
         part: Final = event.get("part")
         if isinstance(part, dict):
             return part
@@ -214,7 +214,7 @@ class XAIRealtimeNormalizer:
         needs_content: Final = event_type in self._EVENTS_NEEDING_CONTENT_INDEX
         if not needs_output and not needs_content:
             return event
-        patch: Final[dict[str, Any]] = {}
+        patch: Final[dict[str, object]] = {}
         if needs_output and "output_index" not in event:
             patch["output_index"] = 0
         if needs_content and "content_index" not in event:
@@ -228,8 +228,8 @@ class XAIRealtimeNormalizer:
     # ---------------------------------------------------------------------------
 
     @staticmethod
-    def _default_ga_usage() -> dict[str, Any]:
-        default_details: Final[dict[str, Any]] = {
+    def _default_ga_usage() -> dict[str, object]:
+        default_details: Final[dict[str, int]] = {
             "cached_tokens": 0,
             "text_tokens": 0,
             "audio_tokens": 0,
@@ -243,7 +243,7 @@ class XAIRealtimeNormalizer:
         }
 
     @staticmethod
-    def _normalize_usage(usage: object, *, empty_as_null: bool) -> dict[str, Any] | None:
+    def _normalize_usage(usage: object, *, empty_as_null: bool) -> dict[str, object] | None:
         """Coerce a usage object into the full OpenAI GA shape.
 
         ``empty_as_null=True`` for ``response.created`` (usage optional).
@@ -253,12 +253,12 @@ class XAIRealtimeNormalizer:
             return None
         if not usage:
             return None if empty_as_null else XAIRealtimeNormalizer._default_ga_usage()
-        default_details: Final[dict[str, Any]] = {
+        default_details: Final[dict[str, int]] = {
             "cached_tokens": 0,
             "text_tokens": 0,
             "audio_tokens": 0,
         }
-        normalized: Final[dict[str, Any]] = {
+        normalized: Final[dict[str, object]] = {
             "total_tokens": usage.get("total_tokens", 0),
             "input_tokens": usage.get("input_tokens", 0),
             "output_tokens": usage.get("output_tokens", 0),
diff --git a/litellm/proxy/analytics_endpoints/cache_activity.py b/litellm/proxy/analytics_endpoints/cache_activity.py
index b87b8eac3ef..952ef5bdcfc 100644
--- a/litellm/proxy/analytics_endpoints/cache_activity.py
+++ b/litellm/proxy/analytics_endpoints/cache_activity.py
@@ -2,16 +2,26 @@ import asyncio
 import json
 from collections.abc import Sequence
 from datetime import datetime
-from typing import TYPE_CHECKING, Final
+from typing import Final, Protocol
 
 from pydantic import BaseModel, TypeAdapter
 
-if TYPE_CHECKING:
-    from litellm.proxy.utils import PrismaClient
-
 UNKNOWN_CALL_TYPE: Final = "Unknown"
 
 
+class _SupportsQueryRaw(Protocol):
+    """The single database operation the cache-activity queries issue."""
+
+    async def query_raw(self, query: str, *args: object) -> Sequence[object]: ...
+
+
+class _SupportsRawQueryDb(Protocol):
+    """A prisma client handle, narrowed to the raw-query surface used here."""
+
+    @property
+    def db(self) -> _SupportsQueryRaw: ...
+
+
 class CacheActivityGroup(BaseModel):
     call_type: str
     api_requests: int
@@ -143,7 +153,7 @@ def compute_totals(groups: Sequence[CacheActivityGroup]) -> CacheActivityTotals:
 
 
 async def get_cache_activity(
-    prisma_client: "PrismaClient",
+    prisma_client: _SupportsRawQueryDb,
     start_date: datetime,
     end_date: datetime,
     key_aliases: Sequence[str],
diff --git a/litellm/proxy/client/cli/commands/model_groups.py b/litellm/proxy/client/cli/commands/model_groups.py
index c904e5bed49..367c2063b6b 100644
--- a/litellm/proxy/client/cli/commands/model_groups.py
+++ b/litellm/proxy/client/cli/commands/model_groups.py
@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from typing import Final, Literal
 
 import click
@@ -5,10 +6,17 @@ import rich
 import rich.table
 
 from ... import Client
+from ._cli_context import cli_context_values
 
 
 def create_client(ctx: click.Context) -> Client:
-    return Client(base_url=ctx.obj["base_url"], api_key=ctx.obj["api_key"])
+    context: Final = cli_context_values(ctx)
+    return Client(base_url=context["base_url"], api_key=context["api_key"])
+
+
+def _rendered_field(group: Mapping[str, object], key: str, default: str) -> str:
+    """The rendered value of one model group field, or ``default`` when the group omits it."""
+    return str(group.get(key, default))
 
 
 @click.group(name="model-groups")
@@ -46,10 +54,10 @@ def list_model_groups(ctx: click.Context, output_format: Literal["table", "json"
 
     for group in groups:
         table.add_row(
-            str(group.get("model_group", "")),
-            str(group.get("mode", "chat")),
-            str(group.get("input_cost_per_token", "")),
-            str(group.get("output_cost_per_token", "")),
+            _rendered_field(group, "model_group", ""),
+            _rendered_field(group, "mode", "chat"),
+            _rendered_field(group, "input_cost_per_token", ""),
+            _rendered_field(group, "output_cost_per_token", ""),
         )
     rich.print(table)
 
diff --git a/litellm/proxy/client/cli/commands/up.py b/litellm/proxy/client/cli/commands/up.py
index b7c02866d6f..45d21b0c0b5 100644
--- a/litellm/proxy/client/cli/commands/up.py
+++ b/litellm/proxy/client/cli/commands/up.py
@@ -166,7 +166,8 @@ def up(ctx: click.Context) -> None:
     is already running (this does not start one for you). Cursor is not
     supported: it has no equivalent file-based config to patch.
     """
-    base_url: Final = ctx.obj["base_url"]
+    ctx_obj: Final[CliContextObj] = ctx.obj
+    base_url: Final = ctx_obj["base_url"]
 
     try:
         _ensure_fresh_login(ctx)
diff --git a/litellm/proxy/common_utils/cache_pydantic_utils.py b/litellm/proxy/common_utils/cache_pydantic_utils.py
index 725c2b61145..3703cf7c916 100644
--- a/litellm/proxy/common_utils/cache_pydantic_utils.py
+++ b/litellm/proxy/common_utils/cache_pydantic_utils.py
@@ -37,7 +37,7 @@ class CacheCodec:
     """
 
     @staticmethod
-    def serialize(value: Any, model_type: type[T] | None = None) -> Any:
+    def serialize(value: object, model_type: type[T] | None = None) -> object:
         """
         Encode a value for DualCache / Redis (``json.dumps``-safe).
 
diff --git a/litellm/proxy/common_utils/proxy_rate_limit_error.py b/litellm/proxy/common_utils/proxy_rate_limit_error.py
index c109da6f571..8d3a587a4cd 100644
--- a/litellm/proxy/common_utils/proxy_rate_limit_error.py
+++ b/litellm/proxy/common_utils/proxy_rate_limit_error.py
@@ -66,7 +66,7 @@ def map_v3_rate_limit_type(
     return None
 
 
-def _coerce_message(detail: Any) -> str:
+def _coerce_message(detail: object) -> str:
     """Best-effort, JSON-friendly stringification of an HTTPException-style detail."""
     if detail is None:
         return ""
@@ -144,7 +144,7 @@ class ProxyRateLimitError(HTTPException, RateLimitError):
     def __init__(
         self,
         detail: Any,
-        headers: Mapping[str, Any] | None = None,
+        headers: Mapping[str, object] | None = None,
         category: str | RateLimitErrorCategory = RateLimitErrorCategory.LITELLM_RATE_LIMIT,
         rate_limit_type: str | RateLimitType | None = None,
         model: str | None = None,
diff --git a/litellm/proxy/container_endpoints/endpoints.py b/litellm/proxy/container_endpoints/endpoints.py
index e852eb5d6f9..c1407979f29 100644
--- a/litellm/proxy/container_endpoints/endpoints.py
+++ b/litellm/proxy/container_endpoints/endpoints.py
@@ -106,7 +106,7 @@ async def create_container(
     # Process request using ProxyBaseLLMRequestProcessing
     processor: Final = ProxyBaseLLMRequestProcessing(data=data)
     try:
-        response: Final = await processor.base_process_llm_request(
+        response: Final[object] = await processor.base_process_llm_request(
             request=request,
             fastapi_response=fastapi_response,
             user_api_key_dict=user_api_key_dict,
@@ -216,7 +216,7 @@ async def list_containers(
         or get_custom_llm_provider_from_request_query(request=request)
         or "openai"
     )
-    data: Final[dict[str, Any]] = {
+    data: Final[dict[str, object]] = {
         "query_params": query_params,
         "model": query_params.get("model"),
         "order": order,
@@ -341,7 +341,7 @@ async def retrieve_container(
     # Process request using ProxyBaseLLMRequestProcessing
     processor: Final = ProxyBaseLLMRequestProcessing(data=data)
     try:
-        return await processor.base_process_llm_request(
+        container: Final[object] = await processor.base_process_llm_request(
             request=request,
             fastapi_response=fastapi_response,
             user_api_key_dict=user_api_key_dict,
@@ -366,6 +366,7 @@ async def retrieve_container(
             proxy_logging_obj=proxy_logging_obj,
             version=version,
         )
+    return container
 
 
 @router.delete(
@@ -446,7 +447,7 @@ async def delete_container(
     # Process request using ProxyBaseLLMRequestProcessing
     processor: Final = ProxyBaseLLMRequestProcessing(data=data)
     try:
-        return await processor.base_process_llm_request(
+        deleted_container: Final[object] = await processor.base_process_llm_request(
             request=request,
             fastapi_response=fastapi_response,
             user_api_key_dict=user_api_key_dict,
@@ -471,6 +472,7 @@ async def delete_container(
             proxy_logging_obj=proxy_logging_obj,
             version=version,
         )
+    return deleted_container
 
 
 # Register JSON-configured container file endpoints
diff --git a/litellm/proxy/db/exception_handler.py b/litellm/proxy/db/exception_handler.py
index f469587ab8e..de20416ad9c 100644
--- a/litellm/proxy/db/exception_handler.py
+++ b/litellm/proxy/db/exception_handler.py
@@ -1,5 +1,5 @@
 from collections.abc import Awaitable, Callable, Iterator
-from typing import Any, Final, TypeVar
+from typing import Final, Protocol, TypeVar
 
 from litellm._logging import verbose_proxy_logger
 from litellm.proxy._types import (
@@ -407,8 +407,20 @@ def _coerce_timeout(value: object, fallback: float) -> float:
 _ReadResultT: Final = TypeVar("_ReadResultT")
 
 
+class _DBReconnectClient(Protocol):
+    """The one method `call_with_db_reconnect_retry` needs from a Prisma client."""
+
+    async def attempt_db_reconnect(
+        self,
+        *,
+        reason: str,
+        timeout_seconds: float | None = None,
+        lock_timeout_seconds: float | None = None,
+    ) -> bool: ...
+
+
 async def call_with_db_reconnect_retry(
-    prisma_client: Any,
+    prisma_client: _DBReconnectClient,
     coro_factory: Callable[[], Awaitable[_ReadResultT]],
     *,
     reason: str,
diff --git a/litellm/proxy/db/spend_log_tool_index.py b/litellm/proxy/db/spend_log_tool_index.py
index 93bbc567430..7a7982481fb 100644
--- a/litellm/proxy/db/spend_log_tool_index.py
+++ b/litellm/proxy/db/spend_log_tool_index.py
@@ -148,4 +148,4 @@ async def flush_tool_usage_transactions(
         except DB_RETRY_SAFE_ERROR_TYPES:
             if attempt >= n_retry_times:
                 raise
-            await asyncio.sleep(2**attempt + random.uniform(0, 1))
+            await asyncio.sleep(2.0**attempt + random.uniform(0, 1))
diff --git a/litellm/proxy/guardrails/guardrail_hooks/crowdstrike_aidr/crowdstrike_aidr.py b/litellm/proxy/guardrails/guardrail_hooks/crowdstrike_aidr/crowdstrike_aidr.py
index f7f500b1adc..db6fc238b9b 100644
--- a/litellm/proxy/guardrails/guardrail_hooks/crowdstrike_aidr/crowdstrike_aidr.py
+++ b/litellm/proxy/guardrails/guardrail_hooks/crowdstrike_aidr/crowdstrike_aidr.py
@@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, Annotated, Final, Literal, NamedTuple, Optiona
 
 from fastapi import HTTPException
 from pydantic import BaseModel, ConfigDict, Field, ValidationError
-from typing_extensions import Any, override
+from typing_extensions import override
 
 from litellm._logging import verbose_proxy_logger
 from litellm.integrations.custom_guardrail import (
@@ -78,7 +78,7 @@ class _GuardChatCompletionsResult(BaseModel):
     """Whether or not the prompt triggered a block detection."""
     transformed: bool | None = None
     """Whether or not the original input was transformed."""
-    detectors: dict[str, Any] | None = None
+    detectors: dict[str, object] | None = None
     """Result of the policy analyzing and input prompt."""
 
 
@@ -146,8 +146,8 @@ def _extract_text_from_message(message: _Message) -> str:
     return "\n".join(part.text for part in content if isinstance(part, _TextContentPart))
 
 
-def _merge_metadata_bags(request_data: Mapping[str, Any]) -> Mapping[str, Any] | None:
-    merged: Final[dict[str, Any]] = {}
+def _merge_metadata_bags(request_data: Mapping[str, object]) -> Mapping[str, object] | None:
+    merged: Final[dict[str, object]] = {}
     present = False
     for bag in (request_data.get("metadata"), request_data.get("litellm_metadata")):
         if isinstance(bag, Mapping):
@@ -313,7 +313,7 @@ class CrowdStrikeAIDRHandler(CustomGuardrail):
         self._set_streaming_params(streaming_params_from_litellm_params(litellm_params))
 
     async def _call_crowdstrike_aidr_guard(
-        self, payload: dict[str, Any], hook_name: str
+        self, payload: dict[str, object], hook_name: str
     ) -> _GuardChatCompletionsResult:
         """
         Makes the API call to the CrowdStrike AIDR AI Guard endpoint.
@@ -423,7 +423,7 @@ class CrowdStrikeAIDRHandler(CustomGuardrail):
         return [_extract_text_from_message(msg) for msg in tail]
 
     async def _call_or_fail_open(
-        self, payload: dict[str, Any], hook_name: str, request_data: dict[str, object]
+        self, payload: dict[str, object], hook_name: str, request_data: dict[str, object]
     ) -> _GuardChatCompletionsResult:
         start_time: Final = time.time()
         try:
@@ -506,7 +506,7 @@ class CrowdStrikeAIDRHandler(CustomGuardrail):
             event_type = "output"
             hook_name = "apply_guardrail (response)"
 
-        ai_guard_payload: Final[dict[str, Any]] = {
+        ai_guard_payload: Final[dict[str, object]] = {
             "guard_input": guard_input.model_dump(mode="json"),
             "event_type": event_type,
         }
@@ -521,7 +521,7 @@ class CrowdStrikeAIDRHandler(CustomGuardrail):
             if user_id:
                 ai_guard_payload["user_id"] = user_id
 
-            extra_info: Final[dict[str, str]] = {}
+            extra_info: Final[dict[str, object]] = {}
             user_email: Final = metadata.get("user_api_key_user_email")
             if user_email:
                 extra_info["user_name"] = user_email
diff --git a/litellm/proxy/guardrails/guardrail_hooks/enkryptai/enkryptai.py b/litellm/proxy/guardrails/guardrail_hooks/enkryptai/enkryptai.py
index 89afecafb0f..efe959bd186 100644
--- a/litellm/proxy/guardrails/guardrail_hooks/enkryptai/enkryptai.py
+++ b/litellm/proxy/guardrails/guardrail_hooks/enkryptai/enkryptai.py
@@ -6,7 +6,7 @@
 # +-------------------------------------------------------------+
 
 import os
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, AsyncIterable
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, Final, Literal, Optional
 
@@ -465,7 +465,7 @@ class EnkryptAIGuardrails(CustomGuardrail):
     async def async_post_call_streaming_iterator_hook(
         self,
         user_api_key_dict: UserAPIKeyAuth,
-        response: Any,
+        response: AsyncIterable[ModelResponseStream],
         request_data: dict,
     ) -> AsyncGenerator[ModelResponseStream, None]:
         """
diff --git a/litellm/proxy/guardrails/guardrail_hooks/qualifire/qualifire.py b/litellm/proxy/guardrails/guardrail_hooks/qualifire/qualifire.py
index d82944c44ed..eceb54681f6 100644
--- a/litellm/proxy/guardrails/guardrail_hooks/qualifire/qualifire.py
+++ b/litellm/proxy/guardrails/guardrail_hooks/qualifire/qualifire.py
@@ -383,7 +383,7 @@ class QualifireGuardrail(CustomGuardrail):
             result: Final = response.json()
 
             # Extract response info for logging
-            qualifire_response: Final = {
+            qualifire_response: Final[dict[str, object]] = {
                 "score": result.get("score"),
                 "status": result.get("status"),
             }
diff --git a/litellm/proxy/guardrails/guardrail_hooks/semantic_guard/route_loader.py b/litellm/proxy/guardrails/guardrail_hooks/semantic_guard/route_loader.py
index 8d5923d1302..b2139779925 100644
--- a/litellm/proxy/guardrails/guardrail_hooks/semantic_guard/route_loader.py
+++ b/litellm/proxy/guardrails/guardrail_hooks/semantic_guard/route_loader.py
@@ -6,6 +6,7 @@ then builds a SemanticRouter for prompt matching.
 """
 
 import os
+from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Final
 
 import yaml
@@ -66,7 +67,7 @@ class SemanticGuardRouteLoader:
         cls,
         route_templates: list[str] | None,
         custom_routes_file: str | None,
-        custom_routes: list[dict[str, Any]] | None,
+        custom_routes: Sequence[Mapping[str, object]] | None,
         global_threshold: float = DEFAULT_SEMANTIC_GUARD_SIMILARITY_THRESHOLD,
     ) -> list["Route"]:
         """Build semantic-router Route objects from templates + custom config."""
diff --git a/litellm/proxy/hooks/parallel_request_limiter.py b/litellm/proxy/hooks/parallel_request_limiter.py
index b313cb64c3f..d41acadc4dd 100644
--- a/litellm/proxy/hooks/parallel_request_limiter.py
+++ b/litellm/proxy/hooks/parallel_request_limiter.py
@@ -27,7 +27,7 @@ if TYPE_CHECKING:
 
     from litellm.proxy.utils import InternalUsageCache as _InternalUsageCache
 
-    Span = _Span | Any
+    Span = _Span
     InternalUsageCache = _InternalUsageCache
 else:
     Span = Any
@@ -75,7 +75,7 @@ class _PROXY_MaxParallelRequestsHandler(CustomLogger):
         current: dict | None,
         request_count_api_key: str,
         rate_limit_type: Literal["key", "model_per_key", "user", "customer", "team"],
-        values_to_update_in_cache: list[tuple[Any, Any]],
+        values_to_update_in_cache: list[tuple[str, object]],
     ) -> dict:
         verbose_proxy_logger.info("Current Usage of %s in this minute: %s", rate_limit_type, current)
         if current is None:
@@ -266,7 +266,7 @@ class _PROXY_MaxParallelRequestsHandler(CustomLogger):
             rpm_limit = sys.maxsize
 
         values_to_update_in_cache: list[
-            tuple[Any, Any]
+            tuple[str, object]
         ] = []  # values that need to get updated in cache, will run a batch_set_cache after this function
 
         # ------------
diff --git a/litellm/proxy/management_endpoints/common_daily_activity.py b/litellm/proxy/management_endpoints/common_daily_activity.py
index ce6a97708ab..6841cadf972 100644
--- a/litellm/proxy/management_endpoints/common_daily_activity.py
+++ b/litellm/proxy/management_endpoints/common_daily_activity.py
@@ -17,6 +17,7 @@ from litellm.proxy.spend_tracking.key_metadata_recovery import (
 )
 from litellm.proxy.spend_tracking.ptu_feature_flag import is_ptu_cost_attribution_enabled
 from litellm.proxy.utils import PrismaClient
+from litellm.repositories.prisma_protocols import TableActions
 from litellm.repositories.table_repositories import DeletedVerificationTokenRepository
 from litellm.repositories.verification_token_repository import (
     VerificationTokenRepository,
@@ -1155,8 +1156,10 @@ async def get_daily_activity(
             include_current_utc_day=include_current_utc_day,
         )
 
+        spend_table: Final[TableActions[DailySpendRecord]] = getattr(prisma_client.db, table_name)
+
         # Get total count for pagination
-        total_count: Final[int] = await getattr(prisma_client.db, table_name).count(where=where_conditions)
+        total_count: Final[int] = await spend_table.count(where=where_conditions)
 
         # Fetch paginated results.
         # ``date`` alone is not a unique sort key -- a busy tenant has many
@@ -1168,7 +1171,7 @@ async def get_daily_activity(
         # total. Adding ``id`` (the row's UUID primary key, present on both
         # LiteLLM_DailyUserSpend and LiteLLM_DailyTeamSpend) as a tiebreaker
         # gives every page a stable cursor (#30164).
-        daily_spend_data: Final[Sequence[DailySpendRecord]] = await getattr(prisma_client.db, table_name).find_many(
+        daily_spend_data: Final[Sequence[DailySpendRecord]] = await spend_table.find_many(
             where=where_conditions,
             order=[
                 {"date": "desc"},
diff --git a/litellm/proxy/ocr_endpoints/endpoints.py b/litellm/proxy/ocr_endpoints/endpoints.py
index ebf4d988fdd..1d0108cc2ba 100644
--- a/litellm/proxy/ocr_endpoints/endpoints.py
+++ b/litellm/proxy/ocr_endpoints/endpoints.py
@@ -2,7 +2,7 @@
 
 import json
 from collections.abc import Mapping
-from typing import Any, Final, cast
+from typing import Final, cast
 
 import orjson
 from fastapi import APIRouter, Depends, HTTPException, Request, Response, UploadFile
@@ -48,7 +48,7 @@ def _build_document_from_upload(
     )
 
 
-def _with_request_format(data: Mapping[str, Any], request: Request) -> Mapping[str, Any]:
+def _with_request_format(data: Mapping[str, object], request: Request) -> Mapping[str, object]:
     """
     Resolve the requested response format from the body or the `x-req-format` header.
 
@@ -90,7 +90,7 @@ def _native_response(response: object, fastapi_response: Response) -> Response |
     )
 
 
-async def _parse_multipart_form(request: Request) -> dict[str, Any]:
+async def _parse_multipart_form(request: Request) -> dict[str, object]:
     """
     Extract OCR data from a multipart form request.
 
@@ -130,7 +130,7 @@ async def _parse_multipart_form(request: Request) -> dict[str, Any]:
         content_type=uploaded_file.content_type,
     )
 
-    data: Final[dict[str, Any]] = {"document": document}
+    data: Final[dict[str, object]] = {"document": document}
 
     for field_name, field_value in form.items():
         if field_name in ("file", "document"):
@@ -154,12 +154,12 @@ async def _parse_multipart_form(request: Request) -> dict[str, Any]:
     return data
 
 
-async def _parse_ocr_request(request: Request) -> Mapping[str, Any]:
+async def _parse_ocr_request(request: Request) -> Mapping[str, object]:
     """Parse an OCR request and apply the `x-req-format` header, if any."""
     return _with_request_format(await _parse_ocr_request_body(request), request)
 
 
-async def _parse_ocr_request_body(request: Request) -> dict[str, Any]:
+async def _parse_ocr_request_body(request: Request) -> dict[str, object]:
     """
     Parse an OCR request, supporting both JSON and multipart form data.
 
@@ -320,7 +320,7 @@ async def ocr(
         # Process request using ProxyBaseLLMRequestProcessing
         processor = ProxyBaseLLMRequestProcessing(data=data)
 
-        response: Final = await processor.base_process_llm_request(
+        response: Final[object] = await processor.base_process_llm_request(
             request=request,
             fastapi_response=fastapi_response,
             user_api_key_dict=user_api_key_dict,
diff --git a/litellm/repositories/budget_repository.py b/litellm/repositories/budget_repository.py
index 62632ffb5f6..205646c8393 100644
--- a/litellm/repositories/budget_repository.py
+++ b/litellm/repositories/budget_repository.py
@@ -2,7 +2,8 @@
 Budget repository for database operations on LiteLLM_BudgetTable.
 """
 
-from typing import TYPE_CHECKING, Any, Final
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Final, Protocol
 
 from litellm.models.budget import LiteLLM_BudgetTable
 from litellm.repositories.base_repository import BaseRepository
@@ -12,12 +13,27 @@ if TYPE_CHECKING:
     from prisma import models as prisma_models
 
 
+class _BudgetDb(Protocol):
+    """The single Prisma table this repository reaches for on ``prisma_client.db``."""
+
+    @property
+    def litellm_budgettable(self) -> TableActions["prisma_models.LiteLLM_BudgetTable"]: ...
+
+
+class _PrismaClientView(Protocol):
+    """The one attribute this repository reads off the untyped Prisma client wrapper."""
+
+    @property
+    def db(self) -> _BudgetDb: ...
+
+
 class BudgetRepository(BaseRepository[LiteLLM_BudgetTable]):
     """Repository for budget database operations."""
 
     @property
     def table(self) -> TableActions["prisma_models.LiteLLM_BudgetTable"]:
-        return self.prisma_client.db.litellm_budgettable
+        client: Final[_PrismaClientView] = self.prisma_client
+        return client.db.litellm_budgettable
 
     @property
     def model_class(self) -> type[LiteLLM_BudgetTable]:
@@ -34,12 +50,12 @@ class BudgetRepository(BaseRepository[LiteLLM_BudgetTable]):
         max_parallel_requests: int | None = None,
         tpm_limit: int | None = None,
         rpm_limit: int | None = None,
-        model_max_budget: dict[str, Any] | None = None,
+        model_max_budget: Mapping[str, object] | None = None,
         budget_duration: str | None = None,
         allowed_models: list[str] | None = None,
     ) -> LiteLLM_BudgetTable:
         """Create a new budget record."""
-        data: Final[dict[str, Any]] = {
+        data: Final[dict[str, object]] = {
             "created_by": created_by,
             "updated_by": created_by,
         }
@@ -71,12 +87,12 @@ class BudgetRepository(BaseRepository[LiteLLM_BudgetTable]):
         max_parallel_requests: int | None = None,
         tpm_limit: int | None = None,
         rpm_limit: int | None = None,
-        model_max_budget: dict[str, Any] | None = None,
+        model_max_budget: Mapping[str, object] | None = None,
         budget_duration: str | None = None,
         allowed_models: list[str] | None = None,
     ) -> LiteLLM_BudgetTable | None:
         """Update an existing budget record."""
-        data: Final[dict[str, Any]] = {"updated_by": updated_by}
+        data: Final[dict[str, object]] = {"updated_by": updated_by}
         if max_budget is not None:
             data["max_budget"] = max_budget
         if soft_budget is not None:
diff --git a/litellm/repositories/organization_repository.py b/litellm/repositories/organization_repository.py
index 5a9bd3724e0..47eb8f4a609 100644
--- a/litellm/repositories/organization_repository.py
+++ b/litellm/repositories/organization_repository.py
@@ -2,7 +2,8 @@
 Organization repository for database operations on LiteLLM_OrganizationTable.
 """
 
-from typing import TYPE_CHECKING, Any, Final
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Final, Protocol
 
 from litellm.models.organization import LiteLLM_OrganizationTable
 from litellm.repositories.base_repository import BaseRepository
@@ -12,12 +13,27 @@ if TYPE_CHECKING:
     from prisma import models as prisma_models
 
 
+class _OrganizationDb(Protocol):
+    """The single Prisma table this repository reaches for on ``prisma_client.db``."""
+
+    @property
+    def litellm_organizationtable(self) -> TableActions["prisma_models.LiteLLM_OrganizationTable"]: ...
+
+
+class _PrismaClientView(Protocol):
+    """The one attribute this repository reads off the untyped Prisma client wrapper."""
+
+    @property
+    def db(self) -> _OrganizationDb: ...
+
+
 class OrganizationRepository(BaseRepository[LiteLLM_OrganizationTable]):
     """Repository for organization database operations."""
 
     @property
     def table(self) -> TableActions["prisma_models.LiteLLM_OrganizationTable"]:
-        return self.prisma_client.db.litellm_organizationtable
+        client: Final[_PrismaClientView] = self.prisma_client
+        return client.db.litellm_organizationtable
 
     @property
     def model_class(self) -> type[LiteLLM_OrganizationTable]:
@@ -39,12 +55,12 @@ class OrganizationRepository(BaseRepository[LiteLLM_OrganizationTable]):
         budget_id: str,
         created_by: str,
         organization_id: str | None = None,
-        metadata: dict[str, Any] | None = None,
+        metadata: Mapping[str, object] | None = None,
         models: list[str] | None = None,
         object_permission_id: str | None = None,
     ) -> LiteLLM_OrganizationTable:
         """Create a new organization."""
-        data: Final[dict[str, Any]] = {
+        data: Final[dict[str, object]] = {
             "organization_alias": organization_alias,
             "budget_id": budget_id,
             "created_by": created_by,
@@ -67,12 +83,12 @@ class OrganizationRepository(BaseRepository[LiteLLM_OrganizationTable]):
         updated_by: str,
         organization_alias: str | None = None,
         budget_id: str | None = None,
-        metadata: dict[str, Any] | None = None,
+        metadata: Mapping[str, object] | None = None,
         models: list[str] | None = None,
         object_permission_id: str | None = None,
     ) -> LiteLLM_OrganizationTable | None:
         """Update an organization."""
-        data: Final[dict[str, Any]] = {"updated_by": updated_by}
+        data: Final[dict[str, object]] = {"updated_by": updated_by}
         if organization_alias is not None:
             data["organization_alias"] = organization_alias
         if budget_id is not None:
diff --git a/litellm/repositories/project_repository.py b/litellm/repositories/project_repository.py
index 48e55efd258..905e813f35e 100644
--- a/litellm/repositories/project_repository.py
+++ b/litellm/repositories/project_repository.py
@@ -2,7 +2,8 @@
 Project repository for database operations on LiteLLM_ProjectTable.
 """
 
-from typing import TYPE_CHECKING, Any, Final
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Final
 
 from litellm.models.project import LiteLLM_ProjectTable
 from litellm.repositories.base_repository import BaseRepository
@@ -43,14 +44,14 @@ class ProjectRepository(BaseRepository[LiteLLM_ProjectTable]):
         description: str | None = None,
         team_id: str | None = None,
         budget_id: str | None = None,
-        metadata: dict[str, Any] | None = None,
+        metadata: Mapping[str, object] | None = None,
         models: list[str] | None = None,
         model_rpm_limit: dict[str, int] | None = None,
         model_tpm_limit: dict[str, int] | None = None,
         object_permission_id: str | None = None,
     ) -> LiteLLM_ProjectTable:
         """Create a new project."""
-        data: Final[dict[str, Any]] = {
+        data: Final[dict[str, object]] = {
             "created_by": created_by,
             "updated_by": created_by,
         }
@@ -85,7 +86,7 @@ class ProjectRepository(BaseRepository[LiteLLM_ProjectTable]):
         description: str | None = None,
         team_id: str | None = None,
         budget_id: str | None = None,
-        metadata: dict[str, Any] | None = None,
+        metadata: Mapping[str, object] | None = None,
         models: list[str] | None = None,
         model_rpm_limit: dict[str, int] | None = None,
         model_tpm_limit: dict[str, int] | None = None,
@@ -93,7 +94,7 @@ class ProjectRepository(BaseRepository[LiteLLM_ProjectTable]):
         object_permission_id: str | None = None,
     ) -> LiteLLM_ProjectTable | None:
         """Update a project."""
-        data: Final[dict[str, Any]] = {"updated_by": updated_by}
+        data: Final[dict[str, object]] = {"updated_by": updated_by}
         if project_alias is not None:
             data["project_alias"] = project_alias
         if description is not None:
diff --git a/litellm/router_strategy/adaptive_router/adaptive_router.py b/litellm/router_strategy/adaptive_router/adaptive_router.py
index 12ccacbbc1d..8727c3a69f0 100644
--- a/litellm/router_strategy/adaptive_router/adaptive_router.py
+++ b/litellm/router_strategy/adaptive_router/adaptive_router.py
@@ -16,6 +16,7 @@ from __future__ import annotations
 import asyncio
 import time
 from collections import OrderedDict
+from collections.abc import Mapping
 from dataclasses import asdict, dataclass
 from typing import Any, Final, cast
 
@@ -122,7 +123,7 @@ class AdaptiveRouter:
                 prefs = self.model_to_prefs.get(model) or _default_prefs()
                 self._cells[(rt, model)] = initial_cell(prefs, rt)
 
-    async def load_state_from_db(self, prisma_client: Any) -> None:
+    async def load_state_from_db(self, prisma_client: object) -> None:
         """Add each row's persisted delta to a freshly computed cold-start prior.
 
         A row holds an accumulated delta, not a full posterior, and can be one-sided
@@ -237,7 +238,7 @@ class AdaptiveRouter:
             cost_weight=self.config.weights.cost,
         )
 
-    async def get_state_snapshot(self) -> dict[str, Any]:
+    async def get_state_snapshot(self) -> dict[str, object]:
         """In-memory snapshot for the introspection endpoint. Cheap; no DB hit."""
         cells: Final = []
         for (rt, model), cell in sorted(self._cells.items(), key=lambda kv: (kv[0][0].value, kv[0][1])):
@@ -278,7 +279,7 @@ class AdaptiveRouter:
 
     @staticmethod
     def _extract_min_quality_tier(
-        request_kwargs: dict[str, Any],
+        request_kwargs: Mapping[str, object],
     ) -> int | None:
         """Pull `min_quality_tier` from request headers or metadata.
 
@@ -484,7 +485,7 @@ class AdaptiveRouter:
             return combined_delta
 
     @staticmethod
-    def _persistable_session_snapshot(state: SessionState) -> dict[str, Any]:
+    def _persistable_session_snapshot(state: SessionState) -> dict[str, object]:
         snapshot: Final = asdict(state)
         for sensitive in (
             "last_user_content",
diff --git a/litellm/router_strategy/adaptive_router/signals.py b/litellm/router_strategy/adaptive_router/signals.py
index c28613b54eb..72e8d27d2bf 100644
--- a/litellm/router_strategy/adaptive_router/signals.py
+++ b/litellm/router_strategy/adaptive_router/signals.py
@@ -92,7 +92,7 @@ class Turn:
 
     user_content: str | None = None
     assistant_content: str | None = None
-    tool_calls: list[dict[str, Any]] = field(default_factory=list)
+    tool_calls: Sequence[Mapping[str, object]] = field(default_factory=list[Mapping[str, object]])
     tool_results: Sequence[Mapping[str, object]] = field(default_factory=list)
     response_status: int | None = None
 
@@ -174,7 +174,7 @@ def _detect_failure(tool_results: Sequence[Mapping[str, object]]) -> bool:
     return False
 
 
-def _signature(call: dict[str, Any]) -> str:
+def _signature(call: Mapping[str, Any]) -> str:
     """Stable signature for loop detection: name + sorted JSON-ish args."""
     name: Final = call.get("name") or call.get("function", {}).get("name", "")
     call_args = call.get("arguments")
@@ -185,7 +185,7 @@ def _signature(call: dict[str, Any]) -> str:
     return f"{name}({call_args})"
 
 
-def _detect_loop(history: list[str], new_calls: list[dict[str, Any]]) -> bool:
+def _detect_loop(history: list[str], new_calls: Sequence[Mapping[str, object]]) -> bool:
     """Fires if any new call's signature appears >= LOOP_REPEAT_THRESHOLD-1 times
     in recent history (so this call would be the Nth)."""
     if not new_calls:
@@ -238,7 +238,7 @@ def detect_response_signals(
     previous_assistant_content: str | None,
     current_assistant_content: str | None,
     tool_call_history: list[str],
-    tool_calls: list[dict[str, Any]],
+    tool_calls: Sequence[Mapping[str, object]],
     tool_results: Sequence[Mapping[str, object]],
     response_status: int | None,
 ) -> SignalDelta:
diff --git a/litellm/router_strategy/adaptive_router/update_queue.py b/litellm/router_strategy/adaptive_router/update_queue.py
index 1b9fce284ac..e28f2379f9c 100644
--- a/litellm/router_strategy/adaptive_router/update_queue.py
+++ b/litellm/router_strategy/adaptive_router/update_queue.py
@@ -19,7 +19,8 @@ to the in-memory aggregator). Flush is async and batched.
 from __future__ import annotations
 
 import asyncio
-from typing import Any, Final
+from collections.abc import Mapping
+from typing import Final
 
 from litellm._logging import verbose_router_logger
 from litellm.repositories.table_repositories import (
@@ -39,7 +40,7 @@ class AdaptiveRouterUpdateQueue:
 
     def __init__(self) -> None:
         self._state_agg: dict[StateKey, dict[str, float]] = {}
-        self._session_agg: dict[SessionKey, dict[str, Any]] = {}
+        self._session_agg: dict[SessionKey, Mapping[str, object]] = {}
         self._lock = asyncio.Lock()
         self._max_state_size_seen = 0
         self._max_session_size_seen = 0
@@ -77,7 +78,7 @@ class AdaptiveRouterUpdateQueue:
         session_id: str,
         router_name: str,
         model_name: str,
-        state_dict: dict[str, Any],
+        state_dict: Mapping[str, object],
     ) -> None:
         """
         Last-write-wins per session row. The state_dict is a snapshot of the
@@ -91,7 +92,7 @@ class AdaptiveRouterUpdateQueue:
 
     # ---- Flushers (called by background task) ----------------------------
 
-    async def flush_state_to_db(self, prisma_client: Any) -> int:
+    async def flush_state_to_db(self, prisma_client: object) -> int:
         """
         Drain state aggregator and apply to LiteLLM_AdaptiveRouterState.
         Returns number of cells flushed.
@@ -147,7 +148,7 @@ class AdaptiveRouterUpdateQueue:
 
         return len(batch)
 
-    async def flush_session_to_db(self, prisma_client: Any) -> int:
+    async def flush_session_to_db(self, prisma_client: object) -> int:
         """
         Drain session aggregator and upsert into LiteLLM_AdaptiveRouterSession.
         Returns number of session rows flushed.
diff --git a/litellm/types/containers/main.py b/litellm/types/containers/main.py
index 6a339fd2eac..62ef524a435 100644
--- a/litellm/types/containers/main.py
+++ b/litellm/types/containers/main.py
@@ -140,7 +140,7 @@ class ContainerFileObject(BaseModel):
     created_at: int
     path: str
     source: str
-    _hidden_params: dict[str, Any] = {}
+    _hidden_params: dict[str, builtins.object] = {}
 
     def __contains__(self, key: str) -> bool:
         return hasattr(self, key)
diff --git a/litellm/types/guardrails.py b/litellm/types/guardrails.py
index 02dee40f2a3..175226f8d4b 100644
--- a/litellm/types/guardrails.py
+++ b/litellm/types/guardrails.py
@@ -2,10 +2,10 @@ from collections.abc import Mapping
 from datetime import datetime
 from enum import Enum
 from types import MappingProxyType
-from typing import Any, Final, Literal
+from typing import Final, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
-from typing_extensions import Required, TypedDict
+from typing_extensions import ReadOnly, Required, TypedDict
 
 from litellm.constants import BEDROCK_APPLY_GUARDRAIL_CHUNK_BUDGET_CHARS
 from litellm.types.proxy.guardrails.guardrail_hooks.akto import (
@@ -935,7 +935,7 @@ class BaseLitellmParams(ContentFilterConfigModel):  # works for new and patch up
         ),
     )
 
-    additional_provider_specific_params: dict[str, Any] | None = Field(
+    additional_provider_specific_params: dict[str, object] | None = Field(
         default=None,
         description="Additional provider-specific parameters for generic guardrail APIs",
     )
@@ -1157,7 +1157,7 @@ class GuardrailEventHooks(str, Enum):
 
 
 class DynamicGuardrailParams(TypedDict):
-    extra_body: dict[str, Any]
+    extra_body: ReadOnly[dict[str, object]]
 
 
 class GUARDRAIL_DEFINITION_LOCATION(str, Enum):
@@ -1188,7 +1188,7 @@ class GuardrailUIAddGuardrailSettings(BaseModel):
     supported_modes: list[str]
     supported_modes_by_provider: dict[str, list[str]]
     pii_entity_categories: list[PiiEntityCategoryMap]
-    content_filter_settings: dict[str, Any] | None = None
+    content_filter_settings: dict[str, object] | None = None
 
 
 class PresidioPerRequestConfig(BaseModel):
@@ -1206,8 +1206,8 @@ class ApplyGuardrailRequest(BaseModel):
     language: str | None = None
     entities: list[PiiEntityType] | None = None
     input_type: str = "request"
-    messages: list[dict[str, Any]] | None = None
-    metadata: dict[str, Any] | None = None
+    messages: list[dict[str, object]] | None = None
+    metadata: dict[str, object] | None = None
 
 
 class ApplyGuardrailResponse(BaseModel):
@@ -1217,4 +1217,4 @@ class ApplyGuardrailResponse(BaseModel):
 class PatchGuardrailRequest(BaseModel):
     guardrail_name: str | None = None
     litellm_params: BaseLitellmParams | None = None
-    guardrail_info: dict[str, Any] | None = None
+    guardrail_info: dict[str, object] | None = None
diff --git a/litellm/types/integrations/prometheus.py b/litellm/types/integrations/prometheus.py
index 8498b6f6d00..323f43f4b01 100644
--- a/litellm/types/integrations/prometheus.py
+++ b/litellm/types/integrations/prometheus.py
@@ -44,7 +44,7 @@ def _sanitize_prometheus_label_name(label: str) -> str:
 _PROMETHEUS_LABEL_VALUE_TRANSLATE_V1: Final = str.maketrans("\n", " ", "\r\u2028\u2029")
 
 
-def _sanitize_prometheus_label_value(value: Any | None) -> str | None:
+def _sanitize_prometheus_label_value(value: object | None) -> str | None:
     """
     Same semantics as :func:`_sanitize_prometheus_label_value`, implemented with
     ``str.translate`` plus a single escape pass instead of chained ``replace``.
@@ -1023,7 +1023,7 @@ class UserAPIKeyLabelValues:
         ``hashed_api_key``. This supports ``**standard_logging_payload`` in tests.
         """
         field_names: Final = {f.name for f in fields(self)}
-        merged: Final[dict[str, Any]] = {}
+        merged: Final[dict[str, object]] = {}
         for f in fields(self):
             if f.default_factory is not MISSING:
                 merged[f.name] = f.default_factory()
@@ -1060,9 +1060,9 @@ class UserAPIKeyLabelValues:
         # stays cheap. (Dataclass default `str()` delegates to `__repr__`.)
         return ""
 
-    def model_dump(self) -> dict[str, Any]:
+    def model_dump(self) -> dict[str, object]:
         """Same shape as the former Pydantic ``model_dump()`` (plain dict, list tags)."""
-        d: Final[dict[str, Any]] = {f.name: getattr(self, f.name) for f in fields(self)}
+        d: Final[dict[str, object]] = {f.name: getattr(self, f.name) for f in fields(self)}
         d["tags"] = list(self.tags)
         d["custom_metadata_labels"] = dict(self.custom_metadata_labels)
         return d
diff --git a/litellm/types/realtime.py b/litellm/types/realtime.py
index 17dc70126f3..bda3865c46a 100644
--- a/litellm/types/realtime.py
+++ b/litellm/types/realtime.py
@@ -78,15 +78,15 @@ class RealtimeSessionConfig(BaseModel):
     type: str | None = None
     model: str | None = None
     instructions: str | None = None
-    audio: dict[str, Any] | None = None
+    audio: dict[str, object] | None = None
     include: list[str] | None = None
     max_output_tokens: int | str | None = None
     output_modalities: list[str] | None = None
-    tool_choice: Any | None = None
-    tools: list[dict[str, Any]] | None = None
-    tracing: Any | None = None
-    truncation: Any | None = None
-    prompt: dict[str, Any] | None = None
+    tool_choice: object | None = None
+    tools: list[dict[str, object]] | None = None
+    tracing: object | None = None
+    truncation: object | None = None
+    prompt: dict[str, object] | None = None
 
 
 class RealtimeClientSecretRequest(BaseModel):
@@ -114,7 +114,7 @@ class RealtimeClientSecretResponse(BaseModel):
 
     expires_at: int | None = None
     value: str
-    session: dict[str, Any] | None = None
+    session: dict[str, object] | None = None
 
 
 class RealtimeTranscriptionSessionRequest(BaseModel):
@@ -151,7 +151,7 @@ class RealtimeTranscriptionSessionResponse(BaseModel):
 
     model_config = {"extra": "allow"}
 
-    client_secret: dict[str, Any] | None = None
+    client_secret: dict[str, object] | None = None
 
 
 class RealtimeErrorDetail(TypedDict):
diff --git a/litellm/vector_store_files/utils.py b/litellm/vector_store_files/utils.py
index 94ad5c0ecdf..8b4bff921f8 100644
--- a/litellm/vector_store_files/utils.py
+++ b/litellm/vector_store_files/utils.py
@@ -1,4 +1,5 @@
-from typing import Any, Final, cast, get_type_hints
+from collections.abc import Mapping
+from typing import Final, cast, get_type_hints
 
 from litellm.types.vector_store_files import (
     VectorStoreFileCreateRequest,
@@ -11,25 +12,25 @@ class VectorStoreFileRequestUtils:
     """Helper utilities for constructing vector store file requests."""
 
     @staticmethod
-    def _filter_params(params: dict[str, Any], model: Any) -> dict[str, Any]:
+    def _filter_params(params: Mapping[str, object], model: type[object]) -> dict[str, object]:
         valid_keys: Final = get_type_hints(model).keys()
         return {key: value for key, value in params.items() if key in valid_keys and value is not None}
 
     @staticmethod
     def get_create_request_params(
-        params: dict[str, Any],
+        params: Mapping[str, object],
     ) -> VectorStoreFileCreateRequest:
         filtered: Final = VectorStoreFileRequestUtils._filter_params(params=params, model=VectorStoreFileCreateRequest)
         return cast(VectorStoreFileCreateRequest, filtered)
 
     @staticmethod
-    def get_list_query_params(params: dict[str, Any]) -> VectorStoreFileListQueryParams:
+    def get_list_query_params(params: Mapping[str, object]) -> VectorStoreFileListQueryParams:
         filtered = VectorStoreFileRequestUtils._filter_params(params=params, model=VectorStoreFileListQueryParams)
         return cast(VectorStoreFileListQueryParams, filtered)
 
     @staticmethod
     def get_update_request_params(
-        params: dict[str, Any],
+        params: Mapping[str, object],
     ) -> VectorStoreFileUpdateRequest:
         filtered: Final = VectorStoreFileRequestUtils._filter_params(params=params, model=VectorStoreFileUpdateRequest)
         return cast(VectorStoreFileUpdateRequest, filtered)