scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 100

chore(typing): clear basedpyright reportAny errors in proxy, router, and main completion paths

Browse source 
Replaces Any-typed seams with real types across the top reportAny/reportExplicitAny
offenders: proxy_server.py, proxy/utils.py, litellm_logging.py, streaming_iterator.py,
streaming_handler.py, llm_http_handler.py, main.py, and router.py. Untyped dicts and
callbacks get TypedDicts, Pydantic TypeAdapter validation, or concrete provider types
instead of Any where the surrounding contract already had a real shape; boundaries
that stayed genuinely heterogeneous (Prisma rows, dynamically loaded callables) are
left as Any rather than forced into a false precision that would surface elsewhere.

reportAny: 19292 -> 18952 (-340)
reportExplicitAny: 6518 -> 6422 (-96)

No other basedpyright rule regresses repo-wide; several also improved as a side
effect (reportUnknownArgumentType -8, reportUnknownVariableType -16,
reportUnknownMemberType -16).
This commit is contained in:
mateo-berri 2026-08-05 08:54:28 +00:00
parent 956d5177d1
commit 58c6f50758
No known key found for this signature in database
12 changed files with 561 additions and 345 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
basedpyright-code-budget.json
View file

@ -1,6 +1,6 @@
{
"reportAny": {
"limit": 29682
"limit": 29002
},
"reportArgumentType": {
"limit": 2645
@ -24,7 +24,7 @@
"limit": 42
},
"reportExplicitAny": {
"limit": 9440
"limit": 9248
},
"reportFunctionMemberAccess": {
"limit": 11
@ -99,19 +99,19 @@
"limit": 0
},
"reportUnknownArgumentType": {
"limit": 45297
"limit": 45281
},
"reportUnknownLambdaType": {
"limit": 113
},
"reportUnknownMemberType": {
"limit": 40411
"limit": 40379
},
"reportUnknownParameterType": {
"limit": 20301
},
"reportUnknownVariableType": {
"limit": 31968
"limit": 31936
},
"reportUnnecessaryCast": {
"limit": 177

37
litellm/litellm_core_utils/litellm_logging.py
View file

@ -175,7 +175,9 @@ from .initialize_dynamic_callback_params import (
from .specialty_caches.dynamic_logging_cache import DynamicLoggingCache
if TYPE_CHECKING:
from litellm.integrations.otel.logger import OpenTelemetryV2
from litellm.llms.base_llm.passthrough.transformation import BasePassthroughConfig
from litellm.router import Router
try:
from litellm_enterprise.enterprise_callbacks.callback_controls import (
EnterpriseCallbackControls,
@ -346,8 +348,12 @@ class Logging(LiteLLMLoggingBaseClass):
self.litellm_call_id = litellm_call_id
self.litellm_trace_id: str = litellm_trace_id if litellm_trace_id else str(uuid.uuid4())
self.function_id = function_id
self.streaming_chunks: list[Any] = [] # for generating complete stream response
self.sync_streaming_chunks: list[Any] = [] # for generating complete stream response
self.streaming_chunks: list[
ModelResponse
] = [] # mutable-ok: accumulates one chunk per stream event over the object's lifetime
self.sync_streaming_chunks: list[
ModelResponse
] = [] # mutable-ok: accumulates one chunk per stream event over the object's lifetime
self.log_raw_request_response = log_raw_request_response
# Initialize dynamic callbacks
@ -392,7 +398,9 @@ class Logging(LiteLLMLoggingBaseClass):
self.caching_details: CachingDetails | None = None
# Passthrough endpoint guardrails config for field targeting
self.passthrough_guardrails_config: dict[str, Any] | None = None
self.passthrough_guardrails_config: dict[str, object] | None = (
None # mutable-ok: assigned once from a caller-supplied dict payload
)
self.model_call_details: dict[str, Any] = {
"litellm_trace_id": self.litellm_trace_id,
@ -1138,20 +1146,15 @@ class Logging(LiteLLMLoggingBaseClass):
self.model_call_details["additional_args"] = additional_args
self.model_call_details["log_event_type"] = "post_api_call"
if self.litellm_request_debug:
attr = "warning"
else:
attr = "debug"
callattr = verbose_logger.warning if self.litellm_request_debug else verbose_logger.debug
if json_logs:
callattr = getattr(verbose_logger, attr)
callattr(
"RAW RESPONSE:\n{}\n\n".format(
self.model_call_details.get("original_response", self.model_call_details)
),
)
else:
callattr = getattr(verbose_logger, attr)
callattr(
"RAW RESPONSE:\n{}\n\n".format(
self.model_call_details.get("original_response", self.model_call_details)
@ -1905,8 +1908,8 @@ class Logging(LiteLLMLoggingBaseClass):
def _is_recognized_call_type_for_logging(
self,
logging_result: Any,
):
logging_result: object,
) -> bool:
"""
Returns True if the call type is recognized for logging (eg. ModelResponse, ModelResponseStream, etc.)
"""
@ -3035,7 +3038,7 @@ class Logging(LiteLLMLoggingBaseClass):
return trace_id
def _get_callback_object(self, service_name: Literal["langfuse"]) -> Any | None:
def _get_callback_object(self, service_name: Literal["langfuse"]) -> LangFuseLogger | None:
"""
Return dynamic callback object.
@ -3548,7 +3551,7 @@ def set_callbacks(callback_list, function_id=None):
def _init_custom_logger_compatible_class(
logging_integration: _custom_logger_compatible_callbacks_literal,
internal_usage_cache: DualCache | None,
llm_router: Any | None, # expect litellm.Router, but typing errors due to circular import
llm_router: "Router | None",
custom_logger_init_args: dict | None = {},
) -> CustomLogger | None:
"""
@ -3936,7 +3939,7 @@ def _init_custom_logger_compatible_class(
dynamic_rate_limiter_obj = _PROXY_DynamicRateLimitHandler(internal_usage_cache=internal_usage_cache)
if llm_router is not None and isinstance(llm_router, litellm.Router):
if llm_router is not None:
dynamic_rate_limiter_obj.update_variables(llm_router=llm_router)
_in_memory_loggers.append(dynamic_rate_limiter_obj)
return dynamic_rate_limiter_obj # type: ignore
@ -3954,7 +3957,7 @@ def _init_custom_logger_compatible_class(
dynamic_rate_limiter_obj_v3 = _PROXY_DynamicRateLimitHandlerV3(internal_usage_cache=internal_usage_cache)
if llm_router is not None and isinstance(llm_router, litellm.Router):
if llm_router is not None:
dynamic_rate_limiter_obj_v3.update_variables(llm_router=llm_router)
_in_memory_loggers.append(dynamic_rate_limiter_obj_v3)
return dynamic_rate_limiter_obj_v3 # type: ignore
@ -4160,7 +4163,9 @@ def _init_custom_logger_compatible_class(
return None
def _maybe_construct_otel_v2(callback_name: str, _in_memory_loggers: list) -> Any | None:
def _maybe_construct_otel_v2(
callback_name: str, _in_memory_loggers: list
) -> "OpenTelemetryV2 | None": # mutable-ok: list is only read here; caller owns the mutable in-memory logger registry
"""If ``LITELLM_OTEL_V2`` is on, build (or reuse) a single ``OpenTelemetryV2``
instance configured via the preset for ``callback_name``.

37
litellm/litellm_core_utils/streaming_handler.py
View file

@ -70,7 +70,7 @@ def _next_sync_or_exhausted(it: Any) -> Any:
return _SYNC_ITER_EXHAUSTED
def is_async_iterable(obj: Any) -> bool:
def is_async_iterable(obj: object) -> bool:
"""
Check if an object is an async iterable (can be used with 'async for').
@ -124,9 +124,8 @@ class CustomStreamWrapper:
self.sent_last_chunk = False
self._stream_created_time: float = time.time()
litellm_params: GenericLiteLLMParams = GenericLiteLLMParams.model_validate(
dict(**self.logging_obj.model_call_details.get("litellm_params", {}))
)
_init_litellm_params = self.logging_obj.model_call_details.get("litellm_params", {})
litellm_params: GenericLiteLLMParams = GenericLiteLLMParams.model_validate(dict(**_init_litellm_params))
self.merge_reasoning_content_in_choices: bool = litellm_params.merge_reasoning_content_in_choices or False
self.sent_first_thinking_block = False
self.sent_last_thinking_block = False
@ -151,7 +150,7 @@ class CustomStreamWrapper:
_api_base = get_api_base(
model=model or "",
optional_params=self.logging_obj.model_call_details.get("litellm_params", {}),
optional_params=_init_litellm_params,
)
self._hidden_params = {
@ -346,9 +345,9 @@ class CustomStreamWrapper:
try:
if not isinstance(chunk, str):
chunk = chunk.decode("utf-8") # DO NOT REMOVE this: This is required for HF inference API + Streaming
text = ""
text: str = ""
is_finished = False
finish_reason = ""
finish_reason: str = ""
print_verbose(f"chunk: {chunk}")
if chunk.startswith("data:"):
data_json = json.loads(chunk[5:])
@ -383,7 +382,7 @@ class CustomStreamWrapper:
chunk = chunk.decode("utf-8")
data_json = json.loads(chunk)
try:
text = data_json["completions"][0]["data"]["text"]
text: str = data_json["completions"][0]["data"]["text"]
is_finished = True
finish_reason = "stop"
return {
@ -398,7 +397,7 @@ class CustomStreamWrapper:
chunk = chunk.decode("utf-8")
data_json = json.loads(chunk)
try:
text = data_json["answer"]
text: str = data_json["answer"]
is_finished = True
finish_reason = "stop"
return {
@ -410,9 +409,9 @@ class CustomStreamWrapper:
raise ValueError(f"Unable to parse response. Original response: {chunk}")
def handle_nlp_cloud_chunk(self, chunk):
text = ""
text: str = ""
is_finished = False
finish_reason = ""
finish_reason: str = ""
try:
if self.model and "dolphin" in self.model:
chunk = self.process_chunk(chunk=chunk)
@ -436,7 +435,7 @@ class CustomStreamWrapper:
chunk = chunk.decode("utf-8")
data_json = json.loads(chunk)
try:
text = data_json["completions"][0]["completion"]
text: str = data_json["completions"][0]["completion"]
is_finished = True
finish_reason = "stop"
return {
@ -449,8 +448,8 @@ class CustomStreamWrapper:
def handle_azure_chunk(self, chunk):
is_finished = False
finish_reason = ""
text = ""
finish_reason: str = ""
text: str = ""
print_verbose(f"chunk: {chunk}")
if "data: [DONE]" in chunk:
text = ""
@ -548,9 +547,9 @@ class CustomStreamWrapper:
def handle_azure_text_completion_chunk(self, chunk):
try:
text = ""
text: str = ""
is_finished = False
finish_reason = None
finish_reason: str | None = None
choices = getattr(chunk, "choices", [])
if len(choices) > 0:
text = choices[0].text
@ -568,9 +567,9 @@ class CustomStreamWrapper:
def handle_openai_text_completion_chunk(self, chunk):
try:
text = ""
text: str = ""
is_finished = False
finish_reason = None
finish_reason: str | None = None
usage = None
choices = getattr(chunk, "choices", [])
if len(choices) > 0:
@ -589,7 +588,7 @@ class CustomStreamWrapper:
except Exception as e:
raise e
def handle_baseten_chunk(self, chunk):
def handle_baseten_chunk(self, chunk) -> str:
try:
chunk = chunk.decode("utf-8")
if len(chunk) > 0:

284
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -19,6 +19,8 @@ from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
import httpx # type: ignore
from openai.types.file_deleted import FileDeleted
from pydantic import TypeAdapter
from typing_extensions import TypedDict
import litellm
import litellm.litellm_core_utils
@ -87,7 +89,11 @@ from litellm.types.containers.main import (
ContainerObject,
DeleteContainerResult,
)
from litellm.types.files import StreamingMediaUploadConfig, TwoStepFileUploadConfig
from litellm.types.files import (
StreamingMediaUploadConfig,
TwoStepFileUploadConfig,
TwoStepFileUploadRequest,
)
from litellm.types.integrations.custom_logger import (
AgenticLoopPlan,
AgenticLoopRequestPatch,
@ -212,6 +218,45 @@ def _responses_api_optional_request_param_names() -> frozenset[str]:
return frozenset(get_type_hints(ResponsesAPIOptionalRequestParams).keys())
_TWO_STEP_FILE_UPLOAD_REQUEST_ADAPTER = TypeAdapter(TwoStepFileUploadRequest)
class _PresignedBatchRequest(TypedDict):
method: str
url: str
headers: dict[str, str] # mutable-ok: TypedDict field mirrors the outgoing HTTP request's header dict
data: str | bytes | dict[str, Any] | None # mutable-ok: TypedDict field mirrors the outgoing HTTP request body dict
_PRESIGNED_BATCH_REQUEST_ADAPTER = TypeAdapter(_PresignedBatchRequest)
def _send_presigned_batch_request(
client: HTTPHandler,
request: _PresignedBatchRequest,
timeout: float | httpx.Timeout | None,
) -> httpx.Response | None:
method = request["method"].lower()
if method == "get":
return getattr(client, method)(url=request["url"], headers=request["headers"], timeout=timeout)
return getattr(client, method)(
url=request["url"], headers=request["headers"], data=request["data"], timeout=timeout
)
async def _asend_presigned_batch_request(
client: AsyncHTTPHandler,
request: _PresignedBatchRequest,
timeout: float | httpx.Timeout | None,
) -> httpx.Response | None:
method = request["method"].lower()
if method == "get":
return await getattr(client, method)(url=request["url"], headers=request["headers"], timeout=timeout)
return await getattr(client, method)(
url=request["url"], headers=request["headers"], data=request["data"], timeout=timeout
)
def _custom_logger_callbacks(logging_obj: LiteLLMLoggingObj) -> list["CustomLogger"]:
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.litellm_logging import (
@ -2007,7 +2052,7 @@ class BaseLLMHTTPHandler:
# Also check for extra_headers in kwargs (from config or direct calls)
extra_headers_from_kwargs = kwargs.get("extra_headers", None)
# Merge all header sources: forwarded < extra_headers < provider_specific
merged_headers = {}
merged_headers: dict[str, str] = {} # mutable-ok: merged in place from multiple header sources below
if forwarded_headers:
merged_headers.update(forwarded_headers)
if extra_headers_from_kwargs:
@ -2370,14 +2415,16 @@ class BaseLLMHTTPHandler:
model: str,
input: str | ResponseInputParam,
custom_llm_provider: str,
response_api_optional_request_params: dict[str, Any],
response_api_optional_request_params: dict[
str, object
], # mutable-ok: parameter mirrors the caller's optional-params dict payload
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
) -> tuple[
str,
str | ResponseInputParam,
str,
dict[str, Any],
dict[str, object],
GenericLiteLLMParams,
]:
if not _has_pre_call_deployment_hook(logging_obj):
@ -2444,7 +2491,9 @@ class BaseLLMHTTPHandler:
model: str,
input: str | ResponseInputParam,
responses_api_provider_config: BaseResponsesAPIConfig,
response_api_optional_request_params: dict[str, Any],
response_api_optional_request_params: dict[
str, object
], # mutable-ok: parameter mirrors the caller's optional-params dict payload
custom_llm_provider: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLoggingObj,
@ -2509,8 +2558,9 @@ class BaseLLMHTTPHandler:
else:
sync_httpx_client = client
request_extra_headers = response_api_optional_request_params.get("extra_headers")
headers = responses_api_provider_config.validate_environment(
headers=response_api_optional_request_params.get("extra_headers", {}) or {},
headers=request_extra_headers if isinstance(request_extra_headers, dict) else {},
model=model,
litellm_params=litellm_params,
)
@ -2519,7 +2569,7 @@ class BaseLLMHTTPHandler:
headers.update(extra_headers)
# Check if streaming is requested
stream = response_api_optional_request_params.get("stream", False)
stream = bool(response_api_optional_request_params.get("stream", False))
api_base = responses_api_provider_config.get_complete_url(
api_base=litellm_params.api_base,
@ -2572,8 +2622,6 @@ class BaseLLMHTTPHandler:
stream=stream,
fake_stream=fake_stream,
)
body_kwargs: dict[str, Any] = {"data": signed_body} if signed_body is not None else {"json": data}
## LOGGING
logging_obj.pre_call(
input=input,
@ -2587,13 +2635,18 @@ class BaseLLMHTTPHandler:
try:
if is_stream_request:
response = sync_httpx_client.post(
url=api_base,
headers=headers,
timeout=timeout or float(response_api_optional_request_params.get("timeout", 0)),
stream=stream,
**body_kwargs,
request_timeout = response_api_optional_request_params.get("timeout", 0)
stream_timeout_value = timeout or float(
request_timeout if isinstance(request_timeout, (int, float)) else 0
)
if signed_body is not None:
response = sync_httpx_client.post(
url=api_base, headers=headers, timeout=stream_timeout_value, stream=stream, data=signed_body
)
else:
response = sync_httpx_client.post(
url=api_base, headers=headers, timeout=stream_timeout_value, stream=stream, json=data
)
if fake_stream is True:
return MockResponsesAPIStreamingIterator(
response=response,
@ -2617,12 +2670,18 @@ class BaseLLMHTTPHandler:
call_type=CallTypes.responses.value,
)
else:
response = sync_httpx_client.post(
url=api_base,
headers=headers,
timeout=timeout or float(response_api_optional_request_params.get("timeout", 0)),
**body_kwargs,
request_timeout = response_api_optional_request_params.get("timeout", 0)
non_stream_timeout_value = timeout or float(
request_timeout if isinstance(request_timeout, (int, float)) else 0
)
if signed_body is not None:
response = sync_httpx_client.post(
url=api_base, headers=headers, timeout=non_stream_timeout_value, data=signed_body
)
else:
response = sync_httpx_client.post(
url=api_base, headers=headers, timeout=non_stream_timeout_value, json=data
)
except Exception as e:
raise self._handle_error(
e=e,
@ -2687,8 +2746,9 @@ class BaseLLMHTTPHandler:
else:
async_httpx_client = client
request_extra_headers = response_api_optional_request_params.get("extra_headers")
headers = responses_api_provider_config.validate_environment(
headers=response_api_optional_request_params.get("extra_headers", {}) or {},
headers=request_extra_headers if isinstance(request_extra_headers, dict) else {},
model=model,
litellm_params=litellm_params,
)
@ -2697,7 +2757,7 @@ class BaseLLMHTTPHandler:
headers.update(extra_headers)
# Check if streaming is requested
stream = response_api_optional_request_params.get("stream", False)
stream = bool(response_api_optional_request_params.get("stream", False))
api_base = responses_api_provider_config.get_complete_url(
api_base=litellm_params.api_base,
@ -2747,7 +2807,6 @@ class BaseLLMHTTPHandler:
stream=stream,
fake_stream=fake_stream,
)
body_kwargs: dict[str, Any] = {"data": signed_body} if signed_body is not None else {"json": data}
## LOGGING
logging_obj.pre_call(
@ -2762,13 +2821,18 @@ class BaseLLMHTTPHandler:
try:
if is_stream_request:
response = await async_httpx_client.post(
url=api_base,
headers=headers,
timeout=timeout or float(response_api_optional_request_params.get("timeout", 0)),
stream=stream,
**body_kwargs,
request_timeout = response_api_optional_request_params.get("timeout", 0)
stream_timeout_value = timeout or float(
request_timeout if isinstance(request_timeout, (int, float)) else 0
)
if signed_body is not None:
response = await async_httpx_client.post(
url=api_base, headers=headers, timeout=stream_timeout_value, stream=stream, data=signed_body
)
else:
response = await async_httpx_client.post(
url=api_base, headers=headers, timeout=stream_timeout_value, stream=stream, json=data
)
if fake_stream is True:
return MockResponsesAPIStreamingIterator(
@ -2794,12 +2858,18 @@ class BaseLLMHTTPHandler:
call_type=CallTypes.responses.value,
)
else:
response = await async_httpx_client.post(
url=api_base,
headers=headers,
timeout=timeout or float(response_api_optional_request_params.get("timeout", 0)),
**body_kwargs,
request_timeout = response_api_optional_request_params.get("timeout", 0)
non_stream_timeout_value = timeout or float(
request_timeout if isinstance(request_timeout, (int, float)) else 0
)
if signed_body is not None:
response = await async_httpx_client.post(
url=api_base, headers=headers, timeout=non_stream_timeout_value, data=signed_body
)
else:
response = await async_httpx_client.post(
url=api_base, headers=headers, timeout=non_stream_timeout_value, json=data
)
except Exception as e:
raise self._handle_error(
@ -3355,16 +3425,19 @@ class BaseLLMHTTPHandler:
"""
if upload_url_location == "headers":
# Google Cloud Storage style - URL in X-Goog-Upload-URL header
upload_url = response.headers.get("X-Goog-Upload-URL")
return upload_url, None
upload_url_header = response.headers.get("X-Goog-Upload-URL")
return (upload_url_header if isinstance(upload_url_header, str) else None), None
else:
# Response body style (e.g., Manus, S3 presigned URLs)
try:
response_data = response.json()
upload_url = response_data.get(upload_url_key)
return upload_url, response_data if upload_url else None
except Exception:
return None, None
if not isinstance(response_data, dict):
return None, None
upload_url_raw = response_data.get(upload_url_key)
upload_url = upload_url_raw if isinstance(upload_url_raw, str) else None
return upload_url, response_data if upload_url else None
def create_file(
self,
@ -3480,7 +3553,7 @@ class BaseLLMHTTPHandler:
):
# Handle pre-signed requests (e.g., from Bedrock S3 uploads)
# Type narrowing: this is a plain dict, not TwoStepFileUploadConfig
presigned_request = cast(dict[str, Any], transformed_request)
presigned_request = _TWO_STEP_FILE_UPLOAD_REQUEST_ADAPTER.validate_python(transformed_request)
upload_response = getattr(sync_httpx_client, presigned_request["method"].lower())(
url=presigned_request["url"],
headers=presigned_request["headers"],
@ -3526,7 +3599,7 @@ class BaseLLMHTTPHandler:
elif isinstance(transformed_request, dict) and "file" in transformed_request:
# Handle multipart form-data uploads (e.g., Anthropic Files API)
# The dict contains tuples suitable for httpx's `files` parameter
file_request = cast(dict[str, Any], transformed_request)
file_request = transformed_request
upload_response = sync_httpx_client.post(
url=api_base,
headers=headers,
@ -3642,7 +3715,7 @@ class BaseLLMHTTPHandler:
):
# Handle pre-signed requests (e.g., from Bedrock S3 uploads)
# Type narrowing: this is a plain dict, not TwoStepFileUploadConfig
presigned_request = cast(dict[str, Any], transformed_request)
presigned_request = _TWO_STEP_FILE_UPLOAD_REQUEST_ADAPTER.validate_python(transformed_request)
upload_response = await getattr(async_httpx_client, presigned_request["method"].lower())(
url=presigned_request["url"],
headers=presigned_request["headers"],
@ -3734,13 +3807,12 @@ class BaseLLMHTTPHandler:
timeout: float | httpx.Timeout | None,
) -> httpx.Response:
headers = {**base_headers, "Content-Type": content_type}
kwargs: dict[str, Any] = {
"headers": headers,
"content": self._iter_in_blocks(body_stream.iter_bytes(), self._MEDIA_UPLOAD_BLOCK_SIZE),
}
if timeout is not None:
kwargs["timeout"] = timeout
resp = client.client.post(url, **kwargs)
resp = client.client.post(
url,
headers=headers,
content=self._iter_in_blocks(body_stream.iter_bytes(), self._MEDIA_UPLOAD_BLOCK_SIZE),
timeout=timeout if timeout is not None else httpx.USE_CLIENT_DEFAULT,
)
self._check_media_upload_response(resp)
return resp
@ -3771,10 +3843,12 @@ class BaseLLMHTTPHandler:
break
yield cast(bytes, block)
kwargs: dict[str, Any] = {"headers": headers, "content": _abody()}
if timeout is not None:
kwargs["timeout"] = timeout
resp = await client.client.post(url, **kwargs)
resp = await client.client.post(
url,
headers=headers,
content=_abody(),
timeout=timeout if timeout is not None else httpx.USE_CLIENT_DEFAULT,
)
await resp.aread()
self._check_media_upload_response(resp)
return resp
@ -3849,10 +3923,10 @@ class BaseLLMHTTPHandler:
try:
if isinstance(transformed_request, dict) and "method" in transformed_request:
# Handle pre-signed requests (e.g., from Bedrock with AWS auth)
batch_response = getattr(sync_httpx_client, transformed_request["method"].lower())(
url=transformed_request["url"],
headers=transformed_request["headers"],
data=transformed_request["data"],
presigned_batch_request = _PRESIGNED_BATCH_REQUEST_ADAPTER.validate_python(transformed_request)
batch_response = _send_presigned_batch_request(
client=sync_httpx_client,
request=presigned_batch_request,
timeout=timeout,
)
elif isinstance(transformed_request, dict):
@ -3937,17 +4011,19 @@ class BaseLLMHTTPHandler:
try:
if isinstance(transformed_request, dict) and "method" in transformed_request:
# Handle pre-signed requests (e.g., from Bedrock with AWS auth)
method = transformed_request["method"].lower()
request_kwargs = {
"url": transformed_request["url"],
"headers": transformed_request["headers"],
}
# Only add data for non-GET requests
if method != "get" and transformed_request.get("data") is not None:
request_kwargs["data"] = transformed_request["data"]
batch_response = getattr(sync_httpx_client, method)(**request_kwargs)
presigned_batch_request = _PRESIGNED_BATCH_REQUEST_ADAPTER.validate_python(transformed_request)
retrieve_method = presigned_batch_request["method"].lower()
if retrieve_method == "get":
batch_response = getattr(sync_httpx_client, retrieve_method)(
url=presigned_batch_request["url"],
headers=presigned_batch_request["headers"],
)
else:
batch_response = getattr(sync_httpx_client, retrieve_method)(
url=presigned_batch_request["url"],
headers=presigned_batch_request["headers"],
data=presigned_batch_request["data"],
)
elif isinstance(transformed_request, dict) and api_base:
# For other providers that use JSON requests
batch_response = sync_httpx_client.get(
@ -4014,10 +4090,10 @@ class BaseLLMHTTPHandler:
try:
if isinstance(transformed_request, dict) and "method" in transformed_request:
# Handle pre-signed requests (e.g., from Bedrock with AWS auth)
batch_response = await getattr(async_httpx_client, transformed_request["method"].lower())(
url=transformed_request["url"],
headers=transformed_request["headers"],
data=transformed_request["data"],
presigned_batch_request = _PRESIGNED_BATCH_REQUEST_ADAPTER.validate_python(transformed_request)
batch_response = await _asend_presigned_batch_request(
client=async_httpx_client,
request=presigned_batch_request,
timeout=timeout,
)
elif isinstance(transformed_request, dict):
@ -4094,17 +4170,19 @@ class BaseLLMHTTPHandler:
try:
if isinstance(transformed_request, dict) and "method" in transformed_request:
# Handle pre-signed requests (e.g., from Bedrock with AWS auth)
method = transformed_request["method"].lower()
request_kwargs = {
"url": transformed_request["url"],
"headers": transformed_request["headers"],
}
# Only add data for non-GET requests
if method != "get" and transformed_request.get("data") is not None:
request_kwargs["data"] = transformed_request["data"]
batch_response = await getattr(async_httpx_client, method)(**request_kwargs)
presigned_batch_request = _PRESIGNED_BATCH_REQUEST_ADAPTER.validate_python(transformed_request)
retrieve_method = presigned_batch_request["method"].lower()
if retrieve_method == "get":
batch_response = await getattr(async_httpx_client, retrieve_method)(
url=presigned_batch_request["url"],
headers=presigned_batch_request["headers"],
)
else:
batch_response = await getattr(async_httpx_client, retrieve_method)(
url=presigned_batch_request["url"],
headers=presigned_batch_request["headers"],
data=presigned_batch_request["data"],
)
elif isinstance(transformed_request, dict) and api_base:
# For other providers that use JSON requests
batch_response = await async_httpx_client.get(
@ -4361,7 +4439,6 @@ class BaseLLMHTTPHandler:
api_key=litellm_params.api_key,
model=model,
)
body_kwargs: dict[str, Any] = {"data": signed_body} if signed_body is not None else {"json": data}
## LOGGING
logging_obj.pre_call(
@ -4375,7 +4452,10 @@ class BaseLLMHTTPHandler:
)
try:
response = sync_httpx_client.post(url=url, headers=headers, timeout=timeout, **body_kwargs)
if signed_body is not None:
response = sync_httpx_client.post(url=url, headers=headers, timeout=timeout, data=signed_body)
else:
response = sync_httpx_client.post(url=url, headers=headers, timeout=timeout, json=data)
except Exception as e:
raise self._handle_error(
@ -4453,7 +4533,6 @@ class BaseLLMHTTPHandler:
api_key=litellm_params.api_key,
model=model,
)
body_kwargs: dict[str, Any] = {"data": signed_body} if signed_body is not None else {"json": data}
## LOGGING
logging_obj.pre_call(
@ -4467,7 +4546,10 @@ class BaseLLMHTTPHandler:
)
try:
response = await async_httpx_client.post(url=url, headers=headers, timeout=timeout, **body_kwargs)
if signed_body is not None:
response = await async_httpx_client.post(url=url, headers=headers, timeout=timeout, data=signed_body)
else:
response = await async_httpx_client.post(url=url, headers=headers, timeout=timeout, json=data)
except Exception as e:
raise self._handle_error(
@ -9453,7 +9535,9 @@ class BaseLLMHTTPHandler:
litellm_params=dict(litellm_params),
extra_body=extra_body,
)
all_optional_params: dict[str, Any] = dict(litellm_params)
all_optional_params: dict[str, object] = dict(
litellm_params
) # mutable-ok: built once via dict() from litellm_params, matching that dict-based contract
all_optional_params.update(vector_store_search_optional_params or {})
headers, signed_json_body = vector_store_provider_config.sign_request(
headers=headers,
@ -9549,7 +9633,9 @@ class BaseLLMHTTPHandler:
extra_body=extra_body,
)
all_optional_params: dict[str, Any] = dict(litellm_params)
all_optional_params: dict[str, object] = dict(
litellm_params
) # mutable-ok: built once via dict() from litellm_params, matching that dict-based contract
all_optional_params.update(vector_store_search_optional_params or {})
headers, signed_json_body = vector_store_provider_config.sign_request(
@ -9869,7 +9955,7 @@ class BaseLLMHTTPHandler:
url = api_base
params: dict[str, Any] = {}
params: dict[str, str | int] = {} # mutable-ok: populated incrementally as query params below
if after is not None:
params["after"] = after
if before is not None:
@ -9947,7 +10033,7 @@ class BaseLLMHTTPHandler:
url = api_base
params: dict[str, Any] = {}
params: dict[str, str | int] = {} # mutable-ok: populated incrementally as query params below
if after is not None:
params["after"] = after
if before is not None:
@ -10010,13 +10096,16 @@ class BaseLLMHTTPHandler:
encoded_vector_store_id = encode_url_path_segment(vector_store_id, field_name="vector_store_id")
url = f"{api_base}/{encoded_vector_store_id}"
request_body: dict[str, Any] = dict(vector_store_update_optional_params)
request_body: dict[str, object] = dict(
vector_store_update_optional_params
) # mutable-ok: built once via dict() from the optional-params payload
# Clean metadata to only include string values (OpenAI requirement)
if "metadata" in request_body and request_body["metadata"] is not None:
metadata_value = request_body.get("metadata")
if isinstance(metadata_value, dict):
from litellm.utils import add_openai_metadata
request_body["metadata"] = add_openai_metadata(request_body["metadata"])
request_body["metadata"] = add_openai_metadata(metadata_value)
if extra_body:
request_body.update(extra_body)
@ -10088,13 +10177,16 @@ class BaseLLMHTTPHandler:
encoded_vector_store_id = encode_url_path_segment(vector_store_id, field_name="vector_store_id")
url = f"{api_base}/{encoded_vector_store_id}"
request_body: dict[str, Any] = dict(vector_store_update_optional_params)
request_body: dict[str, object] = dict(
vector_store_update_optional_params
) # mutable-ok: built once via dict() from the optional-params payload
# Clean metadata to only include string values (OpenAI requirement)
if "metadata" in request_body and request_body["metadata"] is not None:
metadata_value = request_body.get("metadata")
if isinstance(metadata_value, dict):
from litellm.utils import add_openai_metadata
request_body["metadata"] = add_openai_metadata(request_body["metadata"])
request_body["metadata"] = add_openai_metadata(metadata_value)
if extra_body:
request_body.update(extra_body)

153
litellm/main.py
View file

@ -29,6 +29,7 @@ from typing import (
Any,
Literal,
Optional,
TypedDict,
Union,
cast,
get_args,
@ -40,6 +41,8 @@ from litellm._uuid import uuid
if TYPE_CHECKING:
from aiohttp import ClientSession
from litellm.router import Router
import dotenv
import httpx
import openai
@ -350,7 +353,7 @@ class LiteLLM:
class Chat:
def __init__(self, params, router_obj: Any | None):
def __init__(self, params, router_obj: "Router | None"):
self.params = params
if self.params.get("acompletion", False) is True:
self.params.pop("acompletion")
@ -360,7 +363,7 @@ class Chat:
class Completions:
def __init__(self, params, router_obj: Any | None):
def __init__(self, params, router_obj: "Router | None"):
self.params = params
self.router_obj = router_obj
@ -376,7 +379,7 @@ class Completions:
class AsyncCompletions:
def __init__(self, params, router_obj: Any | None):
def __init__(self, params, router_obj: "Router | None"):
self.params = params
self.router_obj = router_obj
@ -1137,13 +1140,18 @@ def _register_custom_pricing_for_request(
)
_CompletionClient = Union[
openai.OpenAI, openai.AsyncOpenAI, openai.AzureOpenAI, openai.AsyncAzureOpenAI, HTTPHandler, AsyncHTTPHandler
]
def _complete_azure(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
_azure_detection_model = ctx._azure_detection_model
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
api_version = ctx.api_version
client = ctx.client
client: _CompletionClient | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
extra_headers = ctx.extra_headers
headers = ctx.headers
@ -1193,6 +1201,9 @@ def _complete_azure(ctx: _CompletionDispatchContext) -> _CompletionDispatchResul
if max_retries is not None:
optional_params["max_retries"] = max_retries
config: dict[
str, object
] # mutable-ok: declared ahead of a branch that assigns it from get_config()'s dict contract
if litellm.AzureOpenAIO1Config().is_o_series_model(model=_azure_detection_model):
## LOAD CONFIG - if set
config = litellm.AzureOpenAIO1Config.get_config()
@ -1275,7 +1286,7 @@ def _complete_azure_text(ctx: _CompletionDispatchContext) -> _CompletionDispatch
api_base = ctx.api_base
api_key = ctx.api_key
api_version = ctx.api_version
client = ctx.client
client: _CompletionClient | None = ctx.client
extra_headers = ctx.extra_headers
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1318,7 +1329,9 @@ def _complete_azure_text(ctx: _CompletionDispatchContext) -> _CompletionDispatch
optional_params["extra_headers"] = extra_headers
## LOAD CONFIG - if set
config = litellm.AzureOpenAIConfig.get_config()
config: dict[str, object] = (
litellm.AzureOpenAIConfig.get_config()
) # mutable-ok: matches get_config()'s plain dict return contract
for k, v in config.items():
if (
k not in optional_params
@ -1367,7 +1380,7 @@ def _complete_deepseek(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1418,7 +1431,7 @@ def _complete_azure_ai(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
extra_headers = ctx.extra_headers
headers = ctx.headers
@ -1574,7 +1587,7 @@ def _complete_text_completion_openai(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: _CompletionClient | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1605,7 +1618,9 @@ def _complete_text_completion_openai(
headers = headers or litellm.headers
## LOAD CONFIG - if set
config = litellm.OpenAITextCompletionConfig.get_config()
config: dict[str, object] = (
litellm.OpenAITextCompletionConfig.get_config()
) # mutable-ok: matches get_config()'s plain dict return contract
for k, v in config.items():
if (
k not in optional_params
@ -1656,7 +1671,7 @@ def _complete_fireworks_ai(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1707,7 +1722,7 @@ def _complete_heroku(ctx: _CompletionDispatchContext) -> _CompletionDispatchResu
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1757,7 +1772,7 @@ def _complete_ragflow(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1807,7 +1822,7 @@ def _complete_xai(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1858,7 +1873,7 @@ def _complete_groq(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1889,7 +1904,9 @@ def _complete_groq(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult
headers = headers or litellm.headers
## LOAD CONFIG - if set
config = litellm.GroqChatConfig.get_config()
config: dict[str, object] = (
litellm.GroqChatConfig.get_config()
) # mutable-ok: matches get_config()'s plain dict return contract
for k, v in config.items():
if (
k not in optional_params
@ -1922,7 +1939,7 @@ def _complete_bedrock_mantle(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -1938,7 +1955,9 @@ def _complete_bedrock_mantle(
api_base = api_base or litellm.api_base or get_secret("BEDROCK_MANTLE_API_BASE")
api_key = api_key or litellm.api_key or get_secret("BEDROCK_MANTLE_API_KEY")
headers = headers or litellm.headers
config = litellm.BedrockMantleChatConfig.get_config()
config: dict[str, object] = (
litellm.BedrockMantleChatConfig.get_config()
) # mutable-ok: matches get_config()'s plain dict return contract
for k, v in config.items():
if k not in optional_params:
optional_params[k] = v
@ -1966,7 +1985,7 @@ def _complete_a2a(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -2029,7 +2048,7 @@ def _complete_gigachat(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -2091,7 +2110,7 @@ def _complete_sap(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -2106,7 +2125,9 @@ def _complete_sap(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
headers = headers or litellm.headers
## LOAD CONFIG - if set
config = litellm.GenAIHubOrchestrationConfig.get_config()
config: dict[str, object] = (
litellm.GenAIHubOrchestrationConfig.get_config()
) # mutable-ok: matches get_config()'s plain dict return contract
for k, v in config.items():
if (
k not in optional_params
@ -2139,7 +2160,7 @@ def _complete_aiohttp_openai(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | ClientSession | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
extra_headers = ctx.extra_headers
headers = ctx.headers
@ -2194,7 +2215,7 @@ def _complete_cometapi(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -2243,7 +2264,7 @@ def _complete_minimax(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -2289,7 +2310,7 @@ def _complete_hosted_vllm(ctx: _CompletionDispatchContext) -> _CompletionDispatc
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -2335,7 +2356,7 @@ def _complete_custom_openai(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
custom_prompt_dict = ctx.custom_prompt_dict
extra_headers = ctx.extra_headers
@ -2401,7 +2422,9 @@ def _complete_custom_openai(
optional_params["metadata"] = openai_metadata
## LOAD CONFIG - if set
config = litellm.OpenAIConfig.get_config()
config: dict[str, object] = (
litellm.OpenAIConfig.get_config()
) # mutable-ok: matches get_config()'s plain dict return contract
for k, v in config.items():
if (
k not in optional_params
@ -2479,7 +2502,7 @@ def _complete_mistral(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -2630,7 +2653,7 @@ def _complete_anthropic(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: _CompletionClient | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
custom_prompt_dict = ctx.custom_prompt_dict
headers = ctx.headers
@ -2929,7 +2952,7 @@ def _complete_huggingface(ctx: _CompletionDispatchContext) -> _CompletionDispatc
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -2972,7 +2995,7 @@ def _complete_oci(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -3007,7 +3030,7 @@ def _complete_compactifai(ctx: _CompletionDispatchContext) -> _CompletionDispatc
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -3083,7 +3106,7 @@ def _complete_databricks(ctx: _CompletionDispatchContext) -> _CompletionDispatch
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
headers = ctx.headers
litellm_params = ctx.litellm_params
logging = ctx.logging
@ -3155,7 +3178,7 @@ def _complete_datarobot(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -3192,7 +3215,7 @@ def _complete_openrouter(ctx: _CompletionDispatchContext) -> _CompletionDispatch
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
headers = ctx.headers
litellm_params = ctx.litellm_params
logging = ctx.logging
@ -3229,7 +3252,9 @@ def _complete_openrouter(ctx: _CompletionDispatchContext) -> _CompletionDispatch
headers = openrouter_headers
## Load Config
config = litellm.OpenrouterConfig.get_config()
config: dict[str, object] = (
litellm.OpenrouterConfig.get_config()
) # mutable-ok: matches get_config()'s plain dict return contract
for k, v in config.items():
if k == "extra_body":
# we use openai 'extra_body' to pass openrouter specific params - transforms, route, models
@ -3271,7 +3296,7 @@ def _complete_vercel_ai_gateway(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
headers = ctx.headers
litellm_params = ctx.litellm_params
logging = ctx.logging
@ -3307,7 +3332,9 @@ def _complete_vercel_ai_gateway(
headers = vercel_headers
## Load Config
config = litellm.VercelAIGatewayConfig.get_config()
config: dict[str, object] = (
litellm.VercelAIGatewayConfig.get_config()
) # mutable-ok: matches get_config()'s plain dict return contract
for k, v in config.items():
if k == "extra_body":
# we use openai 'extra_body' to pass vercel specific params - providerOptions
@ -3349,7 +3376,7 @@ def _complete_vertex_ai_beta(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -3414,7 +3441,7 @@ def _complete_vertex_ai_beta(
def _complete_vertex_ai(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
acompletion = ctx.acompletion
api_base = ctx.api_base
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
custom_prompt_dict = ctx.custom_prompt_dict
headers = ctx.headers
@ -3711,7 +3738,7 @@ def _complete_text_completion_inception(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: _CompletionClient | None = ctx.client
headers = ctx.headers
litellm_params = ctx.litellm_params
logger_fn = ctx.logger_fn
@ -3775,7 +3802,7 @@ def _complete_sagemaker_chat(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -3838,7 +3865,7 @@ def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_prompt_dict = ctx.custom_prompt_dict
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -3962,7 +3989,7 @@ def _complete_watsonx(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_prompt_dict = ctx.custom_prompt_dict
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -4001,7 +4028,7 @@ def _complete_watsonx_text(
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
headers = ctx.headers
litellm_params = ctx.litellm_params
logging = ctx.logging
@ -4113,7 +4140,7 @@ def _complete_ollama(ctx: _CompletionDispatchContext) -> _CompletionDispatchResu
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
headers = ctx.headers
litellm_params = ctx.litellm_params
logging = ctx.logging
@ -4153,7 +4180,7 @@ def _complete_ollama_chat(ctx: _CompletionDispatchContext) -> _CompletionDispatc
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
headers = ctx.headers
litellm_params = ctx.litellm_params
logging = ctx.logging
@ -4268,7 +4295,7 @@ def _complete_cloudflare(ctx: _CompletionDispatchContext) -> _CompletionDispatch
def _complete_petals(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
api_base = ctx.api_base
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
litellm_params = ctx.litellm_params
logger_fn = ctx.logger_fn
logging = ctx.logging
@ -4310,7 +4337,7 @@ def _complete_snowflake(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: _CompletionClient | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -4398,7 +4425,7 @@ def _complete_gdc(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -4437,7 +4464,7 @@ def _complete_bytez(ctx: _CompletionDispatchContext) -> _CompletionDispatchResul
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -4477,7 +4504,7 @@ def _complete_lemonade(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -4517,7 +4544,7 @@ def _complete_ovhcloud(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -4697,7 +4724,7 @@ def _complete_langgraph(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -4746,7 +4773,7 @@ def _complete_langflow(ctx: _CompletionDispatchContext) -> _CompletionDispatchRe
acompletion = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client = ctx.client
client: HTTPHandler | AsyncHTTPHandler | None = ctx.client
custom_llm_provider = ctx.custom_llm_provider
headers = ctx.headers
litellm_params = ctx.litellm_params
@ -4904,7 +4931,7 @@ def completion( # type: ignore
thinking = validate_and_fix_thinking_param(thinking=thinking)
######### unpacking kwargs #####################
args = locals()
args: dict[str, object] = locals() # mutable-ok: locals() itself returns a plain mutable dict
# Set by the responses->completion fallback so completion() does not bridge
# back to the Responses API: that round-trip mutually recurses forever for a
@ -7144,7 +7171,9 @@ def text_completion(
tokenizer = tiktoken.encoding_for_model("text-davinci-003")
## if it's a 2d list - each element in the list is a text_completion() request
if len(prompt) > 0 and isinstance(prompt[0], list):
responses = [None for x in prompt] # init responses
responses: list[TextChoices | None] = [
None for x in prompt
] # mutable-ok: entries are overwritten in place by the thread pool loop below
def process_prompt(i, individual_prompt):
decoded_prompt = tokenizer.decode(individual_prompt)
@ -7161,7 +7190,7 @@ def text_completion(
text_completion_response["object"] = "text_completion"
text_completion_response["created"] = response.get("created", None)
text_completion_response["model"] = response.get("model", None)
return response["choices"][0]
return response.choices[0]
with concurrent.futures.ThreadPoolExecutor() as executor:
completed_futures = [
@ -8442,7 +8471,7 @@ def stream_chunk_builder(
if isinstance(delta_obj, dict):
delta = delta_obj
elif hasattr(delta_obj, "model_dump"):
delta = cast(dict[str, Any], delta_obj.model_dump())
delta = cast(dict[str, object], delta_obj.model_dump())
else:
delta = {}
@ -8618,7 +8647,9 @@ def stream_chunk_builder(
]
if len(provider_specific_chunks) > 0:
combined_provider_fields: dict[str, Any] = {}
combined_provider_fields: dict[
str, object
] = {} # mutable-ok: populated incrementally from provider field lookups below
for chunk in provider_specific_chunks:
fields = chunk["choices"][0]["delta"]["provider_specific_fields"]
if isinstance(fields, dict):

4
litellm/proxy/_lazy_openapi_snapshot.py
View file

@ -118,9 +118,11 @@ def generate_snapshot() -> dict[str, dict]:
break
op["tags"] = [feat.name]
full = ensure_unique_openapi_operation_ids(full, used_operation_ids)
components = full.get("components", {})
schemas = components.get("schemas", {}) if isinstance(components, dict) else {}
fragments[feat.name] = {
"paths": paths,
"components": {"schemas": full.get("components", {}).get("schemas", {})},
"components": {"schemas": schemas},
}
return fragments

123
litellm/proxy/proxy_server.py
View file

@ -23,6 +23,7 @@ from typing import (
Any,
Literal,
Optional,
TypeAlias,
TypedDict,
Union,
cast,
@ -34,7 +35,8 @@ from typing import (
import anyio
import websockets
import websockets.exceptions
from pydantic import BaseModel, Json, JsonValue
from pydantic import BaseModel, Json, JsonValue, TypeAdapter
from pydantic_core import ErrorDetails
from typing_extensions import NotRequired, assert_never
from litellm._uuid import uuid
@ -130,9 +132,11 @@ if TYPE_CHECKING:
from litellm.integrations.opentelemetry import OpenTelemetry
Span = Union[_Span, Any]
_OtelSpanType: TypeAlias = _Span
else:
Span = Any
OpenTelemetry = Any
_OtelSpanType = Any
REALTIME_REQUEST_SCOPE_TEMPLATE: dict[str, Any] = {
"type": "http",
@ -653,7 +657,7 @@ from fastapi.responses import (
RedirectResponse,
StreamingResponse,
)
from fastapi.routing import APIRouter
from fastapi.routing import APIRoute, APIRouter
from fastapi.security import OAuth2PasswordBearer
from fastapi.security.api_key import APIKeyHeader
from fastapi.staticfiles import StaticFiles
@ -720,7 +724,7 @@ def _redact_worker_config_for_logging(worker_config: str | dict[str, JsonValue]
return None
if isinstance(worker_config, dict):
return _redact_secret_values_in_obj(worker_config)
parsed = safe_json_loads(worker_config, default=None)
parsed = TypeAdapter(JsonValue | None).validate_python(safe_json_loads(worker_config, default=None))
if isinstance(parsed, dict):
return safe_dumps(_redact_secret_values_in_obj(parsed))
return worker_config
@ -844,21 +848,14 @@ async def _initialize_shared_aiohttp_session():
_build_aiohttp_keepalive_socket_factory,
)
connector_kwargs: dict[str, Any] = {
"keepalive_timeout": AIOHTTP_KEEPALIVE_TIMEOUT,
"ttl_dns_cache": AIOHTTP_TTL_DNS_CACHE,
}
if AIOHTTP_NEEDS_CLEANUP_CLOSED:
connector_kwargs["enable_cleanup_closed"] = True
if AIOHTTP_CONNECTOR_LIMIT > 0:
connector_kwargs["limit"] = AIOHTTP_CONNECTOR_LIMIT
if AIOHTTP_CONNECTOR_LIMIT_PER_HOST > 0:
connector_kwargs["limit_per_host"] = AIOHTTP_CONNECTOR_LIMIT_PER_HOST
socket_factory = _build_aiohttp_keepalive_socket_factory()
if socket_factory is not None:
connector_kwargs["socket_factory"] = socket_factory
connector = TCPConnector(**connector_kwargs)
connector = TCPConnector(
keepalive_timeout=AIOHTTP_KEEPALIVE_TIMEOUT,
ttl_dns_cache=AIOHTTP_TTL_DNS_CACHE,
enable_cleanup_closed=AIOHTTP_NEEDS_CLEANUP_CLOSED,
limit=AIOHTTP_CONNECTOR_LIMIT if AIOHTTP_CONNECTOR_LIMIT > 0 else 100,
limit_per_host=max(0, AIOHTTP_CONNECTOR_LIMIT_PER_HOST),
socket_factory=_build_aiohttp_keepalive_socket_factory(),
)
session = ClientSession(connector=connector)
verbose_proxy_logger.info(
@ -911,7 +908,9 @@ async def proxy_startup_event(app: FastAPI):
)
_module_path, _func_name = _hook_spec.rsplit(":", 1)
_module = importlib.import_module(_module_path)
_hook_fn = getattr(_module, _func_name)
_hook_fn: object = getattr(_module, _func_name)
if not callable(_hook_fn):
raise TypeError(f"Hook '{_hook_spec}' is not callable")
if inspect.iscoroutinefunction(_hook_fn):
await _hook_fn()
else:
@ -959,9 +958,9 @@ async def proxy_startup_event(app: FastAPI):
await initialize(**worker_config)
else:
# if not, assume it's a json string
worker_config = json.loads(worker_config)
if isinstance(worker_config, dict):
await initialize(**worker_config)
parsed_worker_config = TypeAdapter(JsonValue).validate_python(json.loads(worker_config))
if isinstance(parsed_worker_config, dict):
await initialize(**parsed_worker_config)
# check if DATABASE_URL in environment - load from there
if prisma_client is None:
@ -1162,7 +1161,7 @@ async def proxy_startup_event(app: FastAPI):
await proxy_shutdown_event() # type: ignore[reportGeneralTypeIssues]
def _generate_stable_operation_id(route: Any) -> str:
def _generate_stable_operation_id(route: APIRoute) -> str:
operation_id = re.sub(r"\W", "_", f"{route.name}{route.path_format}")
route_methods = sorted(route.methods or [])
if len(route_methods) == 1:
@ -1199,22 +1198,26 @@ def _strip_operation_id_method_suffix(operation_id: str) -> str:
def ensure_unique_openapi_operation_ids(
openapi_schema: dict[str, Any],
openapi_schema: dict[str, JsonValue], # mutable-ok: parameter mirrors the caller's OpenAPI schema dict payload
reserved_operation_ids: set[str] | None = None,
) -> dict[str, Any]:
operation_entries = []
) -> dict[str, JsonValue]: # mutable-ok: mirrors the OpenAPI schema dict payload this function returns
operation_entries: list[
tuple[str, dict[str, JsonValue], str]
] = [] # mutable-ok: accumulates operation entries in the loop below
operation_id_counts: dict[str, int] = {}
for path_item in openapi_schema.get("paths", {}).values():
if not isinstance(path_item, dict):
continue
for method, operation in path_item.items():
if method not in _OPENAPI_HTTP_METHODS or not isinstance(operation, dict):
paths = openapi_schema.get("paths", {})
if isinstance(paths, dict):
for path_item in paths.values():
if not isinstance(path_item, dict):
continue
operation_id = operation.get("operationId")
if not isinstance(operation_id, str):
continue
operation_entries.append((method, operation, operation_id))
operation_id_counts[operation_id] = operation_id_counts.get(operation_id, 0) + 1
for method, operation in path_item.items():
if method not in _OPENAPI_HTTP_METHODS or not isinstance(operation, dict):
continue
operation_id = operation.get("operationId")
if not isinstance(operation_id, str):
continue
operation_entries.append((method, operation, operation_id))
operation_id_counts[operation_id] = operation_id_counts.get(operation_id, 0) + 1
used_operation_ids = set(reserved_operation_ids or set())
seen_operation_ids: set[str] = set()
@ -1421,7 +1424,7 @@ async def openai_exception_handler(request: Request, exc: ProxyException):
def _close_dangling_otel_server_span(request: Request, status_code: int, exc: Exception | None = None) -> None:
parent_otel_span = getattr(request.state, "parent_otel_span", None)
parent_otel_span: _OtelSpanType | None = getattr(request.state, "parent_otel_span", None)
if parent_otel_span is None:
return
if open_telemetry_logger is None:
@ -1468,13 +1471,16 @@ async def management_problem_exception_handler(request: Request, exc: Management
async def otel_request_validation_exception_handler(request: Request, exc: RequestValidationError):
if request.url.path.startswith(MANAGEMENT_V1_PREFIX):
_close_dangling_otel_server_span(request, 400, exc=exc)
validation_errors: list[ErrorDetails] = list(
exc.errors()
) # mutable-ok: collects validation errors in the loop below
return problem_response(
ProblemDetail(
type=f"{PROBLEM_TYPE_BASE}invalid-query-parameter",
title="Invalid query parameter",
status=400,
detail="; ".join(
f"{'.'.join(str(part) for part in error['loc'][1:])}: {error['msg']}" for error in exc.errors()
f"{'.'.join(str(part) for part in error['loc'][1:])}: {error['msg']}" for error in validation_errors
)
or "The request query parameters are invalid.",
)
@ -2033,7 +2039,9 @@ use_queue = False
health_check_interval = None
health_check_concurrency = None
health_check_details = None
health_check_results: dict[str, int | list[dict[str, Any]]] = {}
health_check_results: dict[
str, int | list[dict[str, object]]
] = {} # mutable-ok: populated incrementally per model in the health check loop below
background_health_check_loop_active = False
background_health_check_cycle_seq = 0
queue: list = []
@ -2340,7 +2348,7 @@ async def _read_spend_counter_estimate(counter_key: str, fallback_spend: float)
redis_clean_miss = False
if spend_counter_cache.redis_cache is not None:
try:
val = await spend_counter_cache.redis_cache.async_get_cache(key=counter_key)
val: str | float | int | None = await spend_counter_cache.redis_cache.async_get_cache(key=counter_key)
if val is not None:
return float(val), True
redis_clean_miss = True
@ -2706,7 +2714,7 @@ async def _get_source_cache_base_spend(
) -> float:
source_cache_keys = [source_cache_key] if isinstance(source_cache_key, str) else source_cache_key
for cache_key in source_cache_keys:
source = await user_api_key_cache.async_get_cache(key=cache_key)
source = TypeAdapter(object).validate_python(await user_api_key_cache.async_get_cache(key=cache_key))
if source is None:
continue
if isinstance(source, dict):
@ -2745,7 +2753,7 @@ async def _ensure_window_spend_counter_initialized(
async def _is_spend_counter_cache_warm(counter_key: str) -> bool:
if spend_counter_cache.redis_cache is not None:
try:
current_value = await spend_counter_cache.redis_cache.async_get_cache(
current_value: str | float | int | None = await spend_counter_cache.redis_cache.async_get_cache(
key=counter_key,
)
if current_value is None:
@ -2817,7 +2825,7 @@ async def update_cache(
Put any alerting logic in here.
"""
values_to_update_in_cache: list[tuple[Any, Any]] = []
values_to_update_in_cache: list[tuple[str, object]] = [] # mutable-ok: accumulates cache updates in the loop below
### UPDATE KEY SPEND ###
async def _update_key_cache(token: str, response_cost: float):
@ -2883,7 +2891,7 @@ async def update_cache(
# Fetch the existing cost for the given user
if _id is None:
continue
cached_user = await user_api_key_cache.async_get_cache(key=_id)
cached_user = TypeAdapter(object).validate_python(await user_api_key_cache.async_get_cache(key=_id))
if cached_user is None:
# do nothing if there is no cache value
return
@ -2906,7 +2914,10 @@ async def update_cache(
)
)
## UPDATE GLOBAL PROXY ##
global_proxy_spend = await user_api_key_cache.async_get_cache(key=GLOBAL_PROXY_SPEND_CACHE_KEY)
global_proxy_spend = TypeAdapter(float | int | None).validate_python(
await user_api_key_cache.async_get_cache(key=GLOBAL_PROXY_SPEND_CACHE_KEY),
strict=True,
)
if global_proxy_spend is None:
# do nothing if not in cache
return
@ -2932,7 +2943,7 @@ async def update_cache(
_id = f"end_user_id:{end_user_id}"
try:
# Fetch the existing cost for the given user
cached_end_user = await user_api_key_cache.async_get_cache(key=_id)
cached_end_user = TypeAdapter(object).validate_python(await user_api_key_cache.async_get_cache(key=_id))
if cached_end_user is None:
# if user does not exist in LiteLLM_UserTable, create a new user
# do nothing if end-user not in api key cache
@ -2973,7 +2984,7 @@ async def update_cache(
_id = f"team_id:{team_id}"
try:
cached_team = await user_api_key_cache.async_get_cache(key=_id)
cached_team = TypeAdapter(object).validate_python(await user_api_key_cache.async_get_cache(key=_id))
if cached_team is None:
# do nothing if team not in api key cache
return
@ -3023,7 +3034,9 @@ async def update_cache(
cache_key = f"tag:{tag_name}"
# Fetch the existing tag object from cache
cached_tag = await user_api_key_cache.async_get_cache(key=cache_key)
cached_tag = TypeAdapter(object).validate_python(
await user_api_key_cache.async_get_cache(key=cache_key)
)
if cached_tag is None:
# do nothing if tag not in api key cache
continue
@ -3527,11 +3540,13 @@ _DB_OVERLAY_REMOTE_MODULE_LIST_FIELDS: dict[str, tuple[str, ...]] = {
}
def _is_remote_module_url(value: Any) -> bool:
def _is_remote_module_url(value: object) -> bool:
return isinstance(value, str) and (value.startswith("s3://") or value.startswith("gcs://"))
def _scrub_guardrail_inner(inner: dict[str, Any]) -> None:
def _scrub_guardrail_inner(
inner: dict[str, JsonValue],
) -> None: # mutable-ok: parameter mirrors the caller's guardrail config dict payload
"""Strip remote-URL entries from a guardrail's ``callbacks`` list
and ``guardrail`` (v2 module-path) field. Mutates in place."""
cbs = inner.get("callbacks")
@ -3551,7 +3566,7 @@ def _scrub_guardrail_inner(inner: dict[str, Any]) -> None:
inner["guardrail"] = None
def _scrub_db_overlay_remote_module_loads(section: str, db_value: Any) -> Any:
def _scrub_db_overlay_remote_module_loads(section: str, db_value: JsonValue) -> JsonValue:
"""Strip ``s3://`` / ``gcs://`` entries from the DB-overlay value for
fields whose contents reach ``get_instance_fn``. The same scheme is
allowed from a YAML config (the documented operator flow) but a
@ -3699,7 +3714,7 @@ def _build_redis_usage_cache(redis_params: Mapping[str, object]) -> RedisCache:
if startup_nodes is None:
env_cluster_nodes = get_secret_str("REDIS_CLUSTER_NODES")
if env_cluster_nodes is not None:
startup_nodes = json.loads(env_cluster_nodes)
startup_nodes = TypeAdapter(object).validate_python(json.loads(env_cluster_nodes))
non_node_params = {key: value for key, value in redis_params.items() if key != "startup_nodes"}
if startup_nodes:
return RedisClusterCache(startup_nodes=startup_nodes, **non_node_params)
@ -15236,7 +15251,7 @@ def _general_settings_ui_litellm_default(
return False if spec["type"] == "Boolean" else None
def _validate_general_settings_ui_litellm_value(field_name: str, value: Any) -> GeneralSettingsUILiteLLMValue:
def _validate_general_settings_ui_litellm_value(field_name: str, value: JsonValue) -> GeneralSettingsUILiteLLMValue:
spec = _GENERAL_SETTINGS_UI_LITELLM_FIELDS[field_name]
field_type = spec["type"]
if value is None or value == "":
@ -15276,7 +15291,7 @@ def _validate_general_settings_ui_litellm_value(field_name: str, value: Any) ->
async def _persist_general_settings_ui_litellm_field(
field_name: str, value: Any, user_api_key_dict: UserAPIKeyAuth
field_name: str, value: JsonValue, user_api_key_dict: UserAPIKeyAuth
) -> dict:
validated = _validate_general_settings_ui_litellm_value(field_name, value)
config = await proxy_config.get_config()

58
litellm/proxy/utils.py
View file

@ -21,6 +21,7 @@ from typing import (
ClassVar,
Literal,
Optional,
TypedDict,
Union,
cast,
overload,
@ -160,6 +161,7 @@ from litellm.repositories.verification_token_repository import (
)
from litellm.secret_managers.main import str_to_bool
from litellm.types.integrations.slack_alerting import DEFAULT_ALERT_TYPES
from litellm.types.llms.base import HiddenParams
from litellm.types.mcp import (
MCPDuringCallResponseObject,
MCPPreCallRequestObject,
@ -345,7 +347,7 @@ def _accepts_litellm_call_info(cb: CustomLogger) -> bool:
return _CALLBACK_ACCEPTS_CALL_INFO[key]
def _enrich_http_exception_with_guardrail_context(exc: BaseException, callback: Any) -> None:
def _enrich_http_exception_with_guardrail_context(exc: BaseException, callback: CustomLogger) -> None:
"""
If `exc` is an HTTPException with a dict `detail`, mutate it in place to
add `guardrail_name` and `guardrail_mode` taken from the callback instance.
@ -377,6 +379,15 @@ def _exception_changes_request_flow(exc: BaseException) -> bool:
return isinstance(exc, (SensitiveDataRouteException, ModifyResponseException))
class _MCPPreCallHookResult(TypedDict):
should_proceed: bool
modified_arguments: (
dict[str, object] | None
) # mutable-ok: parameter mirrors the caller's request-arguments dict payload
error_message: str | None
hidden_params: HiddenParams
@dataclass(frozen=True)
class _CallbackCapabilities:
"""Cached per-hook capability flags derived from ``litellm.callbacks``.
@ -394,11 +405,11 @@ class _CallbackCapabilities:
# Tuple[(resolved_callback, "override" | "apply_guardrail"), ...]
# Ordered the same as ``litellm.callbacks``; used to build the streaming
# iterator chain without re-scanning per request.
iterator_overrides: tuple[tuple[Any, str], ...] = field(default_factory=tuple)
iterator_overrides: tuple[tuple[CustomLogger, str], ...] = field(default_factory=tuple)
# Resolved CustomLogger callbacks in original order. Pre-resolving once
# avoids the per-request ``get_custom_logger_compatible_class`` walk for
# every string entry in ``litellm.callbacks``.
resolved_callbacks: tuple[Any, ...] = field(default_factory=tuple)
resolved_callbacks: tuple[CustomLogger, ...] = field(default_factory=tuple)
class ProxyLogging:
@ -676,12 +687,10 @@ class ProxyLogging:
return synthetic_data
def _convert_llm_result_to_mcp_response(self, llm_result, request_obj) -> Any | None:
def _convert_llm_result_to_mcp_response(self, llm_result, request_obj) -> MCPPreCallResponseObject | None:
"""
Convert LLM guardrail result back to MCP response format.
"""
from litellm.types.mcp import MCPPreCallResponseObject
# If result is an exception, it means the guardrail blocked the request
if isinstance(llm_result, Exception):
return MCPPreCallResponseObject(
@ -802,7 +811,7 @@ class ProxyLogging:
verbose_proxy_logger.error("Error in manual argument parsing: %s", e)
return None
def _convert_llm_result_to_mcp_during_response(self, llm_result, request_obj) -> Any | None:
def _convert_llm_result_to_mcp_during_response(self, llm_result, request_obj) -> MCPDuringCallResponseObject | None:
"""
Convert LLM guardrail result back to MCP during call response format.
"""
@ -848,7 +857,7 @@ class ProxyLogging:
self,
response: MCPPreCallResponseObject,
original_request: MCPPreCallRequestObject,
) -> dict[str, Any]:
) -> _MCPPreCallHookResult:
"""
Parse the response from the pre_mcp_tool_call_hook
@ -856,7 +865,7 @@ class ProxyLogging:
2. Apply any argument modifications
3. Handle validation errors
"""
result = {
result: _MCPPreCallHookResult = {
"should_proceed": response.should_proceed,
"modified_arguments": response.modified_arguments or original_request.arguments,
"error_message": response.error_message,
@ -868,9 +877,6 @@ class ProxyLogging:
"""
Helper function to create MCPPreCallRequestObject from kwargs for standard pre_call_hook.
"""
from litellm.types.llms.base import HiddenParams
from litellm.types.mcp import MCPPreCallRequestObject
user_api_key_auth_dict = self._convert_user_api_key_auth_to_dict(kwargs.get("user_api_key_auth"))
return MCPPreCallRequestObject(
@ -1031,7 +1037,7 @@ class ProxyLogging:
selected_guardrail = llm_router.get_available_guardrail(guardrail_name=guardrail_name)
callback = selected_guardrail.get("callback")
if callback is None:
if not isinstance(callback, CustomGuardrail):
raise ValueError(f"No callback found for guardrail: {guardrail_name}")
return await self._execute_guardrail_hook(
@ -1142,8 +1148,8 @@ class ProxyLogging:
self,
data: dict,
litellm_logging_obj: Any,
prompt_id: Any,
prompt_version: Any,
prompt_id: str,
prompt_version: int | None,
call_type: CallTypesLiteral,
) -> None:
"""Process prompt template if applicable."""
@ -1438,9 +1444,7 @@ class ProxyLogging:
data = result
elif (
_callback is not None
and isinstance(_callback, CustomLogger)
and "async_pre_call_hook" in vars(_callback.__class__)
"async_pre_call_hook" in vars(_callback.__class__)
and _callback.__class__.async_pre_call_hook != CustomLogger.async_pre_call_hook
):
if call_type == "call_mcp_tool" and user_api_key_dict is None:
@ -1614,7 +1618,7 @@ class ProxyLogging:
break
@staticmethod
async def _run_guardrail_with_metrics(callback: Any, coro: Awaitable[Any], hook_type: str) -> Any:
async def _run_guardrail_with_metrics(callback: CustomGuardrail, coro: Awaitable[Any], hook_type: str) -> Any:
"""
Await `coro`, recording its latency and status to the
`litellm_guardrail_latency_seconds` metric under `hook_type`, and
@ -1646,7 +1650,7 @@ class ProxyLogging:
@staticmethod
async def _wrap_streaming_iterator_with_enrichment(
callback: Any, gen: AsyncGenerator[Any, None]
callback: CustomLogger, gen: AsyncGenerator[Any, None]
) -> AsyncGenerator[Any, None]:
"""
Yield from `gen`; if iteration raises an HTTPException with dict detail,
@ -1691,12 +1695,14 @@ class ProxyLogging:
has_streaming_chunk_override = False
has_guardrail = False
has_pre_call_override = False
iterator_overrides: list[tuple[Any, str]] = [] # (callback, kind)
resolved_callbacks: list[Any] = []
iterator_overrides: list[
tuple[CustomLogger, str]
] = [] # mutable-ok: accumulates (callback, kind) overrides in the loop below
resolved_callbacks: list[CustomLogger] = [] # mutable-ok: accumulates resolved callbacks in the loop below
for callback in callbacks:
if isinstance(callback, str):
resolved: Any = litellm.litellm_core_utils.litellm_logging.get_custom_logger_compatible_class(
resolved = litellm.litellm_core_utils.litellm_logging.get_custom_logger_compatible_class(
cast(_custom_logger_compatible_callbacks_literal, callback)
)
else:
@ -2096,7 +2102,7 @@ class ProxyLogging:
"""
litellm_debug_info = getattr(original_exception, "litellm_debug_info", None)
exception_str = str(original_exception)
if litellm_debug_info is not None:
if isinstance(litellm_debug_info, str):
exception_str += litellm_debug_info
asyncio.create_task(
@ -2765,7 +2771,7 @@ class ProxyLogging:
),
)
else:
# kind == "apply_guardrail": route through unified_guardrail
assert isinstance(resolved_callback, CustomGuardrail)
current_response = self._wrap_streaming_iterator_with_enrichment(
resolved_callback,
unified_guardrail.async_post_call_streaming_iterator_hook(
@ -6557,7 +6563,7 @@ def model_dump_with_preserved_fields(
model_dump(exclude_none=True) strips them.
Args:
obj: The Pydantic BaseModel instance to serialize
obj: The ModelResponse / ModelResponseStream instance to serialize
preserve_fields: Deprecated, kept for backward compatibility.
exclude_unset: Whether to exclude fields that were not explicitly set

145
litellm/responses/streaming_iterator.py
View file

@ -9,7 +9,7 @@ from collections.abc import Mapping
from datetime import datetime
from functools import lru_cache
from types import MappingProxyType
from typing import Any, Literal
from typing import TYPE_CHECKING, Any, Literal, Protocol, runtime_checkable
import httpx
from openai._streaming import SSEDecoder
@ -34,6 +34,40 @@ from litellm.types.llms.openai import ResponsesAPIStreamEvents
from litellm.types.utils import CallTypes
from litellm.utils import async_post_call_success_deployment_hook
if TYPE_CHECKING:
from fastapi import WebSocket
from litellm.types.llms.openai import (
ContentPartDoneEvent,
ContentPartDonePartOutputText,
ContentPartDonePartReasoningText,
ContentPartDonePartRefusal,
ResponsesAPIResponse,
ResponsesAPIStreamingResponse,
)
@runtime_checkable
class _ModelDumpable(Protocol):
def model_dump(
self, *, exclude_none: bool = ...
) -> dict[
str, object
]: ... # mutable-ok: Protocol mirrors Pydantic's model_dump(), which returns a plain mutable dict
@runtime_checkable
class _ModelDumpJsonable(Protocol):
def model_dump_json(self, *, exclude_none: bool = ...) -> str: ...
class _BackendWebSocketLike(Protocol):
async def recv(self, decode: bool = ...) -> str | bytes: ...
async def send(self, message: str) -> None: ...
async def close(self) -> None: ...
@lru_cache(maxsize=1)
def _get_openai_response_types():
@ -42,7 +76,7 @@ def _get_openai_response_types():
return openai_types
def _log_background_task_failure(task: asyncio.Task[Any], *, task_name: str) -> None:
def _log_background_task_failure(task: asyncio.Task[object], *, task_name: str) -> None:
if task.cancelled():
return
exception = task.exception()
@ -131,7 +165,7 @@ class BaseResponsesAPIStreamingIterator:
self.logging_obj = logging_obj
self.finished = False
self.responses_api_provider_config = responses_api_provider_config
self.completed_response: Any | None = None
self.completed_response: ResponsesAPIStreamingResponse | None = None
self.start_time = getattr(logging_obj, "start_time", datetime.now())
self._failure_handled = False # Track if failure handler has been called
self._yielded_first_chunk = False
@ -176,7 +210,7 @@ class BaseResponsesAPIStreamingIterator:
llm_provider=self.custom_llm_provider or "",
)
def _process_chunk(self, chunk) -> Any | None:
def _process_chunk(self, chunk: str | None) -> ResponsesAPIStreamingResponse | None:
"""Process a single chunk of data from the stream"""
if not chunk:
return None
@ -196,7 +230,7 @@ class BaseResponsesAPIStreamingIterator:
try:
# Parse the JSON chunk
parsed_chunk = json.loads(chunk)
parsed_chunk: object = json.loads(chunk)
# Format as ResponsesAPIStreamingResponse
if isinstance(parsed_chunk, dict):
@ -401,7 +435,7 @@ class BaseResponsesAPIStreamingIterator:
)
self._handle_failure(exception)
def _record_failed_response_usage(self, response_obj: Any | None) -> None:
def _record_failed_response_usage(self, response_obj: ResponsesAPIResponse | None) -> None:
if response_obj is None or self.logging_obj is None:
return
usage_obj = getattr(response_obj, "usage", None)
@ -451,7 +485,7 @@ class BaseResponsesAPIStreamingIterator:
is_pre_first_chunk=not self._yielded_first_chunk,
)
def _get_completed_response_object(self) -> Any | None:
def _get_completed_response_object(self) -> ResponsesAPIResponse | None:
openai_types = _get_openai_response_types()
completed_response = self.completed_response
if isinstance(completed_response, openai_types.ResponsesAPIResponse):
@ -577,7 +611,7 @@ class BaseResponsesAPIStreamingIterator:
if self.completed_response is None:
return
request_payload: dict[str, Any] = {}
request_payload: dict[str, object] = {} # mutable-ok: populated incrementally from the request payload below
if isinstance(self.request_data, dict):
request_payload.update(self.request_data)
try:
@ -707,7 +741,7 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
def __aiter__(self):
return self
async def __anext__(self) -> Any:
async def __anext__(self) -> ResponsesAPIStreamingResponse:
try:
self._check_max_streaming_duration()
while True:
@ -880,7 +914,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
def _set_events_from_response(
self,
transformed: Any,
transformed: ResponsesAPIResponse,
logging_obj: LiteLLMLoggingObj,
) -> None:
self._events = _build_synthetic_response_events(
@ -894,7 +928,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
def __aiter__(self):
return self
async def __anext__(self) -> Any:
async def __anext__(self) -> ResponsesAPIStreamingResponse:
if self._idx >= len(self._events):
raise StopAsyncIteration
evt = self._events[self._idx]
@ -908,7 +942,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
def __iter__(self):
return self
def __next__(self) -> Any:
def __next__(self) -> ResponsesAPIStreamingResponse:
if self._idx >= len(self._events):
raise StopIteration
evt = self._events[self._idx]
@ -923,7 +957,7 @@ class MockResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
def __init__(
self,
response: Any,
response: ResponsesAPIResponse,
logging_obj: LiteLLMLoggingObj,
request_data: dict[str, Any] | None = None,
call_type: str | None = None,
@ -941,13 +975,15 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
)
self._completed_response_cache_hit = True
self._persist_completed_response_before_logging = False
self._events: list[Any] = []
self._events: list[
ResponsesAPIStreamingResponse
] = [] # mutable-ok: accumulates one event per stream update over the object's lifetime
self._idx = 0
self._set_events_from_response(transformed=response, logging_obj=logging_obj)
def _set_events_from_response(
self,
transformed: Any,
transformed: ResponsesAPIResponse,
logging_obj: LiteLLMLoggingObj,
) -> None:
self._events = _build_synthetic_response_events(
@ -961,7 +997,7 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
def __aiter__(self):
return self
async def __anext__(self) -> Any:
async def __anext__(self) -> ResponsesAPIStreamingResponse:
if self._idx >= len(self._events):
raise StopAsyncIteration
evt = self._events[self._idx]
@ -975,7 +1011,7 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
def __iter__(self):
return self
def __next__(self) -> Any:
def __next__(self) -> ResponsesAPIStreamingResponse:
if self._idx >= len(self._events):
raise StopIteration
evt = self._events[self._idx]
@ -987,8 +1023,10 @@ class CachedResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
return evt
def _dump_response_object(obj: Any) -> dict[str, Any]:
if hasattr(obj, "model_dump"):
def _dump_response_object(
obj: object,
) -> dict[str, Any]: # mutable-ok: returns a plain mutable dict to match model_dump()'s contract
if isinstance(obj, _ModelDumpable):
return obj.model_dump()
if isinstance(obj, dict):
return obj
@ -1000,8 +1038,8 @@ def _build_response_status_event(
"response.created",
"response.in_progress",
],
transformed: Any,
) -> Any:
transformed: ResponsesAPIResponse,
) -> ResponsesAPIStreamingResponse:
openai_types = _get_openai_response_types()
in_progress_response = transformed.model_copy(
deep=True,
@ -1018,10 +1056,10 @@ def _build_content_part_done_event(
output_index: int,
content_index: int,
part_payload: dict[str, Any],
) -> Any | None:
) -> ContentPartDoneEvent | None:
openai_types = _get_openai_response_types()
part_type = part_payload.get("type")
part: Any
part: ContentPartDonePartOutputText | ContentPartDonePartRefusal | ContentPartDonePartReasoningText
if part_type == "output_text":
annotations = [
openai_types.BaseLiteLLMOpenAIResponseObject(**annotation)
@ -1057,7 +1095,7 @@ def _build_content_part_done_event(
def _add_text_like_part_events(
*,
events: list[Any],
events: list[ResponsesAPIStreamingResponse], # mutable-ok: parameter mirrors the caller's mutable event list
item_id: str,
output_index: int,
content_index: int,
@ -1123,13 +1161,15 @@ def _add_text_like_part_events(
def _build_synthetic_response_events(
*,
transformed: Any,
transformed: ResponsesAPIResponse,
logging_obj: LiteLLMLoggingObj,
chunk_size: int,
) -> list[Any]:
) -> list[
ResponsesAPIStreamingResponse
]: # mutable-ok: returns a plain mutable event list to match the caller's accumulator
openai_types = _get_openai_response_types()
if litellm.include_cost_in_streaming_usage and logging_obj is not None:
usage_obj: Any | None = getattr(transformed, "usage", None)
usage_obj = transformed.usage
if usage_obj is not None:
try:
cost: float | None = logging_obj._response_cost_calculator(result=transformed)
@ -1138,13 +1178,15 @@ def _build_synthetic_response_events(
except Exception:
pass
events: list[Any] = [
events: list[
ResponsesAPIStreamingResponse
] = [ # mutable-ok: built once as a mutable event list for the caller to extend
_build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed),
_build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, transformed),
]
sequence_number = 0
for output_index, output_item in enumerate(getattr(transformed, "output", []) or []):
for output_index, output_item in enumerate(transformed.output or []):
output_item_payload = _dump_response_object(output_item)
item_id = str(output_item_payload.get("id") or transformed.id)
item_type = output_item_payload.get("type")
@ -1292,8 +1334,8 @@ class ResponsesWebSocketStreaming:
def __init__(
self,
websocket: Any,
backend_ws: Any,
websocket: WebSocket,
backend_ws: _BackendWebSocketLike,
logging_obj: LiteLLMLoggingObj,
user_api_key_dict: Any | None = None,
request_data: dict | None = None,
@ -1319,12 +1361,16 @@ class ResponsesWebSocketStreaming:
def _should_store_event(self, event_obj: dict) -> bool:
return event_obj.get("type") in RESPONSES_WS_LOGGED_EVENT_TYPES
def _store_event(self, event: Any) -> None:
def _store_event(
self, event: str | bytes | dict[str, object]
) -> None: # mutable-ok: parameter mirrors the caller's event payload, which may be a mutable dict
if isinstance(event, bytes):
event = event.decode("utf-8")
if isinstance(event, str):
try:
event_obj = json.loads(event)
event_obj: dict[str, object] = json.loads(
event
) # mutable-ok: parsed once via json.loads(), which returns a plain mutable dict
except (json.JSONDecodeError, TypeError):
return
else:
@ -1333,15 +1379,17 @@ class ResponsesWebSocketStreaming:
if self._should_store_event(event_obj):
self.messages.append(event_obj)
def _collect_input_from_client_event(self, message: Any) -> None:
def _collect_input_from_client_event(
self, message: str | dict[str, object]
) -> None: # mutable-ok: parameter mirrors the caller's message payload, which may be a mutable dict
"""Extract user input content from response.create for logging."""
try:
if isinstance(message, str):
msg_obj = json.loads(message)
elif isinstance(message, dict):
msg_obj = message
msg_obj: dict[str, object] = json.loads(
message
) # mutable-ok: parsed once via json.loads(), which returns a plain mutable dict
else:
return
msg_obj = message
if msg_obj.get("type") != "response.create":
return
@ -1368,7 +1416,9 @@ class ResponsesWebSocketStreaming:
except (json.JSONDecodeError, AttributeError, TypeError):
pass
def _store_input(self, message: Any) -> None:
def _store_input(
self, message: str | dict[str, object]
) -> None: # mutable-ok: parameter mirrors the caller's message payload, which may be a mutable dict
self._collect_input_from_client_event(message)
if self.logging_obj:
self.logging_obj.pre_call(input=message, api_key="")
@ -1407,7 +1457,10 @@ class ResponsesWebSocketStreaming:
# masked response.completed.
if self.output_guardrail_callbacks:
try:
_evt_type = json.loads(response_str).get("type")
_evt_parsed: dict[str, object] = json.loads(
response_str
) # mutable-ok: parsed once via json.loads(), which returns a plain mutable dict
_evt_type = _evt_parsed.get("type")
except (json.JSONDecodeError, TypeError):
_evt_type = None
if _evt_type in self._DELTA_EVENT_TYPES or _evt_type in self._OUTPUT_DONE_EVENT_TYPES:
@ -1793,7 +1846,7 @@ class ManagedResponsesWebSocketHandler:
def __init__(
self,
websocket: Any,
websocket: WebSocket,
model: str,
logging_obj: LiteLLMLoggingObj,
user_api_key_dict: Any | None = None,
@ -1832,12 +1885,12 @@ class ManagedResponsesWebSocketHandler:
# ------------------------------------------------------------------
@staticmethod
def _serialize_chunk(chunk: Any) -> str | None:
def _serialize_chunk(chunk: object) -> str | None:
"""Serialize a streaming chunk to a JSON string for WebSocket transmission."""
try:
if hasattr(chunk, "model_dump_json"):
if isinstance(chunk, _ModelDumpJsonable):
return chunk.model_dump_json(exclude_none=True)
if hasattr(chunk, "model_dump"):
if isinstance(chunk, _ModelDumpable):
return json.dumps(chunk.model_dump(exclude_none=True), default=str)
if isinstance(chunk, dict):
return json.dumps(chunk, default=str)
@ -1925,7 +1978,9 @@ class ManagedResponsesWebSocketHandler:
return messages
@staticmethod
def _input_to_messages(input_val: Any) -> list[dict[str, Any]]:
def _input_to_messages(
input_val: object,
) -> list[dict[str, Any]]: # mutable-ok: returns a plain mutable list to match callers that append further messages
"""
Normalise the ``input`` field of a ``response.create`` event to a list
of Responses API message dicts.

37
litellm/router.py
View file

@ -249,7 +249,7 @@ if TYPE_CHECKING:
ResponsesAPIResponse,
)
Span = Union[_Span, Any]
Span = _Span
else:
Span = Any
AutoRouter = Any
@ -463,6 +463,13 @@ class Router:
```
"""
self._override_selectors: dict[
str, Any
] = {} # mutable-ok: populated on demand and mutated by the selector-registration helpers below
self._group_selectors: dict[
str, dict[str, Any]
] = {} # mutable-ok: populated on demand and mutated by the selector-registration helpers below
self.set_verbose = set_verbose
self.ignore_invalid_deployments = ignore_invalid_deployments
self.debug_level = debug_level
@ -679,7 +686,7 @@ class Router:
routing_strategy_args=routing_strategy_args,
)
self._init_routing_groups(self._routing_groups_input)
self._override_selectors: dict[str, Any] = {}
self._override_selectors = {}
self._override_selectors_lock = threading.Lock()
self.access_groups = None
## USAGE TRACKING ##
@ -949,7 +956,7 @@ class Router:
self._unregister_router_selectors(
[getattr(self, attr, None) for attr in self._DEFAULT_SELECTOR_ATTR_BY_STRATEGY.values()]
+ list(getattr(self, "_override_selectors", {}).values())
+ list(self._override_selectors.values())
)
self._override_selectors = {}
@ -985,12 +992,12 @@ class Router:
attributes set up in `routing_strategy_init`.
"""
self._unregister_router_selectors(
[sel for selectors in getattr(self, "_group_selectors", {}).values() for sel in selectors.values()]
[sel for selectors in self._group_selectors.values() for sel in selectors.values()]
)
self._routing_groups: dict[str, RoutingGroup] = {}
self._model_to_group: dict[str, str] = {}
self._group_selectors: dict[str, dict[str, Any]] = {}
self._group_selectors = {}
if not groups_input:
return
@ -1910,7 +1917,7 @@ class Router:
async def _run_silent_completion():
await self.acompletion(
model=silent_model,
messages=cast(list[AllMessageValues], messages),
messages=messages,
**silent_kwargs,
)
# Drain any fire-and-forget tasks (e.g. alerting hooks)
@ -2753,7 +2760,7 @@ class Router:
# Trigger the silent request
await self.acompletion(
model=silent_model,
messages=cast(list[AllMessageValues], messages),
messages=messages,
**silent_kwargs,
)
except Exception as e:
@ -3276,7 +3283,11 @@ class Router:
)
)
responses = await asyncio.gather(*_tasks)
final_responses: list[list[Any]] = [[] for _ in range(len(messages))]
final_responses: list[
list[ModelResponse | CustomStreamWrapper | Exception]
] = [ # mutable-ok: populated incrementally by the thread pool loop below
[] for _ in range(len(messages))
]
for response in responses:
if isinstance(response, tuple):
final_responses[response[1]].append(response[0])
@ -4583,7 +4594,7 @@ class Router:
# fallback to the original reference for any non-picklable value.
# The original_generic_function is preserved so the per-attempt
# helper knows which underlying API to call on fallback.
fallback_kwargs: dict[str, Any] = kwargs.copy()
fallback_kwargs = kwargs.copy()
if isinstance(fallback_kwargs.get("litellm_metadata"), dict):
fallback_kwargs["litellm_metadata"] = safe_deep_copy(fallback_kwargs["litellm_metadata"])
if isinstance(fallback_kwargs.get("metadata"), dict):
@ -7079,7 +7090,7 @@ class Router:
except Exception as e:
raise e
async def async_deployment_callback_on_failure(self, kwargs, completion_response: Any | None, start_time, end_time):
async def async_deployment_callback_on_failure(self, kwargs, completion_response, start_time, end_time):
"""
Update RPM usage for a deployment
"""
@ -10999,7 +11010,7 @@ class Router:
self,
model: str,
request_kwargs: dict,
messages: list[dict[str, Any]] | None,
messages: list[dict[str, str]] | None, # mutable-ok: parameter mirrors the caller's message-dict payload
) -> RoutingContext:
"""
Build a RoutingContext for `model`, run it through `self.routing_plugins`
@ -11098,7 +11109,7 @@ class Router:
self,
model: str,
request_kwargs: dict,
messages: list[dict[str, Any]] | None = None,
messages: list[dict[str, str]] | None = None, # mutable-ok: parameter mirrors the caller's message-dict payload
input: str | list | None = None,
specific_deployment: bool | None = False,
) -> PreRoutingHookResponse | None:
@ -11181,7 +11192,7 @@ class Router:
@staticmethod
def _redact_prompt_text_if_needed(
request_kwargs: Mapping[str, Any],
request_kwargs: Mapping[str, object],
routing_decision: StandardLoggingRoutingDecision,
) -> StandardLoggingRoutingDecision:
"""Drop verbatim prompt text from the record when message logging is redacted.

12
ruff-strict-budget.json
View file

@ -9,10 +9,10 @@
"limit": 831
},
"ANN201": {
"limit": 2137
"limit": 2135
},
"ANN202": {
"limit": 941
"limit": 939
},
"ANN204": {
"limit": 724
@ -24,7 +24,7 @@
"limit": 130
},
"ANN401": {
"limit": 1848
"limit": 1756
},
"ASYNC230": {
"limit": 14
@ -135,7 +135,7 @@
"limit": 30
},
"PERF401": {
"limit": 142
"limit": 138
},
"PERF402": {
"limit": 9
@ -222,7 +222,7 @@
"limit": 0
},
"RET504": {
"limit": 702
"limit": 700
},
"RUF010": {
"limit": 0
@ -321,7 +321,7 @@
"limit": 121
},
"TRY300": {
"limit": 879
"limit": 877
},
"UP006": {
"limit": 0

6
type-discipline-budget.json
View file

@ -1,9 +1,9 @@
{
"LIT001": {
"limit": 23349
"limit": 23301
},
"LIT002": {
"limit": 27252
"limit": 27199
},
"LIT003": {
"limit": 292
@ -15,7 +15,7 @@
"limit": 0
},
"LIT006": {
"limit": 1105
"limit": 1095
},
"LIT007": {
"limit": 0