Merge 018a6a90ad into 860bc7811d

litellm/cost_calculator.py

@@ -2017,8 +2017,8 @@ def _deployment_model_info(
         return cast(ModelInfo, registered_deployment_info)  # cast-ok: router registers deployment prices under its id
     if litellm_logging_obj is None:
         return None
-    litellm_params: Final = getattr(litellm_logging_obj, "litellm_params", None)
-    if litellm_params is None:
+    litellm_params: Final = litellm_logging_obj.litellm_params
+    if not litellm_params:
         return None
     return next(
         (
@@ -2036,7 +2036,9 @@ def _ocr_model_info(
     router_model_id: str | None,
 ) -> OCRPricing | None:
     deployment_info: Final = _deployment_model_info(litellm_logging_obj, custom_pricing, router_model_id)
-    litellm_params: Final = getattr(litellm_logging_obj, "litellm_params", None) if custom_pricing else None
+    litellm_params: Final = (
+        litellm_logging_obj.litellm_params if custom_pricing and litellm_logging_obj is not None else None
+    )
     if litellm_params is None:
         return deployment_info
     return _layered_ocr_pricing(litellm_params, deployment_info)

litellm/experimental_mcp_client/tools.py

@@ -129,7 +129,7 @@ async def list_tools_with_pagination(
             )
             tools.extend(result.tools)
 
-            next_cursor = getattr(result, "next_cursor", None)
+            next_cursor = result.next_cursor
             if not isinstance(next_cursor, str) or not next_cursor:
                 return tools
             if next_cursor in seen_cursors:

litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py

@@ -21,7 +21,7 @@ from __future__ import annotations
 
 import os
 from datetime import datetime, timedelta, timezone
-from typing import TYPE_CHECKING, Any, Final, Protocol
+from typing import TYPE_CHECKING, Any, Final
 
 import litellm
 from litellm._logging import verbose_proxy_logger
@@ -35,17 +35,6 @@ else:
     AsyncIOScheduler = Any
 
 
-class _PodLockManager(Protocol):
-    """The subset of PodLockManager this logger drives to serialize the export across pods."""
-
-    @property
-    def redis_cache(self) -> object: ...
-
-    async def acquire_lock(self, cronjob_id: str) -> bool | None: ...
-
-    async def release_lock(self, cronjob_id: str) -> None: ...
-
-
 def _parse_metrics_marker(
     marker: object | None,
 ) -> datetime | None:
@@ -237,13 +226,10 @@ class MavvrikFocusLogger(FocusLogger):
         """Scheduler entry point — uses Mavvrik-specific pod-lock key."""
         from litellm.proxy.proxy_server import proxy_logging_obj  # noqa: PLC0415
 
-        pod_lock_manager: _PodLockManager | None = None
-        if proxy_logging_obj is not None:
-            writer: Final[object] = getattr(proxy_logging_obj, "db_spend_update_writer", None)
-            if writer is not None:
-                pod_lock_manager = getattr(writer, "pod_lock_manager", None)
-
-        if pod_lock_manager and pod_lock_manager.redis_cache:
+        pod_lock_manager: Final = (
+            proxy_logging_obj.db_spend_update_writer.pod_lock_manager if proxy_logging_obj is not None else None
+        )
+        if pod_lock_manager is not None and pod_lock_manager.redis_cache:
             acquired: Final = await pod_lock_manager.acquire_lock(cronjob_id=MAVVRIK_FOCUS_EXPORT_JOB_NAME)
             if not acquired:
                 verbose_proxy_logger.debug("Mavvrik FOCUS export: unable to acquire pod lock")

litellm/integrations/websearch_interception/handler.py

@@ -1849,7 +1849,7 @@ class WebSearchInterceptionLogger(CustomLogger):
         for tool_call in tool_calls:
             # Handle both Anthropic-style input and OpenAI-style function.arguments
             query = None
-            tool_args: dict | None = None  # mutable-ok: the tool call's own arguments dict
+            tool_args: dict[str, object] | None = None  # mutable-ok: the tool call's own arguments dict
             if "input" in tool_call and isinstance(tool_call["input"], dict):
                 tool_args = tool_call["input"]
                 query = tool_args.get("query")

litellm/litellm_core_utils/core_helpers.py

@@ -365,7 +365,7 @@ def _budget_reservation_on_auth_object(user_api_key_auth: object) -> object:
     return getattr(user_api_key_auth, "budget_reservation", None)
 
 
-def budget_reservation_from_metadata(metadata: Mapping[str, object]) -> dict | None:
+def budget_reservation_from_metadata(metadata: Mapping[str, object]) -> dict[str, object] | None:
     stamped: Final = metadata.get("user_api_key_budget_reservation")
     if isinstance(stamped, dict):
         return stamped

litellm/litellm_core_utils/litellm_logging.py

@@ -5173,7 +5173,7 @@ def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list[Custom
     for callback in _in_memory_loggers:
         if (
             isinstance(callback, OpenTelemetryV2)
-            and getattr(callback, "callback_name", None) == callback_name
+            and callback.callback_name == callback_name
             and (serves_a_destination or not _exports_nowhere(callback.config))
         ):
             return callback
@@ -6645,11 +6645,11 @@ def get_standard_logging_object_payload(
             cost_breakdown=request_cost_breakdown,
             autorouter_savings=autorouter_savings,
             autorouter_savings_estimate=(
-                {
+                {  # mutable-ok: spend-log JSON serialization requires plain mappings
                     "version": 3,
                     "status": "unknown",
                     "reason": "pending_projection",
-                }  # mutable-ok: spend-log JSON serialization requires plain mappings
+                }
                 if captured_baseline is not None
                 else (
                     {  # mutable-ok: spend-log JSON serialization requires plain mappings

litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py

@@ -5,7 +5,12 @@ from typing import Final
 from pydantic import TypeAdapter, ValidationError
 from typing_extensions import assert_never
 
-from litellm.types.utils import StandardLoggingZeroCostDiagnostic, Usage
+from litellm.types.utils import (
+    CompletionTokensDetailsWrapper,
+    PromptTokensDetailsWrapper,
+    StandardLoggingZeroCostDiagnostic,
+    Usage,
+)
 
 ZERO_COST_COUNTER_NAME: Final = "litellm_zero_cost_requests_total"
 
@@ -18,8 +23,8 @@ _NESTED_PRICING: Final = TypeAdapter(Mapping[str, object] | tuple[object, ...])
 _MAX_PRICING_DEPTH: Final = 4
 
 
-def _audio_tokens(details: object) -> int:
-    audio_tokens: Final = getattr(details, "audio_tokens", None)
+def _audio_tokens(details: PromptTokensDetailsWrapper | CompletionTokensDetailsWrapper | None) -> int:
+    audio_tokens: Final = details.audio_tokens if details is not None else None
     return audio_tokens if isinstance(audio_tokens, int) and audio_tokens > 0 else 0
 
 

litellm/litellm_core_utils/prompt_templates/common_utils.py

@@ -2003,11 +2003,11 @@ def strip_encrypted_reasoning_from_messages(messages: object) -> None:
     """
     if not isinstance(messages, list):
         return
-    for content in _anthropic_content_lists(cast(list[object], messages)):  # cast-ok: untyped client json
+    for content in anthropic_content_lists(cast(list[object], messages)):  # cast-ok: untyped client json
         _strip_encrypted_reasoning_from_blocks(content)
 
 
-def _anthropic_content_lists(messages: Sequence[object]) -> Iterator[object]:
+def anthropic_content_lists(messages: Sequence[object]) -> Iterator[object]:
     return (
         cast(list[object], content)  # cast-ok: narrowed by isinstance
         for message in messages

litellm/litellm_core_utils/streaming_handler.py

@@ -1329,7 +1329,7 @@ class CustomStreamWrapper:
                 "is_finished": chunk_finish_reason is not None,
                 "finish_reason": chunk_finish_reason,
                 "original_chunk": cached_chunk,
-                "tool_calls": (getattr(cached_choice.delta, "tool_calls", None) if cached_choice is not None else None),
+                "tool_calls": cached_choice.delta.tool_calls if cached_choice is not None else None,
             }
 
             completion_obj["content"] = response_obj["text"]

litellm/llms/a2a/chat/transformation.py

@@ -48,7 +48,7 @@ def _registry_api_key(agent_litellm_params: Mapping[str, object]) -> str | None:
     return configured_api_key if isinstance(configured_api_key, str) else None
 
 
-def _registry_headers(agent_litellm_params: Mapping[str, object]) -> dict[str, Any] | None:
+def _registry_headers(agent_litellm_params: Mapping[str, object]) -> dict[str, object] | None:
     stored_headers: Final = agent_litellm_params.get("headers")
     if not isinstance(stored_headers, Mapping):
         return None

litellm/llms/anthropic/chat/guardrail_translation/handler.py

@@ -199,9 +199,7 @@ def _write_back_system_block(system: object, block_idx: int, response: str) -> N
         return
     text_blocks: Final = tuple(block for block in system if isinstance(block, dict) and block.get("type") == "text")
     if block_idx < len(text_blocks):
-        text_blocks[block_idx]["text"] = (
-            response  # mutable-ok: guardrails rewrite the caller's request payload in place
-        )
+        text_blocks[block_idx]["text"] = response
 
 
 def _write_back_message_text(message: _WritableMessage, target: MessageTextTarget, response: str) -> None:
@@ -211,22 +209,16 @@ def _write_back_message_text(message: _WritableMessage, target: MessageTextTarge
     match target:
         case MessageContentTarget():
             if isinstance(content, str):
-                message["content"] = response  # mutable-ok: guardrails rewrite the caller's request payload in place
+                message["content"] = response
         case ContentBlockTextTarget(content_idx=content_idx):
             if isinstance(content, list):
-                content[content_idx]["text"] = (
-                    response  # mutable-ok: guardrails rewrite the caller's request payload in place
-                )
+                content[content_idx]["text"] = response
         case ToolResultStringTarget(content_idx=content_idx):
             if isinstance(content, list):
-                content[content_idx]["content"] = (
-                    response  # mutable-ok: guardrails rewrite the caller's request payload in place
-                )
+                content[content_idx]["content"] = response
         case ToolResultBlockTextTarget(content_idx=content_idx, block_idx=block_idx):
             if isinstance(content, list):
-                content[content_idx]["content"][block_idx]["text"] = (
-                    response  # mutable-ok: guardrails rewrite the caller's request payload in place
-                )
+                content[content_idx]["content"][block_idx]["text"] = response
         case _:
             assert_never(target)
 
@@ -248,9 +240,9 @@ def _write_back_tool_use(
     block: Final = content[target.content_idx] if isinstance(content, list) else None
     if not isinstance(block, dict):
         return
-    block["input"] = rewritten_input  # mutable-ok: guardrails rewrite the caller's request payload in place
+    block["input"] = rewritten_input
     if shape.name is not None and shape.name != block.get("name"):
-        block["name"] = shape.name  # mutable-ok: guardrails rewrite the caller's request payload in place
+        block["name"] = shape.name
 
 
 @dataclass(frozen=True, slots=True)
@@ -607,9 +599,7 @@ class AnthropicMessagesHandler(BaseTranslation):
             image for one_message in extracted for image in one_message.images
         ]  # mutable-ok: GenericGuardrailAPIInputs takes list[str]
         scanned_tool_calls: Final = tuple(item for one_message in extracted for item in one_message.tool_calls)
-        tool_calls_to_check: Final = [
-            item.tool_call for item in scanned_tool_calls
-        ]  # mutable-ok: GenericGuardrailAPIInputs takes list[ChatCompletionToolCallChunk]
+        tool_calls_to_check: Final = [item.tool_call for item in scanned_tool_calls]
         pre_guardrail_tool_calls: Final = _tool_call_shapes(tool_calls_to_check)
 
         # Step 2: Apply guardrail to all texts and tool calls in batch
@@ -697,9 +687,7 @@ class AnthropicMessagesHandler(BaseTranslation):
 
         return data
 
-    def _hoisted_top_level_system_message(
-        self, data: dict
-    ) -> AllMessageValues | None:  # mutable-ok: API message payload
+    def _hoisted_top_level_system_message(self, data: Mapping[str, object]) -> AllMessageValues | None:
         """Return the system message produced by translating the top-level prompt."""
         system: Final = data.get("system")
         if not system:

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

@@ -11,7 +11,6 @@ from typing import (
     Final,
     Literal,
     Protocol,
-    cast,  # noqa: TID251  # rebuilt message_delta dict spans the ContentBlockDelta/MessageBlockDelta union
     get_args,
 )
 
@@ -27,6 +26,7 @@ from litellm.types.llms.anthropic import (
     ContentBlockDelta,
     ContextManagementResponse,
     MessageBlockDelta,
+    MessageDelta,
     StreamingContentBlockDeltaType,
     UsageDelta,
     UsageIteration,
@@ -1028,26 +1028,22 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
         self,
         processed_chunk: ContentBlockDelta | MessageBlockDelta,
     ) -> ContentBlockDelta | MessageBlockDelta:
-        if processed_chunk.get("type") != "message_delta" or not self._refusal_text:
+        if processed_chunk["type"] != "message_delta" or not self._refusal_text:
             return processed_chunk
-        delta: Final = cast(Mapping[str, object], processed_chunk["delta"])  # cast-ok: keys checked before use
+        delta: Final = processed_chunk["delta"]
         if delta.get("stop_reason") == "max_tokens":
             return processed_chunk
         from litellm.llms.anthropic.experimental_pass_through.messages.utils import (
             refusal_stop_details,
         )
 
-        return cast(  # cast-ok: rebuilt dict matches the message_delta TypedDict shape for this branch
-            ContentBlockDelta | MessageBlockDelta,
-            {  # mutable-ok: fresh translation payload; never mutated after construction
-                **processed_chunk,
-                "delta": {  # mutable-ok: fresh message_delta payload; never mutated after construction
-                    **delta,
-                    "stop_reason": "refusal",
-                    "stop_details": refusal_stop_details(self._refusal_text),
-                },
-            },
-        )
+        refusal_delta: Final[MessageDelta] = {  # mutable-ok: fresh message_delta payload
+            **delta,
+            "stop_reason": "refusal",
+            "stop_details": refusal_stop_details(self._refusal_text),
+        }
+        refusal_chunk: Final[MessageBlockDelta] = {**processed_chunk, "delta": refusal_delta}
+        return refusal_chunk
 
     @staticmethod
     def _delta_has_content(processed_chunk: Mapping[str, object]) -> bool:

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

@@ -37,7 +37,7 @@ def _mapping_field(container: object, key: str) -> object | None:
     """One key of a raw provider payload, or None when the payload is not a mapping."""
     if not isinstance(container, Mapping):
         return None
-    return cast(Mapping[str, object], container).get(key)  # cast-ok: raw payload, callers re-check every value
+    return container.get(key)
 
 
 def _mapping_str_field(container: object, key: str) -> str | None:

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

@@ -171,7 +171,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
         cls,
         summary: Iterable[object],
         encrypted_content: object,
-    ) -> dict[str, Any] | None:  # mutable-ok: API message payload
+    ) -> dict[str, object] | None:  # mutable-ok: API message payload
         """The one Anthropic block for a Responses reasoning item.
 
         The item's encrypted reasoning rides the block's opaque field (`signature`, or
@@ -200,7 +200,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
     @classmethod
     def _assistant_group_to_input_items(
         cls, group: tuple[Mapping[str, object], ...]
-    ) -> tuple[dict[str, Any], ...]:  # mutable-ok: API message payload
+    ) -> tuple[dict[str, object], ...]:  # mutable-ok: API message payload
         first: Final = group[0]
         btype: Final = first.get("type")
         if btype in ("thinking", "redacted_thinking"):

litellm/llms/nvidia_nim/passthrough/transformation.py

@@ -110,7 +110,7 @@ class NvidiaNimPassthroughConfig(BasePassthroughConfig):
         return {
             **headers,
             "Authorization": f"Bearer {api_key}",
-        }  # mutable-ok: base class contract returns dict for httpx
+        }
 
     @staticmethod
     def get_api_base(api_base: str | None = None) -> str | None:

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

@@ -994,7 +994,7 @@ class OpenAIResponsesHandler(BaseTranslation):
 
     def _spread_text_rewrite_over_stream_events(
         self,
-        stream_events: Sequence[Any],
+        stream_events: Sequence[object],
         rewritten_text: str,
         guardrail_name: str,
     ) -> None:

litellm/llms/vercel_ai_gateway/embedding/transformation.py

@@ -161,7 +161,9 @@ class VercelAIGatewayEmbeddingConfig(BaseEmbeddingConfig):
                 optional_params[param] = value
         return optional_params
 
-    def get_error_class(self, error_message: str, status_code: int, headers: Any) -> BaseLLMException:
+    def get_error_class(
+        self, error_message: str, status_code: int, headers: dict[str, str] | httpx.Headers
+    ) -> BaseLLMException:
         """
         Get the error class for Vercel AI Gateway errors.
         """

litellm/main.py

@@ -1182,7 +1182,7 @@ def _is_claude_tool_target(custom_llm_provider: str | None, model: str) -> bool:
     return False
 
 
-def _without_anthropic_only_tool_keys(tool: dict) -> dict:
+def _without_anthropic_only_tool_keys(tool: dict[str, object]) -> dict[str, object]:
     kept: Final = {key: value for key, value in tool.items() if key not in _ANTHROPIC_ONLY_TOOL_KEYS}
     function: Final = tool.get("function")
     if not isinstance(function, dict):
@@ -1193,7 +1193,7 @@ def _without_anthropic_only_tool_keys(tool: dict) -> dict:
     }
 
 
-def _drop_anthropic_only_tool_keys(tools: list[dict] | None) -> list[dict] | None:
+def _drop_anthropic_only_tool_keys(tools: list[dict[str, object]] | None) -> list[dict[str, object]] | None:
     if tools is None:
         return None
     return [_without_anthropic_only_tool_keys(tool) if isinstance(tool, dict) else tool for tool in tools]

litellm/proxy/_experimental/mcp_server/bridge_token_flow.py

@@ -86,8 +86,8 @@ async def oauth_authorization_uses_gateway_credential(request: Request) -> bool:
 
 
 async def _opaque_bearer_is_gateway_credential(token: str) -> bool:
-    from litellm.proxy._experimental.mcp_server.outbound_credentials.envelope import (
-        is_envelope,  # noqa: PLC0415  # envelope imports bridge types
+    from litellm.proxy._experimental.mcp_server.outbound_credentials.envelope import (  # noqa: PLC0415  # envelope imports bridge types
+        is_envelope,
         is_refresh_envelope,
     )
     from litellm.proxy._types import hash_token  # noqa: PLC0415  # proxy import cycle

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -2633,7 +2633,9 @@ def _build_aggregate_protected_resource_response(request: Request) -> dict:
     }
 
 
-def _build_aggregate_authorization_server_response(request: Request, token_exchange_available: bool) -> dict:
+def _build_aggregate_authorization_server_response(
+    request: Request, token_exchange_available: bool
+) -> dict[str, object]:
     """RFC 8414 metadata for the gateway as the aggregate authorization server.
 
     The issuer is ``{base}/mcp`` and must stay equal to the value the

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -3461,7 +3461,7 @@ class MCPServerManager:
                 passthrough_server_ids: Final = [
                     server.server_id
                     for server in self.get_registry().values()
-                    if getattr(server, "auth_type", None) == MCPAuth.true_passthrough
+                    if server.auth_type == MCPAuth.true_passthrough
                 ]
                 combined_servers.update(passthrough_server_ids)
 

litellm/proxy/_experimental/mcp_server/tool_search.py

@@ -128,14 +128,15 @@ def with_mcp_proxy_identity(tool: Tool, server_id: str) -> Tool:
 
 
 def _mcp_proxy_identity(tool: Tool) -> MCPProxyToolIdentity:
-    identity: Final = (tool.meta or {}).get(_MCP_PROXY_IDENTITY_META_KEY)  # mutable-ok: absent metadata default
+    identity: Final = None if tool.meta is None else tool.meta.get(_MCP_PROXY_IDENTITY_META_KEY)
     if not isinstance(identity, Mapping):
         raise TypeError("MCP proxy tool identity is missing")
     server_id: Final = identity.get("server_id")
     tool_name: Final = identity.get("tool_name")
     if not isinstance(server_id, str) or not isinstance(tool_name, str):
         raise TypeError("MCP proxy tool identity is invalid")
-    return {"server_id": server_id, "tool_name": tool_name}  # mutable-ok: TypedDict identity payload
+    resolved: Final[MCPProxyToolIdentity] = {"server_id": server_id, "tool_name": tool_name}
+    return resolved
 
 
 def mcp_proxy_tool_id(tool: Tool) -> str:

litellm/proxy/auth/auth_checks.py

@@ -4073,7 +4073,7 @@ async def get_org_object_for_request(
         )
     except OrganizationNotFoundError:
         return None
-    except Exception as e:  # noqa: BLE001  # only a DB outage may fail auth here, anything else degrades to no org limits
+    except Exception as e:
         if not PrismaDBExceptionHandler.is_database_service_unavailable_error_in_chain(e):
             verbose_proxy_logger.debug("org lookup failed, continuing without org limits", exc_info=True)
             return None

litellm/proxy/common_request_processing.py

@@ -2321,7 +2321,7 @@ class ProxyBaseLLMRequestProcessing:
         return fallbacks if isinstance(fallbacks, list) and fallbacks else None
 
     @staticmethod
-    def _resolve_fallback_models(model: str, fallbacks: list) -> list | None:
+    def _resolve_fallback_models(model: str, fallbacks: list) -> list[str] | None:
         from litellm.router_utils.fallback_event_handlers import get_fallback_model_group
 
         fallback_model_group, generic_fallback_idx = get_fallback_model_group(

litellm/proxy/db/baseline_accounting.py

@@ -486,12 +486,12 @@ class BaselineAccountingStore:
     async def _pages(
         self, db: SupportsRawQueries, scope: str, after_revision: int, withdraw_from: float | None = None
     ) -> AsyncIterator[tuple[_StoredRecord, ...]]:
-        cursor: float | None = None
+        cursor: float | None = None  # rebind-ok: keyset pagination advances after each complete timestamp group
         while page := _RECORDS.validate_python(
             tuple(await db.query_raw(_READ_PAGE, scope, after_revision, cursor, _PAGE_TIMESTAMPS, withdraw_from))
         ):
             yield page
-            cursor = page[-1].started_at  # rebind-ok: keyset pagination advances after each complete timestamp group
+            cursor = page[-1].started_at
 
     async def _withdraw(self, db: SupportsRawQueries, scope: str, started_at: float) -> None:
         async for page in self._pages(db, scope, 0, withdraw_from=started_at):
@@ -623,13 +623,13 @@ async def flush_baseline_accounting(client: PrismaClient) -> None:
     store: Final = BaselineAccountingStore.for_client(client)
     async with client.baseline_accounting_lock:
         batch: Final = tuple(client.baseline_accounting_transactions[:32])
-        client.baseline_accounting_transactions = client.baseline_accounting_transactions[
+        client.baseline_accounting_transactions = client.baseline_accounting_transactions[  # rebind-ok: drain under lock
             32:
-        ]  # rebind-ok: drain under lock
+        ]
         more_queued: Final = bool(client.baseline_accounting_transactions)
     try:
         remaining: Final = await asyncio.wait_for(_flush_records(store, batch), timeout=5)
-    except (Exception, asyncio.CancelledError) as error:  # noqa: BLE001  # unknown acknowledgements can be replayed safely
+    except (Exception, asyncio.CancelledError) as error:
         async with client.baseline_accounting_lock:
             client.baseline_accounting_transactions.extend(batch)
         if isinstance(error, asyncio.CancelledError):

litellm/proxy/guardrails/guardrail_initializers.py

@@ -120,7 +120,7 @@ def initialize_presidio(litellm_params: LitellmParams, guardrail: Guardrail) ->
         _OPTIONAL_PresidioPIIMasking,
     )
 
-    explicit_filter_scope: Final = getattr(litellm_params, "presidio_filter_scope", None)
+    explicit_filter_scope: Final = litellm_params.presidio_filter_scope
     filter_scope: Final = explicit_filter_scope or ("input" if _is_mcp_only_mode(litellm_params.mode) else "both")
     run_input: Final = filter_scope in ("input", "both")
     run_output: Final = filter_scope in ("output", "both")

litellm/proxy/hooks/autorouter_baseline_cache.py

@@ -230,12 +230,12 @@ class AutoRouterBaselineCache(CustomLogger):
 async def invalidate_baseline_cache(logging_obj: Logging, reason: str, *, completed: bool = False) -> None:
     context: Final = logging_obj.baseline_cache_context
     if context is not None:
-        logging_obj.baseline_cache_context = replace(
+        logging_obj.baseline_cache_context = replace(  # rebind-ok: request-owned retry marker
             context, invalidated=reason
-        )  # rebind-ok: request-owned retry marker
-        logging_obj.baseline_observation = context.capture.model_copy(
+        )
+        logging_obj.baseline_observation = context.capture.model_copy(  # rebind-ok: uncertain capture for failure logs
             update=MappingProxyType(
-                {  # rebind-ok: capture uncertainty for failure logging
+                {
                     "observation": context.capture.observation.model_copy(
                         update=MappingProxyType(
                             {

litellm/proxy/hooks/proxy_track_cost_callback.py

@@ -770,7 +770,7 @@ async def _reconcile_budget_reservation_before_db_update(
                 "Failed to invalidate budget reservation counters after pre-persist reconcile failed"
             )
         finally:
-            budget_reservation["finalized"] = True  # rebind-ok: the counter update reads the stamp off the shared dict
+            budget_reservation["finalized"] = True  # rebind-ok: stamps the caller's shared dict for the counter update
 
 
 async def _release_budget_reservation(budget_reservation: dict | None) -> None:

litellm/proxy/litellm_pre_call_utils.py

@@ -137,7 +137,7 @@ def add_otel_trace_id_to_request(
         return
     data["litellm_trace_id"] = trace_id  # rebind-ok: data is an out-param
     if isinstance(metadata, dict):
-        metadata["trace_id"] = trace_id  # rebind-ok: metadata is the request's own out-param dict
+        metadata["trace_id"] = trace_id
 
 
 def _session_id_from_baggage(baggage: str) -> str | None:

litellm/proxy/management_endpoints/common_utils.py

@@ -500,7 +500,7 @@ def _prisma_value(value: object) -> object:
     return list(value) if isinstance(value, tuple) else value
 
 
-def member_budget_patch(source: BaseModel) -> dict[str, Any]:
+def member_budget_patch(source: BaseModel) -> Mapping[str, object]:
     """Map the per-member limit fields a request actually set to their budget-table
     columns (merge-patch: a sent value updates, an explicit null clears, an absent
     field is left untouched)."""
@@ -533,7 +533,7 @@ async def _upsert_budget_and_membership(
     user_id: str,
     existing_budget_id: str | None,
     user_api_key_dict: UserAPIKeyAuth,
-    budget_patch: dict[str, Any],
+    budget_patch: Mapping[str, object],
     team_default_budget_id: str | None = None,
     shared_budget_ids: frozenset[str] | None = None,
 ):
@@ -596,9 +596,9 @@ async def _upsert_budget_and_membership(
         if is_shared_default and not temp_only
         else None
     )
-    source: Final[Mapping[str, Any]] = source_row.model_dump() if source_row is not None else MappingProxyType({})
+    source: Final[Mapping[str, object]] = source_row.model_dump() if source_row is not None else MappingProxyType({})
 
-    create_data: Final[dict[str, Any]] = {  # mutable-ok: Prisma create payloads are dict-shaped
+    create_data: Final[dict[str, object]] = {  # mutable-ok: Prisma create payloads are dict-shaped
         "created_by": user_api_key_dict.user_id or "",
         "updated_by": user_api_key_dict.user_id or "",
         **MappingProxyType(

litellm/proxy/management_helpers/bulk_user_creation.py

@@ -348,7 +348,7 @@ async def _prepare_user(user: _PendingUser, prisma_client: PrismaClient) -> _Pre
         data: Final = {**dumped, "user_id": user.user_id}  # mutable-ok: /user/new defaults helper mutates in place
         data_json: Final = _JSON_OBJECT.validate_python(_update_internal_new_user_params(data, user.request))
         with_permission: Final = _JSON_OBJECT.validate_python(
-            await _set_object_permission(data_json=data_json, prisma_client=prisma_client)  # pyright: ignore[reportUnknownArgumentType]  # validated by the adapter
+            await _set_object_permission(data_json=data_json, prisma_client=prisma_client)
         )
         return _PreparedUser(user, _USER_ROW.validate_python(with_permission))
     except Exception as exc:  # noqa: BLE001  # any preparation failure is reported on this row only
@@ -509,7 +509,7 @@ class _TeamsData(TypedDict):
 def _default_member_budget_id(team: LiteLLM_TeamTable) -> str | None:
     metadata: Final = (
         _JSON_OBJECT.validate_python(
-            team.metadata  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]  # LiteLLM_TeamTable.metadata is a bare dict; validated by the adapter
+            team.metadata  # pyright: ignore[reportUnknownMemberType]  # LiteLLM_TeamTable.metadata is a bare dict; validated by the adapter
         )
         if team.metadata  # pyright: ignore[reportUnknownMemberType]  # same bare dict
         else None
@@ -543,7 +543,7 @@ async def _write_team_roster(
             already_present: Final = frozenset(member.user_id for member in roster if member.user_id)
             new_members: Final = tuple(member for member in members if member.user_id not in already_present)
             budget_ids: Final = tuple(
-                [  # mutable-ok: budgets are created one at a time on the transaction's single connection
+                [
                     await _resolve_member_budget_id(
                         prisma_client=prisma_client,
                         user_api_key_dict=user_api_key_dict,

litellm/proxy/management_helpers/bulk_user_deletion.py

@@ -204,7 +204,7 @@ def _error_message(exc: BaseException) -> str:
     if isinstance(exc, HTTPException) and isinstance(exc.detail, dict):
         return str(exc.detail.get("error", exc.detail))  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType]  # HTTPException.detail is untyped
     if isinstance(exc, HTTPException):
-        return str(exc.detail)  # pyright: ignore[reportUnknownArgumentType]  # HTTPException.detail is untyped
+        return str(exc.detail)
     return str(exc) or type(exc).__name__
 
 

litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_passthrough_logging_handler.py

@@ -460,7 +460,7 @@ class VertexPassthroughLoggingHandler:
     @staticmethod
     def _is_audio_predict_response(
         model: str,
-        json_response: dict,  # mutable-ok: predicate inspects the decoded provider response dictionary without mutation
+        json_response: Mapping[str, object],
     ) -> bool:
         return (
             VertexPassthroughLoggingHandler._get_audio_prediction_count(json_response=json_response) > 0
@@ -469,7 +469,7 @@ class VertexPassthroughLoggingHandler:
 
     @staticmethod
     def _get_audio_prediction_count(
-        json_response: dict,  # mutable-ok: counter inspects the decoded provider response dictionary without mutation
+        json_response: Mapping[str, object],
     ) -> int:
         predictions: Final = json_response.get("predictions")
         if not isinstance(predictions, list):

litellm/proxy/spend_tracking/daily_global_spend_rollup.py

@@ -198,10 +198,6 @@ async def _scan_pending(prisma_client: "PrismaClient") -> _PendingScan:
     return _PendingScan(marker, db_now.now, tuple(_DateRow.model_validate(row).date for row in rows))
 
 
-async def pending_days(prisma_client: "PrismaClient") -> tuple[str, ...]:
-    return (await _scan_pending(prisma_client)).days
-
-
 async def reconcile_day(prisma_client: "PrismaClient", day: str) -> None:
     """Rewrite one day of the global table from the per-key sums. Idempotent: a rerun
     overwrites every group with the same totals."""

litellm/responses/litellm_completion_transformation/transformation.py

@@ -2254,7 +2254,7 @@ class LiteLLMCompletionResponsesConfig:
     ) -> Mapping[str, ResponseFunctionWebSearch]:
         calls: Final[dict[str, ResponseFunctionWebSearch]] = {}  # mutable-ok: indexes provider-built calls
         for choice in chat_completion_response.choices:
-            provider_fields = getattr(choice.message, "provider_specific_fields", None)
+            provider_fields = choice.message.provider_specific_fields
             if not isinstance(provider_fields, Mapping):
                 continue
             web_search_calls = provider_fields.get("web_search_calls")

litellm/responses/streaming_iterator.py

@@ -1360,7 +1360,7 @@ def _billed_terminal_response(
         return None
     usage: Final[object] = response_obj.get("usage")  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # a model_constructed terminal event leaves response as an untyped dict
     return ResponsesAPIResponse.model_construct(
-        **{**response_obj, "usage": usage if usage is not None or estimate is None else estimate()}  # pyright: ignore[reportUnknownArgumentType, reportArgumentType]  # same untyped dict spread
+        **{**response_obj, "usage": usage if usage is not None or estimate is None else estimate()}  # pyright: ignore[reportArgumentType]  # same untyped dict spread
     )
 
 

litellm/router_strategy/complexity_router/complexity_router.py

@@ -1093,7 +1093,7 @@ def _with_classifier_forecast(
     if forecast is None:
         return decision
     verdict: Final = forecast.verdict
-    enriched: Final[StandardLoggingRoutingDecision] = {  # mutable-ok: routing decisions are JSON TypedDict records
+    enriched: Final[StandardLoggingRoutingDecision] = {
         **decision,
         "classifier_crux": verdict.crux,
         "classifier_primary_rule": verdict.primary_rule,
@@ -2484,7 +2484,7 @@ class ComplexityRouter(CustomLogger):
             {"role": "user", "content": opening_task},  # mutable-ok: SDK messages are dict-shaped
         ]
         if latest_follow_up is not None:
-            task_messages.append(  # mutable-ok: the provider SDK requires a concrete message list
+            task_messages.append(
                 {"role": "user", "content": latest_follow_up}  # mutable-ok: SDK messages are dict-shaped
             )
 

litellm/router_strategy/complexity_router/jev_classifier.py

@@ -1,7 +1,7 @@
 from collections.abc import Mapping
 from datetime import datetime, timezone
 from types import MappingProxyType
-from typing import Annotated, Final, Literal, NamedTuple, Protocol
+from typing import Annotated, Final, Literal, NamedTuple, Protocol, TypeAlias
 from uuid import uuid4
 
 import httpx
@@ -24,7 +24,7 @@ from litellm.proxy.pass_through_endpoints.llm_provider_handlers.typesafe_passthr
 from litellm.router_strategy.complexity_router.config import DEFAULT_JEV_INSTRUCTIONS as _DEFAULT_JEV_INSTRUCTIONS
 from litellm.types.utils import AUTOROUTER_CLASSIFIER_CALL_ORIGIN
 
-JevProbability = Annotated[float, Field(ge=0.0, le=1.0)]
+JevProbability: TypeAlias = Annotated[float, Field(ge=0.0, le=1.0)]
 DEFAULT_JEV_INSTRUCTIONS: Final = _DEFAULT_JEV_INSTRUCTIONS
 
 

litellm/router_utils/pre_call_checks/encrypted_content_affinity_check.py

@@ -50,6 +50,7 @@ from litellm.exceptions import (
 from litellm.integrations.custom_logger import CustomLogger, Span
 from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
 from litellm.litellm_core_utils.prompt_templates.common_utils import (
+    anthropic_content_lists,
     encrypted_content_of_block,
     strip_encrypted_reasoning_from_messages,
 )
@@ -155,10 +156,7 @@ class EncryptedContentAffinityCheck(CustomLogger):
             return iter(())
         return (
             cast(Mapping[str, object], block)  # cast-ok: narrowed by isinstance
-            for message in cast(list[object], messages)  # cast-ok: narrowed by isinstance
-            if isinstance(message, Mapping)
-            for content in (cast(Mapping[str, object], message).get("content"),)  # cast-ok: narrowed by isinstance
-            if isinstance(content, list)
+            for content in anthropic_content_lists(cast(list[object], messages))  # cast-ok: narrowed by isinstance
             for block in cast(list[object], content)  # cast-ok: narrowed by isinstance
             if isinstance(block, Mapping)
         )

litellm/rust_bridge/dispatch.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass
-from typing import Final, Generic, TypeVar
+from typing import Final, Generic, TypeAlias, TypeVar
 
 from litellm.rust_bridge import catalog, runtime
 from litellm.rust_bridge.bindings import NativeBinding
@@ -10,11 +10,11 @@ from litellm.rust_bridge.catalog import Route, RouteContext, RouteRule, Rules
 from litellm.rust_bridge.configuration import Decision
 from litellm.rust_bridge.configuration import decision as rollout_decision
 
-RequestT = TypeVar("RequestT")
-NativeT = TypeVar("NativeT")
-ResultT = TypeVar("ResultT")
+RequestT: Final = TypeVar("RequestT")
+NativeT: Final = TypeVar("NativeT")
+ResultT: Final = TypeVar("ResultT")
 
-NativeHook = Callable[[RequestT, tuple[object, ...], Mapping[str, object]], ResultT]
+NativeHook: TypeAlias = Callable[[RequestT, tuple[object, ...], Mapping[str, object]], ResultT]
 
 
 def call_hook(

litellm/rust_bridge/response_metadata.py

@@ -1,10 +1,10 @@
-from typing import TypeVar
+from typing import Final, TypeVar
 
 from litellm.router_utils.add_retry_fallback_headers import (
     _add_headers_to_response,  # pyright: ignore[reportPrivateUsage]  # reuse the proxy's identity-preserving response metadata writer
 )
 
-ResultT = TypeVar("ResultT")
+ResultT: Final = TypeVar("ResultT")
 
 
 def mark_rust_response(response: ResultT) -> ResultT:

litellm/types/llms/bedrock.py

@@ -558,7 +558,6 @@ class AmazonTitanMultimodalEmbeddingResponse(TypedDict):
     message: str  # Specifies any errors that occur during generation.
 
 
-# TwelveLabs Marengo Embed types
 TWELVELABS_EMBEDDING_INPUT_TYPES = Literal["text", "image", "video", "audio"]
 TWELVELABS_EMBEDDING_OPTIONS = Literal["visual-text", "visual-image", "audio"]
 

litellm/types/utils.py

@@ -3864,7 +3864,7 @@ def without_server_derived_pricing(model_info: Mapping[str, Any]) -> Mapping[str
     )
 
 
-def echoed_cost_map_pricing_fields(model_info: Mapping[str, Any]) -> tuple[str, ...]:
+def echoed_cost_map_pricing_fields(model_info: Mapping[str, object]) -> tuple[str, ...]:
     """Pricing fields a stored ``model_info`` blob copied from a ``/model/info`` response.
 
     Only ``litellm.get_model_info`` emits ``key`` (the resolved cost-map entry), so a stored
@@ -3895,7 +3895,7 @@ def echoed_cost_map_fields(
     )
 
 
-def pricing_override_fields(*sources: Mapping[str, Any]) -> tuple[str, ...]:
+def pricing_override_fields(*sources: Mapping[str, object]) -> tuple[str, ...]:
     return tuple(
         sorted(
             frozenset(