scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-08 22:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

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

Browse source
This commit is contained in:
mateo-berri 2026-09-02 23:08:42 +00:00
parent 362fb4cffe
commit 459858829e
89 changed files with 594 additions and 418 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
enterprise/litellm_enterprise/proxy/hooks/managed_vector_stores.py
View file

@ -16,6 +16,7 @@ from litellm.llms.base_llm.managed_resources.utils import (
is_base64_encoded_unified_id,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.utils import LLMResponseTypes
from litellm.types.vector_stores import (
VectorStoreCreateOptionalRequestParams,
VectorStoreCreateResponse,
@ -24,6 +25,7 @@ from litellm.types.vector_stores import (
if TYPE_CHECKING:
from opentelemetry.trace import Span as _Span
from litellm.caching.caching import DualCache
from litellm.proxy.utils import InternalUsageCache as _InternalUsageCache
from litellm.proxy.utils import PrismaClient as _PrismaClient
@ -156,7 +158,7 @@ class _PROXY_LiteLLMManagedVectorStores(
# Create vector store for each model
# Convert TypedDict to Dict[str, Any] for base class compatibility
request_data_dict: Dict[str, Any] = dict(create_request)
request_data_dict: Dict[str, object] = dict(create_request)
responses = await self.create_resource_for_each_model(
llm_router=llm_router,
request_data=request_data_dict,
@ -209,7 +211,7 @@ class _PROXY_LiteLLMManagedVectorStores(
limit: Optional[int] = None,
after: Optional[str] = None,
order: Optional[str] = None,
) -> Dict[str, Any]:
) -> Dict[str, object]:
"""
List vector stores created by a user.
@ -301,7 +303,7 @@ class _PROXY_LiteLLMManagedVectorStores(
async def async_pre_call_hook(
self,
user_api_key_dict: UserAPIKeyAuth,
cache: Any,
cache: "DualCache",
data: Dict,
call_type: str,
) -> Union[Exception, str, Dict, None]:
@ -403,8 +405,8 @@ class _PROXY_LiteLLMManagedVectorStores(
self,
data: Dict,
user_api_key_dict: UserAPIKeyAuth,
response: Any,
) -> Any:
response: LLMResponseTypes,
) -> LLMResponseTypes:
"""
Post-call hook to transform responses.

17
litellm/_redis_credential_provider.py
View file

@ -1,7 +1,7 @@
import asyncio
import threading
import time
from typing import Any, Final
from typing import Final, Protocol
from redis.credentials import CredentialProvider
@ -18,6 +18,19 @@ _token_cache: Final[dict[str, tuple[str, float]]] = {}
_token_cache_lock: Final = threading.Lock()
class AzureAccessToken(Protocol):
"""The ``azure.core.credentials.AccessToken`` shape this module reads."""
@property
def token(self) -> str: ...
class AzureCredential(Protocol):
"""The ``azure-identity`` credential surface this module calls."""
def get_token(self, *scopes: str) -> AzureAccessToken: ...
def _generate_gcp_iam_access_token(service_account: str) -> str:
"""
Generate GCP IAM access token for Redis authentication.
@ -115,7 +128,7 @@ class AzureADCredentialProvider(CredentialProvider):
fail authentication after the initial token expired (~1 hour TTL).
"""
def __init__(self, credential: Any, username: str | None = None) -> None:
def __init__(self, credential: AzureCredential, username: str | None = None) -> None:
self._credential = credential
self._username = username

46
litellm/_service_logger.py
View file

@ -1,6 +1,6 @@
import asyncio
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Any, Final, Protocol
import litellm
from litellm._logging import verbose_logger
@ -24,7 +24,30 @@ else:
UserAPIKeyAuth = Any
def _get_otel_v2_class() -> type | None:
class _ServiceSpanLogger(Protocol):
"""The OTel logger surface this module drives: the two service-span hooks it calls."""
async def async_service_success_hook(
self,
payload: ServiceLoggerPayload,
parent_otel_span: Span | None = None,
start_time: datetime | float | None = None,
end_time: datetime | float | None = None,
event_metadata: dict | None = None,
) -> None: ...
async def async_service_failure_hook(
self,
payload: ServiceLoggerPayload,
error: str | None = "",
parent_otel_span: Span | None = None,
start_time: datetime | float | None = None,
end_time: datetime | float | None = None,
event_metadata: dict | None = None,
) -> None: ...
def _get_otel_v2_class() -> type[_ServiceSpanLogger] | None:
"""Return the ``OpenTelemetryV2`` class, or ``None`` if the OTel SDK is absent.
Imported lazily: ``litellm.integrations.otel.logger`` imports the OpenTelemetry
@ -54,7 +77,7 @@ class ServiceLogging(CustomLogger):
if "prometheus_system" in litellm.service_callback:
self.prometheusServicesLogger = PrometheusServicesLogger()
def _resolve_otel_service_logger(self, callback: Any) -> Any | None:
def _resolve_otel_service_logger(self, callback: object) -> _ServiceSpanLogger | None:
"""Resolve the OTel logger (legacy or V2) to emit a service span on.
Returns the logger instance whose ``async_service_*_hook`` should fire for
@ -69,18 +92,21 @@ class ServiceLogging(CustomLogger):
"""
otel_v2_cls: Final = _get_otel_v2_class()
def _is_otel_logger(obj: Any) -> bool:
def _as_otel_logger(obj: object) -> _ServiceSpanLogger | None:
if isinstance(obj, OpenTelemetry):
return True
return otel_v2_cls is not None and isinstance(obj, otel_v2_cls)
return obj
if otel_v2_cls is not None and isinstance(obj, otel_v2_cls):
return obj
return None
if _is_otel_logger(callback):
return callback
resolved_callback: Final = _as_otel_logger(callback)
if resolved_callback is not None:
return resolved_callback
if callback == "otel":
from litellm.proxy.proxy_server import open_telemetry_logger
if open_telemetry_logger is not None and _is_otel_logger(open_telemetry_logger):
return open_telemetry_logger
if open_telemetry_logger is not None:
return _as_otel_logger(open_telemetry_logger)
return None
def service_success_hook(

3
litellm/a2a_protocol/card_resolver.py
View file

@ -4,6 +4,7 @@ Custom A2A Card Resolver for LiteLLM.
Extends the A2A SDK's card resolver to support multiple well-known paths.
"""
from collections.abc import Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final
@ -152,7 +153,7 @@ class LiteLLMA2ACardResolver(_A2ACardResolver):
async def get_agent_card(
self,
relative_card_path: str | None = None,
http_kwargs: dict[str, Any] | None = None,
http_kwargs: Mapping[str, object] | None = None,
) -> "AgentCard":
"""
Fetch the agent card, trying multiple well-known paths.

14
litellm/a2a_protocol/providers/watsonx_orchestrate/transformation.py
View file

@ -8,7 +8,7 @@ WXO uses a REST API (not A2A/JSON-RPC) with an async-poll execution model:
"""
import asyncio
from collections.abc import AsyncIterator
from collections.abc import AsyncIterator, Mapping
from typing import Any, Final
from uuid import uuid4
@ -51,9 +51,9 @@ class WatsonxOrchestrateTransformation:
wxo_agent_id: str,
text: str,
thread_id: str | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""Build the WXO POST /v1/orchestrate/runs request body."""
body: Final[dict[str, Any]] = {
body: Final[dict[str, object]] = {
"agent_id": wxo_agent_id,
"message": {
"role": "user",
@ -70,7 +70,7 @@ class WatsonxOrchestrateTransformation:
return body
@staticmethod
def extract_text_from_wxo_result(result: Any) -> str:
def extract_text_from_wxo_result(result: object) -> str:
"""
Extract response text from a WXO run result.
@ -103,7 +103,7 @@ class WatsonxOrchestrateTransformation:
return ""
@staticmethod
def extract_text_from_a2a_message_response(a2a_response: dict[str, Any]) -> str:
def extract_text_from_a2a_message_response(a2a_response: Mapping[str, object]) -> str:
result: Final = a2a_response.get("result")
if not isinstance(result, dict):
verbose_logger.warning("WXO: A2A response missing result object")
@ -119,7 +119,7 @@ class WatsonxOrchestrateTransformation:
return ""
@staticmethod
def build_a2a_message_response(request_id: str, text: str) -> dict[str, Any]:
def build_a2a_message_response(request_id: str, text: str) -> dict[str, object]:
"""
Build a standard A2A non-streaming SendMessageResponse (kind=message).
"""
@ -140,7 +140,7 @@ class WatsonxOrchestrateTransformation:
request_id: str,
chunk_size: int = 50,
delay_ms: int = 10,
) -> AsyncIterator[dict[str, Any]]:
) -> AsyncIterator[dict[str, object]]:
"""
Emit standard A2A streaming events from a completed text response.

45
litellm/assistants/utils.py
View file

@ -1,3 +1,4 @@
from collections.abc import Mapping, Sequence
from typing import Final
import litellm
@ -10,20 +11,22 @@ def get_optional_params_add_message(
role: str | None,
content: str | list[MessageContentTextObject | MessageContentImageFileObject | MessageContentImageURLObject] | None,
attachments: list[Attachment] | None,
metadata: dict | None,
metadata: Mapping[str, object] | None,
custom_llm_provider: str,
**kwargs,
):
**kwargs: object,
) -> dict[str, object]:
"""
Azure doesn't support 'attachments' for creating a message
Reference - https://learn.microsoft.com/en-us/azure/ai-services/openai/assistants-reference-messages?tabs=python#create-message
"""
passed_params: Final = locals()
custom_llm_provider = passed_params.pop("custom_llm_provider")
special_params: Final = passed_params.pop("kwargs")
for k, v in special_params.items():
passed_params[k] = v
passed_params: Final[Mapping[str, object]] = {
"role": role,
"content": content,
"attachments": attachments,
"metadata": metadata,
**kwargs,
}
default_params: Final = {
"role": None,
@ -33,10 +36,10 @@ def get_optional_params_add_message(
}
non_default_params = {k: v for k, v in passed_params.items() if (k in default_params and v != default_params[k])}
optional_params = {}
optional_params: dict[str, object] = {}
## raise exception if non-default value passed for non-openai/azure embedding calls
def _check_valid_arg(supported_params):
def _check_valid_arg(supported_params: Sequence[str]) -> Mapping[str, object] | None:
if len(non_default_params.keys()) > 0:
keys: Final = list(non_default_params.keys())
for k in keys:
@ -71,14 +74,18 @@ def get_optional_params_image_gen(
style: str | None = None,
user: str | None = None,
custom_llm_provider: str | None = None,
**kwargs,
):
**kwargs: object,
) -> dict[str, object]:
# retrieve all parameters passed to the function
passed_params: Final = locals()
custom_llm_provider = passed_params.pop("custom_llm_provider")
special_params: Final = passed_params.pop("kwargs")
for k, v in special_params.items():
passed_params[k] = v
passed_params: Final[Mapping[str, object]] = {
"n": n,
"quality": quality,
"response_format": response_format,
"size": size,
"style": style,
"user": user,
**kwargs,
}
default_params: Final = {
"n": None,
@ -90,10 +97,10 @@ def get_optional_params_image_gen(
}
non_default_params = {k: v for k, v in passed_params.items() if (k in default_params and v != default_params[k])}
optional_params = {}
optional_params: dict[str, object] = {}
## raise exception if non-default value passed for non-openai/azure embedding calls
def _check_valid_arg(supported_params):
def _check_valid_arg(supported_params: Sequence[str]) -> Mapping[str, object] | None:
if len(non_default_params.keys()) > 0:
keys: Final = list(non_default_params.keys())
for k in keys:

12
litellm/batches/batch_utils.py
View file

@ -160,7 +160,7 @@ def _classify_output_line_stats(
def _safe_output_line_stats(
entry: Mapping[str, Any],
entry: Mapping[str, object],
custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic", "bedrock"],
model_name: str | None,
model_info: ModelInfo | None,
@ -182,7 +182,7 @@ def _safe_output_line_stats(
def _compute_output_line_stats(
entry: Mapping[str, Any],
entry: Mapping[str, object],
custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic", "bedrock"],
model_name: str | None,
model_info: ModelInfo | None,
@ -213,7 +213,7 @@ def _compute_output_line_stats(
def _output_line_cost(
response_body: Mapping[str, Any],
response_body: Mapping[str, object],
usage: Usage,
custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic", "bedrock"],
model_name: str | None,
@ -556,7 +556,7 @@ def _iter_batch_output_entries(file_content: bytes) -> Iterator[dict]:
def _parse_batch_output_line(line: bytes) -> dict | None:
try:
parsed: Final = json.loads(line)
parsed: Final[object] = json.loads(line)
except ValueError as e:
verbose_logger.warning("skipping malformed batch output line: %s", str(e))
return None
@ -601,7 +601,7 @@ def _count_entry_tokens(
return 0
def _count_prompt_or_input_tokens(model: str, value: Any) -> int:
def _count_prompt_or_input_tokens(model: str, value: object) -> int:
"""Token-count a ``prompt`` / ``input`` field that the OpenAI batch
schema allows in four shapes:
@ -680,7 +680,7 @@ def _get_anthropic_result_from_batch_results_line(batch_results_line: Mapping[st
def _get_response_from_batch_job_output_file(
batch_job_output_file: Mapping[str, Any], custom_llm_provider: str = "openai"
) -> Mapping[str, Any]:
) -> Mapping[str, object]:
"""
Get the response from the batch job output file
"""

12
litellm/compression/compress.py
View file

@ -66,7 +66,7 @@ def _build_retrieval_tools(keys: list[str], call_type: str) -> list[dict]:
return cast(list[dict], anthropic_tools)
def _content_to_text(content: Any) -> str:
def _content_to_text(content: object) -> str:
"""
Convert OpenAI/Anthropic message content blocks to plain text.
@ -78,7 +78,7 @@ def _content_to_text(content: Any) -> str:
Implemented iteratively (stack-based) to avoid unbounded recursion.
"""
parts: Final[list[str]] = []
stack: Final[list[Any]] = [content]
stack: Final[list[object]] = [content]
while stack:
item = stack.pop()
if isinstance(item, str):
@ -111,7 +111,7 @@ def _normalize_messages_for_compression(
f"Unsupported call_type={call_type!r} for compression. Expected one of: {sorted(_SUPPORTED_CALL_TYPES)}."
)
original_messages: Final[list[dict[str, Any]]] = [dict(m) for m in messages]
original_messages: Final[list[dict[str, object]]] = [dict(m) for m in messages]
normalized_messages: Final[list[dict]] = []
for msg in original_messages:
@ -132,7 +132,7 @@ def _extract_last_user_message(messages: list[dict]) -> str:
return ""
def _extract_tool_use_ids(content: Any) -> list[str]:
def _extract_tool_use_ids(content: object) -> list[str]:
if not isinstance(content, list):
return []
tool_use_ids: Final[list[str]] = []
@ -147,7 +147,7 @@ def _extract_tool_use_ids(content: Any) -> list[str]:
return tool_use_ids
def _extract_tool_result_ids(content: Any) -> set[str]:
def _extract_tool_result_ids(content: object) -> set[str]:
if not isinstance(content, list):
return set()
tool_result_ids: Final[set[str]] = set()
@ -337,7 +337,7 @@ def compress(
compression_trigger: int = 200_000,
compression_target: int | None = None,
embedding_model: str | None = None,
embedding_model_params: dict[str, Any] | None = None,
embedding_model_params: Mapping[str, object] | None = None,
compression_cache: DualCache | None = None,
) -> CompressedResult:
"""

14
litellm/containers/endpoint_factory.py
View file

@ -11,7 +11,7 @@ import json
from collections.abc import Callable
from functools import partial
from pathlib import Path
from typing import Any, Final, Literal
from typing import Final, Literal
import litellm
from litellm.constants import request_timeout as DEFAULT_REQUEST_TIMEOUT
@ -56,9 +56,9 @@ def create_sync_endpoint_function(endpoint_config: dict) -> Callable:
def endpoint_func(
timeout: int = 600,
custom_llm_provider: Literal["openai", "azure", "azure_text"] = "openai",
extra_headers: dict[str, Any] | None = None,
extra_query: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_headers: dict[str, object] | None = None,
extra_query: dict[str, object] | None = None,
extra_body: dict[str, object] | None = None,
**kwargs,
):
local_vars: Final = locals()
@ -145,9 +145,9 @@ def create_async_endpoint_function(
async def async_endpoint_func(
timeout: int = 600,
custom_llm_provider: Literal["openai", "azure", "azure_text"] = "openai",
extra_headers: dict[str, Any] | None = None,
extra_query: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_headers: dict[str, object] | None = None,
extra_query: dict[str, object] | None = None,
extra_body: dict[str, object] | None = None,
**kwargs,
):
local_vars: Final = locals()

4
litellm/exceptions.py
View file

@ -85,7 +85,7 @@ _RATE_LIMIT_CATEGORY_VALUES: Final = frozenset(c.value for c in RateLimitErrorCa
_RATE_LIMIT_TYPE_VALUES: Final = frozenset(t.value for t in RateLimitType)
def validate_rate_limit_category(value: Any) -> str | None:
def validate_rate_limit_category(value: object) -> str | None:
"""Return ``value`` only if it matches a known :class:`RateLimitErrorCategory`.
Used at duck-typed read sites (StandardLoggingPayload extraction, Prometheus
@ -100,7 +100,7 @@ def validate_rate_limit_category(value: Any) -> str | None:
return None
def validate_rate_limit_type(value: Any) -> str | None:
def validate_rate_limit_type(value: object) -> str | None:
"""Return ``value`` only if it matches a known :class:`RateLimitType`.
See :func:`validate_rate_limit_category` for the rationale.

14
litellm/fine_tuning/main.py
View file

@ -11,7 +11,7 @@ https://platform.openai.com/docs/api-reference/fine-tuning
import asyncio
import contextvars
import os
from collections.abc import Coroutine
from collections.abc import Coroutine, Mapping
from functools import partial
from typing import Any, Final, Literal
@ -37,8 +37,8 @@ vertex_fine_tuning_apis_instance: Final = VertexFineTuningAPI()
def _prepare_azure_extra_body(
extra_body: dict[str, Any] | None,
kwargs: dict[str, Any],
azure_specific_hyperparams: dict[str, Any],
kwargs: Mapping[str, object],
azure_specific_hyperparams: Mapping[str, object],
) -> dict[str, Any]:
"""
Prepare extra_body for Azure fine-tuning API by combining Azure-specific parameters.
@ -138,7 +138,7 @@ def _build_fine_tuning_job_data(model, training_file, hyperparameters, suffix, v
def _resolve_fine_tuning_timeout(
timeout: Any,
timeout: float | str | httpx.Timeout | None,
custom_llm_provider: str,
) -> float | httpx.Timeout:
"""Normalise a raw timeout value to a float (seconds) or httpx.Timeout for fine-tuning calls."""
@ -163,7 +163,7 @@ def create_fine_tuning_job(
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, str] | None = None,
**kwargs,
) -> LiteLLMFineTuningJob | Coroutine[Any, Any, LiteLLMFineTuningJob]:
) -> LiteLLMFineTuningJob | Coroutine[object, object, LiteLLMFineTuningJob]:
"""
Creates a fine-tuning job which begins the process of creating a new model from a given dataset.
@ -375,7 +375,7 @@ def cancel_fine_tuning_job(
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, str] | None = None,
**kwargs,
) -> LiteLLMFineTuningJob | Coroutine[Any, Any, LiteLLMFineTuningJob]:
) -> LiteLLMFineTuningJob | Coroutine[object, object, LiteLLMFineTuningJob]:
"""
Immediately cancel a fine-tune job.
@ -682,7 +682,7 @@ def retrieve_fine_tuning_job(
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, str] | None = None,
**kwargs,
) -> LiteLLMFineTuningJob | Coroutine[Any, Any, LiteLLMFineTuningJob]:
) -> LiteLLMFineTuningJob | Coroutine[object, object, LiteLLMFineTuningJob]:
"""
Get info about a fine-tuning job.
"""

3
litellm/images/utils.py
View file

@ -1,3 +1,4 @@
from collections.abc import Mapping
from io import BufferedReader, BytesIO
from typing import Any, Final, cast, get_type_hints
@ -61,7 +62,7 @@ class ImageEditRequestUtils:
@staticmethod
def get_requested_image_edit_optional_param(
params: dict[str, Any],
params: Mapping[str, object],
) -> ImageEditOptionalRequestParams:
"""
Filter parameters to only include those defined in ImageEditOptionalRequestParams.

5
litellm/integrations/datadog/datadog_cost_management.py
View file

@ -1,6 +1,7 @@
import asyncio
import os
import time
from collections.abc import Mapping
from datetime import datetime
from typing import Any, Final, cast
@ -181,7 +182,7 @@ class DatadogCostManagementLogger(CustomBatchLogger):
# cast because StandardLoggingMetadata is a TypedDict; we iterate it
# as a generic mapping below.
metadata: Final[dict[str, Any]] = cast(dict[str, Any], log.get("metadata") or {})
metadata: Final[Mapping[str, object]] = cast(dict[str, Any], log.get("metadata") or {})
# Backwards-compat: team/user/model_group preserved regardless of allowlist.
if metadata.get("user_api_key_alias"):
@ -233,7 +234,7 @@ class DatadogCostManagementLogger(CustomBatchLogger):
tags[key] = normalize_datadog_tag_value(value)
@staticmethod
def _add_tag(tags: dict[str, str], key: str, value: Any) -> None:
def _add_tag(tags: dict[str, str], key: str, value: object) -> None:
if value:
tags[key] = str(value)

7
litellm/integrations/dotprompt/dotprompt_manager.py
View file

@ -4,6 +4,7 @@ Builds on top of PromptManagementBase to provide .prompt file support.
"""
import json
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from litellm.integrations.custom_prompt_management import CustomPromptManagement
@ -347,14 +348,14 @@ class DotpromptManager(CustomPromptManagement):
metadata: Final = json_data.get("metadata", {})
self.prompt_manager.add_prompt(prompt_id, content, metadata)
def load_prompts_from_json(self, prompts_data: dict[str, dict[str, Any]]) -> None:
def load_prompts_from_json(self, prompts_data: dict[str, dict[str, object]]) -> None:
"""Load multiple prompts from JSON data."""
self.prompt_manager.load_prompts_from_json_data(prompts_data)
def get_prompts_as_json(self) -> dict[str, dict[str, Any]]:
def get_prompts_as_json(self) -> dict[str, dict[str, object]]:
"""Get all prompts in JSON format."""
return self.prompt_manager.get_all_prompts_as_json()
def convert_prompt_file_to_json(self, file_path: str) -> dict[str, Any]:
def convert_prompt_file_to_json(self, file_path: str) -> Mapping[str, object]:
"""Convert a .prompt file to JSON format."""
return self.prompt_manager.prompt_file_to_json(file_path)

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

@ -102,7 +102,7 @@ class FocusLogger(CustomLogger):
# No time bounds → export all available data
await self._export_all(limit=limit)
async def dry_run_export_usage_data(self, limit: int | None = DEFAULT_DRY_RUN_LIMIT) -> dict[str, Any]:
async def dry_run_export_usage_data(self, limit: int | None = DEFAULT_DRY_RUN_LIMIT) -> dict[str, object]:
"""Return transformed data without uploading."""
engine: Final = self._ensure_engine()
return await engine.dry_run_export_usage_data(limit=limit)
@ -153,7 +153,7 @@ class FocusLogger(CustomLogger):
**trigger_kwargs,
)
def _build_scheduler_trigger(self) -> dict[str, Any]:
def _build_scheduler_trigger(self) -> dict[str, str | int]:
"""Return scheduler configuration for the selected frequency."""
if self.frequency == "interval":
seconds: Final = self.interval_seconds or 60

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

