Merge pull request #39461 from BerriAI/litellm_decrease_anys_opus5_r4

refactor(typing): cut 1,397 Any errors across 183 backend files
This commit is contained in:
Mateo Wang 2026-09-04 21:44:04 -07:00 committed by GitHub
commit 75736323e6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
186 changed files with 1349 additions and 985 deletions

View file

@ -1,9 +1,9 @@
{
"reportAny": {
"limit": 14072
"limit": 13429
},
"reportArgumentType": {
"limit": 2206
"limit": 2198
},
"reportAssignmentType": {
"limit": 319
@ -24,7 +24,7 @@
"limit": 19
},
"reportExplicitAny": {
"limit": 4121
"limit": 3369
},
"reportFunctionMemberAccess": {
"limit": 7
@ -48,16 +48,16 @@
"limit": 30
},
"reportInvalidTypeVarUse": {
"limit": 2
"limit": 1
},
"reportMatchNotExhaustive": {
"limit": 0
},
"reportMissingParameterType": {
"limit": 5601
"limit": 5570
},
"reportMissingTypeArgument": {
"limit": 15284
"limit": 15281
},
"reportMissingTypeStubs": {
"limit": 40
@ -90,7 +90,7 @@
"limit": 8
},
"reportReturnType": {
"limit": 181
"limit": 180
},
"reportTypedDictNotRequiredAccess": {
"limit": 22
@ -105,16 +105,16 @@
"limit": 109
},
"reportUnknownMemberType": {
"limit": 38309
"limit": 38283
},
"reportUnknownParameterType": {
"limit": 19620
"limit": 19584
},
"reportUnknownVariableType": {
"limit": 29844
"limit": 29829
},
"reportUnnecessaryCast": {
"limit": 111
"limit": 110
},
"reportUnnecessaryComparison": {
"limit": 687
@ -123,7 +123,7 @@
"limit": 4
},
"reportUnnecessaryIsInstance": {
"limit": 819
"limit": 816
},
"reportUntypedBaseClass": {
"limit": 0

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.

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

View file

@ -1,7 +1,7 @@
import asyncio
from collections.abc import Callable, Coroutine
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
@ -25,7 +25,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
@ -55,7 +78,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
@ -70,18 +93,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
@staticmethod

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.

View file

@ -6,8 +6,8 @@ completion bridge that would otherwise strip the envelope.
"""
import json
from collections.abc import AsyncIterator
from typing import Any, Final, cast
from collections.abc import AsyncIterator, Mapping
from typing import Any, Final
from litellm._logging import verbose_logger
from litellm.a2a_protocol.providers.bedrock_agentcore.transformation import (
@ -28,7 +28,7 @@ class BedrockAgentCoreA2AHandler:
@staticmethod
async def handle_non_streaming(
request_id: str,
params: dict[str, Any],
params: Mapping[str, object],
litellm_params: dict[str, Any],
agent_extra_headers: dict[str, str] | None = None,
) -> dict[str, Any]:
@ -56,7 +56,7 @@ class BedrockAgentCoreA2AHandler:
verbose_logger.info("BedrockAgentCore A2A: Sending non-streaming request to %s", url)
client: Final = get_async_httpx_client(
llm_provider=cast(Any, httpxSpecialProvider.A2AProvider),
llm_provider=httpxSpecialProvider.A2AProvider,
)
response: Final = await client.post(
url,
@ -74,7 +74,7 @@ class BedrockAgentCoreA2AHandler:
@staticmethod
async def handle_streaming(
request_id: str,
params: dict[str, Any],
params: Mapping[str, object],
litellm_params: dict[str, Any],
agent_extra_headers: dict[str, str] | None = None,
) -> AsyncIterator[dict[str, Any]]:
@ -103,7 +103,7 @@ class BedrockAgentCoreA2AHandler:
verbose_logger.info("BedrockAgentCore A2A: Sending streaming request to %s", url)
client: Final = get_async_httpx_client(
llm_provider=cast(Any, httpxSpecialProvider.A2AProvider),
llm_provider=httpxSpecialProvider.A2AProvider,
)
response: Final = await client.post(
url,

View file

@ -7,7 +7,7 @@ and signs requests via AmazonAgentCoreConfig (SigV4 or JWT).
import json
from collections.abc import AsyncIterator, Mapping
from typing import Any, Final
from typing import Any, Final, Protocol
from litellm._logging import verbose_logger
from litellm.a2a_protocol.litellm_completion_bridge.handler import (
@ -47,6 +47,12 @@ _RESERVED_PREFIX_HEADERS: Final[tuple[str, ...]] = (
)
class _SSELineSource(Protocol):
"""Minimal streaming-response surface used to read SSE lines."""
def aiter_lines(self) -> AsyncIterator[str]: ...
def _filter_reserved_headers(
agent_extra_headers: Mapping[str, str] | None,
) -> dict[str, str] | None:
@ -114,7 +120,7 @@ class BedrockAgentCoreA2ATransformation:
@staticmethod
def get_url_and_signed_request(
request_id: str,
params: dict[str, Any],
params: Mapping[str, object],
litellm_params: dict[str, Any],
method: str = "message/send",
stream: bool = False,
@ -213,7 +219,7 @@ class BedrockAgentCoreA2ATransformation:
return url, signed_headers, signed_body
@staticmethod
async def parse_sse_events(response: Any) -> AsyncIterator[dict[str, Any]]:
async def parse_sse_events(response: _SSELineSource) -> AsyncIterator[dict[str, Any]]:
"""
Parse SSE events from an httpx streaming response.

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.

View file

@ -148,9 +148,9 @@ class A2AStreamingIterator:
except Exception as e:
verbose_logger.debug("Error in A2A streaming completion handler: %s", e)
def _build_logging_result(self, usage: litellm.Usage) -> dict[str, Any]:
def _build_logging_result(self, usage: litellm.Usage) -> dict[str, object]:
"""Build a result dict for logging."""
result: Final[dict[str, Any]] = {
result: Final[dict[str, object]] = {
"id": getattr(self.request, "id", "unknown"),
"jsonrpc": "2.0",
"usage": (usage.model_dump() if hasattr(usage, "model_dump") else dict(usage)),

View file

@ -48,7 +48,7 @@ class A2ARequestUtils:
return " ".join(text_parts)
@staticmethod
def extract_text_from_response(response_dict: dict[str, Any]) -> str:
def extract_text_from_response(response_dict: Mapping[str, object]) -> str:
"""
Extract text content from A2A response result.
@ -111,7 +111,7 @@ class A2ARequestUtils:
@staticmethod
def calculate_usage_from_request_response(
request: "SendMessageRequest | SendStreamingMessageRequest",
response_dict: dict[str, Any],
response_dict: Mapping[str, object],
) -> tuple[int, int, int]:
"""
Calculate token usage from A2A request and response.
@ -170,5 +170,5 @@ def extract_text_from_a2a_message(message: Any) -> str:
return A2ARequestUtils.extract_text_from_message(message)
def extract_text_from_a2a_response(response_dict: dict[str, Any]) -> str:
def extract_text_from_a2a_response(response_dict: Mapping[str, object]) -> str:
return A2ARequestUtils.extract_text_from_response(response_dict)

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:

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
"""

View file

@ -672,7 +672,7 @@ class LLMCachingHandler:
def _async_log_cache_hit_on_callbacks(
self,
logging_obj: LiteLLMLoggingObj,
cached_result: Any,
cached_result: object,
start_time: datetime.datetime,
end_time: datetime.datetime,
cache_hit: bool,
@ -1184,7 +1184,7 @@ class LLMCachingHandler:
logging_obj: LiteLLMLoggingObj,
model: str,
kwargs: dict[str, Any],
cached_result: Any,
cached_result: object,
is_async: bool,
is_embedding: bool = False,
custom_llm_provider: str | None = None,

View file

@ -116,7 +116,7 @@ class RedisSemanticCache(BaseCache):
password = password or os.environ["REDIS_PASSWORD"]
except KeyError as e:
# Raise a more informative exception if any of the required keys are missing
missing_var: Final = e.args[0]
missing_var: Final[object] = e.args[0]
raise ValueError(
f"Missing required Redis configuration: {missing_var}. Provide {missing_var} or redis_url."
) from e
@ -273,7 +273,7 @@ class RedisSemanticCache(BaseCache):
return prompt or None
@classmethod
def _collect_responses_input_text(cls, value: Any, prompt_parts: list[str]) -> None:
def _collect_responses_input_text(cls, value: object, prompt_parts: list[str]) -> None:
value = cls._coerce_response_input_value(value)
if value is None:
return
@ -334,7 +334,7 @@ class RedisSemanticCache(BaseCache):
resolve_embedding_max_input_tokens(self.embedding_max_input_tokens, self.embedding_model, router),
)
def _get_embedding(self, prompt: str, metadata: dict[str, Any] | None = None) -> list[float]:
def _get_embedding(self, prompt: str, metadata: dict[str, object] | None = None) -> list[float]:
"""
Routes through the proxy Router when the embedding model is a Router
deployment so per-deployment auth (e.g. Bedrock aws_role_name) applies,
@ -425,7 +425,7 @@ class RedisSemanticCache(BaseCache):
prompt_embedding: Final = self._get_embedding(prompt, metadata=kwargs.get("metadata"))
store_kwargs: Final[dict[str, Any]] = {
store_kwargs: Final[dict[str, object]] = {
"vector": prompt_embedding,
"filters": self._get_cache_filters(key),
}
@ -504,7 +504,7 @@ class RedisSemanticCache(BaseCache):
print_verbose(f"Error retrieving from Redis semantic cache: {e}")
kwargs.setdefault("metadata", {})["semantic-similarity"] = 0.0
async def _get_async_embedding(self, prompt: str, metadata: dict[str, Any] | None = None) -> list[float]:
async def _get_async_embedding(self, prompt: str, metadata: dict[str, object] | None = None) -> list[float]:
"""
Asynchronously generate an embedding for the given prompt.
@ -571,7 +571,7 @@ class RedisSemanticCache(BaseCache):
# Generate embedding for the value (response) to cache
prompt_embedding: Final = await self._get_async_embedding(prompt, metadata=kwargs.get("metadata"))
store_kwargs: Final[dict[str, Any]] = {
store_kwargs: Final[dict[str, object]] = {
"vector": prompt_embedding,
"filters": self._get_cache_filters(key),
}
@ -665,7 +665,7 @@ class RedisSemanticCache(BaseCache):
aindex: Final = await self.llmcache._get_async_index()
return await aindex.info()
async def async_set_cache_pipeline(self, cache_list: list[tuple[str, Any]], **kwargs: object) -> None:
async def async_set_cache_pipeline(self, cache_list: list[tuple[str, object]], **kwargs: object) -> None:
"""
Asynchronously store multiple values in the semantic cache.

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:
"""

View file

@ -5,6 +5,7 @@ Computes cosine similarity between the query embedding and each message embeddin
"""
import math
from collections.abc import Mapping
from typing import Any, Final
from litellm.caching.dual_cache import DualCache
@ -49,7 +50,7 @@ def embedding_score_messages(
messages: list[dict],
model: str,
cache: DualCache | None = None,
embedding_model_params: dict[str, Any] | None = None,
embedding_model_params: Mapping[str, object] | None = None,
) -> list[float]:
"""
Score each message's semantic similarity to the query using embeddings.

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()

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.

View file

@ -6,17 +6,35 @@ import asyncio
import base64
import os
from collections.abc import Awaitable, Callable, Generator
from contextlib import AbstractAsyncContextManager
from datetime import timedelta
from functools import partial
from importlib import metadata
from typing import Any, Final, TypeVar
from typing import Any, Final, Protocol, TypeAlias, TypeVar
import httpx
from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
from mcp import ClientSession, McpError, ReadResourceResult, Resource, StdioServerParameters
from mcp.client.sse import sse_client
from mcp.client.stdio import stdio_client
from mcp.shared.message import SessionMessage
from typing_extensions import Unpack
streamable_http_client: Any | None = None
_TransportStreams: TypeAlias = tuple[
MemoryObjectReceiveStream[SessionMessage | Exception],
MemoryObjectSendStream[SessionMessage],
Unpack[tuple[object, ...]],
]
_TransportContext: TypeAlias = AbstractAsyncContextManager[_TransportStreams]
class _StreamableHttpClientFactory(Protocol):
"""The ``streamable_http_client`` entry point this module calls on the installed MCP SDK."""
def __call__(self, *, url: str, http_client: httpx.AsyncClient | None) -> _TransportContext: ...
streamable_http_client: _StreamableHttpClientFactory | None = None
try:
import mcp.client.streamable_http as streamable_http_module
@ -216,10 +234,12 @@ class MCPSigV4Auth(httpx.Auth):
aws_region_name: str,
):
"""Call STS AssumeRole and return temporary credentials."""
import time
import boto3
from botocore.credentials import Credentials
session_name: Final = aws_session_name or f"litellm-mcp-{int(__import__('time').time())}"
session_name: Final = aws_session_name or f"litellm-mcp-{int(time.time())}"
sts_kwargs: Final[dict] = {"region_name": aws_region_name}
if aws_access_key_id and aws_secret_access_key:
sts_kwargs["aws_access_key_id"] = aws_access_key_id
@ -315,7 +335,7 @@ class MCPClient:
def _create_transport_context(
self,
) -> tuple[Any, httpx.AsyncClient | None]:
) -> tuple[_TransportContext, httpx.AsyncClient | None]:
"""
Create the appropriate transport context based on transport type.
Returns:
@ -408,7 +428,7 @@ class MCPClient:
async def _execute_session_operation(
self,
transport_ctx: Any,
transport_ctx: _TransportContext,
operation: Callable[[ClientSession], Awaitable[TSessionResult]],
) -> TSessionResult:
"""

View file

@ -14,6 +14,7 @@ from functools import partial
from typing import Any, Final, Literal, cast
import httpx
from openai import AsyncOpenAI, OpenAI
# Type aliases for provider parameters
FileCreateProvider = Literal[
@ -431,7 +432,7 @@ async def afile_delete(
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, str] | None = None,
**kwargs,
) -> Coroutine[Any, Any, FileObject]:
) -> Coroutine[object, object, FileObject]:
"""
Async: Delete file
@ -1002,8 +1003,8 @@ def file_content_streaming(
timeout: float | httpx.Timeout,
logging_obj: LiteLLMLoggingObj | None,
_is_async: bool,
client: Any | None,
) -> FileContentStreamingResult | Coroutine[Any, Any, FileContentStreamingResult]:
client: OpenAI | AsyncOpenAI | None,
) -> FileContentStreamingResult | Coroutine[object, object, FileContentStreamingResult]:
if logging_obj is not None:
logging_obj.model = model or ""
logging_obj.model_call_details["model"] = model or ""
@ -1028,8 +1029,8 @@ def file_content_streaming(
headers=response.headers,
)
response: FileContentStreamingResult | Coroutine[Any, Any, FileContentStreamingResult] = FileContentStreamingResult(
stream_iterator=iter(()), headers={}
response: FileContentStreamingResult | Coroutine[object, object, FileContentStreamingResult] = (
FileContentStreamingResult(stream_iterator=iter(()), headers={})
)
if custom_llm_provider in OPENAI_COMPATIBLE_BATCH_AND_FILES_PROVIDERS:
openai_creds: Final = get_openai_credentials(

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.
"""

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.

View file

@ -2,7 +2,7 @@ import json
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from typing_extensions import override
from typing_extensions import ReadOnly, TypedDict, override
from litellm._logging import verbose_logger
from litellm.integrations.opentelemetry_utils.base_otel_llm_obs_attributes import (
@ -492,12 +492,12 @@ def _sanitize_optional_params(optional_params: dict | None) -> dict:
return optional_params
def _set_metadata_attributes(span: "Span", metadata: Any | None, span_attrs) -> None:
def _set_metadata_attributes(span: "Span", metadata: object | None, span_attrs) -> None:
if metadata is not None:
safe_set_attribute(span, span_attrs.METADATA, safe_dumps(metadata))
def _extract_metadata_tools(metadata: Any | None) -> list | None:
def _extract_metadata_tools(metadata: object | None) -> list | None:
if not isinstance(metadata, dict):
return None
llm_obj: Final = metadata.get("llm")
@ -670,7 +670,22 @@ def _get_tool_calls(message) -> list | None:
return tool_calls if isinstance(tool_calls, list) and tool_calls else None
def _normalize_tool_call(raw_tc) -> dict[str, Any] | None:
class _NormalizedToolCallFunction(TypedDict):
"""The ``function`` sub-object of a normalized tool call."""
name: ReadOnly[object]
arguments: ReadOnly[object]
class _NormalizedToolCall(TypedDict):
"""A tool call reduced to the stable shape the OpenInference emitters read."""
id: ReadOnly[object]
type: ReadOnly[object]
function: ReadOnly[_NormalizedToolCallFunction]
def _normalize_tool_call(raw_tc) -> _NormalizedToolCall | None:
"""Normalize a single tool_call (dict or Pydantic) into a stable shape:
{"id": str|None, "type": str, "function": {"name": str|None, "arguments": str|None}}

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)

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)

View file

@ -3,14 +3,26 @@
from __future__ import annotations
import asyncio
from collections.abc import Mapping
from datetime import timezone
from typing import Any, Final
from typing import Final, TypedDict
import boto3
from typing_extensions import ReadOnly
from .base import FocusDestination, FocusTimeWindow
class _S3ClientKwargs(TypedDict, total=False):
"""Optional boto3 client arguments the destination config may supply."""
region_name: ReadOnly[str]
endpoint_url: ReadOnly[str]
aws_access_key_id: ReadOnly[str]
aws_secret_access_key: ReadOnly[str]
aws_session_token: ReadOnly[str]
class FocusS3Destination(FocusDestination):
"""Handles uploading serialized exports to S3 buckets."""
@ -18,7 +30,7 @@ class FocusS3Destination(FocusDestination):
self,
*,
prefix: str,
config: dict[str, Any] | None = None,
config: Mapping[str, str] | None = None,
) -> None:
config = config or {}
bucket_name: Final = config.get("bucket_name")
@ -47,25 +59,23 @@ class FocusS3Destination(FocusDestination):
key_prefix: Final = "/".join(filter(None, parts))
return f"{key_prefix}/{filename}" if key_prefix else filename
def _client_kwargs(self) -> _S3ClientKwargs:
"""Collect the boto3 client arguments the destination config provides."""
region: Final = self.config.get("region_name")
endpoint: Final = self.config.get("endpoint_url")
key_id: Final = self.config.get("aws_access_key_id")
secret: Final = self.config.get("aws_secret_access_key")
token: Final = self.config.get("aws_session_token")
return {
**(_S3ClientKwargs(region_name=region) if region else _S3ClientKwargs()),
**(_S3ClientKwargs(endpoint_url=endpoint) if endpoint else _S3ClientKwargs()),
**(_S3ClientKwargs(aws_access_key_id=key_id) if key_id else _S3ClientKwargs()),
**(_S3ClientKwargs(aws_secret_access_key=secret) if secret else _S3ClientKwargs()),
**(_S3ClientKwargs(aws_session_token=token) if token else _S3ClientKwargs()),
}
def _upload(self, content: bytes, object_key: str) -> None:
client_kwargs: Final[dict[str, Any]] = {}
region_name: Final = self.config.get("region_name")
if region_name:
client_kwargs["region_name"] = region_name
endpoint_url: Final = self.config.get("endpoint_url")
if endpoint_url:
client_kwargs["endpoint_url"] = endpoint_url
session_kwargs: Final[dict[str, Any]] = {}
for key in (
"aws_access_key_id",
"aws_secret_access_key",
"aws_session_token",
):
if self.config.get(key):
session_kwargs[key] = self.config[key]
s3_client: Final = boto3.client("s3", **client_kwargs, **session_kwargs)
s3_client: Final = boto3.client("s3", **self._client_kwargs())
s3_client.put_object(
Bucket=self.bucket_name,
Key=object_key,

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

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.

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.

View file

@ -47,6 +47,8 @@ import os
import threading
import time
import uuid
from collections.abc import Mapping, Sequence
from datetime import datetime
from typing import Any, Final
import litellm
@ -408,8 +410,8 @@ class NewRelicLogger(CustomLogger):
def _get_duration(
self,
kwargs: dict,
start_time: Any,
end_time: Any,
start_time: datetime | float | None,
end_time: datetime | float | None,
standard_logging_object: StandardLoggingPayload | None = None,
) -> float | None:
"""
@ -438,7 +440,7 @@ class NewRelicLogger(CustomLogger):
self,
kwargs: dict,
standard_logging_object: StandardLoggingPayload | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Extract request parameters like temperature and max_tokens, preferring
StandardLoggingPayload.model_parameters.
@ -450,7 +452,7 @@ class NewRelicLogger(CustomLogger):
else:
source_params = kwargs.get("optional_params") or {}
params: Final = {}
params: Final[dict[str, object]] = {}
temperature: Final = source_params.get("temperature")
if temperature is not None:
@ -502,7 +504,7 @@ class NewRelicLogger(CustomLogger):
response_model: str,
vendor: str,
standard_logging_object: StandardLoggingPayload | None = None,
) -> list[dict[str, Any]]:
) -> Sequence[Mapping[str, object]]:
"""
Extract all messages (request + response) with sequence numbers and timestamps.
@ -512,7 +514,7 @@ class NewRelicLogger(CustomLogger):
Adds timestamps from StandardLoggingPayload (preferred) or kwargs if available
(converted to epoch milliseconds).
"""
messages: Final = []
messages: Final[list[dict[str, object]]] = []
sequence = 0
# Extract timestamps, preferring StandardLoggingPayload
@ -544,7 +546,7 @@ class NewRelicLogger(CustomLogger):
else:
request_messages = kwargs.get("messages") or []
for msg in request_messages:
message_data = {
message_data: dict[str, object] = {
"role": msg.get("role") or "user",
"sequence": sequence,
"response.model": response_model,
@ -599,11 +601,11 @@ class NewRelicLogger(CustomLogger):
num_messages: int,
usage: dict[str, int],
duration: float | None = None,
request_params: dict[str, Any] | None = None,
request_params: Mapping[str, object] | None = None,
):
"""Record LlmChatCompletionSummary event to New Relic."""
try:
event_data: Final = {
event_data: Final[dict[str, object]] = {
"id": request_id,
"request_id": request_id,
"request.model": request_model,
@ -647,7 +649,7 @@ class NewRelicLogger(CustomLogger):
request_id: str,
llm_response_id: str,
trace_id: str | None,
messages: list[dict[str, Any]],
messages: Sequence[Mapping[str, object]],
):
"""Record LlmChatCompletionMessage events to New Relic.
@ -666,7 +668,7 @@ class NewRelicLogger(CustomLogger):
for message in messages:
sequence = message["sequence"]
event_data = {
event_data: dict[str, object] = {
"id": f"{llm_response_id}-{sequence}",
"request_id": request_id,
"completion_id": request_id,

View file

@ -1,7 +1,7 @@
import os
import threading
from collections import OrderedDict
from collections.abc import Callable, Mapping
from collections.abc import Callable, Iterable, Mapping
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import datetime
@ -166,7 +166,7 @@ class OTELMetricAttributeFilter:
exclude_list: list[str] | None = None
def _build_metric_attribute_filter(value: Any) -> OTELMetricAttributeFilter:
def _build_metric_attribute_filter(value: object) -> OTELMetricAttributeFilter:
if isinstance(value, OTELMetricAttributeFilter):
return value
if not isinstance(value, dict):
@ -205,7 +205,7 @@ def _resolve_metric_attribute_filter(
)
def _normalize_team_metadata_keys(value: Any) -> list[str]:
def _normalize_team_metadata_keys(value: str | Iterable[object] | None) -> list[str]:
"""Coerce a team-metadata allowlist from a list or comma-separated string.
config.yaml passes a YAML list; an env var passes a comma-separated string.
@ -1569,7 +1569,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
self.safe_set_attribute(span=span, key=RESPONSE_SERVICE_TIER_ATTRIBUTE, value=served_tier)
@staticmethod
def _team_metadata_json(value: Any, allowed_keys: list[str]) -> str | None:
def _team_metadata_json(value: object, allowed_keys: list[str]) -> str | None:
"""JSON-serialize only the allowlisted sub-keys of a team's metadata.
Returns ``None`` when nothing is allowlisted or no allowlisted key is
@ -3524,7 +3524,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
kwargs={"standard_logging_object": {"error_information": error_information}},
)
def set_preprocessing_duration_attribute(self, span: Span | None, container: Any) -> None:
def set_preprocessing_duration_attribute(self, span: Span | None, container: object) -> None:
"""
Set ``litellm.preprocessing.duration_ms`` (proxy-receive -> first
provider handoff) on the proxy SERVER span. ``litellm_received_at``

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),

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],

View file

@ -10,7 +10,7 @@ import sys
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import replace
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, TypeVar, cast
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, TypeAlias, TypeVar, cast
from pydantic import BaseModel
@ -142,6 +142,9 @@ class _ExcludedLabelMetric:
return self._metric.labels(*kept_values) if kept_values else self._metric
_MetricLike: TypeAlias = "NoOpMetric | _ExcludedLabelMetric | MetricWrapperBase"
def _get_budget_metrics_per_request_timeout() -> float:
raw: Final = os.getenv("PROMETHEUS_BUDGET_METRICS_PER_REQUEST_TIMEOUT")
if raw is None:
@ -1652,7 +1655,7 @@ class PrometheusLogger(CustomLogger):
cache_creation_detail_tokens: Final = PrometheusLogger._resolve_cache_write_tokens(prompt_details)
detail_metrics: Final[list[tuple[Any, DEFINED_PROMETHEUS_METRICS, object]]] = [
detail_metrics: Final[list[tuple[_MetricLike, DEFINED_PROMETHEUS_METRICS, object]]] = [
(
self.litellm_input_cached_tokens_metric,
"litellm_input_cached_tokens_metric",
@ -1705,7 +1708,7 @@ class PrometheusLogger(CustomLogger):
if not isinstance(usage_object, dict):
return
media_metrics: Final[list[tuple[Any, DEFINED_PROMETHEUS_METRICS, object]]] = [
media_metrics: Final[list[tuple[_MetricLike, DEFINED_PROMETHEUS_METRICS, object]]] = [
(
self.litellm_video_duration_seconds_metric,
"litellm_video_duration_seconds_metric",
@ -1727,7 +1730,7 @@ class PrometheusLogger(CustomLogger):
def _inc_sparse_usage_counters(
self,
counters_with_values: Sequence[tuple[Any, DEFINED_PROMETHEUS_METRICS, object]],
counters_with_values: Sequence[tuple[_MetricLike, DEFINED_PROMETHEUS_METRICS, object]],
enum_values: UserAPIKeyLabelValues,
label_context: PrometheusLabelFactoryContext | None = None,
) -> None:
@ -2607,7 +2610,7 @@ class PrometheusLogger(CustomLogger):
for all successful requests (both streaming and non-streaming).
"""
def _safe_get(self, obj: Any, key: str, default: object = None) -> Any:
def _safe_get(self, obj: object, key: str, default: object = None) -> Any:
"""Get value from dict or Pydantic model."""
if obj is None:
return default
@ -2623,7 +2626,7 @@ class PrometheusLogger(CustomLogger):
"""
standard_logging_payload: Final = request_kwargs.get("standard_logging_object", {}) or {}
_litellm_params: Final = request_kwargs.get("litellm_params", {}) or {}
_metadata_raw: Final = self._safe_get(standard_logging_payload, "metadata") or {}
_metadata_raw: Final[object] = self._safe_get(standard_logging_payload, "metadata") or {}
if isinstance(_metadata_raw, dict):
_metadata = _metadata_raw
else:
@ -4215,8 +4218,8 @@ class PrometheusLogger(CustomLogger):
def _safe_duration_seconds(
self,
start_time: Any,
end_time: Any,
start_time: object,
end_time: object,
) -> float | None:
"""
Compute the duration in seconds between two objects.

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.

View file

@ -6,12 +6,13 @@ Native provider tools (like Anthropic's web_search_20250305) are converted
to this format for consistent interception and execution.
"""
from collections.abc import Mapping
from typing import Any, Final
from litellm.constants import LITELLM_WEB_SEARCH_TOOL_NAME
def get_litellm_web_search_tool() -> dict[str, Any]:
def get_litellm_web_search_tool() -> dict[str, object]:
"""
Get the standard LiteLLM web search tool definition.
@ -49,7 +50,7 @@ def get_litellm_web_search_tool() -> dict[str, Any]:
}
def get_litellm_web_search_tool_openai() -> dict[str, Any]:
def get_litellm_web_search_tool_openai() -> dict[str, object]:
"""
Get the standard LiteLLM web search tool definition in OpenAI format.
@ -82,7 +83,7 @@ def get_litellm_web_search_tool_openai() -> dict[str, Any]:
}
def get_litellm_web_search_tool_responses() -> dict[str, Any]:
def get_litellm_web_search_tool_responses() -> dict[str, object]:
"""
Get the standard LiteLLM web search tool definition in Responses API format.
@ -114,7 +115,7 @@ def get_litellm_web_search_tool_responses() -> dict[str, Any]:
}
def is_web_search_tool_responses(tool: dict[str, Any]) -> bool:
def is_web_search_tool_responses(tool: Mapping[str, object]) -> bool:
"""
Check if a tool is a web search tool for the Responses API.
@ -195,7 +196,7 @@ def is_web_search_tool_chat_completion(tool: dict[str, Any]) -> bool:
return False
def is_anthropic_native_web_search_tool(tool: dict[str, Any]) -> bool:
def is_anthropic_native_web_search_tool(tool: Mapping[str, object]) -> bool:
"""
Check if a tool is an Anthropic-native ``web_search_*`` tool.

View file

@ -24,7 +24,7 @@ class WebSearchTransformation:
@staticmethod
def transform_request(
response: Any,
response: object,
stream: bool,
response_format: str = "anthropic",
) -> tuple[bool, list[dict]]:
@ -66,7 +66,7 @@ class WebSearchTransformation:
@staticmethod
def _detect_from_responses_response(
response: Any,
response: object,
) -> tuple[bool, list[dict]]:
"""Parse a Responses API response for ``litellm_web_search`` function calls.
@ -399,7 +399,7 @@ class WebSearchTransformation:
def build_web_search_tool_result_block(
tool_use_id: str,
search_response: SearchResponse | None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Build an Anthropic-native ``web_search_tool_result`` content block.
@ -433,7 +433,7 @@ class WebSearchTransformation:
emitted with an empty result list (signals "search ran, no
results" rather than "search did not run").
"""
items: Final[list[dict[str, Any]]] = []
items: Final[list[dict[str, object]]] = []
if search_response is not None:
results: Final = getattr(search_response, "results", None) or []
for r in results:

View file

@ -6,7 +6,7 @@ Extends InteractionsHTTPHandler so that the shared HTTP infrastructure
duplicated. BaseAgentsAPIConfig stays as pure transform code.
"""
from collections.abc import Coroutine
from collections.abc import Coroutine, Mapping
from typing import Any, Final
import httpx
@ -39,11 +39,11 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
) -> AgentCreateResponse | Coroutine[Any, Any, AgentCreateResponse]:
) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
if _is_async:
return self.async_create_agent(
agents_api_config=agents_api_config,
@ -94,7 +94,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
) -> AgentCreateResponse:
@ -145,7 +145,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
) -> AgentListResponse | Coroutine[Any, Any, AgentListResponse]:
) -> AgentListResponse | Coroutine[object, object, AgentListResponse]:
if _is_async:
return self.async_list_agents(
agents_api_config=agents_api_config,
@ -220,7 +220,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
) -> AgentCreateResponse | Coroutine[Any, Any, AgentCreateResponse]:
) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
if _is_async:
return self.async_get_agent(
agents_api_config=agents_api_config,
@ -299,7 +299,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
) -> AgentDeleteResult | Coroutine[Any, Any, AgentDeleteResult]:
) -> AgentDeleteResult | Coroutine[object, object, AgentDeleteResult]:
if _is_async:
return self.async_delete_agent(
agents_api_config=agents_api_config,
@ -378,7 +378,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
) -> AgentVersionsResponse | Coroutine[Any, Any, AgentVersionsResponse]:
) -> AgentVersionsResponse | Coroutine[object, object, AgentVersionsResponse]:
if _is_async:
return self.async_list_agent_versions(
agents_api_config=agents_api_config,

View file

@ -30,7 +30,7 @@ Usage:
import asyncio
import contextvars
from collections.abc import Coroutine
from collections.abc import Coroutine, Mapping
from functools import partial
from typing import Any, Final
@ -75,7 +75,7 @@ def _make_logging_obj(
model: str,
custom_llm_provider: str,
call_type: str,
optional_params: dict[str, Any],
optional_params: dict[str, object],
) -> LiteLLMLoggingObj:
litellm_logging_obj: Final[LiteLLMLoggingObj] = kwargs.get("litellm_logging_obj")
litellm_call_id: Final[str | None] = kwargs.get("litellm_call_id", None)
@ -102,7 +102,7 @@ async def acreate(
base_environment: InteractionEnvironment | None = None,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentCreateResponse:
@ -146,10 +146,10 @@ def create(
base_environment: InteractionEnvironment | None = None,
custom_llm_provider: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentCreateResponse | Coroutine[Any, Any, AgentCreateResponse]:
) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
"""
Sync: Create a managed agent on the provider side.
@ -244,7 +244,7 @@ def list(
extra_headers: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentListResponse | Coroutine[Any, Any, AgentListResponse]:
) -> AgentListResponse | Coroutine[object, object, AgentListResponse]:
"""Sync: List all agents on the provider side."""
local_vars: Final = locals()
custom_llm_provider = custom_llm_provider or kwargs.get("custom_llm_provider") or "gemini"
@ -320,7 +320,7 @@ def get(
extra_headers: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentCreateResponse | Coroutine[Any, Any, AgentCreateResponse]:
) -> AgentCreateResponse | Coroutine[object, object, AgentCreateResponse]:
"""Sync: Get a specific agent by name."""
local_vars: Final = locals()
custom_llm_provider = custom_llm_provider or kwargs.get("custom_llm_provider") or "gemini"
@ -397,7 +397,7 @@ def delete(
extra_headers: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentDeleteResult | Coroutine[Any, Any, AgentDeleteResult]:
) -> AgentDeleteResult | Coroutine[object, object, AgentDeleteResult]:
"""Sync: Delete a specific agent by name."""
local_vars: Final = locals()
custom_llm_provider = custom_llm_provider or kwargs.get("custom_llm_provider") or "gemini"
@ -474,7 +474,7 @@ def list_versions(
extra_headers: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
**kwargs,
) -> AgentVersionsResponse | Coroutine[Any, Any, AgentVersionsResponse]:
) -> AgentVersionsResponse | Coroutine[object, object, AgentVersionsResponse]:
"""Sync: List versions of a specific agent."""
local_vars: Final = locals()
custom_llm_provider = custom_llm_provider or kwargs.get("custom_llm_provider") or "gemini"

View file

@ -4,7 +4,7 @@ HTTP Handler for Interactions API requests.
This module handles the HTTP communication for the Google Interactions API.
"""
from collections.abc import AsyncIterator, Coroutine, Iterator
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping
from typing import Any, Final
import httpx
@ -96,8 +96,8 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
model: str | None = None,
agent: str | None = None,
input: InteractionInput | None = None,
extra_headers: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
@ -105,7 +105,7 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
) -> (
InteractionsAPIResponse
| Iterator[InteractionsAPIStreamingResponse]
| Coroutine[Any, Any, InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]]
| Coroutine[object, object, InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]]
):
"""
Create a new interaction (synchronous or async based on _is_async flag).
@ -211,8 +211,8 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
model: str | None = None,
agent: str | None = None,
input: InteractionInput | None = None,
extra_headers: dict[str, Any] | None = None,
extra_body: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
stream: bool | None = None,
@ -345,11 +345,11 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
) -> InteractionsAPIResponse | Coroutine[Any, Any, InteractionsAPIResponse]:
) -> InteractionsAPIResponse | Coroutine[object, object, InteractionsAPIResponse]:
"""Get an interaction by ID."""
if _is_async:
return self.async_get_interaction(
@ -407,7 +407,7 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
) -> InteractionsAPIResponse:
@ -464,11 +464,11 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
) -> DeleteInteractionResult | Coroutine[Any, Any, DeleteInteractionResult]:
) -> DeleteInteractionResult | Coroutine[object, object, DeleteInteractionResult]:
"""Delete an interaction by ID."""
if _is_async:
return self.async_delete_interaction(
@ -527,7 +527,7 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
) -> DeleteInteractionResult:
@ -585,11 +585,11 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: HTTPHandler | None = None,
_is_async: bool = False,
) -> CancelInteractionResult | Coroutine[Any, Any, CancelInteractionResult]:
) -> CancelInteractionResult | Coroutine[object, object, CancelInteractionResult]:
"""Cancel an interaction by ID."""
if _is_async:
return self.async_cancel_interaction(
@ -648,7 +648,7 @@ class InteractionsHTTPHandler(_BaseHTTPHandler):
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
client: AsyncHTTPHandler | None = None,
) -> CancelInteractionResult:

