chore(typing): clear 670 basedpyright Any errors by validating provider payloads at the boundary (#45388)

* chore(typing): validate provider payloads at the boundary to clear 464 basedpyright Any errors

Replaces untyped reads of decoded JSON in 48 provider, guardrail and proxy
modules with one pydantic validation at the point the payload enters the
module, so every value read afterwards carries a real type.

reportAny 6111 -> 5663 and reportExplicitAny 1416 -> 1400. No basedpyright,
ruff-strict or type-discipline rule is higher repo-wide or in any single
file, and no cast, ignore or suppression marker was added.

* chore(typing): replace 47 Any annotations that every caller already satisfies with object

Each annotation here is a parameter or return type whose value is only
stored, compared, logged or forwarded, so object describes it exactly and
no function body changes. A site was kept only if no basedpyright,
ruff-strict or type-discipline rule rose in any file that references it.

reportExplicitAny 1400 -> 1353 and reportAny 5663 -> 5655.

* chore(typing): type upstream responses and loose annotations across providers, proxy and integrations

Second pass of the Any reduction. Decoded upstream JSON is validated once at
the provider boundary and read through typed views afterwards, annotations
that every caller already satisfies move from Any to a concrete type, and the
nine search transformations validate each result item on its own so error
text no longer carries the item index.

No cast, no type: ignore, no suppression marker and no budget file edit

* ci(unit): run the brave, google_pse, nlp_cloud and opensandbox tests in the providers shard
This commit is contained in:
Mateo Wang 2026-10-13 22:35:23 +00:00 • committed by GitHub
parent 271b4f0306
commit 0ebec123fd
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
218 changed files with 9258 additions and 484 deletions

View file

@ -233,6 +233,7 @@ jobs:
tests/unit/llms/base_llm
tests/unit/llms/baseten
tests/unit/llms/black_forest_labs
tests/unit/llms/brave
tests/unit/llms/bytez
tests/unit/llms/cerebras
tests/unit/llms/chat
@ -268,6 +269,7 @@ jobs:
tests/unit/llms/gdc
tests/unit/llms/gemini
tests/unit/llms/gigachat
tests/unit/llms/google_pse
tests/unit/llms/github_copilot
tests/unit/llms/gradient_ai
tests/unit/llms/groq
@ -300,6 +302,7 @@ jobs:
tests/unit/llms/nebius
tests/unit/llms/neosantara
tests/unit/llms/nimble
tests/unit/llms/nlp_cloud
tests/unit/llms/novita
tests/unit/llms/nscale
tests/unit/llms/nvidia_nim
@ -310,6 +313,7 @@ jobs:
tests/unit/llms/oobabooga
tests/unit/llms/openai_like
tests/unit/llms/opencode
tests/unit/llms/opensandbox
tests/unit/llms/openrouter
tests/unit/llms/ovhcloud
tests/unit/llms/parallel_ai

View file

@ -117,7 +117,7 @@ class BedrockAgentCoreA2AHandler:
response.raise_for_status()
# Check content type — AgentCore may return JSON instead of SSE
content_type: Final = response.headers.get("content-type", "").lower()
content_type: Final = ", ".join(response.headers.get_list("content-type")).lower()
if "application/json" in content_type:
# Single JSON response fallback (not SSE)

View file

@ -768,7 +768,7 @@ def _get_response_from_batch_job_output_file(
def _batch_response_was_successful(
batch_job_output_file: Mapping[str, Any], custom_llm_provider: str = "openai"
batch_job_output_file: Mapping[str, object], custom_llm_provider: str = "openai"
) -> bool:
"""
Check if the batch job response was successful

View file

@ -158,7 +158,7 @@ class DeepAgentsHandler(BaseHarnessHandler):
agent, deps = self._require_agent()
snapshot = await agent.aget_state(self._run_config(ctx, None))
messages = (snapshot.values or MappingProxyType({})).get("messages") or ()
converted: list[dict[str, Any]] = deps.convert_to_openai_messages( # mutable-ok: LangChain returns a list
converted: list[dict[str, object]] = deps.convert_to_openai_messages( # mutable-ok: LangChain returns a list
messages
)
return converted

View file

@ -50,7 +50,7 @@ class ArgillaLogger(CustomBatchLogger):
if litellm.argilla_transformation_object is None:
raise Exception("'litellm.argilla_transformation_object' is required, to log your payload to Argilla.")
self.validate_argilla_transformation_object(litellm.argilla_transformation_object)
self.argilla_transformation_object = litellm.argilla_transformation_object
self.argilla_transformation_object: Mapping[str, str] = litellm.argilla_transformation_object
self.default_credentials = self.get_credentials_from_env(
argilla_api_key=argilla_api_key,
argilla_dataset_name=argilla_dataset_name,

View file

@ -3,7 +3,7 @@ Arize Phoenix API client for fetching prompt versions from Arize Phoenix.
"""
import urllib.parse
from typing import Any, Final
from typing import Final
from litellm.llms.custom_httpx.http_handler import HTTPHandler
@ -53,7 +53,7 @@ class ArizePhoenixClient:
# Initialize HTTPHandler
self.http_handler = HTTPHandler(disable_default_headers=True)
def get_prompt_version(self, prompt_version_id: str) -> dict[str, Any] | None:
def get_prompt_version(self, prompt_version_id: str) -> dict[str, object] | None:
"""
Fetch a prompt version from Arize Phoenix.

View file

@ -80,7 +80,7 @@ def _json(value: object) -> str:
return value if isinstance(value, str) else json.dumps(value, default=str)
def _json_mapping(value: Mapping[str, Any]) -> str:
def _json_mapping(value: Mapping[str, object]) -> str:
return _json(dict(value))

View file

@ -209,7 +209,7 @@ class CloudZeroStreamer:
return payload
def _convert_cbf_to_api_format(self, row: dict[str, object]) -> dict[str, Any] | None:
def _convert_cbf_to_api_format(self, row: dict[str, object]) -> dict[str, object] | None:
"""Convert CBF row to CloudZero API format - keeping CBF field names as CloudZero expects them."""
try:
# CloudZero expects CBF format field names directly, not converted names

View file

@ -511,7 +511,7 @@ class CustomGuardrail(CustomLogger):
sticky_session_routing=self.sticky_session_routing,
)
def _get_session_id_from_request_data(self, request_data: dict[str, Any]) -> str | None:
def _get_session_id_from_request_data(self, request_data: dict[str, object]) -> str | None:
"""Extract session_id from request data."""
return get_session_id_from_request_data(request_data)

View file

@ -70,7 +70,7 @@ class PromptManager:
self,
prompt_id: str | None = None,
prompt_directory: str | None = None,
prompt_data: dict[str, dict[str, Any]] | None = None,
prompt_data: dict[str, dict[str, object]] | None = None,
prompt_file: str | None = None,
):
self.prompt_directory = Path(prompt_directory) if prompt_directory else None
@ -372,6 +372,6 @@ class PromptManager:
}
return result
def load_prompts_from_json_data(self, prompt_data: dict[str, dict[str, Any]]) -> None:
def load_prompts_from_json_data(self, prompt_data: dict[str, dict[str, object]]) -> None:
"""Load additional prompts from JSON data (merges with existing prompts)."""
self._load_prompts_from_json(prompt_data)

View file

@ -115,7 +115,7 @@ class FocusLogger(CustomLogger):
pod_lock_manager: PodLockManager | None = None
if proxy_logging_obj is not None:
writer: Final = getattr(proxy_logging_obj, "db_spend_update_writer", None)
writer: Final[object] = getattr(proxy_logging_obj, "db_spend_update_writer", None)
if writer is not None:
pod_lock_manager = getattr(writer, "pod_lock_manager", None)

View file

@ -16,6 +16,8 @@ from litellm.integrations.focus.focus_logger import FocusLogger
if TYPE_CHECKING:
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from litellm.proxy.db.db_transaction_queue.pod_lock_manager import PodLockManager
else:
AsyncIOScheduler = Any
@ -91,7 +93,7 @@ class VantageLogger(FocusLogger):
"""
from litellm.proxy.proxy_server import proxy_logging_obj
pod_lock_manager = None
pod_lock_manager: PodLockManager | None = None
if proxy_logging_obj is not None:
writer: Final[object] = getattr(proxy_logging_obj, "db_spend_update_writer", None)
if writer is not None:

View file

@ -7,6 +7,7 @@ from collections.abc import AsyncIterator, Iterator, Mapping
from typing import TYPE_CHECKING, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.azure_ai.common_utils import AZURE_ENTRA_LITELLM_PARAM_KEYS, get_azure_ai_agent_entra_token
from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
@ -30,6 +31,7 @@ if TYPE_CHECKING:
_REGISTRY_PARAMS_KEPT_OUT_OF_OPTIONAL_PARAMS: Final = (
frozenset({"api_key", "api_base", "headers", "model"}) | AZURE_ENTRA_LITELLM_PARAM_KEYS
)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
def _card_declares_no_streaming(agent_card_params: Mapping[str, object]) -> bool:
@ -325,7 +327,7 @@ class A2AConfig(BaseConfig):
# Check for JSON-RPC error
if "error" in response_json:
error: Final = response_json["error"]
error: Final = _JSON_OBJECT.validate_python(response_json["error"])
raise A2AError(
status_code=raw_response.status_code,
message=f"A2A error: {error.get('message', 'Unknown error')}",
@ -333,7 +335,7 @@ class A2AConfig(BaseConfig):
)
# Extract text from A2A response
text: Final = extract_text_from_a2a_response(response_json)
text: Final = extract_text_from_a2a_response(_JSON_OBJECT.validate_python(response_json))
# Populate model response
model_response.choices = [

View file

@ -22,7 +22,7 @@ class A2AError(BaseLLMException):
self,
status_code: int,
message: str,
headers: dict[str, Any] = {},
headers: dict[str, str] = {},
):
super().__init__(
status_code=status_code,

View file

@ -1,6 +1,8 @@
from collections.abc import Mapping, Sequence
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,10 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_CONTAINER: Final[TypeAdapter[Mapping[str, object] | str | Sequence[object]]] = TypeAdapter(
Mapping[str, object] | str | Sequence[object], config=ConfigDict(hide_input_in_errors=True)
)
OPENAI_STYLE_IMAGE_MODEL_PREFIXES: Final[tuple[str, ...]] = ("openai/",)
@ -198,29 +204,33 @@ class AimlImageGenerationConfig(BaseImageGenerationConfig):
if "data" in response_data and isinstance(response_data["data"], list):
# Handle OpenAI-like format: {"data": [{"url": "...", "width": 1024, "height": 768, "content_type": "image/jpeg"}]}
for image in response_data["data"]:
for raw_image in _JSON_CONTAINER.validate_python(response_data["data"]):
image = raw_image if isinstance(raw_image, str) else _JSON_CONTAINER.validate_python(raw_image)
if "url" in image:
url_image = _JSON_OBJECT.validate_python(image)
model_response.data.append(
ImageObject(
b64_json=None,
url=image["url"],
revised_prompt=image.get("revised_prompt"),
url=url_image["url"],
revised_prompt=url_image.get("revised_prompt"),
)
)
elif "b64_json" in image or "image_base64" in image:
b64_image = _JSON_OBJECT.validate_python(image)
model_response.data.append(
ImageObject(
b64_json=image.get("b64_json") or image.get("image_base64"),
b64_json=b64_image.get("b64_json") or b64_image.get("image_base64"),
url=None,
revised_prompt=image.get("revised_prompt"),
revised_prompt=b64_image.get("revised_prompt"),
)
)
elif "output" in response_data and "choices" in response_data["output"]:
for choice in response_data["output"]["choices"]:
for raw_choice in _JSON_CONTAINER.validate_python(response_data["output"]["choices"]):
choice = raw_choice if isinstance(raw_choice, str) else _JSON_CONTAINER.validate_python(raw_choice)
if "image_base64" in choice:
model_response.data.append(
ImageObject(
b64_json=choice["image_base64"],
b64_json=_JSON_OBJECT.validate_python(choice)["image_base64"],
url=None,
)
)
@ -228,23 +238,24 @@ class AimlImageGenerationConfig(BaseImageGenerationConfig):
model_response.data.append(
ImageObject(
b64_json=None,
url=choice["url"],
url=_JSON_OBJECT.validate_python(choice)["url"],
)
)
elif "images" in response_data:
# Handle alternative format: {"images": [{"url": "...", "width": 1024, "height": 768, "content_type": "image/jpeg"}]}
for image in response_data["images"]:
for raw_image in _JSON_CONTAINER.validate_python(response_data["images"]):
image = raw_image if isinstance(raw_image, str) else _JSON_CONTAINER.validate_python(raw_image)
if "url" in image:
model_response.data.append(
ImageObject(
b64_json=None,
url=image["url"],
url=_JSON_OBJECT.validate_python(image)["url"],
)
)
elif "image_base64" in image:
model_response.data.append(
ImageObject(
b64_json=image["image_base64"],
b64_json=_JSON_OBJECT.validate_python(image)["image_base64"],
url=None,
)
)

View file

@ -4,6 +4,8 @@ Anthropic Token Counter implementation using the CountTokens API.
from typing import Any, Final
from pydantic import JsonValue
from litellm._logging import verbose_logger
from litellm.exceptions import AuthenticationError
from litellm.llms.anthropic.count_tokens.handler import AnthropicCountTokensHandler
@ -26,12 +28,12 @@ class AnthropicTokenCounter(BaseTokenCounter):
async def count_tokens(
self,
model_to_use: str,
messages: list[dict[str, Any]] | None,
messages: list[dict[str, JsonValue]] | None,
contents: list[dict[str, object]] | None,
deployment: dict[str, Any] | None = None,
request_model: str = "",
tools: list[dict[str, Any]] | None = None,
system: Any | None = None,
tools: list[dict[str, JsonValue]] | None = None,
system: JsonValue = None,
) -> TokenCountResponse | None:
"""
Count tokens using Anthropic's CountTokens API.

View file

@ -14,7 +14,7 @@ Anthropic Files API endpoints:
import calendar
import time
from collections.abc import Mapping
from collections.abc import Iterable, Mapping
from typing import Final, cast
import httpx
@ -49,6 +49,7 @@ 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))
_JSON_ITEMS: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class AnthropicFilesConfig(BaseFilesConfig):
@ -284,9 +285,9 @@ class AnthropicFilesConfig(BaseFilesConfig):
"last_id": "..."
}
"""
response_json: Final = raw_response.json()
files_data: Final = response_json.get("data", [])
return [self._parse_anthropic_file(f) for f in files_data]
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
files_data: Final = _JSON_ITEMS.validate_python(response_json.get("data", []))
return [self._parse_anthropic_file(_JSON_OBJECT.validate_python(f)) for f in files_data]
def transform_file_content_request(
self,

View file

@ -1347,7 +1347,7 @@ class LiteLLMAnthropicMessagesAdapter:
self,
choices: list[Choices],
tool_name_mapping: dict[str, str] | None = None,
) -> list[dict[str, Any]]:
) -> list[dict[str, object]]:
new_content: Final[list[dict[str, Any]]] = []
for choice, compaction_blocks in (
(choice, _compaction_blocks(_optional_attr(choice.message, "provider_specific_fields")))

View file

@ -4,9 +4,10 @@ Azure AI Anthropic CountTokens API handler.
Uses httpx for HTTP requests with Azure authentication.
"""
from typing import Any, Final
from typing import Final
import httpx
from pydantic import JsonValue
import litellm
from litellm._logging import verbose_logger
@ -28,14 +29,14 @@ class AzureAIAnthropicCountTokensHandler(AzureAIAnthropicCountTokensConfig):
async def handle_count_tokens_request(
self,
model: str,
messages: list[dict[str, Any]],
messages: list[dict[str, JsonValue]],
api_key: str,
api_base: str,
litellm_params: dict[str, object] | None = None,
timeout: float | httpx.Timeout | None = None,
tools: list[dict[str, Any]] | None = None,
tools: list[dict[str, JsonValue]] | None = None,
system: object = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Handle a CountTokens request using httpx with Azure authentication.

View file

@ -5,6 +5,8 @@ Azure AI Anthropic Token Counter implementation using the CountTokens API.
import os
from typing import Any, Final
from pydantic import JsonValue
from litellm._logging import verbose_logger
from litellm.llms.azure_ai.anthropic.count_tokens.handler import (
AzureAIAnthropicCountTokensHandler,
@ -28,11 +30,11 @@ class AzureAIAnthropicTokenCounter(BaseTokenCounter):
async def count_tokens(
self,
model_to_use: str,
messages: list[dict[str, object]] | None,
messages: list[dict[str, JsonValue]] | None,
contents: list[dict[str, object]] | None,
deployment: dict[str, Any] | None = None,
request_model: str = "",
tools: list[dict[str, object]] | None = None,
tools: list[dict[str, JsonValue]] | None = None,
system: object | None = None,
) -> TokenCountResponse | None:
"""
@ -86,12 +88,14 @@ class AzureAIAnthropicTokenCounter(BaseTokenCounter):
)
if result is not None:
return TokenCountResponse(
total_tokens=result.get("input_tokens", 0),
request_model=request_model,
model_used=model_to_use,
tokenizer_type="azure_ai_anthropic_api",
original_response=result,
return TokenCountResponse.model_validate(
{
"total_tokens": result.get("input_tokens", 0),
"request_model": request_model,
"model_used": model_to_use,
"tokenizer_type": "azure_ai_anthropic_api",
"original_response": result,
}
)
except AnthropicError as e:
verbose_logger.warning(

View file

@ -113,7 +113,7 @@ class AzureFoundryFlux2ImageEditConfig(OpenAIImageEditConfig):
for index, reference_image in enumerate(images, start=1)
}
)
request_body: Final[dict[str, Any]] = {
request_body: Final[dict[str, object]] = {
"prompt": prompt,
"model": model,
**reference_images,

View file

@ -10,6 +10,7 @@ they created. Callers with no admin role and no identifying ids at all
are denied so an empty user_id can never select an unscoped query.
"""
from collections.abc import Mapping, Sequence
from typing import Any, Final
from litellm.types.proxy.auth.user_api_key_auth import (
@ -46,7 +47,7 @@ def resolve_resource_owner_id(
return None
def build_list_page(items: list[Any], has_more: bool = False) -> dict[str, Any]:
def build_list_page(items: list[Any], has_more: bool = False) -> dict[str, object]:
"""Build the OpenAI-style paginated list response shape used by managed
file/batch/vector-store listings. ``first_id`` and ``last_id`` are
sourced from each item's ``.id`` attribute."""
@ -61,7 +62,7 @@ def build_list_page(items: list[Any], has_more: bool = False) -> dict[str, Any]:
def build_owner_filter(
user_api_key_dict: UserAPIKeyAuth,
) -> dict[str, Any] | None:
) -> dict[str, str | Sequence[Mapping[str, str]]] | None:
"""Return a Prisma `where` fragment that scopes a managed-resource listing
to records the caller is allowed to see.

View file

@ -7,9 +7,11 @@ The main difference is in the response format: Qwen2 uses "text" field while Qwe
Qwen2 + Invoke API Tutorial: https://docs.aws.amazon.com/bedrock/latest/userguide/invoke-imported-model.html
"""
from collections.abc import Mapping
from typing import TYPE_CHECKING, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.bedrock.chat.invoke_transformations.amazon_qwen3_transformation import (
AmazonQwen3Config,
@ -23,6 +25,9 @@ from litellm.types.utils import ModelResponse, Usage
if TYPE_CHECKING:
from litellm.litellm_core_utils.tokenizer import Encoding as Tokenizer
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_TEXT: Final = TypeAdapter(str, config=ConfigDict(strict=True, hide_input_in_errors=True))
class AmazonQwen2Config(AmazonQwen3Config):
"""
@ -54,10 +59,10 @@ class AmazonQwen2Config(AmazonQwen3Config):
Qwen2 uses "text" field, but we also support "generation" field for compatibility.
"""
try:
response_data: Final = raw_response.json()
response_data: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Extract the generated text - Qwen2 uses "text" field, but also support "generation" for compatibility
generated_text = response_data.get("generation", "") or response_data.get("text", "")
generated_text = _TEXT.validate_python(response_data.get("generation", "") or response_data.get("text", ""))
# Clean up the response (remove assistant start token if present)
generated_text = generated_text.removeprefix("<|im_start|>assistant\n")
@ -71,14 +76,16 @@ class AmazonQwen2Config(AmazonQwen3Config):
# Set usage information if available in response
if "usage" in response_data:
usage_data: Final = response_data["usage"]
usage_data: Final = _JSON_OBJECT.validate_python(response_data["usage"])
setattr(
model_response,
"usage",
Usage(
prompt_tokens=usage_data.get("prompt_tokens", 0),
completion_tokens=usage_data.get("completion_tokens", 0),
total_tokens=usage_data.get("total_tokens", 0),
Usage.model_validate(
{
"prompt_tokens": usage_data.get("prompt_tokens", 0),
"completion_tokens": usage_data.get("completion_tokens", 0),
"total_tokens": usage_data.get("total_tokens", 0),
}
),
)

View file

@ -10,6 +10,7 @@ from collections.abc import Mapping
from typing import TYPE_CHECKING, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.llms.base_llm.chat.transformation import BaseConfig
from litellm.llms.bedrock.chat.invoke_transformations.base_invoke_transformation import (
@ -22,6 +23,9 @@ from litellm.types.utils import ModelResponse, Usage
if TYPE_CHECKING:
from litellm.litellm_core_utils.tokenizer import Encoding as Tokenizer
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_TEXT: Final = TypeAdapter(str, config=ConfigDict(strict=True, hide_input_in_errors=True))
class AmazonQwen3Config(AmazonInvokeConfig, BaseConfig):
"""
@ -179,10 +183,10 @@ class AmazonQwen3Config(AmazonInvokeConfig, BaseConfig):
Transform Qwen3 Bedrock response to OpenAI format
"""
try:
response_data: Final = raw_response.json()
response_data: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Extract the generated text - Qwen3 uses "generation" field
generated_text = response_data.get("generation", "")
generated_text = _TEXT.validate_python(response_data.get("generation", ""))
# Clean up the response (remove assistant start token if present)
generated_text = generated_text.removeprefix("<|im_start|>assistant\n")
@ -196,14 +200,16 @@ class AmazonQwen3Config(AmazonInvokeConfig, BaseConfig):
# Set usage information if available in response
if "usage" in response_data:
usage_data: Final = response_data["usage"]
usage_data: Final = _JSON_OBJECT.validate_python(response_data["usage"])
setattr(
model_response,
"usage",
Usage(
prompt_tokens=usage_data.get("prompt_tokens", 0),
completion_tokens=usage_data.get("completion_tokens", 0),
total_tokens=usage_data.get("total_tokens", 0),
Usage.model_validate(
{
"prompt_tokens": usage_data.get("prompt_tokens", 0),
"completion_tokens": usage_data.get("completion_tokens", 0),
"total_tokens": usage_data.get("total_tokens", 0),
}
),
)

View file

@ -51,6 +51,10 @@ _GUARDRAIL_CONFIG_EXPECTED_FORMAT: Final = (
"{'guardrailIdentifier': str, 'guardrailVersion': str, 'trace': 'enabled'|'disabled'|'enabled_full'}"
)
_INVOKE_PROVIDERS: Final = TypeAdapter(tuple[litellm.BEDROCK_INVOKE_PROVIDERS_LITERAL, ...]).validate_python(
get_args(litellm.BEDROCK_INVOKE_PROVIDERS_LITERAL)
)
def _bedrock_invoke_guardrail_headers(raw_guardrail_config: object) -> "dict[str, str]":
try:
@ -568,7 +572,7 @@ class AmazonInvokeConfig(BaseConfig, BaseAWSLLM):
if provider is not None:
return provider
for provider in get_args(litellm.BEDROCK_INVOKE_PROVIDERS_LITERAL):
for provider in _INVOKE_PROVIDERS:
if provider in model:
return provider
return None

View file

@ -6,15 +6,20 @@ Documentation: https://api-dashboard.search.brave.com/app/documentation/web-sear
from __future__ import annotations
import re
from collections.abc import Iterable, Mapping
from datetime import datetime, timezone
from typing import Final, Literal, TypedDict
import httpx
from dateutil import parser
from pydantic import ConfigDict, TypeAdapter
_ISO_YMD: Final = re.compile(r"^\s*\d{4}[-/]\d{1,2}[-/]\d{1,2}\s*$")
_UNIX_TIMESTAMP: Final = re.compile(r"^\s*-?\d+(\.\d+)?\s*$")
BRAVE_SECTIONS: Final = ["web", "discussions", "faqs", "faq", "news", "videos"]
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -26,7 +31,7 @@ from litellm.secret_managers.main import get_secret_str
def to_yyyy_mm_dd(
s: str | float | None,
s: object,
*,
dayfirst: bool = False,
yearfirst: bool = False,
@ -246,7 +251,7 @@ class BraveSearchConfig(BaseSearchConfig):
"""
Transform Brave Search API response to LiteLLM unified SearchResponse format.
"""
response_json: Final = raw_response.json()
response_body: Final = _JSON_VALUE.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final[list[SearchResult]] = []
@ -254,9 +259,11 @@ class BraveSearchConfig(BaseSearchConfig):
query_params: Final = raw_response.request.url.params if raw_response.request else {}
sections_to_process: Final = self._sections_from_params(dict(query_params))
max_results: Final = max(1, min(int(query_params.get("count", 20)), 20))
response_json: Final = _JSON_OBJECT.validate_python(response_body)
for section in sections_to_process:
for result in response_json.get(section, {}).get("results", []):
section_json = _JSON_OBJECT.validate_python(response_json.get(section, {}))
for raw_result in _JSON_VALUES.validate_python(section_json.get("results", [])):
# Because the `max_results`/`count` parameters do not affect
# the number of "discussion", "faq", "news", or "videos"
# results, we need to manually limit the number of results
@ -264,18 +271,21 @@ class BraveSearchConfig(BaseSearchConfig):
if len(results) >= max_results:
break
result = _JSON_OBJECT.validate_python(raw_result)
title = result.get("title", "")
url = result.get("url", "")
snippet = result.get("description", "")
date = to_yyyy_mm_dd(result.get("page_age") or result.get("age"))
last_updated = to_yyyy_mm_dd(result.get("fetched_content_timestamp", ""))
search_result = SearchResult(
title=title,
url=url,
snippet=snippet,
date=date,
last_updated=last_updated,
search_result = SearchResult.model_validate(
{
"title": title,
"url": url,
"snippet": snippet,
"date": date,
"last_updated": last_updated,
}
)
results.append(search_result)

View file

@ -10,10 +10,11 @@ Convers
Docs - https://docs.cohere.com/v2/reference/embed
"""
from collections.abc import Sized
from collections.abc import Iterable, Mapping, Sized
from typing import Final, Protocol, cast
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm import COHERE_DEFAULT_EMBEDDING_INPUT_TYPE
@ -30,6 +31,9 @@ from litellm.utils import is_base64_encoded
from ..common_utils import CohereError
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_EMBEDDINGS_BY_TYPE: Final = TypeAdapter(Mapping[str, Iterable[object]], config=ConfigDict(hide_input_in_errors=True))
class _SupportsEncode(Protocol):
"""Tokenizer handle: the embedding usage path only encodes text to measure its token length."""
@ -198,7 +202,8 @@ class CohereEmbeddingConfig(BaseEmbeddingConfig):
'usage'
}
"""
embeddings: Final = response_json["embeddings"]
payload: Final = _JSON_OBJECT.validate_python(response_json)
embeddings: Final = _EMBEDDINGS_BY_TYPE.validate_python(payload["embeddings"])
output_data: Final = []
for k, embedding_list in embeddings.items():
for idx, embedding in enumerate(embedding_list):
@ -213,7 +218,7 @@ class CohereEmbeddingConfig(BaseEmbeddingConfig):
setattr(
model_response,
"usage",
self._calculate_usage(input, encoding, response_json.get("meta", {})),
self._calculate_usage(input, encoding, dict(_JSON_OBJECT.validate_python(payload.get("meta", {})))),
)
return model_response

View file

@ -2,9 +2,11 @@
CometAPI Embedding API support - OpenAI compatible
"""
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
@ -15,6 +17,9 @@ from litellm.types.utils import EmbeddingResponse, Usage
from ..common_utils import CometAPIException
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_INT: Final = TypeAdapter(int)
class CometAPIEmbeddingConfig(BaseEmbeddingConfig):
"""
@ -136,9 +141,10 @@ class CometAPIEmbeddingConfig(BaseEmbeddingConfig):
model_response.data = raw_response_json.get("data")
model_response.object = raw_response_json.get("object")
usage_payload: Final = _JSON_OBJECT.validate_python(raw_response_json.get("usage", {}))
usage: Final = Usage(
prompt_tokens=raw_response_json.get("usage", {}).get("prompt_tokens", 0),
total_tokens=raw_response_json.get("usage", {}).get("total_tokens", 0),
prompt_tokens=_INT.validate_python(usage_payload.get("prompt_tokens", 0) or 0),
total_tokens=_INT.validate_python(usage_payload.get("total_tokens", 0) or 0),
)
model_response.usage = usage

View file

@ -147,7 +147,7 @@ class DashScopeEmbeddingConfig(BaseEmbeddingConfig):
)
if "error" in response_json:
error: Final = response_json["error"]
error: Final[object] = response_json["error"]
message: Final = error.get("message", str(error)) if isinstance(error, dict) else str(error)
raise DashScopeError(
status_code=raw_response.status_code,

View file

@ -475,7 +475,7 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
# Move message-level cache_control into a content block when content is a string.
if "cache_control" in _message and isinstance(_message.get("content"), str):
_message = self._move_cache_control_into_string_content_block(_message)
_sanitize_empty_content(cast(dict[str, Any], _message))
_sanitize_empty_content(cast(dict[str, object], _message))
if _is_bare_assistant_message(_message):
continue
new_messages.append(_message)
@ -504,7 +504,7 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
"""
content: Final = message.get("content")
# Create new message with cache_control moved into content block
transformed_message: Final = cast(dict[str, Any], message.copy())
transformed_message: Final = cast(dict[str, object], message.copy())
cache_control: Final = transformed_message.pop("cache_control")
transformed_message["content"] = [
{

View file

@ -4,9 +4,11 @@ Calls DataForSEO SERP API to search the web.
DataForSEO API Reference: https://docs.dataforseo.com/v3/serp/google/organic/live/advanced/?bash
"""
from collections.abc import Iterable, Mapping
from typing import Final, Literal
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.constants import DEFAULT_DATAFORSEO_LOCATION_CODE
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -17,6 +19,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class DataForSEOSearchConfig(BaseSearchConfig):
"""
@ -201,18 +206,21 @@ class DataForSEOSearchConfig(BaseSearchConfig):
if task.get("status_code") == 20000 and "result" in task:
# Result is an array, take first element
if len(task["result"]) > 0:
result: Final = task["result"][0]
result: Final = _JSON_OBJECT.validate_python(task["result"][0])
# Items contain the actual search results
for item in result.get("items", []):
for raw_item in _JSON_VALUES.validate_python(result.get("items", [])):
item = _JSON_OBJECT.validate_python(raw_item)
# Only process organic search results
if item.get("type") == "organic":
search_result = SearchResult(
title=item.get("title", ""),
url=item.get("url", ""),
snippet=item.get("description", ""),
date=None, # DataForSEO doesn't provide date in standard response
last_updated=None,
search_result = SearchResult.model_validate(
{
"title": item.get("title", ""),
"url": item.get("url", ""),
"snippet": item.get("description", ""),
"date": None, # DataForSEO doesn't provide date in standard response
"last_updated": None,
}
)
results.append(search_result)

View file

@ -14,7 +14,7 @@ import json
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Any, Final, Protocol
from litellm.harness.options import DeepAgentsOptions
from litellm.harness.types import (
@ -73,7 +73,7 @@ _APPROVAL_DECISIONS: Final = ("approve", "reject")
def chat_model_kwargs(
ctx: SessionContext,
) -> dict[str, Any]: # mutable-ok: ChatLiteLLM constructor kwargs, splatted as **kwargs
) -> dict[str, object]: # mutable-ok: ChatLiteLLM constructor kwargs, splatted as **kwargs
"""ChatLiteLLM constructor kwargs for gateway or SDK mode."""
if not ctx.model:
raise ValueError("Harness.DEEPAGENTS needs model=")
@ -107,7 +107,7 @@ def blocked_tools(permissions: str, disable_tools: Sequence[str]) -> frozenset[s
def interrupt_config(
permissions: str, blocked: frozenset[str]
) -> dict[str, Any] | None: # mutable-ok: deepagents create_deep_agent(interrupt_on=) takes a dict
) -> dict[str, dict[str, list[str]]] | None: # mutable-ok: deepagents create_deep_agent(interrupt_on=) takes a dict
"""interrupt_on for permissions='ask': approve/reject every mutating built-in."""
if permissions != "ask":
return None
@ -221,9 +221,14 @@ def _message_events(message: object, skip_tools: frozenset[str]) -> tuple[Event,
return ()
class InterruptLike(Protocol):
@property
def id(self) -> str: ...
def interrupts_in(
update: object,
) -> list[Any]: # mutable-ok: returns a list; existing callers/tests compare it to list literals
) -> list[InterruptLike]: # mutable-ok: returns a list; existing callers/tests compare it to list literals
if not isinstance(update, Mapping):
return [] # mutable-ok: returns a list; existing callers/tests compare it to list literals
found = update.get("__interrupt__")
@ -260,7 +265,7 @@ def approval_requests(
return list(kept) # mutable-ok: list return; callers/tests compare to lists
def decision(allowed: bool, reason: str) -> dict[str, Any]: # mutable-ok: LangGraph resume payload (HITL decision dict)
def decision(allowed: bool, reason: str) -> dict[str, str]: # mutable-ok: LangGraph resume payload (HITL decision dict)
if allowed:
return {"type": "approve"} # mutable-ok: LangGraph resume payload (HITL decision dict)
return { # mutable-ok: LangGraph HITL decision
@ -273,7 +278,7 @@ def decision(allowed: bool, reason: str) -> dict[str, Any]: # mutable-ok: LangG
class TurnState:
"""Mutable state across the stream passes of one turn."""
interrupts: tuple[Any, ...] = ()
interrupts: tuple[InterruptLike, ...] = ()
class DeepAgentsHarnessConfig(BaseHarnessConfig):

View file

@ -4,10 +4,12 @@ Calls DuckDuckGo's Instant Answer API to search the web.
DuckDuckGo API Reference: https://duckduckgo.com/api
"""
from collections.abc import Iterable, Mapping
from typing import Final, Literal, TypedDict
from urllib.parse import urlencode
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -17,6 +19,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _DuckDuckGoSearchRequestRequired(TypedDict):
"""Required fields for DuckDuckGo Search API request."""
@ -159,7 +164,7 @@ class DuckDuckGoSearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Extract max_results from the request URL params
query_params: Final = raw_response.request.url.params if raw_response.request else {}
@ -178,17 +183,19 @@ class DuckDuckGoSearchConfig(BaseSearchConfig):
# Check if there's an Abstract with URL
if response_json.get("AbstractURL") and response_json.get("AbstractText"):
abstract_result: Final = SearchResult(
title=response_json.get("Heading", ""),
url=response_json.get("AbstractURL", ""),
snippet=response_json.get("AbstractText", ""),
date=None,
last_updated=None,
abstract_result: Final = SearchResult.model_validate(
{
"title": response_json.get("Heading", ""),
"url": response_json.get("AbstractURL", ""),
"snippet": response_json.get("AbstractText", ""),
"date": None,
"last_updated": None,
}
)
results.append(abstract_result)
# Process RelatedTopics
related_topics: Final = response_json.get("RelatedTopics", [])
related_topics: Final = _JSON_VALUES.validate_python(response_json.get("RelatedTopics", []))
for topic in related_topics:
# Stop if we've reached max_results
if max_results is not None and len(results) >= max_results:

View file

@ -4,9 +4,11 @@ Calls Exa AI's /search endpoint to search the web.
Exa AI API Reference: https://docs.exa.ai/reference/search
"""
from collections.abc import Iterable, Mapping
from typing import Final, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -16,6 +18,10 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
_STRS: Final = TypeAdapter(Iterable[str], config=ConfigDict(hide_input_in_errors=True))
class _ExaAISearchRequestRequired(TypedDict):
"""Required fields for Exa AI Search API request."""
@ -177,20 +183,23 @@ class ExaAISearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final = []
for result in response_json.get("results", []):
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("url", ""),
snippet=result.get("text")
or "\n\n".join(result.get("highlights") or [])
or result.get("summary")
or "",
date=result.get("publishedDate"), # ISO 8601 datetime string
last_updated=None, # Exa AI doesn't provide last_updated in response
for raw_result in _JSON_VALUES.validate_python(response_json.get("results", [])):
result = _JSON_OBJECT.validate_python(raw_result)
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": result.get("text")
or "\n\n".join(_STRS.validate_python(result.get("highlights") or []))
or result.get("summary")
or "",
"date": result.get("publishedDate"), # ISO 8601 datetime string
"last_updated": None, # Exa AI doesn't provide last_updated in response
}
)
results.append(search_result)

View file

@ -7,6 +7,7 @@ from typing import TYPE_CHECKING, Final, Protocol, runtime_checkable
import httpx
from httpx._types import RequestFiles
from pydantic import ConfigDict, TypeAdapter
from litellm.images.utils import ImageEditRequestUtils
from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig
@ -23,6 +24,8 @@ from litellm.types.utils import FileTypes, ImageResponse
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
DEFAULT_BASE_URL: Final[str] = "https://fal.run"
EDIT_SUFFIX: Final[str] = "/edit"
SUPPORTED_OPENAI_PARAMS: Final[tuple[str, ...]] = ("background", "mask", "n", "quality", "size")
@ -163,7 +166,7 @@ class FalAIImageEditConfig(BaseImageEditConfig):
logging_obj: "LiteLLMLoggingObj",
) -> ImageResponse:
try:
response_json: Final = raw_response.json()
response_json: Final = _JSON_VALUE.validate_python(raw_response.json())
except Exception as e:
raise self.get_error_class(
error_message=f"Error parsing Fal AI image edit response: {e}",
@ -171,5 +174,7 @@ class FalAIImageEditConfig(BaseImageEditConfig):
headers=raw_response.headers,
)
model_response: Final = ImageResponse()
model_response.data = list(fal_images_to_image_objects(response_json.get("images", ())))
model_response.data = list(
fal_images_to_image_objects(_JSON_OBJECT.validate_python(response_json).get("images", ()))
)
return model_response

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,9 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class FalAIBriaConfig(FalAIBaseConfig):
"""
@ -206,7 +211,7 @@ class FalAIBriaConfig(FalAIBaseConfig):
}
"""
try:
response_data: Final = raw_response.json()
response_data: Final = _JSON_VALUE.validate_python(raw_response.json())
except Exception as e:
raise self.get_error_class(
error_message=f"Error transforming image generation response: {e}",
@ -218,7 +223,7 @@ class FalAIBriaConfig(FalAIBaseConfig):
model_response.data = []
# Handle Bria response format - uses "image" (singular) not "images"
image_data: Final = response_data.get("image")
image_data: Final = _JSON_OBJECT.validate_python(response_data).get("image")
if image_data and isinstance(image_data, dict):
model_response.data.append(
ImageObject(

View file

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final, cast
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
@ -16,6 +18,9 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class FalAIImagen4Config(FalAIBaseConfig):
"""
@ -203,7 +208,7 @@ class FalAIImagen4Config(FalAIBaseConfig):
}
"""
try:
response_data: Final = raw_response.json()
response_data: Final = _JSON_VALUE.validate_python(raw_response.json())
except Exception as e:
raise self.get_error_class(
error_message=f"Error transforming image generation response: {e}",
@ -215,7 +220,8 @@ class FalAIImagen4Config(FalAIBaseConfig):
model_response.data = []
# Handle Imagen4 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):
@ -239,7 +245,7 @@ class FalAIImagen4Config(FalAIBaseConfig):
hidden_params: Final = cast( # cast-ok: preserve mapping operations on dynamic response metadata
dict[str, object], getattr(model_response, HIDDEN_PARAMS_ATTR)
)
if "seed" in response_data:
hidden_params["seed"] = response_data["seed"]
if "seed" in response_object:
hidden_params["seed"] = response_object["seed"]
return model_response

View file

@ -2,7 +2,7 @@ from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import TypeAdapter
from pydantic import ConfigDict, TypeAdapter
from typing_extensions import ReadOnly, TypedDict
from litellm.llms.base_llm.image_generation.transformation import (
@ -31,6 +31,8 @@ class FalImageProviderSpecificFields(TypedDict, total=False):
_FAL_IMAGE_DATA: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object])
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
def fal_images_to_image_objects(images: object) -> tuple[ImageObject, ...]:
@ -124,7 +126,7 @@ class FalAIBaseConfig(BaseImageGenerationConfig):
Transform the image generation response to the litellm image response
"""
try:
response_data: Final = raw_response.json()
response_data: Final = _JSON_VALUE.validate_python(raw_response.json())
except Exception as e:
raise self.get_error_class(
error_message=f"Error transforming image generation response: {e}",
@ -134,7 +136,9 @@ class FalAIBaseConfig(BaseImageGenerationConfig):
if not model_response.data:
model_response.data = []
model_response.data.extend(fal_images_to_image_objects(response_data.get("images", ())))
model_response.data.extend(
fal_images_to_image_objects(_JSON_OBJECT.validate_python(response_data).get("images", ()))
)
return model_response

View file

@ -8,9 +8,11 @@ or cloud). The search response uses the Firecrawl-compatible envelope
fastCRW API Reference: https://fastcrw.com/docs/rest-api
"""
from collections.abc import Iterable, Mapping
from typing import Final, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -20,6 +22,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _FastCRWSearchRequestRequired(TypedDict):
"""Required fields for fastCRW Search API request."""
@ -157,21 +162,24 @@ class FastCRWSearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
results: Final = []
data: Final = response_json.get("data", [])
if isinstance(data, list):
for result in data:
for raw_result in _JSON_VALUES.validate_python(data):
result = _JSON_OBJECT.validate_python(raw_result)
snippet = result.get("markdown") or result.get("description", "")
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("url", ""),
snippet=snippet,
date=None,
last_updated=None,
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": snippet,
"date": None,
"last_updated": None,
}
)
results.append(search_result)

View file

@ -4,9 +4,11 @@ Calls Firecrawl's /search endpoint to search the web.
Firecrawl API Reference: https://docs.firecrawl.dev/api-reference/endpoint/search
"""
from collections.abc import Iterable, Mapping
from typing import Final, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -16,6 +18,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _FirecrawlSearchRequestRequired(TypedDict):
"""Required fields for Firecrawl Search API request."""
@ -170,7 +175,7 @@ class FirecrawlSearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final = []
@ -179,44 +184,54 @@ class FirecrawlSearchConfig(BaseSearchConfig):
if isinstance(data, list):
# Self-hosted Firecrawl (v1) format: data is a flat list of results
for result in data:
for raw_result in _JSON_VALUES.validate_python(data):
result = _JSON_OBJECT.validate_python(raw_result)
snippet = result.get("markdown") or result.get("description", "")
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("url", ""),
snippet=snippet,
date=None,
last_updated=None,
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": snippet,
"date": None,
"last_updated": None,
}
)
results.append(search_result)
elif isinstance(data, dict):
# Firecrawl Cloud (v2) format: data is a dict with web/news keys
web_results: Final = data.get("web", [])
sections: Final = _JSON_OBJECT.validate_python(data)
web_results: Final = _JSON_VALUES.validate_python(sections.get("web", []))
for result in web_results:
for raw_result in web_results:
result = _JSON_OBJECT.validate_python(raw_result)
# Use markdown if available, otherwise fall back to description
snippet = result.get("markdown") or result.get("description", "")
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("url", ""),
snippet=snippet,
date=None,
last_updated=None,
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": snippet,
"date": None,
"last_updated": None,
}
)
results.append(search_result)
# Process news results if available (they have date field)
news_results: Final = data.get("news", [])
for result in news_results:
news_results: Final = _JSON_VALUES.validate_python(sections.get("news", []))
for raw_result in news_results:
result = _JSON_OBJECT.validate_python(raw_result)
snippet = result.get("markdown") or result.get("snippet", "")
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("url", ""),
snippet=snippet,
date=result.get("date"), # News results include date
last_updated=None,
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": snippet,
"date": result.get("date"), # News results include date
"last_updated": None,
}
)
results.append(search_result)

View file

@ -1,8 +1,9 @@
import json
from collections.abc import AsyncIterator, Iterator, Mapping
from collections.abc import AsyncIterator, Iterable, Iterator, Mapping
from typing import TYPE_CHECKING, Final, Literal, cast
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm._logging import verbose_logger
@ -51,6 +52,10 @@ from ..common_utils import (
if TYPE_CHECKING:
from litellm.litellm_core_utils.tokenizer import Encoding as Tokenizer
_JSON_DICT: Final = TypeAdapter(dict[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)
def _map_reasoning_effort(value: object) -> object:
effort: Final[object] = cast(Mapping[str, object], value).get("effort") if isinstance(value, Mapping) else value
@ -757,7 +762,8 @@ class FireworksAIConfig(FireworksAIMixin, OpenAIGPTConfig):
additional_headers: Final = get_response_headers(raw_response_headers)
response: Final = ModelResponse(**completion_response)
completion_fields: Final = _JSON_DICT.validate_python(completion_response)
response: Final = ModelResponse(**completion_fields)
if response.model is not None:
response.model = "fireworks_ai/" + response.model
@ -771,7 +777,7 @@ class FireworksAIConfig(FireworksAIMixin, OpenAIGPTConfig):
response.hidden_params = {
"additional_headers": additional_headers,
**_extract_fireworks_hidden_params(completion_response),
**_extract_fireworks_hidden_params(completion_fields),
}
return response
@ -832,9 +838,9 @@ class FireworksAIConfig(FireworksAIMixin, OpenAIGPTConfig):
f"Failed to fetch models from Fireworks AI. Status code: {response.status_code}, Response: {response.json()}"
)
models: Final = response.json()["models"]
models: Final = _JSON_OBJECTS.validate_python(_JSON_DICT.validate_python(response.json())["models"])
return ["fireworks_ai/" + model["name"] for model in models]
return ["fireworks_ai/" + _STR.validate_python(model["name"]) for model in models]
@staticmethod
def get_api_key(api_key: str | None = None) -> str | None:

View file

@ -1,4 +1,4 @@
from collections.abc import Iterable, Mapping
from collections.abc import Iterable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -35,6 +35,10 @@ else:
_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))
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_CONTAINER: Final[TypeAdapter[Mapping[str, object] | str | Sequence[object]]] = TypeAdapter(
Mapping[str, object] | str | Sequence[object], config=ConfigDict(hide_input_in_errors=True)
)
class GoogleImageGenConfig(BaseImageGenerationConfig):
@ -199,19 +203,22 @@ class GoogleImageGenConfig(BaseImageGenerationConfig):
# Handle different response formats based on model
if is_gemini_image_model(model):
# Gemini Flash Image Preview models return in candidates format
candidates: Final = response_data.get("candidates", [])
response_object: Final = _JSON_OBJECT.validate_python(response_data)
candidates: Final = _JSON_CONTAINER.validate_python(response_object.get("candidates", []))
for candidate in candidates:
content = candidate.get("content", {})
parts = content.get("parts", [])
for part in parts:
content = _JSON_OBJECT.validate_python(_JSON_OBJECT.validate_python(candidate).get("content", {}))
parts = _JSON_CONTAINER.validate_python(content.get("parts", []))
for raw_part in parts:
part = raw_part if isinstance(raw_part, str) else _JSON_CONTAINER.validate_python(raw_part)
# Look for inlineData with image
if "inlineData" in part:
inline_data = part["inlineData"]
part_object = _JSON_OBJECT.validate_python(part)
inline_data = _JSON_CONTAINER.validate_python(part_object["inlineData"])
if "data" in inline_data:
thought_sig = part.get("thoughtSignature")
thought_sig = part_object.get("thoughtSignature")
model_response.data.append(
ImageObject(
b64_json=inline_data["data"],
b64_json=_JSON_OBJECT.validate_python(inline_data)["data"],
url=None,
provider_specific_fields=(
{"thought_signature": thought_sig} if thought_sig else None
@ -224,12 +231,14 @@ class GoogleImageGenConfig(BaseImageGenerationConfig):
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)
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)
else:
# Original Imagen format - predictions with generated images
predictions: Final = _JSON_OBJECTS.validate_python(response_data.get("predictions", []))
predictions: Final = _JSON_OBJECTS.validate_python(
_JSON_OBJECT.validate_python(response_data).get("predictions", [])
)
for prediction in predictions:
# Google AI returns base64 encoded images in the prediction
model_response.data.append(

View file

@ -4,9 +4,11 @@ Calls Google Programmable Search Engine (PSE) API to search the web.
Google PSE API Reference: https://developers.google.com/custom-search/v1/reference/rest/v1/cse/list
"""
from collections.abc import Iterable, Mapping
from typing import Final, Literal, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -16,6 +18,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _GooglePSESearchRequestRequired(TypedDict):
"""Required fields for Google PSE Search API request."""
@ -239,17 +244,20 @@ class GooglePSESearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final = []
for item in response_json.get("items", []):
search_result = SearchResult(
title=item.get("title", ""),
url=item.get("link", ""),
snippet=item.get("snippet", ""),
date=None, # Google PSE doesn't provide date in standard response
last_updated=None, # Google PSE doesn't provide last_updated in response
for raw_item in _JSON_VALUES.validate_python(response_json.get("items", [])):
item = _JSON_OBJECT.validate_python(raw_item)
search_result = SearchResult.model_validate(
{
"title": item.get("title", ""),
"url": item.get("link", ""),
"snippet": item.get("snippet", ""),
"date": None, # Google PSE doesn't provide date in standard response
"last_updated": None, # Google PSE doesn't provide last_updated in response
}
)
results.append(search_result)

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
@ -11,6 +13,9 @@ from litellm.types.utils import EmbeddingResponse, Usage
from ..common_utils import InfinityError
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_INT: Final = TypeAdapter(int)
class InfinityEmbeddingConfig(BaseEmbeddingConfig):
"""
@ -124,9 +129,10 @@ class InfinityEmbeddingConfig(BaseEmbeddingConfig):
model_response.data = raw_response_json.get("data")
model_response.object = raw_response_json.get("object")
usage_payload: Final = _JSON_OBJECT.validate_python(raw_response_json.get("usage", {}))
usage: Final = Usage(
prompt_tokens=raw_response_json.get("usage", {}).get("prompt_tokens", 0),
total_tokens=raw_response_json.get("usage", {}).get("total_tokens", 0),
prompt_tokens=_INT.validate_python(usage_payload.get("prompt_tokens", 0) or 0),
total_tokens=_INT.validate_python(usage_payload.get("total_tokens", 0) or 0),
)
model_response.usage = usage
return model_response

View file

@ -10,7 +10,7 @@ Non-streaming endpoint: POST /runs/wait
import json
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final, Optional, Union, cast
from typing import TYPE_CHECKING, Any, Final, Optional, Union
import httpx
@ -369,7 +369,7 @@ class LangGraphConfig(BaseConfig):
from litellm.utils import CustomStreamWrapper
if client is None or not isinstance(client, AsyncHTTPHandler):
client = get_async_httpx_client(llm_provider=cast(Any, "langgraph"), params={})
client = get_async_httpx_client(llm_provider="langgraph", params={})
verbose_logger.debug("Making async streaming request to: %s", api_base)

View file

@ -2,10 +2,12 @@
Translate from OpenAI's `/v1/chat/completions` to Lemonade's `/v1/chat/completions`
"""
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
from urllib.parse import quote
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm._logging import verbose_logger
@ -21,6 +23,10 @@ from ...openai_like.chat.transformation import OpenAILikeChatConfig
if TYPE_CHECKING:
from litellm.litellm_core_utils.tokenizer import Encoding as Tokenizer
_JSON_DICT: Final = TypeAdapter(dict[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 LemonadeChatConfig(OpenAILikeChatConfig):
_DEFAULT_API_KEY = "lemonade"
@ -55,7 +61,7 @@ class LemonadeChatConfig(OpenAILikeChatConfig):
response_format: dict | None = None,
tools: list | None = None,
) -> None:
locals_: Final = locals().copy()
locals_: Final[Mapping[str, object]] = dict(locals())
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)
@ -104,8 +110,8 @@ class LemonadeChatConfig(OpenAILikeChatConfig):
f"Failed to fetch models from Lemonade. Status code: {response.status_code}, Response: {response.text}"
)
model_list: Final = response.json().get("data", [])
return ["lemonade/" + model["id"] for model in model_list]
model_list: Final = _JSON_OBJECTS.validate_python(_JSON_DICT.validate_python(response.json()).get("data", []))
return ["lemonade/" + _STR.validate_python(model["id"]) for model in model_list]
@staticmethod
def _get_positive_int(value: Any) -> int | None:
@ -188,10 +194,11 @@ class LemonadeChatConfig(OpenAILikeChatConfig):
verbose_logger.debug("LemonadeError: Could not get model info.")
return self._get_default_model_info(model)
max_input_tokens: Final = self._get_context_window(model_info)
max_output_tokens: Final = self._get_positive_int(model_info.get("max_output_tokens"))
max_tokens: Final = self._get_positive_int(model_info.get("max_tokens"))
provider_specific_entry: Final = self._get_provider_specific_entry(model_info)
model_info_fields: Final = _JSON_DICT.validate_python(model_info)
max_input_tokens: Final = self._get_context_window(model_info_fields)
max_output_tokens: Final = self._get_positive_int(model_info_fields.get("max_output_tokens"))
max_tokens: Final = self._get_positive_int(model_info_fields.get("max_tokens"))
provider_specific_entry: Final = self._get_provider_specific_entry(model_info_fields)
model_info_response: Final = self._get_default_model_info(model)
model_info_response.update(

View file

@ -4,9 +4,11 @@ Calls Linkup's /search endpoint to search the web.
Linkup API Reference: https://docs.linkup.so/pages/documentation/api-reference/endpoint/post-search
"""
from collections.abc import Iterable, Mapping
from typing import Final, Literal, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -16,6 +18,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _LinkupSearchRequestRequired(TypedDict):
"""Required fields for Linkup Search API request."""
@ -165,35 +170,40 @@ class LinkupSearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final = []
# Process results array
raw_results: Final = response_json.get("results", [])
raw_results: Final = _JSON_VALUES.validate_python(response_json.get("results", []))
for result in raw_results:
for raw_result in raw_results:
result = _JSON_OBJECT.validate_python(raw_result)
# Handle both text and image result types
result_type = result.get("type", "text")
if result_type == "text":
search_result = SearchResult(
title=result.get("name", ""),
url=result.get("url", ""),
snippet=result.get("content", ""),
date=None,
last_updated=None,
search_result = SearchResult.model_validate(
{
"title": result.get("name", ""),
"url": result.get("url", ""),
"snippet": result.get("content", ""),
"date": None,
"last_updated": None,
}
)
results.append(search_result)
elif result_type == "image":
# For image results, use the URL as both title and snippet if name not provided
search_result = SearchResult(
title=result.get("name", result.get("url", "")),
url=result.get("url", ""),
snippet=result.get("content", ""),
date=None,
last_updated=None,
search_result = SearchResult.model_validate(
{
"title": result.get("name", result.get("url", "")),
"url": result.get("url", ""),
"snippet": result.get("content", ""),
"date": None,
"last_updated": None,
}
)
results.append(search_result)

View file

@ -11,6 +11,7 @@ from types import MappingProxyType
from typing import TYPE_CHECKING, Final
import httpx
from pydantic import ConfigDict, TypeAdapter, ValidationError
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.text_to_speech.transformation import (
@ -23,6 +24,8 @@ if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.types.llms.openai import HttpxBinaryResponseContent
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class MistralTextToSpeechException(BaseLLMException):
pass
@ -143,7 +146,7 @@ class MistralTextToSpeechConfig(BaseTextToSpeechConfig):
return request_data
def _requested_content_type(self, request: httpx.Request) -> str:
request_body: Final = json.loads(request.content or b"{}")
request_body: Final = _JSON_OBJECT.validate_python(json.loads(request.content or b"{}"))
requested_format: Final = request_body.get("response_format")
if not isinstance(requested_format, str):
return "audio/mpeg"
@ -158,7 +161,9 @@ class MistralTextToSpeechConfig(BaseTextToSpeechConfig):
from litellm.types.llms.openai import HttpxBinaryResponseContent
try:
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
except ValidationError:
raise
except (json.JSONDecodeError, ValueError):
raise MistralTextToSpeechException(
status_code=raw_response.status_code,

View file

@ -1,5 +1,6 @@
import json
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -79,7 +80,7 @@ class NLPCloudConfig(BaseConfig):
num_beams: int | None = None,
num_return_sequences: int | None = None,
) -> None:
locals_: Final = locals().copy()
locals_: Final[Mapping[str, object]] = dict(locals())
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)

View file

@ -759,7 +759,7 @@ class OCIStreamWrapper(CustomStreamWrapper):
self.sent_last_chunk = True
return self.model_response_creator(chunk={"choices": parsed.choices})
def chunk_creator(self, chunk: Any) -> ModelResponseStream | None:
def chunk_creator(self, chunk: object) -> ModelResponseStream | None:
if not isinstance(chunk, str):
raise ValueError(f"Chunk is not a string: {chunk}")
if not chunk.startswith("data:"):

View file

@ -1,6 +1,6 @@
import base64
import io
from typing import Any, Final
from typing import Final
import httpx
@ -174,7 +174,7 @@ class OllamaModelInfo(BaseLLMModelInfo):
model: str,
api_base: str | None = None,
api_key: str | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
from litellm import module_level_client
model = self._strip_ollama_model_prefix(model)
@ -223,7 +223,7 @@ class OllamaModelInfo(BaseLLMModelInfo):
model: str,
api_base: str | None = None,
api_key: str | None = None,
) -> dict[str, Any] | None:
) -> dict[str, object] | None:
if self._is_static_ollama_model(model):
return None
return self.get_runtime_model_info(model=model, api_base=api_base, api_key=api_key)

View file

@ -97,6 +97,7 @@ class _OllamaGenerateResponse(TypedDict):
_OLLAMA_GENERATE_RESPONSE: Final = TypeAdapter(_OllamaGenerateResponse)
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_CODE_FENCE: Final = re.compile(r"^```(?:json)?\s*(.*?)\s*(?:```)?$", re.DOTALL)
_JSON_OBJECT_START: Final = re.compile(r"(?:```(?:json)?\s*)?\{")
_JSON_OBJECT_PARTIAL_START: Final = re.compile(r"`{0,3}|```(?:j|js|jso|json)?\s*")
@ -334,7 +335,9 @@ class OllamaConfig(BaseConfig):
model_response.choices[0].finish_reason = "stop"
else:
try:
response_content: Final[object] = json.loads(_strip_json_code_fence(response_text))
response_content: Final = _JSON_VALUE.validate_python(
json.loads(_strip_json_code_fence(response_text))
)
# Check if this is a function call format with name/arguments structure
if (

View file

@ -6,7 +6,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
from typing import Final
import httpx
@ -33,7 +33,7 @@ class OpenAICountTokensHandler(OpenAICountTokensConfig):
timeout: float | httpx.Timeout | None = None,
tools: list[dict[str, object]] | None = None,
instructions: str | None = None,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Handle a token counting request to OpenAI's Responses API.

View file

@ -81,12 +81,14 @@ class OpenAITokenCounter(BaseTokenCounter):
)
if result is not None:
return TokenCountResponse(
total_tokens=result.get("input_tokens", 0),
request_model=request_model,
model_used=model_to_use,
tokenizer_type="openai_api",
original_response=result,
return TokenCountResponse.model_validate(
{
"total_tokens": result.get("input_tokens", 0),
"request_model": request_model,
"model_used": model_to_use,
"tokenizer_type": "openai_api",
"original_response": result,
}
)
except OpenAIError as e:
verbose_logger.warning("OpenAI CountTokens API error: status=%s, message=%s", e.status_code, e.message)

View file

@ -118,7 +118,7 @@ class OpenAICountTokensConfig:
self,
model: str,
input: str | Sequence[object],
tools: list[dict[str, Any]] | None = None,
tools: list[dict[str, object]] | None = None,
instructions: str | None = None,
) -> dict[str, object]:
"""

View file

@ -6,11 +6,12 @@ Calls done in OpenAI/openai.py as OpenRouter is openai-compatible.
Docs: https://openrouter.ai/docs/parameters
"""
from collections.abc import AsyncIterator, Iterator
from collections.abc import AsyncIterator, Iterator, Mapping
from enum import Enum
from typing import TYPE_CHECKING, Any, Final, cast
from typing import TYPE_CHECKING, Final, cast
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, set_hidden_params
from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
@ -26,6 +27,8 @@ if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.tokenizer import Encoding as Tokenizer
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
class CacheControlSupportedModels(str, Enum):
"""Models that support cache_control in content blocks."""
@ -203,9 +206,9 @@ class OpenrouterConfig(OpenAIGPTConfig):
# Extract cost from OpenRouter response body
# OpenRouter returns cost information in the usage object when usage.include=true
try:
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
if "usage" in response_json and response_json["usage"]:
response_cost: Final = response_json["usage"].get("cost")
response_cost: Final = _JSON_OBJECT.validate_python(response_json["usage"]).get("cost")
if response_cost is not None:
# Store cost in hidden params for the cost calculator to use
if not hasattr(model_response, HIDDEN_PARAMS_ATTR):
@ -218,7 +221,8 @@ class OpenrouterConfig(OpenAIGPTConfig):
additional_headers: Final = cast( # cast-ok: preserve mapping operations on response metadata
dict[str, object], hidden_params["additional_headers"]
)
additional_headers["llm_provider-x-litellm-response-cost"] = float(response_cost)
if isinstance(response_cost, str | int | float):
additional_headers["llm_provider-x-litellm-response-cost"] = float(response_cost)
except Exception:
# If we can't extract cost, continue without it - don't fail the response
pass
@ -237,7 +241,7 @@ class OpenrouterConfig(OpenAIGPTConfig):
streaming_response: Iterator[str] | AsyncIterator[str] | ModelResponse,
sync_stream: bool,
json_mode: bool | None = False,
) -> Any:
) -> "OpenRouterChatCompletionStreamingHandler":
return OpenRouterChatCompletionStreamingHandler(
streaming_response=streaming_response,
sync_stream=sync_stream,

View file

@ -1,9 +1,11 @@
import asyncio
import json
import time
from collections.abc import Mapping
from typing import Final, cast
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.constants import (
OPEN_SANDBOX_API_BASE_ENV_VAR,
@ -37,6 +39,9 @@ DEFAULT_READY_TIMEOUT: Final = OPEN_SANDBOX_READY_TIMEOUT
DEFAULT_POLL_INTERVAL: Final = OPEN_SANDBOX_POLL_INTERVAL
MAX_OUTPUT_BYTES: Final = SANDBOX_MAX_OUTPUT_BYTES
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUE: Final = TypeAdapter(object)
class OpenSandboxSandboxConfig(BaseSandboxConfig):
def _http(self, client: AsyncHTTPHandler | None) -> AsyncHTTPHandler:
@ -92,7 +97,7 @@ class OpenSandboxSandboxConfig(BaseSandboxConfig):
headers=self._lifecycle_headers(key),
json=body,
)
data: Final = response.json()
data: Final = _JSON_OBJECT.validate_python(response.json())
sandbox_id: Final = str(data["id"])
if self._sandbox_state(data) != "Running":
@ -256,7 +261,7 @@ class OpenSandboxSandboxConfig(BaseSandboxConfig):
url=f"{api_base}/sandboxes/{sandbox_id}",
headers=headers,
)
data = response.json()
data = _JSON_VALUE.validate_python(response.json())
state = self._sandbox_state(data)
if state == "Running":
return

View file

@ -3,9 +3,11 @@ This is OpenAI compatible - no transformation is applied
"""
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
@ -16,6 +18,9 @@ from litellm.types.utils import EmbeddingResponse, Usage
from ..utils import OVHCloudException
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_INT: Final = TypeAdapter(int)
class OVHCloudEmbeddingConfig(BaseEmbeddingConfig):
def __init__(self) -> None:
@ -107,9 +112,10 @@ class OVHCloudEmbeddingConfig(BaseEmbeddingConfig):
model_response.data = raw_response_json.get("data")
model_response.object = raw_response_json.get("object")
usage_payload: Final = _JSON_OBJECT.validate_python(raw_response_json.get("usage", {}))
usage: Final = Usage(
prompt_tokens=raw_response_json.get("usage", {}).get("prompt_tokens", 0),
total_tokens=raw_response_json.get("usage", {}).get("total_tokens", 0),
prompt_tokens=_INT.validate_python(usage_payload.get("prompt_tokens", 0) or 0),
total_tokens=_INT.validate_python(usage_payload.get("total_tokens", 0) or 0),
)
model_response.usage = usage

View file

@ -7,6 +7,7 @@ from types import MappingProxyType
from typing import TYPE_CHECKING, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm._logging import verbose_logger
@ -20,6 +21,7 @@ if TYPE_CHECKING:
from litellm.litellm_core_utils.tokenizer import Encoding as Tokenizer
_PERPLEXITY_ATTRIBUTION_HEADERS: Final[Mapping[str, str]] = MappingProxyType({"X-Pplx-Integration": "litellm"})
_JSON_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(strict=True, hide_input_in_errors=True))
class PerplexityChatConfig(OpenAIGPTConfig):
@ -110,7 +112,7 @@ class PerplexityChatConfig(OpenAIGPTConfig):
# Extract and enhance usage with Perplexity-specific fields
try:
raw_response_json: Final = raw_response.json()
raw_response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
self._enhance_usage_with_perplexity_fields(model_response, raw_response_json)
self._add_citations_as_annotations(model_response, raw_response_json)
except Exception as e:

View file

@ -13,9 +13,11 @@ This module decodes them into float arrays for OpenAI-compatible responses.
import base64
import struct
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
@ -24,6 +26,9 @@ from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllEmbeddingInputValues, AllMessageValues
from litellm.types.utils import EmbeddingResponse, Usage
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_INT: Final = TypeAdapter(int)
class PerplexityEmbeddingError(BaseLLMException):
def __init__(
@ -156,15 +161,17 @@ class PerplexityEmbeddingConfig(BaseEmbeddingConfig):
raw_data: Final = raw_response_json.get("data", [])
decoded_data: Final[list[dict[str, object]]] = []
for item in raw_data:
decoded_item = dict(item)
decoded_item["embedding"] = self._decode_base64_embedding(item.get("embedding"))
decoded_item = dict(_JSON_OBJECT.validate_python(item))
decoded_item["embedding"] = self._decode_base64_embedding(decoded_item.get("embedding"))
decoded_data.append(decoded_item)
model_response.data = decoded_data
usage_data: Final = raw_response_json.get("usage", {})
usage_data: Final = _JSON_OBJECT.validate_python(raw_response_json.get("usage", {}))
usage: Final = Usage(
prompt_tokens=usage_data.get("prompt_tokens", 0) or usage_data.get("total_tokens", 0),
total_tokens=usage_data.get("total_tokens", 0),
prompt_tokens=_INT.validate_python(
usage_data.get("prompt_tokens", 0) or usage_data.get("total_tokens", 0) or 0
),
total_tokens=_INT.validate_python(usage_data.get("total_tokens", 0) or 0),
)
model_response.usage = usage
return model_response

View file

@ -2,9 +2,11 @@
Calls Perplexity's /search endpoint to search the web.
"""
from collections.abc import Iterable, Mapping
from typing import Final, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -14,6 +16,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _PerplexitySearchRequestRequired(TypedDict):
"""Required fields for Perplexity Search API request."""
@ -146,17 +151,20 @@ class PerplexitySearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final = []
for result in response_json.get("results", []):
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("url", ""),
snippet=result.get("snippet", ""),
date=result.get("date"),
last_updated=result.get("last_updated"),
for raw_result in _JSON_VALUES.validate_python(response_json.get("results", [])):
result = _JSON_OBJECT.validate_python(raw_result)
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": result.get("snippet", ""),
"date": result.get("date"),
"last_updated": result.get("last_updated"),
}
)
results.append(search_result)