@ -4,6 +4,7 @@ Fetches prompts from any API that implements the /beta/litellm_prompt_management
"""
import json
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -349,7 +350,7 @@ class GenericPromptManager(CustomPromptManagement):
def _apply_variables(
self,
prompt_client: PromptManagementClient,
variables: dict[str, Any],
variables: Mapping[str, object],
) -> PromptManagementClient:
"""
Apply variables to the prompt template.

6
litellm/integrations/humanloop.py
View file

@ -4,7 +4,7 @@ Humanloop integration
https://humanloop.com/
"""
from typing import Any, Final, cast
from typing import Final, cast
import httpx
from typing_extensions import TypedDict
@ -24,7 +24,7 @@ class PromptManagementClient(TypedDict):
prompt_id: str
prompt_template: list[AllMessageValues]
model: str | None
optional_params: dict[str, Any] | None
optional_params: dict[str, object] | None
class HumanLoopPromptManager(DualCache):
@ -36,7 +36,7 @@ class HumanLoopPromptManager(DualCache):
return cast(PromptManagementClient | None, self.get_cache(key=humanloop_prompt_id))
def _compile_prompt_helper(
self, prompt_template: list[AllMessageValues], prompt_variables: dict[str, Any]
self, prompt_template: list[AllMessageValues], prompt_variables: dict[str, object]
) -> list[AllMessageValues]:
"""
Helper function to compile the prompt by substituting variables in the template.

6
litellm/integrations/opentelemetry_utils/gen_ai_semconv.py
View file

