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

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

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-09-20 10:16:48 +00:00
parent 6ef7b86748
commit ccd8997b08
34 changed files with 98 additions and 90 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/a2a_protocol/providers/bedrock_agentcore/transformation.py
View file

@ -85,7 +85,7 @@ def _filter_reserved_headers(
def _request_scoped_runtime_session_id(
params: Mapping[str, Any],
params: Mapping[str, object],
litellm_params: Mapping[str, Any],
) -> str | None:
context_id: Final = get_session_id_from_a2a_params(params)

4
litellm/a2a_protocol/providers/watsonx_orchestrate/config.py
View file

@ -20,7 +20,7 @@ class WatsonxOrchestrateA2AConfig(BaseA2AProviderConfig):
params: dict[str, Any],
api_base: str | None = None,
**kwargs: Any,
) -> dict[str, Any]:
) -> dict[str, object]:
"""Handle a non-streaming A2A request via WXO runs API."""
litellm_params: Final = kwargs.get("litellm_params")
if not litellm_params:
@ -40,7 +40,7 @@ class WatsonxOrchestrateA2AConfig(BaseA2AProviderConfig):
params: dict[str, Any],
api_base: str | None = None,
**kwargs: Any,
) -> AsyncIterator[dict[str, Any]]:
) -> AsyncIterator[dict[str, object]]:
"""Handle a streaming A2A request via WXO streaming runs API."""
litellm_params: Final = kwargs.get("litellm_params")
if not litellm_params:

6
litellm/a2a_protocol/utils.py
View file

@ -17,7 +17,7 @@ class A2ARequestUtils:
"""Utility class for A2A request/response processing."""
@staticmethod
def extract_text_from_message(message: Any) -> str:
def extract_text_from_message(message: object) -> str:
"""
Extract text content from A2A message parts.
@ -142,7 +142,7 @@ class A2ARequestUtils:
return prompt_tokens, completion_tokens, total_tokens
def get_session_id_from_a2a_params(params: Mapping[str, Any]) -> str | None:
def get_session_id_from_a2a_params(params: Mapping[str, object]) -> str | None:
message: Final = params.get("message", {})
if isinstance(message, dict):
return message.get("contextId")
@ -166,7 +166,7 @@ def scope_session_to_principal(session_id: str, principal: str | None) -> str:
# Backwards compatibility aliases
def extract_text_from_a2a_message(message: Any) -> str:
def extract_text_from_a2a_message(message: object) -> str:
return A2ARequestUtils.extract_text_from_message(message)

4
litellm/integrations/gitlab/gitlab_prompt_manager.py
View file

@ -200,8 +200,8 @@ class GitLabTemplateManager:
metadata=metadata,
)
def _parse_yaml_basic(self, yaml_str: str) -> dict[str, Any]:
result: Final[dict[str, Any]] = {}
def _parse_yaml_basic(self, yaml_str: str) -> dict[str, bool | int | float | str]:
result: Final[dict[str, bool | int | float | str]] = {}
for line in yaml_str.split("\n"):
line = line.strip()
if ":" in line and not line.startswith("#"):

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

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

4
litellm/integrations/vantage/vantage_logger.py
View file

@ -59,7 +59,7 @@ class VantageLogger(FocusLogger):
raw_interval,
)
destination_config: Final[dict[str, Any]] = {}
destination_config: Final[dict[str, str]] = {}
if resolved_api_key:
destination_config["api_key"] = resolved_api_key
if resolved_token:
@ -93,7 +93,7 @@ class VantageLogger(FocusLogger):
pod_lock_manager = None
if proxy_logging_obj is not None:
writer: Final = getattr(proxy_logging_obj, "db_spend_update_writer", None)
writer: Final[object] = getattr(proxy_logging_obj, "db_spend_update_writer", None)
if writer is not None:
pod_lock_manager = getattr(writer, "pod_lock_manager", None)

22
litellm/interactions/agents/http_handler.py
View file

@ -7,7 +7,7 @@ duplicated. BaseAgentsAPIConfig stays as pure transform code.
"""
from collections.abc import Coroutine, Mapping
from typing import Any, Final
from typing import Final
import httpx
@ -38,7 +38,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: 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,
@ -93,7 +93,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
extra_headers: 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,
@ -141,7 +141,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
agents_api_config: BaseAgentsAPIConfig,
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,
@ -181,7 +181,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
agents_api_config: BaseAgentsAPIConfig,
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,
) -> AgentListResponse:
@ -216,7 +216,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: 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,
@ -259,7 +259,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: 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,
) -> AgentCreateResponse:
@ -295,7 +295,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: 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,
@ -338,7 +338,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: 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,
) -> AgentDeleteResult:
@ -374,7 +374,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: 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,
@ -417,7 +417,7 @@ class AgentsHTTPHandler(InteractionsHTTPHandler):
name: 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,
) -> AgentVersionsResponse:

4
litellm/litellm_core_utils/logging_utils.py
View file

@ -67,7 +67,9 @@ def _truncate_base64_in_string(value: str) -> str:
return _DATA_URI_RE.sub(_base64_data_uri_replacer, value)
def _truncate_base64_in_value(value: Any) -> Any:
def _truncate_base64_in_value(
value: str | dict[str, object] | list[object] | None,
) -> str | dict[str, object] | list[object] | None:
"""Iteratively truncate base64 data URIs in a JSON-like value (str/list/dict).
Uses an explicit stack instead of recursion to satisfy the project's

4
litellm/litellm_core_utils/url_utils.py
View file

@ -418,7 +418,7 @@ def _extract_redirect_url(response: httpx.Response, request_url: str) -> str:
return str(httpx.URL(request_url).join(location))
def safe_get(client: Any, url: str, **kwargs: Any) -> httpx.Response:
def safe_get(client: _UrlFetcher, url: str, **kwargs: Any) -> httpx.Response:
"""
Fetch a user-supplied URL with SSRF protection on every redirect hop.
@ -461,7 +461,7 @@ def safe_get(client: Any, url: str, **kwargs: Any) -> httpx.Response:
raise SSRFError("Too many redirects")
async def async_safe_get(client: Any, url: str, **kwargs: Any) -> httpx.Response:
async def async_safe_get(client: _AsyncUrlFetcher, url: str, **kwargs: Any) -> httpx.Response:
"""Async version of safe_get."""
if not getattr(litellm, "user_url_validation", True):
kwargs.setdefault("follow_redirects", True)

2
litellm/llms/anthropic/chat/handler.py
View file

@ -596,7 +596,7 @@ class ModelResponseIterator:
self.reasoning_content_chunks: list[str] = []
# Track server tool use inputs and results for code_interpreter_results
self._server_tool_inputs: dict[str, Any] = {}
self._server_tool_inputs: dict[str, object] = {}
self.tool_results: list[dict[str, Any]] = []
self._current_server_tool_id: str | None = None
self._container_id: str | None = None

9
litellm/llms/azure/completion/handler.py
View file

@ -1,6 +1,7 @@
from collections.abc import Callable
from typing import Any, Final
from typing import Final
import httpx
from openai import AsyncAzureOpenAI, AzureOpenAI
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -191,7 +192,7 @@ class AzureTextCompletion(BaseAzureLLM):
model: str,
api_base: str,
data: dict,
timeout: Any,
timeout: float | httpx.Timeout | None,
model_response: ModelResponse,
logging_obj: LiteLLMLoggingObj,
max_retries: int,
@ -253,7 +254,7 @@ class AzureTextCompletion(BaseAzureLLM):
api_version: str,
data: dict,
model: str,
timeout: Any,
timeout: float | httpx.Timeout | None,
azure_ad_token: str | None = None,
client=None,
litellm_params: dict = {},
@ -306,7 +307,7 @@ class AzureTextCompletion(BaseAzureLLM):
api_version: str,
data: dict,
model: str,
timeout: Any,
timeout: float | httpx.Timeout | None,
azure_ad_token: str | None = None,
client=None,
litellm_params: dict = {},

