scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

refactor(types): replace Any with proven types in 137 files (#44478)

Browse source 
* refactor(types): prove runtime types at harness, search, rag and client boundaries

Replace Any with adapter-validated types in the litellm.agent() harness, the
search provider transformations, RAG ingestion and query, the vector store
pre-call hook and registry, the galileo and opik logging integrations and the
proxy client CLI. Each boundary gets unit tests for well-formed and malformed
payloads.

* chore(typing): prove types at more provider boundaries and restore search transformations

Second pass over a2a, embedding, rerank, image, audio and small provider
modules. Search transformations go back to their previous form because
validating their response bodies would change the proxy status for
malformed upstream bodies from 400 to 500.

* refactor(types): prove types at logging, files, rerank, image, audio and management boundaries

Replace Any with validated or annotation-only types in 37 more files: logging
integrations, token counters, provider files/rerank/image generation/audio
transcription transformations, pass-through logging handlers and management
endpoints. No proxy HTTP status or error type changes.

* refactor(types): prove types at repository, spend, files and router boundaries

Replace Any with repository table accessors, validated mappings and
annotation-only types in 33 more files: Prisma repositories, the enterprise
batch and responses cost checkers, budget reservation, files endpoints,
management endpoints, the policy registry, the adaptive and complexity routers
and the secret managers. No proxy HTTP status or error type changes.

* test(types): run the aiohttp transformation test in-process and cover repository row conversion

The aiohttp chat transformation test no longer starts a server. It feeds the
transformation a response whose json() returns the body under test.

The proxy unit shards now exercise stored model rows whose params are JSON
strings and the object permission create and update paths.
This commit is contained in:
Mateo Wang 2026-10-05 18:01:45 +00:00 • committed by GitHub
parent 3b2ed83152
commit d80f8c28ca
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
240 changed files with 6820 additions and 556 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
enterprise/litellm_enterprise/enterprise_callbacks/send_emails/base_email.py
View file

@ -46,6 +46,8 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
WebhookEvent,
)
from litellm.repositories.table_repositories import InvitationLinkRepository
from litellm.repositories.user_repository import UserRepository
from litellm.secret_managers.main import get_secret_bool
from litellm.types.integrations.slack_alerting import LITELLM_LOGO_URL
@ -844,7 +846,7 @@ class BaseEmailLogger(CustomLogger):
)
return None
user_row = await prisma_client.db.litellm_usertable.find_unique(
user_row = await UserRepository(prisma_client).table.find_unique(
where={"user_id": user_id}
)
@ -929,7 +931,7 @@ class BaseEmailLogger(CustomLogger):
try:
# Try to get existing invitation
existing_invitations = (
await prisma_client.db.litellm_invitationlink.find_many(
await InvitationLinkRepository(prisma_client).table.find_many(
where={"user_id": user_id},
order={"created_at": "desc"},
)

18
enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py
View file

@ -15,6 +15,10 @@ from litellm.constants import (
MANAGED_OBJECT_STALENESS_CUTOFF_DAYS,
MAX_OBJECTS_PER_POLL_CYCLE,
)
from litellm.repositories.table_repositories import ManagedObjectRepository
from litellm.repositories.team_repository import TeamRepository
from litellm.repositories.user_repository import UserRepository
from litellm.repositories.verification_token_repository import VerificationTokenRepository
if TYPE_CHECKING:
from prisma import models as prisma_models
@ -58,25 +62,19 @@ class _ManagedObjectRow(Protocol):
def _managed_object_table(prisma_client: "PrismaClient") -> "TableActions[_ManagedObjectRow]":
table: Final[TableActions[_ManagedObjectRow]] = prisma_client.db.litellm_managedobjecttable
return table
return ManagedObjectRepository(prisma_client).table
def _user_table(prisma_client: "PrismaClient") -> "TableActions[prisma_models.LiteLLM_UserTable]":
table: Final[TableActions[prisma_models.LiteLLM_UserTable]] = prisma_client.db.litellm_usertable
return table
return UserRepository(prisma_client).table
def _token_table(prisma_client: "PrismaClient") -> "TableActions[prisma_models.LiteLLM_VerificationToken]":
table: Final[TableActions[prisma_models.LiteLLM_VerificationToken]] = (
prisma_client.db.litellm_verificationtoken
)
return table
return VerificationTokenRepository(prisma_client).table
def _team_table(prisma_client: "PrismaClient") -> "TableActions[prisma_models.LiteLLM_TeamTable]":
table: Final[TableActions[prisma_models.LiteLLM_TeamTable]] = prisma_client.db.litellm_teamtable
return table
return TeamRepository(prisma_client).table
class CheckBatchCost:

4
enterprise/litellm_enterprise/proxy/common_utils/check_responses_cost.py
View file

@ -16,6 +16,7 @@ from litellm.constants import (
MAX_OBJECTS_PER_POLL_CYCLE,
STALE_OBJECT_CLEANUP_BATCH_SIZE,
)
from litellm.repositories.table_repositories import ManagedObjectRepository
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN
@ -43,8 +44,7 @@ class _ManagedObjectRow(Protocol):
def _managed_object_table(prisma_client: "PrismaClient") -> "TableActions[_ManagedObjectRow]":
table: Final[TableActions[_ManagedObjectRow]] = prisma_client.db.litellm_managedobjecttable
return table
return ManagedObjectRepository(prisma_client).table
class CheckResponsesCost:

4
enterprise/litellm_enterprise/proxy/hooks/managed_files.py
View file

@ -1647,7 +1647,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
owner_filter: Final = build_owner_filter(user_api_key_dict)
if owner_filter is None:
return FileListPage(**build_list_page([]))
return FileListPage.model_validate(build_list_page([]))
if after:
cursor_row = await _managed_file_table(self.prisma_client).find_first(
@ -1686,7 +1686,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
cursor_id = chunk[-1].unified_file_id
chunk_size = max(chunk_size, FILE_LIST_CONTINUATION_CHUNK_SIZE)
return FileListPage(**build_list_page(matches[:page_size], has_more=len(matches) > page_size))
return FileListPage.model_validate(build_list_page(matches[:page_size], has_more=len(matches) > page_size))
def _is_batch_polling_enabled(self) -> bool:
"""

9
litellm/a2a_protocol/providers/base.py
View file

@ -4,7 +4,6 @@ Base configuration for A2A protocol providers.
from abc import ABC, abstractmethod
from collections.abc import AsyncIterator
from typing import Any
class BaseA2AProviderConfig(ABC):
@ -19,10 +18,10 @@ class BaseA2AProviderConfig(ABC):
async def handle_non_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Handle non-streaming A2A request.
@ -40,10 +39,10 @@ class BaseA2AProviderConfig(ABC):
async def handle_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs,
) -> AsyncIterator[dict[str, Any]]:
) -> AsyncIterator[dict[str, object]]:
"""
Handle streaming A2A request.

10
litellm/a2a_protocol/providers/bedrock_agentcore/config.py
View file

@ -3,7 +3,7 @@ Bedrock AgentCore A2A provider configuration.
"""
from collections.abc import AsyncIterator
from typing import Any, Final
from typing import Final
from litellm.a2a_protocol.providers.base import BaseA2AProviderConfig
from litellm.a2a_protocol.providers.bedrock_agentcore.handler import (
@ -23,10 +23,10 @@ class BedrockAgentCoreA2AConfig(BaseA2AProviderConfig):
async def handle_non_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs,
) -> dict[str, Any]:
) -> dict[str, object]:
"""Handle non-streaming request to AgentCore A2A agent."""
litellm_params: Final = kwargs.get("litellm_params")
if not litellm_params:
@ -43,10 +43,10 @@ class BedrockAgentCoreA2AConfig(BaseA2AProviderConfig):
async def handle_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs,
) -> AsyncIterator[dict[str, Any]]:
) -> AsyncIterator[dict[str, object]]:
"""Handle streaming request to AgentCore A2A agent."""
litellm_params: Final = kwargs.get("litellm_params")
if not litellm_params:

10
litellm/a2a_protocol/providers/bedrock_agentcore/handler.py
View file

@ -7,7 +7,7 @@ completion bridge that would otherwise strip the envelope.
import json
from collections.abc import AsyncIterator, Mapping
from typing import Any, Final
from typing import Final
from litellm._logging import verbose_logger
from litellm.a2a_protocol.providers.bedrock_agentcore.transformation import (
@ -30,9 +30,9 @@ class BedrockAgentCoreA2AHandler:
async def handle_non_streaming(
request_id: str,
params: Mapping[str, object],
litellm_params: dict[str, Any],
litellm_params: dict[str, object],
agent_extra_headers: dict[str, str] | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Handle non-streaming A2A request to AgentCore.
@ -77,9 +77,9 @@ class BedrockAgentCoreA2AHandler:
async def handle_streaming(
request_id: str,
params: Mapping[str, object],
litellm_params: dict[str, Any],
litellm_params: dict[str, object],
agent_extra_headers: dict[str, str] | None = None,
) -> AsyncIterator[dict[str, Any]]:
) -> AsyncIterator[dict[str, object]]:
"""
Handle streaming A2A request to AgentCore.

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

@ -219,7 +219,7 @@ class BedrockAgentCoreA2ATransformation:
return url, signed_headers, signed_body
@staticmethod
async def parse_sse_events(response: _SSELineSource) -> AsyncIterator[dict[str, Any]]:
async def parse_sse_events(response: _SSELineSource) -> AsyncIterator[dict[str, object]]:
"""
Parse SSE events from an httpx streaming response.

9
litellm/a2a_protocol/providers/langflow/config.py
View file

@ -1,5 +1,4 @@
from collections.abc import AsyncIterator
from typing import Any
from litellm.a2a_protocol.litellm_completion_bridge.handler import (
A2A_USER_API_KEY_HASH_PARAM,
@ -16,10 +15,10 @@ class LangFlowA2AConfig(BaseA2AProviderConfig):
async def handle_non_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs,
) -> dict[str, Any]:
) -> dict[str, object]:
litellm_params = kwargs.get("litellm_params")
if not litellm_params:
raise ValueError(
@ -39,10 +38,10 @@ class LangFlowA2AConfig(BaseA2AProviderConfig):
async def handle_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs,
) -> AsyncIterator[dict[str, Any]]:
) -> AsyncIterator[dict[str, object]]:
litellm_params = kwargs.get("litellm_params")
if not litellm_params:
raise ValueError(

16
litellm/a2a_protocol/providers/pydantic_ai_agents/config.py
View file

@ -2,8 +2,7 @@
Pydantic AI provider configuration.
"""
from collections.abc import AsyncIterator
from typing import Any
from collections.abc import AsyncIterator, Mapping
from litellm.a2a_protocol.providers.base import BaseA2AProviderConfig
from litellm.a2a_protocol.providers.pydantic_ai_agents.handler import PydanticAIHandler
@ -20,9 +19,12 @@ class PydanticAIProviderConfig(BaseA2AProviderConfig):
async def handle_non_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs: Any,
*,
timeout: float = 60.0,
agent_extra_headers: Mapping[str, str] | None = None,
**kwargs: object,
) -> dict[str, object]:
"""Handle non-streaming request to Pydantic AI agent."""
if api_base is None:
@ -31,14 +33,14 @@ class PydanticAIProviderConfig(BaseA2AProviderConfig):
request_id=request_id,
params=params,
api_base=api_base,
timeout=kwargs.get("timeout", 60.0),
agent_extra_headers=kwargs.get("agent_extra_headers"),
timeout=timeout,
agent_extra_headers=agent_extra_headers,
)
async def handle_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs,
) -> AsyncIterator[dict[str, object]]:

16
litellm/a2a_protocol/providers/pydantic_ai_agents/handler.py
View file

@ -5,8 +5,8 @@ Pydantic AI agents follow A2A protocol but don't support streaming natively.
This handler provides fake streaming by converting non-streaming responses into streaming chunks.
"""
from collections.abc import AsyncIterator
from typing import Any, Final
from collections.abc import AsyncIterator, Mapping
from typing import Final
from litellm._logging import verbose_logger
from litellm.a2a_protocol.providers.pydantic_ai_agents.transformation import (
@ -26,11 +26,11 @@ class PydanticAIHandler:
@staticmethod
async def handle_non_streaming(
request_id: str,
params: dict[str, Any],
params: Mapping[str, object],
api_base: str | None = None,
timeout: float = 60.0,
agent_extra_headers: dict[str, str] | None = None,
) -> dict[str, Any]:
agent_extra_headers: Mapping[str, str] | None = None,
) -> dict[str, object]:
"""
Handle non-streaming request to Pydantic AI agent.
@ -63,13 +63,13 @@ class PydanticAIHandler:
@staticmethod
async def handle_streaming(
request_id: str,
params: dict[str, Any],
params: Mapping[str, object],
api_base: str | None = None,
timeout: float = 60.0,
chunk_size: int = 50,
delay_ms: int = 10,
agent_extra_headers: dict[str, str] | None = None,
) -> AsyncIterator[dict[str, Any]]:
agent_extra_headers: Mapping[str, str] | None = None,
) -> AsyncIterator[dict[str, object]]:
"""
Handle streaming request to Pydantic AI agent with fake streaming.

12
litellm/a2a_protocol/providers/pydantic_ai_agents/transformation.py
View file

@ -7,7 +7,7 @@ This module provides fake streaming by converting non-streaming responses into s
import asyncio
from collections.abc import AsyncIterator, Mapping, Sequence
from typing import Any, Final, Protocol, cast, runtime_checkable
from typing import Final, Protocol, runtime_checkable
from uuid import uuid4
from pydantic import TypeAdapter
@ -100,7 +100,7 @@ class PydanticAITransformation:
request_id: str,
max_attempts: int = 30,
poll_interval: float = 0.5,
agent_extra_headers: dict[str, str] | None = None,
agent_extra_headers: Mapping[str, str] | None = None,
) -> dict[str, object]:
"""
Poll for task completion using tasks/get method.
@ -156,7 +156,7 @@ class PydanticAITransformation:
request_id: str,
params: "_SupportsModelDump | _SupportsPydanticDict | Mapping[str, object]",
timeout: float = 60.0,
agent_extra_headers: dict[str, str] | None = None,
agent_extra_headers: Mapping[str, str] | None = None,
) -> dict[str, object]:
"""
Send a request to Pydantic AI agent and return the raw task response.
@ -200,7 +200,7 @@ class PydanticAITransformation:
# Send request to Pydantic AI agent using shared async HTTP client
client: Final = get_async_httpx_client(
llm_provider=cast(Any, "pydantic_ai_agent"),
llm_provider="pydantic_ai_agent",
params={"timeout": timeout},
)
response: Final = await client.post(
@ -242,7 +242,7 @@ class PydanticAITransformation:
request_id: str,
params: "_SupportsModelDump | _SupportsPydanticDict | Mapping[str, object]",
timeout: float = 60.0,
agent_extra_headers: dict[str, str] | None = None,
agent_extra_headers: Mapping[str, str] | None = None,
) -> dict[str, object]:
"""
Send a non-streaming A2A request to Pydantic AI agent and wait for completion.
@ -278,7 +278,7 @@ class PydanticAITransformation:
request_id: str,
params: "_SupportsModelDump | _SupportsPydanticDict | Mapping[str, object]",
timeout: float = 60.0,
agent_extra_headers: dict[str, str] | None = None,
agent_extra_headers: Mapping[str, str] | None = None,
) -> dict[str, object]:
"""
Send a request to Pydantic AI agent and return the raw task response.

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

@ -3,11 +3,11 @@ A2A provider configuration for IBM watsonx Orchestrate (WXO).
"""
from collections.abc import AsyncIterator
from typing import Any, Final
from litellm.a2a_protocol.providers.base import BaseA2AProviderConfig
from litellm.a2a_protocol.providers.watsonx_orchestrate.handler import (
WatsonxOrchestrateHandler,
WXOLitellmParams,
)
@ -17,12 +17,13 @@ class WatsonxOrchestrateA2AConfig(BaseA2AProviderConfig):
async def handle_non_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs: Any,
*,
litellm_params: WXOLitellmParams | None = None,
**kwargs: object,
) -> dict[str, object]:
"""Handle a non-streaming A2A request via WXO runs API."""
litellm_params: Final = kwargs.get("litellm_params")
if not litellm_params:
raise ValueError(
"litellm_params is required for WatsonxOrchestrateA2AConfig "
@ -37,12 +38,13 @@ class WatsonxOrchestrateA2AConfig(BaseA2AProviderConfig):
async def handle_streaming(
self,
request_id: str,
params: dict[str, Any],
params: dict[str, object],
api_base: str | None = None,
**kwargs: Any,
*,
litellm_params: WXOLitellmParams | None = None,
**kwargs: object,
) -> 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:
raise ValueError(
"litellm_params is required for WatsonxOrchestrateA2AConfig "

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

@ -7,7 +7,7 @@ import hashlib
import json
import time
from collections.abc import AsyncIterator
from typing import Any, Final, NamedTuple, Protocol
from typing import Final, NamedTuple, Protocol
import httpx
from typing_extensions import NotRequired, ReadOnly, TypedDict
@ -228,7 +228,7 @@ class WatsonxOrchestrateHandler:
return run_data
@staticmethod
async def _accumulate_wxo_sse_text(response: Any) -> str:
async def _accumulate_wxo_sse_text(response: _SSELineSource) -> str:
source: Final[_WXOView] = {"sse_source": response}
accumulated_text = ""
async for line in source["sse_source"].aiter_lines():

8
litellm/caching/_embedding_router.py
View file

@ -12,7 +12,7 @@ This module is dependency-injected: callers pass the proxy ``llm_router`` and
from __future__ import annotations
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final
import litellm
@ -39,10 +39,10 @@ def resolve_embedding_router(
def build_router_embedding_metadata(
request_metadata: dict[str, Any] | None,
) -> dict[str, Any]:
request_metadata: Mapping[str, object] | None,
) -> Mapping[str, object]:
"""Forward the caller's full metadata, flagged as a semantic-cache embedding."""
metadata: Final[dict[str, Any]] = dict(request_metadata or {})
metadata: Final = dict(request_metadata or {})
metadata["semantic-cache-embedding"] = True
return metadata

2
litellm/files/main.py
View file

@ -331,7 +331,7 @@ async def afile_retrieve(
else:
response = init_response
return OpenAIFileObject(**response.model_dump())
return OpenAIFileObject.model_validate(response.model_dump())
except Exception as e:
raise e

4
litellm/files/streaming.py
View file

@ -34,7 +34,7 @@ class FileContentStreamingResponse:
self.custom_llm_provider = custom_llm_provider
self.logging_obj = logging_obj
self.standard_logging_object: StandardLoggingPayload | None = None
self._hidden_params: dict[str, Any] = {}
self._hidden_params: dict[str, object] = {}
self._logging_completed = False
self._close_completed = False
self._start_time = (
@ -121,7 +121,7 @@ class FileContentStreamingResponse:
return response
def _sync_hidden_params(self) -> None:
litellm_params: dict[str, Any] = {}
litellm_params: dict[str, object] = {}
if self.logging_obj is not None:
litellm_params = self.logging_obj.model_call_details.get("litellm_params", {}) or {}

6
litellm/harness/context.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, TypeAlias
from typing import TYPE_CHECKING, TypeAlias
from pydantic import BaseModel
@ -42,7 +42,7 @@ class SessionContext:
api_base: str | None = None
endpoint: ModelEndpoint | None = None
instructions: str | None = None
tools: Sequence[Callable[..., Any]] = ()
tools: Sequence[Callable[..., object]] = ()
skills: Sequence[str] = ()
disable_tools: Sequence[str] = ()
permissions: PermissionMode = "full"
@ -50,7 +50,7 @@ class SessionContext:
output: type[BaseModel] | None = None
max_turns: int | None = None
timeout: float | None = None
metadata: Mapping[str, Any] = field(default_factory=dict)
metadata: Mapping[str, object] = field(default_factory=dict)
options: HarnessOptions | None = None
# Set by the handler after each turn.
final_text: str = ""

4
litellm/harness/handlers/cli_handler.py
View file

@ -13,7 +13,7 @@ import asyncio
import os
from collections import deque
from collections.abc import AsyncIterator, Sequence
from typing import Any, Final
from typing import Final
from litellm._logging import verbose_logger
from litellm.constants import HARNESS_STDERR_TAIL_LINES, HARNESS_STREAM_READ_CHUNK_BYTES
@ -125,7 +125,7 @@ class CLIHarnessHandler(BaseHarnessHandler):
self._proc = proc
tail: Final[deque[str]] = deque(maxlen=HARNESS_STDERR_TAIL_LINES) # mutable-ok: bounded stderr ring buffer
stderr_task = asyncio.ensure_future(drain_stderr(proc.stderr, tail))
state: Any = self.config.create_stream_state()
state: Final[object] = self.config.create_stream_state()
exit_code: int | None = None
try:
await send_stdin(proc, request.stdin)

8
litellm/harness/options.py
View file

@ -9,7 +9,7 @@ from typing import Any, Literal
@dataclass(frozen=True)
class ClaudeCodeOptions:
config: Mapping[str, Any] = field(default_factory=dict)
config: Mapping[str, object] = field(default_factory=dict)
env: Mapping[str, str] = field(default_factory=dict)
@ -17,20 +17,20 @@ class ClaudeCodeOptions:
class CodexOptions:
reasoning_effort: Literal["low", "medium", "high", "xhigh"] | None = None
web_search: bool = False
config: Mapping[str, Any] = field(default_factory=dict)
config: Mapping[str, object] = field(default_factory=dict)
env: Mapping[str, str] = field(default_factory=dict)
@dataclass(frozen=True)
class OpenCodeOptions:
agent: str = "build"
config: Mapping[str, Any] = field(default_factory=dict)
config: Mapping[str, object] = field(default_factory=dict)
env: Mapping[str, str] = field(default_factory=dict)
@dataclass(frozen=True)
class DeepAgentsOptions:
subagents: Sequence[Any] = ()
subagents: Sequence[object] = ()
recursion_limit: int | None = None

42
litellm/harness/runtime.py
View file

@ -1005,24 +1005,24 @@ def aagent(
`await litellm.aagent(...)` returns a Result. With stream=True it returns an async
iterator of events instead: `async for event in litellm.aagent(..., stream=True)`.
"""
kwargs: dict[str, Any] = { # mutable-ok: forwarded as **kwargs to arun_agent/astream_agent
"sandbox": sandbox,
"model": model,
"api_key": api_key,
"api_base": api_base,
"instructions": instructions,
"tools": tools,
"skills": skills,
"disable_tools": disable_tools,
"permissions": permissions,
"on_approval": on_approval,
"output": output,
"max_turns": max_turns,
"timeout": timeout,
"metadata": metadata,
"options": options,
"install": install,
}
if stream:
return astream_agent(harness, prompt, **kwargs)
return arun_agent(harness, prompt, **kwargs)
call: Final = astream_agent if stream else arun_agent
return call(
harness,
prompt,
sandbox=sandbox,
model=model,
api_key=api_key,
api_base=api_base,
instructions=instructions,
tools=tools,
skills=skills,
disable_tools=disable_tools,
permissions=permissions,
on_approval=on_approval,
output=output,
max_turns=max_turns,
timeout=timeout,
metadata=metadata,
options=options,
install=install,
)

44
litellm/harness/sync.py
View file

@ -8,7 +8,7 @@ import threading
from collections.abc import AsyncIterator, Callable, Coroutine, Mapping, Sequence
from concurrent.futures import Future
from typing import (
Any,
Final,
TypeVar,
)
@ -417,24 +417,24 @@ def agent(
Prefix the model with `litellm_proxy/` to route every model call through your
LiteLLM AI Gateway.
"""
kwargs: dict[str, Any] = { # mutable-ok: forwarded as **kwargs to _run/_stream
"sandbox": sandbox,
"model": model,
"api_key": api_key,
"api_base": api_base,
"instructions": instructions,
"tools": tools,
"skills": skills,
"disable_tools": disable_tools,
"permissions": permissions,
"on_approval": on_approval,
"output": output,
"max_turns": max_turns,
"timeout": timeout,
"metadata": metadata,
"options": options,
"install": install,
}
if stream:
return _stream(harness, prompt, **kwargs)
return _run(harness, prompt, **kwargs)
call: Final = _stream if stream else _run
return call(
harness,
prompt,
sandbox=sandbox,
model=model,
api_key=api_key,
api_base=api_base,
instructions=instructions,
tools=tools,
skills=skills,
disable_tools=disable_tools,
permissions=permissions,
on_approval=on_approval,
output=output,
max_turns=max_turns,
timeout=timeout,
metadata=metadata,
options=options,
install=install,
)

6
litellm/harness/types.py
View file

@ -7,7 +7,7 @@ import json
from collections.abc import Mapping
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Literal
from typing import Literal
from pydantic import BaseModel
@ -71,7 +71,7 @@ class ToolCall:
id: str
name: str
native_name: str
input: Mapping[str, Any]
input: Mapping[str, object]
builtin: bool = True
@ -100,7 +100,7 @@ class Approval:
"""A request to run a tool. The turn waits until allow() or deny() is called."""
tool: str
input: Mapping[str, Any]
input: Mapping[str, object]
_decision: asyncio.Future[tuple[bool, str]] = field(
default_factory=lambda: asyncio.get_event_loop().create_future(),
compare=False,

15
litellm/integrations/arize/_utils.py
View file

@ -2,6 +2,7 @@ import json
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
from pydantic import ConfigDict, TypeAdapter
from typing_extensions import ReadOnly, TypedDict, override
from litellm._logging import verbose_logger
@ -29,6 +30,8 @@ from litellm.integrations._types.open_inference import (
ToolCallAttributes,
)
_JSON_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
class ArizeOTELAttributes(BaseLLMObsOTELAttributes):
@staticmethod
@ -609,9 +612,7 @@ def _coerce_response_obj_for_attrs(response_obj):
text: Final = getattr(response_obj, "text", None)
if isinstance(text, str) and text:
try:
parsed: Final = json.loads(text)
if isinstance(parsed, dict):
return parsed
return _JSON_OBJECT.validate_python(json.loads(text))
except Exception:
pass
return response_obj
@ -1062,9 +1063,7 @@ def _parse_passthrough_response(raw_response_obj, coerced_response_obj, kwargs):
inner = candidate.get("response")
if isinstance(inner, str):
try:
parsed = json.loads(inner)
if isinstance(parsed, dict):
return parsed
return _JSON_OBJECT.validate_python(json.loads(inner))
except Exception:
continue
if isinstance(inner, dict):
@ -1078,9 +1077,7 @@ def _parse_passthrough_response(raw_response_obj, coerced_response_obj, kwargs):
return original
if isinstance(original, str):
try:
parsed = json.loads(original)
if isinstance(parsed, dict):
return parsed
return _JSON_OBJECT.validate_python(json.loads(original))
except Exception:
return None
return None

8
litellm/integrations/datadog/datadog_llm_obs.py
View file

@ -15,6 +15,7 @@ from types import MappingProxyType
from typing import Any, Final, Literal
import httpx
from pydantic import ConfigDict, TypeAdapter, ValidationError
import litellm
from litellm._logging import verbose_logger
@ -57,6 +58,7 @@ from litellm.types.utils import (
_EMPTY_MAPPING: Final[Mapping[str, object]] = MappingProxyType({})
_EMPTY_MESSAGE: Final[Message] = {"role": "", "content": ""}
_MAX_PARSED_TOOL_ARGUMENT_CHARS: Final = 256 * 1024
_JSON_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
_SAFE_REDACTED_MESSAGE_ROLES: Final = frozenset(
{"agent", "assistant", "developer", "function", "model", "system", "tool", "user"}
)
@ -282,8 +284,10 @@ def _to_dd_arguments(raw_arguments: object) -> dict[str, object] | str:
return raw_arguments if isinstance(raw_arguments, dict) else str(raw_arguments)
if len(raw_arguments) > _MAX_PARSED_TOOL_ARGUMENT_CHARS:
return raw_arguments
parsed: Final = safe_json_loads(raw_arguments)
return parsed if isinstance(parsed, dict) else raw_arguments
try:
return _JSON_OBJECT.validate_python(safe_json_loads(raw_arguments))
except ValidationError:
return raw_arguments
def _to_dd_tool_calls(message: Mapping[str, object]) -> tuple[ToolCall, ...]:

15
litellm/integrations/galileo.py
View file

@ -4,12 +4,12 @@ import json
import os
import re
import uuid
from collections.abc import Mapping, Sequence
from collections.abc import Callable, Mapping, Sequence
from datetime import datetime, timezone, tzinfo
from typing import Any, Final, Protocol, cast
import httpx
from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
from typing_extensions import ReadOnly, TypedDict
import litellm
@ -35,6 +35,11 @@ GALILEO_CLOUD_API_BASE_URL: Final = "https://api.galileo.ai"
# unavailable, invalid credentials) cannot leak memory unboundedly.
GALILEO_MAX_IN_MEMORY_RECORDS: Final = 1000
_UNTYPED_VALUE: Final = TypeAdapter(object)
_ZERO_ARGUMENT_CALLABLE: Final[TypeAdapter[Callable[[], object]]] = TypeAdapter(
Callable[[], object], config=ConfigDict(hide_input_in_errors=True)
)
class _GalileoLoginBody(TypedDict):
"""Decoded body of the Galileo login response."""
@ -473,9 +478,9 @@ class GalileoObserve(CustomLogger):
if isinstance(value, str):
return value
def _json_default(obj: Any) -> object:
def _json_default(obj: object) -> object:
if hasattr(obj, "model_dump"):
return obj.model_dump()
return _ZERO_ARGUMENT_CALLABLE.validate_python(getattr(obj, "model_dump", None))()
return str(obj)
return json.dumps(value, default=_json_default)
@ -496,7 +501,7 @@ class GalileoObserve(CustomLogger):
if hasattr(message, "json"):
message_json: Final[object] = message.json()
if isinstance(message_json, str):
return json.loads(message_json)
return _UNTYPED_VALUE.validate_python(json.loads(message_json))
return message_json
return message
return None

8
litellm/integrations/opentelemetry.py
View file

@ -8,6 +8,8 @@ from datetime import datetime
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, TypedDict, cast
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm._logging import verbose_logger
from litellm.integrations._types.open_inference import (
@ -95,6 +97,8 @@ class _ResponseWithUsageView(TypedDict, total=False):
usage: "_UsageCompletionTokensView | None"
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
# Cap on credential-scoped providers held at once; each one owns an exporter thread.
_MAX_DYNAMIC_TRACER_PROVIDERS: Final = 256
@ -1633,7 +1637,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
attributes = self.config.attributes
if attributes is None and self.callback_name in (None, "otel"):
otel_settings: Final = (litellm.callback_settings or {}).get("otel") or {}
raw: Final = otel_settings.get("attributes") if isinstance(otel_settings, dict) else None
raw: Final[object] = otel_settings.get("attributes") if isinstance(otel_settings, dict) else None
if raw is not None:
attributes = _build_metric_attribute_filter(raw)
(
@ -2919,7 +2923,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
import json
try:
_parsed: Final[Mapping[str, object]] = json.loads(_raw_response)
_parsed: Final = _JSON_OBJECT.validate_python(json.loads(_raw_response))
for param, val in _parsed.items():
self.safe_set_attribute(
span=span,

11
litellm/integrations/opik/opik_payload_builder/api.py
View file

@ -1,12 +1,17 @@
"""Public API for Opik payload building."""
from collections.abc import Mapping
from datetime import datetime
from typing import Any, Final
from pydantic import ConfigDict, TypeAdapter
from litellm.integrations.opik import utils
from . import extractors, payload_builders, types
_STANDARD_LOGGING_FIELDS: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
def build_opik_payload(
kwargs: dict[str, Any],
@ -36,12 +41,14 @@ def build_opik_payload(
- First element is TracePayload if creating a new trace, None if attaching to existing
- Second element is always SpanPayload
"""
standard_logging_object: Final = kwargs["standard_logging_object"]
standard_logging_object: Final = _STANDARD_LOGGING_FIELDS.validate_python(kwargs["standard_logging_object"])
# Extract litellm params and metadata
litellm_params: Final = kwargs.get("litellm_params", {}) or {}
litellm_metadata: Final = litellm_params.get("metadata", {}) or {}
standard_logging_metadata: Final = standard_logging_object.get("metadata", {}) or {}
standard_logging_metadata: Final = _STANDARD_LOGGING_FIELDS.validate_python(
standard_logging_object.get("metadata", {}) or {}
)
# Extract and merge Opik metadata
opik_metadata: Final = extractors.extract_opik_metadata(litellm_metadata, standard_logging_metadata)

6
litellm/integrations/opik/opik_payload_builder/extractors.py
View file

@ -4,8 +4,12 @@ import json
from collections.abc import Mapping
from typing import Any, Final
from pydantic import TypeAdapter
from litellm import _logging
_DECODED_JSON: Final = TypeAdapter(object)
def normalize_provider_name(provider: str | None) -> str | None:
"""
@ -149,7 +153,7 @@ def apply_proxy_header_overrides(
thread_id = value
elif param_key == "tags":
try:
parsed_tags: object = json.loads(value)
parsed_tags = _DECODED_JSON.validate_python(json.loads(value))
if isinstance(parsed_tags, list):
tags.extend(parsed_tags)
except (json.JSONDecodeError, TypeError):

15
litellm/integrations/opik/opik_payload_builder/types.py
View file

@ -1,7 +1,8 @@
"""Type definitions for Opik payload building."""
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Final, Literal
from typing import Final, Literal
@dataclass
@ -13,9 +14,9 @@ class TracePayload:
name: str
start_time: str
end_time: str
input: Any
output: Any
metadata: dict[str, Any]
input: object
output: object
metadata: Mapping[str, object]
tags: list[str]
thread_id: str | None = None
@ -32,9 +33,9 @@ class SpanPayload:
model: str
start_time: str
end_time: str
input: Any
output: Any
metadata: dict[str, Any]
input: object
output: object
metadata: Mapping[str, object]
tags: list[str]
usage: dict[str, int]
parent_span_id: str | None = None

9
litellm/integrations/opik/utils.py
View file

@ -2,7 +2,8 @@ import configparser
import os
import time
import uuid
from typing import Any, Final
from collections.abc import Mapping, Sequence
from typing import Final
CONFIG_FILE_PATH_DEFAULT: Final[str] = "~/.opik.config"
@ -93,14 +94,14 @@ def create_usage_object(usage):
return usage_dict
def _remove_nulls(x: dict[str, Any]) -> dict[str, Any]:
def _remove_nulls(x: Mapping[str, object]) -> dict[str, object]:
"""Remove None values from dict."""
return {k: v for k, v in x.items() if v is not None}
def get_traces_and_spans_from_payload(
payload: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
payload: Sequence[Mapping[str, object]],
) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
"""
Separate traces and spans from payload.

4
litellm/integrations/otel/model/config.py
View file

@ -2,7 +2,7 @@
from enum import Enum
from functools import lru_cache
from typing import Annotated, Any, Final
from typing import Annotated, Final
from pydantic import AliasChoices, BaseModel, Field, TypeAdapter, ValidationError, field_validator, model_validator
from pydantic.fields import FieldInfo
@ -315,7 +315,7 @@ class OpenTelemetryV2Config(BaseSettings):
mode="before",
)
@classmethod
def _split_csv(cls, value: Any) -> Any:
def _split_csv(cls, value: object) -> object:
"""Accept a comma-separated string for list fields.
Env vars are strings, but these fields are lists. Pydantic-settings would

4
litellm/integrations/otel/plumbing/metrics.py
View file

@ -307,7 +307,7 @@ class GenAIMetricRecorder:
return common_attrs
def _bounded_attributes(self, kwargs: Mapping[str, Any]) -> MetricAttributes:
def _bounded_attributes(self, kwargs: Mapping[str, object]) -> MetricAttributes:
"""The datapoint attributes, capped at :data:`METRIC_ATTRIBUTE_CEILING`.
The cap runs BEFORE the operator's include/exclude filter so the filter can
@ -322,7 +322,7 @@ class GenAIMetricRecorder:
attributes = None
if self._callback_name in (None, "otel"):
otel_settings: Final = (litellm.callback_settings or {}).get("otel") or {}
raw: Final = otel_settings.get("attributes") if isinstance(otel_settings, dict) else None
raw: Final[object] = otel_settings.get("attributes") if isinstance(otel_settings, dict) else None
if raw is not None:
attributes = _build_metric_attribute_filter(raw)
# A bad filter (include_list + exclude_list both set, an unfilterable name)

4
litellm/integrations/s3_v2.py
View file

@ -595,7 +595,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
and not (self.s3_drop_on_terminal_error and _is_terminal(response))
and attempt < max_retries - 1
):
wait_time = 2**attempt # 1s, 2s
wait_time = 1 << attempt # 1s, 2s
verbose_logger.log(
logging.DEBUG if _in_flush.get() else logging.WARNING,
"S3 upload returned %s, retrying in %ss (attempt %s/%s) key=%s",
@ -897,7 +897,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
and not (self.s3_drop_on_terminal_error and _is_terminal(response))
and attempt < max_retries - 1
):
wait_time = 2**attempt # 1s, 2s
wait_time = 1 << attempt # 1s, 2s
verbose_logger.warning(
"S3 upload returned %s, retrying in %ss (attempt %s/%s) key=%s",
response.status_code,

20
litellm/integrations/vector_store_integrations/vector_store_pre_call_hook.py
View file

@ -6,12 +6,12 @@ It searches the vector store for relevant context, runs the request's pre-call g
over that context, and appends it to the messages.
"""
from collections.abc import Awaitable, Callable, Mapping, Sequence
from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
from dataclasses import dataclass
from itertools import chain
from typing import TYPE_CHECKING, Any, Final, Protocol, cast, get_args
from pydantic import TypeAdapter, ValidationError
from pydantic import ConfigDict, TypeAdapter, ValidationError
from typing_extensions import assert_never
import litellm
@ -44,10 +44,12 @@ else:
LiteLLMLoggingObj = Any
SEARCH_FAILURES_FIELD: Final = "vector_store_search_failures"
_PROVIDER_FIELDS_ATTRIBUTE: Final = "provider_specific_fields"
_DEFAULT_FAILURE_MODE: Final[VectorStoreSearchFailureMode] = "annotate"
_FAILURE_MODE_ADAPTER: Final = TypeAdapter(VectorStoreSearchFailureMode)
_OBJECT_ADAPTER: Final = TypeAdapter(object)
_STR_KEYED_ADAPTER: Final = TypeAdapter(dict[str, object])
_ITERABLE_ADAPTER: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
_GUARDRAIL_KEYS_THE_PROXY_MERGES_INTO_METADATA: Final = frozenset(
{"guardrails", "guardrail_config", "policies", "include_guardrail_response"}
)
@ -475,7 +477,7 @@ class VectorStorePreCallHook(CustomLogger):
async def async_post_call_streaming_deployment_hook(
self,
request_data: dict,
response_chunk: Any,
response_chunk: object,
call_type: CallTypes | None,
) -> object | None:
"""
@ -496,15 +498,17 @@ class VectorStorePreCallHook(CustomLogger):
return response_chunk
# Add search results to streaming chunk
if hasattr(response_chunk, "choices") and response_chunk.choices:
for choice in response_chunk.choices:
if hasattr(choice, "delta") and choice.delta:
provider_fields = getattr(choice.delta, "provider_specific_fields", None) or {}
choices: Final[object] = getattr(response_chunk, "choices", None)
if choices:
for choice in _ITERABLE_ADAPTER.validate_python(choices):
delta: object = getattr(choice, "delta", None)
if delta:
provider_fields = getattr(delta, _PROVIDER_FIELDS_ATTRIBUTE, None) or {}
if search_results:
provider_fields["search_results"] = search_results
if search_failures:
provider_fields[SEARCH_FAILURES_FIELD] = search_failures
choice.delta.provider_specific_fields = provider_fields
setattr(delta, _PROVIDER_FIELDS_ATTRIBUTE, provider_fields)
# Return modified chunk
return response_chunk

14
litellm/llms/a2a/chat/transformation.py
View file

@ -3,8 +3,8 @@ A2A Protocol Transformation for LiteLLM
"""
import uuid
from collections.abc import Iterator, Mapping
from typing import TYPE_CHECKING, Any, Final
from collections.abc import AsyncIterator, Iterator, Mapping
from typing import TYPE_CHECKING, Final
import httpx
@ -72,9 +72,9 @@ class A2AConfig(BaseConfig):
agent_name: str,
api_base: str | None,
api_key: str | None,
headers: dict[str, Any] | None,
optional_params: dict[str, Any],
) -> tuple[str | None, str | None, dict[str, Any] | None]:
headers: dict[str, object] | None,
optional_params: dict[str, object],
) -> tuple[str | None, str | None, dict[str, object] | None]:
"""
Resolve agent configuration from the registry for a registered agent.
@ -376,7 +376,7 @@ class A2AConfig(BaseConfig):
def get_model_response_iterator(
self,
streaming_response: Iterator | Any,
streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
sync_stream: bool,
json_mode: bool | None = False,
) -> BaseModelResponseIterator:
@ -397,7 +397,7 @@ class A2AConfig(BaseConfig):
json_mode=json_mode,
)
def _openai_message_to_a2a_message(self, message: dict[str, Any]) -> dict[str, Any]:
def _openai_message_to_a2a_message(self, message: Mapping[str, object]) -> dict[str, object]:
"""
Convert OpenAI message to A2A message format.

10
litellm/llms/aiohttp_openai/chat/transformation.py
View file

@ -7,9 +7,11 @@ https://github.com/BerriAI/litellm/issues/6592
New config to ensure we introduce this without causing breaking changes for users
"""
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
from aiohttp import ClientResponse
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.openai_like.chat.transformation import OpenAILikeChatConfig
from litellm.types.llms.openai import AllMessageValues
@ -23,6 +25,10 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECTS: Final = TypeAdapter(
Iterable[Mapping[str, object]], config=ConfigDict(strict=True, hide_input_in_errors=True)
)
class AiohttpOpenAIChatConfig(OpenAILikeChatConfig):
def get_complete_url(
@ -73,7 +79,9 @@ class AiohttpOpenAIChatConfig(OpenAILikeChatConfig):
) -> ModelResponse:
_json_response: Final = await raw_response.json()
model_response.id = _json_response.get("id")
model_response.choices = [Choices(**choice) for choice in _json_response.get("choices")]
model_response.choices = [
Choices.model_validate(choice) for choice in _JSON_OBJECTS.validate_python(_json_response.get("choices"))
]
model_response.created = _json_response.get("created")
model_response.model = _json_response.get("model")
model_response.object = _json_response.get("object")

2
litellm/llms/anthropic/count_tokens/token_counter.py
View file

@ -27,7 +27,7 @@ class AnthropicTokenCounter(BaseTokenCounter):
self,
model_to_use: str,
messages: list[dict[str, Any]] | None,
contents: list[dict[str, Any]] | None,
contents: list[dict[str, object]] | None,
deployment: dict[str, Any] | None = None,
request_model: str = "",
tools: list[dict[str, Any]] | None = None,

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

@ -19,6 +19,7 @@ from typing import Final, cast
import httpx
from openai.types.file_deleted import FileDeleted
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.prompt_templates.common_utils import extract_file_data
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
@ -47,6 +48,8 @@ ANTHROPIC_FILES_API_BASE: Final = "https://api.anthropic.com"
ANTHROPIC_FILES_BETA_HEADER: Final = "files-api-2025-04-14"
ANTHROPIC_MESSAGE_BATCH_ID_PREFIX: Final = "msgbatch_"
_JSON_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
class AnthropicFilesConfig(BaseFilesConfig):
"""
@ -211,7 +214,7 @@ class AnthropicFilesConfig(BaseFilesConfig):
"created_at": "2025-01-01T00:00:00Z"
}
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
return self._parse_anthropic_file(response_json)
def transform_retrieve_file_request(
@ -230,7 +233,7 @@ class AnthropicFilesConfig(BaseFilesConfig):
logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
) -> OpenAIFileObject:
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
return self._parse_anthropic_file(response_json)
def transform_delete_file_request(
@ -249,13 +252,9 @@ class AnthropicFilesConfig(BaseFilesConfig):
logging_obj: LiteLLMLoggingObj,
litellm_params: dict,
) -> FileDeleted:
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
file_id: Final = response_json.get("id", "")
return FileDeleted(
id=file_id,
deleted=True,
object="file",
)
return FileDeleted.model_validate({"id": file_id, "deleted": True, "object": "file"})
def transform_list_files_request(
self,

18
litellm/llms/azure/audio_transcription/transformation.py
View file

@ -5,10 +5,12 @@ Maps OpenAI-compatible audio transcription calls to Azure Speech REST
recognition for short audio.
"""
from typing import Any, Final
from collections.abc import Mapping
from typing import Final
from urllib.parse import urlencode, urlparse
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
from litellm.llms.base_llm.audio_transcription.transformation import (
@ -23,6 +25,8 @@ from litellm.types.llms.openai import (
)
from litellm.types.utils import FileTypes, TranscriptionResponse
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class AzureSpeechAudioTranscriptionException(BaseLLMException):
pass
@ -127,7 +131,8 @@ class AzureSpeechAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
raw_response: httpx.Response,
) -> TranscriptionResponse:
response_json: Final = raw_response.json()
recognition_status: Final = response_json.get("RecognitionStatus")
payload: Final = _JSON_OBJECT.validate_python(response_json)
recognition_status: Final = payload.get("RecognitionStatus")
if recognition_status is not None and recognition_status != "Success":
raise AzureSpeechAudioTranscriptionException(
message=(f"Azure AI Speech transcription failed with RecognitionStatus={recognition_status}."),
@ -135,7 +140,7 @@ class AzureSpeechAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
headers=raw_response.headers,
)
text: Final = self._extract_text(response_json)
text: Final = self._extract_text(payload)
response: Final = TranscriptionResponse(text=text)
response._hidden_params = response_json
return response
@ -194,9 +199,10 @@ class AzureSpeechAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
return "detailed"
return "simple"
def _extract_text(self, response_json: dict[str, Any]) -> str:
if isinstance(response_json.get("DisplayText"), str):
return response_json["DisplayText"]
def _extract_text(self, response_json: Mapping[str, object]) -> str:
display_text: Final = response_json.get("DisplayText")
if isinstance(display_text, str):
return display_text
nbest: Final = response_json.get("NBest")
if isinstance(nbest, list) and nbest:

2
litellm/llms/azure_ai/anthropic/count_tokens/handler.py
View file

@ -30,7 +30,7 @@ class AzureAIAnthropicCountTokensHandler(AzureAIAnthropicCountTokensConfig):
messages: list[dict[str, Any]],
api_key: str,
api_base: str,
litellm_params: dict[str, Any] | None = None,
litellm_params: dict[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
tools: list[dict[str, Any]] | None = None,
system: object = None,

4
litellm/llms/base_llm/harness/transformation.py
View file

@ -18,7 +18,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar
from typing import TYPE_CHECKING, ClassVar, Generic, TypeVar
from litellm.harness.errors import HarnessError, OptionsMismatch
from litellm.harness.types import Capabilities, Event, Harness
@ -133,7 +133,7 @@ class BaseCLIHarnessConfig(BaseHarnessConfig[OptionsT], Generic[OptionsT, Stream
"""Fresh per-turn parser state."""
@abstractmethod
def transform_stream_line(self, line: Mapping[str, Any], state: StreamStateT) -> Sequence[Event]:
def transform_stream_line(self, line: Mapping[str, object], state: StreamStateT) -> Sequence[Event]:
"""One decoded JSON line from stdout to zero or more events. Pure."""
@abstractmethod

16
litellm/llms/bedrock/batches/handler.py
View file

@ -1,6 +1,6 @@
from collections.abc import Mapping, Sequence
from datetime import datetime
from typing import TYPE_CHECKING, Any, Final, cast
from typing import TYPE_CHECKING, Final, Literal
from openai.types.batch import BatchRequestCounts
from openai.types.batch import Metadata as OpenAIBatchMetadata
@ -17,7 +17,12 @@ if TYPE_CHECKING:
# AWS Bedrock model-invocation-job statuses → OpenAI Batch statuses.
# Mirrors the mapping used by `BedrockBatchesConfig.transform_create_batch_response`
# so create / retrieve return consistent statuses.
_BEDROCK_MIJ_STATUS_TO_OPENAI: Final = {
_BEDROCK_MIJ_STATUS_TO_OPENAI: Final[
Mapping[
str,
Literal["validating", "failed", "in_progress", "finalizing", "completed", "expired", "cancelling", "cancelled"],
]
] = {
"Submitted": "validating",
"Validating": "validating",
"Scheduled": "validating",
@ -92,7 +97,7 @@ def _record_counts_from_response(response: Mapping[str, object]) -> BatchRequest
)
def _to_epoch(value: Any) -> int | None:
def _to_epoch(value: object) -> int | None:
if value is None:
return None
if isinstance(value, (int, float)):
@ -349,10 +354,7 @@ class BedrockBatchesHandler:
)
bedrock_status: Final = str(response.get("status", ""))
openai_status: Final = cast(
Any,
_BEDROCK_MIJ_STATUS_TO_OPENAI.get(bedrock_status, "in_progress"),
)
openai_status: Final = _BEDROCK_MIJ_STATUS_TO_OPENAI.get(bedrock_status, "in_progress")
input_uri: Final = response.get("inputDataConfig", {}).get("s3InputDataConfig", {}).get("s3Uri", "")
output_prefix: Final = response.get("outputDataConfig", {}).get("s3OutputDataConfig", {}).get("s3Uri", "")

3
litellm/llms/bedrock/image_edit/stability_transformation.py
View file

@ -25,6 +25,7 @@ import base64
from typing import TYPE_CHECKING, Any, Final
import httpx
from httpx._types import RequestFiles
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
from litellm.llms.bedrock.common_utils import BedrockError
@ -165,7 +166,7 @@ class BedrockStabilityImageEditConfig(BaseImageEditConfig):
image_edit_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[dict, Any]:
) -> tuple[dict, RequestFiles]:
"""
Transform OpenAI-style request to Bedrock Stability request format.

10
litellm/llms/bedrock/image_generation/amazon_titan_transformation.py
View file

@ -80,9 +80,7 @@ class AmazonTitanImageGenerationConfig:
non_default_params: dict,
optional_params: dict,
):
from typing import Any
image_generation_config: Final[dict[str, Any]] = {}
image_generation_config: Final[dict[str, object]] = {}
for k, v in non_default_params.items():
if k == "size" and v is not None:
width, height = v.split("x")
@ -106,11 +104,9 @@ class AmazonTitanImageGenerationConfig:
text: str,
optional_params: dict,
) -> AmazonTitanImageGenerationRequestBody:
from typing import Any
image_generation_config = optional_params.pop("imageGenerationConfig", {})
negative_text: Final = optional_params.pop("negativeText", None)
text_to_image_params: Final[dict[str, Any]] = {"text": text}
text_to_image_params: Final = AmazonTitanTextToImageParams(text=text)
if negative_text:
text_to_image_params["negativeText"] = negative_text
task_type: Final = optional_params.pop("taskType", "TEXT_IMAGE")
@ -121,7 +117,7 @@ class AmazonTitanImageGenerationConfig:
}
return AmazonTitanImageGenerationRequestBody(
taskType=task_type,
textToImageParams=AmazonTitanTextToImageParams(**text_to_image_params),
textToImageParams=text_to_image_params,
imageGenerationConfig=AmazonNovaCanvasImageGenerationConfig(**image_generation_config),
)

9
litellm/llms/bedrock/rerank/handler.py
View file

@ -2,6 +2,7 @@ import json
from typing import TYPE_CHECKING, Any, Final, cast
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging
@ -24,6 +25,8 @@ if TYPE_CHECKING:
else:
AWSPreparedRequest = Any
_JSON_DICT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
class BedrockRerankHandler(BaseAWSLLM):
async def arerank(
@ -55,13 +58,13 @@ class BedrockRerankHandler(BaseAWSLLM):
except httpx.TimeoutException:
raise BedrockError(status_code=408, message="Timeout error occurred.")
return BedrockRerankConfig()._transform_response(response.json())
return BedrockRerankConfig()._transform_response(_JSON_DICT.validate_python(response.json()))
def rerank(
self,
model: str,
query: str,
documents: list[str | dict[str, Any]],
documents: list[str | dict[str, object]],
optional_params: dict,
logging_obj: LitellmLogging,
top_n: int | None = None,
@ -136,7 +139,7 @@ class BedrockRerankHandler(BaseAWSLLM):
api_key="",
)
response_json: Final = response.json()
response_json: Final = _JSON_DICT.validate_python(response.json())
return BedrockRerankConfig()._transform_response(response_json)

7
litellm/llms/bedrock/search/transformation.py
View file

@ -37,6 +37,7 @@ from collections.abc import Iterator, Mapping, Sequence
from typing import Final
import httpx
from pydantic import TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -81,6 +82,8 @@ _SSE_EVENT_SEPARATOR: Final = re.compile(r"\r?\n[ \t]*\r?\n")
_SSE_LINE_PREFIXES: Final = ("event:", "data:", ":", "id:", "retry:")
_JSON_VALUE: Final = TypeAdapter(object)
def _gateway_host_match(api_base: str) -> re.Match[str] | None:
return _GATEWAY_HOST_PATTERN.fullmatch(httpx.URL(api_base).host)
@ -128,7 +131,7 @@ def _parse_result_items(raw_text: object) -> tuple[Mapping[str, object], ...]:
if not isinstance(raw_text, str):
return ()
try:
parsed: Final = json.loads(raw_text)
parsed: Final = _JSON_VALUE.validate_python(json.loads(raw_text))
except json.JSONDecodeError:
return ()
return _result_items(parsed)
@ -147,7 +150,7 @@ def _iter_sse_events(text: str) -> Iterator[Mapping[str, object]]:
if not payload:
continue
try:
parsed = json.loads(payload)
parsed = _JSON_VALUE.validate_python(json.loads(payload))
except json.JSONDecodeError:
continue
if isinstance(parsed, dict):

8
litellm/llms/black_forest_labs/image_generation/transformation.py
View file

@ -8,9 +8,11 @@ API Reference: https://docs.bfl.ai/
"""
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
@ -36,6 +38,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class BlackForestLabsImageGenerationConfig(BaseImageGenerationConfig):
"""
@ -218,7 +222,7 @@ class BlackForestLabsImageGenerationConfig(BaseImageGenerationConfig):
https://docs.bfl.ai/flux_models/flux_1_1_pro
"""
# Build request body with prompt
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"prompt": prompt,
}
@ -275,7 +279,7 @@ class BlackForestLabsImageGenerationConfig(BaseImageGenerationConfig):
message=f"Error parsing BFL response: {e}",
)
result: Final = response_data.get("result", {})
result: Final = _JSON_OBJECT.validate_python(response_data).get("result", {})
if not model_response.data:
model_response.data = []

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

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.openai.common_utils import OpenAIError
@ -20,6 +22,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(strict=True, hide_input_in_errors=True))
class ClarifaiConfig(OpenAIGPTConfig):
"""
@ -111,7 +115,7 @@ class ClarifaiConfig(OpenAIGPTConfig):
headers=raw_response.headers,
) from e
response: Final = ModelResponse(**completion_response)
response: Final = ModelResponse(**_JSON_OBJECT.validate_python(completion_response))
if response.model is not None:
response.model = "clarifai/" + model

8
litellm/llms/claude_code/harness/transformation.py
View file

@ -16,6 +16,8 @@ from dataclasses import dataclass, field
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final
from pydantic import ConfigDict, TypeAdapter
from litellm.harness.errors import HarnessError, OptionsMismatch
from litellm.harness.options import ClaudeCodeOptions
from litellm.harness.types import (
@ -48,6 +50,7 @@ if TYPE_CHECKING:
CLAUDE_BINARY: Final = "claude"
SYNTHETIC_MODEL: Final = "<synthetic>"
_BLOCK: Final = TypeAdapter(Mapping[object, object], config=ConfigDict(hide_input_in_errors=True))
BASE_COMMAND: Final = ("-p", "--output-format", "stream-json", "--verbose", "--input-format", "text")
@ -157,7 +160,7 @@ def _stringify_block(block: object) -> str:
return json.dumps(block, ensure_ascii=False)
def _message_blocks(event: Mapping[str, object]) -> Sequence[Any]:
def _message_blocks(event: Mapping[str, object]) -> Sequence[object]:
message: Final = event.get("message")
content: Final = message.get("content") if isinstance(message, Mapping) else None
if isinstance(content, str):
@ -211,8 +214,7 @@ def _user_events(event: Mapping[str, object]) -> Sequence[Event]:
output=stringify_tool_output(block.get("content")),
is_error=bool(block.get("is_error", False)),
)
for block in _message_blocks(event)
if _is_tool_result(block)
for block in map(_BLOCK.validate_python, filter(_is_tool_result, _message_blocks(event)))
)
)

16
litellm/llms/codex/harness/transformation.py
View file

@ -17,6 +17,8 @@ from dataclasses import dataclass, field
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final
from pydantic import ConfigDict, TypeAdapter
from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
from litellm.harness.errors import HarnessError, OptionsMismatch
from litellm.harness.options import CodexOptions
@ -61,6 +63,7 @@ MANAGED_CONFIG_KEYS: Final = frozenset(
)
_BARE_TOML_KEY: Final = re.compile(r"^[A-Za-z0-9_-]+$")
_TOOL_ITEM_TYPES: Final = frozenset({"command_execution", "file_change", "web_search", "mcp_tool_call"})
_CHANGES: Final = TypeAdapter(tuple[Mapping[object, object], ...], config=ConfigDict(hide_input_in_errors=True))
@dataclass
@ -92,7 +95,7 @@ def _tool_input(item: Mapping[str, Any]) -> tuple[str, str, Mapping[str, Any], b
return name, tool, tool_args, False
def _tool_output(item: Mapping[str, Any]) -> tuple[str, bool]:
def _tool_output(item: Mapping[str, object]) -> tuple[str, bool]:
"""(output text, is_error) for a completed tool-like item."""
item_type = item.get("type")
status = item.get("status")
@ -101,7 +104,8 @@ def _tool_output(item: Mapping[str, Any]) -> tuple[str, bool]:
is_error = status == "failed" or (exit_code is not None and exit_code != 0)
return str(item.get("aggregated_output") or ""), is_error
if item_type == "file_change":
lines = (f"{c.get('kind', '')} {c.get('path', '')}".strip() for c in item.get("changes") or ())
changes: Final = _CHANGES.validate_python(item.get("changes") or ())
lines: Final = (f"{c.get('kind', '')} {c.get('path', '')}".strip() for c in changes)
return "\n".join(lines), status == "failed"
if item_type == "web_search":
return "", status == "failed"
@ -118,7 +122,7 @@ def _tool_output(item: Mapping[str, Any]) -> tuple[str, bool]:
def _tool_item_events(
item_id: str, item: Mapping[str, Any], completed: bool, state: CodexStreamState
item_id: str, item: Mapping[str, object], completed: bool, state: CodexStreamState
) -> Iterator[Event]:
if item_id not in state.started:
state.started.add(item_id)
@ -129,7 +133,7 @@ def _tool_item_events(
yield ToolResult(id=item_id, output=output, is_error=is_error)
def _item_events(event_type: str, item: Mapping[str, Any], state: CodexStreamState) -> Sequence[Event]:
def _item_events(event_type: str, item: Mapping[str, object], state: CodexStreamState) -> Sequence[Event]:
item_type = item.get("type")
item_id = str(item.get("id") or "")
completed = event_type == "item.completed"
@ -181,7 +185,7 @@ def _config_override(key: object, value: object) -> str:
return f"{dotted}={toml_value(value)}"
def config_overrides(config: Mapping[str, Any]) -> Sequence[str]:
def config_overrides(config: Mapping[str, object]) -> Sequence[str]:
"""`-c` override strings for CodexOptions.config, rejecting managed keys."""
overrides: Final = (_config_override(key, value) for key, value in config.items())
return list(overrides) # mutable-ok: public helper; tests compare to a list
@ -310,7 +314,7 @@ class CodexHarnessConfig(BaseCLIHarnessConfig):
def create_stream_state(self) -> CodexStreamState:
return CodexStreamState()
def transform_stream_line(self, line: Mapping[str, Any], state: CodexStreamState) -> Sequence[Event]:
def transform_stream_line(self, line: Mapping[str, object], state: CodexStreamState) -> Sequence[Event]:
"""turn.completed usage is ignored on purpose: the session endpoint accounts it."""
event_type = line.get("type")
if event_type == "thread.started":

9
litellm/llms/cohere/rerank/transformation.py
View file

@ -1,7 +1,8 @@
from collections.abc import Mapping
from typing import Any, Final
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -12,6 +13,8 @@ from litellm.types.rerank import OptionalRerankParams, RerankRequest, RerankResp
from ..common_utils import CohereError
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class CohereRerankConfig(BaseRerankConfig):
"""
@ -51,7 +54,7 @@ class CohereRerankConfig(BaseRerankConfig):
model: str,
drop_params: bool,
query: str,
documents: list[str | dict[str, Any]],
documents: list[str | dict[str, object]],
custom_llm_provider: str | None = None,
top_n: int | None = None,
rank_fields: list[str] | None = None,
@ -148,7 +151,7 @@ class CohereRerankConfig(BaseRerankConfig):
except Exception:
raise CohereError(message=raw_response.text, status_code=raw_response.status_code)
return RerankResponse(**raw_response_json)
return RerankResponse.model_validate(_JSON_OBJECT.validate_python(raw_response_json))
def get_error_class(self, error_message: str, status_code: int, headers: dict | httpx.Headers) -> BaseLLMException:
return CohereError(message=error_message, status_code=status_code)

8
litellm/llms/cometapi/image_generation/transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
@ -20,6 +22,9 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
class CometAPIImageGenerationConfig(BaseImageGenerationConfig):
DEFAULT_BASE_URL: str = "https://api.cometapi.com"
@ -155,7 +160,8 @@ class CometAPIImageGenerationConfig(BaseImageGenerationConfig):
# CometAPI returns OpenAI-compatible format
# Expected format: {"created": timestamp, "data": [{"url": "...", "b64_json": "..."}]}
if "data" in response_data:
for image_data in response_data["data"]:
payload: Final = _JSON_OBJECT.validate_python(response_data)
for image_data in _JSON_OBJECTS.validate_python(payload["data"]):
image_obj = ImageObject(
b64_json=image_data.get("b64_json"),
url=image_data.get("url"),

16
litellm/llms/dashscope/image_generation/transformation.py
View file

@ -23,9 +23,11 @@ Response format:
}
"""
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
@ -45,6 +47,9 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
DEFAULT_API_BASE: Final = "https://dashscope-intl.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation"
CHAT_COMPATIBLE_MODE_PATH: Final = "/compatible-mode/v1"
@ -192,9 +197,10 @@ class DashScopeImageGenerationConfig(BaseImageGenerationConfig):
# DashScope can return API-level errors in a 200 response body.
# Example: {"code": "InvalidParameter", "message": "Size not supported"}
if "code" in response_data and "output" not in response_data:
response_object: Final = _JSON_OBJECT.validate_python(response_data)
if "code" in response_object and "output" not in response_object:
raise self.get_error_class(
error_message=str(response_data.get("message", response_data)),
error_message=str(response_object.get("message", response_object)),
status_code=raw_response.status_code,
headers=raw_response.headers,
)
@ -202,9 +208,11 @@ class DashScopeImageGenerationConfig(BaseImageGenerationConfig):
if not model_response.data:
model_response.data = []
choices: Final = response_data.get("output", {}).get("choices", [])
output: Final = _JSON_OBJECT.validate_python(response_object.get("output", {}))
choices: Final = _JSON_OBJECTS.validate_python(output.get("choices", []))
for choice in choices:
content_list = choice.get("message", {}).get("content", [])
message = _JSON_OBJECT.validate_python(choice.get("message", {}))
content_list = _JSON_OBJECTS.validate_python(message.get("content", []))
for content_item in content_list:
image_url = content_item.get("image")
if image_url:

23
litellm/llms/dashscope/rerank/transformation.py
View file

@ -26,10 +26,11 @@ as supported only for gte-rerank-v2 / qwen3-vl-rerank.
Docs - https://help.aliyun.com/zh/model-studio/text-rerank-api
"""
from collections.abc import Mapping
from typing import Any, Final
from collections.abc import Iterable, Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm._uuid import uuid
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -48,6 +49,11 @@ from ..common_utils import DashScopeError, resolve_dashscope_family_rerank_api_b
DEFAULT_RERANK_URL: Final = "https://dashscope.aliyuncs.com/compatible-api/v1/reranks"
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_OPTIONAL_INT: Final[TypeAdapter[int | None]] = TypeAdapter(int | None)
_STR: Final = TypeAdapter(str)
class DashScopeRerankConfig(BaseRerankConfig):
"""
@ -117,7 +123,7 @@ class DashScopeRerankConfig(BaseRerankConfig):
model: str,
drop_params: bool,
query: str,
documents: list[str | dict[str, Any]],
documents: list[str | dict[str, object]],
custom_llm_provider: str | None = None,
top_n: int | None = None,
rank_fields: list[str] | None = None,
@ -197,7 +203,8 @@ class DashScopeRerankConfig(BaseRerankConfig):
message=response_json.get("message", str(response_json)),
)
results: Final = response_json.get("results")
payload: Final = _JSON_OBJECT.validate_python(response_json)
results: Final = payload.get("results")
if results is None:
raise DashScopeError(
status_code=raw_response.status_code,
@ -210,7 +217,7 @@ class DashScopeRerankConfig(BaseRerankConfig):
# "document": {"text": "..."}
# which already matches LiteLLM's RerankResponseDocument shape.
transformed_results: Final[list[dict]] = []
for r in results:
for r in _JSON_OBJECTS.validate_python(results):
item: dict[str, object] = {
"index": r["index"],
"relevance_score": r["relevance_score"],
@ -223,14 +230,14 @@ class DashScopeRerankConfig(BaseRerankConfig):
item["document"] = {"text": doc}
transformed_results.append(item)
usage: Final = response_json.get("usage") or {}
total_tokens: Final = usage.get("total_tokens")
usage: Final = _JSON_OBJECT.validate_python(payload.get("usage") or {})
total_tokens: Final = _OPTIONAL_INT.validate_python(usage.get("total_tokens"))
billed_units: Final = RerankBilledUnits(total_tokens=total_tokens)
tokens: Final = RerankTokens(input_tokens=total_tokens)
meta: Final = RerankResponseMeta(billed_units=billed_units, tokens=tokens)
return RerankResponse(
id=response_json.get("id") or str(uuid.uuid4()),
id=_STR.validate_python(payload.get("id") or str(uuid.uuid4())),
results=transformed_results,
meta=meta,
)

6
litellm/llms/deepagents/harness/transformation.py
View file

@ -173,7 +173,7 @@ def stream_events(
]
def tool_call_event(call: Mapping[str, Any]) -> ToolCall:
def tool_call_event(call: Mapping[str, object]) -> ToolCall:
native = str(call.get("name") or "")
args = call.get("args")
return ToolCall(
@ -185,7 +185,7 @@ def tool_call_event(call: Mapping[str, Any]) -> ToolCall:
)
def _node_messages(update: Mapping[Any, Any]) -> Iterator[object]:
def _node_messages(update: Mapping[object, object]) -> Iterator[object]:
for node, delta in update.items():
if node not in _EVENT_NODES or not isinstance(delta, Mapping):
continue
@ -231,7 +231,7 @@ def interrupts_in(
return list(items) # mutable-ok: list return; callers/tests compare to lists
def final_ai_text(messages: Sequence[Any]) -> str:
def final_ai_text(messages: Sequence[object]) -> str:
for message in reversed(messages):
if getattr(message, "type", None) == "ai":
text = content_text(getattr(message, "content", ""))

9
litellm/llms/deepgram/audio_transcription/transformation.py
View file

@ -2,10 +2,12 @@
Translates from OpenAI's `/v1/audio/transcriptions` to Deepgram's `/v1/listen`
"""
from collections.abc import Iterable, Mapping
from typing import Final
from urllib.parse import urlencode
from httpx import Headers, Response
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
from litellm.llms.base_llm.chat.transformation import BaseLLMException
@ -22,6 +24,9 @@ from ...base_llm.audio_transcription.transformation import (
)
from ..common_utils import DeepgramException
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
class DeepgramAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
def get_supported_openai_params(self, model: str) -> list[OpenAIAudioTranscriptionOptionalParams]:
@ -105,7 +110,7 @@ class DeepgramAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
response["task"] = "transcribe"
# Use detected_language if available, otherwise default to "en"
detected_language: Final = first_channel.get("detected_language")
detected_language: Final = _JSON_OBJECT.validate_python(first_channel).get("detected_language")
response["language"] = detected_language if detected_language else "en"
response["duration"] = response_json["metadata"]["duration"]
@ -114,7 +119,7 @@ class DeepgramAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
if "words" in first_alternative:
response["words"] = [
{"word": word["word"], "start": word["start"], "end": word["end"]}
for word in first_alternative["words"]
for word in _JSON_OBJECTS.validate_python(first_alternative["words"])
]
# Store full response in hidden params

36
litellm/llms/e2b/sandbox/transformation.py
View file

@ -8,9 +8,11 @@ Talks to e2b's REST API directly over httpx (no e2b SDK dependency):
"""
import json
from collections.abc import Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.sandbox.transformation import (
SANDBOX_MAX_OUTPUT_BYTES,
@ -32,6 +34,12 @@ JUPYTER_PORT: Final = 49999
DEFAULT_SANDBOX_TIMEOUT: Final = 300
MAX_OUTPUT_BYTES: Final = SANDBOX_MAX_OUTPUT_BYTES
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_MESSAGE: Final[TypeAdapter[Mapping[str, object] | None]] = TypeAdapter(
Mapping[str, object] | None, config=ConfigDict(hide_input_in_errors=True)
)
_TEXT: Final = TypeAdapter(str, config=ConfigDict(strict=True, hide_input_in_errors=True))
class E2BSandboxConfig(BaseSandboxConfig):
def _http(self, client: AsyncHTTPHandler | None) -> AsyncHTTPHandler:
@ -73,12 +81,14 @@ class E2BSandboxConfig(BaseSandboxConfig):
headers={"X-API-Key": key, "Content-Type": "application/json"},
json=body,
)
data: Final = response.json()
data: Final = _JSON_OBJECT.validate_python(response.json())
handle: Final = ContainerHandle(
id=data["sandboxID"],
provider="e2b",
domain=data.get("domain") or E2B_DEFAULT_DOMAIN,
handle: Final = ContainerHandle.model_validate(
{
"id": data["sandboxID"],
"provider": "e2b",
"domain": data.get("domain") or E2B_DEFAULT_DOMAIN,
}
)
handle._hidden_params = {
"envd_access_token": data.get("envdAccessToken"),
@ -158,7 +168,7 @@ class E2BSandboxConfig(BaseSandboxConfig):
def _parse_lines(lines: list[str]) -> CodeExecutionResult:
def _try_parse(stripped: str):
try:
return json.loads(stripped)
return _MESSAGE.validate_python(json.loads(stripped))
except json.JSONDecodeError:
return None
@ -178,10 +188,12 @@ class E2BSandboxConfig(BaseSandboxConfig):
None,
)
return CodeExecutionResult(
stdout="".join(m.get("text", "") for m in of_type("stdout")),
stderr="".join(m.get("text", "") for m in of_type("stderr")),
results=[{k: v for k, v in m.items() if k != "type"} for m in of_type("result")],
error=error,
execution_count=execution_count,
return CodeExecutionResult.model_validate(
{
"stdout": "".join(_TEXT.validate_python(m.get("text", "")) for m in of_type("stdout")),
"stderr": "".join(_TEXT.validate_python(m.get("text", "")) for m in of_type("stderr")),
"results": [{k: v for k, v in m.items() if k != "type"} for m in of_type("result")],
"error": error,
"execution_count": execution_count,
}
)

14
litellm/llms/elevenlabs/audio_transcription/transformation.py
View file

@ -2,9 +2,11 @@
Translates from OpenAI's `/v1/audio/transcriptions` to ElevenLabs's `/v1/speech-to-text`
"""
from collections.abc import Iterable, Mapping
from typing import Final
from httpx import Headers, Response
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
@ -22,6 +24,9 @@ from ...base_llm.audio_transcription.transformation import (
)
from ..common_utils import ElevenLabsException
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
@property
@ -115,21 +120,22 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
"""
try:
response_json: Final = raw_response.json()
response_object: Final = _JSON_OBJECT.validate_python(response_json)
# Extract the main transcript text
text: Final = response_json.get("text", "")
text: Final = response_object.get("text", "")
# Create TranscriptionResponse object
response: Final = TranscriptionResponse(text=text)
# Add additional metadata matching OpenAI format
response["task"] = "transcribe"
response["language"] = response_json.get("language_code", "unknown")
response["language"] = response_object.get("language_code", "unknown")
# Map ElevenLabs words to OpenAI format
if "words" in response_json:
if "words" in response_object:
response["words"] = []
for word_data in response_json["words"]:
for word_data in _JSON_OBJECTS.validate_python(response_object["words"]):
# Only include actual words, skip spacing and audio events
if word_data.get("type") == "word":
response["words"].append(

19
litellm/llms/fal_ai/image_generation/flux_pro_v11_ultra_transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
from litellm.types.utils import ImageResponse
@ -15,6 +17,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class FalAIFluxProV11UltraConfig(FalAIBaseConfig):
"""
@ -228,16 +232,17 @@ class FalAIFluxProV11UltraConfig(FalAIBaseConfig):
if not model_response.data:
model_response.data = []
images: Final = response_data.get("images", [])
response_object: Final = _JSON_OBJECT.validate_python(response_data)
images: Final = response_object.get("images", [])
model_response.data.extend(fal_images_to_image_objects(images))
# Add additional metadata from Flux Pro response
if hasattr(model_response, "_hidden_params"):
if "seed" in response_data:
model_response._hidden_params["seed"] = response_data["seed"]
if "timings" in response_data:
model_response._hidden_params["timings"] = response_data["timings"]
if "has_nsfw_concepts" in response_data:
model_response._hidden_params["has_nsfw_concepts"] = response_data["has_nsfw_concepts"]
if "seed" in response_object:
model_response._hidden_params["seed"] = response_object["seed"]
if "timings" in response_object:
model_response._hidden_params["timings"] = response_object["timings"]
if "has_nsfw_concepts" in response_object:
model_response._hidden_params["has_nsfw_concepts"] = response_object["has_nsfw_concepts"]
return model_response

11
litellm/llms/fal_ai/image_generation/ideogram_v3_transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
from litellm.types.utils import ImageObject, ImageResponse
@ -15,6 +17,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class FalAIIdeogramV3Config(FalAIBaseConfig):
"""
@ -169,7 +173,8 @@ class FalAIIdeogramV3Config(FalAIBaseConfig):
if not model_response.data:
model_response.data = []
images: Final = response_data.get("images", [])
response_object: Final = _JSON_OBJECT.validate_python(response_data)
images: Final = response_object.get("images", [])
if isinstance(images, list):
for image_entry in images:
if isinstance(image_entry, dict):
@ -184,7 +189,7 @@ class FalAIIdeogramV3Config(FalAIBaseConfig):
)
)
if hasattr(model_response, "_hidden_params") and "seed" in response_data:
model_response._hidden_params["seed"] = response_data["seed"]
if hasattr(model_response, "_hidden_params") and "seed" in response_object:
model_response._hidden_params["seed"] = response_object["seed"]
return model_response

8
litellm/llms/fal_ai/image_generation/recraft_v3_transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
from litellm.types.utils import ImageObject, ImageResponse
@ -15,6 +17,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class FalAIRecraftV3Config(FalAIBaseConfig):
"""
@ -89,7 +93,7 @@ class FalAIRecraftV3Config(FalAIBaseConfig):
return optional_params
def _map_image_size(self, size: str) -> Any:
def _map_image_size(self, size: str) -> str | Mapping[str, int]:
"""
Map OpenAI size format to Recraft v3 image_size format.
@ -203,7 +207,7 @@ class FalAIRecraftV3Config(FalAIBaseConfig):
model_response.data = []
# Handle Recraft v3 response format
images: Final = response_data.get("images", [])
images: Final = _JSON_OBJECT.validate_python(response_data).get("images", [])
if isinstance(images, list):
for image_data in images:
if isinstance(image_data, dict):

21
litellm/llms/fal_ai/image_generation/stable_diffusion_transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
from litellm.types.utils import ImageObject, ImageResponse
@ -15,6 +17,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class FalAIStableDiffusionConfig(FalAIBaseConfig):
"""
@ -124,7 +128,7 @@ class FalAIStableDiffusionConfig(FalAIBaseConfig):
return optional_params
def _map_image_size(self, size: str) -> Any:
def _map_image_size(self, size: str) -> str | Mapping[str, int]:
"""
Map OpenAI size format to Stable Diffusion image_size format.
@ -243,7 +247,8 @@ class FalAIStableDiffusionConfig(FalAIBaseConfig):
model_response.data = []
# Handle Stable Diffusion response format
images: Final = response_data.get("images", [])
response_object: Final = _JSON_OBJECT.validate_python(response_data)
images: Final = response_object.get("images", [])
if isinstance(images, list):
for image_data in images:
if isinstance(image_data, dict):
@ -264,11 +269,11 @@ class FalAIStableDiffusionConfig(FalAIBaseConfig):
# Add additional metadata from Stable Diffusion response
if hasattr(model_response, "_hidden_params"):
if "seed" in response_data:
model_response._hidden_params["seed"] = response_data["seed"]
if "timings" in response_data:
model_response._hidden_params["timings"] = response_data["timings"]
if "has_nsfw_concepts" in response_data:
model_response._hidden_params["has_nsfw_concepts"] = response_data["has_nsfw_concepts"]
if "seed" in response_object:
model_response._hidden_params["seed"] = response_object["seed"]
if "timings" in response_object:
model_response._hidden_params["timings"] = response_object["timings"]
if "has_nsfw_concepts" in response_object:
model_response._hidden_params["has_nsfw_concepts"] = response_object["has_nsfw_concepts"]
return model_response

55
litellm/llms/fireworks_ai/rerank/transformation.py
View file

@ -4,10 +4,11 @@ Fireworks AI Rerank API transformation
Reference: https://docs.fireworks.ai/inference-api-reference/rerank
"""
from collections.abc import Mapping
from typing import Any, Final
from collections.abc import Iterable, Mapping, Sequence
from typing import Final
import httpx
from pydantic import BaseModel, ConfigDict, TypeAdapter
from litellm._uuid import uuid
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -23,6 +24,26 @@ from litellm.types.rerank import (
)
class _FireworksAIUsageFields(BaseModel):
model_config = ConfigDict(extra="ignore", frozen=True)
total_tokens: int | None = 0
prompt_tokens: int | None = 0
completion_tokens: int | None = 0
class _FireworksAIResultFields(BaseModel):
model_config = ConfigDict(extra="ignore", frozen=True, hide_input_in_errors=True)
index: int | float | str
relevance_score: int | float | str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_STR: Final = TypeAdapter(str)
class FireworksAIRerankConfig(FireworksAIMixin, BaseRerankConfig):
"""
Fireworks AI Rerank API configuration
@ -59,7 +80,7 @@ class FireworksAIRerankConfig(FireworksAIMixin, BaseRerankConfig):
model: str,
drop_params: bool,
query: str,
documents: list[str | dict[str, Any]],
documents: Sequence[str | Mapping[str, object]],
custom_llm_provider: str | None = None,
top_n: int | None = None,
rank_fields: list[str] | None = None,
@ -204,23 +225,18 @@ class FireworksAIRerankConfig(FireworksAIMixin, BaseRerankConfig):
# }
# Extract usage information
usage: Final = raw_response_json.get("usage", {})
_billed_units: Final = RerankBilledUnits(search_units=usage.get("total_tokens", 0))
_tokens: Final = RerankTokens(
input_tokens=usage.get("prompt_tokens", 0),
output_tokens=usage.get("completion_tokens", 0),
)
rerank_meta: Final = RerankResponseMeta(billed_units=_billed_units, tokens=_tokens)
response_json: Final = _JSON_OBJECT.validate_python(raw_response_json)
usage: Final = _JSON_OBJECT.validate_python(response_json.get("usage", {}))
# Extract results - Fireworks AI uses "data" instead of "results"
_results: Final[list[dict] | None] = raw_response_json.get("data") or raw_response_json.get("results")
_results: Final = response_json.get("data") or response_json.get("results")
if _results is None:
raise ValueError(f"No results found in the response={raw_response_json}")
rerank_results: Final[list[RerankResponseResult]] = []
for result in _results:
for result in _JSON_OBJECTS.validate_python(_results):
# Validate required fields exist
if not all(key in result for key in ["index", "relevance_score"]):
raise ValueError(f"Missing required fields in the result={result}")
@ -239,9 +255,10 @@ class FireworksAIRerankConfig(FireworksAIMixin, BaseRerankConfig):
document = RerankResponseDocument(text=str(text))
# Create typed result
fields = _FireworksAIResultFields.model_validate(result)
rerank_result = RerankResponseResult(
index=int(result["index"]),
relevance_score=float(result["relevance_score"]),
index=int(fields.index),
relevance_score=float(fields.relevance_score),
)
# Only add document if it exists
@ -250,7 +267,15 @@ class FireworksAIRerankConfig(FireworksAIMixin, BaseRerankConfig):
rerank_results.append(rerank_result)
response_id: Final = raw_response_json.get("id") or str(uuid.uuid4())
usage_fields: Final = _FireworksAIUsageFields.model_validate(usage)
_billed_units: Final = RerankBilledUnits(search_units=usage_fields.total_tokens)
_tokens: Final = RerankTokens(
input_tokens=usage_fields.prompt_tokens,
output_tokens=usage_fields.completion_tokens,
)
rerank_meta: Final = RerankResponseMeta(billed_units=_billed_units, tokens=_tokens)
response_id: Final = _STR.validate_python(response_json.get("id") or str(uuid.uuid4()))
return RerankResponse(
id=response_id,

11
litellm/llms/gemini/image_generation/transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
@ -31,6 +33,9 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_JSON_DICT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
class GoogleImageGenConfig(BaseImageGenerationConfig):
DEFAULT_BASE_URL: str = "https://generativelanguage.googleapis.com/v1beta"
@ -216,13 +221,15 @@ class GoogleImageGenConfig(BaseImageGenerationConfig):
# Extract usage metadata for Gemini models
if "usageMetadata" in response_data:
model_response.usage = transform_gemini_image_usage(response_data["usageMetadata"])
model_response.usage = transform_gemini_image_usage(
_JSON_DICT.validate_python(response_data["usageMetadata"])
)
web_search_requests: Final = get_gemini_image_web_search_requests(response_data)
if web_search_requests and model_response.usage is not None:
setattr(model_response.usage, "web_search_requests", web_search_requests)
else:
# Original Imagen format - predictions with generated images
predictions: Final = response_data.get("predictions", [])
predictions: Final = _JSON_OBJECTS.validate_python(response_data.get("predictions", []))
for prediction in predictions:
# Google AI returns base64 encoded images in the prediction
model_response.data.append(

6
litellm/llms/gigachat/embedding/transformation.py
View file

@ -8,9 +8,11 @@ API Documentation: https://developers.sber.ru/docs/ru/gigachat/api/reference/res
from __future__ import annotations
import types
from collections.abc import Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm import LlmProviders
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -22,6 +24,8 @@ from litellm.types.utils import EmbeddingResponse
from ..authenticator import get_access_token
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class GigaChatEmbeddingError(BaseLLMException):
"""GigaChat Embedding API error."""
@ -165,7 +169,7 @@ class GigaChatEmbeddingConfig(BaseEmbeddingConfig):
"total_tokens": total_tokens,
}
return EmbeddingResponse(**response_json)
return EmbeddingResponse.model_validate(_JSON_OBJECT.validate_python(response_json))
def validate_environment(
self,

5
litellm/llms/github_copilot/embedding/transformation.py
View file

@ -11,6 +11,7 @@ import os
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm._logging import verbose_logger
from litellm.exceptions import AuthenticationError
@ -33,6 +34,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_RESPONSE_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
class GithubCopilotEmbeddingConfig(BaseEmbeddingConfig):
"""
@ -154,7 +157,7 @@ class GithubCopilotEmbeddingConfig(BaseEmbeddingConfig):
logging_obj.post_call(original_response=raw_response.text)
# GitHub Copilot returns standard OpenAI-compatible embedding response
response_json: Final = raw_response.json()
response_json: Final = _RESPONSE_OBJECT.validate_python(raw_response.json())
return convert_to_model_response_object(
response_object=response_json,

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

@ -7,9 +7,11 @@ Docs - https://jina.ai/embeddings/
"""
import types
from collections.abc import Mapping
from typing import Final, cast
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm import LlmProviders
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -22,6 +24,8 @@ from litellm.utils import is_base64_encoded
from ..common_utils import JinaAIError
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class JinaAIEmbeddingConfig(BaseEmbeddingConfig):
"""
@ -139,7 +143,7 @@ class JinaAIEmbeddingConfig(BaseEmbeddingConfig):
additional_args={"complete_input_dict": request_data},
original_response=response_json,
)
return EmbeddingResponse(**response_json)
return EmbeddingResponse.model_validate(_JSON_OBJECT.validate_python(response_json))
def validate_environment(
self,

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

@ -6,10 +6,11 @@ Why separate file? Make it easy to see how transformation works
Docs - https://jina.ai/reranker
"""
from collections.abc import Mapping, Sequence
from collections.abc import Iterable, Mapping, Sequence
from typing import Final
from httpx import URL, Response
from pydantic import ConfigDict, TypeAdapter
from litellm._uuid import uuid
from litellm.llms.base_llm.chat.transformation import LiteLLMLoggingObj
@ -23,6 +24,12 @@ from litellm.types.rerank import (
)
from litellm.types.utils import ModelInfo
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_BILLED_UNITS: Final = TypeAdapter(RerankBilledUnits)
_TOKENS: Final = TypeAdapter(RerankTokens)
_STR: Final = TypeAdapter(str)
class JinaAIRerankConfig(BaseRerankConfig):
def get_supported_cohere_rerank_params(self, model: str) -> list:
@ -100,13 +107,11 @@ class JinaAIRerankConfig(BaseRerankConfig):
logging_obj.post_call(original_response=raw_response.text)
_json_response: Final = raw_response.json()
_json_response: Final = _JSON_OBJECT.validate_python(raw_response.json())
_billed_units: Final = RerankBilledUnits(**_json_response.get("usage", {}))
_tokens: Final = RerankTokens(**_json_response.get("usage", {}))
rerank_meta: Final = RerankResponseMeta(billed_units=_billed_units, tokens=_tokens)
usage: Final = _JSON_OBJECT.validate_python(_json_response.get("usage", {}))
_results: Final[list[dict] | None] = _json_response.get("results")
_results: Final = _json_response.get("results")
if _results is None:
raise ValueError(f"No results found in the response={_json_response}")
@ -115,7 +120,7 @@ class JinaAIRerankConfig(BaseRerankConfig):
# Jina AI returns: {"index": 0, "relevance_score": 0.72, "document": "hello"}
# LiteLLM expects: {"index": 0, "relevance_score": 0.72, "document": {"text": "hello"}}
transformed_results: Final = []
for result in _results:
for result in _JSON_OBJECTS.validate_python(_results):
transformed_result = {
"index": result["index"],
"relevance_score": result["relevance_score"],
@ -128,8 +133,12 @@ class JinaAIRerankConfig(BaseRerankConfig):
transformed_result["document"] = result["document"]
transformed_results.append(transformed_result)
_billed_units: Final = _BILLED_UNITS.validate_python(usage)
_tokens: Final = _TOKENS.validate_python(usage)
rerank_meta: Final = RerankResponseMeta(billed_units=_billed_units, tokens=_tokens)
return RerankResponse(
id=_json_response.get("id") or str(uuid.uuid4()),
id=_STR.validate_python(_json_response.get("id") or str(uuid.uuid4())),
results=transformed_results,
meta=rerank_meta,
) # Return response

40
litellm/llms/manus/files/transformation.py
View file

@ -11,10 +11,12 @@ Reference: https://open.manus.im/docs/openai-compatibility#file-management
"""
import time
from typing import Any, Final
from collections.abc import Iterable, Mapping
from typing import Final
import httpx
from openai.types.file_deleted import FileDeleted
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm._logging import verbose_logger
@ -39,6 +41,10 @@ from litellm.types.utils import LlmProviders
MANUS_API_BASE: Final = "https://api.manus.im"
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(strict=True, hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_TEXT: Final = TypeAdapter(str, config=ConfigDict(strict=True, hide_input_in_errors=True))
class ManusFilesConfig(BaseFilesConfig):
"""
@ -337,8 +343,7 @@ class ManusFilesConfig(BaseFilesConfig):
litellm_params: dict,
) -> FileDeleted:
"""Transform delete file response."""
response_json: Final = raw_response.json()
return FileDeleted(**response_json)
return FileDeleted.model_validate(_JSON_OBJECT.validate_python(raw_response.json()))
def transform_list_files_request(
self,
@ -366,19 +371,20 @@ class ManusFilesConfig(BaseFilesConfig):
litellm_params: dict,
) -> list[OpenAIFileObject]:
"""Transform list files response."""
response_json: Final = raw_response.json()
files_data: Final = response_json.get("data", [])
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
files_data: Final = _JSON_OBJECTS.validate_python(response_json.get("data", []))
return [self._parse_file_dict(f) for f in files_data]
def _parse_file_dict(self, file_dict: dict[str, Any]) -> OpenAIFileObject:
def _parse_file_dict(self, file_dict: Mapping[str, object]) -> OpenAIFileObject:
"""Parse a file dict into OpenAIFileObject."""
created_at_str: Final = file_dict.get("created_at", "")
if created_at_str:
created_at_text: Final = _TEXT.validate_python(created_at_str)
try:
created_at = int(
time.mktime(
time.strptime(
created_at_str.replace("Z", "+00:00")[:19],
created_at_text.replace("Z", "+00:00")[:19],
"%Y-%m-%dT%H:%M:%S",
)
)
@ -388,15 +394,17 @@ class ManusFilesConfig(BaseFilesConfig):
else:
created_at = int(time.time())
return OpenAIFileObject(
id=file_dict.get("id", ""),
bytes=file_dict.get("bytes", 0),
created_at=created_at,
filename=file_dict.get("filename", ""),
object="file",
purpose=file_dict.get("purpose", "assistants"),
status=file_dict.get("status", "uploaded"),
status_details=file_dict.get("status_details"),
return OpenAIFileObject.model_validate(
{
"id": file_dict.get("id", ""),
"bytes": file_dict.get("bytes", 0),
"created_at": created_at,
"filename": file_dict.get("filename", ""),
"object": "file",
"purpose": file_dict.get("purpose", "assistants"),
"status": file_dict.get("status", "uploaded"),
"status_details": file_dict.get("status_details"),
}
)
def transform_file_content_request(

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

@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from httpx import Headers
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.llms.base_llm.chat.transformation import BaseLLMException
@ -26,6 +27,9 @@ else:
LiteLLMLoggingObj = Any
HttpxBinaryResponseContent = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_STR: Final = TypeAdapter(str, config=ConfigDict(strict=True, hide_input_in_errors=True))
class MinimaxException(BaseLLMException):
"""Custom exception for MiniMax API errors"""
@ -299,7 +303,7 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
try:
# Parse JSON response
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# MiniMax API response format check
# The API can return different structures:
@ -320,7 +324,7 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
# Extract audio data
# MiniMax returns audio in "data" field
data: Final = response_json.get("data", {})
data: Final = _JSON_OBJECT.validate_python(response_json.get("data", {}))
# Check if response contains a URL (output_format='url')
audio_url: Final = data.get("audio_url", None)
@ -334,7 +338,7 @@ class MinimaxTextToSpeechConfig(BaseTextToSpeechConfig):
)
# Get hex-encoded audio data
audio_hex: Final = data.get("audio", "") or response_json.get("audio_file", "")
audio_hex: Final = _STR.validate_python(data.get("audio", "") or response_json.get("audio_file", "") or "")
if not audio_hex:
raise MinimaxException(

13
litellm/llms/modelscope/image_generation/transformation.py
View file

@ -6,9 +6,11 @@ Handles transformation between OpenAI-compatible format and ModelScope API forma
API Reference: https://modelscope.cn/docs/model-service/API-Inference/intro
"""
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from typing_extensions import override
from litellm.llms.base_llm.chat.transformation import BaseLLMException
@ -29,6 +31,9 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = object
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
class ModelScopeImageGenerationConfig(BaseImageGenerationConfig):
"""
@ -177,8 +182,10 @@ class ModelScopeImageGenerationConfig(BaseImageGenerationConfig):
)
# Check for errors in response
if "error" in response_data:
error_msg: Final = response_data["error"].get("message", str(response_data["error"]))
response_object: Final = _JSON_OBJECT.validate_python(response_data)
if "error" in response_object:
error: Final = _JSON_OBJECT.validate_python(response_object["error"])
error_msg: Final = error.get("message", str(error))
raise self.get_error_class(
error_message=f"ModelScope error: {error_msg}",
status_code=raw_response.status_code,
@ -186,7 +193,7 @@ class ModelScopeImageGenerationConfig(BaseImageGenerationConfig):
)
# Extract images from response
data_list: Final = response_data.get("data", [])
data_list: Final = _JSON_OBJECTS.validate_python(response_object.get("data", []))
if not model_response.data:
model_response.data = []

26
litellm/llms/nvidia_nim/rerank/transformation.py
View file

@ -1,7 +1,8 @@
from collections.abc import Mapping
from typing import Any, Final, Literal
from typing import Final, Literal
import httpx
from pydantic import ConfigDict, TypeAdapter
from typing_extensions import Required, TypedDict
import litellm
@ -44,6 +45,14 @@ class NvidiaNimRerankResponse(TypedDict):
rankings: Required[list[NvidiaNimRankingResult]]
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_NUMBER: Final[TypeAdapter[bool | int | float]] = TypeAdapter(
bool | int | float, config=ConfigDict(strict=True, hide_input_in_errors=True)
)
_BILLED_UNITS: Final = TypeAdapter(RerankBilledUnits)
_STR: Final = TypeAdapter(str)
class NvidiaNimRerankConfig(BaseRerankConfig):
"""
Reference: https://docs.api.nvidia.com/nim/reference/nvidia-llama-3_2-nv-rerankqa-1b-v2-infer
@ -115,7 +124,7 @@ class NvidiaNimRerankConfig(BaseRerankConfig):
model: str,
drop_params: bool,
query: str,
documents: list[str | dict[str, Any]],
documents: list[str | dict[str, object]],
custom_llm_provider: str | None = None,
top_n: int | None = None,
rank_fields: list[str] | None = None,
@ -133,7 +142,7 @@ class NvidiaNimRerankConfig(BaseRerankConfig):
Nvidia NIM specific params (passed through as-is from non_default_params):
- truncate: How to truncate input if too long (NONE, END)
"""
optional_nvidia_nim_rerank_params: Final[dict[str, Any]] = {
optional_nvidia_nim_rerank_params: Final[dict[str, object]] = {
"query": query,
"documents": documents,
}
@ -327,15 +336,18 @@ class NvidiaNimRerankConfig(BaseRerankConfig):
# Construct metadata with billed_units
# Nvidia NIM uses "usage" field with "total_tokens"
usage: Final = raw_response_json.get("usage", {})
total_tokens: Final = usage.get("total_tokens", 0)
payload: Final = _JSON_OBJECT.validate_python(raw_response_json)
usage: Final = _JSON_OBJECT.validate_python(payload.get("usage", {}))
total_tokens: Final = _NUMBER.validate_python(usage.get("total_tokens", 0))
billed_units: Final[RerankBilledUnits] = {"total_tokens": total_tokens if total_tokens > 0 else len(results)}
billed_units: Final = _BILLED_UNITS.validate_python(
{"total_tokens": total_tokens if total_tokens > 0 else len(results)}
)
meta: Final[RerankResponseMeta] = {"billed_units": billed_units}
return RerankResponse(
id=raw_response_json.get("id") or str(uuid.uuid4()),
id=_STR.validate_python(payload.get("id") or str(uuid.uuid4())),
results=results,
meta=meta,
)

10
litellm/llms/nvidia_riva/audio_transcription/transformation.py
View file

@ -102,7 +102,7 @@ class NvidiaRivaAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
if endpointing_config is not None:
recognition_config["endpointing_config"] = endpointing_config
request_payload: Final[dict[str, Any]] = {
request_payload: Final[dict[str, object]] = {
"recognition_config": recognition_config,
"response_format": optional_params.get("response_format") or "json",
"timestamp_granularities": optional_params.get("timestamp_granularities"),
@ -135,7 +135,7 @@ class NvidiaRivaAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
# gRPC auth is constructed in the handler, not via HTTP headers.
return headers
def _build_recognition_config_dict(self, model: str, optional_params: dict) -> dict[str, Any]:
def _build_recognition_config_dict(self, model: str, optional_params: dict) -> dict[str, object]:
"""
Build the Riva ``RecognitionConfig`` shape as a plain dict.
@ -159,7 +159,7 @@ class NvidiaRivaAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
"profanity_filter": optional_params.get("profanity_filter", False),
}
def _build_endpointing_config_dict(self, optional_params: dict) -> dict[str, Any] | None:
def _build_endpointing_config_dict(self, optional_params: dict) -> dict[str, object] | None:
"""
Translate an OpenAI-style ``chunking_strategy`` into Riva's
``EndpointingConfig`` shape, or pass through an explicit
@ -177,7 +177,7 @@ class NvidiaRivaAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
return None
if isinstance(chunking, dict) and chunking.get("type") == "server_vad":
config: Final[dict[str, Any]] = {}
config: Final[dict[str, object]] = {}
if "threshold" in chunking:
threshold: Final = float(chunking["threshold"])
config["start_threshold"] = threshold
@ -245,7 +245,7 @@ class NvidiaRivaAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
response["task"] = "transcribe"
if response_format == "verbose_json":
words: Final[list[dict[str, Any]]] = []
words: Final[list[dict[str, object]]] = []
if timestamp_granularities and "word" in timestamp_granularities:
for item in final_results:
for word in item.get("words", []) or []:

6
litellm/llms/oci/embed/transformation.py
View file

@ -22,9 +22,11 @@ Supported models:
Reference: https://docs.oracle.com/en-us/iaas/api/#/en/generative-ai-inference/latest/EmbedTextResult/EmbedText
"""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.llms.base_llm.chat.transformation import BaseLLMException
@ -52,6 +54,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
# OCI sends up to 96 texts per embedText request (Cohere limit).
OCI_EMBED_BATCH_LIMIT: Final = 96
@ -275,7 +279,7 @@ class OCIEmbedConfig(BaseEmbeddingConfig):
)
try:
parsed: Final = OCIEmbedResponse(**json_response)
parsed: Final = OCIEmbedResponse.model_validate(_JSON_OBJECT.validate_python(json_response))
except Exception as e:
raise OCIError(
status_code=500,

5
litellm/llms/openai/responses/count_tokens/handler.py
View file

@ -5,6 +5,7 @@ Uses httpx for HTTP requests to OpenAI's /v1/responses/input_tokens endpoint.
"""
import json
from collections.abc import Sequence
from typing import Any, Final
import httpx
@ -26,11 +27,11 @@ class OpenAICountTokensHandler(OpenAICountTokensConfig):
async def handle_count_tokens_request(
self,
model: str,
input: str | list[Any],
input: str | Sequence[object],
api_key: str,
api_base: str | None = None,
timeout: float | httpx.Timeout | None = None,
tools: list[dict[str, Any]] | None = None,
tools: list[dict[str, object]] | None = None,
instructions: str | None = None,
) -> dict[str, Any]:
"""

8
litellm/llms/opencode/harness/transformation.py
View file

@ -165,7 +165,7 @@ def _as_dict(value: object) -> Mapping[str, Any]:
return value if isinstance(value, dict) else MappingProxyType({})
def _tool_events(part: Mapping[str, Any]) -> Sequence[Event]:
def _tool_events(part: Mapping[str, object]) -> Sequence[Event]:
native = str(part.get("tool") or "")
call_id = str(part.get("callID") or part.get("id") or "")
state = _as_dict(part.get("state"))
@ -195,7 +195,7 @@ def _error_message(error: object) -> str:
return str(error.get("name") or "opencode reported an error")
def validate_user_config(config: Mapping[str, Any]) -> None:
def validate_user_config(config: Mapping[str, object]) -> None:
"""Reject OpenCodeOptions.config keys LiteLLM manages (or that bypass permissions)."""
for key in config:
if key in MANAGED_CONFIG_KEYS:
@ -241,7 +241,7 @@ def build_opencode_config(
user_config: Mapping[str, Any] | None = None,
instructions_path: str | None = None,
skills_path: str | None = None,
) -> Mapping[str, Any]:
) -> Mapping[str, object]:
"""The full opencode config: user config underneath, LiteLLM-managed keys on top."""
user: Final = user_config or MappingProxyType({})
validate_user_config(user)
@ -379,7 +379,7 @@ class OpenCodeHarnessConfig(BaseCLIHarnessConfig):
def create_stream_state(self) -> OpenCodeStreamState:
return OpenCodeStreamState()
def transform_stream_line(self, line: Mapping[str, Any], state: OpenCodeStreamState) -> Sequence[Event]:
def transform_stream_line(self, line: Mapping[str, object], state: OpenCodeStreamState) -> Sequence[Event]:
"""step_finish token counts are ignored on purpose: the session endpoint accounts usage."""
session_id = line.get("sessionID")
if session_id and state.session_id is None:

20
litellm/llms/openrouter/image_generation/transformation.py
View file

@ -27,9 +27,11 @@ Response format:
}
"""
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.llms.base_llm.chat.transformation import BaseLLMException
@ -55,6 +57,11 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_DICT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_STR: Final = TypeAdapter(str, config=ConfigDict(strict=True, hide_input_in_errors=True))
class OpenRouterImageGenerationConfig(BaseImageGenerationConfig):
"""
@ -355,15 +362,16 @@ class OpenRouterImageGenerationConfig(BaseImageGenerationConfig):
model_response.data = []
try:
choices: Final = response_json.get("choices", [])
response_object: Final = _JSON_DICT.validate_python(response_json)
choices: Final = _JSON_OBJECTS.validate_python(response_object.get("choices", []))
for choice in choices:
message = choice.get("message", {})
images = message.get("images", [])
message = _JSON_OBJECT.validate_python(choice.get("message", {}))
images = _JSON_OBJECTS.validate_python(message.get("images", []))
for image_data in images:
image_url_obj = image_data.get("image_url", {})
image_url = image_url_obj.get("url")
image_url_obj = _JSON_OBJECT.validate_python(image_data.get("image_url", {}))
image_url = _STR.validate_python(image_url_obj.get("url") or "")
if image_url:
if image_url.startswith("data:"):
@ -389,7 +397,7 @@ class OpenRouterImageGenerationConfig(BaseImageGenerationConfig):
)
# Extract and set usage and cost information
self._set_usage_and_cost(model_response, response_json, model)
self._set_usage_and_cost(model_response, response_object, model)
return model_response

11
litellm/llms/ovhcloud/audio_transcription/transformation.py
View file

@ -5,9 +5,11 @@ Our unified API follows the OpenAI standard.
More information on our website: https://endpoints.ai.cloud.ovh.net
"""
from collections.abc import Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
from litellm.llms.base_llm.audio_transcription.transformation import (
@ -24,6 +26,8 @@ from litellm.types.utils import FileTypes, TranscriptionResponse
from ..utils import OVHCloudException
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class OVHCloudAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
def get_supported_openai_params(self, model: str) -> list[OpenAIAudioTranscriptionOptionalParams]:
@ -145,7 +149,8 @@ class OVHCloudAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
headers=raw_response.headers,
)
text: Final = response_json.get("text") or response_json.get("transcript") or ""
payload: Final = _JSON_OBJECT.validate_python(response_json)
text: Final = payload.get("text") or payload.get("transcript") or ""
response: Final = TranscriptionResponse(text=text)
# OVHCloud field migration (deadline: 2026-05-11):
@ -153,9 +158,7 @@ class OVHCloudAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
# Prefer `seconds`, fall back to `duration`, normalize to `duration`
# so downstream consumers see a consistent key.
duration: Final = (
response_json["seconds"]
if "seconds" in response_json and response_json["seconds"] is not None
else response_json.get("duration")
payload["seconds"] if "seconds" in payload and payload["seconds"] is not None else payload.get("duration")
)
if duration is not None:
response_json["duration"] = duration

8
litellm/llms/recraft/image_generation/transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
@ -21,6 +23,9 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
class RecraftImageGenerationConfig(BaseImageGenerationConfig):
DEFAULT_BASE_URL: str = "https://external.api.recraft.ai"
@ -141,7 +146,8 @@ class RecraftImageGenerationConfig(BaseImageGenerationConfig):
if not model_response.data:
model_response.data = []
for image_data in response_data["data"]:
payload: Final = _JSON_OBJECT.validate_python(response_data)
for image_data in _JSON_OBJECTS.validate_python(payload["data"]):
model_response.data.append(
ImageObject(
url=image_data.get("url", None),

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

@ -1,6 +1,8 @@
from collections.abc import Iterable
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.constants import REPLICATE_MODEL_NAME_WITH_ID_LENGTH
@ -26,6 +28,8 @@ if TYPE_CHECKING:
else:
LoggingClass = Any
_TEXTS: Final = TypeAdapter(Iterable[str], config=ConfigDict(strict=True, hide_input_in_errors=True))
class ReplicateConfig(BaseConfig):
"""
@ -253,7 +257,7 @@ class ReplicateConfig(BaseConfig):
message=f"LiteLLM Error - prediction not succeeded - {raw_response_json}",
headers=raw_response.headers,
)
outputs: Final = raw_response_json.get("output", [])
outputs: Final = _TEXTS.validate_python(raw_response_json.get("output", []))
response_str = "".join(outputs)
if len(response_str) == 0: # edge case, where result from replicate is empty
response_str = " "

7
litellm/llms/sagemaker/common_utils.py
View file

@ -1,9 +1,10 @@
import functools
import json
from collections.abc import AsyncIterator, Iterator
from collections.abc import AsyncIterator, Iterator, Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm import verbose_logger
@ -11,6 +12,8 @@ from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.utils import GenericStreamingChunk as GChunk
from litellm.types.utils import StreamingChatCompletionChunk
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
def _load_sagemaker_response_stream_shape():
try:
@ -18,7 +21,7 @@ def _load_sagemaker_response_stream_shape():
from botocore.model import ServiceModel
loader: Final = Loader()
service_dict: Final = loader.load_service_model("sagemaker-runtime", "service-2")
service_dict: Final = _JSON_OBJECT.validate_python(loader.load_service_model("sagemaker-runtime", "service-2"))
return ServiceModel(service_dict).shape_for("InvokeEndpointWithResponseStreamOutput")
except Exception as e:
verbose_logger.warning(

6
litellm/llms/snowflake/embedding/transformation.py
View file

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.chat.transformation import BaseLLMException
@ -10,6 +12,8 @@ from litellm.types.utils import EmbeddingResponse
from ..utils import SnowflakeBaseConfig, SnowflakeException
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class SnowflakeEmbeddingConfig(SnowflakeBaseConfig, BaseEmbeddingConfig):
"""
@ -53,7 +57,7 @@ class SnowflakeEmbeddingConfig(SnowflakeBaseConfig, BaseEmbeddingConfig):
# convert embeddings to 1d array
for item in response_json["data"]:
item["embedding"] = item["embedding"][0]
returned_response: Final = EmbeddingResponse(**response_json)
returned_response: Final = EmbeddingResponse.model_validate(_JSON_OBJECT.validate_python(response_json))
returned_response.model = "snowflake/" + (returned_response.model or "")

9
litellm/llms/stability/image_generation/transformation.py
View file

@ -6,9 +6,11 @@ Handles transformation between OpenAI-compatible format and Stability AI API for
API Reference: https://platform.stability.ai/docs/api-reference
"""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
@ -33,6 +35,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class StabilityImageGenerationConfig(BaseImageGenerationConfig):
"""
@ -234,7 +238,8 @@ class StabilityImageGenerationConfig(BaseImageGenerationConfig):
)
# Check finish_reason
finish_reason: Final = response_data.get("finish_reason", "")
payload: Final = _JSON_OBJECT.validate_python(response_data)
finish_reason: Final = payload.get("finish_reason", "")
if finish_reason == "CONTENT_FILTERED":
raise self.get_error_class(
error_message="Content was filtered by Stability AI safety systems",
@ -246,7 +251,7 @@ class StabilityImageGenerationConfig(BaseImageGenerationConfig):
model_response.data = []
# Extract image from response
image_b64: Final = response_data.get("image")
image_b64: Final = payload.get("image")
if image_b64:
model_response.data.append(
ImageObject(

5
litellm/llms/tinyfish/search/transformation.py
View file

@ -26,6 +26,7 @@ from litellm.secret_managers.main import get_secret_str
_UrlEncodableParams: Final = TypeAdapter(dict[str, str | int | float | bool])
_StrList: Final = TypeAdapter(list[str])
_StrFrozenSet: Final = TypeAdapter(frozenset[str])
_DecodedJson: Final = TypeAdapter(object)
_TINYFISH_PARAMS_KEY: Final = "_tinyfish_params"
_TINYFISH_DOCS_URL: Final = "https://docs.tinyfish.ai/search-api"
@ -218,7 +219,7 @@ class TinyfishSearchConfig(BaseSearchConfig):
)
try:
raw_json: Final[object] = raw_response.json() # any-ok: httpx Response.json() -> Any
raw_json: Final = _DecodedJson.validate_python(raw_response.json())
except json.JSONDecodeError:
raise self._wrap_error(
error_message=f"Expected JSON response, got: {raw_response.text[:200]}",
@ -276,7 +277,7 @@ class TinyfishSearchConfig(BaseSearchConfig):
# for other envelope shapes (CDN HTML pages, other JSON envelopes, plain text).
inner_message = error_message
try:
body: Final[object] = json.loads(error_message) # any-ok: json.loads -> Any
body: Final = _DecodedJson.validate_python(json.loads(error_message))
if isinstance(body, dict):
error_obj: Final[object] = body.get("error") # any-ok: untyped dict
if isinstance(error_obj, dict):

14
litellm/llms/together_ai/rerank/handler.py
View file

@ -4,7 +4,9 @@ Re rank api
LiteLLM supports the re rank API format, no paramter transformation occurs
"""
from typing import Any, Final
from typing import Final
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.llms.base import BaseLLM
@ -15,6 +17,8 @@ from litellm.llms.custom_httpx.http_handler import (
from litellm.llms.together_ai.rerank.transformation import TogetherAIRerankConfig
from litellm.types.rerank import RerankRequest, RerankResponse
_JSON_DICT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
def _rerank_url(api_base: str) -> str:
return f"{api_base.rstrip('/')}/rerank"
@ -27,7 +31,7 @@ class TogetherAIRerank(BaseLLM):
api_key: str,
api_base: str,
query: str,
documents: list[str | dict[str, Any]],
documents: list[str | dict[str, object]],
top_n: int | None = None,
rank_fields: list[str] | None = None,
return_documents: bool | None = True,
@ -66,13 +70,13 @@ class TogetherAIRerank(BaseLLM):
if response.status_code != 200:
raise Exception(response.text)
_json_response: Final = response.json()
_json_response: Final = _JSON_DICT.validate_python(response.json())
return TogetherAIRerankConfig()._transform_response(_json_response)
async def async_rerank( # New async method
self,
request_data_dict: dict[str, Any],
request_data_dict: dict[str, object],
api_key: str,
api_base: str,
) -> RerankResponse:
@ -91,6 +95,6 @@ class TogetherAIRerank(BaseLLM):
if response.status_code != 200:
raise Exception(response.text)
_json_response: Final = response.json()
_json_response: Final = _JSON_DICT.validate_python(response.json())
return TogetherAIRerankConfig()._transform_response(_json_response)

11
litellm/llms/vertex_ai/image_generation/image_generation_handler.py
View file

@ -1,8 +1,10 @@
import json
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from openai.types.image import Image
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.llms.custom_httpx.http_handler import (
@ -17,11 +19,13 @@ from litellm.types.utils import ImageResponse
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
_PREDICTIONS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
class VertexImageGeneration(VertexLLM):
def process_image_generation_response(
self,
json_response: dict[str, Any],
json_response: Mapping[str, object],
model_response: ImageResponse,
model: str | None = None,
) -> ImageResponse:
@ -32,12 +36,11 @@ class VertexImageGeneration(VertexLLM):
model=model,
)
predictions: Final = json_response["predictions"]
predictions: Final = _PREDICTIONS.validate_python(json_response["predictions"])
response_data: Final[list[Image]] = []
for prediction in predictions:
bytes_base64_encoded = prediction["bytesBase64Encoded"]
image_object = Image(b64_json=bytes_base64_encoded)
image_object = Image.model_validate({"b64_json": prediction["bytesBase64Encoded"]})
response_data.append(image_object)
model_response.data = response_data

12
litellm/llms/vertex_ai/image_generation/vertex_gemini_transformation.py
View file

@ -1,7 +1,9 @@
import os
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm._logging import verbose_logger
@ -31,6 +33,9 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_DICT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class VertexAIGeminiImageGenerationConfig(BaseImageGenerationConfig, VertexLLM):
"""
@ -321,10 +326,11 @@ class VertexAIGeminiImageGenerationConfig(BaseImageGenerationConfig, VertexLLM):
)
)
if usage_metadata := response_data.get("usageMetadata", None):
model_response.usage = self._transform_image_usage(usage_metadata)
response_object: Final = _JSON_OBJECT.validate_python(response_data)
if usage_metadata := response_object.get("usageMetadata", None):
model_response.usage = self._transform_image_usage(_JSON_DICT.validate_python(usage_metadata))
web_search_requests: Final = get_gemini_image_web_search_requests(response_data)
web_search_requests: Final = get_gemini_image_web_search_requests(response_object)
if web_search_requests and model_response.usage is not None:
setattr(model_response.usage, "web_search_requests", web_search_requests)

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

@ -6,11 +6,12 @@ Reference: https://cloud.google.com/text-to-speech/docs/reference/rest/v1/text/s
"""
import base64
from collections.abc import Coroutine
from collections.abc import Coroutine, Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, TypeAlias, Union
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.exceptions import UnsupportedParamsError
@ -45,6 +46,9 @@ else:
_LyriaVoice: TypeAlias = str | dict | None
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_STR: Final = TypeAdapter(str, config=ConfigDict(hide_input_in_errors=True))
class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
"""
@ -465,14 +469,14 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
from litellm.types.llms.openai import HttpxBinaryResponseContent
# Parse JSON response
_json_response: Final = raw_response.json()
_json_response: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Get base64-encoded audio content
response_content: Final = _json_response.get("audioContent")
if not response_content:
raise ValueError("No audioContent in Vertex AI TTS response")
binary_data: Final = base64.b64decode(response_content)
binary_data: Final = base64.b64decode(_STR.validate_python(response_content))
media_type: Final = speech_media_type_from_audio_bytes(binary_data)
response: Final = httpx.Response(
status_code=200,

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

@ -4,10 +4,11 @@ Transformation logic for Voyage AI's /v1/rerank endpoint.
Docs - https://docs.voyageai.com/docs/reranker
"""
from collections.abc import Mapping, Sequence
from collections.abc import Iterable, Mapping, Sequence
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm._uuid import uuid
from litellm.llms.base_llm.chat.transformation import LiteLLMLoggingObj
@ -23,6 +24,11 @@ from litellm.types.utils import ModelInfo
from ..embedding.transformation import VoyageError
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_OPTIONAL_INT: Final[TypeAdapter[int | None]] = TypeAdapter(int | None)
_STR: Final = TypeAdapter(str)
class VoyageRerankConfig(BaseRerankConfig):
def get_supported_cohere_rerank_params(self, model: str) -> list:
@ -103,13 +109,14 @@ class VoyageRerankConfig(BaseRerankConfig):
)
# Voyage AI returns results in "data" key, not "results"
_results: Final[list[dict] | None] = _json_response.get("data")
payload: Final = _JSON_OBJECT.validate_python(_json_response)
_results: Final = payload.get("data")
if _results is None:
raise ValueError(f"No results found in the response={_json_response}")
# Transform to LiteLLM format
transformed_results: Final = []
for result in _results:
for result in _JSON_OBJECTS.validate_python(_results):
transformed_result: dict[str, object] = {
"index": result["index"],
"relevance_score": result["relevance_score"],
@ -121,14 +128,14 @@ class VoyageRerankConfig(BaseRerankConfig):
transformed_result["document"] = result["document"]
transformed_results.append(transformed_result)
usage: Final = _json_response.get("usage", {})
total_tokens: Final = usage.get("total_tokens", 0)
usage: Final = _JSON_OBJECT.validate_python(payload.get("usage", {}))
total_tokens: Final = _OPTIONAL_INT.validate_python(usage.get("total_tokens", 0))
_billed_units: Final = RerankBilledUnits(total_tokens=total_tokens)
_tokens: Final = RerankTokens(input_tokens=total_tokens, output_tokens=0)
rerank_meta: Final = RerankResponseMeta(billed_units=_billed_units, tokens=_tokens)
return RerankResponse(
id=_json_response.get("id") or str(uuid.uuid4()),
id=_STR.validate_python(payload.get("id") or str(uuid.uuid4())),
results=transformed_results,
meta=rerank_meta,
)

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

@ -4,9 +4,11 @@ Translates from OpenAI's `/v1/audio/transcriptions` to IBM WatsonX's `/ml/v1/aud
WatsonX follows the OpenAI spec for audio transcription.
"""
from collections.abc import Mapping
from typing import Final
from httpx import Response
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
@ -25,6 +27,8 @@ from ...openai.transcriptions.whisper_transformation import (
)
from ..common_utils import IBMWatsonXMixin
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class IBMWatsonXAudioTranscriptionConfig(IBMWatsonXMixin, OpenAIWhisperAudioTranscriptionConfig):
"""
@ -174,8 +178,9 @@ class IBMWatsonXAudioTranscriptionConfig(IBMWatsonXMixin, OpenAIWhisperAudioTran
# Extract only valid fields for TranscriptionResponse.__init__()
# TranscriptionResponse only accepts 'text' and 'usage' in __init__()
text: Final = raw_response_json.get("text")
usage: Final = raw_response_json.get("usage")
response_object: Final = _JSON_OBJECT.validate_python(raw_response_json)
text: Final = response_object.get("text")
usage: Final = response_object.get("usage")
# Create response with only valid fields
response_kwargs: Final = {}
@ -187,14 +192,14 @@ class IBMWatsonXAudioTranscriptionConfig(IBMWatsonXMixin, OpenAIWhisperAudioTran
if not response_kwargs:
raise ValueError(
"Invalid response format. Received response does not match the expected format. Got: ",
raw_response_json,
response_object,
)
response: Final = TranscriptionResponse(**response_kwargs)
# Add other fields using dictionary-style assignment (like duration, task, etc.)
# Skip fields that TranscriptionResponse doesn't accept in __init__()
for key, value in raw_response_json.items():
for key, value in response_object.items():
if key not in [
"text",
"usage",

9
litellm/llms/watsonx/embed/transformation.py
View file

@ -2,9 +2,11 @@
Translates from OpenAI's `/v1/embeddings` to IBM's `/text/embeddings` route.
"""
from collections.abc import Iterable, Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.embedding.transformation import (
BaseEmbeddingConfig,
@ -16,6 +18,9 @@ from litellm.types.utils import EmbeddingResponse, Usage
from ..common_utils import IBMWatsonXMixin, _get_api_params
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_TOKEN_COUNT: Final = TypeAdapter(int)
class IBMWatsonXEmbeddingConfig(IBMWatsonXMixin, BaseEmbeddingConfig):
def get_supported_openai_params(self, model: str) -> list:
@ -95,7 +100,7 @@ class IBMWatsonXEmbeddingConfig(IBMWatsonXMixin, BaseEmbeddingConfig):
json_resp: Final = raw_response.json()
if model_response is None:
model_response = EmbeddingResponse(model=json_resp.get("model_id", None))
results: Final = json_resp.get("results", [])
results: Final = _JSON_OBJECTS.validate_python(json_resp.get("results", []))
embedding_response: Final = []
for idx, result in enumerate(results):
embedding_response.append(
@ -107,7 +112,7 @@ class IBMWatsonXEmbeddingConfig(IBMWatsonXMixin, BaseEmbeddingConfig):
)
model_response.object = "list"
model_response.data = embedding_response
input_tokens: Final = json_resp.get("input_token_count", 0)
input_tokens: Final = _TOKEN_COUNT.validate_python(json_resp.get("input_token_count", 0) or 0)
setattr(
model_response,
"usage",

17
litellm/llms/watsonx/rerank/transformation.py
View file

@ -5,10 +5,11 @@ Docs - https://cloud.ibm.com/apidocs/watsonx-ai#text-rerank
"""
import uuid
from collections.abc import Mapping, Sequence
from collections.abc import Iterable, Mapping, Sequence
from typing import Final, cast
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.chat.transformation import LiteLLMLoggingObj
from litellm.llms.base_llm.rerank.transformation import BaseRerankConfig
@ -24,6 +25,11 @@ from litellm.types.rerank import (
from ..common_utils import IBMWatsonXMixin, _generate_watsonx_token, _get_api_params
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_OPTIONAL_INT: Final[TypeAdapter[int | None]] = TypeAdapter(int | None)
_STR: Final = TypeAdapter(str)
class IBMWatsonXRerankConfig(IBMWatsonXMixin, BaseRerankConfig):
"""
@ -171,13 +177,14 @@ class IBMWatsonXRerankConfig(IBMWatsonXMixin, BaseRerankConfig):
headers=raw_response.headers,
)
_results: Final[list[dict] | None] = raw_response_json.get("results")
payload: Final = _JSON_OBJECT.validate_python(raw_response_json)
_results: Final = payload.get("results")
if _results is None:
raise ValueError(f"No results found in the response={raw_response_json}")
transformed_results: Final = []
for result in _results:
for result in _JSON_OBJECTS.validate_python(_results):
transformed_result: dict[str, object] = {
"index": result["index"],
"relevance_score": result["score"],
@ -191,11 +198,11 @@ class IBMWatsonXRerankConfig(IBMWatsonXMixin, BaseRerankConfig):
transformed_results.append(transformed_result)
response_id: Final = raw_response_json.get("id") or str(uuid.uuid4())
response_id: Final = _STR.validate_python(payload.get("id") or str(uuid.uuid4()))
# Extract usage information
_tokens: Final = RerankTokens(
input_tokens=raw_response_json.get("input_token_count", 0),
input_tokens=_OPTIONAL_INT.validate_python(payload.get("input_token_count", 0)),
)
rerank_meta: Final = RerankResponseMeta(tokens=_tokens)

11
litellm/proxy/_experimental/mcp_server/db.py
View file

@ -8,7 +8,7 @@ from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, TypedDict, cast
from fastapi import HTTPException
from pydantic import TypeAdapter
from pydantic import ConfigDict, TypeAdapter
from typing_extensions import ReadOnly
from litellm._logging import verbose_proxy_logger
@ -162,6 +162,8 @@ _CLIENT_FORWARDED_AUTH_TYPES: Final["frozenset[str]"] = frozenset({"true_passthr
# Minted token material that must never survive a client rotation on a persisted row.
_MINTED_TOKEN_CREDENTIAL_FIELDS: Final["frozenset[str]"] = frozenset({"access_token", "refresh_token", "expires_in"})
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
def _bind_submitted_oauth_client(
credentials: Mapping[str, object], issuer: str | None, url: str | None
@ -1916,7 +1918,7 @@ def mcp_oauth_token_identity(server: object) -> tuple[object, ...]:
creds: Final = getattr(server, "credentials", None)
if isinstance(creds, str):
try:
parsed: dict[str, object] | None = json.loads(creds)
parsed: Mapping[str, object] | None = _JSON_OBJECT.validate_python(json.loads(creds))
except ValueError:
parsed = None
else:
@ -2382,13 +2384,10 @@ def _decode_user_env_vars(stored: str) -> dict[str, str]:
"re-enter them rather than silently forwarding ciphertext"
)
return {}
parsed: dict[str, object] | None
try:
parsed = json.loads(decrypted)
parsed: Final = _JSON_OBJECT.validate_python(json.loads(decrypted))
except (ValueError, TypeError):
return {}
if not isinstance(parsed, dict):
return {}
return {str(k): str(v) for k, v in parsed.items()}

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

@ -19,7 +19,7 @@ from urllib.parse import urlparse
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import ValidationError
from pydantic import TypeAdapter, ValidationError
import litellm
from litellm._logging import verbose_proxy_logger
@ -78,6 +78,9 @@ _PASCAL_TO_WIRE: Final[Mapping[str, str]] = {
}
_DECODED_JSON: Final = TypeAdapter(object)
def _sse_event(payload: object) -> str:
"""Frame a JSON-RPC object as a single A2A SSE event (``data: <json>\\n\\n``)."""
return f"data: {json.dumps(payload)}\n\n"
@ -91,7 +94,7 @@ def _to_jsonrpc_object(chunk: object) -> object:
"""
if isinstance(chunk, (str, bytes, bytearray)):
try:
return json.loads(chunk)
return _DECODED_JSON.validate_python(json.loads(chunk))
except (json.JSONDecodeError, UnicodeDecodeError):
return chunk
if hasattr(chunk, "model_dump"):

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

@ -3,9 +3,14 @@ from collections.abc import Iterator
from typing import Any, Final
import requests
from pydantic import ConfigDict, TypeAdapter
from .exceptions import UnauthorizedError
_SSE_LINE: Final[TypeAdapter[bytes | bytearray]] = TypeAdapter(
bytes | bytearray, config=ConfigDict(arbitrary_types_allowed=True, strict=True, hide_input_in_errors=True)
)
class ChatClient:
def __init__(self, base_url: str, api_key: str | None = None, timeout: int = 600):
@ -172,7 +177,7 @@ class ChatClient:
# Parse SSE stream
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
line = _SSE_LINE.validate_python(line).decode("utf-8")
if line.startswith("data: "):
data_str = line[6:] # Remove 'data: ' prefix
if data_str.strip() == "[DONE]":

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

@ -8,6 +8,7 @@ from urllib.parse import urlencode
import click
import requests
from pydantic import ConfigDict, TypeAdapter
from rich.console import Console
from rich.table import Table
from typing_extensions import NotRequired, ReadOnly, TypedDict, assert_never
@ -121,6 +122,8 @@ class CliAuthResult(TypedDict):
_TeamMapping: Final = TypeVar("_TeamMapping", bound=Mapping[str, object])
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
KEYRING_INSTALL_HINT: Final = "pip install 'litellm[cli]'"
KEYRING_ENABLE_HINT: Final = "keyring --enable (or unset PYTHON_KEYRING_BACKEND)"
@ -485,7 +488,7 @@ def prompt_team_selection_fallback(
def _response_error_detail(response: requests.Response) -> str | None:
try:
body: Final[dict[str, object] | list[object] | str | int | float | bool | None] = response.json()
body: Final = _JSON_OBJECT.validate_python(response.json())
except ValueError:
return None
detail: Final = body.get("detail") if isinstance(body, dict) else None

3
litellm/proxy/client/cli/commands/debug.py
View file

@ -115,6 +115,7 @@ class RequestResponsePayload(BaseModel):
_SESSION_PAGE: Final = TypeAdapter(SessionLogsPage)
_PAYLOAD: Final[TypeAdapter[RequestResponsePayload | None]] = TypeAdapter(RequestResponsePayload | None)
_JSON: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
_BACKTICK_RUNS: Final = TypeAdapter(tuple[str, ...])
_SESSION_PAGE_SIZE: Final = 100
_TRANSPORT_BODY_CHARS: Final = 500
@ -192,7 +193,7 @@ def _fmt_json(value: JsonValue, max_chars: int) -> str:
def _fenced(text: str, info: str = "") -> tuple[str, str, str]:
longest_run: Final = max((len(run) for run in re.findall(r"`+", text)), default=0)
longest_run: Final = max((len(run) for run in _BACKTICK_RUNS.validate_python(re.findall(r"`+", text))), default=0)
fence: Final = "`" * max(3, longest_run + 1)
return (f"{fence}{info}", text, fence)

10
litellm/proxy/db/model_insights_tasks.py
View file

@ -1,14 +1,20 @@
import json
from collections.abc import Mapping
from functools import lru_cache
from pathlib import Path
from typing import Final
from pydantic import ConfigDict, TypeAdapter
from litellm.types.model_insights import ModelInsightTask
_TASKS_FILE: Final = Path(__file__).resolve().parent.parent / "model_insights_tasks.json"
_TASK_ENTRIES: Final = TypeAdapter(
Mapping[str, Mapping[str, object]], config=ConfigDict(strict=True, hide_input_in_errors=True)
)
@lru_cache(maxsize=1)
def load_model_insight_tasks() -> dict[str, ModelInsightTask]:
raw: Final = json.loads(_TASKS_FILE.read_text())
return {name: ModelInsightTask(task_type=name, **entry) for name, entry in raw.items()}
raw: Final = _TASK_ENTRIES.validate_python(json.loads(_TASKS_FILE.read_text()))
return {name: ModelInsightTask.model_validate(dict(task_type=name, **entry)) for name, entry in raw.items()}

2
litellm/proxy/fine_tuning_endpoints/endpoints.py
View file

@ -426,7 +426,7 @@ async def list_fine_tuning_jobs(
route_type=CallTypes.alist_fine_tuning_jobs.value,
)
response: Any | None = None
response: object = None
if target_model_names and isinstance(target_model_names, str):
target_model_names_list: Final = target_model_names.split(",")
if len(target_model_names_list) != 1:

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