@ -117,7 +117,7 @@ class OTELGenAISemconvMixin:
if TYPE_CHECKING:
config: "OpenTelemetryConfig"
def safe_set_attribute(self, span: Span, key: str, value: Any) -> None: ...
def safe_set_attribute(self, span: Span, key: str, value: object) -> None: ...
def _capture_in_event(self) -> bool: ...
@ -195,13 +195,13 @@ class OTELGenAISemconvMixin:
if value:
self.safe_set_attribute(span=span, key=semconv_key, value=value)
def _build_inference_details_attrs(self, kwargs: dict, response_obj: dict, provider: str) -> dict[str, Any]:
def _build_inference_details_attrs(self, kwargs: dict, response_obj: dict, provider: str) -> dict[str, str]:
"""Build the attribute payload for the inference-details event.
Always includes provider/operation; input/output messages are added
only when content capture is enabled and non-empty. Mixin-internal.
"""
attrs: Final[dict[str, Any]] = {
attrs: Final[dict[str, str]] = {
"event_name": _INFERENCE_DETAILS_EVENT_NAME,
"gen_ai.provider.name": provider,
"gen_ai.operation.name": self._gen_ai_operation_name(kwargs),

8
litellm/integrations/opik/opik_payload_builder/payload_builders.py
View file

@ -15,8 +15,8 @@ def build_trace_payload(
response_obj: dict[str, Any],
start_time: datetime,
end_time: datetime,
input_data: Any,
output_data: Any,
input_data: object,
output_data: object,
metadata: dict[str, object],
tags: list[str],
thread_id: str | None,
@ -45,8 +45,8 @@ def build_span_payload(
response_obj: dict[str, Any],
start_time: datetime,
end_time: datetime,
input_data: Any,
output_data: Any,
input_data: object,
output_data: object,
metadata: dict[str, object],
tags: list[str],
usage: dict[str, int],

5
litellm/integrations/weave/weave_otel.py
View file

@ -3,6 +3,7 @@ from __future__ import annotations
import base64
import json
import os
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from opentelemetry.trace import Status, StatusCode
@ -59,7 +60,7 @@ class WeaveLLMObsOTELAttributes(BaseLLMObsOTELAttributes):
safe_set_attribute(span, OpenInferenceSpanAttributes.INPUT_VALUE, json.dumps(prompt))
def _set_weave_specific_attributes(span: Span, kwargs: dict[str, Any], response_obj: Any):
def _set_weave_specific_attributes(span: Span, kwargs: Mapping[str, Any], response_obj: Any):
"""
Sets Weave-specific metadata attributes onto the OTEL span.
@ -169,7 +170,7 @@ def get_weave_otel_config() -> WeaveOtelConfig:
)
def set_weave_otel_attributes(span: Span, kwargs: dict[str, Any], response_obj: Any):
def set_weave_otel_attributes(span: Span, kwargs: Mapping[str, object], response_obj: object):
"""
Sets OpenTelemetry span attributes for Weave observability.
Uses the same attribute setting logic as other OTEL integrations for consistency.

17
litellm/litellm_core_utils/dot_notation_indexing.py
View file

@ -23,12 +23,13 @@ Used by JWT Auth to get the user role from the token, and by
additional_drop_params to remove nested fields from optional parameters.
"""
from collections.abc import Mapping
from typing import Any, Final, TypeVar
T = TypeVar("T")
def get_nested_value(data: dict[str, Any], key_path: str, default: T | None = None) -> T | None:
def get_nested_value(data: Mapping[str, object], key_path: str, default: T | None = None) -> T | None:
"""
Retrieves a value from a nested dictionary using dot notation.
@ -107,7 +108,7 @@ def _parse_path_segments(path: str) -> list:
def _delete_nested_value_custom(
data: dict[str, Any] | list[Any],
data: dict[str, object] | list[object],
segments: list,
segment_index: int = 0,
) -> None:
@ -168,13 +169,15 @@ def _delete_nested_value_custom(
if segment in data:
next_segment: Final = segments[segment_index + 1] if segment_index + 1 < len(segments) else None
child: Final = data[segment]
# If next segment is array notation, current field should be list
if next_segment and (next_segment.startswith("[")):
if isinstance(data[segment], list):
_delete_nested_value_custom(data[segment], segments, segment_index + 1)
if isinstance(child, list):
_delete_nested_value_custom(child, segments, segment_index + 1)
# Otherwise navigate into dict
elif isinstance(data[segment], dict):
_delete_nested_value_custom(data[segment], segments, segment_index + 1)
elif isinstance(child, dict):
_delete_nested_value_custom(child, segments, segment_index + 1)
def delete_nested_value(
@ -182,7 +185,7 @@ def delete_nested_value(
path: str,
depth: int = 0,
max_depth: int = 20,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Delete a field from nested data using JSONPath notation.

6
litellm/litellm_core_utils/json_validation_rule.py
View file

@ -5,10 +5,10 @@ from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
def normalize_json_schema_types(
schema: dict[str, Any] | list[Any] | Any,
schema: object,
depth: int = 0,
max_depth: int = DEFAULT_MAX_RECURSE_DEPTH,
) -> dict[str, Any] | list[Any] | Any:
) -> object:
"""
Normalize JSON schema types from uppercase to lowercase format.
@ -47,7 +47,7 @@ def normalize_json_schema_types(
return [normalize_json_schema_types(item, depth + 1, max_depth) for item in schema]
if isinstance(schema, dict):
normalized_schema: Final[dict[str, Any]] = {}
normalized_schema: Final[dict[str, object]] = {}
for key, value in schema.items():
if key == "type" and isinstance(value, str) and value in type_mapping:

2
litellm/litellm_core_utils/logging_utils.py
View file

@ -184,7 +184,7 @@ def _get_parent_otel_span_from_logging_obj(
def convert_litellm_response_object_to_str(
response_obj: Any | LiteLLMModelResponse,
response_obj: object,
) -> str | None:
"""
Get the string of the response object from LiteLLM

2
litellm/litellm_core_utils/safe_json_dumps.py
View file

@ -30,7 +30,7 @@ def safe_dumps(
def _transform(key: str | None, value: str) -> str:
return value if value_transform is None else value_transform(key, value)
def _serialize(obj: Any, seen: set, depth: int, key: str | None = None) -> Any:
def _serialize(obj: object, seen: set[int], depth: int, key: str | None = None) -> Any:
# Check for maximum depth.
if depth > max_depth:
return "MaxDepthExceeded"

3
litellm/llms/a2a/common_utils.py
View file

@ -2,6 +2,7 @@
Common utilities for A2A (Agent-to-Agent) Protocol
"""
from collections.abc import Mapping
from typing import Any, Final
from pydantic import BaseModel
@ -91,7 +92,7 @@ def extract_text_from_a2a_message(message: dict[str, Any], depth: int = 0, max_d
return " ".join(text_parts)
def extract_text_from_a2a_response(response_dict: dict[str, Any], max_depth: int = 10) -> str:
def extract_text_from_a2a_response(response_dict: Mapping[str, object], max_depth: int = 10) -> str:
"""
Extract text content from A2A response result.

10
litellm/llms/anthropic/experimental_pass_through/messages/interceptors/advisor.py
View file

@ -266,7 +266,7 @@ def _make_synthetic_advisor_tool() -> dict:
}
def _find_advisor_tool_use(response: Any) -> dict | None:
def _find_advisor_tool_use(response: object) -> dict | None:
"""Return the first tool_use block with name='advisor', or None."""
content: Final = response.get("content") if isinstance(response, dict) else []
if not isinstance(content, list):
@ -277,7 +277,7 @@ def _find_advisor_tool_use(response: Any) -> dict | None:
return None
def _extract_response_text(response: Any) -> str:
def _extract_response_text(response: object) -> str:
"""Extract concatenated text from all text blocks in a response."""
content: Final = response.get("content") if isinstance(response, dict) else []
if not isinstance(content, list):
@ -291,7 +291,7 @@ _PROVIDER_SPECIFIC_KEYS: Final = frozenset({"provider_specific_fields"})
def _build_advisor_context(
messages: list[dict],
executor_response: Any,
executor_response: object,
advisor_use_block: dict,
) -> list[dict]:
"""
@ -327,7 +327,7 @@ def _build_advisor_context(
def _inject_advisor_turn(
messages: list[dict],
executor_response: Any,
executor_response: object,
advisor_use_block: dict,
advisor_text: str,
) -> list[dict]:
@ -355,7 +355,7 @@ def _inject_advisor_turn(
def _inject_max_uses_error(
messages: list[dict],
executor_response: Any,
executor_response: object,
advisor_use_block: dict,
) -> list[dict]:
"""

10
litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py
View file

@ -42,9 +42,9 @@ class AnthropicResponsesStreamWrapper:
self._pending_tool_ids: dict[str, str] = {} # item_id -> call_id / name accumulator
self._sent_message_start = False
self._sent_message_stop = False
self._chunk_queue: deque = deque()
self._chunk_queue: deque[dict[str, object]] = deque()
def _make_message_start(self) -> dict[str, Any]:
def _make_message_start(self) -> dict[str, object]:
return {
"type": "message_start",
"message": {
@ -68,7 +68,7 @@ class AnthropicResponsesStreamWrapper:
self._current_block_index += 1
return self._current_block_index
def _open_block(self, item_id: str | None, content_block: Mapping[str, Any]) -> int:
def _open_block(self, item_id: str | None, content_block: Mapping[str, object]) -> int:
block_idx = self._next_block_index()
if item_id:
self._item_id_to_block_index[item_id] = block_idx
@ -81,7 +81,7 @@ class AnthropicResponsesStreamWrapper:
)
return block_idx
def _process_event(self, event: Any) -> None:
def _process_event(self, event: object) -> None:
"""Convert one Responses API event into zero or more Anthropic chunks queued for emission."""
event_type = getattr(event, "type", None)
if event_type is None and isinstance(event, dict):
@ -247,7 +247,7 @@ class AnthropicResponsesStreamWrapper:
def __aiter__(self) -> "AnthropicResponsesStreamWrapper":
return self
async def __anext__(self) -> dict[str, Any]:
async def __anext__(self) -> dict[str, object]:
# Return any queued chunks first
if self._chunk_queue:
return self._chunk_queue.popleft()

4
litellm/llms/anthropic/files/transformation.py
View file

@ -14,7 +14,7 @@ Anthropic Files API endpoints:
import calendar
import time
from typing import Any, Final, cast
from typing import Final, cast
import httpx
from openai.types.file_deleted import FileDeleted
@ -226,7 +226,7 @@ class AnthropicFilesConfig(BaseFilesConfig):
) -> tuple[str, dict]:
api_base: Final = AnthropicModelInfo.get_api_base(litellm_params.get("api_base")) or ANTHROPIC_FILES_API_BASE
url: Final = f"{api_base.rstrip('/')}/v1/files"
params: Final[dict[str, Any]] = {}
params: Final[dict[str, str]] = {}
if purpose:
params["purpose"] = purpose
return url, params

13
litellm/llms/aws_polly/text_to_speech/transformation.py
View file

@ -20,6 +20,7 @@ from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.types.llms.openai import HttpxBinaryResponseContent
else:
LiteLLMLoggingObj = Any
@ -75,15 +76,15 @@ class AWSPollyTextToSpeechConfig(BaseTextToSpeechConfig, BaseAWSLLM):
litellm_params_dict: dict,
logging_obj: "LiteLLMLoggingObj",
timeout: float | httpx.Timeout,
extra_headers: dict[str, Any] | None,
base_llm_http_handler: Any,
extra_headers: dict[str, object] | None,
base_llm_http_handler: "BaseLLMHTTPHandler",
aspeech: bool,
api_base: str | None,
api_key: str | None,
**kwargs: Any,
**kwargs: object,
) -> Union[
"HttpxBinaryResponseContent",
Coroutine[Any, Any, "HttpxBinaryResponseContent"],
Coroutine[object, object, "HttpxBinaryResponseContent"],
]:
"""
Dispatch method to handle AWS Polly TTS requests
@ -251,7 +252,7 @@ class AWSPollyTextToSpeechConfig(BaseTextToSpeechConfig, BaseAWSLLM):
def _sign_polly_request(
self,
request_body: dict[str, Any],
request_body: dict[str, object],
endpoint_url: str,
litellm_params: dict,
) -> tuple[dict[str, str], str]:
@ -337,7 +338,7 @@ class AWSPollyTextToSpeechConfig(BaseTextToSpeechConfig, BaseAWSLLM):
engine: Final = optional_params.get("engine", self.DEFAULT_ENGINE)
# Build request body
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"Engine": engine,
"OutputFormat": output_format,
"Text": input,

14
litellm/llms/azure/realtime/handler.py
View file

@ -6,7 +6,7 @@ This requires websockets, and is currently only supported on LiteLLM Proxy.
from collections.abc import Mapping
from types import MappingProxyType
from typing import Any, Final, cast
from typing import Any, Final, Protocol, cast
from litellm._logging import _redact_string, verbose_proxy_logger
from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
@ -31,6 +31,12 @@ async def forward_messages(client_ws: Any, backend_ws: Any):
pass
class _ProxyClientWebSocket(Protocol):
"""Client-facing websocket handle: this path only closes it after a failed handshake."""
async def close(self, code: int = ..., reason: str | None = ...) -> None: ...
class AzureOpenAIRealtime(AzureChatCompletion):
@staticmethod
def get_auth_headers(api_key: str | None, azure_ad_token: str | None) -> Mapping[str, str]:
@ -104,17 +110,17 @@ class AzureOpenAIRealtime(AzureChatCompletion):
async def async_realtime(
self,
model: str,
websocket: Any,
websocket: _ProxyClientWebSocket,
logging_obj: LiteLLMLogging,
api_base: str | None = None,
api_key: str | None = None,
api_version: str | None = None,
azure_ad_token: str | None = None,
client: Any | None = None,
client: object | None = None,
timeout: float | None = None,
realtime_protocol: str | None = None,
query_params: RealtimeQueryParams | None = None,
user_api_key_dict: Any | None = None,
user_api_key_dict: object | None = None,
litellm_metadata: dict | None = None,
):
import websockets

6
litellm/llms/azure/responses/transformation.py
View file

@ -96,7 +96,7 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
# Then filter out status from message items
if isinstance(validated_input, list):
filtered_input: Final[list[Any]] = []
filtered_input: Final[list[object]] = []
for item in validated_input:
if isinstance(item, dict) and item.get("type") == "message":
# Filter out status field from message items
@ -123,7 +123,7 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
if "tools" in response_api_optional_request_params and isinstance(
response_api_optional_request_params["tools"], list
):
new_tools: Final[list[dict[str, Any]]] = []
new_tools: Final[list[dict[str, object]]] = []
for tool in response_api_optional_request_params["tools"]:
if isinstance(tool, dict) and "function" in tool:
new_tool: dict[str, Any] = deepcopy(tool)
@ -291,7 +291,7 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
url: Final = self._construct_url_for_response_id_in_path(
api_base=api_base, response_id=response_id, path_suffix="/input_items"
)
params: Final[dict[str, Any]] = {}
params: Final[dict[str, str | int]] = {}
if after is not None:
params["after"] = after
if before is not None:

8
litellm/llms/azure_ai/anthropic/count_tokens/token_counter.py
View file

@ -28,12 +28,12 @@ class AzureAIAnthropicTokenCounter(BaseTokenCounter):
async def count_tokens(
self,
model_to_use: str,
messages: list[dict[str, Any]] | None,
contents: list[dict[str, Any]] | None,
messages: list[dict[str, object]] | None,
contents: list[dict[str, object]] | None,
deployment: dict[str, Any] | None = None,
request_model: str = "",
tools: list[dict[str, Any]] | None = None,
system: Any | None = None,
tools: list[dict[str, object]] | None = None,
system: object | None = None,
) -> TokenCountResponse | None:
"""
Count tokens using Azure AI Anthropic's CountTokens API.

24
litellm/llms/base_llm/agents/transformation.py
View file

@ -10,7 +10,7 @@ InteractionsHTTPHandler).
"""
from abc import ABC, abstractmethod
from typing import Any
from collections.abc import Mapping
import httpx
@ -35,7 +35,7 @@ class BaseAgentsAPIConfig(ABC):
def get_complete_url(
self,
api_base: str | None,
litellm_params: dict[str, Any],
litellm_params: Mapping[str, object],
) -> str:
"""Return the full URL for POST /agents (create)."""
@ -43,7 +43,7 @@ class BaseAgentsAPIConfig(ABC):
def validate_environment(
self,
headers: dict[str, str],
litellm_params: dict[str, Any],
litellm_params: dict[str, object],
) -> dict[str, str]:
"""Validate credentials and return auth headers."""
@ -51,8 +51,8 @@ class BaseAgentsAPIConfig(ABC):
def transform_create_request(
self,
name: str,
litellm_params: dict[str, Any],
) -> dict[str, Any]:
litellm_params: Mapping[str, object],
) -> dict[str, object]:
"""Map name + litellm_params to the provider's create-agent body."""
@abstractmethod
@ -71,8 +71,8 @@ class BaseAgentsAPIConfig(ABC):
def transform_list_request(
self,
api_base: str | None,
litellm_params: dict[str, Any],
) -> tuple[str, dict[str, Any]]:
litellm_params: Mapping[str, object],
) -> tuple[str, dict[str, object]]:
"""Return (url, query_params) for GET /agents."""
@abstractmethod
@ -91,8 +91,8 @@ class BaseAgentsAPIConfig(ABC):
self,
name: str,
api_base: str | None,
litellm_params: dict[str, Any],
) -> tuple[str, dict[str, Any]]:
litellm_params: Mapping[str, object],
) -> tuple[str, dict[str, object]]:
"""Return (url, query_params) for GET /agents/{name}."""
@abstractmethod
@ -112,7 +112,7 @@ class BaseAgentsAPIConfig(ABC):
self,
name: str,
api_base: str | None,
litellm_params: dict[str, Any],
litellm_params: Mapping[str, object],
) -> str:
"""Return the URL for DELETE /agents/{name}."""
@ -133,8 +133,8 @@ class BaseAgentsAPIConfig(ABC):
self,
name: str,
api_base: str | None,
litellm_params: dict[str, Any],
) -> tuple[str, dict[str, Any]]:
litellm_params: Mapping[str, object],
) -> tuple[str, dict[str, object]]:
"""Return (url, query_params) for GET /agents/{name}/versions."""
@abstractmethod

12
litellm/llms/base_llm/guardrail_translation/utils.py
View file

@ -2,13 +2,13 @@ from __future__ import annotations
import json
from collections.abc import Callable, Iterator, Sequence
from typing import Any, Final, TypeVar
from typing import Final, TypeVar
from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUsage
from litellm.types.llms.openai import AllMessageValues, ResponseAPIUsage
def _anthropic_stream_chunk_events(item: Any) -> list[dict]:
def _anthropic_stream_chunk_events(item: object) -> list[dict]:
if isinstance(item, dict):
return [item]
if isinstance(item, bytes):
@ -36,7 +36,7 @@ def _anthropic_stream_chunk_events(item: Any) -> list[dict]:
return events
def _usage_from_anthropic_stream_chunks(original_response: list[Any]) -> AnthropicUsage | None:
def _usage_from_anthropic_stream_chunks(original_response: Sequence[object]) -> AnthropicUsage | None:
input_tokens = 0
output_tokens = 0
found_usage = False
@ -79,7 +79,7 @@ def _usage_tokens(usage_obj: object, key: str, fallback_key: str) -> int:
return int(getattr(usage_obj, key, getattr(usage_obj, fallback_key, 0)) or 0)
def blocked_response_usage(original_response: Any | None) -> AnthropicUsage:
def blocked_response_usage(original_response: object) -> AnthropicUsage:
"""
Token usage for a synthetic guardrail-blocked response.
@ -179,7 +179,7 @@ def blocked_responses_stream_usage(original_response: object) -> ResponseAPIUsag
return blocked_responses_api_usage(completed)
def effective_skip_system_message_for_guardrail(guardrail_to_apply: Any) -> bool:
def effective_skip_system_message_for_guardrail(guardrail_to_apply: object) -> bool:
per: Final = getattr(guardrail_to_apply, "skip_system_message_in_guardrail", None)
if per is not None:
return bool(per)
@ -188,7 +188,7 @@ def effective_skip_system_message_for_guardrail(guardrail_to_apply: Any) -> bool
return bool(getattr(litellm, "skip_system_message_in_guardrail", False))
def effective_skip_tool_message_for_guardrail(guardrail_to_apply: Any) -> bool:
def effective_skip_tool_message_for_guardrail(guardrail_to_apply: object) -> bool:
per: Final = getattr(guardrail_to_apply, "skip_tool_message_in_guardrail", None)
if per is not None:
return bool(per)

23
litellm/llms/base_llm/vector_store_files/transformation.py
View file

@ -1,4 +1,5 @@
from abc import ABC, abstractmethod
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any
import httpx
@ -43,10 +44,10 @@ class BaseVectorStoreFilesConfig(ABC):
self,
*,
operation: str,
non_default_params: dict[str, Any],
optional_params: dict[str, Any],
non_default_params: Mapping[str, object],
optional_params: Mapping[str, object],
drop_params: bool,
) -> dict[str, Any]:
) -> Mapping[str, object]:
"""Map non-default OpenAI params to provider-specific params."""
return optional_params
@ -87,7 +88,7 @@ class BaseVectorStoreFilesConfig(ABC):
vector_store_id: str,
create_request: VectorStoreFileCreateRequest,
api_base: str,
) -> tuple[str, dict[str, Any]]: ...
) -> tuple[str, dict[str, object]]: ...
@abstractmethod
def transform_create_vector_store_file_response(
@ -103,7 +104,7 @@ class BaseVectorStoreFilesConfig(ABC):
vector_store_id: str,
query_params: VectorStoreFileListQueryParams,
api_base: str,
) -> tuple[str, dict[str, Any]]: ...
) -> tuple[str, dict[str, object]]: ...
@abstractmethod
def transform_list_vector_store_files_response(
@ -119,7 +120,7 @@ class BaseVectorStoreFilesConfig(ABC):
vector_store_id: str,
file_id: str,
api_base: str,
) -> tuple[str, dict[str, Any]]: ...
) -> tuple[str, dict[str, object]]: ...
@abstractmethod
def transform_retrieve_vector_store_file_response(
@ -135,7 +136,7 @@ class BaseVectorStoreFilesConfig(ABC):
vector_store_id: str,
file_id: str,
api_base: str,
) -> tuple[str, dict[str, Any]]: ...
) -> tuple[str, dict[str, object]]: ...
@abstractmethod
def transform_retrieve_vector_store_file_content_response(
@ -152,7 +153,7 @@ class BaseVectorStoreFilesConfig(ABC):
file_id: str,
update_request: VectorStoreFileUpdateRequest,
api_base: str,
) -> tuple[str, dict[str, Any]]: ...
) -> tuple[str, dict[str, object]]: ...
@abstractmethod
def transform_update_vector_store_file_response(
@ -168,7 +169,7 @@ class BaseVectorStoreFilesConfig(ABC):
vector_store_id: str,
file_id: str,
api_base: str,
) -> tuple[str, dict[str, Any]]: ...
) -> tuple[str, dict[str, object]]: ...
@abstractmethod
def transform_delete_vector_store_file_response(
@ -196,8 +197,8 @@ class BaseVectorStoreFilesConfig(ABC):
self,
*,
headers: dict[str, str],
optional_params: dict[str, Any],
request_data: dict[str, Any],
optional_params: Mapping[str, object],
request_data: Mapping[str, object],
api_base: str,
api_key: str | None = None,
) -> tuple[dict[str, str], bytes | None]:

22
litellm/llms/bedrock/base_aws_llm.py
View file

@ -4,7 +4,7 @@ import json
import os
import re
import urllib.parse
from collections.abc import Callable
from collections.abc import Callable, Mapping
from datetime import datetime
from threading import Lock
from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, cast, get_args
@ -125,7 +125,7 @@ class BaseAWSLLM:
return get_ssl_verify(ssl_verify=ssl_verify)
def get_cache_key(self, credential_args: dict[str, str | None]) -> str:
def get_cache_key(self, credential_args: Mapping[str, str | bool | None]) -> str:
"""
Generate a unique cache key based on the credential arguments.
"""
@ -135,8 +135,8 @@ class BaseAWSLLM:
def _get_or_set_cached_credentials(
self,
credential_args: dict[str, str | None],
credential_fetcher: Callable[[], tuple[Any, int | None]],
credential_args: Mapping[str, str | bool | None],
credential_fetcher: Callable[[], tuple[Credentials, int | None]],
) -> Any:
"""
Read-through IAM cache on the process-wide ``DualCache``.
@ -271,7 +271,19 @@ class BaseAWSLLM:
aws_external_id,
)
args: Final = {k: v for k, v in locals().items() if k.startswith("aws_") or k == "ssl_verify"}
args: Final = {
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key,
"aws_session_token": aws_session_token,
"aws_region_name": aws_region_name,
"aws_session_name": aws_session_name,
"aws_profile_name": aws_profile_name,
"aws_role_name": aws_role_name,
"aws_web_identity_token": aws_web_identity_token,
"aws_sts_endpoint": aws_sts_endpoint,
"aws_external_id": aws_external_id,
"ssl_verify": ssl_verify,
}
#########################################################
# Handle diff boto3 auth flows