8
litellm/llms/azure_ai/ocr/document_intelligence/transformation.py
View file

@ -12,7 +12,7 @@ import asyncio
import re
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Final
from urllib.parse import quote
import httpx
@ -127,7 +127,7 @@ class AzureDocumentIntelligenceOCRConfig(BaseOCRConfig):
def map_ocr_params(
self,
non_default_params: dict,
non_default_params: Mapping[str, object],
optional_params: dict,
model: str,
) -> dict:
@ -164,7 +164,7 @@ class AzureDocumentIntelligenceOCRConfig(BaseOCRConfig):
raise UnsupportedParamsError(message=f"{e}", model=model, llm_provider="azure_ai") from e
@staticmethod
def _normalize_pages_param(pages: Any) -> str:
def _normalize_pages_param(pages: object) -> str:
"""
Convert a caller-provided `pages` value to Azure DI's query-string
form. Azure expects 1-based page numbers, grammar: `^(\\d+(-\\d+)?)(,\\s*(\\d+(-\\d+)?))*$`.
@ -412,7 +412,7 @@ class AzureDocumentIntelligenceOCRConfig(BaseOCRConfig):
raise ValueError("Document URL is required")
# Build Azure DI request
data: Final[dict[str, Any]] = {}
data: Final[dict[str, str]] = {}
# Check if it's a data URI (base64)
if document_url.startswith("data:"):

2
litellm/llms/base_llm/guardrail_translation/base_translation.py
View file

@ -81,7 +81,7 @@ class BaseTranslation(ABC):
@staticmethod
def transform_user_api_key_dict_to_metadata(
user_api_key_dict: Any | None,
user_api_key_dict: Optional["UserAPIKeyAuth"],
) -> dict[str, object]:
"""
Transform user_api_key_dict to a metadata dict with prefixed keys.

10
litellm/llms/bedrock/batches/transformation.py
View file

@ -2,7 +2,7 @@ import os
import re
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final, Literal, cast
from typing import TYPE_CHECKING, Final, Literal, cast
from httpx import Headers, Response
from pydantic import TypeAdapter, ValidationError
@ -170,7 +170,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
create_batch_data: CreateBatchRequest,
optional_params: dict,
litellm_params: dict,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Transform the batch creation request to Bedrock format.
@ -354,7 +354,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
)
@staticmethod
def _get_openai_compatible_batch_metadata(metadata: Any) -> dict[str, str]:
def _get_openai_compatible_batch_metadata(metadata: object) -> dict[str, str]:
"""
OpenAI Batch metadata only accepts string values.
"""
@ -379,7 +379,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
batch_id: str,
optional_params: dict,
litellm_params: dict,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Transform batch retrieval request for Bedrock.
@ -523,7 +523,7 @@ class BedrockBatchesConfig(BaseAWSLLM, BaseBatchesConfig):
)
# Enrich metadata with useful Bedrock fields
enriched_metadata_raw: Final[dict[str, Any]] = {
enriched_metadata_raw: Final[dict[str, object]] = {
"jobName": response_data.get("jobName"),
"clientRequestToken": response_data.get("clientRequestToken"),
"modelId": response_data.get("modelId"),

2
litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py
View file

@ -110,7 +110,7 @@ class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig):
headers: dict,
) -> dict:
input_prompt: Final = self._convert_messages_to_prompt(messages=messages)
request_data: Final[dict[str, Any]] = {"inputPrompt": input_prompt}
request_data: Final[dict[str, object]] = {"inputPrompt": input_prompt}
media_source: Final = self._build_media_source(optional_params)
if media_source is not None:

6
litellm/llms/bytez/chat/transformation.py
View file

