scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-24 00:52:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 91

refactor(types): replace Any with proven types in 32 files

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-09-21 10:51:46 +00:00
parent e484a7c89c
commit 8795be0a65
32 changed files with 127 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/_logging.py
View file

@ -5,6 +5,7 @@ import logging
import os
import re
import sys
from collections.abc import Sequence
from datetime import datetime
from logging import Formatter
from typing import Any, Final, TextIO
@ -186,7 +187,8 @@ class SecretRedactionFilter(logging.Filter):
record.stack_info = _redact_string(record.stack_info) # rebind-ok: a Filter scrubs records in place
# Redact extra fields passed via logger.debug("msg", extra={...})
for key, value in list(record.__dict__.items()):
record_items: Final[Sequence[tuple[str, object]]] = list(record.__dict__.items())
for key, value in record_items:
if key in _STANDARD_RECORD_ATTRS:
continue
if isinstance(value, str):
@ -507,7 +509,7 @@ handler.addFilter(_secret_filter)
handler.addFilter(_correlation_filter)
def _try_parse_json_message(message: str) -> dict[str, Any] | None:
def _try_parse_json_message(message: str) -> dict[str, object] | None:
"""
Try to parse a log message as JSON. Returns parsed dict if valid, else None.
Handles messages that are entirely valid JSON (e.g. json.dumps output).
@ -585,7 +587,7 @@ class JsonFormatter(Formatter):
def format(self, record):
message_str: Final = record.getMessage()
json_record: Final[dict[str, Any]] = {
json_record: Final[dict[str, object]] = {
"message": message_str,
"level": record.levelname,
"timestamp": self.formatTime(record),

12
litellm/a2a_protocol/streaming_iterator.py
View file

@ -5,7 +5,7 @@ A2A Streaming Iterator with token tracking and logging support.
import asyncio
from collections.abc import AsyncIterator
from datetime import datetime
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Final
import litellm
from litellm._logging import verbose_logger
@ -15,7 +15,7 @@ from litellm.litellm_core_utils.asyncify import asyncify
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
if TYPE_CHECKING:
from a2a.types import SendStreamingMessageRequest, SendStreamingMessageResponse
from a2a.compat.v0_3.types import SendStreamingMessageRequest, SendStreamingMessageResponse
class A2AStreamingIterator:
@ -39,9 +39,9 @@ class A2AStreamingIterator:
self.start_time = datetime.now()
# Collect chunks for token counting
self.chunks: list[Any] = []
self.chunks: list[SendStreamingMessageResponse] = []
self.collected_text_parts: list[str] = []
self.final_chunk: Any | None = None
self.final_chunk: SendStreamingMessageResponse | None = None
def __aiter__(self):
return self
@ -69,7 +69,7 @@ class A2AStreamingIterator:
await self._handle_stream_complete()
raise
def _collect_text_from_chunk(self, chunk: Any) -> None:
def _collect_text_from_chunk(self, chunk: "SendStreamingMessageResponse") -> None:
"""Extract text from a streaming chunk and add to collected parts."""
try:
chunk_dict: Final = chunk.model_dump(mode="json", exclude_none=True) if hasattr(chunk, "model_dump") else {}
@ -79,7 +79,7 @@ class A2AStreamingIterator:
except Exception:
verbose_logger.debug("Failed to extract text from A2A streaming chunk")
def _is_completed_chunk(self, chunk: Any) -> bool:
def _is_completed_chunk(self, chunk: "SendStreamingMessageResponse") -> bool:
"""Check if chunk indicates stream completion."""
try:
chunk_dict: Final = chunk.model_dump(mode="json", exclude_none=True) if hasattr(chunk, "model_dump") else {}

4
litellm/integrations/focus/focus_logger.py
View file

@ -15,6 +15,8 @@ from .destinations import FocusTimeWindow
if TYPE_CHECKING:
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
from .export_engine import FocusExportEngine
else:
AsyncIOScheduler = Any
@ -111,7 +113,7 @@ class FocusLogger(CustomLogger):
"""Entry point for scheduler jobs to run export cycle with locking."""
from litellm.proxy.proxy_server import proxy_logging_obj
pod_lock_manager = None
pod_lock_manager: PodLockManager | None = None
if proxy_logging_obj is not None:
writer: Final = getattr(proxy_logging_obj, "db_spend_update_writer", None)
if writer is not None:

2
litellm/integrations/generic_prompt_management/generic_prompt_manager.py
View file

@ -58,7 +58,7 @@ class GenericPromptManager(CustomPromptManagement):
api_key: str | None = None,
timeout: int = 30,
prompt_id: str | None = None,
additional_provider_specific_query_params: dict[str, Any] | None = None,
additional_provider_specific_query_params: Mapping[str, object] | None = None,
**kwargs,
):
"""