6
litellm/llms/custom_httpx/container_handler.py
View file

@ -141,7 +141,7 @@ def _build_query_params(
def _error_message_from_response(response: httpx.Response) -> str:
try:
body: Final = response.json()
body: Final[object] = response.json()
except ValueError:
return response.text
@ -330,7 +330,7 @@ class GenericContainerHandler:
timeout: float | httpx.Timeout = 600,
client: HTTPHandler | AsyncHTTPHandler | None = None,
**kwargs: object,
) -> Any:
) -> ContainerEndpointResponse:
"""Synchronous request handler."""
endpoint_config: Final = _get_endpoint_config(endpoint_name)
if not endpoint_config:
@ -410,7 +410,7 @@ class GenericContainerHandler:
timeout: float | httpx.Timeout = 600,
client: HTTPHandler | AsyncHTTPHandler | None = None,
**kwargs: object,
) -> Any:
) -> ContainerEndpointResponse:
"""Asynchronous request handler."""
endpoint_config: Final = _get_endpoint_config(endpoint_name)
if not endpoint_config:

6
litellm/llms/gemini/google_genai/transformation.py
View file

@ -117,7 +117,7 @@ class GoogleGenAIConfig(BaseGoogleGenAIGenerateContentConfig, VertexLLM):
_snake_to_camel,
)
_generate_content_config_dict: Final[dict[str, Any]] = {}
_generate_content_config_dict: Final[dict[str, object]] = {}
supported_google_genai_params: Final = self.get_supported_generate_content_optional_params(model)
# Create a set with both camelCase and snake_case versions for faster lookup
supported_params_set: Final = set(supported_google_genai_params)
@ -175,7 +175,7 @@ class GoogleGenAIConfig(BaseGoogleGenAIGenerateContentConfig, VertexLLM):
def _get_common_auth_components(
self,
litellm_params: dict,
) -> tuple[Any, str | None, str | None]:
) -> tuple[str | None, str | None, str | None]:
"""
Get common authentication components used by both sync and async methods.
@ -193,7 +193,7 @@ class GoogleGenAIConfig(BaseGoogleGenAIGenerateContentConfig, VertexLLM):
auth_header: str | None,
vertex_project: str | None,
vertex_location: str | None,
vertex_credentials: Any,
vertex_credentials: str | None,
stream: bool,
api_base: str | None,
litellm_params: dict,

6
litellm/llms/jina_ai/rerank/transformation.py
View file

@ -6,8 +6,8 @@ Why separate file? Make it easy to see how transformation works
Docs - https://jina.ai/reranker
"""
from collections.abc import Mapping
from typing import Any, Final
from collections.abc import Mapping, Sequence
from typing import Final
from httpx import URL, Response
@ -39,7 +39,7 @@ class JinaAIRerankConfig(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,

30
litellm/llms/litellm_proxy/skills/handler.py
View file

@ -6,7 +6,7 @@ Used by the transformation layer and skills injection hook.
"""
import uuid
from typing import Any, Final
from typing import Final
from litellm._logging import verbose_logger
from litellm.caching.in_memory_cache import InMemoryCache
@ -76,7 +76,7 @@ class LiteLLMSkillsHandler:
# this module FastAPI-free per the project layering rule.
raise ValueError("Unable to record skill ownership: caller has no identity scope.")
skill_data: Final[dict[str, Any]] = {
skill_data: Final[dict[str, object]] = {
"skill_id": skill_id,
"display_title": data.display_title,
"description": data.description,
@ -115,22 +115,24 @@ class LiteLLMSkillsHandler:
verbose_logger.debug("LiteLLMSkillsHandler: Listing skills with limit=%s, offset=%s", limit, offset)
find_many_kwargs: Final[dict[str, Any]] = {
"take": limit,
"skip": offset,
"order": {"created_at": "desc"},
}
if user_api_key_dict is not None and not is_proxy_admin(user_api_key_dict):
owner_scopes: Final = get_resource_owner_scopes(user_api_key_dict)
if not owner_scopes:
return []
find_many_kwargs["where"] = {"created_by": {"in": owner_scopes}}
owner_scopes: Final = (
get_resource_owner_scopes(user_api_key_dict)
if user_api_key_dict is not None and not is_proxy_admin(user_api_key_dict)
else None
)
if owner_scopes is not None and not owner_scopes:
return []
skills: Final = await SkillsRepository(prisma_client).table.find_many(**find_many_kwargs)
skills: Final = await SkillsRepository(prisma_client).table.find_many(
take=limit,
skip=offset,
order={"created_at": "desc"},
where={"created_by": {"in": owner_scopes}} if owner_scopes else None,
)
return [_prisma_skill_to_litellm(s) for s in skills]
@staticmethod
async def _load_skill(skill_id: str) -> Any | None:
async def _load_skill(skill_id: str) -> object | None:
"""Cache-first read of the Prisma skill row. Owner-scope filtering
happens on the cached row, so the cache is per-skill not per-caller.
"""

39
litellm/llms/litellm_proxy/skills/sandbox_executor.py
View file

@ -7,11 +7,40 @@ Supports Docker, Podman, and Kubernetes backends.
import base64
import os
from typing import Any, Final
from typing import Any, Final, Protocol, TypedDict
from typing_extensions import ReadOnly
from litellm._logging import verbose_logger
class _SandboxRunResult(Protocol):
"""Result of running code inside an llm-sandbox session."""
@property
def exit_code(self) -> int: ...
@property
def stdout(self) -> str | None: ...
class _SandboxSession(Protocol):
"""The subset of an llm-sandbox session used while collecting generated files."""
def run(self, code: str, /) -> _SandboxRunResult: ...
def copy_from_runtime(self, src: str, dest: str, /) -> object: ...
class _GeneratedFile(TypedDict):
"""A file produced inside the sandbox and carried back out as base64."""
name: ReadOnly[str]
path: ReadOnly[str]
content_base64: ReadOnly[str]
mime_type: ReadOnly[str]
class SkillsSandboxExecutor:
"""
Executes skill code in llm-sandbox Docker container.
@ -77,7 +106,7 @@ class SkillsSandboxExecutor:
try:
# Create sandbox session
session_kwargs: Final[dict[str, Any]] = {
session_kwargs: Final[dict[str, object]] = {
"lang": "python",
"verbose": False,
}
@ -197,9 +226,9 @@ sys.path.insert(0, '/sandbox')
def _collect_generated_files(
self,
session: Any,
session: _SandboxSession,
original_files: dict[str, bytes],
) -> list[dict[str, Any]]:
) -> list[_GeneratedFile]:
"""
Collect files generated during execution.
@ -213,7 +242,7 @@ sys.path.insert(0, '/sandbox')
Returns:
List of generated files with base64 content
"""
generated_files: Final[list[dict[str, Any]]] = []
generated_files: Final[list[_GeneratedFile]] = []
try:
import tempfile

17
litellm/llms/openai/fine_tuning/handler.py
View file

@ -1,13 +1,14 @@
from collections.abc import Coroutine
from typing import Any, Final, cast
from collections.abc import Coroutine, Mapping
from typing import Final, cast
import httpx
from openai import AsyncAzureOpenAI, AsyncOpenAI, AzureOpenAI, OpenAI
from openai.types.fine_tuning import FineTuningJob
from litellm._logging import verbose_logger
from litellm.types.utils import LiteLLMFineTuningJob
_AZURE_STATUS_MAP: Final = {
_AZURE_STATUS_MAP: Final[Mapping[object, str]] = {
"pending": "queued",
"notRunning": "queued",
"running": "running",
@ -20,7 +21,7 @@ _AZURE_STATUS_MAP: Final = {
# because LiteLLMFineTuningJob schema has no intermediate cancellation state.
def _normalize_fine_tuning_job_dict(data: dict[str, Any], is_azure: bool = False) -> dict[str, Any]:
def _normalize_fine_tuning_job_dict(data: dict[str, object], is_azure: bool = False) -> dict[str, object]:
"""
Normalize Azure OpenAI FineTuningJob response to match OpenAI schema.
@ -47,7 +48,7 @@ def _normalize_fine_tuning_job_dict(data: dict[str, Any], is_azure: bool = False
return normalized
def _litellm_fine_tuning_job_from_response(response: Any, is_azure: bool = False) -> LiteLLMFineTuningJob:
def _litellm_fine_tuning_job_from_response(response: FineTuningJob, is_azure: bool = False) -> LiteLLMFineTuningJob:
return LiteLLMFineTuningJob(**_normalize_fine_tuning_job_dict(response.model_dump(), is_azure=is_azure))
@ -111,7 +112,7 @@ class OpenAIFineTuningAPI:
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,
@ -159,7 +160,7 @@ class OpenAIFineTuningAPI:
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,
@ -258,7 +259,7 @@ class OpenAIFineTuningAPI:
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,

4
litellm/llms/openai/image_variations/handler.py
View file

@ -104,7 +104,7 @@ class OpenAIImageVariationsHandler:
status_code: Final = getattr(e, "status_code", 500)
error_headers = getattr(e, "headers", None)
error_text: Final = getattr(e, "text", str(e))
error_response: Final = getattr(e, "response", None)
error_response: Final[object] = getattr(e, "response", None)
if error_headers is None and error_response:
error_headers = getattr(error_response, "headers", None)
raise OpenAIError(status_code=status_code, message=error_text, headers=error_headers)
@ -221,7 +221,7 @@ class OpenAIImageVariationsHandler:
status_code: Final = getattr(e, "status_code", 500)
error_headers = getattr(e, "headers", None)
error_text: Final = getattr(e, "text", str(e))
error_response: Final = getattr(e, "response", None)
error_response: Final[object] = getattr(e, "response", None)
if error_headers is None and error_response:
error_headers = getattr(error_response, "headers", None)
raise OpenAIError(status_code=status_code, message=error_text, headers=error_headers)

9
litellm/llms/openai/realtime/handler.py
View file

@ -4,6 +4,7 @@ This file contains the calling OpenAI's `/v1/realtime` endpoint.
This requires websockets, and is currently only supported on LiteLLM Proxy.
"""
import ssl
from typing import Any, Final, cast
from litellm._logging import _redact_string, verbose_logger
@ -56,7 +57,7 @@ class OpenAIRealtime(OpenAIChatCompletion):
headers["OpenAI-Beta"] = "realtime=v1"
return headers
def _get_ssl_config(self, url: str) -> Any:
def _get_ssl_config(self, url: str) -> bool | str | ssl.SSLContext | None:
"""
Get SSL configuration for WebSocket connection.
Override this in subclasses to customize SSL behavior.
@ -111,12 +112,12 @@ class OpenAIRealtime(OpenAIChatCompletion):
logging_obj: LiteLLMLogging,
api_base: str | None = None,
api_key: str | None = None,
client: Any | None = None,
client: object | None = None,
timeout: float | None = None,
query_params: RealtimeQueryParams | None = None,
user_api_key_dict: Any | None = None,
user_api_key_dict: object | None = None,
litellm_metadata: dict | None = None,
**kwargs: Any,
**kwargs: object,
):
import websockets
from websockets.asyncio.client import ClientConnection

8
litellm/llms/openai/responses/count_tokens/token_counter.py
View file

@ -32,12 +32,12 @@ class OpenAITokenCounter(BaseTokenCounter):
async def count_tokens(
self,
model_to_use: str,
messages: list[dict[str, Any]] | None,
contents: list[dict[str, Any]] | None,
messages: list[dict[str, object]] | None,
contents: list[dict[str, object]] | None,
deployment: dict[str, Any] | None = None,
request_model: str = "",
tools: list[dict[str, Any]] | None = None,
system: Any | None = None,
tools: list[dict[str, object]] | None = None,
system: object = None,
) -> TokenCountResponse | None:
"""
Count tokens using OpenAI's Responses API /input_tokens endpoint.

25
litellm/llms/openai/vector_store_files/transformation.py
View file

@ -1,4 +1,5 @@
from typing import Any, Final, cast
from collections.abc import Mapping
from typing import Final, cast
import httpx
@ -22,7 +23,7 @@ from litellm.types.vector_store_files import (
from litellm.utils import add_openai_metadata
def _clean_dict(source: dict[str, Any]) -> dict[str, Any]:
def _clean_dict(source: Mapping[str, object]) -> dict[str, object]:
return {k: v for k, v in source.items() if v is not None}
@ -30,7 +31,7 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
ASSISTANTS_HEADER_KEY = "OpenAI-Beta"
ASSISTANTS_HEADER_VALUE = "assistants=v2"
def get_auth_credentials(self, litellm_params: dict[str, Any]) -> VectorStoreFileAuthCredentials:
def get_auth_credentials(self, litellm_params: Mapping[str, object]) -> VectorStoreFileAuthCredentials:
api_key: Final = litellm_params.get("api_key")
if api_key is None:
raise ValueError("api_key is required")
@ -82,7 +83,7 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
*,
api_base: str | None,
vector_store_id: str,
litellm_params: dict[str, Any],
litellm_params: Mapping[str, object],
) -> str:
base_url = (
api_base
@ -101,8 +102,8 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
vector_store_id: str,
create_request: VectorStoreFileCreateRequest,
api_base: str,
) -> tuple[str, dict[str, Any]]:
payload: Final[dict[str, Any]] = _clean_dict(dict(create_request))
) -> tuple[str, dict[str, object]]:
payload: Final[dict[str, object]] = _clean_dict(dict(create_request))
attributes: Final = payload.get("attributes")
if isinstance(attributes, dict):
filtered_attributes: Final = add_openai_metadata(attributes)
@ -133,7 +134,7 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
vector_store_id: str,
query_params: VectorStoreFileListQueryParams,
api_base: str,
) -> tuple[str, dict[str, Any]]:
) -> tuple[str, dict[str, object]]:
params: Final = _clean_dict(dict(query_params))
return api_base, params
@ -157,7 +158,7 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
vector_store_id: str,
file_id: str,
api_base: str,
) -> tuple[str, dict[str, Any]]:
) -> tuple[str, dict[str, object]]:
encoded_file_id: Final = encode_url_path_segment(file_id, field_name="file_id")
return f"{api_base}/{encoded_file_id}", {}
@ -181,7 +182,7 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
vector_store_id: str,
file_id: str,
api_base: str,
) -> tuple[str, dict[str, Any]]:
) -> tuple[str, dict[str, object]]:
encoded_file_id: Final = encode_url_path_segment(file_id, field_name="file_id")
return f"{api_base}/{encoded_file_id}/content", {}
@ -206,8 +207,8 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
file_id: str,
update_request: VectorStoreFileUpdateRequest,
api_base: str,
) -> tuple[str, dict[str, Any]]:
payload: Final[dict[str, Any]] = dict(update_request)
) -> tuple[str, dict[str, object]]:
payload: Final[dict[str, object]] = dict(update_request)
attributes: Final = payload.get("attributes")
if isinstance(attributes, dict):
filtered_attributes: Final = add_openai_metadata(attributes)
@ -238,7 +239,7 @@ class OpenAIVectorStoreFilesConfig(BaseVectorStoreFilesConfig):
vector_store_id: str,
file_id: str,
api_base: str,
) -> tuple[str, dict[str, Any]]:
) -> tuple[str, dict[str, object]]:
encoded_file_id: Final = encode_url_path_segment(file_id, field_name="file_id")
return f"{api_base}/{encoded_file_id}", {}