View file

@ -1,7 +1,7 @@
import asyncio
import json
import time
from collections.abc import Callable
from collections.abc import Callable, Iterable
from typing import Final
from pydantic import ConfigDict, TypeAdapter
@ -23,6 +23,8 @@ from .transformation import ReplicateConfig
replicate_config: Final = ReplicateConfig()
_JSON_OBJECT: Final = TypeAdapter(dict[str, object], config=ConfigDict(strict=True))
_POLLED_PREDICTION: Final = TypeAdapter(dict[str, object], config=ConfigDict(strict=True, hide_input_in_errors=True))
_TEXTS: Final = TypeAdapter(Iterable[str], config=ConfigDict(strict=True, hide_input_in_errors=True))
# Function to handle prediction response (streaming)
@ -38,11 +40,11 @@ def handle_prediction_response_streaming(
print_verbose(f"replicate: polling endpoint: {prediction_url}")
response = http_client.get(prediction_url, headers=headers)
if response.status_code == 200:
response_data = response.json()
response_data = _POLLED_PREDICTION.validate_python(response.json())
status = response_data["status"]
if "output" in response_data:
try:
output_string = "".join(response_data["output"])
output_string = "".join(_TEXTS.validate_python(response_data["output"]))
except Exception:
raise ReplicateError(
status_code=422,

View file

@ -1,4 +1,4 @@
from collections.abc import Iterable
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -29,6 +29,7 @@ else:
LoggingClass = Any
_TEXTS: Final = TypeAdapter(Iterable[str], config=ConfigDict(strict=True, hide_input_in_errors=True))
_JSON_DICT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
class ReplicateConfig(BaseConfig):
@ -79,7 +80,7 @@ class ReplicateConfig(BaseConfig):
seed: int | None = None,
debug: bool | None = None,
) -> None:
locals_: Final = locals().copy()
locals_: Final[Mapping[str, object]] = dict(locals())
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)
@ -250,7 +251,7 @@ class ReplicateConfig(BaseConfig):
original_response=raw_response.text,
additional_args={"complete_input_dict": request_data},
)
raw_response_json: Final = raw_response.json()
raw_response_json: Final = _JSON_DICT.validate_python(raw_response.json())
if raw_response_json.get("status") != "succeeded":
raise ReplicateError(
status_code=422,

View file

@ -3,9 +3,11 @@ This is OpenAI compatible - no transformation is applied
"""
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
@ -16,6 +18,9 @@ from litellm.types.utils import EmbeddingResponse, Usage
from ..common_utils import SambaNovaError
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_INT: Final = TypeAdapter(int)
class SambaNovaEmbeddingConfig(BaseEmbeddingConfig):
def __init__(self) -> None:
@ -124,9 +129,10 @@ class SambaNovaEmbeddingConfig(BaseEmbeddingConfig):
model_response.data = raw_response_json.get("data")
model_response.object = raw_response_json.get("object")
usage_payload: Final = _JSON_OBJECT.validate_python(raw_response_json.get("usage", {}))
usage: Final = Usage(
prompt_tokens=raw_response_json.get("usage", {}).get("prompt_tokens", 0),
total_tokens=raw_response_json.get("usage", {}).get("total_tokens", 0),
prompt_tokens=_INT.validate_python(usage_payload.get("prompt_tokens", 0) or 0),
total_tokens=_INT.validate_python(usage_payload.get("total_tokens", 0) or 0),
)
model_response.usage = usage

View file

@ -7,6 +7,7 @@ from functools import cached_property
from typing import TYPE_CHECKING, Any, Final, Union
import httpx
from pydantic import BaseModel
import litellm
from litellm.types.llms.openai import AllMessageValues
@ -54,7 +55,7 @@ _SAP_MODEL_PARAMS_EXCLUDED_KEYS: Final[frozenset[str]] = frozenset(
def validate_dict(
data: dict, model: type
data: dict, model: type[BaseModel]
) -> dict: # mutable-ok: pydantic validation boundary; both input and output are untyped wire dicts
return model(**data).model_dump(by_alias=True, exclude_unset=True)

View file

@ -4,10 +4,12 @@ Calls SearchAPI.io's Google Search API endpoint.
SearchAPI.io API Reference: https://www.searchapi.io/docs/google
"""
from collections.abc import Iterable, Mapping
from typing import Final, Literal, TypedDict, cast
from urllib.parse import urlencode
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -17,6 +19,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _SearchAPIRequestRequired(TypedDict):
"""Required fields for SearchAPI.io request."""
@ -213,24 +218,27 @@ class SearchAPIConfig(BaseSearchConfig):
- organic_results[].snippet → SearchResult.snippet
- organic_results[].date → SearchResult.date
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final[list[SearchResult]] = []
# Process organic results
for result in response_json.get("organic_results", []):
for raw_result in _JSON_VALUES.validate_python(response_json.get("organic_results", [])):
result = _JSON_OBJECT.validate_python(raw_result)
title = result.get("title", "")
url = result.get("link", "")
snippet = result.get("snippet", "")
date = result.get("date") # SearchAPI.io provides date in some results
search_result = SearchResult(
title=title,
url=url,
snippet=snippet,
date=date,
last_updated=None, # SearchAPI.io doesn't provide last_updated
search_result = SearchResult.model_validate(
{
"title": title,
"url": url,
"snippet": snippet,
"date": date,
"last_updated": None, # SearchAPI.io doesn't provide last_updated
}
)
results.append(search_result)

View file

@ -4,9 +4,11 @@ Calls SearXNG's /search endpoint to search the web.
SearXNG API Reference: https://docs.searxng.org/dev/search_api.html
"""
from collections.abc import Iterable, Mapping
from typing import Final, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -16,6 +18,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _SearXNGSearchRequestRequired(TypedDict):
"""Required fields for SearXNG Search API request."""
@ -204,22 +209,25 @@ class SearXNGSearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
# Note: SearXNG doesn't natively support limiting results via API params
# It returns ~20 results per page by default
results: Final = []
for result in response_json.get("results", []):
for raw_result in _JSON_VALUES.validate_python(response_json.get("results", [])):
result = _JSON_OBJECT.validate_python(raw_result)
# Get date from either publishedDate or pubdate field
date = result.get("publishedDate") or result.get("pubdate")
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("url", ""),
snippet=result.get("content", ""), # SearXNG uses "content" for snippet
date=date,
last_updated=None, # SearXNG doesn't provide last_updated in response
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": result.get("content", ""), # SearXNG uses "content" for snippet
"date": date,
"last_updated": None, # SearXNG doesn't provide last_updated in response
}
)
results.append(search_result)

View file

@ -4,9 +4,11 @@ Calls Serper's /search endpoint to search Google.
Serper API Reference: https://serper.dev
"""
from collections.abc import Iterable, Mapping
from typing import Final, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -16,6 +18,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _SerperSearchRequestRequired(TypedDict):
"""Required fields for Serper Search API request."""
@ -156,16 +161,19 @@ class SerperSearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
results: Final = []
for result in response_json.get("organic", []):
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("link", ""),
snippet=result.get("snippet", ""),
date=result.get("date"),
last_updated=None,
for raw_result in _JSON_VALUES.validate_python(response_json.get("organic", [])):
result = _JSON_OBJECT.validate_python(raw_result)
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("link", ""),
"snippet": result.get("snippet", ""),
"date": result.get("date"),
"last_updated": None,
}
)
results.append(search_result)