17
litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py
View file

@ -21,7 +21,7 @@ from __future__ import annotations
import os
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Any, Final, Protocol
import litellm
from litellm._logging import verbose_proxy_logger
@ -35,6 +35,17 @@ 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:
@ -226,9 +237,9 @@ 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 = None
pod_lock_manager: _PodLockManager | None = None
if proxy_logging_obj is not None:
writer: Final = getattr(proxy_logging_obj, "db_spend_update_writer", 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)

7
litellm/integrations/otel/presets/agentops.py
View file

@ -9,9 +9,12 @@ this preset registers a custom exporter (``kind="agentops"``) that mints the JWT
worker thread, off any event loop — and caches it for the process lifetime.
"""
from collections.abc import Sequence
from typing import Any, Final
import httpx
from opentelemetry.sdk.trace import ReadableSpan
from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
@ -71,7 +74,7 @@ def agentops_preset(
)
def _build_agentops_exporter(spec: ExporterSpec) -> Any:
def _build_agentops_exporter(spec: ExporterSpec) -> SpanExporter:
"""Factory for the ``agentops`` exporter kind: a lazy-auth OTLP/HTTP exporter."""
from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
OTLPSpanExporter,
@ -106,7 +109,7 @@ def _build_agentops_exporter(spec: ExporterSpec) -> Any:
except Exception as e:
verbose_logger.debug("AgentOps JWT fetch failed: %s", e)
def export(self, spans: Any) -> Any:
def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
self._ensure_authenticated()
return super().export(spans)

13
litellm/integrations/otel/runtime.py
View file

@ -8,13 +8,16 @@ identity unconditionally.
"""
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from contextlib import AbstractContextManager, contextmanager
from functools import cache
from typing import Any, Final
from typing import TYPE_CHECKING, Final
if TYPE_CHECKING:
from opentelemetry.trace import Span
@cache
def _otel_runtime() -> "tuple[Callable[[str], Any], Callable[..., None]] | None":
def _otel_runtime() -> "tuple[Callable[[str], AbstractContextManager[Span | None]], Callable[..., None]] | None":
"""Resolve the SDK-backed hooks once and cache the outcome, absence included.
CPython never caches a failed import, so without this memoization every call
@ -29,7 +32,7 @@ def _otel_runtime() -> "tuple[Callable[[str], Any], Callable[..., None]] | None"
@contextmanager
def phase_span(name: str) -> "Iterator[Any]":
def phase_span(name: str) -> "Iterator[Span | None]":
"""Run a request phase inside a live active span so its DB/service calls nest.
Yields ``None`` (a plain no-op) when the OTel SDK is unavailable or V2 is not
@ -43,7 +46,7 @@ def phase_span(name: str) -> "Iterator[Any]":
yield span
def seed_request_identity(user_api_key_dict: Any, model: Any = None) -> None:
def seed_request_identity(user_api_key_dict: object, model: object = None) -> None:
"""Seed request-identity Baggage at the auth boundary (no-op without V2)."""
runtime: Final = _otel_runtime()
if runtime is None:

16
litellm/integrations/prometheus_helpers/bounded_prometheus_series_tracker.py
View file