25
litellm/llms/predibase/chat/transformation.py
View file

@ -18,6 +18,8 @@ from litellm.types.utils import Choices, Message, ModelResponse, Usage
from ..common_utils import PredibaseError
if TYPE_CHECKING:
import tiktoken
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
LiteLLMLoggingObj = _LiteLLMLoggingObj
@ -64,8 +66,23 @@ class PredibaseConfig(BaseConfig):
typical_p: float | None = None,
watermark: bool | None = None,
) -> None:
locals_: Final = locals().copy()
for key, value in locals_.items():
locals_: Final = (
("best_of", best_of),
("decoder_input_details", decoder_input_details),
("details", details),
("max_new_tokens", max_new_tokens),
("repetition_penalty", repetition_penalty),
("return_full_text", return_full_text),
("seed", seed),
("stop", stop),
("temperature", temperature),
("top_k", top_k),
("top_p", top_p),
("truncate", truncate),
("typical_p", typical_p),
("watermark", watermark),
)
for key, value in locals_:
if key != "self" and value is not None:
setattr(self.__class__, key, value)
@ -133,7 +150,7 @@ class PredibaseConfig(BaseConfig):
messages: list[AllMessageValues],
optional_params: dict,
litellm_params: dict,
encoding: Any,
encoding: "tiktoken.Encoding | None",
api_key: str | None = None,
json_mode: bool | None = None,
) -> ModelResponse:
@ -217,7 +234,7 @@ class PredibaseConfig(BaseConfig):
# Keep usage calculation non-blocking if token counting fails.
pass
output_text: Final = model_response["choices"][0]["message"].get("content", "")
if output_text is not None and len(output_text) > 0:
if encoding is not None and output_text is not None and len(output_text) > 0:
completion_tokens = 0
try:
completion_tokens = len(encoding.encode(model_response["choices"][0]["message"].get("content", "")))

6
litellm/llms/soniox/audio_transcription/transformation.py
View file

@ -152,7 +152,7 @@ class SonioxAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
and for filling in `file_id`/`audio_url`. This method exists so the
config can be exercised in isolation by unit tests.
"""
body: Final[dict[str, Any]] = {"model": model}
body: Final[dict[str, object]] = {"model": model}
for key in SONIOX_PASSTHROUGH_PARAMS:
value = optional_params.get(key)
@ -247,9 +247,9 @@ class SonioxAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
# For verbose_json, include word-level timing from tokens.
if response_format == "verbose_json" and tokens:
words: Final[list[dict[str, Any]]] = []
words: Final[list[dict[str, object]]] = []
for token in tokens:
word_entry: dict[str, Any] = {"word": token.get("text", "")}
word_entry: dict[str, object] = {"word": token.get("text", "")}
if token.get("start_ms") is not None:
word_entry["start"] = float(token["start_ms"]) / 1000.0
if token.get("end_ms") is not None:

8
litellm/llms/vertex_ai/rag_engine/ingestion.py
View file

@ -14,7 +14,7 @@ Key differences from OpenAI:
from __future__ import annotations
import os
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Final
from litellm import get_secret_str
from litellm._logging import verbose_logger
@ -26,12 +26,12 @@ if TYPE_CHECKING:
from litellm.types.rag import RAGIngestOptions
def _get_str_or_none(value: Any) -> str | None:
def _get_str_or_none(value: object) -> str | None:
"""Cast config value to Optional[str]."""
return str(value) if value is not None else None
def _get_int(value: Any, default: int) -> int:
def _get_int(value: str | float | None, default: int) -> int:
"""Cast config value to int with default."""
if value is None:
return default
@ -205,7 +205,7 @@ class VertexAIRAGIngestion(BaseRAGIngestion):
)
verbose_logger.info("Import started asynchronously")
def _build_transformation_config(self) -> Any:
def _build_transformation_config(self) -> object:
"""
Build Vertex AI TransformationConfig from unified chunking_strategy.

4
litellm/llms/vertex_ai/videos/transformation.py
View file

@ -265,7 +265,9 @@ class VertexAIVideoConfig(BaseVideoConfig, VertexBase):
# Extract Vertex AI parameters using safe helpers from VertexBase
# Use safe_get_* methods that don't mutate litellm_params dict
# Ensure litellm_params is a dict for type checking
params_dict: Final[dict[str, Any]] = cast(dict[str, Any], litellm_params) if litellm_params is not None else {}
params_dict: Final[dict[str, object]] = (
cast(dict[str, object], litellm_params) if litellm_params is not None else {}
)
vertex_project: Final = VertexBase.safe_get_vertex_ai_project(litellm_params=params_dict)
vertex_credentials: Final = VertexBase.safe_get_vertex_ai_credentials(litellm_params=params_dict)

6
litellm/llms/voyage/embedding/transformation_multimodal.py
View file

@ -6,7 +6,7 @@ containing content blocks, unlike standard Voyage embeddings which use
/v1/embeddings and a string/list `input` field.
"""
from typing import Any, Final
from typing import Final
import httpx
@ -98,7 +98,7 @@ class VoyageMultimodalEmbeddingConfig(BaseEmbeddingConfig):
)
return {"Authorization": f"Bearer {api_key}"}
def _normalize_content_item(self, item: dict[str, Any]) -> dict[str, Any]:
def _normalize_content_item(self, item: dict[str, object]) -> dict[str, object]:
item_type: Final = item.get("type")
if item_type == "image_url":
image_url = item.get("image_url")
@ -115,7 +115,7 @@ class VoyageMultimodalEmbeddingConfig(BaseEmbeddingConfig):
return {"type": "image_url", "image_url": image_url}
return item
def _normalize_input_item(self, item: Any) -> dict[str, Any]:
def _normalize_input_item(self, item: object) -> object:
if isinstance(item, str):
return {"content": [{"type": "text", "text": item}]}
if isinstance(item, dict) and "content" in item:

4
litellm/llms/voyage/rerank/transformation.py
View file

@ -43,7 +43,7 @@ class VoyageRerankConfig(BaseRerankConfig):
instruction: str | None = None,
) -> dict:
# Voyage AI uses 'top_k' instead of 'top_n'
optional_params: Final[dict[str, Any]] = {"query": query, "documents": documents}
optional_params: Final[dict[str, object]] = {"query": query, "documents": documents}
if top_n is not None:
optional_params["top_k"] = top_n
if return_documents is not None:
@ -109,7 +109,7 @@ class VoyageRerankConfig(BaseRerankConfig):
# Transform to LiteLLM format
transformed_results: Final = []
for result in _results:
transformed_result: dict[str, Any] = {
transformed_result: dict[str, object] = {
"index": result["index"],
"relevance_score": result["relevance_score"],
}

2
litellm/llms/xai/chat/transformation.py
View file

@ -196,7 +196,7 @@ class XAIChatConfig(OpenAIGPTConfig):
streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
sync_stream: bool,
json_mode: bool | None = False,
) -> Any:
) -> "XAIChatCompletionStreamingHandler":
return XAIChatCompletionStreamingHandler(
streaming_response=streaming_response,
sync_stream=sync_stream,

14
litellm/llms/xai/responses/transformation.py
View file

@ -1,4 +1,5 @@
from typing import Any, Final
from collections.abc import Mapping
from typing import Final
import litellm
from litellm._logging import verbose_logger
@ -8,7 +9,6 @@ from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfi
from litellm.llms.xai.common_utils import XAIModelInfo
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import ResponsesAPIOptionalRequestParams
from litellm.types.llms.xai import XAIWebSearchTool, XAIXSearchTool
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
@ -44,7 +44,7 @@ class XAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
return supported_params
def _transform_web_search_tool(self, tool: dict[str, Any]) -> XAIWebSearchTool | dict[str, Any]:
def _transform_web_search_tool(self, tool: Mapping[str, object]) -> Mapping[str, object]:
"""
Transform web_search tool to XAI format.
@ -55,7 +55,7 @@ class XAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
XAI does NOT support search_context_size (OpenAI-specific).
"""
xai_tool: Final[dict[str, Any]] = {"type": "web_search"}
xai_tool: Final[dict[str, object]] = {"type": "web_search"}
# Remove search_context_size if present (not supported by XAI)
if "search_context_size" in tool:
@ -83,7 +83,7 @@ class XAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
return xai_tool
def _transform_x_search_tool(self, tool: dict[str, Any]) -> XAIXSearchTool | dict[str, Any]:
def _transform_x_search_tool(self, tool: Mapping[str, object]) -> Mapping[str, object]:
"""
Transform x_search tool to XAI format.
@ -95,7 +95,7 @@ class XAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
- enable_image_understanding
- enable_video_understanding
"""
xai_tool: Final[dict[str, Any]] = {"type": "x_search"}
xai_tool: Final[dict[str, object]] = {"type": "x_search"}
# Handle allowed_x_handles
if "allowed_x_handles" in tool:
@ -157,7 +157,7 @@ class XAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
if not isinstance(tools_list, list):
tools_list = [tools_list]
transformed_tools: Final[list[Any]] = []
transformed_tools: Final[list[object]] = []
for tool in tools_list:
if isinstance(tool, dict):
tool_type = tool.get("type")

10
litellm/proxy/caching_routes.py
View file

@ -1,4 +1,4 @@
from typing import Any, Final
from typing import Final
from fastapi import APIRouter, Depends, HTTPException, Request
@ -19,7 +19,7 @@ router: Final = APIRouter(
)
def _extract_cache_params() -> dict[str, Any]:
def _extract_cache_params() -> dict[str, object]:
"""
Safely extracts and cleans cache parameters.
@ -56,8 +56,8 @@ async def cache_ping():
"""
Endpoint for checking if cache can be pinged
"""
litellm_cache_params: dict[str, Any] = {}
cleaned_cache_params: dict[str, Any] = {}
litellm_cache_params: dict[str, object] = {}
cleaned_cache_params: dict[str, object] = {}
if litellm.cache is None:
raise ProxyException(
message=safe_dumps(
@ -162,7 +162,7 @@ async def cache_delete(request: Request):
)
def _get_redis_client_info(cache_instance) -> tuple[list, int]:
def _get_redis_client_info(cache_instance: RedisCache) -> tuple[list[object], int]:
"""
Helper function to safely get Redis client list information.

4
litellm/proxy/client/chat.py
View file

@ -73,7 +73,7 @@ class ChatClient:
url: Final = f"{self._base_url}/chat/completions"
# Build request data with required fields
data: Final[dict[str, Any]] = {"model": model, "messages": messages}
data: Final[dict[str, object]] = {"model": model, "messages": messages}
# Add optional parameters if provided
if temperature is not None:
@ -143,7 +143,7 @@ class ChatClient:
url: Final = f"{self._base_url}/chat/completions"
# Build request data with required fields
data: Final[dict[str, Any]] = {"model": model, "messages": messages, "stream": True}
data: Final[dict[str, object]] = {"model": model, "messages": messages, "stream": True}
# Add optional parameters if provided
if temperature is not None:

10
litellm/proxy/client/cli/commands/agents.py
View file

@ -8,7 +8,7 @@ from typing import Final
import click
import requests
from .auth import context_secret_vault, get_stored_api_key, login
from .auth import CliContextObj, context_secret_vault, get_stored_api_key, login
from .cmd_quoting import quote_for_cmd
ANTHROPIC_BASE_URL_ENV: Final = "ANTHROPIC_BASE_URL"
@ -289,8 +289,9 @@ def _is_interactive() -> bool:
def resolve_api_key(ctx: click.Context) -> str:
base_url: Final = ctx.obj["base_url"]
api_key = ctx.obj.get("api_key")
ctx_obj: Final[CliContextObj] = ctx.obj
base_url: Final = ctx_obj["base_url"]
api_key = ctx_obj.get("api_key")
if api_key:
return api_key
@ -312,7 +313,8 @@ _SKIP_VERIFY_HELP: Final = "Skip the pre-launch key check against the proxy."
def _launch(ctx: click.Context, binary: str, args: Sequence[str], *, skip_verify: bool) -> None:
base_url: Final = ctx.obj["base_url"]
ctx_obj: Final[CliContextObj] = ctx.obj
base_url: Final = ctx_obj["base_url"]
started_interactive: Final = _is_interactive()
api_key: Final = resolve_api_key(ctx)

17
litellm/proxy/client/keys.py
View file

@ -1,4 +1,5 @@
import builtins
from collections.abc import Mapping
from typing import Any, Final
import requests
@ -72,7 +73,7 @@ class KeysManagementClient:
requests.exceptions.RequestException: If the request fails with any other error
"""
url: Final = f"{self._base_url}/key/list"
params: Final[dict[str, Any]] = {}
params: Final[dict[str, int | str]] = {}
# Add optional query parameters
if page is not None:
@ -119,9 +120,9 @@ class KeysManagementClient:
team_id: str | None = None,
user_id: str | None = None,
budget_id: str | None = None,
config: dict[str, Any] | None = None,
config: Mapping[str, object] | None = None,
return_request: bool = False,
) -> dict[str, Any] | requests.Request:
) -> dict[str, object] | requests.Request:
"""
Generate an API key based on the provided data.
@ -149,7 +150,7 @@ class KeysManagementClient:
"""
url: Final = f"{self._base_url}/key/generate"
data: Final[dict[str, Any]] = {}
data: Final[dict[str, object]] = {}
if models is not None:
data["models"] = models
if aliases is not None:
@ -189,7 +190,7 @@ class KeysManagementClient:
keys: builtins.list[str] | None = None,
key_aliases: builtins.list[str] | None = None,
return_request: bool = False,
) -> dict[str, Any] | requests.Request:
) -> dict[str, object] | requests.Request:
"""
Delete existing keys
@ -238,7 +239,7 @@ class KeysManagementClient:
key_alias: str | None = None,
team_id: str | None = None,
user_id: str | None = None,
) -> dict[str, Any] | requests.Request:
) -> dict[str, object] | requests.Request:
"""
Update an existing API key's parameters.
@ -261,7 +262,7 @@ class KeysManagementClient:
"""
url: Final = f"{self._base_url}/key/update"
data: Final[dict[str, Any]] = {"key": key}
data: Final[dict[str, object]] = {"key": key}
if key_alias is not None:
data["key_alias"] = key_alias
@ -288,7 +289,7 @@ class KeysManagementClient:
except Exception:
raise Exception(f"Error updating key: {response_text}")
def info(self, key: str, return_request: bool = False) -> dict[str, Any] | requests.Request:
def info(self, key: str, return_request: bool = False) -> dict[str, object] | requests.Request:
"""
Get information about API keys.

