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

Merge pull request #42127 from BerriAI/litellm_any_sweep_20260920

Browse source 
refactor(types): replace Any with proven types in 30 files
This commit is contained in:
Mateo Wang 2026-09-20 05:23:51 -07:00 • committed by GitHub
commit cbcccd531c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
30 changed files with 84 additions and 80 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("#"):

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

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

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:

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.