@ -3,7 +3,13 @@ from __future__ import annotations
import time
from collections import OrderedDict
from threading import RLock
from typing import Any, Final
from typing import Final, Protocol
class _RemovableMetric(Protocol):
"""The one prometheus-client metric method this tracker calls."""
def remove(self, *labelvalues: object) -> None: ...
class BoundedPrometheusSeriesTracker:
@ -21,7 +27,7 @@ class BoundedPrometheusSeriesTracker:
def track_series(
self,
metric: Any,
metric: _RemovableMetric,
metric_name: str,
label_values: tuple[str | None, ...],
max_series: int | None,
@ -60,7 +66,7 @@ class BoundedPrometheusSeriesTracker:
break
del series[tracked_label_values]
def remove_series(self, metric: object, label_values: tuple[str | None, ...]) -> bool:
def remove_series(self, metric: _RemovableMetric, label_values: tuple[str | None, ...]) -> bool:
"""Drop one child series, True when it is gone (removed or never existed)."""
return self._remove_metric_child(metric, label_values)
@ -82,7 +88,7 @@ class BoundedPrometheusSeriesTracker:
def _remove_metric_series(
self,
metric: Any,
metric: _RemovableMetric,
series: OrderedDict[tuple[str | None, ...], float],
label_values: tuple[str | None, ...],
) -> None:
@ -90,7 +96,7 @@ class BoundedPrometheusSeriesTracker:
series.pop(label_values, None)
@staticmethod
def _remove_metric_child(metric: Any, label_values: tuple[str | None, ...]) -> bool:
def _remove_metric_child(metric: _RemovableMetric, label_values: tuple[str | None, ...]) -> bool:
"""
Remove the Prometheus child for ``label_values`` and report whether the
tracker should commit the matching state change.

2
litellm/integrations/vector_store_integrations/vector_store_pre_call_hook.py
View file

@ -406,7 +406,7 @@ class VectorStorePreCallHook(CustomLogger):
request_data: dict,
response_chunk: Any,
call_type: CallTypes | None,
) -> Any | None:
) -> object | None:
"""
Add search results to the final streaming chunk.

9
litellm/integrations/weights_biases.py
View file

@ -4,6 +4,7 @@ imported_openAIResponse = True
try:
import io
import logging
from collections.abc import Mapping
from typing import Any, Literal, Protocol, TypeVar
from wandb.sdk.data_types import trace_tree
@ -43,7 +44,7 @@ try:
@staticmethod
def results_to_trace_tree(
request: dict[str, Any],
request: Mapping[str, object],
response: OpenAIResponse,
results: list[trace_tree.Result],
time_elapsed: float,
@ -73,7 +74,7 @@ try:
def _resolve_edit(
self,
request: dict[str, Any],
request: Mapping[str, object],
response: OpenAIResponse,
time_elapsed: float,
) -> trace_tree.WBTraceTree:
@ -91,7 +92,7 @@ try:
def _resolve_completion(
self,
request: dict[str, Any],
request: Mapping[str, object],
response: OpenAIResponse,
time_elapsed: float,
) -> trace_tree.WBTraceTree:
@ -134,7 +135,7 @@ try:
def _request_response_result_to_trace(
self,
request: dict[str, Any],
request: Mapping[str, object],
response: OpenAIResponse,
request_str: str,
choices: list[str],

6
litellm/litellm_core_utils/llm_cost_calc/usage_object_transformation.py
View file

@ -50,7 +50,7 @@ _INTERACTIONS_MODALITY_FIELDS: Final[Mapping[str, str]] = MappingProxyType(
)
def _modality_field(entry: Mapping[str, Any]) -> str | None:
def _modality_field(entry: Mapping[str, object]) -> str | None:
return _INTERACTIONS_MODALITY_FIELDS.get(str(entry.get("modality", "")).lower())
@ -58,7 +58,7 @@ def _token_count(value: object) -> int:
return value if isinstance(value, int) else 0
def _modality_token_sums(entries: Sequence[Mapping[str, Any]]) -> Mapping[str, int]:
def _modality_token_sums(entries: Sequence[Mapping[str, object]]) -> Mapping[str, int]:
fields: Final = frozenset(field for entry in entries if (field := _modality_field(entry)) is not None)
return MappingProxyType(
{
@ -68,7 +68,7 @@ def _modality_token_sums(entries: Sequence[Mapping[str, Any]]) -> Mapping[str, i
)
def _google_search_query_count(usage_object: Mapping[str, Any]) -> int:
def _google_search_query_count(usage_object: Mapping[str, object]) -> int:
entries: Final = usage_object.get("grounding_tool_count")
if not isinstance(entries, Sequence):
return 0

2
litellm/litellm_core_utils/safe_json_dumps.py
View file

@ -85,7 +85,7 @@ def safe_json_structure(
def safe_dumps(
data: Any,
data: object,
max_depth: int = DEFAULT_MAX_RECURSE_DEPTH,
value_transform: Callable[[str | None, str], str] | None = None,
) -> str:

10
litellm/llms/azure/fine_tuning/handler.py
View file

@ -1,5 +1,5 @@
from collections.abc import Coroutine
from typing import Any, Final, cast
from typing import Final, cast
import httpx
from openai import AsyncAzureOpenAI, AsyncOpenAI, AzureOpenAI, OpenAI
@ -19,7 +19,7 @@ class AzureOpenAIFineTuningAPI(OpenAIFineTuningAPI, BaseAzureLLM):
"""
@staticmethod
def _ensure_training_type(create_fine_tuning_job_data: dict[str, Any]) -> None:
def _ensure_training_type(create_fine_tuning_job_data: dict[str, object]) -> None:
"""
Azure requires trainingType in extra_body. Default to 1 (supervised) if omitted.
"""
@ -66,7 +66,7 @@ class AzureOpenAIFineTuningAPI(OpenAIFineTuningAPI, BaseAzureLLM):
max_retries: int | None,
organization: str | None,
client: OpenAI | AsyncOpenAI | AzureOpenAI | AsyncAzureOpenAI | None = None,
) -> LiteLLMFineTuningJob | Coroutine[Any, Any, LiteLLMFineTuningJob]:
) -> LiteLLMFineTuningJob | Coroutine[object, object, LiteLLMFineTuningJob]:
self._ensure_training_type(create_fine_tuning_job_data)
openai_client: Final[OpenAI | AsyncOpenAI | AzureOpenAI | AsyncAzureOpenAI | None] = self.get_openai_client(
@ -109,7 +109,7 @@ class AzureOpenAIFineTuningAPI(OpenAIFineTuningAPI, BaseAzureLLM):
max_retries: int | None,
organization: str | None,
client: OpenAI | AsyncOpenAI | AzureOpenAI | AsyncAzureOpenAI | None = None,
) -> LiteLLMFineTuningJob | Coroutine[Any, Any, LiteLLMFineTuningJob]:
) -> LiteLLMFineTuningJob | Coroutine[object, object, LiteLLMFineTuningJob]:
openai_client: Final[OpenAI | AsyncOpenAI | AzureOpenAI | AsyncAzureOpenAI | None] = self.get_openai_client(
api_key=api_key,
api_base=api_base,
@ -149,7 +149,7 @@ class AzureOpenAIFineTuningAPI(OpenAIFineTuningAPI, BaseAzureLLM):
max_retries: int | None,
organization: str | None,
client: OpenAI | AsyncOpenAI | AzureOpenAI | AsyncAzureOpenAI | None = None,
) -> LiteLLMFineTuningJob | Coroutine[Any, Any, LiteLLMFineTuningJob]:
) -> LiteLLMFineTuningJob | Coroutine[object, object, LiteLLMFineTuningJob]:
openai_client: Final[OpenAI | AsyncOpenAI | AzureOpenAI | AsyncAzureOpenAI | None] = self.get_openai_client(
api_key=api_key,
api_base=api_base,

2
litellm/llms/codestral/completion/transformation.py
View file

@ -29,7 +29,7 @@ class CodestralTextCompletionConfig(OpenAITextCompletionConfig):
random_seed: int | None = None,
stop: str | None = None,
) -> None:
locals_: Final = locals().copy()
locals_: Final[dict[str, object]] = locals().copy()
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)

4
litellm/llms/databricks/common_utils.py
View file

@ -12,7 +12,7 @@ Authentication priority:
import os
import re
from typing import Any, Final, Literal
from typing import Final, Literal
from urllib.parse import urlsplit, urlunsplit
from litellm.llms.base_llm.chat.transformation import BaseLLMException
@ -48,7 +48,7 @@ class DatabricksBase:
]
@classmethod
def redact_sensitive_data(cls, data: Any) -> Any:
def redact_sensitive_data(cls, data: object) -> object:
"""
Redact sensitive information (tokens, secrets) from data before logging.

4
litellm/llms/gemini/realtime/transformation.py
View file

@ -453,7 +453,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
return normalized
@staticmethod
def _finalize_gemini_live_setup(model: str, setup: dict[str, Any]) -> dict[str, Any]:
def _finalize_gemini_live_setup(model: str, setup: dict[str, object]) -> dict[str, object]:
generation_config: Final = setup.get("generationConfig")
if isinstance(generation_config, dict):
modalities: Final = generation_config.get("responseModalities")
@ -1172,7 +1172,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
def map_openai_event(
self,
key: str,
value: Any,
value: object,
current_delta_type: ALL_DELTA_TYPES | None,
) -> OpenAIRealtimeEventTypes | ResponsesAPIStreamEvents:
if isinstance(value, dict):

2
litellm/llms/jina_ai/embedding/transformation.py
View file

@ -31,7 +31,7 @@ class JinaAIEmbeddingConfig(BaseEmbeddingConfig):
def __init__(
self,
) -> None:
locals_: Final = locals().copy()
locals_: Final[dict[str, object]] = locals().copy()
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)

4
litellm/llms/openrouter/embedding/transformation.py
View file

@ -170,7 +170,9 @@ class OpenrouterEmbeddingConfig(BaseEmbeddingConfig):
optional_params[param] = value
return optional_params
def get_error_class(self, error_message: str, status_code: int, headers: Any) -> Any:
def get_error_class(
self, error_message: str, status_code: int, headers: dict[str, str] | httpx.Headers
) -> OpenRouterException:
"""
Get the error class for OpenRouter errors.
"""

4
litellm/llms/reducto/common.py
View file

@ -3,6 +3,8 @@ import binascii
from collections import defaultdict
from typing import TYPE_CHECKING, Any, Final, NoReturn
import httpx
from litellm.constants import request_timeout
REDUCTO_API_BASE: Final = "https://platform.reducto.ai"
@ -62,7 +64,7 @@ def extract_file_id_or_bytes(
return None, raw_bytes, mime
def _extract_file_id_from_upload_response(response: Any) -> str:
def _extract_file_id_from_upload_response(response: httpx.Response) -> str:
try:
payload: Final = response.json()
except ValueError as exc:

3
litellm/llms/vercel_ai_gateway/embedding/transformation.py
View file

@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.embedding.transformation import BaseEmbeddingConfig
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllEmbeddingInputValues
@ -160,7 +161,7 @@ class VercelAIGatewayEmbeddingConfig(BaseEmbeddingConfig):
optional_params[param] = value
return optional_params
def get_error_class(self, error_message: str, status_code: int, headers: Any) -> Any:
def get_error_class(self, error_message: str, status_code: int, headers: Any) -> BaseLLMException:
"""
Get the error class for Vercel AI Gateway errors.
"""

2
litellm/llms/vertex_ai/agent_engine/transformation.py
View file

@ -205,7 +205,7 @@ class VertexAgentEngineConfig(BaseConfig, VertexBase):
session_id: Final = self._get_session_id(optional_params)
# Build the input
input_data: Final[dict[str, Any]] = {
input_data: Final[dict[str, str]] = {
"message": prompt,
"user_id": user_id,
}

7
litellm/proxy/a2a/discovery.py
View file

@ -14,6 +14,7 @@ fetcher dispatches by ``discovery_mode``:
pure-A2A fallback strategy returns 404 for these deployments.
"""
from collections.abc import Mapping
from enum import Enum
from typing import Any, Final
from urllib.parse import urlencode
@ -55,7 +56,7 @@ def _normalize_base_url(base_url: str) -> str:
def _build_langgraph_platform_paths(
params: dict[str, Any] | None,
params: Mapping[str, object] | None,
) -> tuple[str, ...]:
"""Build the paths to try for LangGraph Platform discovery.
@ -71,7 +72,7 @@ def _build_langgraph_platform_paths(
return tuple(f"{path}?{query}" for path in AGENT_CARD_WELL_KNOWN_PATHS)
def _paths_for_mode(mode: DiscoveryMode, params: dict[str, Any] | None) -> tuple[str, ...]:
def _paths_for_mode(mode: DiscoveryMode, params: Mapping[str, object] | None) -> tuple[str, ...]:
if mode == DiscoveryMode.WELL_KNOWN_FALLBACK:
return AGENT_CARD_WELL_KNOWN_PATHS
if mode == DiscoveryMode.LANGGRAPH_PLATFORM:
@ -83,7 +84,7 @@ async def fetch_well_known_card(
base_url: str,
*,
discovery_mode: DiscoveryMode = DiscoveryMode.WELL_KNOWN_FALLBACK,
params: dict[str, Any] | None = None,
params: Mapping[str, object] | None = None,
timeout: float = DEFAULT_DISCOVERY_TIMEOUT_SECONDS,
headers: dict[str, str] | None = None,
) -> dict[str, Any]:

11
litellm/proxy/agent_endpoints/databricks_oauth.py
View file

@ -25,8 +25,9 @@ Config example::
import asyncio
import base64
import hashlib
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Final
from typing import Final
import httpx
@ -43,7 +44,7 @@ _TOKEN_EXPIRY_BUFFER_SECONDS: Final = 60
_DEFAULT_TTL_SECONDS: Final = 3600
def _resolve_secret(value: Any) -> str | None:
def _resolve_secret(value: object) -> str | None:
"""Resolve a config value, expanding ``os.environ/`` references."""
if not isinstance(value, str):
return None
@ -75,7 +76,7 @@ class DatabricksAppOAuthConfig:
def parse_databricks_oauth_config(
litellm_params: dict[str, Any] | None,
litellm_params: Mapping[str, object] | None,
) -> DatabricksAppOAuthConfig | None:
"""Build a Databricks App OAuth config from an agent's ``litellm_params``.
@ -191,7 +192,7 @@ class DatabricksAppOAuthTokenCache(InMemoryCache):
except httpx.HTTPError as exc:
raise ValueError(f"Databricks App OAuth token request failed: {exc}") from exc
body: Final = response.json()
body: Final[object] = response.json()
if not isinstance(body, dict):
raise ValueError(
f"Databricks App OAuth token response returned non-object JSON (got {type(body).__name__})"
@ -215,7 +216,7 @@ databricks_app_oauth_token_cache: Final = DatabricksAppOAuthTokenCache()
async def resolve_databricks_app_auth_header(
litellm_params: dict[str, Any] | None,
litellm_params: Mapping[str, object] | None,
) -> dict[str, str] | None:
"""Return ``{"Authorization": "Bearer <token>"}`` for a Databricks App agent.

13
litellm/proxy/client/cli/main.py
View file

@ -9,7 +9,15 @@ from litellm._version import version as litellm_version
from litellm.proxy.client.health import HealthManagementClient
from .commands.agents import agent_commands
from .commands.auth import auth_group, context_secret_vault, get_stored_api_key, login, logout, whoami
from .commands.auth import (
CliContextObj,
auth_group,
context_secret_vault,
get_stored_api_key,
login,
logout,
whoami,
)
from .commands.autoroute.commands import autoroute_group
from .commands.chat import chat
from .commands.config import config_commands, get_config_value, hidden_command_names
@ -126,7 +134,8 @@ def cli(ctx: click.Context, show_version: bool, base_url: str | None, api_key: s
@click.pass_context
def version(ctx: click.Context):
"""Show the LiteLLM Proxy CLI and server version."""
print_version(ctx.obj.get("base_url"), ctx.obj.get("api_key"))
ctx_obj: Final[CliContextObj] = ctx.obj
print_version(ctx_obj.get("base_url"), ctx_obj.get("api_key"))
# Add authentication commands as top-level commands

2
litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/__init__.py
View file

@ -8,7 +8,7 @@ if TYPE_CHECKING:
from litellm.types.guardrails import Guardrail, LitellmParams
def _get_config_value(litellm_params: Any, optional_params: Any, attribute_name: str) -> Any | None:
def _get_config_value(litellm_params: "LitellmParams", optional_params: object, attribute_name: str) -> Any | None:
if optional_params is not None:
value: Final = (
optional_params.get(attribute_name)

4
litellm/proxy/guardrails/guardrail_hooks/onyx/onyx.py
View file

@ -6,7 +6,7 @@
# +-------------------------------------------------------------+
import os
import uuid
from typing import TYPE_CHECKING, Any, Final, Literal, Optional
from typing import TYPE_CHECKING, Final, Literal, Optional
import httpx
from fastapi import HTTPException
@ -63,7 +63,7 @@ class OnyxGuardrail(CustomGuardrail):
async def _validate_with_guard_server(
self,
payload: Any,
payload: object,
input_type: Literal["request", "response"],
conversation_id: str,
) -> dict:

6
litellm/proxy/hooks/responses_id_security.py
View file

@ -40,7 +40,7 @@ _UNMANAGED_RESPONSE_ID_DETAIL: Final = (
_PROXY_ADMIN_ROLES: Final = frozenset({LitellmUserRoles.PROXY_ADMIN, LitellmUserRoles.PROXY_ADMIN.value})
def _proxy_general_settings() -> Mapping[str, Any]:
def _proxy_general_settings() -> Mapping[str, object]:
from litellm.proxy.proxy_server import general_settings
return general_settings
@ -107,7 +107,7 @@ def _is_responses_api_create_route(request_route: str | None) -> bool:
class ResponsesIDSecurity(CustomLogger):
def __init__(
self,
general_settings_reader: Callable[[], Mapping[str, Any]] = _proxy_general_settings,
general_settings_reader: Callable[[], Mapping[str, object]] = _proxy_general_settings,
signing_key_reader: Callable[[], str | None] = _proxy_signing_key,
) -> None:
self._general_settings_reader: Final = general_settings_reader
@ -307,7 +307,7 @@ class ResponsesIDSecurity(CustomLogger):
data: dict,
user_api_key_dict: "UserAPIKeyAuth",
response: LLMResponseTypes,
) -> Any:
) -> LLMResponseTypes:
"""
Queue response IDs for batch processing instead of writing directly to DB.

5
litellm/proxy/logging_endpoints/callback_logs_endpoints.py
View file

@ -15,6 +15,7 @@ self-describing `StandardLoggingPayload`, so completions/responses can use it to
"""
import uuid
from collections.abc import Mapping
from datetime import datetime, timezone
from typing import Any, Final
@ -48,7 +49,7 @@ class CallbackLogsReplayer:
"""
@staticmethod
def _epoch_to_datetime(value: Any) -> datetime:
def _epoch_to_datetime(value: object) -> datetime:
"""`StandardLoggingPayload` stores startTime/endTime as float epoch seconds."""
if isinstance(value, (int, float)):
return datetime.fromtimestamp(float(value), tz=timezone.utc)
@ -114,7 +115,7 @@ class CallbackLogsReplayer:
return logging_obj
@staticmethod
def _response_obj_from_payload(payload: dict[str, Any]) -> dict[str, Any]:
def _response_obj_from_payload(payload: Mapping[str, object]) -> dict[str, object]:
"""Minimal response object so usage-derived spend-log fields resolve."""
return {
"id": payload.get("id"),

8
litellm/proxy/management_endpoints/management_v1/spend_logs.py
View file

@ -1,7 +1,7 @@
"""`/management/v1/spend_logs` facets."""
from datetime import datetime, timezone
from typing import Annotated, Any, Final, Literal
from typing import Annotated, Final, Literal
from fastapi import APIRouter, Depends, Query, Request
@ -39,7 +39,7 @@ async def _spend_log_scope_clause(
user_api_key_dict: UserAPIKeyAuth,
prisma_client: PrismaClient,
next_param_index: int,
) -> tuple[str | None, tuple[Any, ...]]:
) -> tuple[str | None, tuple[str | list[str], ...]]:
"""SQL predicate restricting the facet to spend logs this caller may read.
Returns ``(None, ())`` for a proxy admin. Mirrors the scoping ``/spend/logs/ui``
@ -101,8 +101,8 @@ async def _list_spend_log_facet(
)
column_sql: Final = "end_user" if column == "end_user" else '"user"'
window_params: Final[tuple[Any, ...]] = (_as_utc(start_time), _as_utc(end_time))
search_params: Final[tuple[Any, ...]] = (f"%{escape_like(q)}%",) if q else ()
window_params: Final[tuple[datetime, datetime]] = (_as_utc(start_time), _as_utc(end_time))
search_params: Final[tuple[str, ...]] = (f"%{escape_like(q)}%",) if q else ()
search_clause: Final = (f"{column_sql} ILIKE ${len(window_params) + 1} ESCAPE '\\'",) if q else ()
scope_clause, scope_params = await _spend_log_scope_clause(

4
litellm/proxy/management_helpers/object_permission_utils.py
View file

@ -8,7 +8,7 @@ from collections.abc import Mapping, Sequence
from collections.abc import Set as AbstractSet
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Optional
from typing import TYPE_CHECKING, Final, Optional
from fastapi import HTTPException, status
from pydantic import TypeAdapter
@ -230,7 +230,7 @@ def _dedupe_preserving_order(values: list[str]) -> list[str]:
return result
def _mcp_server_identifier_matches(server: Any, identifier: str) -> bool:
def _mcp_server_identifier_matches(server: object, identifier: str) -> bool:
return identifier in {
getattr(server, "server_id", None),
getattr(server, "alias", None),

2
litellm/proxy/pass_through_endpoints/llm_provider_handlers/gemini_passthrough_logging_handler.py
View file

@ -147,7 +147,7 @@ class GeminiPassthroughLoggingHandler:
- Creates standard logging object
- Logs in litellm callbacks
"""
kwargs: dict[str, Any] = {}
kwargs: dict[str, object] = {}
model = model or GeminiPassthroughLoggingHandler.extract_model_from_url(url_route)
complete_streaming_response: Final = GeminiPassthroughLoggingHandler._build_complete_streaming_response(
all_chunks=all_chunks,

10
litellm/proxy/public_endpoints/public_endpoints.py
View file

@ -199,9 +199,13 @@ def _build_endpoints(raw: _ProvidersFile) -> list[_EndpointEntry]:
return result
_PROVIDERS_FILE_ADAPTER: Final = TypeAdapter(_ProvidersFile)
_PROVIDER_CREATE_FIELDS_ADAPTER: Final = TypeAdapter(list[ProviderCreateInfo])
def _load_endpoints() -> list[_EndpointEntry]:
raw: Final[_ProvidersFile] = json.loads(
files("litellm").joinpath("provider_endpoints_support_backup.json").read_text(encoding="utf-8")
raw: Final = _PROVIDERS_FILE_ADAPTER.validate_python(
json.loads(files("litellm").joinpath("provider_endpoints_support_backup.json").read_text(encoding="utf-8"))
)
return _build_endpoints(raw)
@ -398,7 +402,7 @@ async def get_provider_fields() -> list[ProviderCreateInfo]:
)
with open(provider_create_fields_path, "r") as f:
provider_create_fields: Final = json.load(f)
provider_create_fields: Final = _PROVIDER_CREATE_FIELDS_ADAPTER.validate_python(json.load(f))
return provider_create_fields