20
litellm/proxy/common_utils/performance_utils.py
View file

@ -15,10 +15,24 @@ import inspect
import threading
from collections.abc import Callable
from pathlib import Path as PathLib
from typing import Any, Final
from types import ModuleType
from typing import Final, Protocol, TextIO
from litellm._logging import verbose_proxy_logger
class _LineProfiler(Protocol):
"""The line_profiler.LineProfiler surface this module drives."""
def __call__(self, func: Callable[..., object]) -> Callable[..., object]: ...
def add_function(self, func: Callable[..., object]) -> object: ...
def dump_stats(self, filename: str) -> object: ...
def print_stats(self, stream: TextIO) -> object: ...
# Global profiling state
_profile_lock: Final = threading.Lock()
_profiler = None
@ -27,7 +41,7 @@ _sample_counter = 0
_sample_counter_lock: Final = threading.Lock()
# Global line_profiler state
_line_profiler: Any | None = None
_line_profiler: _LineProfiler | None = None
_line_profiler_lock: Final = threading.Lock()
_wrapped_functions: Final[dict[str, Callable]] = {} # Store original functions
@ -157,7 +171,7 @@ def enable_line_profiler() -> None:
verbose_proxy_logger.info("Line profiler enabled")
def wrap_function_with_line_profiler(module: Any, function_name: str) -> bool:
def wrap_function_with_line_profiler(module: ModuleType, function_name: str) -> bool:
"""Dynamically wrap a function with line_profiler.
Args:

8
litellm/proxy/container_endpoints/endpoints.py
View file

@ -1,6 +1,6 @@
#### Container Endpoints #####
from typing import Any, Final
from typing import Final
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi.responses import ORJSONResponse
@ -208,7 +208,7 @@ async def list_containers(
# Read query parameters
query_params: Final = dict(request.query_params)
data: Final[dict[str, Any]] = {"query_params": query_params}
data: Final[dict[str, object]] = {"query_params": query_params}
# Extract custom_llm_provider using priority chain
custom_llm_provider: Final = (
@ -312,7 +312,7 @@ async def retrieve_container(
)
# Include container_id in request data
data: Final[dict[str, Any]] = {"container_id": container_id}
data: Final[dict[str, object]] = {"container_id": container_id}
# Extract custom_llm_provider using priority chain
custom_llm_provider = (
@ -417,7 +417,7 @@ async def delete_container(
)
# Include container_id in request data
data: Final[dict[str, Any]] = {"container_id": container_id}
data: Final[dict[str, object]] = {"container_id": container_id}
# Extract custom_llm_provider using priority chain
custom_llm_provider = (

4
litellm/proxy/db/exception_handler.py
View file

@ -204,7 +204,7 @@ class PrismaDBExceptionHandler:
if isinstance(e, prisma.errors.PrismaError):
return False
tb = getattr(e, "__traceback__", None)
tb = e.__traceback__ if hasattr(e, "__traceback__") else None
while tb is not None:
if tb.tb_frame.f_globals.get("__name__", "").startswith("prisma.engine"):
return True
@ -318,7 +318,7 @@ _DEFAULT_RECONNECT_TIMEOUT_SECONDS: Final = 2.0
_DEFAULT_RECONNECT_LOCK_TIMEOUT_SECONDS: Final = 0.1
def _coerce_timeout(value: Any, fallback: float) -> float:
def _coerce_timeout(value: object, fallback: float) -> float:
"""Return `value` if it is a real int/float, else `fallback`. Guards
against tests that mock `prisma_client` and leave the timeout slots as
MagicMock instances."""

4
litellm/proxy/guardrails/guardrail_hooks/azure/base.py
View file

@ -45,7 +45,7 @@ class AzureGuardrailBase:
self.api_base = api_base
self.api_version: str = kwargs.get("api_version") or "2024-09-01"
async def _post_to_content_safety(self, endpoint_path: str, request_body: dict[str, Any]) -> dict[str, Any]:
async def _post_to_content_safety(self, endpoint_path: str, request_body: dict[str, object]) -> dict[str, Any]:
"""POST to an Azure Content Safety endpoint with standard auth headers.
Args:
@ -94,7 +94,7 @@ class AzureGuardrailBase:
# Tokenize into alternating non-whitespace and whitespace runs so
# that original newlines, tabs, and multiple spaces are preserved
# within each chunk.
tokens: Final = re.findall(r"\S+|\s+", text)
tokens: Final = [match.group(0) for match in re.finditer(r"\S+|\s+", text)]
chunks: Final[list[str]] = []
current_chunk = ""

18
litellm/proxy/guardrails/guardrail_hooks/azure/text_moderation.py
View file

@ -3,7 +3,7 @@
Azure Text Moderation Native Guardrail Integrationfor LiteLLM
"""
from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, Union, cast
from typing import TYPE_CHECKING, Any, ClassVar, Final, Literal, cast
from fastapi import HTTPException
@ -14,18 +14,18 @@ from litellm.integrations.custom_guardrail import (
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import CallTypesLiteral, GenericGuardrailAPIInputs
from litellm.types.utils import CallTypesLiteral, GenericGuardrailAPIInputs, LLMResponseTypes
from .base import AzureGuardrailBase
if TYPE_CHECKING:
from litellm.caching.caching import DualCache
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.types.llms.openai import AllMessageValues
from litellm.types.proxy.guardrails.guardrail_hooks.azure.azure_text_moderation import (
AzureTextModerationGuardrailResponse,
)
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
from litellm.types.utils import EmbeddingResponse, ImageResponse, ModelResponse
class AzureContentSafetyTextModerationGuardrail(AzureGuardrailBase, CustomGuardrail):
@ -219,10 +219,10 @@ class AzureContentSafetyTextModerationGuardrail(AzureGuardrailBase, CustomGuardr
async def async_pre_call_hook(
self,
user_api_key_dict: "UserAPIKeyAuth",
cache: Any,
cache: "DualCache",
data: dict[str, Any],
call_type: CallTypesLiteral,
) -> dict[str, Any] | None:
) -> dict[str, object] | None:
"""
Pre-call hook to scan user prompts before sending to LLM.
@ -251,8 +251,8 @@ class AzureContentSafetyTextModerationGuardrail(AzureGuardrailBase, CustomGuardr
self,
data: dict,
user_api_key_dict: "UserAPIKeyAuth",
response: Union[Any, "ModelResponse", "EmbeddingResponse", "ImageResponse"],
) -> Any:
response: LLMResponseTypes,
) -> LLMResponseTypes:
from litellm.types.utils import Choices, ModelResponse
if isinstance(response, ModelResponse) and response.choices:
@ -267,7 +267,7 @@ class AzureContentSafetyTextModerationGuardrail(AzureGuardrailBase, CustomGuardr
)
return response
async def async_post_call_streaming_hook(self, user_api_key_dict: UserAPIKeyAuth, response: str) -> Any:
async def async_post_call_streaming_hook(self, user_api_key_dict: UserAPIKeyAuth, response: str) -> str:
try:
if response is not None and len(response) > 0:
await self.async_make_request(
@ -281,7 +281,7 @@ class AzureContentSafetyTextModerationGuardrail(AzureGuardrailBase, CustomGuardr
return f"data: {error_returned}\n\n"
def _message_content_to_text(content: Any) -> str:
def _message_content_to_text(content: object) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):

6
litellm/proxy/guardrails/guardrail_hooks/block_code_execution/__init__.py
View file

@ -1,6 +1,6 @@
"""Block Code Execution guardrail: blocks or masks fenced code blocks by language."""
from typing import TYPE_CHECKING, Any, Final, Literal, cast
from typing import TYPE_CHECKING, Final, Literal, cast
from litellm.types.guardrails import GuardrailEventHooks, SupportedGuardrailIntegrations
@ -20,8 +20,8 @@ def _get_param(
litellm_params: "LitellmParams",
guardrail: "Guardrail",
key: str,
default: Any = None,
) -> Any:
default: object = None,
) -> object:
"""Get a param from litellm_params, with fallback to raw guardrail litellm_params (for extra fields not on LitellmParams)."""
value: Final = getattr(litellm_params, key, default)
if value is not None:

6
litellm/proxy/guardrails/guardrail_hooks/custom_code/primitives.py
View file

@ -8,7 +8,7 @@ and provide safe, sandboxed functionality for common guardrail operations.
import json
import re
from collections.abc import Mapping, Sequence
from typing import Any, Final
from typing import Final
from urllib.parse import urlparse
import httpx
@ -16,7 +16,7 @@ from pydantic import JsonValue
from typing_extensions import ReadOnly, TypedDict
from litellm._logging import verbose_proxy_logger
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, get_async_httpx_client
from litellm.types.llms.custom_http import httpxSpecialProvider
# =============================================================================
@ -508,7 +508,7 @@ async def http_request(
async def _execute_http_request(
client: Any,
client: AsyncHTTPHandler,
method: str,
url: str,
headers: dict[str, str] | None,

10
litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/generic_guardrail_api.py
View file

@ -7,6 +7,7 @@
import fnmatch
import os
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final, Literal, Optional
import httpx
@ -23,6 +24,7 @@ from litellm.llms.custom_httpx.http_handler import (
httpxSpecialProvider,
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.llms.openai import ChatCompletionToolParam
from litellm.types.proxy.guardrails.guardrail_hooks.generic_guardrail_api import (
GenericGuardrailAPIMetadata,
GenericGuardrailAPIRequest,
@ -73,7 +75,7 @@ def _header_value_allowed(
def _sanitize_inbound_headers(
headers: Any,
headers: object,
extra_allowlist: set[str] | None = None,
) -> dict[str, str] | None:
"""
@ -175,7 +177,7 @@ class GenericGuardrailAPI(CustomGuardrail):
headers: dict[str, Any] | None = None,
api_base: str | None = None,
api_key: str | None = None,
additional_provider_specific_params: dict[str, Any] | None = None,
additional_provider_specific_params: Mapping[str, object] | None = None,
unreachable_fallback: Literal["fail_closed", "fail_open"] = "fail_closed",
fail_on_error: bool | None = True,
extra_headers: list | None = None,
@ -318,8 +320,8 @@ class GenericGuardrailAPI(CustomGuardrail):
self,
*,
texts: list,
images: Any,
tools: Any,
images: list[str] | None,
tools: list[ChatCompletionToolParam] | None,
guardrail_response: GenericGuardrailAPIResponse,
) -> GenericGuardrailAPIInputs:
# Action is NONE or no modifications needed

5
litellm/proxy/guardrails/guardrail_hooks/model_armor/model_armor.py
View file

@ -33,6 +33,7 @@ from litellm.proxy.guardrails.guardrail_hooks.model_armor.file_scanning import (
)
from litellm.types.guardrails import GuardrailEventHooks, LitellmParams
from litellm.types.llms.openai import AllMessageValues
from litellm.types.llms.vertex_ai import VERTEX_CREDENTIALS_TYPES
from litellm.types.utils import (
CallTypes,
CallTypesLiteral,
@ -97,7 +98,7 @@ class ModelArmorGuardrail(CustomGuardrail, VertexBase):
template_id: str | None = None,
project_id: str | None = None,
location: str | None = None,
credentials: Any | None = None,
credentials: VERTEX_CREDENTIALS_TYPES | None = None,
api_endpoint: str | None = None,
sanitize_error_detail: "bool | None" = True,
**kwargs,
@ -147,7 +148,7 @@ class ModelArmorGuardrail(CustomGuardrail, VertexBase):
else:
return {"modelResponseData": {"text": content}}
def _extract_content_from_response(self, response: Any | ModelResponse) -> str:
def _extract_content_from_response(self, response: object) -> str:
"""
Extract text content from model response.

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

@ -72,7 +72,7 @@ class NomaBlockedMessage(HTTPException):
},
)
def _is_result_true(self, result_obj: dict[str, Any] | None) -> bool:
def _is_result_true(self, result_obj: dict[str, object] | None) -> bool:
"""
Check if a result object has a "result" field that is True.
@ -454,7 +454,7 @@ class NomaGuardrail(CustomGuardrail):
return False
def _is_result_true(self, result_obj: dict[str, Any] | None) -> bool:
def _is_result_true(self, result_obj: dict[str, object] | None) -> bool:
"""
Check if a result object has a "result" field that is True.

6
litellm/proxy/guardrails/guardrail_hooks/pangea/pangea.py
View file

@ -35,14 +35,14 @@ class PangeaGuardrailMissingSecrets(Exception):
class _TextCompletionRequest:
def __init__(self, body):
def __init__(self, body: dict[str, object]) -> None:
self.body = body
def get_messages(self) -> list[dict]:
return [{"role": "user", "content": self.body["prompt"]}]
# This mutates the original dict, but we'll still return it anyways
def update_original_body(self, prompt_messages: list[dict]) -> Any:
def update_original_body(self, prompt_messages: list[dict]) -> dict[str, object]:
assert len(prompt_messages) == 1
self.body["prompt"] = prompt_messages[0]["content"]
return self.body
@ -159,7 +159,7 @@ class PangeaHandler(CustomGuardrail):
call_type: str,
):
transformer = None
messages: Any = None
messages: object = None
if call_type == "text_completion" or call_type == "atext_completion":
transformer = _TextCompletionRequest(data)
messages = transformer.get_messages()

2
litellm/proxy/guardrails/guardrail_hooks/panw_prisma_airs/panw_prisma_airs.py
View file

@ -721,7 +721,7 @@ class PanwPrismaAirsHandler(CustomGuardrail):
}
}
def _record_scan_id(self, request_data: dict[str, Any], scan_result: Mapping[str, object]) -> None:
def _record_scan_id(self, request_data: dict[str, object], scan_result: Mapping[str, object]) -> None:
"""Surface the AIRS scan id on the response, so allowed calls are auditable too."""
scan_id: Final = scan_result.get("scan_id")
add_guardrail_scan_id(request_data=request_data, scan_id=str(scan_id) if scan_id else None)

16
litellm/proxy/guardrails/usage_tracking.py
View file

