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

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

Browse source
This commit is contained in:
mateo-berri 2026-09-02 15:35:01 +00:00
parent f94bd6d903
commit 362fb4cffe
42 changed files with 336 additions and 213 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
litellm/a2a_protocol/providers/bedrock_agentcore/handler.py
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,

4
litellm/a2a_protocol/streaming_iterator.py
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)),

7
litellm/a2a_protocol/utils.py
View file

@ -2,6 +2,7 @@
Utility functions for A2A protocol.
"""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import litellm
@ -46,7 +47,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.
@ -109,7 +110,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.
@ -145,5 +146,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)

4
litellm/caching/caching_handler.py
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,

12
litellm/experimental_mcp_client/client.py
View file

@ -5,7 +5,7 @@ LiteLLM Proxy uses this MCP Client to connnect to other MCP servers.
import asyncio
import base64
import os
from collections.abc import Awaitable, Callable, Generator, Sequence
from collections.abc import Awaitable, Callable, Generator
from contextlib import AbstractAsyncContextManager
from datetime import timedelta
from functools import partial
@ -13,11 +13,19 @@ from importlib import metadata
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
_TransportContext: TypeAlias = AbstractAsyncContextManager[Sequence[Any]]
_TransportStreams: TypeAlias = tuple[
MemoryObjectReceiveStream[SessionMessage | Exception],
MemoryObjectSendStream[SessionMessage],
Unpack[tuple[object, ...]],
]
_TransportContext: TypeAlias = AbstractAsyncContextManager[_TransportStreams]
class _StreamableHttpClientFactory(Protocol):

3
litellm/files/main.py
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[
@ -1002,7 +1003,7 @@ def file_content_streaming(
timeout: float | httpx.Timeout,
logging_obj: LiteLLMLoggingObj | None,
_is_async: bool,
client: Any | None,
client: OpenAI | AsyncOpenAI | None,
) -> FileContentStreamingResult | Coroutine[object, object, FileContentStreamingResult]:
if logging_obj is not None:
logging_obj.model = model or ""

23
litellm/integrations/arize/_utils.py
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}}

50
litellm/integrations/focus/destinations/s3_destination.py
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,

13
litellm/integrations/prometheus.py
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:
@ -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:

30
litellm/interactions/http_handler.py
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:

32
litellm/llms/anthropic/experimental_pass_through/messages/fake_stream_iterator.py
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,
}

12
litellm/llms/bedrock/chat/invoke_handler.py
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,

13
litellm/llms/cohere/embed/transformation.py
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()

4
litellm/llms/dashscope/rerank/transformation.py
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"],
}

4
litellm/llms/dataforseo/search/transformation.py
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):

22
litellm/llms/elevenlabs/text_to_speech/transformation.py
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 = {}

4
litellm/llms/fireworks_ai/rerank/transformation.py
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,
}

4
litellm/llms/gemini/count_tokens/handler.py
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.

12
litellm/llms/gigachat/file_handler.py
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

16
litellm/llms/huggingface/embedding/handler.py
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),

2
litellm/llms/minimax/text_to_speech/transformation.py
View file

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

2
litellm/llms/openai/vector_stores/transformation.py
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"

11
litellm/llms/pass_through/guardrail_translation/handler.py
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.
"""
@ -130,9 +131,9 @@ class PassThroughEndpointHandler(BaseTranslation):
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.
@ -239,9 +240,9 @@ class LlmPassthroughRouteHandler(BaseTranslation):
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:

17
litellm/llms/vertex_ai/text_to_speech/transformation.py
View file