View file

@ -34,8 +34,8 @@ class LiteLLMResponsesInteractionsConfig:
model: str,
input: InteractionInput | None,
optional_params: InteractionsAPIOptionalRequestParams,
**kwargs,
) -> dict[str, Any]:
**kwargs: object,
) -> dict[str, object]:
"""
Transform an Interactions API request to a Responses API request.
@ -45,7 +45,7 @@ class LiteLLMResponsesInteractionsConfig:
- tools -> tools (similar format)
- generation_config -> temperature, top_p, etc.
"""
responses_request: Final[dict[str, Any]] = {
responses_request: Final[dict[str, object]] = {
"model": model,
}
@ -201,15 +201,15 @@ class LiteLLMResponsesInteractionsConfig:
- Extract usage
"""
# Extract text from outputs and build both `outputs` (legacy) and `steps` (new schema).
outputs: Final[list[dict[str, Any]]] = []
steps: Final[list[dict[str, Any]]] = []
outputs: Final[list[dict[str, object]]] = []
steps: Final[list[dict[str, object]]] = []
if hasattr(responses_response, "output") and responses_response.output:
for output_item in responses_response.output:
# Use getattr with None default to safely access content
content = getattr(output_item, "content", None)
if content is not None:
content_items = content if isinstance(content, list) else [content]
model_output_contents: list[dict[str, Any]] = []
model_output_contents: list[dict[str, object]] = []
for content_item in content_items:
# Check if content_item has text attribute
text = getattr(content_item, "text", None)
@ -264,7 +264,7 @@ class LiteLLMResponsesInteractionsConfig:
# Add usage if available
# Map Responses API usage (input_tokens, output_tokens) to Interactions API spec format
# (total_input_tokens, total_output_tokens)
usage: Final = getattr(responses_response, "usage", None)
usage: Final[object] = getattr(responses_response, "usage", None)
if usage:
interactions_response_dict["usage"] = {
"total_input_tokens": getattr(usage, "input_tokens", 0),

View file

@ -229,7 +229,7 @@ def create(
) -> (
InteractionsAPIResponse
| Iterator[InteractionsAPIStreamingResponse]
| Coroutine[Any, Any, InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]]
| Coroutine[object, object, InteractionsAPIResponse | AsyncIterator[InteractionsAPIStreamingResponse]]
):
"""
Sync: Create a new interaction using Google's Interactions API.
@ -406,7 +406,7 @@ def get(
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
) -> InteractionsAPIResponse | Coroutine[Any, Any, InteractionsAPIResponse]:
) -> InteractionsAPIResponse | Coroutine[object, object, InteractionsAPIResponse]:
"""Sync: Get an interaction by its ID."""
local_vars: Final = locals()
custom_llm_provider = custom_llm_provider or "gemini"
@ -510,7 +510,7 @@ def delete(
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
) -> DeleteInteractionResult | Coroutine[Any, Any, DeleteInteractionResult]:
) -> DeleteInteractionResult | Coroutine[object, object, DeleteInteractionResult]:
"""Sync: Delete an interaction by its ID."""
local_vars: Final = locals()
custom_llm_provider = custom_llm_provider or "gemini"
@ -612,7 +612,7 @@ def cancel(
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
) -> CancelInteractionResult | Coroutine[Any, Any, CancelInteractionResult]:
) -> CancelInteractionResult | Coroutine[object, object, CancelInteractionResult]:
"""Sync: Cancel an interaction by its ID."""
local_vars: Final = locals()
custom_llm_provider = custom_llm_provider or "gemini"