View file

@ -4,9 +4,11 @@ Calls Tavily's /search endpoint to search the web.
Tavily API Reference: https://docs.tavily.com/documentation/api-reference/endpoint/search
"""
from collections.abc import Iterable, Mapping
from typing import Final, TypedDict
import httpx
from pydantic import ConfigDict, TypeAdapter
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.search.transformation import (
@ -16,6 +18,9 @@ from litellm.llms.base_llm.search.transformation import (
)
from litellm.secret_managers.main import get_secret_str
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
class _TavilySearchRequestRequired(TypedDict):
"""Required fields for Tavily Search API request."""
@ -174,17 +179,20 @@ class TavilySearchConfig(BaseSearchConfig):
Returns:
SearchResponse with standardized format
"""
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
# Transform results to SearchResult objects
results: Final = []
for result in response_json.get("results", []):
search_result = SearchResult(
title=result.get("title", ""),
url=result.get("url", ""),
snippet=result.get("content", ""), # Tavily uses "content" instead of "snippet"
date=None, # Tavily doesn't provide date in response
last_updated=None, # Tavily doesn't provide last_updated in response
for raw_result in _JSON_VALUES.validate_python(response_json.get("results", [])):
result = _JSON_OBJECT.validate_python(raw_result)
search_result = SearchResult.model_validate(
{
"title": result.get("title", ""),
"url": result.get("url", ""),
"snippet": result.get("content", ""), # Tavily uses "content" instead of "snippet"
"date": None, # Tavily doesn't provide date in response
"last_updated": None, # Tavily doesn't provide last_updated in response
}
)
results.append(search_result)

View file

@ -672,7 +672,7 @@ def _openai_batch_jsonl_entry_to_vertex_embeddings_rows(
def _openai_batch_jsonl_entry_to_vertex_rows(
openai_entry: dict[str, Any],
map_openai_to_vertex_params: Callable[[dict[str, Any]], dict[str, object]],
map_openai_to_vertex_params: Callable[[dict[str, object]], dict[str, object]],
) -> tuple[Mapping[str, object], ...]:
"""
Transforms a single OpenAI JSONL batch entry into the Vertex rows it maps to.

View file

@ -5,6 +5,7 @@ from datetime import datetime
from typing import Final, Literal
import httpx
from pydantic import TypeAdapter
import litellm
from litellm._logging import verbose_logger
@ -23,6 +24,8 @@ from litellm.types.llms.vertex_ai import (
)
from litellm.types.utils import LiteLLMFineTuningJob
_JSON_VALUE: Final = TypeAdapter(object)
class VertexFineTuningAPI(VertexLLM):
"""
@ -341,4 +344,4 @@ class VertexFineTuningAPI(VertexLLM):
f"Error creating fine tuning job. Status code: {response.status_code}. Response: {response.text}"
)
return response.json()
return _JSON_VALUE.validate_python(response.json())

View file

@ -1,13 +1,14 @@
import base64
import json
import os
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from io import BufferedRandom, BufferedReader, BytesIO
from pathlib import Path
from typing import TYPE_CHECKING, Any, Final, cast
import httpx
from httpx._types import RequestFiles
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH
@ -26,6 +27,11 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_CONTAINER: Final[TypeAdapter[Mapping[str, object] | str | Sequence[object]]] = TypeAdapter(
Mapping[str, object] | str | Sequence[object], config=ConfigDict(hide_input_in_errors=True)
)
class VertexAIImagenImageEditConfig(BaseImageEditConfig, VertexLLM):
"""
@ -208,15 +214,18 @@ class VertexAIImagenImageEditConfig(BaseImageEditConfig, VertexLLM):
headers=raw_response.headers,
)
predictions: Final = response_json.get("predictions", [])
predictions: Final = _JSON_CONTAINER.validate_python(response_json.get("predictions", []))
data_list: Final[list[ImageObject]] = []
for prediction in predictions:
for raw_prediction in predictions:
prediction = (
raw_prediction if isinstance(raw_prediction, str) else _JSON_CONTAINER.validate_python(raw_prediction)
)
# Imagen returns images as bytesBase64Encoded
if "bytesBase64Encoded" in prediction:
data_list.append(
ImageObject(
b64_json=prediction["bytesBase64Encoded"],
b64_json=_JSON_OBJECT.validate_python(prediction)["bytesBase64Encoded"],
url=None,
)
)

View file

@ -1,5 +1,5 @@
import os
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -35,6 +35,9 @@ else:
_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_CONTAINER: Final[TypeAdapter[Mapping[str, object] | str | Sequence[object]]] = TypeAdapter(
Mapping[str, object] | str | Sequence[object], config=ConfigDict(hide_input_in_errors=True)
)
class VertexAIGeminiImageGenerationConfig(BaseImageGenerationConfig, VertexLLM):
@ -308,25 +311,27 @@ class VertexAIGeminiImageGenerationConfig(BaseImageGenerationConfig, VertexLLM):
model_response.data = []
# Gemini image generation models return in candidates format
candidates: Final = response_data.get("candidates", [])
response_object: Final = _JSON_OBJECT.validate_python(response_data)
candidates: Final = _JSON_CONTAINER.validate_python(response_object.get("candidates", []))
for candidate in candidates:
content = candidate.get("content", {})
parts = content.get("parts", [])
for part in parts:
content = _JSON_OBJECT.validate_python(_JSON_OBJECT.validate_python(candidate).get("content", {}))
parts = _JSON_CONTAINER.validate_python(content.get("parts", []))
for raw_part in parts:
part = raw_part if isinstance(raw_part, str) else _JSON_CONTAINER.validate_python(raw_part)
# Look for inlineData with image
if "inlineData" in part:
inline_data = part["inlineData"]
part_object = _JSON_OBJECT.validate_python(part)
inline_data = _JSON_CONTAINER.validate_python(part_object["inlineData"])
if "data" in inline_data:
thought_sig = part.get("thoughtSignature")
thought_sig = part_object.get("thoughtSignature")
model_response.data.append(
ImageObject(
b64_json=inline_data["data"],
b64_json=_JSON_OBJECT.validate_python(inline_data)["data"],
url=None,
provider_specific_fields=({"thought_signature": thought_sig} if thought_sig else None),
)
)
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))

View file

@ -1,7 +1,9 @@
import os
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.llms.base_llm.image_generation.transformation import (
@ -27,6 +29,12 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_CONTAINER: Final[TypeAdapter[Mapping[str, object] | str | Sequence[object]]] = TypeAdapter(
Mapping[str, object] | str | Sequence[object], config=ConfigDict(hide_input_in_errors=True)
)
class VertexAIImagenImageGenerationConfig(BaseImageGenerationConfig, VertexLLM):
"""
@ -221,7 +229,7 @@ class VertexAIImagenImageGenerationConfig(BaseImageGenerationConfig, VertexLLM):
Transform Imagen image generation response to litellm ImageResponse format
"""
try:
response_data: Final = raw_response.json()
response_data: Final = _JSON_VALUE.validate_python(raw_response.json())
except Exception as e:
raise self.get_error_class(
error_message=f"Error transforming image generation response: {e}",
@ -233,13 +241,18 @@ class VertexAIImagenImageGenerationConfig(BaseImageGenerationConfig, VertexLLM):
model_response.data = []
# Imagen format - predictions with generated images
predictions: Final = response_data.get("predictions", [])
for prediction in predictions:
predictions: Final = _JSON_CONTAINER.validate_python(
_JSON_OBJECT.validate_python(response_data).get("predictions", [])
)
for raw_prediction in predictions:
prediction = (
raw_prediction if isinstance(raw_prediction, str) else _JSON_CONTAINER.validate_python(raw_prediction)
)
# Imagen returns images as bytesBase64Encoded
if "bytesBase64Encoded" in prediction:
model_response.data.append(
ImageObject(
b64_json=prediction["bytesBase64Encoded"],
b64_json=_JSON_OBJECT.validate_python(prediction)["bytesBase64Encoded"],
url=None,
)
)

View file

@ -1,6 +1,8 @@
from collections.abc import Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
from typing_extensions import TypedDict
import litellm
@ -12,6 +14,9 @@ from litellm.llms.openai.openai import HttpxBinaryResponseContent
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import VertexLLM
from litellm.types.llms.vertex_ai import VERTEX_CREDENTIALS_TYPES
_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 VertexInput(TypedDict, total=False):
text: str | None
@ -150,12 +155,12 @@ class VertexTextToSpeechAPI(VertexLLM):
if response.status_code != 200:
raise Exception(f"Request failed with status code {response.status_code}, {response.text}")
############ Process the response ############
_json_response: Final = response.json()
_json_response: Final = _JSON_OBJECT.validate_python(response.json())
response_content: Final = _json_response["audioContent"]
# Decode base64 to get binary content
binary_data: Final = base64.b64decode(response_content)
binary_data: Final = base64.b64decode(_STR.validate_python(response_content))
# Create an httpx.Response object
response = httpx.Response(
@ -187,12 +192,12 @@ class VertexTextToSpeechAPI(VertexLLM):
if response.status_code != 200:
raise Exception(f"Request did not return a 200 status code: {response.status_code}, {response.text}")
_json_response: Final = response.json()
_json_response: Final = _JSON_OBJECT.validate_python(response.json())
response_content: Final = _json_response["audioContent"]
# Decode base64 to get binary content
binary_data: Final = base64.b64decode(response_content)
binary_data: Final = base64.b64decode(_STR.validate_python(response_content))
# Create an httpx.Response object
response = httpx.Response(

View file

@ -47,6 +47,7 @@ else:
_LyriaVoice: TypeAlias = str | dict | None
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_ARRAY: Final = TypeAdapter(tuple[object, ...], config=ConfigDict(hide_input_in_errors=True))
_STR: Final = TypeAdapter(str, config=ConfigDict(hide_input_in_errors=True))
@ -650,27 +651,37 @@ class VertexAILyriaTextToSpeechConfig(VertexAITextToSpeechConfig):
) -> "HttpxBinaryResponseContent":
from litellm.types.llms.openai import HttpxBinaryResponseContent
response_json: Final = raw_response.json()
response_json: Final = _JSON_OBJECT.validate_python(raw_response.json())
base_model: Final = model.removeprefix("vertex_ai/")
model_info: Final = self._get_model_info(model=model)
audio_data: str | None = None # rebind-ok: response parsing discovers audio data in provider-specific shapes
mime_type: str | None = None # rebind-ok: response parsing discovers the MIME type beside the audio payload
audio_data: object = None # rebind-ok: response parsing discovers audio data in provider-specific shapes
mime_type: object = None # rebind-ok: response parsing discovers the MIME type beside the audio payload
if model_info["vertex_ai_audio_api"] == "lyria_predict":
predictions: Final = response_json.get("predictions") or ()
predictions: Final = _JSON_ARRAY.validate_python(response_json.get("predictions") or ())
if predictions:
audio_data = predictions[0].get("audioContent") or predictions[0].get("bytesBase64Encoded")
mime_type = predictions[0].get("mimeType") # rebind-ok: predict response supplies its audio MIME type
prediction: Final = _JSON_OBJECT.validate_python(predictions[0])
audio_data = prediction.get("audioContent") or prediction.get("bytesBase64Encoded")
mime_type = prediction.get("mimeType") # rebind-ok: predict response supplies its audio MIME type
else:
for step in response_json.get("steps") or response_json.get("outputs") or ():
content_items = step.get("content") or () if step.get("type") == "model_output" else (step,)
steps: Final = _JSON_ARRAY.validate_python(response_json.get("steps") or response_json.get("outputs") or ())
for step in map(_JSON_OBJECT.validate_python, steps):
content_items = (
map(_JSON_OBJECT.validate_python, _JSON_ARRAY.validate_python(step.get("content") or ()))
if step.get("type") == "model_output"
else (step,)
)
for content in content_items:
if content.get("type") == "audio" and content.get("data"):
audio_data = content["data"]
mime_type = content.get("mime_type")
if audio_data is None:
raise ValueError(f"No generated audio found in Vertex AI {base_model} response")
binary_data: Final = base64.b64decode(audio_data)
media_type: Final = mime_type or speech_media_type_from_audio_bytes(binary_data) or DEFAULT_SPEECH_MEDIA_TYPE
binary_data: Final = base64.b64decode(_STR.validate_python(audio_data))
media_type: Final = (
(_STR.validate_python(mime_type) if mime_type else None)
or speech_media_type_from_audio_bytes(binary_data)
or DEFAULT_SPEECH_MEDIA_TYPE
)
return HttpxBinaryResponseContent(
httpx.Response(
status_code=raw_response.status_code,

View file

@ -1,6 +1,7 @@
from typing import Final, Literal
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
@ -17,6 +18,9 @@ from litellm.types.utils import EmbeddingResponse
from .types import *
_JSON_VALUE: Final = TypeAdapter(object)
_JSON_DICT: Final = TypeAdapter(dict[str, object], config=ConfigDict(hide_input_in_errors=True))
class VertexEmbedding(VertexBase):
def __init__(self) -> None:
@ -123,12 +127,12 @@ class VertexEmbedding(VertexBase):
except httpx.TimeoutException:
raise VertexAIError(status_code=408, message="Timeout error occurred.")
_json_response: Final = response.json()
_json_response: Final = _JSON_VALUE.validate_python(response.json())
## LOGGING POST-CALL
logging_obj.post_call(input=input, api_key=None, original_response=_json_response)
model_response = litellm.vertexAITextEmbeddingConfig.transform_vertex_response_to_openai(
response=_json_response, model=model, model_response=model_response
response=_JSON_DICT.validate_python(_json_response), model=model, model_response=model_response
)
return model_response
@ -215,12 +219,12 @@ class VertexEmbedding(VertexBase):
except httpx.TimeoutException:
raise VertexAIError(status_code=408, message="Timeout error occurred.")
_json_response: Final = response.json()
_json_response: Final = _JSON_VALUE.validate_python(response.json())
## LOGGING POST-CALL
logging_obj.post_call(input=input, api_key=None, original_response=_json_response)
model_response = litellm.vertexAITextEmbeddingConfig.transform_vertex_response_to_openai(
response=_json_response, model=model, model_response=model_response
response=_JSON_DICT.validate_python(_json_response), model=model, model_response=model_response
)
return model_response

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
@ -9,6 +11,9 @@ from litellm.llms.voyage.common_utils import get_default_base_url, get_voyage_ap
from litellm.types.llms.openai import AllEmbeddingInputValues, AllMessageValues
from litellm.types.utils import EmbeddingResponse, Usage
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_INT: Final = TypeAdapter(int)
class VoyageError(BaseLLMException):
def __init__(
@ -123,9 +128,10 @@ class VoyageEmbeddingConfig(BaseEmbeddingConfig):
model_response.data = raw_response_json.get("data")
model_response.object = raw_response_json.get("object")
usage_payload: Final = _JSON_OBJECT.validate_python(raw_response_json.get("usage", {}))
usage: Final = Usage(
prompt_tokens=raw_response_json.get("usage", {}).get("total_tokens", 0),
total_tokens=raw_response_json.get("usage", {}).get("total_tokens", 0),
prompt_tokens=_INT.validate_python(usage_payload.get("total_tokens", 0) or 0),
total_tokens=_INT.validate_python(usage_payload.get("total_tokens", 0) or 0),
)
model_response.usage = usage
return model_response

View file

@ -7,6 +7,7 @@ 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
@ -15,6 +16,9 @@ from litellm.llms.voyage.common_utils import get_default_base_url, get_voyage_ap
from litellm.types.llms.openai import AllEmbeddingInputValues, AllMessageValues
from litellm.types.utils import EmbeddingResponse, Usage
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_INT: Final = TypeAdapter(int)
class VoyageError(BaseLLMException):
def __init__(
@ -149,9 +153,10 @@ class VoyageContextualEmbeddingConfig(BaseEmbeddingConfig):
model_response.data = raw_response_json.get("data")
model_response.object = raw_response_json.get("object")
usage_payload: Final = _JSON_OBJECT.validate_python(raw_response_json.get("usage", {}))
usage: Final = Usage(
prompt_tokens=raw_response_json.get("usage", {}).get("total_tokens", 0),
total_tokens=raw_response_json.get("usage", {}).get("total_tokens", 0),
prompt_tokens=_INT.validate_python(usage_payload.get("total_tokens", 0) or 0),
total_tokens=_INT.validate_python(usage_payload.get("total_tokens", 0) or 0),
)
model_response.usage = usage
return model_response

View file

@ -6,9 +6,11 @@ containing content blocks, unlike standard Voyage embeddings which use
/v1/embeddings and a string/list `input` field.
"""
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
@ -17,6 +19,9 @@ from litellm.llms.voyage.common_utils import get_default_base_url, get_voyage_ap
from litellm.types.llms.openai import AllEmbeddingInputValues, AllMessageValues
from litellm.types.utils import EmbeddingResponse, Usage
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_INT: Final = TypeAdapter(int)
class VoyageMultimodalEmbeddingError(BaseLLMException):
def __init__(
@ -155,8 +160,8 @@ class VoyageMultimodalEmbeddingConfig(BaseEmbeddingConfig):
model_response.data = raw_response_json.get("data")
model_response.object = raw_response_json.get("object")
usage_payload: Final = raw_response_json.get("usage", {})
total_tokens: Final = usage_payload.get("total_tokens", 0)
usage_payload: Final = _JSON_OBJECT.validate_python(raw_response_json.get("usage", {}))
total_tokens: Final = _INT.validate_python(usage_payload.get("total_tokens", 0) or 0)
model_response.usage = Usage(
prompt_tokens=total_tokens,
total_tokens=total_tokens,

View file

@ -1,5 +1,5 @@
import time
from collections.abc import AsyncIterator, Iterator
from collections.abc import AsyncIterator, Iterator, Mapping
from datetime import datetime
from typing import TYPE_CHECKING, Any, Final
@ -102,7 +102,7 @@ class IBMWatsonXAIConfig(IBMWatsonXMixin, BaseConfig):
stream: bool | None = None,
**kwargs,
) -> None:
locals_: Final = locals().copy()
locals_: Final[Mapping[str, object]] = dict(locals())
for key, value in locals_.items():
if key != "self" and value is not None:
setattr(self.__class__, key, value)

View file

@ -1,6 +1,8 @@
from collections.abc import Iterable, Mapping
from typing import Final
import httpx
from pydantic import ConfigDict, TypeAdapter
import litellm
from litellm.llms.base_llm.base_utils import BaseLLMModelInfo
@ -9,6 +11,9 @@ from litellm.types.llms.openai import AllMessageValues
from litellm.types.utils import ProviderSpecificModelInfo
USD_TICKS_PER_DOLLAR: Final = 10_000_000_000
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_VALUES: Final = TypeAdapter(Iterable[object], config=ConfigDict(hide_input_in_errors=True))
_MODEL_ID: Final = TypeAdapter(str, config=ConfigDict(strict=True, hide_input_in_errors=True))
def xai_reported_cost_in_usd(cost_in_usd_ticks: object) -> float | None:
@ -97,11 +102,11 @@ class XAIModelInfo(BaseLLMModelInfo):
f"Failed to fetch models from XAI. Status code: {response.status_code}, Response: {response.text}"
)
models: Final = response.json()["data"]
models: Final = _JSON_VALUES.validate_python(_JSON_OBJECT.validate_python(response.json())["data"])
litellm_model_names: Final = []
for model in models:
stripped_model_name = model["id"]
stripped_model_name = _MODEL_ID.validate_python(_JSON_OBJECT.validate_python(model)["id"])
litellm_model_name = "xai/" + stripped_model_name
litellm_model_names.append(litellm_model_name)
return litellm_model_names

View file

@ -142,7 +142,7 @@ class MCPGuardrailTranslationHandler(BaseTranslation):
data: dict[str, Any],
guardrail_to_apply: "CustomGuardrail",
litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
) -> dict[str, Any]:
) -> dict[str, object]:
mcp_tool_name: Final = data.get("mcp_tool_name") or data.get("name")
mcp_arguments: Final[object] = data.get("mcp_arguments") or data.get("arguments")
mcp_tool_description: Final = data.get("mcp_tool_description") or data.get("description")
@ -228,9 +228,9 @@ class MCPGuardrailTranslationHandler(BaseTranslation):
response: "CallToolResult",
guardrail_to_apply: "CustomGuardrail",
litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
user_api_key_dict: Any | None = None,
user_api_key_dict: object | None = None,
request_data: dict | None = None,
) -> Any:
) -> object:
"""Scan the text content of an MCP tool result and write masked text back.
The content list is rewritten in place (only the entries the guardrail

View file

@ -6023,7 +6023,7 @@ class MCPServerManager:
server=server,
)
hook_result: Final[dict[str, Any]] = {}
hook_result: Final[dict[str, object]] = {}
if proxy_logging_obj is None:
return hook_result

View file

@ -157,7 +157,7 @@ def _sanitize_path_parameter_value(param_value: object, param_name: str) -> str:
return quote(value_str, safe="")
def load_openapi_spec(filepath: str) -> dict[str, Any]:
def load_openapi_spec(filepath: str) -> dict[str, object]:
"""
Sync wrapper. For URL specs, use the shared/custom MCP httpx client.
"""

View file

@ -3,7 +3,7 @@ import json
from collections.abc import Callable, Generator, Mapping
from contextlib import contextmanager
from contextvars import ContextVar
from typing import TYPE_CHECKING, Any, Final
from typing import TYPE_CHECKING, Final
from litellm._logging import verbose_logger
from litellm.proxy.types_utils.utils import get_instance_fn
@ -56,7 +56,7 @@ class MCPToolRegistry:
self,
name: str,
description: str,
input_schema: dict[str, Any],
input_schema: dict[str, object],
handler: Callable,
*,
server_id: str | None = None,
@ -117,7 +117,7 @@ class MCPToolRegistry:
def load_tools_from_config(
self,
mcp_tools_config: dict[str, Any] | None = None,
mcp_tools_config: dict[str, object] | None = None,
config_file_path: str | None = None,
) -> None:
"""