@ -335,10 +335,10 @@ class BytezChatConfig(BaseConfig):
class BytezCustomStreamWrapper(CustomStreamWrapper):
def chunk_creator(self, chunk: Any):
def chunk_creator(self, chunk: object):
try:
model_response: Final = self.model_response_creator()
response_obj: dict[str, Any] = {}
response_obj: dict[str, object] = {}
response_obj = {
"text": chunk,
@ -346,7 +346,7 @@ class BytezCustomStreamWrapper(CustomStreamWrapper):
"finish_reason": "",
}
completion_obj: Final[dict[str, Any]] = {"content": chunk}
completion_obj: Final[dict[str, object]] = {"content": chunk}
return self.return_processed_chunk_logic(
completion_obj=completion_obj,

4
litellm/llms/custom_httpx/aiohttp_handler.py
View file

@ -1,5 +1,5 @@
import ssl
from collections.abc import Callable
from collections.abc import AsyncIterable, Callable, Iterable
from typing import TYPE_CHECKING, Any, Final, cast
import aiohttp
@ -212,7 +212,7 @@ class BaseLLMAIOHTTPHandler:
litellm_params: dict,
stream: bool = False,
files: dict | None = None,
content: Any = None,
content: str | bytes | Iterable[bytes] | AsyncIterable[bytes] | None = None,
params: dict | None = None,
) -> httpx.Response:
max_retry_on_unprocessable_entity_error: Final = provider_config.max_retry_on_unprocessable_entity_error

2
litellm/llms/deprecated_providers/aleph_alpha.py
View file

@ -146,7 +146,7 @@ class AlephAlphaConfig:
setattr(self.__class__, key, value)
@classmethod
def get_config(cls):
def get_config(cls) -> dict[str, object]:
return {
k: v
for k, v in cls.__dict__.items()

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

@ -170,7 +170,7 @@ class LemonadeChatConfig(OpenAILikeChatConfig):
model: str,
api_base: str | None = None,
api_key: str | None = None,
) -> Any:
) -> dict[str, object]:
if model.startswith("lemonade/"):
model = model.split("/", 1)[1]

2
litellm/llms/openai/image_edit/transformation.py
View file

@ -66,7 +66,7 @@ class OpenAIImageEditConfig(BaseImageEditConfig):
def _add_image_to_files(
self,
files_list: list[tuple[str, Any]],
image: Any,
image: object,
field_name: str,
) -> None:
"""Add an image to the files list with appropriate content type"""

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

@ -78,7 +78,7 @@ class RunwayMLTextToSpeechConfig(BaseTextToSpeechConfig):
aspeech: bool,
api_base: str | None,
api_key: str | None,
**kwargs: Any,
**kwargs: object,
) -> Union[
"HttpxBinaryResponseContent",
Coroutine[object, object, "HttpxBinaryResponseContent"],

6
litellm/llms/vertex_ai/files/transformation.py
View file

@ -651,7 +651,7 @@ def _openai_batch_jsonl_entry_to_vertex_embeddings_rows(
def _openai_batch_jsonl_entry_to_vertex_rows(
openai_entry: dict[str, Any],
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, Any]],
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, object]],
) -> tuple[Mapping[str, object], ...]:
"""
Transforms a single OpenAI JSONL batch entry into the Vertex rows it maps to.
@ -774,7 +774,7 @@ class _OpenAIToVertexBatchUploadStream(BaseFileUploadStream):
def __init__(
self,
openai_file_content: FileTypes,
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, Any]],
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, object]],
) -> None:
self._openai_file_content = openai_file_content
self._map_openai_to_vertex_params = map_openai_to_vertex_params
@ -948,7 +948,7 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
def _map_openai_to_vertex_params(
self,
openai_request_body: dict[str, Any],
) -> dict[str, Any]:
) -> dict[str, object]:
"""
wrapper to call VertexGeminiConfig.map_openai_params
"""

4
litellm/llms/vertex_ai/gemini_embeddings/batch_embed_content_handler.py
View file

@ -3,7 +3,7 @@ Google AI Studio /batchEmbedContents Embeddings Endpoint
"""
import json
from typing import TYPE_CHECKING, Any, Final, Literal
from typing import TYPE_CHECKING, Final, Literal
import httpx
@ -210,7 +210,7 @@ class GoogleBatchEmbeddings(VertexLLM):
)
### TRANSFORMATION (sync path) ###
request_data: Any
request_data: VertexAIBatchEmbeddingsRequestBody | dict[str, object]
if use_embed_content:
resolved_files = {}
if api_key:

2
litellm/llms/vertex_ai/vertex_ai_non_gemini.py
View file

@ -64,7 +64,7 @@ def _get_client_from_cache(client_cache_key: str):
return litellm.in_memory_llm_clients_cache.get_cache(client_cache_key)
def _set_client_in_cache(client_cache_key: str, vertex_llm_model: Any):
def _set_client_in_cache(client_cache_key: str, vertex_llm_model: object):
litellm.in_memory_llm_clients_cache.set_cache(
key=client_cache_key,
value=vertex_llm_model,

4
litellm/llms/watsonx/audio_transcription/transformation.py
View file

@ -4,7 +4,7 @@ Translates from OpenAI's `/v1/audio/transcriptions` to IBM WatsonX's `/ml/v1/aud
WatsonX follows the OpenAI spec for audio transcription.
"""
from typing import Any, Final
from typing import Final
from httpx import Response
@ -124,7 +124,7 @@ class IBMWatsonXAudioTranscriptionConfig(IBMWatsonXMixin, OpenAIWhisperAudioTran
}
# Convert TypedDict to regular dict for AudioTranscriptionRequestData
form_data_dict: Final[dict[str, Any]] = dict(form_data)
form_data_dict: Final[dict[str, object]] = dict(form_data)
return AudioTranscriptionRequestData(data=form_data_dict, files=files)

16
litellm/proxy/a2a/agent_card.py
View file

@ -10,7 +10,7 @@ and uses LiteLLM auth.
import re
from collections.abc import Mapping
from copy import deepcopy
from typing import Any, Final, Literal
from typing import Final, Literal
SupportedA2AVersion = Literal["0.3", "1.0"]
@ -44,7 +44,7 @@ def normalize_protocol_version(version: object) -> SupportedA2AVersion | None:
return next((supported for supported in SUPPORTED_A2A_PROTOCOL_VERSIONS if supported == major_minor), None)
def resolve_served_protocol_version(card: Mapping[str, Any] | None) -> str:
def resolve_served_protocol_version(card: Mapping[str, object] | None) -> str:
"""Return the validated protocol version an agent card pins, else the default."""
normalized: Final = normalize_protocol_version(card.get("protocolVersion") if card else None)
return normalized if normalized is not None else LITELLM_A2A_PROTOCOL_VERSION
@ -53,7 +53,7 @@ def resolve_served_protocol_version(card: Mapping[str, Any] | None) -> str:
# Security scheme exposed by the LiteLLM-fronted agent card. Always replaces
# whatever upstream advertised — the client must authenticate to the proxy,
# not the upstream agent.
LITELLM_SECURITY_SCHEMES: Final[dict[str, dict[str, Any]]] = {
LITELLM_SECURITY_SCHEMES: Final[dict[str, dict[str, str]]] = {
"LiteLLMKey": {
"type": "http",
"scheme": "bearer",
@ -112,7 +112,7 @@ _ALLOWED_TOP_LEVEL_KEYS: Final = {
"url",
}
_DEFAULT_SKILLS: Final[list[dict[str, Any]]] = [
_DEFAULT_SKILLS: Final[list[dict[str, str | list[str]]]] = [
{
"id": "chat",
"name": "Chat",
@ -129,7 +129,7 @@ _DEFAULT_MODES: Final[list[str]] = ["text"]
_DEFAULT_AGENT_VERSION: Final = "1.0.0"
def _filter_capabilities(upstream_capabilities: Any) -> dict[str, Any]:
def _filter_capabilities(upstream_capabilities: object) -> dict[str, object]:
"""Return a capabilities dict containing only allowlisted, truthy keys."""
if not isinstance(upstream_capabilities, dict):
return {}
@ -143,13 +143,13 @@ def _default_litellm_provider(proxy_base_url: str) -> dict[str, str]:
def merge_agent_card(
upstream_card: Mapping[str, Any] | None,
upstream_card: Mapping[str, object] | None,
*,
proxy_url: str,
proxy_base_url: str,
name: str | None = None,
description: str | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Build the LiteLLM-fronted agent card.
@ -169,7 +169,7 @@ def merge_agent_card(
A dict suitable for serving as the proxy's agent card. Only keys in
the v1.0 AgentCard schema (plus ``supportedInterfaces``) are emitted.
"""
base: Final[dict[str, Any]] = deepcopy(dict(upstream_card)) if upstream_card else {}
base: Final[dict[str, object]] = deepcopy(dict(upstream_card)) if upstream_card else {}
# Keep the upstream ``url`` on the stored card: the runtime A2A
# invocation path reads it from ``agent_card_params`` to know where to

2
litellm/proxy/agent_endpoints/a2a_endpoints.py
View file

@ -880,7 +880,7 @@ async def invoke_agent_a2a(
logging_obj._enqueue_deferred_logging = None
_enqueue_fn()
response_dict: Final[dict[str, Any]] = (
response_dict: Final[dict[str, object]] = (
response.model_dump(mode="json", exclude_none=True)
if hasattr(response, "model_dump")
else response

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

@ -1,3 +1,4 @@
from collections.abc import Mapping
from typing import Any, Final
import requests
@ -69,8 +70,8 @@ class CredentialsManagementClient:
def create(
self,
credential_name: str,
credential_info: dict[str, Any],
credential_values: dict[str, Any],
credential_info: Mapping[str, object],
credential_values: Mapping[str, object],
return_request: bool = False,
) -> dict[str, Any] | requests.Request:
"""

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

@ -7,7 +7,7 @@
import json
import os
from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict
from typing import TYPE_CHECKING, Final, Literal, TypedDict
from fastapi import HTTPException
@ -181,7 +181,7 @@ class GuardrailsAI(CustomGuardrail):
): # raise exception if invalid, return a str for the user to receive - if rejected, or return a modified dictionary for passing into litellm
return await self.process_input(data=data, call_type=call_type)
async def async_logging_hook(self, kwargs: dict, result: Any, call_type: str) -> tuple[dict, Any]:
async def async_logging_hook(self, kwargs: dict, result: object, call_type: str) -> tuple[dict, object]:
if call_type == "acompletion" or call_type == "completion":
kwargs = await self.process_input(data=kwargs, call_type=call_type)

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

@ -24,7 +24,7 @@ from litellm.llms.custom_httpx.http_handler import (
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.llms.openai import AllMessageValues
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolCallChunk
from litellm.types.proxy.guardrails.guardrail_hooks.base import (
GuardrailConfigModel,
)
@ -36,7 +36,7 @@ from litellm.types.proxy.guardrails.guardrail_hooks.singulr import (
ToolCall,
ToolCallFunction,
)
from litellm.types.utils import CallTypes, GenericGuardrailAPIInputs
from litellm.types.utils import CallTypes, ChatCompletionMessageToolCall, GenericGuardrailAPIInputs
_DEFAULT_API_BASE: Final = "http://localhost:8003"
_GUARD_ENDPOINT: Final = "/api/v1/ai-gateway/litellm-v2"
@ -339,7 +339,7 @@ class SingulrGuardrail(CustomGuardrail):
return inputs
@staticmethod
def _build_tool_call(tool_call: Mapping[str, Any]) -> "ToolCall | None":
def _build_tool_call(tool_call: ChatCompletionToolCallChunk | ChatCompletionMessageToolCall) -> "ToolCall | None":
tool_call_id: Final = tool_call.get("id")
fun: Final = tool_call.get("function")
if not tool_call_id or not fun:

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

@ -8,7 +8,7 @@ from collections.abc import Mapping, Sequence
from collections.abc import Set as AbstractSet
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Optional
from typing import TYPE_CHECKING, Final, Optional
from fastapi import HTTPException, status
from pydantic import TypeAdapter
@ -156,12 +156,13 @@ async def handle_update_object_permission_common(
if prisma_client is None:
raise ValueError("Prisma client not found")
new_object_permission: dict | str | None = data_json.pop("object_permission", None)
if new_object_permission is None:
raw_object_permission: Final[dict | str | None] = data_json.pop("object_permission", None)
if raw_object_permission is None:
return None
if isinstance(new_object_permission, str):
new_object_permission = json.loads(new_object_permission)
new_object_permission: Final[object] = (
json.loads(raw_object_permission) if isinstance(raw_object_permission, str) else raw_object_permission
)
upsert: Final = await prepare_object_permission_upsert(
new_object_permission=new_object_permission if isinstance(new_object_permission, dict) else {},
@ -230,7 +231,7 @@ def _dedupe_preserving_order(values: list[str]) -> list[str]:
return result
def _mcp_server_identifier_matches(server: Any, identifier: str) -> bool:
def _mcp_server_identifier_matches(server: object, identifier: str) -> bool:
return identifier in {
getattr(server, "server_id", None),
getattr(server, "alias", None),

12
litellm/proxy/policy_engine/pipeline_executor.py
View file

@ -8,7 +8,7 @@ pass/fail actions (allow, block, next, modify_response) and data forwarding.
import copy
import time
from collections.abc import Callable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar
from typing import TYPE_CHECKING, Final, Literal, TypeVar
from pydantic import BaseModel
@ -314,11 +314,11 @@ class PipelineExecutor:
steps: list[PipelineStep],
mode: str,
data: dict,
user_api_key_dict: Any,
user_api_key_dict: "UserAPIKeyAuth",
call_type: str,
policy_name: str,
raw_request_snapshot: dict | None = None, # mutable-ok: same request-payload shape as data
streaming_chunks: list[Any] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
streaming_chunks: list[object] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
endpoint_translation: "BaseTranslation | None" = None,
) -> PipelineExecutionResult:
"""
@ -490,10 +490,10 @@ class PipelineExecutor:
step: PipelineStep,
mode: str,
data: dict,
user_api_key_dict: Any,
user_api_key_dict: "UserAPIKeyAuth",
call_type: str,
raw_request_snapshot: dict | None = None, # mutable-ok: same request-payload shape as data
streaming_chunks: list[Any] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
streaming_chunks: list[object] | None = None, # mutable-ok: shared buffered-stream chunks, read per step
endpoint_translation: "BaseTranslation | None" = None,
) -> tuple[
Literal["pass", "fail", "error"],
@ -722,7 +722,7 @@ def _extract_error_message(e: Exception) -> str:
if isinstance(e, ModifyResponseException):
return str(e)
if HTTPException is not None and isinstance(e, HTTPException):
detail: Final = getattr(e, "detail", None)
detail: Final[object] = getattr(e, "detail", None)
if detail:
return str(detail)
return str(e)

2
litellm/router_strategy/adaptive_router/hooks.py
View file

@ -86,7 +86,7 @@ def _resolve_session_key(kwargs: dict[str, Any]) -> str | None:
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def _last_user_content(messages: list[dict[str, Any]] | None) -> str | None:
def _last_user_content(messages: Sequence[Mapping[str, object]] | None) -> str | None:
if not messages:
return None
for msg in reversed(messages):

4
litellm/router_strategy/auto_router/auto_router.py
View file

@ -3,7 +3,7 @@ Auto-Routing Strategy that works with a Semantic Router Config
"""
import asyncio
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, Optional
from pydantic import BaseModel, ConfigDict
@ -158,7 +158,7 @@ class AutoRouter(CustomLogger):
return await asyncio.shield(build_task)
@staticmethod
def _extract_text_from_messages(messages: list[dict[str, Any]]) -> str:
def _extract_text_from_messages(messages: Sequence[Mapping[str, object]]) -> str:
"""
Extract text content from the last user message for routing.