View file

@ -419,7 +419,7 @@ def safe_deep_copy(data):
if litellm.safe_memory_mode is True:
return data
litellm_parent_otel_span: Any | None = None
litellm_parent_otel_span: object | None = None
# Step 1: Remove the litellm_parent_otel_span
litellm_parent_otel_span = None
if isinstance(data, dict):
@ -510,7 +510,7 @@ def independent_snapshot(
}
def filter_exceptions_from_params(data: Any, max_depth: int = 20) -> Any:
def filter_exceptions_from_params(data: object, max_depth: int = 20) -> Any:
"""
Recursively filter out Exception objects and callable objects from dicts/lists.
@ -542,7 +542,7 @@ def filter_exceptions_from_params(data: Any, max_depth: int = 20) -> Any:
return None
if isinstance(data, dict):
result: Final[dict[str, Any]] = {}
result: Final[dict[str, object]] = {}
for k, v in data.items():
# Skip exception and callable values
if isinstance(v, Exception) or (callable(v) and not isinstance(v, type)):
@ -556,7 +556,7 @@ def filter_exceptions_from_params(data: Any, max_depth: int = 20) -> Any:
continue
return result
elif isinstance(data, list):
result_list: Final[list[Any]] = []
result_list: Final[list[object]] = []
for item in data:
# Skip exception and callable items
if isinstance(item, Exception) or (callable(item) and not isinstance(item, type)):
@ -624,7 +624,7 @@ def redact_nested_match_and_regex_keys(
# Iterative traversal; `seen` guards against cyclic refs preserved by deepcopy.
try:
seen: Final[set] = set()
stack: Final[list[Any]] = [redacted]
stack: Final[list[object]] = [redacted]
while stack:
node = stack.pop()
node_id = id(node)

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.

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:

View file

@ -168,7 +168,7 @@ class ResponseMetadata:
def update_response_metadata(
result: Any,
result: object,
logging_obj: LiteLLMLoggingObject,
model: str | None,
kwargs: dict,

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

View file

@ -1708,8 +1708,8 @@ def _find_server_tool_result(
def convert_to_anthropic_tool_invoke(
tool_calls: list[ChatCompletionAssistantToolCall],
web_search_results: list[Any] | None = None,
tool_results: list[Any] | None = None,
web_search_results: Sequence[object] | None = None,
tool_results: Sequence[object] | None = None,
) -> list[AnthropicMessagesToolUseParam | dict[str, Any]]:
"""
OpenAI tool invokes:
@ -5349,7 +5349,7 @@ class NormalizedToolCall(TypedDict):
arguments: dict[str, object]
def _parse_tool_call_arguments(raw: Any, tool_name: str | None, context: str) -> dict[str, object]:
def _parse_tool_call_arguments(raw: object, tool_name: str | None, context: str) -> dict[str, object]:
# Anthropic's tool_use blocks already carry a parsed dict in "input";
# chat completions and the Responses API carry a JSON string that may be
# truncated by the model, so route those through the repair-aware parser.

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"

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.