@ -172,7 +172,7 @@ def _guardrail_status_to_action(status: str | None) -> str:
return "passed"
def _parse_guardrail_info_from_payload(payload: Mapping[str, Any]) -> Sequence[Mapping[str, Any]]:
def _parse_guardrail_info_from_payload(payload: Mapping[str, object]) -> Sequence[Mapping[str, Any]]:
"""Extract guardrail_information from spend log payload metadata."""
meta = payload.get("metadata")
if not meta:
@ -197,7 +197,7 @@ def _date_str(dt: datetime) -> str:
return dt.astimezone(timezone.utc).strftime("%Y-%m-%d")
def _parse_payload_start_time(payload: Mapping[str, Any]) -> datetime | None:
def _parse_payload_start_time(payload: Mapping[str, object]) -> datetime | None:
start_time: Final = payload.get("startTime")
if isinstance(start_time, datetime):
return start_time
@ -209,7 +209,9 @@ def _parse_payload_start_time(payload: Mapping[str, Any]) -> datetime | None:
return None
def _iter_usage_unit_increments(logs_to_process: Sequence[Mapping[str, Any]]) -> Iterator[tuple[_UsageUnitKey, int]]:
def _iter_usage_unit_increments(
logs_to_process: Sequence[Mapping[str, object]],
) -> Iterator[tuple[_UsageUnitKey, int]]:
for payload in logs_to_process:
start_time = _parse_payload_start_time(payload)
if not payload.get("request_id") or start_time is None:
@ -227,7 +229,7 @@ def _iter_usage_unit_increments(logs_to_process: Sequence[Mapping[str, Any]]) ->
yield _UsageUnitKey(guardrail_id, date_key, team_id, api_key, str(unit_name)), units
def _sum_usage_unit_increments(logs_to_process: Sequence[Mapping[str, Any]]) -> Mapping[_UsageUnitKey, int]:
def _sum_usage_unit_increments(logs_to_process: Sequence[Mapping[str, object]]) -> Mapping[_UsageUnitKey, int]:
ordered: Final = sorted(_iter_usage_unit_increments(logs_to_process), key=itemgetter(0))
return MappingProxyType(
{key: sum(units for _, units in group) for key, group in groupby(ordered, key=itemgetter(0))}
@ -284,7 +286,7 @@ async def _upsert_metrics_row(prisma_client: PrismaClient, key: _MetricsKey, agg
async def process_spend_logs_guardrail_usage(
prisma_client: PrismaClient,
logs_to_process: list[dict[str, Any]],
logs_to_process: Sequence[Mapping[str, object]],
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
pending: PendingRollups = _PENDING_ROLLUPS,
) -> None:
@ -295,7 +297,7 @@ async def process_spend_logs_guardrail_usage(
if not logs_to_process:
return
# Aggregate daily metrics by (guardrail_id, date). Latency/score metrics dropped.
daily_guardrail: Final[dict[_MetricsKey, dict[str, Any]]] = defaultdict(
daily_guardrail: Final[dict[_MetricsKey, dict[str, int]]] = defaultdict(
lambda: {
"requests_evaluated": 0,
"passed_count": 0,
@ -303,7 +305,7 @@ async def process_spend_logs_guardrail_usage(
"flagged_count": 0,
}
)
index_rows: Final[list[dict[str, Any]]] = []
index_rows: Final[list[dict[str, object]]] = []
for payload in logs_to_process:
request_id = payload.get("request_id")

9
litellm/proxy/health_check_utils/shared_health_check_manager.py
View file

@ -1,6 +1,7 @@
import asyncio
import json
import time
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final
from litellm._logging import verbose_proxy_logger
@ -143,8 +144,8 @@ class SharedHealthCheckManager:
async def cache_health_check_results(
self,
healthy_endpoints: list[dict[str, Any]],
unhealthy_endpoints: list[dict[str, Any]],
healthy_endpoints: Sequence[Mapping[str, object]],
unhealthy_endpoints: Sequence[Mapping[str, object]],
) -> None:
"""
Cache health check results in Redis.
@ -336,14 +337,14 @@ class SharedHealthCheckManager:
verbose_proxy_logger.error("Error checking health check lock status: %s", str(e))
return False
async def get_health_check_status(self) -> dict[str, Any]:
async def get_health_check_status(self) -> dict[str, object]:
"""
Get the current status of health check coordination.
Returns:
Dict containing status information
"""
status: Final = {
status: Final[dict[str, object]] = {
"pod_id": self.pod_id,
"redis_available": self.redis_cache is not None,
"lock_ttl": self.lock_ttl,

8
litellm/proxy/hooks/batch_rate_limiter.py
View file

@ -62,6 +62,7 @@ from litellm.proxy.hooks.rate_limiter_utils import resolve_llm_provider_for_rate
if TYPE_CHECKING:
from opentelemetry.trace import Span as _Span
from litellm.caching.caching import DualCache
from litellm.proxy.hooks.parallel_request_limiter_v3 import (
RateLimitDescriptor as _RateLimitDescriptor,
)
@ -73,8 +74,9 @@ if TYPE_CHECKING:
)
from litellm.proxy.utils import InternalUsageCache as _InternalUsageCache
from litellm.router import Router as _Router
from litellm.types.llms.openai import HttpxBinaryResponseContent
Span = _Span | Any
Span = _Span
InternalUsageCache = _InternalUsageCache
Router = _Router
ParallelRequestLimiter = _ParallelRequestLimiter
@ -1011,7 +1013,7 @@ class _PROXY_BatchRateLimiter(CustomLogger):
self,
file_id: str,
user_api_key_dict: UserAPIKeyAuth,
) -> Any:
) -> "HttpxBinaryResponseContent":
"""
Fetch file content from managed files hook.
@ -1062,7 +1064,7 @@ class _PROXY_BatchRateLimiter(CustomLogger):
async def async_pre_call_hook(
self,
user_api_key_dict: UserAPIKeyAuth,
cache: Any,
cache: "DualCache",
data: dict,
call_type: str,
) -> Exception | str | dict | None:

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

@ -1,7 +1,7 @@
import asyncio
import json
from datetime import datetime, timezone
from typing import Any, Final
from typing import Final
import litellm
from litellm._logging import verbose_proxy_logger
@ -89,8 +89,8 @@ class KeyManagementEventHooks:
@staticmethod
async def async_key_updated_hook(
data: UpdateKeyRequest,
existing_key_row: Any,
response: Any,
existing_key_row: LiteLLM_VerificationToken,
response: object,
user_api_key_dict: UserAPIKeyAuth,
litellm_changed_by: str | None = None,
):

7
litellm/proxy/management_endpoints/common_utils.py
View file

@ -1,4 +1,5 @@
import math
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final, Optional, Union
from fastapi import HTTPException, status
@ -438,7 +439,7 @@ _TEAM_MEMBER_BUDGET_LIMIT_FIELDS: Final = (
)
def _is_set_budget_value(value: Any) -> bool:
def _is_set_budget_value(value: object) -> bool:
if value is None:
return False
if isinstance(value, list) and len(value) == 0:
@ -446,7 +447,7 @@ def _is_set_budget_value(value: Any) -> bool:
return True
def _has_meaningful_budget_limit(budget_values: dict[str, Any]) -> bool:
def _has_meaningful_budget_limit(budget_values: Mapping[str, object]) -> bool:
"""A budget is meaningful if at least one limit is actually set; an empty
list (no model restriction) and None both count as unset."""
return any(_is_set_budget_value(budget_values.get(field)) for field in _TEAM_MEMBER_BUDGET_LIMIT_FIELDS)
@ -590,7 +591,7 @@ def _update_metadata_field(updated_kv: dict, field_name: str) -> None:
updated_kv["metadata"] = {field_name: _value}
def _has_non_empty_value(value: Any) -> bool:
def _has_non_empty_value(value: object) -> bool:
"""Check if a value has real content (not None, not empty list, not blank string)."""
if value is None:
return False

11
litellm/proxy/realtime_endpoints/endpoints.py
View file

@ -2,7 +2,7 @@
import json
import time
from typing import Any, Final
from typing import TYPE_CHECKING, Any, Final
import httpx
from fastapi import APIRouter, Depends, HTTPException, Request, Response
@ -24,6 +24,9 @@ from litellm.types.realtime import (
RealtimeTranscriptionSessionResponse,
)
if TYPE_CHECKING:
from litellm.router import Router
router: Final = APIRouter()
_REALTIME_TOKEN_VERSION: Final = "realtime_v1"
@ -38,7 +41,7 @@ def _coerce_realtime_session_type(session_type: str | None) -> str:
return "realtime"
def _append_model_candidate(candidates: list[str], model: Any) -> None:
def _append_model_candidate(candidates: list[str], model: object) -> None:
if isinstance(model, str) and model and model not in candidates:
candidates.append(model)
@ -116,7 +119,7 @@ async def _prepare_client_secret_session(
req: RealtimeClientSecretRequest,
user_api_key_dict: UserAPIKeyAuth,
llm_model_list: list | None,
llm_router: Any,
llm_router: "Router | None",
) -> tuple[str, dict | None, str]:
session_type: Final = _coerce_realtime_session_type(req.session.type if req.session else None)
session_data: Final[dict | None] = req.session.model_dump(exclude_none=True) if req.session else None
@ -171,7 +174,7 @@ def _encode_realtime_token_payload(
Encode metadata with the upstream ephemeral key so /realtime/calls can
route without requiring model as a query param.
"""
payload: Final[dict[str, Any]] = {
payload: Final[dict[str, str | int | None]] = {
"v": _REALTIME_TOKEN_VERSION,
"ephemeral_key": ephemeral_key,
"model_id": model_id,

2
litellm/proxy/response_polling/polling_handler.py
View file

@ -89,7 +89,7 @@ class ResponsePollingHandler:
error: dict | None = None,
incomplete_details: dict | None = None,
reasoning: dict | None = None,
tool_choice: Any | None = None,
tool_choice: object | None = None,
tools: list | None = None,
output: list | None = None,
# Additional ResponsesAPIResponse fields

7
litellm/proxy/vector_store_endpoints/utils.py
View file

@ -1,6 +1,7 @@
import json
import re
from typing import Any, Final, Literal
from collections.abc import Mapping
from typing import Final, Literal
from fastapi import HTTPException, Request
@ -291,8 +292,8 @@ def _does_endpoint_match(endpoint_path: str, request_path: str) -> bool:
def check_vector_store_permission(
index_name: str,
permission: str,
key_metadata: dict[str, Any] | None,
team_metadata: dict[str, Any] | None,
key_metadata: Mapping[str, object] | None,
team_metadata: Mapping[str, object] | None,
) -> bool:
"""
Check if a specific permission is allowed for a given vector store index.

10
litellm/rag/ingestion/bedrock_ingestion.py
View file

@ -14,7 +14,7 @@ from __future__ import annotations
import asyncio
import json
import uuid
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Final
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.aws_partition import get_aws_arn_prefix
@ -26,12 +26,12 @@ if TYPE_CHECKING:
from litellm.types.rag import RAGIngestOptions
def _get_str_or_none(value: Any) -> str | None:
def _get_str_or_none(value: object) -> str | None:
"""Cast config value to Optional[str]."""
return str(value) if value is not None else None
def _get_int(value: Any, default: int) -> int:
def _get_int(value: str | float | None, default: int) -> int:
"""Cast config value to int with default."""
if value is None:
return default
@ -122,7 +122,7 @@ class BedrockRAGIngestion(BaseRAGIngestion, BaseAWSLLM):
self._config_initialized = False
# Track resources we create (for cleanup if needed)
self._created_resources: dict[str, Any] = {}
self._created_resources: dict[str, object] = {}
async def _ensure_config_initialized(self):
"""Lazily initialize KB config - either detect from existing or create new."""
@ -233,7 +233,7 @@ class BedrockRAGIngestion(BaseRAGIngestion, BaseAWSLLM):
verbose_logger.debug("Creating S3 bucket: %s", bucket_name)
create_params: Final[dict[str, Any]] = {"Bucket": bucket_name}
create_params: Final[dict[str, object]] = {"Bucket": bucket_name}
if self.aws_region_name != "us-east-1":
create_params["CreateBucketConfiguration"] = {"LocationConstraint": self.aws_region_name}

2
litellm/repositories/table_repositories.py
View file

@ -21,7 +21,7 @@ class PrismaTableRepository(Generic[RowT_co]):
table_name: str
def __init__(self, prisma_client: Any):
def __init__(self, prisma_client: object):
self._prisma_client = prisma_client
@property

2
litellm/router_strategy/lowest_latency.py
View file

@ -14,7 +14,7 @@ from litellm.types.utils import LiteLLMPydanticObjectBase
if TYPE_CHECKING:
from opentelemetry.trace import Span as _Span
Span = _Span | Any
Span = _Span
else:
Span = Any

21
litellm/router_utils/pre_call_checks/encrypted_content_affinity_check.py
View file

@ -37,7 +37,7 @@ Safe to enable globally:
"""
import time
from typing import TYPE_CHECKING, Any, Final, Optional, cast
from typing import TYPE_CHECKING, Final, Optional, Protocol, cast
import httpx
@ -51,11 +51,20 @@ from litellm.integrations.custom_logger import CustomLogger, Span
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.router_utils.cooldown_cache import CooldownCacheValue
from litellm.types.llms.openai import AllMessageValues
from litellm.types.router import Deployment
if TYPE_CHECKING:
from litellm.router import Router
class _SupportsActiveCooldowns(Protocol):
"""Cooldown-cache handle: this check only reads back the currently active cooldowns."""
async def async_get_active_cooldowns(
self, model_ids: list[str], parent_otel_span: Span | None
) -> list[tuple[str, CooldownCacheValue]]: ...
class EncryptedContentAffinityCheck(CustomLogger):
"""
Routes follow-up Responses API requests to the deployment that produced
@ -99,7 +108,7 @@ class EncryptedContentAffinityCheck(CustomLogger):
)
@staticmethod
def _extract_model_id_from_input(request_input: Any) -> str | None:
def _extract_model_id_from_input(request_input: object) -> str | None:
"""
Scan ``input`` items for litellm-encoded encrypted-content markers and
return the ``model_id`` embedded in the first one found.
@ -151,7 +160,7 @@ class EncryptedContentAffinityCheck(CustomLogger):
@staticmethod
def _encryption_boundary_key(
litellm_params: Any,
litellm_params: object,
) -> tuple | None:
"""
``(api_base, api_key)`` pair identifying an Azure resource. Two
@ -179,7 +188,7 @@ class EncryptedContentAffinityCheck(CustomLogger):
self,
healthy_deployments: list[dict],
model_id: str,
) -> tuple[list[dict], Any]:
) -> tuple[list[dict], Deployment | None]:
"""
Deployments in ``healthy_deployments`` sharing the originating
deployment's ``(api_base, api_key)``, alongside the originating
@ -289,7 +298,7 @@ class EncryptedContentAffinityCheck(CustomLogger):
self,
model: str,
model_id: str,
originating: Any,
originating: Deployment | None,
parent_otel_span: Span | None,
) -> Exception:
# Public error messages intentionally omit the originating ``model_id`` so
@ -347,7 +356,7 @@ class EncryptedContentAffinityCheck(CustomLogger):
) -> CooldownCacheValue | None:
if self.router is None:
return None
cooldown_cache: Final = getattr(self.router, "cooldown_cache", None)
cooldown_cache: Final[_SupportsActiveCooldowns | None] = getattr(self.router, "cooldown_cache", None)
if cooldown_cache is None:
return None
try:

4
litellm/router_utils/prompt_caching_cache.py
View file

@ -18,7 +18,7 @@ if TYPE_CHECKING:
from litellm.router import Router
litellm_router = Router
Span = _Span | Any
Span = _Span
else:
Span = Any
litellm_router = Any
@ -34,7 +34,7 @@ class PromptCachingCache:
self.in_memory_cache = InMemoryCache()
@staticmethod
def serialize_object(obj: Any) -> Any:
def serialize_object(obj: Any) -> object:
"""Helper function to serialize Pydantic objects, dictionaries, or fallback to string."""
if hasattr(obj, "dict"):
# If the object is a Pydantic model, use its `dict()` method

4
litellm/secret_managers/custom_secret_manager_loader.py
View file

@ -38,7 +38,9 @@ def load_custom_secret_manager(config_file_path: str | None = None) -> None:
"CustomSecretManagerException - key_management_settings is required with custom_secret_manager field"
)
custom_secret_manager_path: Final = getattr(litellm._key_management_settings, "custom_secret_manager", None)
custom_secret_manager_path: Final[str | None] = getattr(
litellm._key_management_settings, "custom_secret_manager", None
)
if not custom_secret_manager_path:
raise ValueError(

42
litellm/types/containers/main.py
View file

@ -1,7 +1,9 @@
import builtins
from collections.abc import Mapping
from typing import Any, Literal
from pydantic import BaseModel
from typing_extensions import TypedDict
from typing_extensions import ReadOnly, TypedDict
class ExpiresAfter(BaseModel):
@ -23,15 +25,15 @@ class ContainerObject(BaseModel):
name: str | None = None
_hidden_params: dict[str, Any] = {}
def __contains__(self, key) -> bool:
def __contains__(self, key: str) -> bool:
# Define custom behavior for the 'in' operator
return hasattr(self, key)
def get(self, key, default=None):
def get(self, key: str, default: builtins.object = None) -> builtins.object:
# Custom .get() method to access attributes with a default value if the attribute doesn't exist
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key: str) -> builtins.object:
# Allow dictionary-style access to attributes
return getattr(self, key)
@ -50,13 +52,13 @@ class DeleteContainerResult(BaseModel):
object: Literal["container.deleted"]
deleted: bool
def __contains__(self, key) -> bool:
def __contains__(self, key: str) -> bool:
return hasattr(self, key)
def get(self, key, default=None):
def get(self, key: str, default: builtins.object = None) -> builtins.object:
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key: str) -> builtins.object:
return getattr(self, key)
def json(self, **kwargs):
@ -75,13 +77,13 @@ class ContainerListResponse(BaseModel):
last_id: str | None = None
has_more: bool
def __contains__(self, key) -> bool:
def __contains__(self, key: str) -> bool:
return hasattr(self, key)
def get(self, key, default=None):
def get(self, key: str, default: builtins.object = None) -> builtins.object:
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key: str) -> builtins.object:
return getattr(self, key)
def json(self, **kwargs):
@ -98,7 +100,7 @@ class ContainerCreateOptionalRequestParams(TypedDict, total=False):
Params here: https://platform.openai.com/docs/api-reference/containers/create
"""
expires_after: dict[str, Any] | None # ExpiresAfter object
expires_after: ReadOnly[Mapping[str, object] | None] # ExpiresAfter object
file_ids: list[str] | None
extra_headers: dict[str, str] | None
extra_body: dict[str, str] | None
@ -140,13 +142,13 @@ class ContainerFileObject(BaseModel):
source: str
_hidden_params: dict[str, Any] = {}
def __contains__(self, key) -> bool:
def __contains__(self, key: str) -> bool:
return hasattr(self, key)
def get(self, key, default=None):
def get(self, key: str, default: builtins.object = None) -> builtins.object:
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key: str) -> builtins.object:
return getattr(self, key)
def json(self, **kwargs):
@ -165,13 +167,13 @@ class ContainerFileListResponse(BaseModel):
last_id: str | None = None
has_more: bool
def __contains__(self, key) -> bool:
def __contains__(self, key: str) -> bool:
return hasattr(self, key)
def get(self, key, default=None):
def get(self, key: str, default: builtins.object = None) -> builtins.object:
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key: str) -> builtins.object:
return getattr(self, key)
def json(self, **kwargs):
@ -189,13 +191,13 @@ class DeleteContainerFileResponse(BaseModel):
object: Literal["container.file.deleted", "container_file.deleted"]
deleted: bool
def __contains__(self, key) -> bool:
def __contains__(self, key: str) -> bool:
return hasattr(self, key)
def get(self, key, default=None):
def get(self, key: str, default: builtins.object = None) -> builtins.object:
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key: str) -> builtins.object:
return getattr(self, key)
def json(self, **kwargs):