@ -29,6 +29,7 @@ from litellm.types.llms.vertex_ai_text_to_speech import (
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
@ -131,19 +132,19 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
model: str,
input: str,
voice: str | dict | None,
optional_params: dict,
litellm_params_dict: dict,
optional_params: dict[str, object],
litellm_params_dict: dict[str, object],
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 Vertex AI TTS requests
@ -227,7 +228,7 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
Returns:
Tuple of (mapped_voice_str, mapped_params)
"""
mapped_params: Final[dict[str, Any]] = {}
mapped_params: Final[dict[str, object]] = {}
##########################################################
# Map voice using helper
@ -428,7 +429,7 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
speakingRate=speaking_rate,
)
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"input": dict(vertex_input),
"voice": dict(vertex_voice),
"audioConfig": dict(vertex_audio_config),

14
litellm/proxy/client/cli/commands/auth.py
View file

@ -1,8 +1,8 @@
import sys
import time
import webbrowser
from collections.abc import Callable, Mapping
from typing import Any, Final
from collections.abc import Callable, Mapping, Sequence
from typing import Any, Final, TypeVar
from urllib.parse import urlencode
import click
@ -112,6 +112,8 @@ class CliAuthResult(TypedDict):
team_id: str | None
_TeamMapping: Final = TypeVar("_TeamMapping", bound=Mapping[str, object])
KEYRING_INSTALL_HINT: Final = "pip install 'litellm[cli]'"
KEYRING_ENABLE_HINT: Final = "keyring --enable (or unset PYTHON_KEYRING_BACKEND)"
@ -353,7 +355,7 @@ def get_key_input():
return None
def display_interactive_team_selection(teams: list[dict[str, Any]], selected_index: int = 0) -> None:
def display_interactive_team_selection(teams: Sequence[Mapping[str, Any]], selected_index: int = 0) -> None:
"""Display teams with one highlighted for selection"""
console: Final = Console()
@ -391,7 +393,7 @@ def display_interactive_team_selection(teams: list[dict[str, Any]], selected_ind
console.print(f" Budget: [dim]{budget_str}[/dim]\n")
def prompt_team_selection(teams: list[dict[str, Any]]) -> dict[str, Any] | None:
def prompt_team_selection(teams: Sequence[_TeamMapping]) -> _TeamMapping | None:
"""Interactive team selection with arrow keys"""
if not teams:
return None
@ -441,8 +443,8 @@ def prompt_team_selection(teams: list[dict[str, Any]]) -> dict[str, Any] | None:
def prompt_team_selection_fallback(
teams: list[dict[str, Any]],
) -> dict[str, Any] | None:
teams: Sequence[_TeamMapping],
) -> _TeamMapping | None:
"""Fallback team selection for non-interactive environments"""
if not teams:
return None

22
litellm/proxy/client/cli/commands/models.py
View file

@ -1,17 +1,32 @@
# stdlib imports
import re
from collections import defaultdict
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Final, Literal
from typing import TYPE_CHECKING, Any, Final, Literal
# third party imports
import click
import rich
import yaml
from typing_extensions import NotRequired, ReadOnly, TypedDict
# local imports
from ... import Client
from ._cli_context import cli_context_values
if TYPE_CHECKING:
from rich.console import JustifyMethod
class _ModelInfoColumnConfig(TypedDict):
"""Rendering config for one column of the ``models info`` table."""
header: ReadOnly[str]
style: ReadOnly[str]
justify: NotRequired[ReadOnly["JustifyMethod"]]
get_value: ReadOnly[Callable[..., str]]
@dataclass
@ -84,7 +99,8 @@ def format_cost_per_1k_tokens(cost: float | None) -> str:
def create_client(ctx: click.Context) -> Client:
"""Helper function to create a client from context."""
return Client(base_url=ctx.obj["base_url"], api_key=ctx.obj["api_key"])
context: Final = cli_context_values(ctx)
return Client(base_url=context["base_url"], api_key=context["api_key"])
@click.group()
@ -216,7 +232,7 @@ def get_models_info(ctx: click.Context, output_format: Literal["table", "json"],
table: Final = rich.table.Table(title="Models Information")
# Define all possible columns with their configurations
column_configs: Final[dict[str, dict[str, Any]]] = {
column_configs: Final[dict[str, _ModelInfoColumnConfig]] = {
"public_model": {
"header": "Public Model",
"style": "cyan",

13
litellm/proxy/client/cli/commands/users.py
View file

@ -4,6 +4,7 @@ import click
import rich
from ... import UsersManagementClient
from ._cli_context import cli_context_values
@click.group()
@ -15,7 +16,8 @@ def users():
@click.pass_context
def list_users(ctx: click.Context):
"""List all users"""
client: Final = UsersManagementClient(base_url=ctx.obj["base_url"], api_key=ctx.obj["api_key"])
context: Final = cli_context_values(ctx)
client: Final = UsersManagementClient(base_url=context["base_url"], api_key=context["api_key"])
users = client.list_users()
if isinstance(users, dict) and "users" in users:
users = users["users"]
@ -46,7 +48,8 @@ def list_users(ctx: click.Context):
@click.pass_context
def get_user(ctx: click.Context, user_id: str):
"""Get information about a specific user"""
client: Final = UsersManagementClient(base_url=ctx.obj["base_url"], api_key=ctx.obj["api_key"])
context: Final = cli_context_values(ctx)
client: Final = UsersManagementClient(base_url=context["base_url"], api_key=context["api_key"])
result: Final = client.get_user(user_id=user_id)
rich.print_json(data=result)
@ -60,7 +63,8 @@ def get_user(ctx: click.Context, user_id: str):
@click.pass_context
def create_user(ctx: click.Context, email, role, alias, team, max_budget):
"""Create a new user"""
client: Final = UsersManagementClient(base_url=ctx.obj["base_url"], api_key=ctx.obj["api_key"])
context: Final = cli_context_values(ctx)
client: Final = UsersManagementClient(base_url=context["base_url"], api_key=context["api_key"])
user_data: Final = {
"user_email": email,
"user_role": role,
@ -80,6 +84,7 @@ def create_user(ctx: click.Context, email, role, alias, team, max_budget):
@click.pass_context
def delete_user(ctx: click.Context, user_ids):
"""Delete one or more users by user_id"""
client: Final = UsersManagementClient(base_url=ctx.obj["base_url"], api_key=ctx.obj["api_key"])
context: Final = cli_context_values(ctx)
client: Final = UsersManagementClient(base_url=context["base_url"], api_key=context["api_key"])
result: Final = client.delete_user(list(user_ids))
rich.print_json(data=result)

5
litellm/proxy/client/http_client.py
View file

@ -1,5 +1,6 @@
"""HTTP client for making requests to the LiteLLM proxy server."""
from collections.abc import Mapping
from typing import Any, Final
import requests
@ -25,8 +26,8 @@ class HTTPClient:
method: str,
uri: str,
*,
data: dict[str, Any] | list | bytes | None = None,
json: dict[str, Any] | list | None = None,
data: Mapping[str, object] | list | bytes | None = None,
json: Mapping[str, object] | list | None = None,
headers: dict[str, str] | None = None,
**kwargs: Any,
) -> Any:

9
litellm/proxy/client/models.py
View file

@ -1,4 +1,5 @@
import builtins
from collections.abc import Mapping
from typing import Any, Final
import requests
@ -68,8 +69,8 @@ class ModelsManagementClient:
def new(
self,
model_name: str,
model_params: dict[str, Any],
model_info: dict[str, Any] | None = None,
model_params: Mapping[str, object],
model_info: Mapping[str, object] | None = None,
return_request: bool = False,
) -> dict[str, Any] | requests.Request:
"""
@ -245,8 +246,8 @@ class ModelsManagementClient:
def update(
self,
model_id: str,
model_params: dict[str, Any],
model_info: dict[str, Any] | None = None,
model_params: Mapping[str, object],
model_info: Mapping[str, object] | None = None,
return_request: bool = False,
) -> dict[str, Any] | requests.Request:
"""

6
litellm/proxy/container_endpoints/handler_factory.py
View file

@ -207,7 +207,7 @@ async def _process_binary_request(
processor: Final = ProxyBaseLLMRequestProcessing(data=data)
try:
content: Final = await processor.base_process_llm_request(
content: Final[object] = await processor.base_process_llm_request(
request=request,
fastapi_response=fastapi_response,
user_api_key_dict=user_api_key_dict,
@ -268,7 +268,7 @@ async def _process_multipart_upload_request(
user_api_key_dict: UserAPIKeyAuth,
route_type: str,
container_id: str,
):
) -> object:
"""Process multipart file upload requests."""
from litellm.proxy.common_utils.http_parsing_utils import (
convert_upload_files_to_file_data,
@ -357,7 +357,7 @@ async def _process_request(
user_api_key_dict: UserAPIKeyAuth,
route_type: str,
path_params: dict[str, str],
):
) -> object:
"""Common request processing logic."""
from litellm.proxy.proxy_server import (
general_settings,

8
litellm/proxy/guardrails/guardrail_hooks/cato_networks/cato_networks.py
View file

@ -299,7 +299,7 @@ class CatoNetworksGuardrail(CustomGuardrail):
return data
if action_type == "monitor_action":
verbose_proxy_logger.info("Cato: monitor action")
elif action_type == "block_action":
elif action_type == "block_action" and required_action is not None:
self._handle_block_action(res.get("analysis_result", {}), required_action)
elif action_type == "anonymize_action":
return self._anonymize_request(res, data)
@ -310,7 +310,7 @@ class CatoNetworksGuardrail(CustomGuardrail):
def _handle_block_action(
self,
analysis_result: _CatoAnalysisResult,
required_action: Any,
required_action: _CatoRequiredAction,
) -> None:
detection_message: Final = required_action.get("detection_message", None)
verbose_proxy_logger.info(
@ -410,7 +410,7 @@ class CatoNetworksGuardrail(CustomGuardrail):
res: Final[_CatoAnalyzeResponse] = response.json()
required_action: Final = res.get("required_action")
action_type: Final = required_action and required_action.get("action_type", None)
if action_type and action_type == "block_action":
if action_type == "block_action" and required_action is not None:
self._handle_block_action_on_output(res.get("analysis_result", {}), required_action)
redacted_chat: Final = res.get("redacted_chat", None)
@ -425,7 +425,7 @@ class CatoNetworksGuardrail(CustomGuardrail):
def _handle_block_action_on_output(
self,
analysis_result: _CatoAnalysisResult,
required_action: Any,
required_action: _CatoRequiredAction,
) -> None:
detection_message: Final = required_action.get("detection_message", None)
verbose_proxy_logger.info(

8
litellm/proxy/guardrails/guardrail_hooks/dynamoai/dynamoai.py
View file

@ -6,7 +6,7 @@
# +-------------------------------------------------------------+
import os
from collections.abc import AsyncGenerator
from collections.abc import AsyncGenerator, AsyncIterable
from datetime import datetime
from typing import Any, Final
@ -188,7 +188,7 @@ class DynamoAIGuardrails(CustomGuardrail):
applied_policies: Final = response.get("appliedPolicies", [])
violations_detected: Final[list[str]] = []
violation_details: Final[dict[str, Any]] = {}
violation_details: Final[dict[str, object]] = {}
# For now, only handle BLOCK action
if final_action == "BLOCK":
@ -404,7 +404,7 @@ class DynamoAIGuardrails(CustomGuardrail):
# to avoid sending empty content to DynamoAI (e.g., during tool calls)
if isinstance(response, litellm.ModelResponse):
has_text_content = False
dynamoai_messages: Final[list[dict[str, Any]]] = []
dynamoai_messages: Final[list[dict[str, str]]] = []
for choice in response.choices:
if isinstance(choice, litellm.Choices):
@ -446,7 +446,7 @@ class DynamoAIGuardrails(CustomGuardrail):
async def async_post_call_streaming_iterator_hook(
self,
user_api_key_dict: UserAPIKeyAuth,
response: Any,
response: AsyncIterable[ModelResponseStream],
request_data: dict,
) -> AsyncGenerator[ModelResponseStream, None]:
"""

7
litellm/proxy/guardrails/guardrail_hooks/singulr/singulr.py
View file

@ -4,6 +4,7 @@ from urllib.parse import urlparse
import httpx
import pydantic
from typing_extensions import TypedDict, Unpack
from litellm._logging import verbose_proxy_logger
from litellm.exceptions import GuardrailRaisedException
@ -34,6 +35,10 @@ _GUARD_ENDPOINT: Final = "/api/v1/ai-gateway/litellm"
_DEFAULT_TIMEOUT: Final = 30.0
class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
"""Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
class SingulrGuardrail(CustomGuardrail):
def __init__(
self,
@ -43,7 +48,7 @@ class SingulrGuardrail(CustomGuardrail):
singulr_guardrail_id: str | None = None,
block_on_error: bool | None = None,
timeout: float | None = None,
**kwargs: Any,
**kwargs: Unpack[_CustomGuardrailOptions],
) -> None:
self.singulr_api_key = singulr_api_key or os.environ.get("SINGULR_API_KEY")
self.singulr_api_base = (singulr_api_base or os.environ.get("SINGULR_API_BASE") or _DEFAULT_API_BASE).rstrip(

9
litellm/proxy/guardrails/guardrail_hooks/tool_policy/tool_policy_guardrail.py
View file

@ -20,9 +20,10 @@ Configuration in proxy config YAML:
mode: post_call
"""
from typing import TYPE_CHECKING, Any, Final, Literal, Optional
from typing import TYPE_CHECKING, Final, Literal, Optional
from fastapi import HTTPException
from typing_extensions import TypedDict, Unpack
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import (
@ -39,6 +40,10 @@ if TYPE_CHECKING:
GUARDRAIL_NAME: Final = "tool_policy"
class _CustomGuardrailOptions(TypedDict, total=False, extra_items=object):
"""Base-class constructor options this guardrail forwards untouched to CustomGuardrail."""
def _get_request_object_permission_ids(
request_data: dict,
) -> tuple[str | None, str | None]:
@ -106,7 +111,7 @@ class ToolPolicyGuardrail(CustomGuardrail):
ToolPolicyRegistry (synced from DB).
"""
def __init__(self, **kwargs: Any) -> None:
def __init__(self, **kwargs: Unpack[_CustomGuardrailOptions]) -> None:
if "supported_event_hooks" not in kwargs:
kwargs["supported_event_hooks"] = [
GuardrailEventHooks.pre_call,

13
litellm/proxy/guardrails/usage_endpoints.py
View file

@ -355,6 +355,11 @@ def _to_dict(value: object) -> dict[str, Any]:
return {}
def _field_str(mapping: Mapping[str, object], key: str, default: str) -> str:
"""Stringify `mapping[key]`, falling back to `default` when the key is absent."""
return str(mapping.get(key, default))
def _get_guardrail_attrs(g: "_DbOrConfigGuardrail") -> tuple[Any, str]:
"""Get (guardrail_id, display_name) from guardrail - handles Prisma model or dict."""
gid: Final = _get_guardrail_field(g, "guardrail_id")
@ -383,9 +388,9 @@ def _guardrail_overview_rows(
req, blocked = a["requests"], a["blocked"]
fail_rate = (100.0 * blocked / req) if req else 0.0
litellm_params = _to_dict(_get_guardrail_field(g, "litellm_params"))
provider = str(litellm_params.get("guardrail", "Unknown"))
provider = _field_str(litellm_params, "guardrail", "Unknown")
guardrail_info = _to_dict(_get_guardrail_field(g, "guardrail_info"))
gtype = str(guardrail_info.get("type", "Guardrail"))
gtype = _field_str(guardrail_info, "type", "Guardrail")
prev_fail = 0.0
for k in lookup_keys:
if k in prev_agg:
@ -624,8 +629,8 @@ async def guardrails_usage_detail(
return UsageDetailResponse(
guardrail_id=guardrail_id,
guardrail_name=_guardrail_name or guardrail_id,
type=str(guardrail_info.get("type", "Guardrail")),
provider=str(litellm_params.get("guardrail", "Unknown")),
type=_field_str(guardrail_info, "type", "Guardrail"),
provider=_field_str(litellm_params, "guardrail", "Unknown"),
requestsEvaluated=requests,
failRate=round(fail_rate, 1),
avgScore=None,

36
litellm/proxy/management_endpoints/usage_endpoints/ai_usage_chat.py
View file

@ -6,7 +6,7 @@ usage/spend data by querying the aggregated daily activity endpoints.
import json
from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence
from datetime import date
from typing import Any, Final, Literal, Protocol, cast, overload
from typing import Any, Final, Literal, NamedTuple, Protocol, cast, overload
from typing_extensions import ReadOnly, TypedDict
@ -82,6 +82,15 @@ class _DayDump(TypedDict, total=False):
breakdown: ReadOnly[Mapping[str, Mapping[str, _EntityEntry]]]
class _EntityTotal(NamedTuple):
"""Running per-entity totals accumulated while summarising a usage dump."""
alias: str
spend: float
requests: float
tokens: float
class _UsageDump(Protocol):
@overload
def get(self, key: Literal["metadata"], default: Mapping[str, float], /) -> Mapping[str, float]: ...
@ -241,7 +250,7 @@ def _parse_csv_ids(raw: str | None) -> list[str] | None:
async def _query_activity(
table_name: str,
entity_id_field: str,
entity_id: Any | None,
entity_id: str | list[str] | None,
start_date: str,
end_date: str,
*,
@ -382,23 +391,22 @@ def _summarise_entity_data(data: _UsageDump, entity_label: str) -> str:
if not results:
return f"No {entity_label} usage data found for the given date range."
totals: Final[dict[str, dict[str, Any]]] = {}
totals: Final[dict[str, _EntityTotal]] = {}
for day in results:
for eid, entry in day.get("breakdown", {}).get("entities", {}).items():
if eid not in totals:
alias = entry.get("metadata", {}).get("alias", eid)
totals[eid] = {"alias": alias, "spend": 0.0, "requests": 0, "tokens": 0}
previous = totals.get(eid)
m = entry.get("metrics", {})
totals[eid]["spend"] += m.get("spend", 0)
totals[eid]["requests"] += m.get("api_requests", 0)
totals[eid]["tokens"] += m.get("total_tokens", 0)
totals[eid] = _EntityTotal(
alias=previous.alias if previous is not None else entry.get("metadata", {}).get("alias", eid),
spend=(previous.spend if previous is not None else 0.0) + m.get("spend", 0),
requests=(previous.requests if previous is not None else 0) + m.get("api_requests", 0),
tokens=(previous.tokens if previous is not None else 0) + m.get("total_tokens", 0),
)
lines: Final = [f"{entity_label} Usage ({len(totals)} {entity_label.lower()}s):", ""]
for eid, d in sorted(totals.items(), key=lambda x: -x[1]["spend"]):
label = d["alias"] if d["alias"] != eid else eid
lines.append(
f"- {label} (ID: {eid}): ${d['spend']:.4f} | {int(d['requests'])} reqs | {int(d['tokens'])} tokens"
)
for eid, d in sorted(totals.items(), key=lambda x: -x[1].spend):
label = d.alias if d.alias != eid else eid
lines.append(f"- {label} (ID: {eid}): ${d.spend:.4f} | {int(d.requests)} reqs | {int(d.tokens)} tokens")
return "\n".join(lines)

8
litellm/realtime_api/main.py
View file

@ -42,6 +42,8 @@ from ..llms.xai.realtime.handler import XAIRealtime
from ..utils import client as wrapper_client
if TYPE_CHECKING:
from fastapi import WebSocket
from litellm.llms.base_llm.realtime.http_transformation import BaseRealtimeHTTPConfig
azure_realtime: Final = AzureOpenAIRealtime()
@ -332,12 +334,12 @@ async def _resolve_vertex_access_token_bounded(
@wrapper_client
async def _arealtime(
model: str,
websocket: Any, # fastapi websocket
websocket: "WebSocket", # fastapi websocket
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,
query_params: RealtimeQueryParams | None = None,
**kwargs,
@ -574,7 +576,7 @@ _TRANSCRIPTION_QUERY_PARAMS: Final[RealtimeQueryParams] = {"intent": "transcript
def _azure_realtime_health_protocol(
model: str, realtime_protocol: str | None, model_params: Mapping[str, Any]
model: str, realtime_protocol: str | None, model_params: Mapping[str, object]
) -> tuple[str, RealtimeQueryParams | None]:
query_params: Final = _TRANSCRIPTION_QUERY_PARAMS if _is_transcription_only_realtime_model(model, "azure") else None
configured_raw: Final = (

7
litellm/responses/utils.py
View file

@ -1,7 +1,7 @@
import base64
import re
from collections.abc import Iterable, Mapping, Sequence
from typing import Any, Final, Optional, Union, cast, get_type_hints, overload
from typing import Any, Final, Optional, TypeVar, Union, cast, get_type_hints, overload
from pydantic import BaseModel
from typing_extensions import TypeIs # noqa: TID251 # narrows untyped wire payloads without a runtime conversion
@ -59,6 +59,9 @@ def _as_input_text_part(part: object) -> object:
return part
_RequestInputT: Final = TypeVar("_RequestInputT")
class ResponsesAPIRequestUtils:
"""Helper utils for constructing ResponseAPI requests"""
@ -502,7 +505,7 @@ class ResponsesAPIRequestUtils:
return response
@staticmethod
def _restore_encrypted_content_item_ids_in_input(request_input: object) -> Any:
def _restore_encrypted_content_item_ids_in_input(request_input: _RequestInputT) -> _RequestInputT:
"""Decode litellm-encoded item IDs in request input back to original IDs.
Called before forwarding the request to the upstream provider so the

13
litellm/router_strategy/adaptive_router/signals.py
View file

@ -13,6 +13,7 @@ bounded list of recent tool call signatures.
from __future__ import annotations
import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from typing import Any, Final
@ -92,7 +93,7 @@ class Turn:
user_content: str | None = None
assistant_content: str | None = None
tool_calls: list[dict[str, Any]] = field(default_factory=list)
tool_results: list[dict[str, Any]] = field(default_factory=list)
tool_results: Sequence[Mapping[str, object]] = field(default_factory=list[Mapping[str, object]])
response_status: int | None = None
@ -104,7 +105,7 @@ _TOKEN_RE: Final = re.compile(r"[A-Za-z0-9]+")
def _tokens(text: str | None) -> set[str]:
if not text:
return set()
return {t.lower() for t in _TOKEN_RE.findall(text)}
return {match.group(0).lower() for match in _TOKEN_RE.finditer(text)}
def _jaccard(a: set[str], b: set[str]) -> float:
@ -160,7 +161,7 @@ def _detect_satisfaction(curr_user: str | None) -> bool:
return any(p.search(curr_user) for p in _SATISFACTION_PATTERNS)
def _detect_failure(tool_results: list[dict[str, Any]]) -> bool:
def _detect_failure(tool_results: Sequence[Mapping[str, object]]) -> bool:
"""Any tool result explicitly flagged as an error.
We do NOT treat empty content as failure — many tools legitimately return
@ -209,7 +210,7 @@ _EXHAUSTION_KEYWORDS: Final = (
)
def _detect_exhaustion(status: int | None, tool_results: list[dict[str, Any]]) -> bool:
def _detect_exhaustion(status: int | None, tool_results: Sequence[Mapping[str, object]]) -> bool:
if status is not None and status in _EXHAUSTION_STATUSES:
return True
for r in tool_results:
@ -222,7 +223,7 @@ def _detect_exhaustion(status: int | None, tool_results: list[dict[str, Any]]) -
def detect_user_feedback(
previous_user_content: str | None,
current_user_content: str | None,
tool_results: list[dict[str, Any]],
tool_results: Sequence[Mapping[str, object]],
allow_satisfaction: bool,
) -> SignalDelta:
return SignalDelta(
@ -238,7 +239,7 @@ def detect_response_signals(
current_assistant_content: str | None,
tool_call_history: list[str],
tool_calls: list[dict[str, Any]],
tool_results: list[dict[str, Any]],
tool_results: Sequence[Mapping[str, object]],
response_status: int | None,
) -> SignalDelta:
return SignalDelta(

8
litellm/router_utils/cooldown_handlers.py
View file

@ -259,7 +259,7 @@ def _should_run_cooldown_logic(
litellm_router_instance: LitellmRouter,
deployment: str | None,
exception_status: str | int,
original_exception: Any,
original_exception: Exception,
time_to_cooldown: float | None = None,
) -> bool:
"""
@ -318,7 +318,7 @@ def _should_cooldown_deployment(
litellm_router_instance: LitellmRouter,
deployment: str,
exception_status: str | int,
original_exception: Any,
original_exception: Exception,
requested_model_group: str | None = None,
) -> bool:
"""
@ -412,7 +412,7 @@ def _should_cooldown_deployment(
def _set_cooldown_deployments(
litellm_router_instance: LitellmRouter,
original_exception: Any,
original_exception: Exception,
exception_status: str | int,
deployment: str | None = None,
time_to_cooldown: float | None = None,
@ -547,7 +547,7 @@ def _get_cooldown_deployments(litellm_router_instance: LitellmRouter, parent_ote
def should_cooldown_based_on_allowed_fails_policy(
litellm_router_instance: LitellmRouter,
deployment: str,
original_exception: Any,
original_exception: Exception,
allowed_fails_override: int | None = None,
cooldown_time_override: float | None = None,
cache_key_suffix: str | None = None,

16
litellm/skills/main.py
View file

@ -72,7 +72,7 @@ def _get_litellm_skills_handler():
async def acreate_skill(
files: list[Any] | None = None,
display_title: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: Mapping[str, object] | None = None,
extra_body: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
@ -135,7 +135,7 @@ async def acreate_skill(
def create_skill(
files: list[Any] | None = None,
display_title: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: Mapping[str, object] | None = None,
extra_body: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
@ -262,7 +262,7 @@ async def alist_skills(
limit: int | None = None,
page: str | None = None,
source: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -325,7 +325,7 @@ def list_skills(
limit: int | None = None,
page: str | None = None,
source: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -443,7 +443,7 @@ def list_skills(
@client
async def aget_skill(
skill_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -500,7 +500,7 @@ async def aget_skill(
@client
def get_skill(
skill_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -607,7 +607,7 @@ def get_skill(
@client
async def adelete_skill(
skill_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -664,7 +664,7 @@ async def adelete_skill(
@client
def delete_skill(
skill_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: Mapping[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,

34
litellm/vector_store_files/main.py
View file

@ -2,7 +2,7 @@
import asyncio
import contextvars
from collections.abc import Coroutine
from collections.abc import Coroutine, Mapping
from functools import partial
from typing import Any, Final
@ -57,9 +57,9 @@ async def acreate(
vector_store_id: str,
file_id: str,
attributes: VectorStoreFileAttributes | None = None,
chunking_strategy: dict[str, Any] | None = None,
extra_headers: dict[str, Any] | None = None,
extra_query: dict[str, Any] | None = None,
chunking_strategy: Mapping[str, object] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: Mapping[str, object] | None = None,
extra_body: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -109,9 +109,9 @@ def create(
vector_store_id: str,
file_id: str,
attributes: VectorStoreFileAttributes | None = None,
chunking_strategy: dict[str, Any] | None = None,
extra_headers: dict[str, Any] | None = None,
extra_query: dict[str, Any] | None = None,
chunking_strategy: Mapping[str, object] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: Mapping[str, object] | None = None,
extra_body: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -187,7 +187,7 @@ async def alist(
filter: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -240,7 +240,7 @@ def list(
filter: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_query: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -308,7 +308,7 @@ async def aretrieve(
*,
vector_store_id: str,
file_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -351,7 +351,7 @@ def retrieve(
*,
vector_store_id: str,
file_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -417,7 +417,7 @@ async def aretrieve_content(
*,
vector_store_id: str,
file_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -459,7 +459,7 @@ def retrieve_content(
*,
vector_store_id: str,
file_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -526,7 +526,7 @@ async def aupdate(
vector_store_id: str,
file_id: str,
attributes: VectorStoreFileAttributes,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -572,7 +572,7 @@ def update(
vector_store_id: str,
file_id: str,
attributes: VectorStoreFileAttributes,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
extra_body: dict[str, Any] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
@ -646,7 +646,7 @@ async def adelete(
*,
vector_store_id: str,
file_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,
@ -688,7 +688,7 @@ def delete(
*,
vector_store_id: str,
file_id: str,
extra_headers: dict[str, Any] | None = None,
extra_headers: dict[str, str] | None = None,
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
**kwargs,