View file

@ -4,7 +4,7 @@ import copy
import json
import traceback
from collections import deque
from collections.abc import AsyncIterator, Iterator, Sequence
from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
from typing import (
TYPE_CHECKING,
Any,
@ -423,7 +423,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
augmented["usage"] = augmented_usage
return augmented
def _next_compaction_event(self) -> dict[str, Any] | None:
def _next_compaction_event(self) -> dict[str, object] | None:
"""Return the next compaction content-block SSE event, or ``None``.
Anthropic delivers compaction as a single delta (no token-by-token
@ -462,7 +462,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
"delta": {"type": "compaction_delta", "content": summary_content},
}
stop_event: Final = {
stop_event: Final[dict[str, object]] = {
"type": "content_block_stop",
"index": compaction_index,
}
@ -994,7 +994,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
self.current_content_block_index += 1
@staticmethod
def _delta_has_content(processed_chunk: dict[str, Any]) -> bool:
def _delta_has_content(processed_chunk: Mapping[str, object]) -> bool:
"""Return True if a translated chunk carries a non-empty
``content_block_delta`` payload.

View file

@ -9,6 +9,7 @@ the LLM doesn't make a tool call, and we need to return a stream to the user.
"""
import json
from collections.abc import Mapping
from typing import Any, Final, cast
from litellm.types.llms.anthropic_messages.anthropic_response import (
@ -38,7 +39,7 @@ class FakeAnthropicMessagesStreamIterator:
self.chunks = self._create_streaming_chunks()
self.current_index = 0
def _create_content_block_chunks(self, block_dict: dict[str, Any], index: int) -> list[bytes]:
def _create_content_block_chunks(self, block_dict: Mapping[str, object], index: int) -> list[bytes]:
"""Build SSE chunks for a single content block."""
chunks: Final = []
block_type: Final = block_dict.get("type")
@ -133,14 +134,14 @@ class FakeAnthropicMessagesStreamIterator:
response_dict: Final = cast(dict[str, Any], self.response)
# 1. message_start event
usage: Final = response_dict.get("usage", {})
usage: Final = self.response.get("usage")
message_start: Final = {
"type": "message_start",
"message": {
"id": response_dict.get("id"),
"id": self.response.get("id"),
"type": "message",
"role": response_dict.get("role", "assistant"),
"model": response_dict.get("model"),
"role": self.response.get("role", "assistant"),
"model": self.response.get("model"),
"content": [],
"stop_reason": None,
"stop_sequence": None,
@ -161,21 +162,24 @@ class FakeAnthropicMessagesStreamIterator:
# 5. message_delta event (with final usage and stop_reason)
# Include cache usage fields so clients that only read message_delta
# (like Claude Code's SDK) see the full input token breakdown.
delta_usage: Final[dict[str, Any]] = {
delta_usage: Final[dict[str, int]] = {
"output_tokens": usage.get("output_tokens", 0) if usage else 0,
}
if usage:
if usage.get("input_tokens") is not None:
delta_usage["input_tokens"] = usage["input_tokens"]
if usage.get("cache_creation_input_tokens") is not None:
delta_usage["cache_creation_input_tokens"] = usage["cache_creation_input_tokens"]
if usage.get("cache_read_input_tokens") is not None:
delta_usage["cache_read_input_tokens"] = usage["cache_read_input_tokens"]
input_tokens: Final = usage.get("input_tokens")
if input_tokens is not None:
delta_usage["input_tokens"] = input_tokens
cache_creation_input_tokens: Final = usage.get("cache_creation_input_tokens")
if cache_creation_input_tokens is not None:
delta_usage["cache_creation_input_tokens"] = cache_creation_input_tokens
cache_read_input_tokens: Final = usage.get("cache_read_input_tokens")
if cache_read_input_tokens is not None:
delta_usage["cache_read_input_tokens"] = cache_read_input_tokens
message_delta: Final = {
"type": "message_delta",
"delta": {
"stop_reason": response_dict.get("stop_reason"),
"stop_sequence": response_dict.get("stop_sequence"),
"stop_reason": self.response.get("stop_reason"),
"stop_sequence": self.response.get("stop_sequence"),
},
"usage": delta_usage,
}

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]:
"""

View file

@ -82,7 +82,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
Processes both `system` and `messages` content blocks.
"""
def _sanitize(cache_control: Any) -> None:
def _sanitize(cache_control: object) -> None:
if isinstance(cache_control, dict):
cache_control.pop("scope", None)
@ -147,7 +147,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
return system_param
@staticmethod
def _as_system_content_blocks(value: Any) -> list:
def _as_system_content_blocks(value: object) -> list:
if value is None:
return []
if isinstance(value, list):
@ -157,7 +157,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
return [value]
@staticmethod
def _is_system_role_message(message: Any) -> bool:
def _is_system_role_message(message: object) -> bool:
return isinstance(message, dict) and message.get("role") == "system"
_CONVERTED_SYSTEM_NOTE: Final = (

View file

@ -48,9 +48,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": {
@ -74,7 +74,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
@ -87,7 +87,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):
@ -253,7 +253,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()

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

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,

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

View file

@ -105,7 +105,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
@ -132,7 +132,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)
@ -300,7 +300,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:

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.

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

View file

@ -55,7 +55,7 @@ class BaseTranslation(ABC):
@staticmethod
def transform_user_api_key_dict_to_metadata(
user_api_key_dict: Any | None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Transform user_api_key_dict to a metadata dict with prefixed keys.
@ -78,7 +78,7 @@ class BaseTranslation(ABC):
return {}
# Transform keys to be prefixed with 'user_api_key_'
transformed: Final = {}
transformed: Final[dict[str, object]] = {}
for key, value in user_dict.items():
# Skip None values and internal fields
if value is None or key.startswith("_"):
@ -174,7 +174,7 @@ class BaseTranslation(ABC):
self,
exc: "ModifyResponseException",
stream_started: bool = False,
responses_so_far: Sequence[Any] | None = None,
responses_so_far: Sequence[object] | None = None,
) -> Sequence[bytes] | None:
"""
Build the streaming chunks that deliver a guardrail block message and
@ -197,8 +197,8 @@ class BaseTranslation(ABC):
def build_stream_error_items(
self,
exc: "HTTPException",
responses_so_far: Sequence[Any] | None = None,
) -> Sequence[Any] | None:
responses_so_far: Sequence[object] | None = None,
) -> Sequence[object] | None:
"""
Build the stream items that surface a guardrail HTTPException (a block
with the default exception-on-block config, or a failed scan) after the

View file

@ -2,7 +2,7 @@ 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 pydantic import BaseModel
@ -10,7 +10,7 @@ from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUs
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):
@ -38,7 +38,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
@ -81,7 +81,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.
@ -191,7 +191,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)
@ -200,7 +200,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)

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]:

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, overload
@ -137,7 +137,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.
"""
@ -147,8 +147,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``.
@ -283,7 +283,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