View file

@ -303,7 +303,7 @@ async def _a2a_sse_event_source(
error_body: Final = await resp.aread()
error_event: Mapping[str, object] | None = None
try:
parsed: Final = json.loads(error_body)
parsed: Final = _DECODED_JSON.validate_python(json.loads(error_body))
if isinstance(parsed, dict) and "error" in parsed:
error_event = normalize_a2a_jsonrpc_response(parsed, request_id=request_id)
except Exception:

View file

@ -86,7 +86,7 @@ def _anthropic_error_json_response(exc: ProxyException, request: Request) -> JSO
return JSONResponse(status_code=status_code, content=content, headers=exc.headers)
def _strip_total_tokens_from_anthropic_response(response: Any) -> None:
def _strip_total_tokens_from_anthropic_response(response: object) -> None:
"""Remove the OpenAI-flavored `usage.total_tokens` field that LiteLLM
injects into Anthropic /v1/messages responses.

View file

@ -199,7 +199,7 @@ class IPAddressUtils:
@staticmethod
def is_request_https(
request: Request,
general_settings: Mapping[str, Any] | None = None,
general_settings: Mapping[str, object] | None = None,
) -> bool:
"""
Whether this request's PUBLIC-facing origin is HTTPS, for deciding

View file

@ -2,7 +2,7 @@
## Common checks for /v1/models and `/model/info`
import copy
from collections.abc import Sequence
from typing import Any, Final
from typing import Final
import litellm
from litellm._logging import verbose_proxy_logger
@ -336,8 +336,8 @@ def get_known_models_from_wildcard(wildcard_model: str, litellm_params: LiteLLM_
def expand_wildcard_deployments_for_model_info(
deployments: list[dict[str, Any]],
) -> list[dict[str, Any]]:
deployments: list[dict[str, object]],
) -> list[dict[str, object]]:
"""Expand wildcard deployments into one row per known provider model.
PR #30025 changed /model/info to read from llm_router.model_list (correct,

View file

@ -1,4 +1,4 @@
from typing import Any, Final
from typing import Final
from fastapi import Request
@ -22,7 +22,7 @@ def _get_proxy_general_settings() -> dict[str, object]:
def get_trusted_proxy_cidrs(
general_settings: dict[str, Any] | None = None,
general_settings: dict[str, object] | None = None,
) -> list[str]:
"""Operator-configured trusted reverse-proxy CIDRs, normalized to strings.
@ -48,7 +48,7 @@ def _get_direct_client_ip(request: Request) -> str | None:
def require_trusted_proxy_request(
*,
request: Request,
general_settings: dict[str, Any] | None = None,
general_settings: dict[str, object] | None = None,
feature_name: str,
setting_name: str = TRUSTED_PROXY_RANGES_KEY,
) -> None:

View file

@ -40,7 +40,9 @@ def _extract_cache_params() -> dict[str, object]:
return {}
try:
cache_params: Final = vars(litellm.cache.cache)
cleaned_params: Final = HealthCheckCacheParams(**cache_params).model_dump() if cache_params else {}
cleaned_params: Final = (
HealthCheckCacheParams.model_validate(dict(**cache_params)).model_dump() if cache_params else {}
)
return masker.mask_dict(cleaned_params)
except (AttributeError, TypeError) as e:
verbose_proxy_logger.debug("Error extracting cache params: %s", e)

View file

@ -3,7 +3,7 @@ import sys
import time
import webbrowser
from collections.abc import Callable, Mapping, Sequence
from typing import Any, Final, TypeVar
from typing import Any, Final, Protocol, TypeVar
from urllib.parse import urlencode
import click
@ -96,6 +96,24 @@ class CliContextObj(TypedDict):
api_key_from_token_file: ReadOnly[NotRequired[bool]]
class _CliContextObjView(Protocol):
@property
def obj(self) -> CliContextObj: ...
class _OptionalCliContextObjView(Protocol):
@property
def obj(self) -> CliContextObj | None: ...
def _cli_context_obj(view: _CliContextObjView) -> CliContextObj:
return view.obj
def _optional_cli_context_obj(view: _OptionalCliContextObjView) -> CliContextObj | None:
return view.obj
class CliPollData(TypedDict, total=False):
status: str
key: str
@ -197,7 +215,7 @@ def keychain_unreadable_notice(vault: SecretVault) -> str:
def context_secret_vault(ctx: click.Context) -> SecretVault:
"""Where this invocation reads and writes secret material; injectable through ctx.obj for tests"""
ctx_obj: Final[CliContextObj | None] = ctx.obj
ctx_obj: Final = _optional_cli_context_obj(ctx)
if ctx_obj is None:
return SYSTEM_KEYRING
return ctx_obj.get("secret_vault") or SYSTEM_KEYRING
@ -880,7 +898,7 @@ def login(ctx: click.Context, config_claude: bool, pkce: bool) -> None:
"""Login to LiteLLM proxy using SSO authentication"""
from litellm.constants import LITELLM_CLI_SOURCE_IDENTIFIER
ctx_obj: Final[CliContextObj] = ctx.obj
ctx_obj: Final = _cli_context_obj(ctx)
base_url: Final = ctx_obj["base_url"]
if config_claude:
settings_path: Final = claude_settings_path(os.environ)
@ -1030,7 +1048,7 @@ def print_token(ctx: click.Context):
# explicitly pointed us at a server, trust whichever one `lite login`
# actually issued this token for -- that's the whole point of not
# needing a wrapper command.
ctx_obj: Final[CliContextObj] = ctx.obj
ctx_obj: Final = _cli_context_obj(ctx)
issued_for_this_server: Final = token_data.get("base_url") == ctx_obj.get("base_url", "").rstrip("/")
if ctx_obj.get("base_url_explicit") and not issued_for_this_server:
click.echo("Not authenticated for this server. Run 'lite login'.", err=True)

View file

@ -1,13 +1,32 @@
import json as json_lib
from typing import Final
from typing import Final, Protocol
import click
import requests
import rich
from typing_extensions import ReadOnly, TypedDict
from ...http_client import HTTPClient
class _CliContext(TypedDict):
base_url: ReadOnly[str]
api_key: ReadOnly[str | None]
class _CliContextView(Protocol):
@property
def obj(self) -> _CliContext: ...
class _JsonBodyView(TypedDict):
body: ReadOnly[object]
def _cli_context(view: _CliContextView) -> _CliContext:
return view.obj
@click.group()
def http():
"""Make HTTP requests to the LiteLLM proxy server"""
@ -79,21 +98,23 @@ def request(
# If not JSON, use as raw data
request_data = data
client: Final = HTTPClient(ctx.obj["base_url"], ctx.obj["api_key"])
client: Final = HTTPClient(_cli_context(ctx)["base_url"], _cli_context(ctx)["api_key"])
try:
response: Final = client.request(
method=method,
uri=uri,
data=request_data,
json=json_data,
headers=headers,
)
rich.print_json(data=response)
response: Final[_JsonBodyView] = {
"body": client.request(
method=method,
uri=uri,
data=request_data,
json=json_data,
headers=headers,
)
}
rich.print_json(data=response["body"])
except requests.exceptions.HTTPError as e:
click.echo(f"Error: HTTP {e.response.status_code}", err=True)
try:
error_body: Final = e.response.json()
rich.print_json(data=error_body)
error_body: Final[_JsonBodyView] = {"body": e.response.json()}
rich.print_json(data=error_body["body"])
except json_lib.JSONDecodeError:
click.echo(e.response.text, err=True)
raise click.Abort()

View file

@ -2,7 +2,7 @@ import builtins
import json
from collections.abc import Mapping, Sequence
from datetime import datetime
from typing import Any, Final, Literal
from typing import Any, Final, Literal, Protocol
import click
import requests
@ -20,8 +20,9 @@ class _CliContext(TypedDict):
api_key: ReadOnly[str | None]
class _CliContextView(TypedDict):
obj: ReadOnly[_CliContext]
class _CliContextView(Protocol):
@property
def obj(self) -> _CliContext: ...
class _KeyRowsView(TypedDict):
@ -32,6 +33,10 @@ class _JsonBodyView(TypedDict):
body: ReadOnly[object]
def _cli_context(view: _CliContextView) -> _CliContext:
return view.obj
@click.group()
def keys():
"""Manage API keys for the LiteLLM proxy server"""
@ -74,8 +79,8 @@ def list(
return_full_object: bool,
):
"""List all API keys"""
context: Final[_CliContextView] = {"obj": ctx.obj}
client: Final = KeysManagementClient(context["obj"]["base_url"], context["obj"]["api_key"])
context: Final = _cli_context(ctx)
client: Final = KeysManagementClient(context["base_url"], context["api_key"])
response: Final = client.list(
page=page,
size=size,
@ -140,8 +145,8 @@ def generate(
config: str | None,
):
"""Generate a new API key"""
context: Final[_CliContextView] = {"obj": ctx.obj}
client: Final = KeysManagementClient(context["obj"]["base_url"], context["obj"]["api_key"])
context: Final = _cli_context(ctx)
client: Final = KeysManagementClient(context["base_url"], context["api_key"])
try:
models_list: Final = [m.strip() for m in models.split(",")] if models else None
aliases_dict: Final = json.loads(aliases) if aliases else None
@ -177,8 +182,8 @@ def generate(
@click.pass_context
def delete(ctx: click.Context, keys: str | None, key_aliases: str | None):
"""Delete API keys by key or alias"""
context: Final[_CliContextView] = {"obj": ctx.obj}
client: Final = KeysManagementClient(context["obj"]["base_url"], context["obj"]["api_key"])
context: Final = _cli_context(ctx)
client: Final = KeysManagementClient(context["base_url"], context["api_key"])
keys_list: Final = [k.strip() for k in keys.split(",")] if keys else None
aliases_list: Final = [a.strip() for a in key_aliases.split(",")] if key_aliases else None
try:
@ -377,8 +382,8 @@ def import_keys(
# Create clients for both source and destination
source_client: Final = KeysManagementClient(source_base_url, source_api_key)
context: Final[_CliContextView] = {"obj": ctx.obj}
dest_client: Final = KeysManagementClient(context["obj"]["base_url"], context["obj"]["api_key"])
context: Final = _cli_context(ctx)
dest_client: Final = KeysManagementClient(context["base_url"], context["api_key"])
try:
# Get all keys from source instance with pagination

View file

@ -378,7 +378,7 @@ def _print_summary_table(provider_counts):
rich.print(summary_table)
def get_model_list_from_yaml_file(yaml_file: str) -> list[dict[str, Any]]:
def get_model_list_from_yaml_file(yaml_file: str) -> list[dict[str, object]]:
"""Load and validate the model list from a YAML file."""
with open(yaml_file, "r") as f:
data: Final = yaml.safe_load(f)

View file

@ -1,7 +1,7 @@
# stdlib imports
# third party imports
from typing import Final
from typing import Final, Protocol
import click
@ -37,6 +37,15 @@ from .commands.users import users
from .interface import interactive_shell
class _CliContextObjView(Protocol):
@property
def obj(self) -> CliContextObj: ...
def _cli_context_obj(view: _CliContextObjView) -> CliContextObj:
return view.obj
def print_version(base_url: str, api_key: str | None):
"""Print CLI and server version info."""
click.echo(f"LiteLLM Proxy CLI Version: {litellm_version}")
@ -135,7 +144,7 @@ def cli(ctx: click.Context, show_version: bool, base_url: str | None, api_key: s
@click.pass_context
def version(ctx: click.Context):
"""Show the LiteLLM Proxy CLI and server version."""
ctx_obj: Final[CliContextObj] = ctx.obj
ctx_obj: Final = _cli_context_obj(ctx)
print_version(ctx_obj.get("base_url"), ctx_obj.get("api_key"))

View file

@ -27,7 +27,7 @@ import httpx
import orjson
from fastapi import HTTPException, Request, status
from fastapi.responses import JSONResponse, Response, StreamingResponse
from pydantic import BaseModel, TypeAdapter, ValidationError
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
from starlette.types import Receive, Scope, Send
from typing_extensions import Never
@ -556,6 +556,7 @@ def _stream_usage_tracking_updates(
}
@runtime_checkable
class _UpstreamHttpResponse(Protocol):
@property
def status_code(self) -> int: ...
@ -570,6 +571,11 @@ def _as_upstream_response(response: _UpstreamHttpResponse) -> _UpstreamHttpRespo
return response
_UPSTREAM_RESPONSE: Final = TypeAdapter(
_UpstreamHttpResponse, config=ConfigDict(arbitrary_types_allowed=True, hide_input_in_errors=True)
)
class _ReadsHeaderValues(Protocol):
def get(self, key: str, default: str = "") -> str: ...
@ -2562,7 +2568,7 @@ class ProxyBaseLLMRequestProcessing:
@staticmethod
def _response_cost_from_logging_obj(
*,
response: Any,
response: object,
logging_obj: LiteLLMLoggingObj,
) -> float | str:
"""
@ -2618,7 +2624,7 @@ class ProxyBaseLLMRequestProcessing:
is_streaming_request: bool | None = False,
contents: list[object] | None = None,
skip_pre_call_logic: bool = False,
) -> Any:
) -> object:
"""Run the request, sending SSE keepalives while the upstream is still silent.
Everything below this point, the upstream call included, happens before the
@ -3261,7 +3267,7 @@ class ProxyBaseLLMRequestProcessing:
if isinstance(result, Response):
return result
upstream: Final = _as_upstream_response(result)
upstream: Final = _UPSTREAM_RESPONSE.validate_python(result)
content: Final = await upstream.aread()
return Response(
content=content,

View file

@ -545,7 +545,7 @@ async def convert_upload_files_to_file_data(
return data
async def get_request_body(request: Request) -> dict[str, Any]:
async def get_request_body(request: Request) -> dict[str, object]:
"""
Read the request body and parse it as JSON.
"""

View file

@ -5,7 +5,7 @@ FastAPI 0.120+ has stricter schema generation that fails on certain types like o
This module provides a compatibility layer to handle these cases gracefully.
"""
from typing import Any, Final
from typing import Final
from litellm._logging import verbose_proxy_logger
@ -16,7 +16,7 @@ def get_openapi_schema_with_compat(
version: str,
description: str,
routes: list,
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Generate OpenAPI schema with compatibility handling for FastAPI 0.120+.

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