chore(techdebt): clear fresh tech debt from the last 24 hours (rolling, 2026-09-06 to 2026-09-24)

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/arize/arize.py

@@ -112,7 +112,7 @@ def _sampling_rate_for_request(self, kwargs: Mapping[str, object], var: str) ->
         if value is None or value in ("", "None"):
             return None
         try:
-            rate = float(value)
+            rate: Final = float(value)
         except (TypeError, ValueError):
             verbose_logger.warning(
                 "ArizeLogger: %s value %r is not a number; exporting the request",

litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py

@@ -21,7 +21,7 @@
 
 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 @@
     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 @@ async def initialize_mavvrik_focus_export_job(self) -> None:
         """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 @@ async def _build_chat_completion_request_patch(
         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

@@ -5191,7 +5191,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
@@ -6663,7 +6663,7 @@ 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",

litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py

@@ -5,7 +5,12 @@
 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 @@
 _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 @@ def _dispatch_provider_chunk(
                 "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

@@ -685,7 +685,7 @@ async def process_input_messages(
 
         return data
 
-    def _hoisted_top_level_system_message(self, data: dict) -> AllMessageValues | None:
+    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 @@
     Final,
     Literal,
     Protocol,
-    cast,  # noqa: TID251  # rebuilt message_delta dict spans the ContentBlockDelta/MessageBlockDelta union
     get_args,
 )
 
@@ -27,6 +26,7 @@
     ContentBlockDelta,
     ContextManagementResponse,
     MessageBlockDelta,
+    MessageDelta,
     StreamingContentBlockDeltaType,
     UsageDelta,
     UsageIteration,
@@ -1028,26 +1028,22 @@ def _with_refusal_stop_details(
         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] = {
+            **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

@@ -169,7 +169,7 @@ def _thinking_block_from_reasoning_item(
         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
@@ -198,7 +198,7 @@ def _assistant_block_group_key(indexed_block: tuple[int, Mapping[str, object]])
     @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/openai/responses/guardrail_translation/handler.py

@@ -994,7 +994,7 @@ async def process_output_streaming_response(
 
     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

@@ -7,6 +7,7 @@
 Docs: https://vercel.com/docs/ai-gateway/openai-compat/embeddings
 """
 
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any, Final
 
 import httpx
@@ -161,12 +162,14 @@ def map_openai_params(
                 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: Mapping[str, str] | httpx.Headers
+    ) -> BaseLLMException:
         """
         Get the error class for Vercel AI Gateway errors.
         """
         return VercelAIGatewayException(
             message=error_message,
             status_code=status_code,
-            headers=headers,
+            headers=headers if isinstance(headers, httpx.Headers) else httpx.Headers(headers),
         )

litellm/llms/vertex_ai/gemini/vertex_and_google_ai_studio_gemini.py

@@ -286,7 +286,7 @@ def _is_gemini_3_or_newer(model: str) -> bool:
         """
         Check if the model is Gemini 3 or newer.
         """
-        model_name = model.split("/")[-1].lower()
+        model_name: Final = model.split("/")[-1].lower()
         is_vertex_fine_tuned_model: Final = model_name.isdigit() or (
             model.startswith("gemini/") and not model_name.startswith("gemini-")
         )

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/passthrough/main.py

@@ -540,7 +540,9 @@ def llm_passthrough_route(
     )
 
     ## IS STREAMING REQUEST
-    _streaming_request_data: dict = data if isinstance(data, dict) else (json if isinstance(json, dict) else {})
+    _streaming_request_data: Final[dict[str, object]] = (
+        data if isinstance(data, dict) else (json if isinstance(json, dict) else {})
+    )
     is_streaming_request: Final = provider_config.is_streaming_request(
         endpoint=endpoint,
         request_data=_streaming_request_data,

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

@@ -2634,7 +2634,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

@@ -3490,7 +3490,7 @@ async def get_allowed_mcp_servers(
                 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