View file

@ -163,7 +163,7 @@ async def make_call(
json_mode: bool | None = False,
bedrock_invoke_provider: litellm.BEDROCK_INVOKE_PROVIDERS_LITERAL | None = None,
stream_chunk_size: int | None = None,
) -> tuple[Any, httpx.Headers]:
) -> "tuple[MockResponseIterator | AsyncIterator[GChunk | ModelResponseStream | dict], httpx.Headers]":
try:
if client is None:
client = get_async_httpx_client(
@ -199,7 +199,9 @@ async def make_call(
messages=messages,
encoding=litellm.encoding,
)
completion_stream: Any = MockResponseIterator(model_response=model_response, json_mode=json_mode)
completion_stream: MockResponseIterator | AsyncIterator[GChunk | ModelResponseStream | dict] = (
MockResponseIterator(model_response=model_response, json_mode=json_mode)
)
elif bedrock_invoke_provider == "anthropic":
decoder: AWSEventStreamDecoder = AmazonAnthropicClaudeStreamDecoder(
model=model,
@ -248,7 +250,7 @@ def make_sync_call(
json_mode: bool | None = False,
bedrock_invoke_provider: litellm.BEDROCK_INVOKE_PROVIDERS_LITERAL | None = None,
stream_chunk_size: int | None = None,
) -> tuple[Any, httpx.Headers]:
) -> "tuple[MockResponseIterator | Iterator[GChunk | ModelResponseStream | dict], httpx.Headers]":
try:
if client is None:
client = _get_httpx_client(
@ -283,7 +285,9 @@ def make_sync_call(
messages=messages,
encoding=litellm.encoding,
)
completion_stream: Any = MockResponseIterator(model_response=model_response, json_mode=json_mode)
completion_stream: MockResponseIterator | Iterator[GChunk | ModelResponseStream | dict] = (
MockResponseIterator(model_response=model_response, json_mode=json_mode)
)
elif bedrock_invoke_provider == "anthropic":
decoder: AWSEventStreamDecoder = AmazonAnthropicClaudeStreamDecoder(
model=model,

View file

@ -52,8 +52,8 @@ _BEDROCK_AWS_AUTH_PARAMETER_KEYS: Final[tuple[str, ...]] = (
def merge_bedrock_aws_request_params(
litellm_params: Mapping[str, Any],
optional_params: Mapping[str, Any],
litellm_params: Mapping[str, object],
optional_params: Mapping[str, object],
) -> dict[str, Any]:
"""Merge deployment and request parameters without allowing auth escalation.
@ -303,7 +303,7 @@ def normalize_json_schema_custom_types_to_object(schema: dict) -> None:
Uses an explicit stack (not recursion) to satisfy recursive-function guards in CI.
"""
stack: Final[list[Any]] = [schema]
stack: Final[list[object]] = [schema]
seen: Final[set[int]] = set()
while stack:
node = stack.pop()
@ -913,7 +913,7 @@ def _get_bedrock_converse_strict_tools_flag(base_model: str) -> bool | None:
return None
def normalize_bedrock_opus_output_config_effort(model: str, output_config: Any) -> None:
def normalize_bedrock_opus_output_config_effort(model: str, output_config: object) -> None:
"""
Normalize Anthropic ``output_config.effort`` values for Bedrock Opus ids.
@ -1436,6 +1436,11 @@ class BedrockEventStreamDecoderBase:
return chunk.decode()
def _decoded_json_value(raw: str) -> object:
"""Decode a JSON document into an opaque value for isinstance narrowing."""
return json.loads(raw)
def get_anthropic_beta_from_headers(headers: dict) -> list[str]:
"""
Extract anthropic-beta header values and convert them to a list.
@ -1463,7 +1468,7 @@ def get_anthropic_beta_from_headers(headers: dict) -> list[str]:
anthropic_beta_header = anthropic_beta_header.strip()
if anthropic_beta_header.startswith("[") and anthropic_beta_header.endswith("]"):
try:
parsed: Final = json.loads(anthropic_beta_header)
parsed: Final = _decoded_json_value(anthropic_beta_header)
if isinstance(parsed, list):
return [str(beta).strip() for beta in parsed]
except json.JSONDecodeError:
@ -1476,8 +1481,8 @@ def get_anthropic_beta_from_headers(headers: dict) -> list[str]:
def resolve_s3_encryption_key_id(
litellm_params: Mapping[str, Any],
optional_params: Mapping[str, Any] | None = None,
litellm_params: Mapping[str, object],
optional_params: Mapping[str, object] | None = None,
) -> str | None:
"""
Resolve the SSE-KMS key configured for Bedrock batch/file S3 objects.

View file

@ -47,7 +47,7 @@ def _nova_canvas_task_body(
task_type: str | None,
mask_prompt: str | None,
out_painting_mode: str | None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""Build InvokeModel body task section (without imageGenerationConfig)."""
if task_type == "BACKGROUND_REMOVAL":
return {
@ -60,7 +60,7 @@ def _nova_canvas_task_body(
"OUTPAINTING requires either a mask image or a mask prompt. "
"Pass mask=<file> or maskPrompt=<str> in the request."
)
out_params: Final[dict[str, Any]] = {
out_params: Final[dict[str, object]] = {
"image": image_b64,
"text": text,
}
@ -79,7 +79,7 @@ def _nova_canvas_task_body(
# Honour explicit IMAGE_VARIATION even when a mask is present (mask is ignored
# for this task type; callers use INPAINTING when they want mask semantics).
if task_type == "IMAGE_VARIATION":
var_params_explicit: Final[dict[str, Any]] = {
var_params_explicit: Final[dict[str, object]] = {
"images": [image_b64],
"text": text,
}
@ -100,7 +100,7 @@ def _nova_canvas_task_body(
"or omit taskType for automatic routing (mask → INPAINTING, else IMAGE_VARIATION)."
)
if mask_b64 is not None or mask_prompt is not None or task_type == "INPAINTING":
in_params: Final[dict[str, Any]] = {"image": image_b64, "text": text}
in_params: Final[dict[str, object]] = {"image": image_b64, "text": text}
if mask_prompt is not None:
in_params["maskPrompt"] = mask_prompt
elif mask_b64 is not None:
@ -114,7 +114,7 @@ def _nova_canvas_task_body(
"See https://docs.aws.amazon.com/nova/latest/userguide/image-gen-req-resp-structure.html"
)
return {"taskType": "INPAINTING", "inPaintingParams": in_params}
var_params: Final[dict[str, Any]] = {
var_params: Final[dict[str, object]] = {
"images": [image_b64],
"text": text,
}
@ -250,9 +250,9 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
image_edit_optional_params: ImageEditOptionalRequestParams,
model: str,
drop_params: bool,
) -> dict[str, Any]:
) -> dict[str, object]:
supported: Final = set(self.get_supported_openai_params(model))
mapped: Final[dict[str, Any]] = dict(image_edit_optional_params)
mapped: Final[dict[str, object]] = dict(image_edit_optional_params)
_size: Final = mapped.pop("size", None)
if _size is not None and isinstance(_size, str) and "x" in _size:
w, h = _size.split("x", 1)
@ -327,7 +327,7 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
cfg_scale: Final = op.pop("cfgScale", None)
seed: Final = op.pop("seed", None)
image_generation_config: Final[dict[str, Any]] = {}
image_generation_config: Final[dict[str, object]] = {}
nested_igc: Final = op.pop("imageGenerationConfig", None)
if isinstance(nested_igc, dict):
image_generation_config.update(nested_igc)

View file

@ -203,7 +203,7 @@ class BedrockVectorStoreConfig(BaseVectorStoreConfig, BaseAWSLLM):
encoded_vector_store_id: Final = encode_url_path_segment(vector_store_id, field_name="vector_store_id")
url: Final = f"{api_base}/{encoded_vector_store_id}/retrieve"
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"retrievalQuery": BedrockKBRetrievalQuery(text=query),
}
@ -288,7 +288,7 @@ class BedrockVectorStoreConfig(BaseVectorStoreConfig, BaseAWSLLM):
data_source_id: Final = metadata.get("x-amz-bedrock-kb-data-source-id", "unknown") if metadata else "unknown"
return f"bedrock-kb-document-{data_source_id}"
def _get_attributes_from_metadata(self, metadata: dict[str, Any]) -> dict[str, Any]:
def _get_attributes_from_metadata(self, metadata: dict[str, object]) -> dict[str, object]:
"""
Extract all attributes from Bedrock KB metadata.
Returns a copy of the metadata dictionary.

View file

@ -84,7 +84,7 @@ class BlackForestLabsImageEditConfig(BaseImageEditConfig):
BFL-specific params are passed through directly.
"""
optional_params: Final[dict[str, Any]] = {}
optional_params: Final[dict[str, object]] = {}
# Pass through BFL-specific params
bfl_params: Final = [
@ -246,7 +246,7 @@ class BlackForestLabsImageEditConfig(BaseImageEditConfig):
b64_image: Final = base64.b64encode(image_bytes).decode("utf-8")
# Build request body
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"prompt": prompt,
"input_image": b64_image,
}

View file

@ -10,7 +10,8 @@ Convers
Docs - https://docs.cohere.com/v2/reference/embed
"""
from typing import Any, Final, cast
from collections.abc import Sized
from typing import Final, Protocol, cast
import httpx
@ -30,6 +31,12 @@ from litellm.utils import is_base64_encoded
from ..common_utils import CohereError
class _SupportsEncode(Protocol):
"""Tokenizer handle: the embedding usage path only encodes text to measure its token length."""
def encode(self, text: str, /) -> Sized: ...
class CohereEmbeddingConfig(BaseEmbeddingConfig):
"""
Reference: https://docs.cohere.com/v2/reference/embed
@ -133,7 +140,7 @@ class CohereEmbeddingConfig(BaseEmbeddingConfig):
),
)
def _calculate_usage(self, input: list[str], encoding: Any, meta: dict) -> Usage:
def _calculate_usage(self, input: list[str], encoding: _SupportsEncode, meta: dict) -> Usage:
input_tokens = 0
text_tokens: Final[int | None] = meta.get("billed_units", {}).get("input_tokens")
@ -169,7 +176,7 @@ class CohereEmbeddingConfig(BaseEmbeddingConfig):
data: dict | CohereEmbeddingRequest,
model_response: EmbeddingResponse,
model: str,
encoding: Any,
encoding: _SupportsEncode,
input: list,
) -> EmbeddingResponse:
response_json: Final = response.json()

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
@ -340,7 +340,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:
@ -420,7 +420,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:

View file

@ -148,7 +148,7 @@ class DashScopeRerankConfig(BaseRerankConfig):
if "documents" not in optional_rerank_params:
raise ValueError("documents is required for DashScope rerank")
request: Final[dict[str, Any]] = {
request: Final[dict[str, object]] = {
"model": model,
"query": optional_rerank_params["query"],
"documents": optional_rerank_params["documents"],
@ -209,7 +209,7 @@ class DashScopeRerankConfig(BaseRerankConfig):
# which already matches LiteLLM's RerankResponseDocument shape.
transformed_results: Final[list[dict]] = []
for r in results:
item: dict[str, Any] = {
item: dict[str, object] = {
"index": r["index"],
"relevance_score": r["relevance_score"],
}

View file

@ -4,7 +4,7 @@ Calls DataForSEO SERP API to search the web.
DataForSEO API Reference: https://docs.dataforseo.com/v3/serp/google/organic/live/advanced/?bash
"""
from typing import Any, Final, Literal
from typing import Final, Literal
import httpx
@ -126,7 +126,7 @@ class DataForSEOSearchConfig(BaseSearchConfig):
List[Dict]: Request body for DataForSEO API (array of task objects as required by API)
"""
# DataForSEO expects an array of task objects
task: Final[dict[str, Any]] = {}
task: Final[dict[str, object]] = {}
# Convert query to string if it's a list
if isinstance(query, list):

View file

@ -80,8 +80,8 @@ class ElevenLabsTextToSpeechConfig(BaseTextToSpeechConfig):
def _resolve_voice_id(
self,
voice: str | dict[str, Any] | None,
params: dict[str, Any],
voice: str | dict[str, object] | None,
params: dict[str, object],
) -> str:
"""
Determine the ElevenLabs voice_id based on provided voice input or parameters.
@ -115,17 +115,17 @@ class ElevenLabsTextToSpeechConfig(BaseTextToSpeechConfig):
optional_params: dict,
voice: str | dict | None = None,
drop_params: bool = False,
kwargs: dict[str, Any] | None = None,
kwargs: dict[str, object] | None = None,
) -> tuple[str | None, dict]:
"""
Map OpenAI parameters to ElevenLabs TTS parameters
"""
mapped_params: Final[dict[str, Any]] = {}
query_params: Final[dict[str, Any]] = {}
mapped_params: Final[dict[str, object]] = {}
query_params: Final[dict[str, object]] = {}
# Work on a copy so we don't mutate the caller's dictionary
params: Final = dict(optional_params) if optional_params else {}
passthrough_kwargs: Final[dict[str, Any]] = kwargs if kwargs is not None else {}
passthrough_kwargs: Final[dict[str, object]] = kwargs if kwargs is not None else {}
# Extract voice identifier
mapped_voice: Final = self._resolve_voice_id(voice, params)
@ -205,7 +205,7 @@ class ElevenLabsTextToSpeechConfig(BaseTextToSpeechConfig):
params: Final = dict(optional_params) if optional_params else {}
extra_body: Final = params.pop("extra_body", None)
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"text": input,
"model_id": model,
}
@ -229,10 +229,10 @@ class ElevenLabsTextToSpeechConfig(BaseTextToSpeechConfig):
def _add_elevenlabs_specific_params(
self,
mapped_voice: str,
query_params: dict[str, Any],
mapped_params: dict[str, Any],
kwargs: dict[str, Any] | None,
remaining_params: dict[str, Any],
query_params: dict[str, object],
mapped_params: dict[str, object],
kwargs: dict[str, object] | None,
remaining_params: dict[str, object],
) -> None:
if kwargs is None:
kwargs = {}

View file

@ -67,11 +67,11 @@ class FireworksAIRerankConfig(FireworksAIMixin, BaseRerankConfig):
max_chunks_per_doc: int | None = None,
max_tokens_per_doc: int | None = None,
instruction: str | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Map Cohere rerank params to Fireworks AI rerank params
"""
params: Final[dict[str, Any]] = {
params: Final[dict[str, object]] = {
"query": query,
"documents": documents,
}

View file

@ -9,6 +9,7 @@ Proxies the Gemini v1beta Agents API:
GET /v1beta/agents/{name}/versions list versions
"""
from collections.abc import Mapping
from typing import Any, Final
import httpx
@ -87,7 +88,7 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
def get_complete_url(
self,
api_base: str | None,
litellm_params: dict[str, Any],
litellm_params: Mapping[str, object],
) -> str:
return f"{self._base_url(api_base)}/agents"
@ -132,9 +133,9 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
def transform_create_request(
self,
name: str,
litellm_params: dict[str, Any],
) -> dict[str, Any]:
body: Final[dict[str, Any]] = {"name": name}
litellm_params: Mapping[str, object],
) -> dict[str, object]:
body: Final[dict[str, object]] = {"name": name}
for key in _GEMINI_AGENT_BODY_KEYS:
value = litellm_params.get(key)
if value is not None:
@ -174,10 +175,10 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
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]]:
url: Final = f"{self._base_url(api_base)}/agents"
params: Final[dict[str, Any]] = {}
params: Final[dict[str, object]] = {}
if litellm_params.get("page_size"):
params["pageSize"] = litellm_params["page_size"]
if litellm_params.get("page_token"):
@ -207,8 +208,8 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
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]]:
url: Final = f"{self._base_url(api_base)}/agents/{name}"
return url, {}
@ -236,7 +237,7 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
self,
name: str,
api_base: str | None,
litellm_params: dict[str, Any],
litellm_params: Mapping[str, object],
) -> str:
return f"{self._base_url(api_base)}/agents/{name}"
@ -262,10 +263,10 @@ class GeminiAgentsConfig(BaseAgentsAPIConfig):
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]]:
url: Final = f"{self._base_url(api_base)}/agents/{name}/versions"
params: Final[dict[str, Any]] = {}
params: Final[dict[str, object]] = {}
if litellm_params.get("page_size"):
params["pageSize"] = litellm_params["page_size"]
if litellm_params.get("page_token"):

View file

@ -58,9 +58,9 @@ class GoogleAIStudioTokenCounter:
self,
api_base: str | None = None,
api_key: str | None = None,
headers: dict[str, Any] | None = None,
headers: dict[str, object] | None = None,
model: str = "",
litellm_params: dict[str, Any] | None = None,
litellm_params: dict[str, object] | None = None,
) -> tuple[dict[str, Any], str]:
"""
Returns a Tuple of headers and url for the Google Gen AI Studio countTokens endpoint.

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,

View file

@ -50,13 +50,21 @@ def _parse_data_url(data_url: str) -> tuple[bytes, str, str] | None:
return content_bytes, content_type, ext
def _content_type_or_default(headers: Mapping[str, str]) -> str:
"""Return the response's ``content-type`` header, falling back to ``image/jpeg`` when absent."""
try:
return headers["content-type"]
except KeyError:
return "image/jpeg"
def _download_image_sync(url: str) -> tuple[bytes, str, str]:
"""Download image from URL synchronously."""
client: Final = _get_httpx_client(params={"ssl_verify": False})
response: Final = client.get(url)
response.raise_for_status()
content_type: Final = response.headers.get("content-type", "image/jpeg")
content_type: Final = _content_type_or_default(response.headers)
ext: Final = content_type.split("/")[-1].split(";")[0] or "jpg"
return response.content, content_type, ext
@ -71,7 +79,7 @@ async def _download_image_async(url: str) -> tuple[bytes, str, str]:
response: Final = await client.get(url)
response.raise_for_status()
content_type: Final = response.headers.get("content-type", "image/jpeg")
content_type: Final = _content_type_or_default(response.headers)
ext: Final = content_type.split("/")[-1].split(";")[0] or "jpg"
return response.content, content_type, ext

View file

@ -1,7 +1,7 @@
import json
import os
from collections.abc import Callable
from typing import Any, Final, Literal, get_args
from collections.abc import Sequence
from typing import Final, Literal, Protocol, get_args
import httpx
@ -29,6 +29,12 @@ hf_tasks_embeddings: Final = (
)
class _SupportsTokenEncode(Protocol):
"""Token encoder handle. Only ``encode`` is ever called on it here."""
def encode(self, text: str, *, disallowed_special: tuple[str, ...]) -> Sequence[int]: ...
def get_hf_task_embedding_for_model(model: str, task_type: str | None, api_base: str) -> str | None:
if task_type is not None:
if task_type in get_args(hf_tasks_embeddings):
@ -173,7 +179,7 @@ class HuggingFaceEmbedding(BaseLLM):
model_response: EmbeddingResponse,
model: str,
input: list,
encoding: Any,
encoding: _SupportsTokenEncode,
) -> EmbeddingResponse:
output_data: Final = []
if "similarities" in embeddings:
@ -234,7 +240,7 @@ class HuggingFaceEmbedding(BaseLLM):
api_base: str,
api_key: str | None,
headers: dict,
encoding: Callable,
encoding: _SupportsTokenEncode,
client: AsyncHTTPHandler | None = None,
):
## TRANSFORMATION ##
@ -294,7 +300,7 @@ class HuggingFaceEmbedding(BaseLLM):
optional_params: dict,
litellm_params: dict,
logging_obj: LiteLLMLoggingObj,
encoding: Callable,
encoding: _SupportsTokenEncode,
api_key: str | None = None,
api_base: str | None = None,
timeout: float | httpx.Timeout = httpx.Timeout(None),

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,

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.
"""

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

View file

@ -5,6 +5,7 @@ Maps OpenAI TTS spec to MiniMax TTS API (WebSocket-based HTTP API)
Reference: https://platform.minimax.io/docs
"""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -86,8 +87,8 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
def _resolve_voice_id(
self,
voice: str | dict[str, Any] | None,
params: dict[str, Any],
voice: str | Mapping[str, object] | None,
params: dict[str, object],
) -> str:
"""
Determine the MiniMax voice_id based on provided voice input or parameters.
@ -122,12 +123,12 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
optional_params: dict,
voice: str | dict | None = None,
drop_params: bool = False,
kwargs: dict[str, Any] | None = None,
kwargs: Mapping[str, object] | None = None,
) -> tuple[str | None, dict]:
"""
Map OpenAI parameters to MiniMax TTS parameters
"""
mapped_params: Final[dict[str, Any]] = {}
mapped_params: Final[dict[str, object]] = {}
# Work on a copy so we don't mutate the caller's dictionary
params: Final = dict(optional_params) if optional_params else {}
@ -242,7 +243,7 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
# Output format: 'url' or 'hex' (default is 'hex')
output_format: Final = params.pop("output_format", "hex")
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"model": model,
"text": input,
"stream": False, # HTTP endpoint doesn't support streaming

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,

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)

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

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.

View file

@ -117,16 +117,16 @@ class OpenAICountTokensConfig:
def transform_request_to_count_tokens(
self,
model: str,
input: str | list[Any],
input: str | Sequence[object],
tools: list[dict[str, Any]] | None = None,
instructions: str | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Transform request to OpenAI Responses API token counting format.
The Responses API uses `input` (not `messages`) and `instructions` (not `system`).
"""
request: Final[dict[str, Any]] = {
request: Final[dict[str, object]] = {
"model": model,
"input": input,
}
@ -145,7 +145,7 @@ class OpenAICountTokensConfig:
"Authorization": f"Bearer {api_key}",
}
def validate_request(self, model: str, input: str | list[Any]) -> None:
def validate_request(self, model: str, input: str | Sequence[object]) -> None:
if not model:
raise ValueError("model parameter is required")
@ -155,18 +155,18 @@ class OpenAICountTokensConfig:
@staticmethod
def _transform_tools_for_responses_api(
tools: list[dict[str, Any]],
) -> list[dict[str, Any]]:
) -> list[dict[str, object]]:
"""
Transform OpenAI chat tools format to Responses API tools format.
Chat format: {"type": "function", "function": {"name": "...", "parameters": {...}}}
Responses format: {"type": "function", "name": "...", "parameters": {...}}
"""
transformed: Final = []
transformed: Final[list[dict[str, object]]] = []
for tool in tools:
if tool.get("type") == "function" and "function" in tool:
func = tool["function"]
item: dict[str, Any] = {
item: dict[str, object] = {
"type": "function",
"name": func.get("name", ""),
"description": func.get("description", ""),
@ -191,7 +191,7 @@ class OpenAICountTokensConfig:
(input_items, instructions) tuple where instructions is extracted
from system/developer messages.
"""
input_items: Final[list[dict[str, Any]]] = []
input_items: Final[list[dict[str, object]]] = []
instructions_parts: Final[list[str]] = []
for msg in messages:

View file

@ -311,7 +311,7 @@ def _patch_or_convert_request_fields(
return _RequestFields(input=tuple(input_items), instructions=converted_instructions)
def _next_stream_sequence_number(responses_so_far: Sequence[Any] | None) -> int:
def _next_stream_sequence_number(responses_so_far: Sequence[object] | None) -> int:
sequence_numbers: Final = (
item.get("sequence_number") if isinstance(item, dict) else getattr(item, "sequence_number", None)
for item in reversed(responses_so_far or ())
@ -484,7 +484,7 @@ class OpenAIResponsesHandler(BaseTranslation):
def _extract_input_text_and_images(
self,
message: Any,
message: Mapping[str, object],
msg_idx: int,
texts_to_check: list[str],
images_to_check: list[str],
@ -845,8 +845,8 @@ class OpenAIResponsesHandler(BaseTranslation):
def build_stream_error_items(
self,
exc: "HTTPException",
responses_so_far: Sequence[Any] | None = None,
) -> Sequence[Any] | None:
responses_so_far: Sequence[object] | None = None,
) -> Sequence[object] | None:
from litellm.proxy.common_request_processing import (
serialize_http_exception_detail,
)

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}", {}

View file

@ -98,7 +98,7 @@ class OpenAIVectorStoreConfig(BaseVectorStoreConfig):
api_base: str,
litellm_logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
extra_body: dict[str, Any] | None = None,
extra_body: dict[str, object] | None = None,
) -> tuple[str, dict]:
encoded_vector_store_id: Final = encode_url_path_segment(vector_store_id, field_name="vector_store_id")
url: Final = f"{api_base}/{encoded_vector_store_id}/search"

View file

@ -571,8 +571,8 @@ class OpenAIVideoConfig(BaseVideoConfig):
def _add_image_to_files(
self,
files_list: list[tuple[str, Any]],
image: Any,
files_list: list[tuple[str, FileTypes]],
image: FileContent,
field_name: str,
) -> None:
"""Add an image to the files list with appropriate content type"""

View file

@ -152,7 +152,7 @@ class OpenRouterImageEditConfig(BaseImageEditConfig):
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[dict, RequestFiles]:
content_parts: Final[list[dict[str, Any]]] = []
content_parts: Final[list[dict[str, object]]] = []
# Add source image(s) as base64 data URLs
if image is not None:
@ -174,7 +174,7 @@ class OpenRouterImageEditConfig(BaseImageEditConfig):
if prompt:
content_parts.append({"type": "text", "text": prompt})
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"model": model,
"messages": [
{

View file

@ -6,6 +6,7 @@ It uses the field targeting configuration from litellm_logging_obj
to extract specific fields for guardrail processing.
"""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final, Optional
from litellm._logging import verbose_proxy_logger
@ -89,7 +90,7 @@ class PassThroughEndpointHandler(BaseTranslation):
data: dict,
guardrail_to_apply: "CustomGuardrail",
litellm_logging_obj: Optional["LiteLLMLoggingObj"] = None,
) -> Any:
) -> Mapping[str, object]:
"""
Process input by applying guardrails to targeted fields or full payload.
"""
@ -127,12 +128,12 @@ class PassThroughEndpointHandler(BaseTranslation):
async def process_output_response(
self,
response: Any,
response: object,
guardrail_to_apply: "CustomGuardrail",
litellm_logging_obj: Optional["LiteLLMLoggingObj"] = None,
user_api_key_dict: Any | None = None,
user_api_key_dict: Optional["UserAPIKeyAuth"] = None,
request_data: dict | None = None,
) -> Any:
) -> object:
"""
Process output response by applying guardrails to targeted fields.
@ -236,12 +237,12 @@ class LlmPassthroughRouteHandler(BaseTranslation):
async def process_output_response(
self,
response: Any,
response: object,
guardrail_to_apply: "CustomGuardrail",
litellm_logging_obj: Optional["LiteLLMLoggingObj"] = None,
user_api_key_dict: Any | None = None,
user_api_key_dict: Optional["UserAPIKeyAuth"] = None,
request_data: dict | None = None,
) -> Any:
) -> object:
provider: Final = (request_data or {}).get("custom_llm_provider")
handler_cls: Final = _get_provider_handlers().get(provider or "")
if handler_cls is None:

View file

@ -13,7 +13,7 @@ This module decodes them into float arrays for OpenAI-compatible responses.
import base64
import struct
from typing import Any, Final
from typing import Final
import httpx
@ -117,7 +117,7 @@ class PerplexityEmbeddingConfig(BaseEmbeddingConfig):
}
@staticmethod
def _decode_base64_embedding(embedding_value: Any) -> list[float]:
def _decode_base64_embedding(embedding_value: object) -> object:
"""
Decode a Perplexity embedding into a list of floats.
@ -154,7 +154,7 @@ class PerplexityEmbeddingConfig(BaseEmbeddingConfig):
model_response.object = raw_response_json.get("object", "list")
raw_data: Final = raw_response_json.get("data", [])
decoded_data: Final[list[dict[str, Any]]] = []
decoded_data: Final[list[dict[str, object]]] = []
for item in raw_data:
decoded_item = dict(item)
decoded_item["embedding"] = self._decode_base64_embedding(item.get("embedding"))

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,9 +66,24 @@ class PredibaseConfig(BaseConfig):
typical_p: float | None = None,
watermark: bool | None = None,
) -> None:
locals_: Final = locals().copy()
for key, value in locals_.items():
if key != "self" and value is not None:
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 value is not None:
setattr(self.__class__, key, value)
@classmethod
@ -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", "")))

View file

@ -1,3 +1,4 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -91,7 +92,7 @@ class RAGFlowVectorStoreConfig(BaseVectorStoreConfig):
api_base: str,
litellm_logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
extra_body: dict[str, Any] | None = None,
extra_body: Mapping[str, object] | None = None,
) -> tuple[str, dict]:
"""RAGFlow vector stores are management-only, search is not supported."""
raise NotImplementedError("RAGFlow vector stores support dataset management only, not search/retrieval")
@ -121,7 +122,7 @@ class RAGFlowVectorStoreConfig(BaseVectorStoreConfig):
raise ValueError("name is required for RAGFlow dataset creation")
# Build request body
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"name": name,
}

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:

Some files were not shown because too many files have changed in this diff Show more