28
litellm/types/llms/oci.py
View file

@ -1,7 +1,7 @@
from __future__ import annotations
from enum import Enum
from typing import Any, Literal
from typing import Literal
from pydantic import BaseModel, SerializeAsAny
@ -105,9 +105,9 @@ class OCIChatRequestPayload(BaseModel):
# Honoured by GPT-5 family, Gemini 2.5, Grok reasoning variants,
# Cohere Command-A-Reasoning. Ignored by non-reasoning models.
reasoningEffort: str | None = None
responseFormat: dict[str, Any] | None = None
toolChoice: str | dict[str, Any] | None = None
logitBias: dict[str, Any] | None = None
responseFormat: dict[str, object] | None = None
toolChoice: str | dict[str, object] | None = None
logitBias: dict[str, object] | None = None
logProbs: int | None = None
@ -163,7 +163,7 @@ class OCIResponseChoice(BaseModel):
# reasoning phase without producing any visible content.
message: OCIMessage | None = None
finishReason: str | None = None
logprobs: dict[str, Any] | None = None
logprobs: dict[str, object] | None = None
class OCIChatResponse(BaseModel):
@ -275,7 +275,7 @@ class CohereToolCall(BaseModel):
"""Tool call made by Cohere model."""
name: str
parameters: dict[str, Any]
parameters: dict[str, object]
class CohereToolResult(BaseModel):
@ -286,7 +286,7 @@ class CohereToolResult(BaseModel):
"""
call: CohereToolCall
outputs: list[dict[str, Any]]
outputs: list[dict[str, object]]
class CohereChatRequest(BaseModel):
@ -318,12 +318,12 @@ class CohereChatRequest(BaseModel):
# OCI Cohere responseFormat is {"type": "TEXT" | "JSON_OBJECT", "schema"?: ...};
# there is no JSON_SCHEMA type. The shape is built in
# OCIChatConfig._normalize_response_format.
responseFormat: dict[str, Any] | None = None
responseFormat: dict[str, object] | None = None
preambleOverride: str | None = None
documents: list[dict[str, Any]] | None = None
documents: list[dict[str, object]] | None = None
searchQueriesOnly: bool | None = None
searchEntryPoint: str | None = None
grounding: dict[str, Any] | None = None
grounding: dict[str, object] | None = None
isEcho: bool | None = None
isSearchQueriesOnly: bool | None = None
isRawPrompting: bool | None = None
@ -333,7 +333,7 @@ class CohereChatRequest(BaseModel):
citationQuality: str | None = None
maxInputTokens: int | None = None
isStream: bool | None = None
streamOptions: dict[str, Any] | None = None
streamOptions: dict[str, object] | None = None
class CohereUsage(BaseModel):
@ -342,8 +342,8 @@ class CohereUsage(BaseModel):
promptTokens: int
completionTokens: int
totalTokens: int
promptTokensDetails: dict[str, Any] | None = None
completionTokensDetails: dict[str, Any] | None = None
promptTokensDetails: dict[str, object] | None = None
completionTokensDetails: dict[str, object] | None = None
class CohereCitation(BaseModel):
@ -378,7 +378,7 @@ class CohereChatResponse(BaseModel):
# Optional fields
chatHistory: list[CohereMessage] | None = None
citations: list[CohereCitation] | None = None
documents: list[dict[str, Any]] | None = None
documents: list[dict[str, object]] | None = None
errorMessage: str | None = None
isSearchRequired: bool | None = None
prompt: str | None = None

33
litellm/types/llms/openai_evals.py
View file

@ -2,7 +2,8 @@
Type definitions for OpenAI Evals API
"""
from typing import Any, Literal
import builtins
from typing import Literal
from pydantic import BaseModel
from typing_extensions import Required, TypedDict
@ -15,7 +16,7 @@ class DataSourceConfigCustom(TypedDict, total=False):
type: Required[Literal["custom"]]
"""Data source type - custom"""
item_schema: Required[dict[str, Any]]
item_schema: Required[dict[str, object]]
"""JSON schema describing the structure of each row in the dataset"""
include_sample_schema: bool | None
@ -28,7 +29,7 @@ class DataSourceConfigLogs(TypedDict, total=False):
type: Required[Literal["logs"]]
"""Data source type - logs"""
metadata: dict[str, Any] | None
metadata: dict[str, object] | None
"""Optional metadata for filtering logs"""
@ -38,7 +39,7 @@ class DataSourceConfigStoredCompletions(TypedDict, total=False):
type: Required[Literal["stored_completions"]]
"""Data source type - stored_completions (deprecated)"""
metadata: dict[str, Any] | None
metadata: dict[str, object] | None
"""Optional metadata for filtering stored completions"""
@ -93,7 +94,7 @@ class CreateEvalRequest(TypedDict, total=False):
testing_criteria: Required[list[GraderConfig]]
"""List of graders for all eval runs"""
metadata: dict[str, Any] | None
metadata: dict[str, object] | None
"""Set of 16 key-value pairs that can be attached to an object (max 64 char keys, 512 char values)"""
@ -103,7 +104,7 @@ class UpdateEvalRequest(TypedDict, total=False):
name: str | None
"""Updated name"""
metadata: dict[str, Any] | None
metadata: dict[str, object] | None
"""Updated metadata"""
@ -145,13 +146,13 @@ class Eval(BaseModel):
name: str | None = None
"""The name of the evaluation"""
data_source_config: dict[str, Any]
data_source_config: dict[str, builtins.object]
"""Configuration for the data source"""
testing_criteria: list[dict[str, Any]]
testing_criteria: list[dict[str, builtins.object]]
"""List of graders for the evaluation"""
metadata: dict[str, Any] | None = None
metadata: dict[str, builtins.object] | None = None
"""Additional metadata"""
@ -227,7 +228,7 @@ class DataSourceInlineConfig(TypedDict, total=False):
type: Required[Literal["inline"]]
"""Data source type - inline"""
samples: Required[list[dict[str, Any]]]
samples: Required[list[dict[str, object]]]
"""List of inline samples to use for the run"""
@ -259,13 +260,13 @@ class CompletionConfig(TypedDict, total=False):
class CreateRunRequest(TypedDict, total=False):
"""Request parameters for creating a run"""
data_source: Required[dict[str, Any]]
data_source: Required[dict[str, object]]
"""Data source configuration for the run (can be jsonl, completions, or responses type)"""
name: str | None
"""Optional name for the run"""
metadata: dict[str, Any] | None
metadata: dict[str, object] | None
"""Optional metadata for the run"""
@ -330,7 +331,7 @@ class Run(BaseModel):
status: Literal["queued", "running", "completed", "failed", "cancelled"]
"""Current status of the run"""
data_source: dict[str, Any]
data_source: dict[str, builtins.object]
"""Data source configuration used for the run"""
eval_id: str
@ -348,7 +349,7 @@ class Run(BaseModel):
model: str | None = None
"""Model used for the run, if any"""
per_model_usage: Any | None = None
per_model_usage: builtins.object | None = None
"""Model usage details per model, if available"""
per_testing_criteria_results: list[PerTestingCriteriaResult] | None = None
@ -363,10 +364,10 @@ class Run(BaseModel):
shared_with_openai: bool | None = None
"""Whether run is shared with OpenAI"""
metadata: dict[str, Any] | None = None
metadata: dict[str, builtins.object] | None = None
"""Additional metadata"""
error: dict[str, Any] | None = None
error: dict[str, builtins.object] | None = None
"""Error details if the run failed"""

12
litellm/types/proxy/management_endpoints/scim_v2.py
View file

@ -36,7 +36,7 @@ class SCIMResource(BaseModel):
schemas: list[str]
id: str | None = None
externalId: str | None = None
meta: dict[str, Any] | None = None
meta: dict[str, object] | None = None
class SCIMUserName(BaseModel):
@ -119,7 +119,7 @@ class SCIMUser(SCIMResource):
)
@model_serializer(mode="wrap")
def _omit_absent_optional_blocks(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]:
def _omit_absent_optional_blocks(self, handler: SerializerFunctionWrapHandler) -> dict[str, object]:
dumped: Final = handler(self)
if self.enterprise_user is None:
dumped.pop(SCIM_ENTERPRISE_USER_SCHEMA, None)
@ -169,7 +169,7 @@ class SCIMListResponse(BaseModel):
class SCIMPatchOperation(BaseModel):
op: str
path: str | None = None
value: Any | None = None
value: object | None = None
@field_validator("op", mode="before")
@classmethod
@ -203,7 +203,7 @@ class SCIMServiceProviderConfig(BaseModel):
changePassword: SCIMFeature = SCIMFeature(supported=False)
sort: SCIMFeature = SCIMFeature(supported=False)
etag: SCIMFeature = SCIMFeature(supported=False)
authenticationSchemes: list[dict[str, Any]] | None = None
authenticationSchemes: list[dict[str, object]] | None = None
meta: dict[str, Any] | None = None
@ -231,7 +231,7 @@ class SCIMResourceType(BaseModel):
schema_: str # "schema" is a reserved name in Pydantic context
schemaExtensions: list[SCIMSchemaExtension] | None = None
meta: dict[str, Any] | None = None
meta: dict[str, object] | None = None
def model_dump(self, **kwargs):
d: Final = super().model_dump(**kwargs)
@ -266,4 +266,4 @@ class SCIMSchema(BaseModel):
name: str
description: str | None = None
attributes: list[SCIMSchemaAttribute] = []
meta: dict[str, Any] | None = None
meta: dict[str, object] | None = None

19
litellm/types/videos/main.py
View file

@ -1,3 +1,4 @@
import builtins
from typing import Any, Literal
from openai.types.audio.transcription_create_params import FileTypes
@ -14,14 +15,14 @@ class VideoObject(BaseModel):
created_at: int | None = None
completed_at: int | None = None
expires_at: int | None = None
error: dict[str, Any] | None = None
error: dict[str, builtins.object] | None = None
progress: int | None = None
remixed_from_video_id: str | None = None
seconds: str | None = None
size: str | None = None
model: str | None = None
usage: dict[str, Any] | None = None
_hidden_params: dict[str, Any] = {}
_hidden_params: dict[str, builtins.object] = {}
def __contains__(self, key) -> bool:
# Define custom behavior for the 'in' operator
@ -31,7 +32,7 @@ class VideoObject(BaseModel):
# Custom .get() method to access attributes with a default value if the attribute doesn't exist
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key) -> builtins.object:
# Allow dictionary-style access to attributes
return getattr(self, key)
@ -47,7 +48,7 @@ class VideoResponse(BaseModel):
"""Response object for video generation requests."""
data: list[VideoObject]
hidden_params: dict[str, Any] = {}
hidden_params: dict[str, object] = {}
def __contains__(self, key) -> bool:
return hasattr(self, key)
@ -55,7 +56,7 @@ class VideoResponse(BaseModel):
def get(self, key, default=None):
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key) -> object:
return getattr(self, key)
def json(self, **kwargs):
@ -73,8 +74,8 @@ class VideoCreateOptionalRequestParams(TypedDict, total=False):
"""
input_reference: FileTypes | None # File reference for input image
image: Any | None # Image for image-to-video; dict with gcsUri/bytesBase64Encoded, or file-like object
parameters: dict[str, Any] | None # Provider-specific parameters block passed directly to the API
image: object | None # Image for image-to-video; dict with gcsUri/bytesBase64Encoded, or file-like object
parameters: dict[str, object] | None # Provider-specific parameters block passed directly to the API
model: str | None
resolution: ReadOnly[str | None]
seconds: str | None
@ -110,7 +111,7 @@ class CharacterObject(BaseModel):
object: Literal["character"] = "character"
created_at: int
name: str
_hidden_params: dict[str, Any] = {}
_hidden_params: dict[str, builtins.object] = {}
def __contains__(self, key) -> bool:
return hasattr(self, key)
@ -118,7 +119,7 @@ class CharacterObject(BaseModel):
def get(self, key, default=None):
return getattr(self, key, default)
def __getitem__(self, key):
def __getitem__(self, key) -> builtins.object:
return getattr(self, key)
def json(self, **kwargs):