Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/common_request_processing.py: 34%
1514 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import asyncio
2import contextlib
3import json
4import logging
5import math
6from collections.abc import AsyncGenerator, Awaitable, Callable, Mapping, Sequence
7from datetime import datetime
8from functools import lru_cache
9from types import MappingProxyType
10from typing import (
11 TYPE_CHECKING,
12 Any,
13 Final,
14 Literal,
15 NamedTuple,
16 Protocol,
17 TypeAlias,
18 TypeVar,
19 overload,
20 runtime_checkable,
21)
22from urllib.parse import urlparse
24import anyio
25import httpx
26import orjson
27from fastapi import HTTPException, Request, status
28from fastapi.responses import JSONResponse, Response, StreamingResponse
29from pydantic import BaseModel, TypeAdapter, ValidationError
30from starlette.types import Receive, Scope, Send
32import litellm
33from litellm._logging import redact_internal_details_from_client_message, verbose_proxy_logger
34from litellm._uuid import uuid
35from litellm.constants import (
36 DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE,
37 DEFAULT_MAX_RECURSE_DEPTH,
38 LITELLM_DETAILED_TIMING,
39 LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED,
40 MAX_LITELLM_CALL_ID_LENGTH,
41 MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG,
42 NON_INFERENCE_CALL_TYPES,
43 RETURN_RAW_MODEL_NAME_METADATA_KEY,
44 STREAM_SSE_DATA_PREFIX,
45 STREAM_SSE_KEEPALIVE_PING_BYTES,
46 UNSAFE_PROXY_RESPONSE_HEADERS,
47)
48from litellm.integrations.custom_guardrail import CustomGuardrail
49from litellm.litellm_core_utils.bug_report import (
50 allowlisted,
51 bug_report_notice,
52 should_report_bug,
53 strip_bug_report_notice,
54)
55from litellm.litellm_core_utils.core_helpers import (
56 get_or_create_metadata_bucket,
57 independent_snapshot,
58 is_expected_client_error,
59 redact_nested_match_and_regex_keys,
60)
61from litellm.litellm_core_utils.dd_tracing import NullTracer, tracer
62from litellm.litellm_core_utils.get_supported_openai_params import (
63 get_supported_openai_params,
64)
65from litellm.litellm_core_utils.internal_call_metadata import is_unbilled_non_inference_call_from_params
66from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
67from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import guardrail_information_cost
68from litellm.litellm_core_utils.llm_response_utils.get_headers import (
69 get_response_headers,
70)
71from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
72from litellm.litellm_core_utils.served_output_texts import (
73 record_served_output_texts,
74 served_output_texts,
75)
76from litellm.litellm_core_utils.streaming_handler import (
77 backfill_missing_cache_usage_fields,
78)
79from litellm.proxy._types import LiteLLMRoutes, ProxyErrorTypes, ProxyException, UserAPIKeyAuth
80from litellm.proxy.auth.auth_checks import (
81 can_key_call_resolved_model,
82 request_skips_budget_checks,
83 tag_max_budget_check_for_tags,
84)
85from litellm.proxy.auth.auth_utils import check_response_size_is_safe, get_request_route
86from litellm.proxy.bug_report_config import build_proxy_bug_report
87from litellm.proxy.common_utils.callback_utils import (
88 get_logging_caching_headers,
89 get_remaining_tokens_and_requests_from_request_data,
90)
91from litellm.proxy.common_utils.error_body_call_id import JSON_OBJECT, error_body_call_id, with_call_id
92from litellm.proxy.common_utils.http_parsing_utils import (
93 get_client_requested_model,
94 get_tags_from_request_body,
95)
96from litellm.proxy.common_utils.openai_error_payload import (
97 LITELLM_CALL_ID_HEADER,
98 attribute_of,
99 error_status_code,
100 openai_error_param,
101 openai_error_type,
102)
103from litellm.proxy.common_utils.sse_keepalive import (
104 SSE_COMMENT_PING_BYTES,
105 coerce_keepalive_interval,
106 resolve_ttft_keepalive_interval,
107 wrap_sse_stream_with_keepalive_pings,
108)
109from litellm.proxy.dd_span_tagger import DDSpanTagger
110from litellm.proxy.guardrails.auto_router_compression import arm_pre_call as _arm_auto_router_compression
111from litellm.proxy.native_compaction import with_proxy_compaction_executor
112from litellm.proxy.route_llm_request import route_request
113from litellm.proxy.utils import ProxyLogging, _check_and_merge_model_level_guardrails
114from litellm.router import Router
115from litellm.router_utils.add_retry_fallback_headers import get_hidden_params_dict
116from litellm.router_utils.common_utils import resolve_model_group_alias
117from litellm.types.guardrails import GuardrailEventHooks
118from litellm.types.router import RouterRateLimitError
119from litellm.types.router_weights import validate_router_weights
121_LateResponseT = TypeVar("_LateResponseT", bound=Response)
122_LlmCallT = TypeVar("_LlmCallT")
124KNOWN_PROXY_ROUTES: Final = frozenset(
125 route for member in LiteLLMRoutes for route in member.value if route.startswith("/")
126)
128ProxyRouteType: TypeAlias = Literal[
129 "acompletion",
130 "aembedding",
131 "aresponses",
132 "_arealtime",
133 "_aresponses_websocket",
134 "acreate_realtime_client_secret",
135 "arealtime_calls",
136 "aget_responses",
137 "adelete_responses",
138 "acancel_responses",
139 "acompact_responses",
140 "acreate_batch",
141 "aretrieve_batch",
142 "alist_batches",
143 "acancel_batch",
144 "afile_content",
145 "afile_retrieve",
146 "afile_delete",
147 "atext_completion",
148 "acreate_fine_tuning_job",
149 "acancel_fine_tuning_job",
150 "alist_fine_tuning_jobs",
151 "aretrieve_fine_tuning_job",
152 "alist_input_items",
153 "aimage_edit",
154 "agenerate_content",
155 "agenerate_content_stream",
156 "allm_passthrough_route",
157 "avector_store_search",
158 "avector_store_create",
159 "avector_store_retrieve",
160 "avector_store_list",
161 "avector_store_update",
162 "avector_store_delete",
163 "avector_store_file_create",
164 "avector_store_file_list",
165 "avector_store_file_retrieve",
166 "avector_store_file_content",
167 "avector_store_file_update",
168 "avector_store_file_delete",
169 "aocr",
170 "asearch",
171 "adecisions",
172 "avideo_generation",
173 "avideo_list",
174 "avideo_status",
175 "avideo_content",
176 "avideo_remix",
177 "avideo_create_character",
178 "avideo_get_character",
179 "avideo_edit",
180 "avideo_extension",
181 "acreate_container",
182 "alist_containers",
183 "aingest",
184 "aretrieve_container",
185 "adelete_container",
186 "aupload_container_file",
187 "alist_container_files",
188 "aretrieve_container_file",
189 "adelete_container_file",
190 "aretrieve_container_file_content",
191 "acreate_skill",
192 "alist_skills",
193 "aget_skill",
194 "adelete_skill",
195 "anthropic_messages",
196 "acreate_interaction",
197 "aget_interaction",
198 "adelete_interaction",
199 "acancel_interaction",
200 "acreate_agent",
201 "alist_agents",
202 "aget_agent",
203 "adelete_agent",
204 "alist_agent_versions",
205 "asend_message",
206 "call_mcp_tool",
207 "acreate_eval",
208 "alist_evals",
209 "aget_eval",
210 "aupdate_eval",
211 "adelete_eval",
212 "acancel_eval",
213 "acreate_run",
214 "alist_runs",
215 "aget_run",
216 "acancel_run",
217 "adelete_run",
218]
219from litellm.llms.anthropic.chat.transformation import AnthropicConfig
221# Type alias for streaming chunk serializer (chunk after hooks + cost injection -> wire format)
222StreamChunkSerializer = Callable[[object], str]
223# Type alias for streaming error serializer (ProxyException -> wire format)
224StreamErrorSerializer = Callable[[ProxyException], str]
226if TYPE_CHECKING: 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true
227 from litellm.proxy.proxy_server import ProxyConfig as _ProxyConfig
229 ProxyConfig = _ProxyConfig
230else:
231 ProxyConfig = Any
232from litellm.proxy.anthropic_endpoints.streaming_model_restamp import (
233 AnthropicStreamModelRestamper,
234)
235from litellm.proxy.litellm_pre_call_utils import (
236 add_litellm_data_to_request,
237 refresh_proxy_server_request_body_snapshot,
238 reject_url_valued_destination,
239)
240from litellm.proxy.policy_engine.response_retrieval import attach_post_call_pipelines_to_retrieval
241from litellm.types.utils import (
242 ModelResponse,
243 ModelResponseStream,
244 StandardLoggingPayloadErrorInformation,
245 Usage,
246)
248# Datadog streaming spans are a no-op when ddtrace is not enabled, but the
249# ``with tracer.trace(...)`` context manager still allocates a NullSpan and
250# runs __enter__/__exit__ for every streamed chunk. Resolve once at import so
251# the per-chunk hot path can skip the context manager entirely when tracing
252# is off (the default).
253_DD_STREAMING_TRACE_ENABLED: Final = not isinstance(tracer, NullTracer)
256_CLIENT_DISCONNECTED_ERROR_INFORMATION: Final[StandardLoggingPayloadErrorInformation] = {
257 "error_code": str(LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED),
258 "error_message": "Client disconnected the request",
259 "error_class": "ClientDisconnected",
260}
263def _withheld_provider_output(response: object) -> bool:
264 return getattr(response, "has_buffered_provider_output", False) is True
267def resolve_litellm_call_id(client_call_id: str | None) -> str:
268 if client_call_id is not None and 0 < len(client_call_id) <= MAX_LITELLM_CALL_ID_LENGTH: 268 ↛ 269line 268 didn't jump to line 269 because the condition on line 268 was never true
269 return client_call_id
270 return str(uuid.uuid4())
273def _should_return_raw_model_name(request_data: dict[str, object]) -> bool:
274 return any(
275 isinstance(metadata, dict) and metadata.get(RETURN_RAW_MODEL_NAME_METADATA_KEY) is True
276 for metadata in (request_data.get("metadata"), request_data.get("litellm_metadata"))
277 )
280def _apply_client_disconnect_metadata(target_metadata: dict[str, object] | None) -> None:
281 if target_metadata is None:
282 return
283 target_metadata["client_disconnected"] = True
284 target_metadata["error_information"] = dict(_CLIENT_DISCONNECTED_ERROR_INFORMATION)
287async def _record_streaming_client_disconnect_if_needed(
288 request: Request | None,
289 request_data: dict,
290 client_disconnected: bool = False,
291) -> bool:
292 if not client_disconnected:
293 if request is None:
294 return False
295 try:
296 disconnected: Final = await request.is_disconnected()
297 except Exception: # noqa: BLE001
298 return False
299 if not disconnected:
300 return False
302 logging_obj: Final = request_data.get("litellm_logging_obj")
303 if logging_obj is not None:
304 litellm_params: Final = logging_obj.model_call_details.setdefault("litellm_params", {})
305 _lp_metadata = litellm_params.get("metadata")
306 if _lp_metadata is None:
307 _lp_metadata = {}
308 litellm_params["metadata"] = _lp_metadata
309 _apply_client_disconnect_metadata(_lp_metadata)
311 _mcd_metadata = logging_obj.model_call_details.get("metadata")
312 if _mcd_metadata is None:
313 _mcd_metadata = {}
314 logging_obj.model_call_details["metadata"] = _mcd_metadata
315 _apply_client_disconnect_metadata(_mcd_metadata)
317 _rd_metadata = request_data.get("metadata")
318 if _rd_metadata is None:
319 _rd_metadata = {}
320 request_data["metadata"] = _rd_metadata
321 _apply_client_disconnect_metadata(_rd_metadata)
323 _rd_litellm_params = request_data.get("litellm_params")
324 if _rd_litellm_params is None:
325 _rd_litellm_params = {}
326 request_data["litellm_params"] = _rd_litellm_params
327 _rd_lp_metadata = _rd_litellm_params.get("metadata")
328 if _rd_lp_metadata is None:
329 _rd_lp_metadata = {}
330 _rd_litellm_params["metadata"] = _rd_lp_metadata
331 _apply_client_disconnect_metadata(_rd_lp_metadata)
333 verbose_proxy_logger.debug(
334 "Recorded streaming client disconnect with error_code=499 for litellm_call_id=%s",
335 request_data.get("litellm_call_id"),
336 )
337 return True
340def _deferred_stream_logging_is_armed(request_data: dict) -> bool:
341 logging_obj: Final = request_data.get("litellm_logging_obj")
342 if logging_obj is None:
343 return False
344 return (
345 getattr(logging_obj, "_on_deferred_stream_complete", None) is not None
346 and getattr(logging_obj, "_deferred_stream_complete_args", None) is not None
347 )
350def _assembled_model_came_from_a_later_chunk(chunks: Sequence[object], assembled_model: object) -> bool:
351 """Report whether stream_chunk_builder picked a model the first chunk did not carry.
353 Azure Model Router puts the routed model on the chunks after the first one, and the
354 proxy deliberately leaves those chunks unrestamped so the builder can recover it.
356 A stored chunk that carries usage is a pre-restamp copy of the one the proxy saw, so
357 an alias-restamped stream reaches the builder with the same shape: a first chunk that
358 disagrees with the rest. Those two are only told apart by what the client asked for.
359 """
360 first_chunk: Final = chunks[0]
361 first_chunk_model: Final = (
362 first_chunk.get("model") if isinstance(first_chunk, dict) else getattr(first_chunk, "model", None)
363 )
364 return (
365 isinstance(first_chunk_model, str)
366 and isinstance(assembled_model, str)
367 and bool(assembled_model)
368 and assembled_model != first_chunk_model
369 )
372def _assembled_model_is_the_name_the_client_asked_for(
373 request_data: Mapping[str, object],
374 assembled_model: object,
375) -> bool:
376 """Report whether the assembled model is the public name the proxy stamps onto chunks.
378 That stamp is what leaves an unpriced alias on the partial response, so the deployment's
379 own model has to go back on before the row is costed. Pre-call processing rewrites
380 `request_data["model"]` for aliasing and routing, so the client's own name wins when it
381 is there, in the same order the proxy picks the name it stamps.
382 """
383 client_requested_model: Final = request_data.get("_litellm_client_requested_model")
384 stamped_model: Final = (
385 client_requested_model if isinstance(client_requested_model, str) else request_data.get("model")
386 )
387 return isinstance(stamped_model, str) and assembled_model == stamped_model
390async def _bill_partial_streamed_spend_on_disconnect(request_data: dict, response: object) -> bool:
391 """
392 A client disconnect throws GeneratorExit/CancelledError into the streaming
393 generator, so neither the success nor the failure logging callback fires
394 and the chunks already streamed (plus any sub-call cost folded into the
395 logging object) would never reach spend tracking. Assemble the partial
396 response from the wrapper's collected chunks and dispatch success logging
397 for it; dispatch_success_handlers dedups against a natural end-of-stream
398 dispatch via has_dispatched_final_stream_success.
400 Awaited directly by the shielded cleanup rather than scheduled with
401 create_task: the client is already gone so the extra latency is harmless,
402 and an unrooted task could be garbage-collected before it bills.
404 Returns True when a disconnect-time success event owns the request's
405 max_parallel_requests slot release (one was dispatched here, or one had
406 already been dispatched for this stream), so the caller can skip the
407 explicit slot release and avoid a double release. Returns False when no
408 success event fired (logging disabled, nothing streamed, or assembly
409 failed) and the caller must release the slot itself.
410 """
411 if litellm.disable_streaming_logging is True:
412 return False
413 logging_obj: Final = request_data.get("litellm_logging_obj")
414 if not isinstance(logging_obj, LiteLLMLoggingObj):
415 return False
416 if logging_obj.model_call_details.get("has_dispatched_final_stream_success"):
417 # A natural end-of-stream success event already fired and released the
418 # slot; do not bill again, and let the caller skip the slot release.
419 return True
420 chunks: Final[object] = getattr(response, "chunks", None)
421 if not isinstance(chunks, list) or not chunks:
422 return False
423 verbose_proxy_logger.debug(
424 "Billing partial streamed spend for %s chunks after client disconnect, litellm_call_id=%s",
425 len(chunks),
426 request_data.get("litellm_call_id"),
427 )
428 messages: Final[object] = getattr(response, "messages", None)
429 try:
430 partial_response: Final = litellm.stream_chunk_builder(
431 chunks=chunks,
432 messages=messages if isinstance(messages, list) else None,
433 logging_obj=logging_obj,
434 )
435 except Exception as e: # noqa: BLE001 # partial billing is best-effort; never break stream teardown
436 verbose_proxy_logger.debug("Failed to assemble partial streamed response for disconnect billing: %s", e)
437 return False
438 if partial_response is None:
439 return False
440 wrapper_model: Final = getattr(response, "model", None)
441 builder_recovered_the_routed_model: Final = _assembled_model_came_from_a_later_chunk(
442 chunks, partial_response.model
443 ) and not _assembled_model_is_the_name_the_client_asked_for(request_data, partial_response.model)
444 if isinstance(wrapper_model, str) and wrapper_model and not builder_recovered_the_routed_model:
445 partial_response.model = wrapper_model
446 partial_usage: Final = getattr(partial_response, "usage", None)
447 if isinstance(partial_usage, Usage):
448 backfill_missing_cache_usage_fields(partial_usage)
449 try:
450 await logging_obj.dispatch_success_handlers(
451 partial_response,
452 cache_hit=False,
453 start_time=None,
454 end_time=None,
455 prefer_async_handlers=True,
456 )
457 except Exception as e: # noqa: BLE001 # partial billing is best-effort; never break stream teardown
458 verbose_proxy_logger.debug("Failed to dispatch disconnect billing event: %s", e)
459 return False
460 return True
463async def _cancel_pending_gather_tasks(tasks: Sequence["asyncio.Task[object]"]) -> None:
464 pending_tasks: Final = [task for task in tasks if not task.done()]
465 for task in pending_tasks: 465 ↛ 466line 465 didn't jump to line 466 because the loop on line 465 never started
466 task.cancel()
467 for task in pending_tasks: 467 ↛ 468line 467 didn't jump to line 468 because the loop on line 467 never started
468 try:
469 await task
470 except (asyncio.CancelledError, Exception): # noqa: BLE001
471 pass
474@lru_cache(maxsize=512)
475def _litellm_model_supports_stream_options(litellm_model: str) -> bool:
476 try:
477 supported_params: Final = get_supported_openai_params(model=litellm_model)
478 except Exception: # noqa: BLE001 # unmapped or malformed model strings must disable injection, not fail the request
479 return False
480 return supported_params is not None and "stream_options" in supported_params
483def _model_deployments_support_stream_options(
484 model: object,
485 llm_router: Router | None,
486 team_id: str | None,
487) -> bool:
488 if not isinstance(model, str): 488 ↛ 489line 488 didn't jump to line 489 because the condition on line 488 was never true
489 return False
490 deployment_models: Final = (
491 llm_router.resolved_litellm_models(model, team_id=team_id) if llm_router is not None else ()
492 )
493 candidate_models: Final = deployment_models if deployment_models else (model,)
494 return all(_litellm_model_supports_stream_options(m) for m in candidate_models)
497def _stream_usage_tracking_updates(
498 data: Mapping[str, object],
499 general_settings: Mapping[str, object],
500 route_type: str,
501 supports_stream_options: Callable[[], bool],
502) -> Mapping[str, object]:
503 scrub: Final = {"_litellm_strip_stream_usage": False} if "_litellm_strip_stream_usage" in data else {}
504 if data.get("stream", False) is not True:
505 return scrub
506 always_include: Final = general_settings.get("always_include_stream_usage")
507 stream_options: Final = data.get("stream_options")
508 if always_include is True: 508 ↛ 509line 508 didn't jump to line 509 because the condition on line 508 was never true
509 if "stream_options" not in data:
510 return {**scrub, "stream_options": {"include_usage": True}}
511 if isinstance(stream_options, dict) and "include_usage" not in stream_options:
512 return {**scrub, "stream_options": {**stream_options, "include_usage": True}}
513 return scrub
514 if always_include is False or route_type != "acompletion":
515 return scrub
516 if isinstance(stream_options, dict) and stream_options.get("include_usage") is True: 516 ↛ 517line 516 didn't jump to line 517 because the condition on line 516 was never true
517 return scrub
518 if not supports_stream_options(): 518 ↛ 520line 518 didn't jump to line 520 because the condition on line 518 was always true
519 return scrub
520 merged_stream_options: Final = {**stream_options} if isinstance(stream_options, dict) else {}
521 return {
522 "stream_options": {**merged_stream_options, "include_usage": True},
523 "_litellm_strip_stream_usage": True,
524 }
527class _UpstreamHttpResponse(Protocol):
528 @property
529 def status_code(self) -> int: ... 529 ↛ exitline 529 didn't return from function 'status_code' because
531 @property
532 def headers(self) -> httpx.Headers: ... 532 ↛ exitline 532 didn't return from function 'headers' because
534 async def aread(self) -> bytes: ... 534 ↛ exitline 534 didn't return from function 'aread' because
537def _as_upstream_response(response: _UpstreamHttpResponse) -> _UpstreamHttpResponse:
538 return response
541class _ReadsHeaderValues(Protocol):
542 def get(self, key: str, default: str = "") -> str: ... 542 ↛ exitline 542 didn't return from function 'get' because
545def _as_header_reader(headers: _ReadsHeaderValues) -> _ReadsHeaderValues:
546 return headers
549class _DispatchesSuccessHandlers(Protocol):
550 async def dispatch_success_handlers( 550 ↛ exitline 550 didn't return from function 'dispatch_success_handlers' because
551 self,
552 result: object = None,
553 start_time: object = None,
554 end_time: object = None,
555 cache_hit: object = None,
556 prefer_async_handlers: bool = False,
557 ) -> None: ...
560def _as_success_dispatcher(logging_obj: _DispatchesSuccessHandlers) -> _DispatchesSuccessHandlers:
561 return logging_obj
564def serialize_http_exception_detail(
565 detail: object,
566) -> tuple[str, dict | None]:
567 """
568 Convert an HTTPException.detail value into (message, structured_fields)
569 for ProxyException / SSE error frames.
571 Dict-detail HTTPExceptions raised by guardrails were previously str()-mangled
572 into a Python repr blob, producing unparseable error responses on both the
573 streaming and non-streaming proxy surfaces. This helper extracts a clean
574 human-readable message while preserving the full payload as structured
575 fields, so the dominant guardrail shapes (`{"error": "..."}` flat and
576 `{"error": {"message": "..."}}` nested) both round-trip cleanly.
577 """
578 if isinstance(detail, str): 578 ↛ 579line 578 didn't jump to line 579 because the condition on line 578 was never true
579 return detail, None
580 if isinstance(detail, dict): 580 ↛ 592line 580 didn't jump to line 592 because the condition on line 580 was always true
581 err: Final = detail.get("error")
582 if isinstance(err, str): 582 ↛ 584line 582 didn't jump to line 584 because the condition on line 582 was always true
583 return err, detail
584 if isinstance(err, dict):
585 nested_msg: Final = err.get("message")
586 if isinstance(nested_msg, str):
587 return nested_msg, detail
588 msg: Final = detail.get("message")
589 if isinstance(msg, str):
590 return msg, detail
591 return json.dumps(detail), detail
592 return str(detail), None
595def proxy_exception_from_http_exception(exc: HTTPException, headers: dict[str, str]) -> ProxyException:
596 raw_detail: Final = attribute_of(exc, "detail", str(exc))
597 message, structured_fields = serialize_http_exception_detail(raw_detail)
598 existing_fields: Final = getattr(exc, "provider_specific_fields", None) or {}
599 merged_fields: Final = {**existing_fields, **structured_fields} if structured_fields else (existing_fields or None)
600 error_status: Final = error_status_code(exc, status.HTTP_400_BAD_REQUEST)
601 return ProxyException(
602 message=message,
603 type=openai_error_type(exc, error_status),
604 param=openai_error_param(exc),
605 code=error_status,
606 provider_specific_fields=merged_fields,
607 headers=headers,
608 )
611def _collect_response_file_search_vector_store_ids(data: Mapping[str, object]) -> set[str]:
612 vector_store_ids: Final[set[str]] = set()
613 tools: Final = data.get("tools")
614 if not isinstance(tools, list):
615 return vector_store_ids
617 for tool in tools: 617 ↛ 618line 617 didn't jump to line 618 because the loop on line 617 never started
618 if not isinstance(tool, dict) or tool.get("type") != "file_search":
619 continue
620 ids = tool.get("vector_store_ids") or []
621 if not isinstance(ids, list):
622 raise HTTPException(
623 status_code=400,
624 detail={"error": "file_search.vector_store_ids must be a list of strings"},
625 )
626 for vector_store_id in ids:
627 if not isinstance(vector_store_id, str) or not vector_store_id:
628 raise HTTPException(
629 status_code=400,
630 detail={"error": "file_search.vector_store_ids must be a list of strings"},
631 )
632 vector_store_ids.add(vector_store_id)
634 return vector_store_ids
637async def _authorize_response_file_search_vector_stores(
638 data: Mapping[str, object],
639 user_api_key_dict: UserAPIKeyAuth,
640) -> None:
641 vector_store_ids: Final = _collect_response_file_search_vector_store_ids(data)
642 if not vector_store_ids: 642 ↛ 645line 642 didn't jump to line 645 because the condition on line 642 was always true
643 return
645 from litellm.proxy.vector_store_endpoints.utils import (
646 assert_user_can_access_vector_store_id,
647 )
649 for vector_store_id in sorted(vector_store_ids):
650 await assert_user_can_access_vector_store_id(
651 vector_store_id=vector_store_id,
652 user_api_key_dict=user_api_key_dict,
653 )
656async def _resolve_per_request_model_group_alias(
657 requested_model: object,
658 router_settings: Mapping[str, object],
659 user_api_key_dict: UserAPIKeyAuth,
660 llm_router: Router,
661) -> str | None:
662 """
663 Resolve ``router_settings.model_group_alias`` coming from a key or team.
665 The Router only ever resolves aliases from its own instance attribute, which
666 holds the global config map and is shared across requests, so a per-request
667 map has to be applied here instead of being forwarded to the Router.
669 Auth already rewrote the body through this map for LLM API routes, so this is
670 a fallback for callers that skipped it; the target is authorized in its own
671 right before the rewrite, so a key that may not call it gets the usual 403.
673 Returns the target model group, or None when no alias applies.
674 """
675 if not isinstance(requested_model, str):
676 return None
677 target: Final = resolve_model_group_alias(router_settings.get("model_group_alias"), requested_model)
678 if target is None or target == requested_model:
679 return None
680 await can_key_call_resolved_model(
681 model=target,
682 llm_model_list=llm_router.model_list,
683 valid_token=user_api_key_dict,
684 llm_router=llm_router,
685 )
686 return target
689_REQUEST_MODEL: Final[TypeAdapter[str | list[str] | None]] = TypeAdapter(str | list[str] | None)
692def _request_model(data: Mapping[str, object]) -> str | list[str] | None:
693 try:
694 return _REQUEST_MODEL.validate_python(data.get("model"), strict=True)
695 except ValidationError:
696 return None
699async def _enforce_guardrail_added_tag_budgets(
700 data: Mapping[str, object],
701 tags_before_guardrails: frozenset[str],
702 route: str,
703 llm_router: Router | None,
704 user_api_key_dict: UserAPIKeyAuth,
705 proxy_logging_obj: ProxyLogging,
706) -> None:
707 added_tags: Final = tuple(
708 tag for tag in get_tags_from_request_body(request_body=data) if tag not in tags_before_guardrails
709 )
710 if not added_tags or request_skips_budget_checks(route=route, model=_request_model(data), llm_router=llm_router): 710 ↛ 712line 710 didn't jump to line 712 because the condition on line 710 was always true
711 return
712 from litellm.proxy.proxy_server import prisma_client, user_api_key_cache
714 try:
715 await tag_max_budget_check_for_tags(
716 tags=added_tags,
717 prisma_client=prisma_client,
718 user_api_key_cache=user_api_key_cache,
719 proxy_logging_obj=proxy_logging_obj,
720 valid_token=user_api_key_dict,
721 )
722 except litellm.BudgetExceededError as e:
723 raise ProxyException(
724 message=e.message,
725 type=ProxyErrorTypes.budget_exceeded,
726 param=None,
727 code=e.status_code,
728 ) from e
731async def _parse_event_data_for_error(event_line: str | bytes) -> int | None:
732 """Parses an event line and returns an error code if present, else None."""
733 event_line = event_line.decode("utf-8") if isinstance(event_line, bytes) else event_line
734 if event_line.startswith("data: "):
735 json_str: Final = event_line[len("data: ") :].strip()
736 if not json_str or json_str == "[DONE]": # handle empty data or [DONE] message
737 return None
738 try:
739 data: Final = orjson.loads(json_str)
740 if isinstance(data, dict) and "error" in data and isinstance(data["error"], dict):
741 error_code_raw: Final = data["error"].get("code")
742 error_code: int | None = None
744 if isinstance(error_code_raw, int):
745 error_code = error_code_raw
746 elif isinstance(error_code_raw, str):
747 try:
748 error_code = int(error_code_raw)
749 except ValueError:
750 verbose_proxy_logger.warning(
751 "Error code is a string but not a valid integer: %s", error_code_raw
752 )
753 # Not a valid integer string, treat as if no valid code was found for this check
755 # Ensure error_code is a valid HTTP status code
756 if error_code is not None and 100 <= error_code <= 599:
757 return error_code
758 elif error_code_raw is not None: # Log if original code was present but not valid
759 verbose_proxy_logger.warning("Error has invalid or non-convertible code: %s", error_code_raw)
760 except (orjson.JSONDecodeError, json.JSONDecodeError):
761 # not a known error chunk
762 pass
763 return None
766def _extract_error_from_sse_chunk(event_line: str | bytes) -> dict:
767 """
768 Extract error dictionary from SSE format chunk.
770 Args:
771 event_line: SSE format event line, e.g. "data: {"error": {...}}\n\n"
773 Returns:
774 Error dictionary in OpenAI API format
775 """
776 event_line = event_line.decode("utf-8") if isinstance(event_line, bytes) else event_line
778 # Default error format
779 default_error: Final = {
780 "message": "Unknown error",
781 "type": "internal_server_error",
782 "param": None,
783 "code": "500",
784 }
786 if event_line.startswith("data: "):
787 json_str: Final = event_line[len("data: ") :].strip()
788 if not json_str or json_str == "[DONE]":
789 return default_error
791 try:
792 data: Final = orjson.loads(json_str)
793 if isinstance(data, dict) and "error" in data:
794 error_obj: Final = data["error"]
795 if isinstance(error_obj, dict):
796 return error_obj
797 except (orjson.JSONDecodeError, json.JSONDecodeError):
798 pass
800 return default_error
803class _UpstreamClosingStreamingResponse(StreamingResponse):
804 """StreamingResponse that always closes its body iterator and the wrapped
805 upstream generator.
807 When the client disconnects mid-stream, Starlette abandons the body
808 iterator without calling aclose(), leaving the upstream LLM connection
809 open until garbage collection; the backend (e.g. vLLM) keeps generating
810 into a dead pipe. The upstream generator is closed directly (not via the
811 body iterator) because aclose() on a never-started generator skips its
812 body, so a cascade through it would be a no-op if the client disconnects
813 before the first chunk is sent.
814 """
816 def __init__(
817 self,
818 content: AsyncGenerator[str, None],
819 *,
820 media_type: str | None = None,
821 headers: Mapping[str, str] | None = None,
822 status_code: int = status.HTTP_200_OK,
823 upstream_generator: AsyncGenerator[str, None] | None = None,
824 ) -> None:
825 super().__init__(content, status_code=status_code, headers=headers, media_type=media_type)
826 self._upstream_generator = upstream_generator
828 @property
829 def upstream_generator(self) -> AsyncGenerator[str, None] | None:
830 """The upstream LLM stream, for a caller that has to run this response's cleanup itself."""
831 return self._upstream_generator
833 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
834 try:
835 await super().__call__(scope, receive, send)
836 finally:
837 with anyio.CancelScope(shield=True):
838 for target in (self.body_iterator, self._upstream_generator):
839 aclose = getattr(target, "aclose", None)
840 if aclose is None:
841 continue
842 try:
843 await aclose()
844 except BaseException as e:
845 verbose_proxy_logger.debug("error closing streaming generator: %s", e)
848class _ClientDisconnectedBeforeFirstChunk(Exception):
849 """Client went away during create_response's first-chunk buffering window.
851 The upstream LLM stream has already been closed by the time this is raised.
852 """
855async def _wait_for_http_disconnect(request: Request) -> None:
856 try:
857 while True:
858 message = await request.receive()
859 if message.get("type") == "http.disconnect":
860 return
861 except asyncio.CancelledError:
862 raise
863 except Exception as exc: # noqa: BLE001
864 verbose_proxy_logger.warning(
865 "create_response: request.receive() raised %s; first-chunk disconnect monitoring disabled for this request",
866 exc,
867 )
868 # A receive() failure must not masquerade as a disconnect.
869 await asyncio.Event().wait()
872async def _buffer_first_chunk_honoring_disconnect(
873 generator: AsyncGenerator[str, None],
874 request: Request | None,
875) -> str:
876 """Fetch the first streamed chunk, cancelling the upstream LLM call if the
877 client disconnects before it arrives.
879 create_response buffers the first chunk to detect error-only streams before
880 handing the StreamingResponse to Starlette, which only begins listening for
881 client disconnects once it is serving that response. A disconnect during a
882 long time-to-first-token would otherwise leave the upstream call running
883 until the request timeout (LIT-3568). Cancelling the fetch propagates into
884 async_streaming_data_generator, whose finally block records the 499 and
885 closes the upstream stream.
886 """
887 if request is None:
888 return await generator.__anext__()
890 chunk_task: Final[asyncio.Task[str]] = asyncio.ensure_future(generator.__anext__())
891 disconnect_task: Final[asyncio.Task[None]] = asyncio.ensure_future(_wait_for_http_disconnect(request))
892 try:
893 await asyncio.wait({chunk_task, disconnect_task}, return_when=asyncio.FIRST_COMPLETED)
894 # A completed disconnect_task has already consumed the http.disconnect
895 # message, so Starlette's later listen_for_disconnect would never see it.
896 # Take the cancellation path whenever a disconnect was observed, even if
897 # the first chunk landed in the same scheduler turn.
898 disconnect_observed: Final = disconnect_task.done()
899 finally:
900 disconnect_task.cancel()
901 try:
902 await disconnect_task
903 except BaseException: # noqa: BLE001
904 pass
906 if not disconnect_observed and chunk_task.done() and not chunk_task.cancelled():
907 return chunk_task.result()
909 chunk_task.cancel()
910 with anyio.CancelScope(shield=True):
911 try:
912 await chunk_task
913 except BaseException: # noqa: BLE001
914 pass
915 try:
916 await generator.aclose()
917 except BaseException as exc: # noqa: BLE001
918 verbose_proxy_logger.debug("create_response: error closing generator on disconnect: %s", exc)
919 verbose_proxy_logger.info("create_response: client disconnected before first chunk, upstream LLM request cancelled")
920 raise _ClientDisconnectedBeforeFirstChunk()
923def sse_error_payload(exc: BaseException) -> tuple[int, Mapping[str, object]]:
924 """Build the ProxyException-shaped ``{"error": ...}`` body used in SSE error frames.
926 Matches ``ProxyException.to_dict()`` so streaming and non-streaming error frames
927 are byte-identical.
928 """
929 # Preserve status code from HTTPException (e.g. guardrail blocks)
930 error_status: Final = error_status_code(exc, status.HTTP_500_INTERNAL_SERVER_ERROR)
931 raw_detail: Final = attribute_of(exc, "detail", "Error processing stream start")
932 message, structured_fields = serialize_http_exception_detail(raw_detail)
934 existing_fields: Final = getattr(exc, "provider_specific_fields", None) or {}
935 merged_fields: Final = {**existing_fields, **structured_fields} if structured_fields else (existing_fields or None)
937 error_obj: Final = {
938 "message": message,
939 "type": openai_error_type(exc, error_status),
940 "param": openai_error_param(exc),
941 "code": str(error_status),
942 }
943 if not merged_fields:
944 return error_status, error_obj
945 return error_status, {**error_obj, "provider_specific_fields": merged_fields}
948def _sse_error_frames(error_obj: Mapping[str, object]) -> tuple[str, str]:
949 """The two frames an SSE stream ends with once it can no longer raise."""
950 return f"data: {json.dumps({'error': error_obj})}\n\n", "data: [DONE]\n\n"
953def _sse_stream_headers(headers: Mapping[str, str]) -> Mapping[str, str]:
954 """`headers` plus the two that stop reverse proxies from buffering SSE (issue #28384)."""
955 return MappingProxyType({**headers, **_TTFT_KEEPALIVE_HEADERS})
958async def _resolve_stream_headers(
959 headers: Mapping[str, str], refresh_headers: Callable[[], Awaitable[Mapping[str, str]]] | None
960) -> Mapping[str, str]:
961 if refresh_headers is None:
962 return headers
963 try:
964 return await refresh_headers()
965 except Exception as e: # noqa: BLE001 # a stream whose first chunk is already paid for must not fail over its headers
966 verbose_proxy_logger.exception("Error refreshing streaming response headers: %s", e)
967 return headers
970_NO_GENERAL_SETTINGS: Final[Mapping[str, object]] = MappingProxyType({})
973async def create_response(
974 generator: AsyncGenerator[str, None],
975 media_type: str,
976 headers: Mapping[str, str],
977 default_status_code: int = status.HTTP_200_OK,
978 request: Request | None = None,
979 refresh_headers: Callable[[], Awaitable[Mapping[str, str]]] | None = None,
980 general_settings: Mapping[str, object] = _NO_GENERAL_SETTINGS,
981) -> StreamingResponse | JSONResponse:
982 """
983 Create streaming response, checking if the first chunk is an error.
984 If the first chunk is an error, return a standard JSON error response.
985 Otherwise, return StreamingResponse and stream all content.
987 ``refresh_headers`` is consulted once the first chunk has been buffered, for
988 callers whose headers can only be known then. ``general_settings`` decides whether
989 the first-chunk error body also carries the ``x-litellm-call-id`` header's value.
990 """
991 first_chunk_value: str | None = None
992 final_status_code = default_status_code
994 try:
995 # Handle coroutine that returns a generator
996 if asyncio.iscoroutine(generator):
997 generator = await generator
999 # Now get the first chunk from the actual generator
1000 first_chunk_value = await _buffer_first_chunk_honoring_disconnect(generator, request)
1001 resolved_headers: Final = await _resolve_stream_headers(headers, refresh_headers)
1003 if first_chunk_value is not None:
1004 try:
1005 error_code_from_chunk: Final = await _parse_event_data_for_error(first_chunk_value)
1006 if error_code_from_chunk is not None:
1007 # First chunk is an error, stream hasn't really started yet
1008 # Should return standard JSON error response instead of SSE format
1009 final_status_code = error_code_from_chunk
1010 verbose_proxy_logger.debug(
1011 "Error detected in first stream chunk. Returning JSON error response with status code: %s",
1012 final_status_code,
1013 )
1015 # Parse error content
1016 error_dict: Final = with_call_id(
1017 JSON_OBJECT.validate_python(_extract_error_from_sse_chunk(first_chunk_value)),
1018 error_body_call_id(general_settings, resolved_headers.get(LITELLM_CALL_ID_HEADER)),
1019 )
1021 # Consume and close generator (avoid resource leak)
1022 try:
1023 await generator.aclose()
1024 except Exception:
1025 pass
1027 # Return JSON format error response
1028 return JSONResponse(
1029 status_code=final_status_code,
1030 content={"error": error_dict},
1031 headers=resolved_headers,
1032 )
1033 except Exception as e:
1034 verbose_proxy_logger.debug("Error parsing first chunk value: %s", e)
1036 except _ClientDisconnectedBeforeFirstChunk:
1037 # Client vanished during the time-to-first-token wait; the upstream
1038 # stream is already closed. Return a 499 the (now-gone) client never reads.
1039 return JSONResponse(
1040 status_code=LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED,
1041 content={
1042 "error": {
1043 "message": _CLIENT_DISCONNECT_DETAIL,
1044 "type": "client_disconnect",
1045 "param": None,
1046 "code": str(LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED),
1047 }
1048 },
1049 headers=headers,
1050 )
1051 except StopAsyncIteration:
1052 # Generator was empty. Default status
1053 async def empty_gen() -> AsyncGenerator[str, None]:
1054 if False:
1055 yield
1057 return StreamingResponse(
1058 empty_gen(),
1059 media_type=media_type,
1060 headers=_sse_stream_headers(await _resolve_stream_headers(headers, refresh_headers)),
1061 status_code=default_status_code,
1062 )
1063 except Exception as e:
1064 # Unexpected error consuming first chunk.
1065 verbose_proxy_logger.exception("Error consuming first chunk from generator: %s", e)
1067 error_status, error_obj = sse_error_payload(e)
1069 async def error_gen_message() -> AsyncGenerator[str, None]:
1070 for frame in _sse_error_frames(error_obj):
1071 yield frame
1073 return StreamingResponse(
1074 error_gen_message(),
1075 media_type=media_type,
1076 headers=_sse_stream_headers(await _resolve_stream_headers(headers, refresh_headers)),
1077 status_code=error_status,
1078 )
1080 async def combined_generator() -> AsyncGenerator[str, None]:
1081 if not _DD_STREAMING_TRACE_ENABLED:
1082 # Fast path: no per-chunk span object / context-manager overhead.
1083 if first_chunk_value is not None:
1084 yield first_chunk_value
1085 async for chunk in generator:
1086 yield chunk
1087 return
1088 if first_chunk_value is not None:
1089 with tracer.trace(DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE):
1090 yield first_chunk_value
1091 async for chunk in generator:
1092 with tracer.trace(DD_TRACER_STREAMING_CHUNK_YIELD_RESOURCE):
1093 yield chunk
1095 return _UpstreamClosingStreamingResponse(
1096 combined_generator(),
1097 media_type=media_type,
1098 headers=_sse_stream_headers(resolved_headers),
1099 status_code=final_status_code,
1100 upstream_generator=generator,
1101 )
1104_TTFT_KEEPALIVE_HEADERS: Final[Mapping[str, str]] = MappingProxyType(
1105 {
1106 "Cache-Control": "no-cache",
1107 "X-Accel-Buffering": "no",
1108 }
1109)
1112def ttft_keepalive_interval(request_data: Mapping[str, object], llm_router: Router | None = None) -> float | None:
1113 """The operator's keepalive interval, but only for a request that asked to stream.
1115 Resolved through the deployments the request could land on, so a deployment's
1116 `keepalive_seconds: 0` stays the hard disable it is documented to be rather
1117 than being switched back on by the global default.
1118 """
1119 if request_data.get("stream") is not True:
1120 return None
1121 requested_model: Final = request_data.get("model")
1122 deployments: Final = (
1123 llm_router.get_model_list(model_name=requested_model) or ()
1124 if llm_router is not None and isinstance(requested_model, str)
1125 else ()
1126 )
1127 return resolve_ttft_keepalive_interval(deployments, litellm.sse_keepalive_ping_interval_seconds)
1130async def _aclose_late_response(produced: Response) -> None:
1131 """Run the cleanup Starlette would have run, for a response it never called.
1133 Closing an already-closed async generator is a no-op, so this is safe to call
1134 from both the relay's own teardown and the outer one.
1135 """
1136 if not isinstance(produced, StreamingResponse):
1137 return
1138 targets: Final = (
1139 (produced.body_iterator, produced.upstream_generator)
1140 if isinstance(produced, _UpstreamClosingStreamingResponse)
1141 else (produced.body_iterator,)
1142 )
1143 for target in targets:
1144 aclose = getattr(target, "aclose", None)
1145 if aclose is None:
1146 continue
1147 try:
1148 await aclose()
1149 except BaseException as exc: # noqa: BLE001 # teardown must not mask why the stream ended
1150 verbose_proxy_logger.debug("error closing relayed streaming generator: %s", exc)
1153async def _relay_late_response(produced: Response) -> AsyncGenerator[bytes, None]:
1154 """Replay a Response that was built after a keepalive had already opened the wire."""
1155 if not isinstance(produced, StreamingResponse):
1156 # The status line is already on the wire, so a non-streaming body, an error
1157 # body included, can only reach the client as an SSE frame.
1158 yield b"data: " + (bytes(produced.body) or b"{}") + b"\n\n"
1159 yield b"data: [DONE]\n\n"
1160 return
1162 try:
1163 async for chunk in produced.body_iterator:
1164 yield chunk.encode("utf-8") if isinstance(chunk, str) else bytes(chunk)
1165 finally:
1166 # Starlette never called this response, so the cleanup its __call__ would
1167 # have run has to happen here or the upstream LLM connection leaks.
1168 with anyio.CancelScope(shield=True):
1169 await _aclose_late_response(produced)
1172async def _sanitized_late_failure(
1173 exc: Exception,
1174 on_late_failure: "Callable[[Exception], Awaitable[HTTPException | None]] | None",
1175) -> Exception:
1176 """Report a late failure and return whatever should reach the client.
1178 ``post_call_failure_hook`` lets a callback replace the client-facing error, by
1179 returning a replacement or by raising one, and both are used elsewhere in this
1180 module. Serializing the original would leak provider detail a deployment had
1181 configured away, so the hook's answer wins. A callback that fails some other
1182 way is a bug in the callback, not a reason to lose the real error.
1183 """
1184 if on_late_failure is None:
1185 return exc
1186 try:
1187 replacement: Final = await on_late_failure(exc)
1188 except HTTPException as raised_replacement:
1189 return raised_replacement
1190 except Exception as hook_failure: # noqa: BLE001 # a broken callback must not replace the real error
1191 verbose_proxy_logger.exception("post_call_failure_hook raised while reporting a late failure: %s", hook_failure)
1192 return exc
1193 return replacement if replacement is not None else exc
1196async def open_sse_before_first_byte(
1197 produce_response: Awaitable[_LateResponseT],
1198 ping_interval_seconds: float | str | None,
1199 media_type: str = "text/event-stream",
1200 on_late_failure: Callable[[Exception], Awaitable[HTTPException | None]] | None = None,
1201) -> _LateResponseT | StreamingResponse:
1202 """Write SSE keepalive comments while the upstream LLM call is still in flight.
1204 The whole time-to-first-token is spent inside `produce_response`: the upstream
1205 withholds its response headers until it emits its first token, so nothing has
1206 entered the ASGI response phase yet and the proxy writes zero bytes. An
1207 intermediary with an idle read timeout (AWS ALB and nginx both default to 60s)
1208 then drops a connection that is perfectly healthy.
1210 When `produce_response` does not finish within one interval, the response is
1211 opened immediately and `: ping` comments, which every conformant SSE client
1212 ignores, fill the wire until the real response is ready to be replayed onto it.
1213 Committing the status line that early is the cost: a failure discovered after
1214 the first ping reaches the client as an SSE error frame under a 200 rather than
1215 as an HTTP error status, and LiteLLM's own `x-litellm-*` response headers are
1216 not yet known. Both are why this stays off until an operator sets an interval.
1217 """
1218 interval: Final = coerce_keepalive_interval(ping_interval_seconds)
1219 if interval is None: 1219 ↛ 1222line 1219 didn't jump to line 1222 because the condition on line 1219 was always true
1220 return await produce_response
1222 produce_task: Final = asyncio.ensure_future(produce_response)
1223 await asyncio.wait((produce_task,), timeout=interval)
1224 if produce_task.done():
1225 # Fast path: the upstream answered inside one interval, so nothing was
1226 # written early and this is byte-identical to not being wrapped at all.
1227 return produce_task.result()
1229 async def keepalive_then_relay() -> AsyncGenerator[bytes, None]:
1230 try:
1231 while not produce_task.done():
1232 yield SSE_COMMENT_PING_BYTES
1233 await asyncio.wait((produce_task,), timeout=interval)
1234 try:
1235 produced: Final = produce_task.result()
1236 except Exception as exc: # noqa: BLE001 # the status line is already sent; surface it as a frame
1237 verbose_proxy_logger.exception(
1238 "request failed after its SSE keepalive had opened the response: %s", exc
1239 )
1240 # The caller's own `except` never sees this, so its failure hook
1241 # would never fire and the failure would go unaudited. The hook
1242 # also gets to sanitize what reaches the client, by returning or
1243 # raising a replacement, so its answer decides the frame.
1244 _, error_obj = sse_error_payload(await _sanitized_late_failure(exc, on_late_failure))
1245 for frame in _sse_error_frames(error_obj):
1246 yield frame.encode()
1247 return
1248 async for chunk in _relay_late_response(produced):
1249 yield chunk
1250 finally:
1251 if not produce_task.done():
1252 produce_task.cancel()
1253 with anyio.CancelScope(shield=True):
1254 with contextlib.suppress(BaseException):
1255 await produce_task
1256 elif not produce_task.cancelled():
1257 # The upstream may have answered while nobody was draining this
1258 # relay, e.g. the client vanished first. Nothing else holds that
1259 # response, so its stream only gets closed here.
1260 with anyio.CancelScope(shield=True):
1261 with contextlib.suppress(BaseException):
1262 await _aclose_late_response(produce_task.result())
1264 verbose_proxy_logger.info(
1265 "no upstream response after %ss, opening the SSE response early and sending keepalives", interval
1266 )
1267 return StreamingResponse(
1268 keepalive_then_relay(),
1269 media_type=media_type,
1270 headers=_TTFT_KEEPALIVE_HEADERS,
1271 )
1274def _is_azure_model_router_request(model: str, hidden_params: Mapping[str, object] | None = None) -> bool:
1275 """
1276 Check if a request went down the Azure Model Router route.
1278 ``model`` here is what the *client* sent, a model group alias with no ``model_router/``
1279 prefix, so matching on it alone only works when the operator happened to put "model-router"
1280 in the alias. Where the response is in hand its stamp answers this outright, so callers
1281 should pass ``hidden_params``.
1283 Args:
1284 model: The requested model name
1285 hidden_params: ``_hidden_params`` from the response, when the caller has it
1287 Returns:
1288 bool: True if this is an Azure Model Router request
1289 """
1290 from litellm.llms.azure_ai.common_utils import AzureFoundryModelInfo
1292 return AzureFoundryModelInfo.is_model_router_call(model=model, hidden_params=hidden_params)
1295def _override_openai_response_model(
1296 *,
1297 response_obj: object,
1298 requested_model: str,
1299 log_context: str,
1300 return_raw_model_name: bool = False,
1301) -> None:
1302 """
1303 Force the OpenAI-compatible `model` field in the response to match what the client requested.
1305 LiteLLM internally prefixes some provider/deployment model identifiers (e.g. `hosted_vllm/...`).
1306 That internal identifier should not be returned to clients in the OpenAI `model` field.
1308 Note: This is intentionally verbose at debug level. A model mismatch is a useful signal that an
1309 internal model identifier is being stamped/preserved somewhere in the request/response pipeline.
1310 We log mismatches as debug (and then restamp to the client-requested value) so these paths stay
1311 observable for maintainers without breaking client compatibility or alarming operators.
1313 Responses that omit an OpenAI-style `model` field are left unchanged (silent return),
1314 including dict responses with no `model` key.
1316 Exceptions:
1317 1. If a fallback occurred (indicated by x-litellm-attempted-fallbacks header),
1318 we preserve the actual model that was used (the fallback model).
1319 2. If the request was to an Azure Model Router, we preserve the actual model
1320 that was used (e.g., gpt-5-nano-2025-08-07) instead of the router model.
1321 3. If this was a fastest_response batch completion, use the winning model's
1322 model group name instead of the comma-separated list the client sent.
1323 """
1324 if return_raw_model_name or not requested_model:
1325 return
1327 hidden_params: Final = get_hidden_params_dict(response_obj)
1328 if isinstance(hidden_params, dict):
1329 # Check if a fallback occurred - if so, preserve the actual model used
1330 fallback_headers: Final = hidden_params.get("additional_headers", {}) or {}
1331 attempted_fallbacks: Final = fallback_headers.get("x-litellm-attempted-fallbacks", None)
1332 if attempted_fallbacks is not None and attempted_fallbacks > 0:
1333 verbose_proxy_logger.debug(
1334 "%s: fallback detected (attempted_fallbacks=%d), preserving actual model used instead of overriding to requested model.",
1335 log_context,
1336 attempted_fallbacks,
1337 )
1338 return
1340 # For fastest_response batch completions, use the winning model's group
1341 # name rather than the comma-separated list the client sent.
1342 if hidden_params.get("fastest_response_batch_completion"):
1343 winning_model: Final = fallback_headers.get("x-litellm-model-group")
1344 if winning_model:
1345 verbose_proxy_logger.debug(
1346 "%s: fastest_response detected, using winning model group=%r instead of requested=%r.",
1347 log_context,
1348 winning_model,
1349 requested_model,
1350 )
1351 requested_model = winning_model
1352 else:
1353 verbose_proxy_logger.debug(
1354 "%s: fastest_response detected but no model group header found, preserving actual model from response.",
1355 log_context,
1356 )
1357 return
1359 # Check if this is an Azure Model Router request - if so, preserve the actual model used
1360 if _is_azure_model_router_request(requested_model, hidden_params):
1361 verbose_proxy_logger.debug(
1362 "%s: Azure Model Router detected - preserving actual model used from response instead of overriding to router model.",
1363 log_context,
1364 )
1365 return
1367 if isinstance(response_obj, dict):
1368 if "model" not in response_obj:
1369 return
1370 downstream_model = response_obj.get("model")
1371 if downstream_model != requested_model:
1372 verbose_proxy_logger.debug(
1373 "%s: response model mismatch - requested=%r downstream=%r. Overriding response['model'] to requested model.",
1374 log_context,
1375 requested_model,
1376 downstream_model,
1377 )
1378 response_obj["model"] = requested_model
1379 return
1381 if not hasattr(response_obj, "model"):
1382 return
1384 downstream_model = getattr(response_obj, "model", None)
1385 if downstream_model != requested_model:
1386 verbose_proxy_logger.debug(
1387 "%s: response model mismatch - requested=%r downstream=%r. Overriding response.model to requested model.",
1388 log_context,
1389 requested_model,
1390 downstream_model,
1391 )
1393 try:
1394 setattr(response_obj, "model", requested_model)
1395 except Exception as e:
1396 verbose_proxy_logger.debug(
1397 "%s: failed to override response.model=%r on response_type=%s. error=%s",
1398 log_context,
1399 requested_model,
1400 type(response_obj),
1401 str(e),
1402 exc_info=True,
1403 )
1406_METADATA_BUCKET_KEYS: Final = ("metadata", "litellm_metadata")
1407_RESPONSE_REDACTED_KEYS: Final = ("keyword", "snippet", "match", "regex")
1410def _request_metadata_buckets(request_data: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:
1411 return tuple(bucket for key in _METADATA_BUCKET_KEYS if isinstance(bucket := request_data.get(key), Mapping))
1414def include_guardrail_response_requested(request_data: Mapping[str, object]) -> bool:
1415 return any(bucket.get("include_guardrail_response") is True for bucket in _request_metadata_buckets(request_data))
1418def attach_guardrail_information(response: object, request_data: Mapping[str, object]) -> object:
1419 recorded: Final[Sequence[object]] = next(
1420 (
1421 entries
1422 for bucket in _request_metadata_buckets(request_data)
1423 if isinstance(
1424 entries := bucket.get("standard_logging_guardrail_information"),
1425 list,
1426 )
1427 ),
1428 (),
1429 )
1430 guardrail_information: Final = [ # mutable-ok: response list contract
1431 redact_nested_match_and_regex_keys(entry, keys=_RESPONSE_REDACTED_KEYS)
1432 for entry in recorded
1433 if isinstance(entry, dict)
1434 ]
1435 if isinstance(response, dict):
1436 return response | MappingProxyType({"guardrail_information": guardrail_information})
1437 if isinstance(response, BaseModel) and response.model_config.get("extra") == "allow":
1438 return response.model_copy(update=MappingProxyType({"guardrail_information": guardrail_information}))
1439 return response
1442class CostBreakdownHeaderValues(NamedTuple):
1443 original_cost: float | None = None
1444 discount_amount: float | None = None
1445 margin_total_amount: float | None = None
1446 margin_percent: float | None = None
1447 input_cost: float | None = None
1448 output_cost: float | None = None
1449 cache_read_cost: float | None = None
1450 cache_creation_cost: float | None = None
1451 reasoning_cost: float | None = None
1452 tool_usage_cost: float | None = None
1455def _uncached_input_cost(
1456 input_cost: float | None,
1457 cache_read_cost: float | None,
1458 cache_creation_cost: float | None,
1459) -> float | None:
1460 """The stored input cost nests the cache costs inside it; headers advertise the additive split instead."""
1461 if input_cost is None:
1462 return None
1463 return input_cost - (cache_read_cost or 0.0) - (cache_creation_cost or 0.0)
1466_ZERO_COST_BREAKDOWN: Final = CostBreakdownHeaderValues(
1467 original_cost=0.0,
1468 discount_amount=0.0,
1469 margin_total_amount=0.0,
1470 margin_percent=0.0,
1471 input_cost=0.0,
1472 output_cost=0.0,
1473 tool_usage_cost=0.0,
1474)
1475"""The component split a call priced at zero advertises, so a client reading the cost headers off a
1476read or management route still finds the whole family rather than a partially populated one."""
1479def _totals_to_zero(response_cost: float | str | None) -> bool:
1480 """Whether the total these headers carry is zero, counting a total no route ever priced as one.
1482 A component split is only reported as zero alongside a total that agrees with it, so a read
1483 that did price normally never advertises a real total beside an all-zero split.
1484 """
1485 if response_cost is None or response_cost == "": 1485 ↛ 1486line 1485 didn't jump to line 1486 because the condition on line 1485 was never true
1486 return True
1487 try:
1488 return float(response_cost) == 0.0
1489 except (TypeError, ValueError):
1490 return False
1493def _get_cost_breakdown_from_logging_obj(
1494 litellm_logging_obj: LiteLLMLoggingObj | None,
1495 response_cost: float | str | None = None,
1496) -> CostBreakdownHeaderValues:
1497 """Extract discount, margin, and per-component cost information from logging object's cost breakdown.
1499 A non-inference call that priced at zero never records a breakdown, so its components are
1500 reported as zero here. Any such call that did price normally (retrieving a background response,
1501 and the cost poller's read of one) reports the breakdown it stored, or nothing at all when the
1502 breakdown has not landed yet.
1503 """
1504 if not litellm_logging_obj or not hasattr(litellm_logging_obj, "cost_breakdown"):
1505 return CostBreakdownHeaderValues()
1507 cost_breakdown: Final = litellm_logging_obj.cost_breakdown
1508 if not cost_breakdown: 1508 ↛ 1513line 1508 didn't jump to line 1513 because the condition on line 1508 was always true
1509 if litellm_logging_obj.call_type in NON_INFERENCE_CALL_TYPES and _totals_to_zero(response_cost):
1510 return _ZERO_COST_BREAKDOWN
1511 return CostBreakdownHeaderValues()
1513 return CostBreakdownHeaderValues(
1514 original_cost=cost_breakdown.get("original_cost"),
1515 discount_amount=cost_breakdown.get("discount_amount"),
1516 margin_total_amount=cost_breakdown.get("margin_total_amount"),
1517 margin_percent=cost_breakdown.get("margin_percent"),
1518 input_cost=_uncached_input_cost(
1519 input_cost=cost_breakdown.get("input_cost"),
1520 cache_read_cost=cost_breakdown.get("cache_read_cost"),
1521 cache_creation_cost=cost_breakdown.get("cache_creation_cost"),
1522 ),
1523 output_cost=cost_breakdown.get("output_cost"),
1524 cache_read_cost=cost_breakdown.get("cache_read_cost"),
1525 cache_creation_cost=cost_breakdown.get("cache_creation_cost"),
1526 reasoning_cost=cost_breakdown.get("reasoning_cost"),
1527 tool_usage_cost=cost_breakdown.get("tool_usage_cost"),
1528 )
1531def _classifier_cost_from_request_data(request_data: Mapping[str, object] | None) -> float | None:
1532 """Cost of the auto-router's LLM classifier call, read from the request's routing_decision.
1534 The pre-routing hook records the decision in `litellm_metadata` on messages/batch-style
1535 routes and in `metadata` on chat-style routes, so both buckets are consulted, in the same
1536 precedence `get_or_create_metadata_bucket` writes them.
1537 """
1538 from litellm.proxy.spend_tracking.savings import classifier_cost_from_decision
1540 data: Final = request_data or {}
1541 for metadata_key in ("litellm_metadata", "metadata"):
1542 metadata = data.get(metadata_key)
1543 if not isinstance(metadata, dict):
1544 continue
1545 decision = metadata.get("routing_decision")
1546 if not isinstance(decision, dict): 1546 ↛ 1548line 1546 didn't jump to line 1548 because the condition on line 1546 was always true
1547 continue
1548 cost = classifier_cost_from_decision(decision)
1549 if cost is None:
1550 continue
1551 return cost
1552 return None
1555def _has_attribute_error_in_chain(exc: Exception) -> bool:
1556 """Walk the exception chain to find an AttributeError at any depth.
1558 Checks __cause__, __context__, and the litellm-specific original_exception
1559 attribute iteratively. Depth is capped at DEFAULT_MAX_RECURSE_DEPTH to
1560 avoid infinite loops from circular exception references.
1561 """
1562 stack: Final[list[BaseException]] = [exc]
1563 seen: Final[set[int]] = set()
1564 depth = 0
1565 while stack and depth < DEFAULT_MAX_RECURSE_DEPTH:
1566 current = stack.pop()
1567 exc_id = id(current)
1568 if exc_id in seen: 1568 ↛ 1569line 1568 didn't jump to line 1569 because the condition on line 1568 was never true
1569 continue
1570 seen.add(exc_id)
1571 if isinstance(current, AttributeError):
1572 return True
1573 for attr in ("__cause__", "__context__", "original_exception"):
1574 inner = getattr(current, attr, None)
1575 if inner is not None and isinstance(inner, BaseException):
1576 stack.append(inner)
1577 depth += 1
1578 return False
1581_CLIENT_DISCONNECT_DETAIL: Final = "Client disconnected the request"
1584@runtime_checkable
1585class _CarriesLitellmCallId(Protocol):
1586 litellm_call_id: str | None
1589def request_litellm_call_id(data: Mapping[str, object]) -> str | None:
1590 logging_obj: Final = data.get("litellm_logging_obj")
1591 logged_id: Final = logging_obj.litellm_call_id if isinstance(logging_obj, _CarriesLitellmCallId) else None
1592 call_id: Final = logged_id or data.get("litellm_call_id")
1593 return call_id if isinstance(call_id, str) else None
1596def log_llm_api_exception(e: Exception, litellm_call_id: str | None) -> None:
1597 if getattr(e, "status_code", None) == 499 and getattr(e, "detail", None) == _CLIENT_DISCONNECT_DETAIL: 1597 ↛ 1598line 1597 didn't jump to line 1598 because the condition on line 1597 was never true
1598 verbose_proxy_logger.info(
1599 "litellm.proxy.proxy_server._handle_llm_api_exception(): client disconnected, "
1600 "upstream LLM request cancelled - litellm_call_id=%s",
1601 litellm_call_id,
1602 extra=MappingProxyType({"litellm_call_id": litellm_call_id}),
1603 )
1604 return
1605 log_fn: Final = (
1606 verbose_proxy_logger.error
1607 if is_expected_client_error(e) and not litellm.log_client_error_tracebacks
1608 else verbose_proxy_logger.exception
1609 )
1610 log_fn(
1611 "litellm.proxy.proxy_server._handle_llm_api_exception(): Exception occured - litellm_call_id=%s - %s",
1612 litellm_call_id,
1613 e,
1614 extra=MappingProxyType({"litellm_call_id": litellm_call_id}),
1615 )
1618async def _cancel_llm_call_on_client_disconnect(
1619 request: Request,
1620 llm_api_call: "asyncio.Future[_LlmCallT]",
1621 disconnect_event: asyncio.Event,
1622) -> None:
1623 try:
1624 while True:
1625 message = await request.receive()
1626 if message["type"] == "http.disconnect":
1627 disconnect_event.set()
1628 llm_api_call.cancel()
1629 return
1630 except Exception as exc:
1631 verbose_proxy_logger.warning(
1632 "cancel_on_disconnect: request.receive() raised %s; upstream LLM call will not be cancelled on disconnect",
1633 exc,
1634 )
1637async def _await_llm_call_cancelling_on_disconnect(
1638 request: Request,
1639 llm_api_call: "asyncio.Future[_LlmCallT]",
1640) -> _LlmCallT:
1641 disconnect_event: Final = asyncio.Event()
1642 monitor: Final = asyncio.create_task(_cancel_llm_call_on_client_disconnect(request, llm_api_call, disconnect_event))
1643 try:
1644 return await llm_api_call
1645 except asyncio.CancelledError:
1646 if disconnect_event.is_set():
1647 raise HTTPException(
1648 status_code=499,
1649 detail=_CLIENT_DISCONNECT_DETAIL,
1650 )
1651 raise
1652 finally:
1653 monitor.cancel()
1656def _timing_values(
1657 *,
1658 hidden_params: Mapping[str, object],
1659 logging_obj: LiteLLMLoggingObj | None,
1660 use_logging_obj: bool,
1661) -> Mapping[str, object]:
1662 """Both timing values from one source, so the two headers always describe the same window.
1664 /v1/messages returns a plain dict and the Anthropic / Responses bridge stream wrappers carry no
1665 ``_hidden_params``, so ``update_response_metadata`` leaves their timing on the logging object.
1666 """
1667 if hidden_params.get("_response_ms") is not None or not use_logging_obj or logging_obj is None:
1668 return hidden_params
1669 return getattr(logging_obj, "response_timing_metrics", None) or {} # mutable-ok: empty fallback
1672class ProxyBaseLLMRequestProcessing:
1673 def __init__(self, data: dict):
1674 self.data = data
1675 self._tags_before_guardrails: frozenset[str] | None = None
1677 @property
1678 def litellm_call_id(self) -> str | None:
1679 return request_litellm_call_id(self.data)
1681 @staticmethod
1682 def _merge_passthrough_streaming_headers(
1683 response_headers: httpx.Headers | dict | None,
1684 custom_headers: Mapping[str, str],
1685 ) -> dict:
1686 """
1687 Merge upstream passthrough headers with proxy/custom headers.
1689 Proxy/custom headers win on key collisions.
1690 """
1691 excluded_headers: Final = { # mutable-ok: set of header names to exclude from forwarding
1692 "transfer-encoding",
1693 "content-encoding",
1694 "set-cookie",
1695 "connection",
1696 "keep-alive",
1697 "proxy-authenticate",
1698 "proxy-authorization",
1699 "te",
1700 "trailer",
1701 "upgrade",
1702 }
1704 merged_headers: Final = { # mutable-ok: dict comprehension for merged headers forwarded to httpx
1705 key: value for key, value in dict(response_headers or {}).items() if key.lower() not in excluded_headers
1706 }
1707 merged_headers.update(custom_headers)
1708 return merged_headers
1710 @staticmethod
1711 def get_custom_headers(
1712 *,
1713 user_api_key_dict: UserAPIKeyAuth,
1714 call_id: str | None = None,
1715 model_id: str | None = None,
1716 cache_key: str | None = None,
1717 api_base: str | None = None,
1718 version: str | None = None,
1719 model_region: str | None = None,
1720 response_cost: float | str | None = None,
1721 hidden_params: Mapping[str, object] | None = None,
1722 fastest_response_batch_completion: bool | None = None,
1723 request_data: dict | None = {},
1724 timeout: float | httpx.Timeout | None = None,
1725 litellm_logging_obj: LiteLLMLoggingObj | None = None,
1726 read_timing_from_logging_obj: bool = True,
1727 **kwargs,
1728 ) -> dict:
1729 exclude_values: Final = {"", None, "None"}
1730 hidden_params = hidden_params or {}
1731 resolved_call_id: Final = (
1732 call_id or hidden_params.get("litellm_call_id") or (request_data or {}).get("litellm_call_id")
1733 )
1734 timing_values: Final = _timing_values(
1735 hidden_params=hidden_params,
1736 logging_obj=litellm_logging_obj,
1737 use_logging_obj=read_timing_from_logging_obj,
1738 )
1740 cost_breakdown: Final = _get_cost_breakdown_from_logging_obj(
1741 litellm_logging_obj=litellm_logging_obj, response_cost=response_cost
1742 )
1744 # Calculate updated spend for header (include current response_cost)
1745 current_spend: Final = user_api_key_dict.spend or 0.0
1746 updated_spend = current_spend
1747 if response_cost is not None:
1748 try:
1749 # Convert response_cost to float if it's a string
1750 cost_value: Final = float(response_cost) if isinstance(response_cost, str) else response_cost
1751 if cost_value > 0: 1751 ↛ 1752line 1751 didn't jump to line 1752 because the condition on line 1751 was never true
1752 updated_spend = current_spend + cost_value
1753 except (ValueError, TypeError):
1754 # If conversion fails, use original spend
1755 pass
1757 model_name: Final = ProxyBaseLLMRequestProcessing._get_deployment_model_name(litellm_logging_obj)
1758 classifier_cost: Final = _classifier_cost_from_request_data(request_data)
1760 headers: Final = {
1761 "x-litellm-call-id": resolved_call_id,
1762 "x-litellm-model-id": model_id,
1763 "x-litellm-model-name": model_name,
1764 "x-litellm-cache-key": cache_key,
1765 "x-litellm-model-api-base": (
1766 api_base.split("?")[0] if api_base else None
1767 ), # don't include query params, risk of leaking sensitive info
1768 "x-litellm-version": version,
1769 "x-litellm-model-region": model_region,
1770 "x-litellm-response-cost": str(response_cost),
1771 "x-litellm-response-cost-original": (
1772 str(cost_breakdown.original_cost) if cost_breakdown.original_cost is not None else None
1773 ),
1774 "x-litellm-response-cost-discount-amount": (
1775 str(cost_breakdown.discount_amount) if cost_breakdown.discount_amount is not None else None
1776 ),
1777 "x-litellm-response-cost-margin-amount": (
1778 str(cost_breakdown.margin_total_amount) if cost_breakdown.margin_total_amount is not None else None
1779 ),
1780 "x-litellm-response-cost-margin-percent": (
1781 str(cost_breakdown.margin_percent) if cost_breakdown.margin_percent is not None else None
1782 ),
1783 "x-litellm-response-cost-input": (
1784 str(cost_breakdown.input_cost) if cost_breakdown.input_cost is not None else None
1785 ),
1786 "x-litellm-response-cost-output": (
1787 str(cost_breakdown.output_cost) if cost_breakdown.output_cost is not None else None
1788 ),
1789 "x-litellm-response-cost-cache-read": (
1790 str(cost_breakdown.cache_read_cost) if cost_breakdown.cache_read_cost is not None else None
1791 ),
1792 "x-litellm-response-cost-cache-creation": (
1793 str(cost_breakdown.cache_creation_cost) if cost_breakdown.cache_creation_cost is not None else None
1794 ),
1795 "x-litellm-response-cost-reasoning": (
1796 str(cost_breakdown.reasoning_cost) if cost_breakdown.reasoning_cost is not None else None
1797 ),
1798 "x-litellm-response-cost-tool-usage": (
1799 str(cost_breakdown.tool_usage_cost) if cost_breakdown.tool_usage_cost is not None else None
1800 ),
1801 "x-litellm-classifier-cost": (str(classifier_cost) if classifier_cost is not None else None),
1802 "x-litellm-key-tpm-limit": str(user_api_key_dict.tpm_limit),
1803 "x-litellm-key-rpm-limit": str(user_api_key_dict.rpm_limit),
1804 "x-litellm-key-max-budget": str(user_api_key_dict.max_budget),
1805 "x-litellm-key-spend": str(updated_spend),
1806 "x-litellm-response-duration-ms": str(timing_values.get("_response_ms")),
1807 "x-litellm-overhead-duration-ms": str(timing_values.get("litellm_overhead_time_ms")),
1808 "x-litellm-callback-duration-ms": str(hidden_params.get("callback_duration_ms", None)),
1809 **(
1810 {
1811 "x-litellm-timing-pre-processing-ms": str(hidden_params.get("timing_pre_processing_ms", None)),
1812 "x-litellm-timing-llm-api-ms": str(hidden_params.get("timing_llm_api_ms", None)),
1813 "x-litellm-timing-post-processing-ms": str(hidden_params.get("timing_post_processing_ms", None)),
1814 "x-litellm-timing-message-copy-ms": str(hidden_params.get("timing_message_copy_ms", None)),
1815 }
1816 if LITELLM_DETAILED_TIMING
1817 else {}
1818 ),
1819 "x-litellm-fastest_response_batch_completion": (
1820 str(fastest_response_batch_completion) if fastest_response_batch_completion is not None else None
1821 ),
1822 "x-litellm-timeout": str(timeout) if timeout is not None else None,
1823 **{k: str(v) for k, v in kwargs.items()},
1824 }
1825 if request_data:
1826 remaining_tokens_header: Final = get_remaining_tokens_and_requests_from_request_data(request_data)
1827 headers.update(remaining_tokens_header)
1829 logging_caching_headers: Final = get_logging_caching_headers(request_data)
1830 if logging_caching_headers:
1831 headers.update(logging_caching_headers)
1833 try:
1834 return {key: str(value) for key, value in headers.items() if value not in exclude_values}
1835 except Exception as e:
1836 verbose_proxy_logger.error("Error setting custom headers: %s", e)
1837 return {}
1839 @staticmethod
1840 async def build_litellm_proxy_success_headers_from_llm_response(
1841 *,
1842 response: object,
1843 request_data: dict,
1844 request: Request,
1845 user_api_key_dict: UserAPIKeyAuth,
1846 logging_obj: LiteLLMLoggingObj,
1847 version: str | None,
1848 proxy_logging_obj: ProxyLogging,
1849 ) -> dict[str, str]:
1850 """
1851 Build LiteLLM proxy response headers for routes that call the LLM directly
1852 (e.g. Google native :generateContent) instead of base_process_llm_request.
1853 """
1854 if isinstance(response, dict):
1855 hidden_params = get_hidden_params_dict(response)
1856 else:
1857 hidden_params = getattr(response, "_hidden_params", None) or {}
1858 if not isinstance(hidden_params, dict):
1859 hidden_params = {}
1861 model_id: Final = ProxyBaseLLMRequestProcessing._get_model_id_from_response(hidden_params, request_data)
1863 cache_key: Final = hidden_params.get("cache_key", None) or ""
1864 api_base: Final = hidden_params.get("api_base", None) or ""
1865 response_cost: Final = hidden_params.get("response_cost", None) or ""
1866 fastest_response_batch_completion: Final = hidden_params.get("fastest_response_batch_completion", None)
1867 additional_headers: Final = hidden_params.get("additional_headers", {}) or {}
1869 custom_headers: Final = ProxyBaseLLMRequestProcessing.get_custom_headers(
1870 user_api_key_dict=user_api_key_dict,
1871 call_id=logging_obj.litellm_call_id,
1872 model_id=model_id,
1873 cache_key=cache_key,
1874 api_base=api_base,
1875 version=version,
1876 response_cost=response_cost,
1877 model_region=getattr(user_api_key_dict, "allowed_model_region", ""),
1878 fastest_response_batch_completion=fastest_response_batch_completion,
1879 request_data=request_data,
1880 hidden_params=hidden_params,
1881 litellm_logging_obj=logging_obj,
1882 **additional_headers,
1883 )
1885 callback_headers: Final = await proxy_logging_obj.post_call_response_headers_hook(
1886 data=request_data,
1887 user_api_key_dict=user_api_key_dict,
1888 response=response,
1889 request_headers=dict(request.headers),
1890 )
1891 if callback_headers:
1892 custom_headers.update(callback_headers)
1894 return custom_headers
1896 async def common_processing_pre_call_logic(
1897 self,
1898 request: Request,
1899 general_settings: dict,
1900 user_api_key_dict: UserAPIKeyAuth,
1901 proxy_logging_obj: ProxyLogging,
1902 proxy_config: ProxyConfig,
1903 route_type: Literal[
1904 "acompletion",
1905 "aembedding",
1906 "aresponses",
1907 "_arealtime",
1908 "_aresponses_websocket",
1909 "acreate_realtime_client_secret",
1910 "arealtime_calls",
1911 "aget_responses",
1912 "adelete_responses",
1913 "acancel_responses",
1914 "acompact_responses",
1915 "acreate_batch",
1916 "aretrieve_batch",
1917 "alist_batches",
1918 "acancel_batch",
1919 "afile_content",
1920 "afile_retrieve",
1921 "afile_delete",
1922 "atext_completion",
1923 "acreate_fine_tuning_job",
1924 "acancel_fine_tuning_job",
1925 "alist_fine_tuning_jobs",
1926 "aretrieve_fine_tuning_job",
1927 "alist_input_items",
1928 "aimage_edit",
1929 "agenerate_content",
1930 "agenerate_content_stream",
1931 "allm_passthrough_route",
1932 "avector_store_search",
1933 "avector_store_create",
1934 "avector_store_retrieve",
1935 "avector_store_list",
1936 "avector_store_update",
1937 "avector_store_delete",
1938 "avector_store_file_create",
1939 "avector_store_file_list",
1940 "avector_store_file_retrieve",
1941 "avector_store_file_content",
1942 "avector_store_file_update",
1943 "avector_store_file_delete",
1944 "aocr",
1945 "asearch",
1946 "adecisions",
1947 "avideo_generation",
1948 "avideo_list",
1949 "avideo_status",
1950 "avideo_content",
1951 "avideo_remix",
1952 "avideo_create_character",
1953 "avideo_get_character",
1954 "avideo_edit",
1955 "avideo_extension",
1956 "acreate_container",
1957 "alist_containers",
1958 "aingest",
1959 "aretrieve_container",
1960 "adelete_container",
1961 "aupload_container_file",
1962 "alist_container_files",
1963 "aretrieve_container_file",
1964 "adelete_container_file",
1965 "aretrieve_container_file_content",
1966 "acreate_skill",
1967 "alist_skills",
1968 "aget_skill",
1969 "adelete_skill",
1970 "anthropic_messages",
1971 "acreate_interaction",
1972 "aget_interaction",
1973 "adelete_interaction",
1974 "acancel_interaction",
1975 "acreate_agent",
1976 "alist_agents",
1977 "aget_agent",
1978 "adelete_agent",
1979 "alist_agent_versions",
1980 "asend_message",
1981 "call_mcp_tool",
1982 "acreate_eval",
1983 "alist_evals",
1984 "aget_eval",
1985 "aupdate_eval",
1986 "adelete_eval",
1987 "acancel_eval",
1988 "acreate_run",
1989 "alist_runs",
1990 "aget_run",
1991 "acancel_run",
1992 "adelete_run",
1993 "apply_guardrail",
1994 ],
1995 version: str | None = None,
1996 user_model: str | None = None,
1997 user_temperature: float | None = None,
1998 user_request_timeout: float | None = None,
1999 user_max_tokens: int | None = None,
2000 user_api_base: str | None = None,
2001 model: str | None = None,
2002 llm_router: Router | None = None,
2003 rate_limited_model: str | None = None,
2004 skip_guardrails: bool = False,
2005 ) -> tuple[dict, LiteLLMLoggingObj]:
2006 start_time: Final = datetime.now() # start before calling guardrail hooks
2008 requested_model: Final = self.data.get("model")
2009 if requested_model is not None and not isinstance(requested_model, str):
2010 raise ProxyException(
2011 message="'model' must be a string.",
2012 type=ProxyErrorTypes.bad_request_error,
2013 param="model",
2014 code=status.HTTP_400_BAD_REQUEST,
2015 )
2016 self.data = await add_litellm_data_to_request(
2017 data=self.data,
2018 request=request,
2019 general_settings=general_settings,
2020 user_api_key_dict=user_api_key_dict,
2021 version=version,
2022 proxy_config=proxy_config,
2023 )
2024 if not general_settings.get("expose_fallback_errors_to_caller"): 2024 ↛ 2026line 2024 didn't jump to line 2026 because the condition on line 2024 was always true
2025 self.data.pop("include_fallback_errors", None)
2026 if route_type in {"aresponses", "_aresponses_websocket"}:
2027 await _authorize_response_file_search_vector_stores(
2028 data=self.data,
2029 user_api_key_dict=user_api_key_dict,
2030 )
2031 # Calculate request queue time after add_litellm_data_to_request
2032 # which sets arrival_time in proxy_server_request. Ends at start_time
2033 # (not a freshly captured time.time() here) so this window is exactly
2034 # [arrival_time, start_time], with zero overlap with the
2035 # litellm_request_total_latency_metric window of [start_time, end_time] --
2036 # otherwise the few lines of add_litellm_data_to_request's own work would
2037 # be double-counted across both metrics.
2038 proxy_server_request: Final = self.data.get("proxy_server_request", {})
2039 arrival_time: Final = proxy_server_request.get("arrival_time")
2040 queue_time_seconds = None
2041 if arrival_time is not None: 2041 ↛ 2045line 2041 didn't jump to line 2045 because the condition on line 2041 was always true
2042 queue_time_seconds = start_time.timestamp() - arrival_time
2044 # Store queue time in metadata after add_litellm_data_to_request to ensure it's preserved
2045 if queue_time_seconds is not None: 2045 ↛ 2055line 2045 didn't jump to line 2055 because the condition on line 2045 was always true
2046 from litellm.proxy.litellm_pre_call_utils import _get_metadata_variable_name
2048 _metadata_variable_name: Final = _get_metadata_variable_name(request)
2049 if _metadata_variable_name not in self.data: 2049 ↛ 2050line 2049 didn't jump to line 2050 because the condition on line 2049 was never true
2050 self.data[_metadata_variable_name] = {}
2051 if not isinstance(self.data[_metadata_variable_name], dict): 2051 ↛ 2052line 2051 didn't jump to line 2052 because the condition on line 2051 was never true
2052 self.data[_metadata_variable_name] = {}
2053 self.data[_metadata_variable_name]["queue_time_seconds"] = queue_time_seconds
2055 if isinstance(model, str):
2056 reject_url_valued_destination("model", model)
2058 self.data["model"] = (
2059 general_settings.get("completion_model", None) # server default
2060 or user_model # model name passed via cli args
2061 or model # for azure deployments
2062 or self.data.get("model", None) # default passed in http request
2063 )
2065 # override with user settings, these are params passed via cli
2066 if user_temperature: 2066 ↛ 2067line 2066 didn't jump to line 2067 because the condition on line 2066 was never true
2067 self.data["temperature"] = user_temperature
2068 if user_request_timeout: 2068 ↛ 2069line 2068 didn't jump to line 2069 because the condition on line 2068 was never true
2069 self.data["request_timeout"] = user_request_timeout
2070 if user_max_tokens: 2070 ↛ 2071line 2070 didn't jump to line 2071 because the condition on line 2070 was never true
2071 self.data["max_tokens"] = user_max_tokens
2072 if user_api_base: 2072 ↛ 2073line 2072 didn't jump to line 2073 because the condition on line 2072 was never true
2073 self.data["api_base"] = user_api_base
2075 ### MODEL ALIAS MAPPING ###
2076 # check if model name in model alias map
2077 # get the actual model name
2078 if isinstance(self.data["model"], str) and self.data["model"] in litellm.model_alias_map: 2078 ↛ 2079line 2078 didn't jump to line 2079 because the condition on line 2078 was never true
2079 self.data["model"] = litellm.model_alias_map[self.data["model"]]
2081 # Check key-specific aliases
2082 if ( 2082 ↛ 2088line 2082 didn't jump to line 2088 because the condition on line 2082 was never true
2083 isinstance(self.data["model"], str)
2084 and user_api_key_dict.aliases
2085 and isinstance(user_api_key_dict.aliases, dict)
2086 and self.data["model"] in user_api_key_dict.aliases
2087 ):
2088 self.data["model"] = user_api_key_dict.aliases[self.data["model"]]
2090 # Apply hierarchical router_settings (Key > Team)
2091 # Global router_settings are already on the Router object itself.
2092 # This sits with the other alias rewrites, and ahead of the guardrail
2093 # merge and the pre-call hooks, so everything that keys off the model
2094 # group -- model-level guardrails, per-model budgets and rate limits,
2095 # the logging object -- sees the group that will actually serve.
2096 if llm_router is not None and proxy_config is not None:
2097 from litellm.proxy.proxy_server import prisma_client
2099 router_settings: Final = await proxy_config._get_hierarchical_router_settings(
2100 user_api_key_dict=user_api_key_dict,
2101 prisma_client=prisma_client,
2102 proxy_logging_obj=proxy_logging_obj,
2103 )
2105 # If router_settings found (from key or team), apply them
2106 # Pass settings as per-request overrides instead of creating a new Router
2107 # This avoids expensive Router instantiation on each request
2108 if router_settings is not None: 2108 ↛ 2109line 2108 didn't jump to line 2109 because the condition on line 2108 was never true
2109 self.data["router_settings_override"] = router_settings
2110 try:
2111 self.data["_router_weights"] = validate_router_weights(router_settings.get("weights"))
2112 except ValidationError:
2113 self.data["_router_weights"] = None
2114 verbose_proxy_logger.warning(
2115 "Ignoring invalid saved router weights; update team/key router_settings"
2116 )
2117 alias_target: Final = await _resolve_per_request_model_group_alias(
2118 requested_model=self.data.get("model"),
2119 router_settings=router_settings,
2120 user_api_key_dict=user_api_key_dict,
2121 llm_router=llm_router,
2122 )
2123 if alias_target is not None:
2124 self.data["model"] = alias_target
2126 self.data["litellm_call_id"] = resolve_litellm_call_id(request.headers.get("x-litellm-call-id"))
2127 DDSpanTagger.tag_call_id(self.data.get("litellm_call_id"))
2128 DDSpanTagger.tag_request(
2129 user_api_key_dict=user_api_key_dict,
2130 requested_model=self.data.get("model"),
2131 )
2133 ### AUTO STREAM USAGE TRACKING ###
2134 self.data.update(
2135 _stream_usage_tracking_updates(
2136 data=self.data,
2137 general_settings=general_settings,
2138 route_type=route_type,
2139 supports_stream_options=lambda: _model_deployments_support_stream_options(
2140 model=self.data.get("model"),
2141 llm_router=llm_router,
2142 team_id=user_api_key_dict.team_id,
2143 ),
2144 )
2145 )
2146 ### CALL HOOKS ### - modify/reject incoming data before calling the model
2148 ## LOGGING OBJECT ## - initialize logging object for logging success/failure events for call
2149 ## IMPORTANT Note: - initialize this before running pre-call checks. Ensures we log rejected requests to langfuse.
2150 logging_obj, self.data = litellm.utils.function_setup(
2151 original_function=route_type,
2152 rules_obj=litellm.utils.Rules(),
2153 start_time=start_time,
2154 **self.data,
2155 )
2157 self.data["litellm_logging_obj"] = logging_obj
2159 # Merge model-level guardrails before pre_call_hook so DB/UI-configured
2160 # guardrails actually execute on pre_call. Without this, guardrails set
2161 # via litellm_params.guardrails are only honored on post_call paths
2162 # (#29652, partial fix in #23774 covered non-streaming post_call only).
2163 # trust_client_model_info=False on pre_call: route_request hasn't run
2164 # and add_litellm_data_to_request preserves client-supplied
2165 # model_info when allow_client_pricing_override is set, so a caller
2166 # could otherwise spoof an unguarded model_info.id while requesting
2167 # a guarded alias and bypass guardrails (veria-ai HIGH on #29654).
2168 merged_for_requested: Final = (
2169 self.data
2170 if rate_limited_model is None
2171 else _check_and_merge_model_level_guardrails(
2172 data=self.data, llm_router=llm_router, trust_client_model_info=False, model_alias=rate_limited_model
2173 )
2174 )
2175 self.data = _check_and_merge_model_level_guardrails(
2176 data=merged_for_requested,
2177 llm_router=llm_router,
2178 trust_client_model_info=False,
2179 )
2181 # An auto router with its own compression policy is authoritative for this
2182 # request: suppress every other compression guardrail and arm whichever one
2183 # the policy names for the model call, before those guardrails get a chance
2184 # to run below.
2185 await _arm_auto_router_compression(data=self.data, llm_router=llm_router)
2187 if self._tags_before_guardrails is None: 2187 ↛ 2189line 2187 didn't jump to line 2189 because the condition on line 2187 was always true
2188 self._tags_before_guardrails = frozenset(get_tags_from_request_body(request_body=self.data))
2189 self.data = await proxy_logging_obj.pre_call_hook(
2190 user_api_key_dict=user_api_key_dict,
2191 data=self.data,
2192 call_type=route_type,
2193 skip_guardrails=skip_guardrails,
2194 )
2195 await _enforce_guardrail_added_tag_budgets(
2196 data=self.data,
2197 tags_before_guardrails=self._tags_before_guardrails,
2198 route=get_request_route(request=request),
2199 llm_router=llm_router,
2200 user_api_key_dict=user_api_key_dict,
2201 proxy_logging_obj=proxy_logging_obj,
2202 )
2203 if route_type == "aget_responses":
2204 attach_post_call_pipelines_to_retrieval(
2205 data=self.data,
2206 user_api_key_dict=user_api_key_dict,
2207 llm_router=llm_router,
2208 )
2210 # Refresh AFTER pre_call_hook: guardrails (e.g. Presidio PII masking) may
2211 # have mutated `self.data` in place, and the audit-trail snapshot taken in
2212 # add_litellm_data_to_request predates that mutation.
2213 refresh_proxy_server_request_body_snapshot(self.data, guardrails_applied=True)
2214 verbose_proxy_logger.debug("receiving data: %s", self.data)
2216 if "messages" in self.data and self.data["messages"]:
2217 logging_obj.update_messages(self.data["messages"])
2219 return self.data, logging_obj
2221 async def _pre_call_with_fallbacks(
2222 self,
2223 request: Request,
2224 general_settings: dict,
2225 proxy_logging_obj: ProxyLogging,
2226 user_api_key_dict: UserAPIKeyAuth,
2227 version: str | None,
2228 proxy_config: ProxyConfig,
2229 user_model: str | None,
2230 user_temperature: float | None,
2231 user_request_timeout: float | None,
2232 user_max_tokens: int | None,
2233 user_api_base: str | None,
2234 model: str | None,
2235 route_type: str,
2236 llm_router: Router | None,
2237 ) -> tuple[dict, LiteLLMLoggingObj]:
2238 from litellm.proxy.common_utils.proxy_rate_limit_error import ProxyRateLimitError
2240 configured_fallbacks: Final = (
2241 self._configured_fallbacks(llm_router=llm_router, user_api_key_dict=user_api_key_dict)
2242 if llm_router is not None
2243 else None
2244 )
2245 pristine: Final = independent_snapshot(self.data) if configured_fallbacks else None
2247 try:
2248 return await self.common_processing_pre_call_logic(
2249 request=request,
2250 general_settings=general_settings,
2251 proxy_logging_obj=proxy_logging_obj,
2252 user_api_key_dict=user_api_key_dict,
2253 version=version,
2254 proxy_config=proxy_config,
2255 user_model=user_model,
2256 user_temperature=user_temperature,
2257 user_request_timeout=user_request_timeout,
2258 user_max_tokens=user_max_tokens,
2259 user_api_base=user_api_base,
2260 model=model,
2261 route_type=route_type,
2262 llm_router=llm_router,
2263 )
2264 except ProxyRateLimitError as original_exc:
2265 rate_limited_data: Final = self.data
2266 original_model: Final = rate_limited_data.get("model")
2267 if (
2268 pristine is None
2269 or not configured_fallbacks
2270 or rate_limited_data.get("disable_fallbacks")
2271 or not isinstance(original_model, str)
2272 ):
2273 raise
2275 fallback_models: Final = self._resolve_fallback_models(
2276 model=original_model,
2277 fallbacks=configured_fallbacks,
2278 )
2279 if not fallback_models:
2280 raise
2282 verbose_proxy_logger.info(
2283 "Local rate limit hit for model=%s, attempting fallbacks: %s",
2284 original_model,
2285 fallback_models,
2286 )
2287 try:
2288 for fallback_model in fallback_models:
2289 if fallback_model == original_model:
2290 continue
2291 self.data = independent_snapshot(pristine)
2292 self.data["model"] = fallback_model
2293 try:
2294 return await self.common_processing_pre_call_logic(
2295 request=request,
2296 general_settings=general_settings,
2297 proxy_logging_obj=proxy_logging_obj,
2298 user_api_key_dict=user_api_key_dict,
2299 version=version,
2300 proxy_config=proxy_config,
2301 user_model=user_model,
2302 user_temperature=user_temperature,
2303 user_request_timeout=user_request_timeout,
2304 user_max_tokens=user_max_tokens,
2305 user_api_base=user_api_base,
2306 model=fallback_model,
2307 route_type=route_type,
2308 llm_router=llm_router,
2309 rate_limited_model=original_model,
2310 )
2311 except ProxyRateLimitError:
2312 continue
2313 except BaseException:
2314 self.data = rate_limited_data
2315 raise
2317 self.data = rate_limited_data
2318 raise original_exc
2320 @staticmethod
2321 def _configured_fallbacks(llm_router: Router, user_api_key_dict: UserAPIKeyAuth) -> list | None:
2322 key_router_settings: Final = user_api_key_dict.router_settings
2323 key_fallbacks: Final = key_router_settings.get("fallbacks") if isinstance(key_router_settings, dict) else None
2324 fallbacks: Final = key_fallbacks if key_fallbacks is not None else llm_router.fallbacks
2325 return fallbacks if isinstance(fallbacks, list) and fallbacks else None
2327 @staticmethod
2328 def _resolve_fallback_models(model: str, fallbacks: list) -> list[str] | None:
2329 from litellm.router_utils.fallback_event_handlers import get_fallback_model_group
2331 fallback_model_group, generic_fallback_idx = get_fallback_model_group(
2332 fallbacks=fallbacks,
2333 model_group=model,
2334 )
2335 if fallback_model_group is not None:
2336 return fallback_model_group
2337 return fallbacks[generic_fallback_idx]["*"] if generic_fallback_idx is not None else None
2339 @staticmethod
2340 def _get_model_id_from_response(hidden_params: Mapping[str, object], data: Mapping[str, object]) -> str:
2341 """Extract model_id from hidden_params with fallback to litellm_metadata."""
2342 model_id = hidden_params.get("model_id", None) or ""
2343 if not model_id: 2343 ↛ 2347line 2343 didn't jump to line 2347 because the condition on line 2343 was always true
2344 litellm_metadata: Final = data.get("litellm_metadata")
2345 model_info: Final = litellm_metadata.get("model_info") if isinstance(litellm_metadata, Mapping) else None
2346 model_id = (model_info.get("id") or "") if isinstance(model_info, Mapping) else ""
2347 return str(model_id) if model_id else ""
2349 def _stream_response_headers(
2350 self,
2351 *,
2352 hidden_params: Mapping[str, object],
2353 user_api_key_dict: UserAPIKeyAuth,
2354 logging_obj: LiteLLMLoggingObj,
2355 version: str | None,
2356 callback_headers: Mapping[str, str],
2357 ) -> Mapping[str, str]:
2358 """The streaming response headers describing `hidden_params`' deployment."""
2359 return MappingProxyType(
2360 {
2361 **ProxyBaseLLMRequestProcessing.get_custom_headers(
2362 user_api_key_dict=user_api_key_dict,
2363 call_id=logging_obj.litellm_call_id,
2364 model_id=self._get_model_id_from_response(hidden_params, self.data),
2365 cache_key=hidden_params.get("cache_key") or "",
2366 api_base=hidden_params.get("api_base") or "",
2367 version=version,
2368 response_cost=hidden_params.get("response_cost") or "",
2369 model_region=getattr(user_api_key_dict, "allowed_model_region", ""),
2370 fastest_response_batch_completion=hidden_params.get("fastest_response_batch_completion"),
2371 request_data=self.data,
2372 hidden_params=hidden_params,
2373 litellm_logging_obj=logging_obj,
2374 **(hidden_params.get("additional_headers") or MappingProxyType({})),
2375 ),
2376 **callback_headers,
2377 }
2378 )
2380 @staticmethod
2381 def _get_deployment_model_name(
2382 litellm_logging_obj: LiteLLMLoggingObj | None,
2383 ) -> str | None:
2384 """Extract the underlying deployment model string (e.g. ``azure/gpt-4o``).
2386 The router rewrites the response ``model`` field to the model-group alias
2387 the client requested, so neither the response body nor the existing
2388 headers expose the concrete deployment model. The router records it under
2389 ``litellm_params`` metadata as ``deployment``, so read it back from there.
2390 """
2391 litellm_params: Final = getattr(litellm_logging_obj, "litellm_params", None)
2392 if not isinstance(litellm_params, dict):
2393 return None
2394 for key in ("litellm_metadata", "metadata"):
2395 metadata = litellm_params.get(key, {}) or {}
2396 deployment = metadata.get("deployment")
2397 if deployment: 2397 ↛ 2398line 2397 didn't jump to line 2398 because the condition on line 2397 was never true
2398 return deployment
2399 return None
2401 @staticmethod
2402 def _response_cost_from_logging_obj(
2403 *,
2404 response: Any,
2405 logging_obj: LiteLLMLoggingObj,
2406 ) -> float | str:
2407 """
2408 Recover the response cost when the response never recorded one in its
2409 ``_hidden_params``: Anthropic /v1/messages returns a TypedDict that cannot
2410 hold the attribute at all, and Google :generateContent carries
2411 ``_hidden_params`` but no synchronously-populated ``response_cost``. In both
2412 cases the cost is read back from the logging object instead, recomputing from
2413 the same calculator only when it has not been stored yet.
2414 """
2415 stored_cost: Final = logging_obj.model_call_details.get("response_cost")
2416 if isinstance(stored_cost, (int, float)):
2417 return float(stored_cost)
2418 recomputed_cost: Final = logging_obj._response_cost_calculator(result=response)
2419 return recomputed_cost if isinstance(recomputed_cost, (int, float)) else ""
2421 def _debug_log_request_payload(self) -> None:
2422 """Log request payload at DEBUG level, truncating if too large."""
2423 if not verbose_proxy_logger.isEnabledFor(logging.DEBUG): 2423 ↛ 2425line 2423 didn't jump to line 2425 because the condition on line 2423 was always true
2424 return
2425 _payload_str: Final = json.dumps(self.data, default=str)
2426 if len(_payload_str) > MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG:
2427 verbose_proxy_logger.debug(
2428 "Request received by LiteLLM: payload too large to log (%d bytes, limit %d). Keys: %s",
2429 len(_payload_str),
2430 MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG,
2431 (list(self.data.keys()) if isinstance(self.data, dict) else type(self.data).__name__),
2432 )
2433 else:
2434 verbose_proxy_logger.debug(
2435 "Request received by LiteLLM:\n%s",
2436 _payload_str,
2437 )
2439 async def base_process_llm_request(
2440 self,
2441 request: Request,
2442 fastapi_response: Response,
2443 user_api_key_dict: UserAPIKeyAuth,
2444 route_type: ProxyRouteType,
2445 proxy_logging_obj: ProxyLogging,
2446 general_settings: dict[str, object],
2447 proxy_config: ProxyConfig,
2448 select_data_generator: Callable[..., object] | None = None,
2449 llm_router: Router | None = None,
2450 model: str | None = None,
2451 user_model: str | None = None,
2452 user_temperature: float | None = None,
2453 user_request_timeout: float | None = None,
2454 user_max_tokens: int | None = None,
2455 user_api_base: str | None = None,
2456 version: str | None = None,
2457 is_streaming_request: bool | None = False,
2458 contents: list[object] | None = None,
2459 skip_pre_call_logic: bool = False,
2460 ) -> Any:
2461 """Run the request, sending SSE keepalives while the upstream is still silent.
2463 Everything below this point, the upstream call included, happens before the
2464 proxy can write a byte, so a slow time-to-first-token leaves the response
2465 idle. See ``open_sse_before_first_byte``; unwrapped unless an operator sets
2466 ``litellm_settings.sse_keepalive_ping_interval_seconds``.
2467 """
2469 async def _audit_late_failure(exc: Exception) -> HTTPException | None:
2470 # Once a keepalive is on the wire this can no longer raise, so the
2471 # caller's `except` never runs its own post_call_failure_hook.
2472 return await proxy_logging_obj.post_call_failure_hook(
2473 user_api_key_dict=user_api_key_dict,
2474 original_exception=exc,
2475 request_data=self.data,
2476 )
2478 return await open_sse_before_first_byte(
2479 self._process_llm_request(
2480 request=request,
2481 fastapi_response=fastapi_response,
2482 user_api_key_dict=user_api_key_dict,
2483 route_type=route_type,
2484 proxy_logging_obj=proxy_logging_obj,
2485 general_settings=general_settings,
2486 proxy_config=proxy_config,
2487 select_data_generator=select_data_generator,
2488 llm_router=llm_router,
2489 model=model,
2490 user_model=user_model,
2491 user_temperature=user_temperature,
2492 user_request_timeout=user_request_timeout,
2493 user_max_tokens=user_max_tokens,
2494 user_api_base=user_api_base,
2495 version=version,
2496 is_streaming_request=is_streaming_request,
2497 contents=contents,
2498 skip_pre_call_logic=skip_pre_call_logic,
2499 ),
2500 ping_interval_seconds=ttft_keepalive_interval(self.data, llm_router),
2501 on_late_failure=_audit_late_failure,
2502 )
2504 async def _process_llm_request(
2505 self,
2506 request: Request,
2507 fastapi_response: Response,
2508 user_api_key_dict: UserAPIKeyAuth,
2509 route_type: ProxyRouteType,
2510 proxy_logging_obj: ProxyLogging,
2511 general_settings: dict[str, object],
2512 proxy_config: ProxyConfig,
2513 select_data_generator: Callable[..., object] | None = None,
2514 llm_router: Router | None = None,
2515 model: str | None = None,
2516 user_model: str | None = None,
2517 user_temperature: float | None = None,
2518 user_request_timeout: float | None = None,
2519 user_max_tokens: int | None = None,
2520 user_api_base: str | None = None,
2521 version: str | None = None,
2522 is_streaming_request: bool | None = False,
2523 contents: list[object] | None = None, # Add contents parameter
2524 skip_pre_call_logic: bool = False,
2525 ) -> Any:
2526 """
2527 Common request processing logic for both chat completions and responses API endpoints
2528 """
2529 client_model: Final = get_client_requested_model(request) or self.data.get("model")
2530 requested_model_from_client: Final[str | None] = client_model if isinstance(client_model, str) else None
2531 self._debug_log_request_payload()
2533 if skip_pre_call_logic: 2533 ↛ 2534line 2533 didn't jump to line 2534 because the condition on line 2533 was never true
2534 logging_obj = self.data.get("litellm_logging_obj")
2535 if logging_obj is None:
2536 raise ValueError(
2537 "skip_pre_call_logic=True requires litellm_logging_obj to be set in data. "
2538 "Ensure common_processing_pre_call_logic was called before using this parameter."
2539 )
2540 else:
2541 self.data, logging_obj = await self._pre_call_with_fallbacks(
2542 request=request,
2543 general_settings=general_settings,
2544 proxy_logging_obj=proxy_logging_obj,
2545 user_api_key_dict=user_api_key_dict,
2546 version=version,
2547 proxy_config=proxy_config,
2548 user_model=user_model,
2549 user_temperature=user_temperature,
2550 user_request_timeout=user_request_timeout,
2551 user_max_tokens=user_max_tokens,
2552 user_api_base=user_api_base,
2553 model=model,
2554 route_type=route_type,
2555 llm_router=llm_router,
2556 )
2558 # Defer async logging when post-call guardrails are configured so the
2559 # StandardLoggingPayload is built after guardrails write to metadata.
2560 # Cache the result to avoid scanning litellm.callbacks twice.
2561 _post_call_guardrails_active: Final = self._has_post_call_guardrails()
2563 # Non-streaming: defer the create_task in wrapper_async so the
2564 # SLP is built after guardrails write to metadata. Streaming
2565 # uses a separate closure mechanism (see below).
2566 #
2567 # Edge case: if _is_streaming_request is False but the response
2568 # turns out to be a CustomStreamWrapper (rare provider behavior),
2569 # wrapper_async exits early before the _defer_async_logging block
2570 # so _enqueue_deferred_logging is never stored — the finally
2571 # block is a no-op. The CSW path handles this correctly via
2572 # _on_deferred_stream_complete, which fires its own logging.
2573 if _post_call_guardrails_active and not self._is_streaming_request( 2573 ↛ 2576line 2573 didn't jump to line 2576 because the condition on line 2573 was never true
2574 data=self.data, is_streaming_request=is_streaming_request
2575 ):
2576 logging_obj._defer_async_logging = True
2578 tasks: Final = []
2579 # Start the moderation check (during_call_hook) as early as possible
2580 # This gives it a head start to mask/validate input while the proxy handles routing
2581 tasks.append(
2582 asyncio.create_task(
2583 proxy_logging_obj.during_call_hook(
2584 data=self.data,
2585 user_api_key_dict=user_api_key_dict,
2586 call_type=route_type,
2587 )
2588 )
2589 )
2591 # Pass contents if provided
2592 if contents: 2592 ↛ 2593line 2592 didn't jump to line 2593 because the condition on line 2592 was never true
2593 self.data["contents"] = contents
2595 ### ROUTE THE REQUEST ###
2596 # Do not change this - it should be a constant time fetch - ALWAYS
2597 llm_call: Final = await route_request(
2598 data=self.data,
2599 route_type=route_type,
2600 llm_router=llm_router,
2601 user_model=user_model,
2602 user_api_key_dict=user_api_key_dict,
2603 )
2604 llm_call_task: Final = asyncio.create_task(with_proxy_compaction_executor(llm_call, request))
2605 tasks.append(llm_call_task)
2607 llm_responses: Final = asyncio.gather(*tasks) # run the moderation check in parallel to the actual llm api call
2609 try:
2610 if general_settings.get("cancel_on_disconnect", False): 2610 ↛ 2611line 2610 didn't jump to line 2611 because the condition on line 2610 was never true
2611 responses = await _await_llm_call_cancelling_on_disconnect(request, llm_responses)
2612 else:
2613 responses = await llm_responses
2614 finally:
2615 await _cancel_pending_gather_tasks(tasks)
2617 response = responses[1]
2619 _exception_raised = False
2620 try:
2621 hidden_params = get_hidden_params_dict(response)
2622 model_id: Final = self._get_model_id_from_response(hidden_params, self.data)
2624 cache_key, api_base, response_cost = (
2625 hidden_params.get("cache_key", None) or "",
2626 hidden_params.get("api_base", None) or "",
2627 hidden_params.get("response_cost", None) or "",
2628 )
2629 fastest_response_batch_completion, additional_headers = (
2630 hidden_params.get("fastest_response_batch_completion", None),
2631 hidden_params.get("additional_headers", {}) or {},
2632 )
2634 # Post Call Processing
2635 if llm_router is not None: 2635 ↛ 2637line 2635 didn't jump to line 2637 because the condition on line 2635 was always true
2636 self.data["deployment"] = llm_router.get_deployment(model_id=model_id)
2637 asyncio.create_task(
2638 proxy_logging_obj.update_request_status(
2639 litellm_call_id=self.data.get("litellm_call_id", ""),
2640 status="success",
2641 )
2642 )
2643 if self._is_streaming_request( 2643 ↛ 2646line 2643 didn't jump to line 2646 because the condition on line 2643 was never true
2644 data=self.data, is_streaming_request=is_streaming_request
2645 ) or self._is_streaming_response(response): # use generate_responses to stream responses
2646 selected_data_generator: AsyncGenerator[str, None] | None = None
2647 # Call response headers hook for streaming success
2648 stream_callback_headers: Final = await proxy_logging_obj.post_call_response_headers_hook(
2649 data=self.data,
2650 user_api_key_dict=user_api_key_dict,
2651 response=response,
2652 request_headers=dict(request.headers),
2653 )
2654 custom_headers: Final = self._stream_response_headers(
2655 hidden_params=hidden_params,
2656 user_api_key_dict=user_api_key_dict,
2657 logging_obj=logging_obj,
2658 version=version,
2659 callback_headers=stream_callback_headers or MappingProxyType({}),
2660 )
2662 async def refresh_stream_headers() -> Mapping[str, str]:
2663 """`custom_headers` rebuilt for whichever deployment served the stream."""
2664 return self._stream_response_headers(
2665 hidden_params=(
2666 get_hidden_params_dict(response)
2667 if getattr(response, "fallback_headers_adopted", False)
2668 else hidden_params
2669 ),
2670 user_api_key_dict=user_api_key_dict,
2671 logging_obj=logging_obj,
2672 version=version,
2673 callback_headers=stream_callback_headers or MappingProxyType({}),
2674 )
2676 # Preserve the original client-requested model (pre-alias mapping) for downstream
2677 # streaming generators. Pre-call processing can rewrite `self.data["model"]` for
2678 # aliasing/routing, but the OpenAI-compatible response `model` field should reflect
2679 # what the client sent.
2680 if requested_model_from_client:
2681 self.data["_litellm_client_requested_model"] = requested_model_from_client
2683 if _post_call_guardrails_active:
2684 self._arm_deferred_stream_dispatch(
2685 response=response,
2686 route_type=route_type,
2687 user_api_key_dict=user_api_key_dict,
2688 logging_obj=logging_obj,
2689 )
2691 if route_type == "allm_passthrough_route":
2692 upstream_response_headers: Final = getattr(response, "headers", None)
2693 streaming_headers: Final = (
2694 ProxyBaseLLMRequestProcessing._merge_passthrough_streaming_headers(
2695 response_headers=upstream_response_headers,
2696 custom_headers=custom_headers,
2697 )
2698 if upstream_response_headers is not None
2699 else custom_headers
2700 )
2702 # Check if response is an async generator
2703 if self._is_streaming_response(response):
2704 if asyncio.iscoroutine(response):
2705 generator = await response
2706 else:
2707 generator = response
2709 if (
2710 self._has_post_call_guardrails_for_passthrough()
2711 and self._passthrough_endpoint_has_stream_guardrail_handler()
2712 ):
2713 body_bytes: Final = b"".join([chunk async for chunk in generator])
2714 modified_bytes: Final = await self._handle_event_stream_allm_passthrough_route(
2715 body_bytes=body_bytes,
2716 proxy_logging_obj=proxy_logging_obj,
2717 user_api_key_dict=user_api_key_dict,
2718 )
2719 response_headers: Final = {
2720 k: v for k, v in custom_headers.items() if k.lower() != "content-length"
2721 }
2722 return Response(
2723 content=modified_bytes,
2724 status_code=status.HTTP_200_OK,
2725 media_type=self._passthrough_event_stream_media_type(),
2726 headers=response_headers,
2727 )
2729 # For passthrough routes, stream directly without error parsing
2730 # since we're dealing with raw binary data (e.g., AWS event streams)
2731 return _UpstreamClosingStreamingResponse(
2732 content=generator, # pyright: ignore[reportArgumentType] # generator-configured StreamingResponse
2733 status_code=getattr(response, "status_code", status.HTTP_200_OK),
2734 media_type=self._passthrough_event_stream_media_type(),
2735 headers=streaming_headers,
2736 )
2737 else:
2738 _early = await self._handle_non_streaming_allm_passthrough_route(
2739 response=response,
2740 proxy_logging_obj=proxy_logging_obj,
2741 user_api_key_dict=user_api_key_dict,
2742 custom_headers=custom_headers,
2743 request_headers=dict(request.headers),
2744 )
2745 if _early is not None:
2746 return _early
2747 return StreamingResponse(
2748 content=response.aiter_bytes(),
2749 status_code=response.status_code,
2750 headers=streaming_headers,
2751 )
2752 elif route_type == "anthropic_messages":
2753 # Check if response is actually a streaming response (async generator)
2754 # Non-streaming responses (dict) should be returned directly
2755 # This handles cases like websearch_interception agentic loop
2756 # which returns a non-streaming dict even for streaming requests
2757 if self._is_streaming_response(response):
2758 self._arm_detached_stream_failure_hook(
2759 logging_obj=logging_obj,
2760 user_api_key_dict=user_api_key_dict,
2761 proxy_logging_obj=proxy_logging_obj,
2762 )
2763 selected_data_generator = ProxyBaseLLMRequestProcessing.async_sse_data_generator(
2764 response=response,
2765 user_api_key_dict=user_api_key_dict,
2766 request_data=self.data,
2767 proxy_logging_obj=proxy_logging_obj,
2768 request=request,
2769 restamp_model=(
2770 None if _should_return_raw_model_name(self.data) else requested_model_from_client
2771 ),
2772 )
2773 selected_data_generator = wrap_sse_stream_with_keepalive_pings(
2774 stream=selected_data_generator,
2775 ping_interval_seconds=litellm.anthropic_sse_ping_interval_seconds,
2776 )
2777 # Non-streaming response - fall through to normal response handling
2778 elif select_data_generator:
2779 selected_data_generator = select_data_generator(
2780 response=response,
2781 user_api_key_dict=user_api_key_dict,
2782 request_data=self.data,
2783 request=request,
2784 )
2785 if route_type == "aresponses":
2786 # Streaming /v1/responses returns here without
2787 # reaching the non-streaming ownership tail below.
2788 # Wrap the SSE generator so container ownership is
2789 # written once the upstream iterator finishes
2790 # assembling ``completed_response`` — otherwise
2791 # code-interpreter containers created during the
2792 # stream stay unregistered and follow-up file API
2793 # calls 403. Covers the background-polling path
2794 # too, which loops ``body_iterator`` end-to-end.
2795 selected_data_generator = (
2796 ProxyBaseLLMRequestProcessing._wrap_responses_stream_for_container_ownership(
2797 original_stream_response=response,
2798 wrapped_generator=selected_data_generator,
2799 user_api_key_dict=user_api_key_dict,
2800 )
2801 )
2802 if selected_data_generator is not None:
2803 return await create_response(
2804 generator=selected_data_generator,
2805 media_type="text/event-stream",
2806 headers=custom_headers,
2807 request=request,
2808 refresh_headers=refresh_stream_headers,
2809 general_settings=general_settings,
2810 )
2812 ### CALL HOOKS ### - modify outgoing data
2813 # If we reach here with a streaming closure still set, it means
2814 # no early-return route consumed the CSW (hypothetical fallthrough).
2815 # Clear the closure so guardrails run inline as before — this
2816 # preserves blocking behavior and avoids double invocation.
2817 if getattr(logging_obj, "_on_deferred_stream_complete", None): 2817 ↛ 2818line 2817 didn't jump to line 2818 because the condition on line 2817 was never true
2818 logging_obj._on_deferred_stream_complete = None
2820 if route_type == "allm_passthrough_route": 2820 ↛ 2821line 2820 didn't jump to line 2821 because the condition on line 2820 was never true
2821 _non_streaming_custom_headers: Final = ProxyBaseLLMRequestProcessing.get_custom_headers(
2822 user_api_key_dict=user_api_key_dict,
2823 call_id=logging_obj.litellm_call_id,
2824 model_id=model_id,
2825 cache_key=cache_key,
2826 api_base=api_base,
2827 version=version,
2828 response_cost=response_cost,
2829 model_region=getattr(user_api_key_dict, "allowed_model_region", ""),
2830 fastest_response_batch_completion=fastest_response_batch_completion,
2831 request_data=self.data,
2832 hidden_params=hidden_params,
2833 litellm_logging_obj=logging_obj,
2834 **additional_headers,
2835 )
2836 _early = await self._handle_non_streaming_allm_passthrough_route(
2837 response=response,
2838 proxy_logging_obj=proxy_logging_obj,
2839 user_api_key_dict=user_api_key_dict,
2840 custom_headers=_non_streaming_custom_headers,
2841 request_headers=dict(request.headers),
2842 )
2843 if _early is not None:
2844 return _early
2846 response = await proxy_logging_obj.post_call_success_hook(
2847 data=self.data,
2848 user_api_key_dict=user_api_key_dict,
2849 response=response,
2850 )
2851 record_served_output_texts(logging_obj.model_call_details, served_output_texts(response))
2852 except Exception:
2853 _exception_raised = True
2854 raise
2855 finally:
2856 ProxyBaseLLMRequestProcessing._flush_deferred_async_logging(
2857 logging_obj=logging_obj,
2858 exception_raised=_exception_raised,
2859 )
2861 # Streaming cleanup: if an exception occurred AND the deferred
2862 # streaming closure is still set, no streaming route will
2863 # consume the CSW — the closure is orphaned. Clear it and
2864 # fire logging directly to avoid silent loss.
2865 #
2866 # On normal streaming returns the closure must stay: CSW calls
2867 # it at stream end. _exception_raised is function-scoped and
2868 # immune to outer exception context, avoiding false positives.
2869 if _exception_raised: 2869 ↛ 2870line 2869 didn't jump to line 2870 because the condition on line 2869 was never true
2870 _deferred_fn: Final = getattr(logging_obj, "_on_deferred_stream_complete", None)
2871 if _deferred_fn is not None:
2872 logging_obj._on_deferred_stream_complete = None
2873 try:
2874 asyncio.create_task(
2875 logging_obj.dispatch_success_handlers(
2876 response,
2877 cache_hit=None,
2878 start_time=None,
2879 end_time=None,
2880 prefer_async_handlers=True,
2881 )
2882 )
2883 except Exception as e:
2884 verbose_proxy_logger.exception("Error in orphaned streaming async logging: %s", e)
2886 hidden_params = get_hidden_params_dict(response) # get any updated response headers
2887 additional_headers = hidden_params.get("additional_headers", {}) or {}
2889 recover_response_cost: Final = not response_cost and hidden_params.get("response_cost") is None
2890 computed_cost_for_headers: Final = (
2891 self._response_cost_from_logging_obj(response=response, logging_obj=logging_obj) or ""
2892 if recover_response_cost
2893 else response_cost
2894 )
2895 llm_cost_for_headers: Final = (
2896 0.0
2897 if is_unbilled_non_inference_call_from_params(logging_obj.call_type, logging_obj.litellm_params, response)
2898 else computed_cost_for_headers
2899 )
2900 _, request_metadata_bucket = get_or_create_metadata_bucket(self.data)
2901 guardrail_cost_for_headers: Final = guardrail_information_cost(
2902 request_metadata_bucket.get("standard_logging_guardrail_information")
2903 )
2904 response_cost_for_headers: Final = (
2905 (llm_cost_for_headers if isinstance(llm_cost_for_headers, (int, float)) else 0.0)
2906 + guardrail_cost_for_headers
2907 if guardrail_cost_for_headers > 0
2908 else llm_cost_for_headers
2909 )
2911 # Always return the client-requested model name (not provider-prefixed internal identifiers)
2912 # for OpenAI-compatible responses.
2913 if requested_model_from_client: 2913 ↛ 2914line 2913 didn't jump to line 2914 because the condition on line 2913 was never true
2914 _override_openai_response_model(
2915 response_obj=response,
2916 requested_model=requested_model_from_client,
2917 log_context=f"litellm_call_id={logging_obj.litellm_call_id}",
2918 return_raw_model_name=_should_return_raw_model_name(self.data),
2919 )
2921 fastapi_response.headers.update(
2922 ProxyBaseLLMRequestProcessing.get_custom_headers(
2923 user_api_key_dict=user_api_key_dict,
2924 call_id=logging_obj.litellm_call_id,
2925 model_id=model_id,
2926 cache_key=cache_key,
2927 api_base=api_base,
2928 version=version,
2929 response_cost=response_cost_for_headers,
2930 model_region=getattr(user_api_key_dict, "allowed_model_region", ""),
2931 fastest_response_batch_completion=fastest_response_batch_completion,
2932 request_data=self.data,
2933 hidden_params=hidden_params,
2934 litellm_logging_obj=logging_obj,
2935 **additional_headers,
2936 )
2937 )
2939 if isinstance(response, dict):
2940 response.pop("_hidden_params", None)
2942 if include_guardrail_response_requested(self.data): 2942 ↛ 2943line 2942 didn't jump to line 2943 because the condition on line 2942 was never true
2943 response = attach_guardrail_information( # rebind-ok: response tail rebinds the copied response
2944 response=response, request_data=self.data
2945 )
2947 # Call response headers hook for non-streaming success
2948 callback_headers = await proxy_logging_obj.post_call_response_headers_hook(
2949 data=self.data,
2950 user_api_key_dict=user_api_key_dict,
2951 response=response,
2952 request_headers=dict(request.headers),
2953 )
2954 if callback_headers: 2954 ↛ 2955line 2954 didn't jump to line 2955 because the condition on line 2954 was never true
2955 fastapi_response.headers.update(callback_headers)
2957 await check_response_size_is_safe(response=response)
2959 if route_type in {"aresponses", "aget_responses"}: 2959 ↛ 2960line 2959 didn't jump to line 2960 because the condition on line 2959 was never true
2960 await ProxyBaseLLMRequestProcessing._record_container_owners_from_responses_if_needed(
2961 response=response,
2962 user_api_key_dict=user_api_key_dict,
2963 )
2965 return response
2967 @staticmethod
2968 async def _record_container_owners_from_responses_if_needed(
2969 response: object,
2970 user_api_key_dict: UserAPIKeyAuth,
2971 ) -> None:
2972 """Register code-interpreter containers so follow-up file APIs pass ownership checks."""
2973 from litellm.proxy.container_endpoints.ownership import (
2974 record_container_owners_from_responses_response,
2975 )
2977 if response is None:
2978 return
2980 try:
2981 await record_container_owners_from_responses_response(
2982 response=response,
2983 user_api_key_dict=user_api_key_dict,
2984 )
2985 except Exception as e:
2986 verbose_proxy_logger.exception(
2987 "Container ownership recording failed after responses call: %s",
2988 e,
2989 )
2991 @staticmethod
2992 def _extract_completed_responses_response(stream_response: object) -> object:
2993 """Pull the assembled ``ResponsesAPIResponse`` off a streaming iterator.
2995 ``ResponsesAPIStreamingIterator`` stores the terminal stream event
2996 (``response.completed`` / ``response.incomplete`` / ``response.failed``)
2997 in ``completed_response``; the actual response body hangs off
2998 that event's ``.response`` attribute. Some iterators store the
2999 ``ResponsesAPIResponse`` directly. Handle both shapes so the
3000 container-ownership recording path can walk ``.output`` either way.
3001 """
3002 completed: Final = attribute_of(stream_response, "completed_response")
3003 if completed is None:
3004 return None
3005 response_obj: Final = attribute_of(completed, "response")
3006 if response_obj is not None:
3007 return response_obj
3008 return completed
3010 @staticmethod
3011 async def _wrap_responses_stream_for_container_ownership(
3012 original_stream_response: object,
3013 wrapped_generator: Any,
3014 user_api_key_dict: UserAPIKeyAuth,
3015 ):
3016 """Forward SSE chunks, then record container ownership at stream end.
3018 Streaming ``/v1/responses`` short-circuits out of
3019 ``base_process_llm_request`` before the non-streaming ownership
3020 tail runs, so without this wrap the
3021 ``LiteLLM_ManagedObjectTable`` row for any container created
3022 during the stream is never written and follow-up file API calls
3023 return 403.
3024 """
3025 try:
3026 async for chunk in wrapped_generator:
3027 yield chunk
3028 finally:
3029 try:
3030 completed_obj: Final = ProxyBaseLLMRequestProcessing._extract_completed_responses_response(
3031 original_stream_response
3032 )
3033 if completed_obj is not None:
3034 await ProxyBaseLLMRequestProcessing._record_container_owners_from_responses_if_needed(
3035 response=completed_obj,
3036 user_api_key_dict=user_api_key_dict,
3037 )
3038 else:
3039 # Silent skip caused #30210: the proxy's Router wrapper
3040 # of the responses streaming iterator wasn't propagating
3041 # ``completed_response``, so this hook recorded nothing
3042 # and follow-up /v1/containers/<id>/files calls 403'd
3043 # for non-admin keys with no proxy-side hint. Log a
3044 # warning so future regressions of the same shape
3045 # surface in operator logs.
3046 verbose_proxy_logger.warning(
3047 "Container ownership recording skipped on streaming "
3048 "/v1/responses: no completed_response on stream "
3049 "iterator %s. If this stream created any tool "
3050 "container (e.g. code_interpreter), follow-up "
3051 "/v1/containers/<id>/files calls will 403 for "
3052 "non-admin keys.",
3053 type(original_stream_response).__name__,
3054 )
3055 except Exception as e:
3056 verbose_proxy_logger.exception(
3057 "Container ownership recording failed after streaming responses call: %s",
3058 e,
3059 )
3061 async def base_passthrough_process_llm_request(
3062 self,
3063 request: Request,
3064 fastapi_response: Response,
3065 user_api_key_dict: UserAPIKeyAuth,
3066 proxy_logging_obj: ProxyLogging,
3067 general_settings: dict,
3068 proxy_config: ProxyConfig,
3069 select_data_generator: Callable,
3070 llm_router: Router | None = None,
3071 model: str | None = None,
3072 user_model: str | None = None,
3073 user_temperature: float | None = None,
3074 user_request_timeout: float | None = None,
3075 user_max_tokens: int | None = None,
3076 user_api_base: str | None = None,
3077 version: str | None = None,
3078 ):
3079 from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
3080 HttpPassThroughEndpointHelpers,
3081 )
3083 result: Final = await self.base_process_llm_request(
3084 request=request,
3085 fastapi_response=fastapi_response,
3086 user_api_key_dict=user_api_key_dict,
3087 route_type="allm_passthrough_route",
3088 proxy_logging_obj=proxy_logging_obj,
3089 llm_router=llm_router,
3090 general_settings=general_settings,
3091 proxy_config=proxy_config,
3092 select_data_generator=select_data_generator,
3093 model=model,
3094 user_model=user_model,
3095 user_temperature=user_temperature,
3096 user_request_timeout=user_request_timeout,
3097 user_max_tokens=user_max_tokens,
3098 user_api_base=user_api_base,
3099 version=version,
3100 )
3102 # Check if result is actually a streaming response by inspecting its type
3103 if isinstance(result, StreamingResponse):
3104 return result
3106 # base_process_llm_request may return a FastAPI Response directly after
3107 # post-call guardrails buffer and rewrite JSON (e.g. Bedrock Converse passthrough).
3108 if isinstance(result, Response):
3109 return result
3111 upstream: Final = _as_upstream_response(result)
3112 content: Final = await upstream.aread()
3113 return Response(
3114 content=content,
3115 status_code=upstream.status_code,
3116 headers=HttpPassThroughEndpointHelpers.get_response_headers(
3117 headers=upstream.headers,
3118 custom_headers=dict(fastapi_response.headers),
3119 ),
3120 )
3122 def _arm_detached_stream_failure_hook(
3123 self,
3124 logging_obj: LiteLLMLoggingObj,
3125 user_api_key_dict: "UserAPIKeyAuth",
3126 proxy_logging_obj: ProxyLogging,
3127 ) -> None:
3128 """Let a stream that fails after the client left still reach ``post_call_failure_hook``.
3130 The client-facing generator reports a mid-stream failure itself, but once
3131 the client disconnects that generator is gone and the detached upstream
3132 drain is the only code that sees the provider error. It fires this closure
3133 so the failed spend is still written and the budget reservation released;
3134 a replacement error the hook raises has no client left to reach.
3135 """
3136 request_data: Final = self.data
3138 async def _on_detached_stream_failure(exc: Exception) -> None:
3139 try:
3140 await proxy_logging_obj.post_call_failure_hook(
3141 user_api_key_dict=user_api_key_dict,
3142 original_exception=exc,
3143 request_data=request_data,
3144 )
3145 except HTTPException:
3146 return
3148 logging_obj._on_detached_stream_failure = _on_detached_stream_failure
3150 def _is_streaming_response(self, response: object) -> bool:
3151 """
3152 Check if the response object is actually a streaming response by inspecting its type.
3154 This uses standard Python inspection to detect streaming/async iterator objects
3155 rather than relying on specific wrapper classes.
3156 """
3157 import inspect
3158 from collections.abc import AsyncGenerator, AsyncIterator
3160 # Check if it's an async generator (most reliable)
3161 if inspect.isasyncgen(response): 3161 ↛ 3162line 3161 didn't jump to line 3162 because the condition on line 3161 was never true
3162 return True
3164 # Check if it implements the async iterator protocol
3165 if isinstance(response, (AsyncIterator, AsyncGenerator)): 3165 ↛ 3166line 3165 didn't jump to line 3166 because the condition on line 3165 was never true
3166 return True
3168 return False
3170 def _is_streaming_request(self, data: dict, is_streaming_request: bool | None = False) -> bool:
3171 """
3172 Check if the request is a streaming request.
3174 1. is_streaming_request is a dynamic param passed in
3175 2. if "stream" in data and data["stream"] is True
3176 """
3177 if is_streaming_request is True: 3177 ↛ 3178line 3177 didn't jump to line 3178 because the condition on line 3177 was never true
3178 return True
3179 if "stream" in data and data["stream"] is True: 3179 ↛ 3180line 3179 didn't jump to line 3180 because the condition on line 3179 was never true
3180 return True
3181 return False
3183 @staticmethod
3184 def _has_post_call_guardrails() -> bool:
3185 """
3186 True when a guardrail explicitly registers post_call. event_hook=None
3187 matches all hooks in should_run_guardrail but must not defer async logging
3188 on non-streaming /chat/completions (no post_call_success_hook flush path).
3189 """
3190 for cb in litellm.callbacks:
3191 if not isinstance(cb, CustomGuardrail): 3191 ↛ 3193line 3191 didn't jump to line 3193 because the condition on line 3191 was always true
3192 continue
3193 if cb.event_hook is None:
3194 continue
3195 if cb._event_hook_is_event_type(GuardrailEventHooks.post_call):
3196 return True
3197 return False
3199 def _has_post_call_guardrails_for_passthrough(self) -> bool:
3200 """
3201 True when a post_call guardrail will actually run for THIS request.
3203 Mirrors the gate in ProxyLogging.post_call_success_hook
3204 (should_run_guardrail against the request's merged guardrails) so that a
3205 guardrail registered globally but not configured for this key/team does
3206 not force the passthrough stream to be buffered into a single
3207 non-streaming response. An event_hook=None guardrail still counts here
3208 because should_run_guardrail treats it as matching every hook.
3209 """
3210 from litellm.proxy.proxy_server import llm_router
3211 from litellm.proxy.utils import _check_and_merge_model_level_guardrails
3213 guardrail_data: Final = _check_and_merge_model_level_guardrails(data=self.data, llm_router=llm_router)
3214 for cb in litellm.callbacks:
3215 if not isinstance(cb, CustomGuardrail):
3216 continue
3217 if cb.should_run_guardrail(
3218 data=guardrail_data,
3219 event_type=GuardrailEventHooks.post_call,
3220 ):
3221 return True
3222 return False
3224 def _passthrough_endpoint_has_stream_guardrail_handler(self) -> bool:
3225 """
3226 True when the resolved passthrough provider AND endpoint have an
3227 event-stream guardrail handler that can rewrite buffered frames. Only such
3228 endpoints may have their stream buffered for post-call guardrails; every
3229 other endpoint must keep streaming so the response is not silently turned
3230 into a non-streaming body when no content modification would occur (e.g.
3231 Bedrock invoke-with-response-stream, whose frames the Converse handler
3232 leaves untouched).
3233 """
3234 from litellm.llms.pass_through.guardrail_translation.handler import (
3235 LlmPassthroughRouteHandler,
3236 )
3238 return LlmPassthroughRouteHandler.supports_event_stream_de_anonymization(
3239 self.data.get("custom_llm_provider"),
3240 self.data.get("endpoint"),
3241 )
3243 def _passthrough_event_stream_media_type(self) -> str | None:
3244 """
3245 Content-type for a passthrough event-stream response, resolved from the
3246 provider handler so the proxy stays provider-agnostic. Mirrors the
3247 upstream content-type the non-streaming path forwards, since the
3248 streaming generator carries no headers of its own. Used for both the
3249 buffered (guardrail-rewritten) and the unbuffered relay paths so
3250 clients that enforce the event-stream content-type (e.g. Claude Code on
3251 Bedrock invoke-with-response-stream) see the correct header instead of
3252 no content-type at all, which they fall back to reading as
3253 application/octet-stream. Returns None for providers with no
3254 event-stream media type, leaving the response headers unchanged.
3255 """
3256 from litellm.llms.pass_through.guardrail_translation.handler import (
3257 LlmPassthroughRouteHandler,
3258 )
3260 return LlmPassthroughRouteHandler.event_stream_media_type(self.data.get("custom_llm_provider"))
3262 async def _handle_non_streaming_allm_passthrough_route(
3263 self,
3264 response: _UpstreamHttpResponse,
3265 proxy_logging_obj: "ProxyLogging",
3266 user_api_key_dict: "UserAPIKeyAuth",
3267 custom_headers: Mapping[str, str],
3268 request_headers: dict[str, str],
3269 ) -> Response | None:
3270 if not self._has_post_call_guardrails_for_passthrough():
3271 return None
3273 import json as _json
3275 from litellm.llms.pass_through.guardrail_translation.handler import (
3276 LlmPassthroughRouteHandler,
3277 )
3278 from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
3279 HttpPassThroughEndpointHelpers,
3280 )
3282 upstream: Final = _as_upstream_response(response)
3283 try:
3284 response_status: Final[int] = upstream.status_code
3285 content_type: Final[str] = _as_header_reader(upstream.headers).get("content-type", "")
3286 except AttributeError:
3287 return None
3289 if response_status >= 300:
3290 return None
3292 is_event_stream: Final = LlmPassthroughRouteHandler.is_event_stream_response(
3293 self.data.get("custom_llm_provider"), content_type
3294 )
3295 if not is_event_stream and "application/json" not in content_type:
3296 return None
3298 response_headers: Final = HttpPassThroughEndpointHelpers.get_response_headers(
3299 headers=upstream.headers,
3300 custom_headers=custom_headers,
3301 )
3302 callback_headers: Final = await proxy_logging_obj.post_call_response_headers_hook(
3303 data=self.data,
3304 user_api_key_dict=user_api_key_dict,
3305 response=upstream,
3306 request_headers=request_headers,
3307 )
3308 if callback_headers:
3309 response_headers.update(callback_headers)
3311 if is_event_stream:
3312 body_bytes = await upstream.aread()
3313 modified_bytes: Final = await self._handle_event_stream_allm_passthrough_route(
3314 body_bytes=body_bytes,
3315 proxy_logging_obj=proxy_logging_obj,
3316 user_api_key_dict=user_api_key_dict,
3317 )
3318 return Response(
3319 content=modified_bytes,
3320 status_code=response_status,
3321 media_type=content_type,
3322 headers=response_headers,
3323 )
3325 body_bytes = await upstream.aread()
3326 try:
3327 parsed: Final = _json.loads(body_bytes)
3328 except (_json.JSONDecodeError, UnicodeDecodeError):
3329 return Response(
3330 content=body_bytes,
3331 status_code=response_status,
3332 media_type="application/json",
3333 headers=response_headers,
3334 )
3335 processed: Final = await proxy_logging_obj.post_call_success_hook(
3336 data=self.data,
3337 user_api_key_dict=user_api_key_dict,
3338 response=parsed,
3339 )
3340 if isinstance(processed, dict):
3341 content = _json.dumps(processed).encode()
3342 else:
3343 verbose_proxy_logger.debug(
3344 "allm_passthrough_route: post_call_success_hook returned %s, leaving JSON response unmodified",
3345 type(processed).__name__,
3346 )
3347 content = body_bytes
3348 return Response(
3349 content=content,
3350 status_code=response_status,
3351 media_type="application/json",
3352 headers=response_headers,
3353 )
3355 async def _handle_event_stream_allm_passthrough_route(
3356 self,
3357 body_bytes: bytes,
3358 proxy_logging_obj: "ProxyLogging",
3359 user_api_key_dict: "UserAPIKeyAuth",
3360 ) -> bytes:
3361 from litellm.llms.pass_through.guardrail_translation.handler import (
3362 LlmPassthroughRouteHandler,
3363 )
3365 return await LlmPassthroughRouteHandler.de_anonymize_event_stream(
3366 body_bytes=body_bytes,
3367 proxy_logging_obj=proxy_logging_obj,
3368 user_api_key_dict=user_api_key_dict,
3369 data=self.data,
3370 )
3372 @staticmethod
3373 def _flush_deferred_async_logging(
3374 logging_obj: Any,
3375 exception_raised: bool,
3376 ) -> None:
3377 """
3378 Fire the deferred async-success closure stored by wrapper_async, then
3379 clear the slot.
3381 Called from the finally block around post_call_success_hook so the
3382 StandardLoggingPayload is built after post-call guardrails write to
3383 metadata (deferred logging is enabled for non-streaming requests with
3384 a registered post_call guardrail).
3386 On exception (e.g. a post-call guardrail blocks the response), skip
3387 firing the closure — the exception propagates to post_call_failure_hook
3388 which writes its own failure spend log via async_failure_handler.
3389 Firing both produced a duplicate (Success + Failure) entry per request,
3390 with the Success row exposing the blocked LLM response.
3392 For streaming early-returns the closure is never stored (wrapper_async
3393 returns before the deferred block in litellm/utils.py), so this is a
3394 no-op there.
3396 Extracted as a static method so tests can exercise the production
3397 gating logic directly rather than reimplementing the finally block.
3398 """
3399 if getattr(logging_obj, "call_type", None) in ("ocr", "aocr"): 3399 ↛ 3400line 3399 didn't jump to line 3400 because the condition on line 3399 was never true
3400 pending: Final = getattr(logging_obj, "_native_pending_logging", None)
3401 if pending is not None:
3402 logging_obj._native_pending_logging = None # rebind-ok: consume the native OCR release signal once
3403 pending.release(not exception_raised)
3404 _enqueue_fn: Final = getattr(logging_obj, "_enqueue_deferred_logging", None)
3405 if _enqueue_fn is None: 3405 ↛ 3407line 3405 didn't jump to line 3407 because the condition on line 3405 was always true
3406 return
3407 logging_obj._enqueue_deferred_logging = None
3408 if exception_raised:
3409 return
3410 try:
3411 _enqueue_fn()
3412 except Exception as e:
3413 verbose_proxy_logger.exception("Error firing deferred logging: %s", e)
3415 def _arm_deferred_stream_dispatch(
3416 self,
3417 response: object,
3418 route_type: str,
3419 user_api_key_dict: "UserAPIKeyAuth",
3420 logging_obj: LiteLLMLoggingObj,
3421 ) -> None:
3422 """
3423 Streaming with post-call guardrails active: attach a closure that
3424 ProxyLogging._fire_deferred_stream_logging fires after all guardrail
3425 end-of-stream blocks complete, so the spend log sees
3426 guardrail_information.
3428 Two closure shapes, matching who owns logging for the stream:
3429 - CustomStreamWrapper (chat completions) stores
3430 (assembled_response, cache_hit); the closure also runs
3431 non-apply_guardrail post-call hooks via
3432 _run_deferred_stream_guardrails.
3433 - Every other anthropic_messages/aresponses stream gets a closure
3434 that dispatches on the stored args shape, because the arming site
3435 cannot tell the producers apart: native iterators store a single
3436 ready-made logging coroutine to enqueue, while bridged streams
3437 (LiteLLMCompletionStreamingIterator, and the plain SSE generator
3438 AnthropicStreamWrapper returns for bridged /v1/messages) share
3439 their inner CustomStreamWrapper's logging_obj and so store
3440 (assembled_response, cache_hit); for those the closure only
3441 dispatches success logging, matching the route's pre-existing
3442 hook surface.
3444 Raw async generators from passthrough routes bypass both and would
3445 orphan the closure, so they are not armed here.
3447 The router wraps iterators that cannot carry _hidden_params in
3448 HiddenParamsAsyncIteratorWrapper, so class sniffing runs on the
3449 unwrapped inner iterator.
3450 """
3451 from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
3452 from litellm.router_utils.add_retry_fallback_headers import HiddenParamsAsyncIteratorWrapper
3454 unwrapped: Final = response._inner if isinstance(response, HiddenParamsAsyncIteratorWrapper) else response
3456 if isinstance(unwrapped, CustomStreamWrapper):
3457 # Intentionally a live reference (not a copy) — mirrors
3458 # ProxyLogging.post_call_success_hook which also mutates
3459 # data["guardrail_to_apply"] during iteration.
3460 _captured_data: Final = self.data
3461 _captured_user_api_key_dict: Final = user_api_key_dict
3462 _captured_logging_obj: Final = logging_obj
3464 async def _on_deferred_stream_complete(assembled_response: object, cache_hit: object) -> None:
3465 await ProxyBaseLLMRequestProcessing._run_deferred_stream_guardrails(
3466 captured_data=_captured_data,
3467 captured_user_api_key_dict=_captured_user_api_key_dict,
3468 captured_logging_obj=_captured_logging_obj,
3469 assembled_response=assembled_response,
3470 cache_hit=cache_hit,
3471 )
3473 logging_obj._on_deferred_stream_complete = _on_deferred_stream_complete
3474 return
3476 if route_type not in ("anthropic_messages", "aresponses") or not self._is_streaming_response(response):
3477 return
3479 from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
3481 _captured_native_logging_obj: Final = logging_obj
3483 async def _on_deferred_native_stream_complete(*args: object) -> None:
3484 match args:
3485 case (logging_coroutine,) if asyncio.iscoroutine(logging_coroutine):
3486 GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(async_coroutine=logging_coroutine)
3487 case (assembled_response, cache_hit):
3488 await _as_success_dispatcher(_captured_native_logging_obj).dispatch_success_handlers(
3489 assembled_response,
3490 cache_hit=cache_hit,
3491 start_time=None,
3492 end_time=None,
3493 prefer_async_handlers=True,
3494 )
3495 case _:
3496 verbose_proxy_logger.error(
3497 "Deferred stream logging dropped: unexpected stored args shape %s",
3498 tuple(type(arg).__name__ for arg in args),
3499 )
3501 logging_obj._on_deferred_stream_complete = _on_deferred_native_stream_complete
3503 @staticmethod
3504 async def _run_deferred_stream_guardrails(
3505 captured_data: dict,
3506 captured_user_api_key_dict: "UserAPIKeyAuth",
3507 captured_logging_obj: LiteLLMLoggingObj,
3508 assembled_response: Any,
3509 cache_hit: object,
3510 ) -> None:
3511 """
3512 Run non-streaming post-call guardrail hooks on an assembled streaming
3513 response, then fire success logging via ``dispatch_success_handlers``.
3515 Called by ProxyLogging._fire_deferred_stream_logging after the full
3516 streaming pipeline (including unified_guardrail end-of-stream blocks)
3517 has completed.
3519 Guardrails routed through unified_guardrail are skipped, since they already ran
3520 via its streaming iterator, and so are guardrails a post_call policy pipeline
3521 manages, since the pipeline ran them against the buffered stream. Guardrails
3522 that override async_post_call_success_hook directly run here, including those
3523 that implement apply_guardrail but keep their native lifecycle hooks.
3525 This is audit-only — content has already been delivered to the client.
3527 Extracted as a static method so tests can call the production
3528 implementation directly rather than reimplementing the closure.
3529 """
3530 _response = assembled_response
3531 try:
3532 from litellm.proxy.proxy_server import llm_router as _global_llm_router
3533 from litellm.proxy.utils import (
3534 _check_and_merge_model_level_guardrails,
3535 stream_gated_guardrail_names,
3536 )
3538 guardrail_data = _check_and_merge_model_level_guardrails(data=captured_data, llm_router=_global_llm_router)
3539 stream_gated: Final = stream_gated_guardrail_names(captured_data, captured_user_api_key_dict)
3540 for cb in litellm.callbacks:
3541 if not isinstance(cb, CustomGuardrail):
3542 continue
3543 if cb.guardrail_name in stream_gated:
3544 continue
3545 if not cb.should_run_guardrail(
3546 data=guardrail_data,
3547 event_type=GuardrailEventHooks.post_call,
3548 ):
3549 continue
3550 try:
3551 guardrail_result = None
3552 if "apply_guardrail" in type(cb).__dict__ and not cb.use_native_lifecycle_hooks:
3553 # Skip — unified-routed guardrails already ran via
3554 # unified_guardrail's end-of-stream block in the
3555 # streaming iterator pipeline. Running them again
3556 # here would duplicate the guardrail API call
3557 # (e.g. double OpenAI Moderation charges).
3558 continue
3559 if "async_post_call_streaming_iterator_hook" in type(cb).__dict__:
3560 # Skip — the guardrail already scanned the assembled
3561 # response via its own streaming iterator hook in the
3562 # streaming pipeline. re running this function async_post_call_success_hook
3563 # here would duplicate the scan and can spuriously block the guardrail that already passed / failed.
3564 continue
3565 else:
3566 guardrail_result = await cb.async_post_call_success_hook(
3567 user_api_key_dict=captured_user_api_key_dict,
3568 data=guardrail_data,
3569 response=_response,
3570 )
3571 if guardrail_result is not None:
3572 _response = guardrail_result
3573 except Exception as e:
3574 verbose_proxy_logger.exception(
3575 "Error running post-call guardrail %s on streaming response: %s",
3576 getattr(cb, "guardrail_name", type(cb).__name__),
3577 e,
3578 )
3579 if isinstance(e, HTTPException) and hasattr(captured_logging_obj, "model_call_details"):
3580 captured_logging_obj.model_call_details.setdefault("metadata", {})["guardrail_blocked"] = True
3581 except Exception as e:
3582 verbose_proxy_logger.exception(
3583 "Error in deferred streaming guardrail initialization: %s",
3584 e,
3585 )
3586 finally:
3587 try:
3588 # Proxy streaming always runs in async context and proxy spend
3589 # logging is async-only; force async dispatch so DB/spend
3590 # callbacks fire regardless of the call-type heuristic in
3591 # _is_sync_litellm_request (which only recognizes a subset of
3592 # async markers stored in litellm_params).
3593 asyncio.create_task(
3594 _as_success_dispatcher(captured_logging_obj).dispatch_success_handlers(
3595 _response,
3596 cache_hit=cache_hit,
3597 start_time=None,
3598 end_time=None,
3599 prefer_async_handlers=True,
3600 )
3601 )
3602 except Exception as e:
3603 verbose_proxy_logger.exception(
3604 "Error in deferred streaming success logging: %s",
3605 e,
3606 )
3608 def _apply_router_cooldown_retry_after(self, headers: dict, e: Exception) -> None:
3609 if isinstance(e, RouterRateLimitError) and e.cooldown_time > 0: 3609 ↛ 3610line 3609 didn't jump to line 3610 because the condition on line 3609 was never true
3610 headers["retry-after"] = str(math.ceil(e.cooldown_time))
3612 async def _handle_llm_api_exception(
3613 self,
3614 e: Exception,
3615 user_api_key_dict: UserAPIKeyAuth,
3616 proxy_logging_obj: ProxyLogging,
3617 version: str | None = None,
3618 ):
3619 """Raises ProxyException (OpenAI API compatible) if an exception is raised"""
3620 log_llm_api_exception(e, self.litellm_call_id)
3621 # Allow callbacks to transform the error response
3622 transformed_exception: Final = await proxy_logging_obj.post_call_failure_hook(
3623 user_api_key_dict=user_api_key_dict,
3624 original_exception=e,
3625 request_data=self.data,
3626 )
3627 # Use transformed exception if callback returned one, otherwise use original
3628 if transformed_exception is not None: 3628 ↛ 3629line 3628 didn't jump to line 3629 because the condition on line 3628 was never true
3629 e = transformed_exception
3630 litellm_debug_info: Final = getattr(e, "litellm_debug_info", "")
3631 verbose_proxy_logger.debug(
3632 "\033[1;31mAn error occurred: %s %s\n\n Debug this by setting `--debug`, e.g. `litellm --model gpt-3.5-turbo --debug`",
3633 e,
3634 litellm_debug_info,
3635 )
3637 timeout: Final = getattr(
3638 e, "timeout", None
3639 ) # returns the timeout set by the wrapper. Used for testing if model-specific timeout are set correctly
3640 _litellm_logging_obj: Final[LiteLLMLoggingObj | None] = self.data.get("litellm_logging_obj", None)
3642 # Attempt to get model_id from logging object
3643 #
3644 # Note: We check the direct model_info path first (not nested in metadata) because that's where the router sets it.
3645 # The nested metadata path is only a fallback for cases where model_info wasn't set at the top level.
3646 model_id: Final = self.maybe_get_model_id(_litellm_logging_obj)
3648 custom_headers: Final = ProxyBaseLLMRequestProcessing.get_custom_headers(
3649 user_api_key_dict=user_api_key_dict,
3650 call_id=self.litellm_call_id,
3651 model_id=model_id,
3652 version=version,
3653 response_cost=0,
3654 model_region=getattr(user_api_key_dict, "allowed_model_region", ""),
3655 request_data=self.data,
3656 timeout=timeout,
3657 litellm_logging_obj=_litellm_logging_obj,
3658 # a failed request reports no timing, matching /v1/chat/completions
3659 read_timing_from_logging_obj=False,
3660 )
3661 headers = getattr(e, "headers", None) or {}
3662 if not headers: 3662 ↛ 3668line 3662 didn't jump to line 3668 because the condition on line 3662 was always true
3663 _response: Final = attribute_of(e, "response")
3664 _response_headers: Final = getattr(_response, "headers", None) if _response is not None else None
3665 _provider_headers: Final = _response_headers or getattr(e, "litellm_response_headers", None)
3666 if _provider_headers:
3667 headers = get_response_headers(dict(_provider_headers))
3668 headers.update(custom_headers)
3670 # Call response headers hook for failure
3671 try:
3672 callback_headers: Final = await proxy_logging_obj.post_call_response_headers_hook(
3673 data=self.data,
3674 user_api_key_dict=user_api_key_dict,
3675 response=None,
3676 request_headers=(self.data.get("proxy_server_request") or {}).get("headers", {}),
3677 )
3678 if callback_headers: 3678 ↛ 3679line 3678 didn't jump to line 3679 because the condition on line 3678 was never true
3679 headers.update(callback_headers)
3680 except Exception:
3681 pass
3683 safe_headers: Final = {k: v for k, v in headers.items() if k.lower() not in UNSAFE_PROXY_RESPONSE_HEADERS}
3685 self._apply_router_cooldown_retry_after(safe_headers, e)
3687 if isinstance(e, ProxyException):
3688 e.headers = {
3689 **{k: v for k, v in e.headers.items() if k.lower() not in UNSAFE_PROXY_RESPONSE_HEADERS},
3690 **{k: v if isinstance(v, str) else str(v) for k, v in safe_headers.items()},
3691 }
3692 raise e
3694 if isinstance(e, HTTPException):
3695 raise proxy_exception_from_http_exception(e, safe_headers)
3696 elif isinstance(e, httpx.HTTPStatusError): 3696 ↛ 3699line 3696 didn't jump to line 3699 because the condition on line 3696 was never true
3697 # Handle httpx.HTTPStatusError - extract actual error from response
3698 # This matches the original behavior before the refactor in commit 511d435f6f
3699 http_status_error: Final[httpx.HTTPStatusError] = e
3700 error_body: Final = await http_status_error.response.aread()
3701 error_text: Final = error_body.decode("utf-8")
3703 error_headers: Final = { # mutable-ok: HTTPException takes a plain header dict
3704 k: v if isinstance(v, str) else str(v) for k, v in safe_headers.items()
3705 }
3706 raise HTTPException(
3707 status_code=http_status_error.response.status_code,
3708 detail={"error": error_text},
3709 headers=error_headers,
3710 )
3711 error_msg: Final = f"{e}"
3712 # Check for AttributeError in the exception chain.
3713 # The AttributeError may be wrapped in multiple layers
3714 # (e.g. AttributeError -> OpenAIException -> APIConnectionError),
3715 # so walk __cause__, __context__, and original_exception recursively.
3716 has_attribute_error: Final = _has_attribute_error_in_chain(e)
3718 if has_attribute_error:
3719 raise ProxyException(
3720 message=f"Invalid request format: {error_msg}",
3721 type="invalid_request_error",
3722 param=None,
3723 code=status.HTTP_400_BAD_REQUEST,
3724 headers=safe_headers,
3725 )
3726 # Extract status_code from the exception if it carries one.
3727 # Provider exceptions (NotFoundError, BadRequestError, GeminiError,
3728 # VertexAIError, etc.) all have a status_code attribute reflecting
3729 # the upstream API response. Use it to return the correct HTTP code
3730 # instead of defaulting to 500.
3731 _exc_status_code: Final = getattr(e, "status_code", None)
3732 if _exc_status_code is not None and isinstance(_exc_status_code, int) and 400 <= _exc_status_code <= 599:
3733 _code = _exc_status_code
3734 else:
3735 _code = status.HTTP_500_INTERNAL_SERVER_ERROR
3736 if should_report_bug(e): 3736 ↛ 3752line 3736 didn't jump to line 3752 because the condition on line 3736 was always true
3737 proxy_server_request: Final = self.data.get("proxy_server_request")
3738 request_url: Final = (
3739 proxy_server_request.get("url") if isinstance(proxy_server_request, Mapping) else None
3740 )
3741 request_path: Final = urlparse(str(request_url)).path if request_url is not None else None
3742 verbose_proxy_logger.error(
3743 bug_report_notice(
3744 build_proxy_bug_report(
3745 e,
3746 call_type=allowlisted(request_path, KNOWN_PROXY_ROUTES),
3747 custom_llm_provider=self.data.get("custom_llm_provider"),
3748 stream=self.data.get("stream"),
3749 )
3750 )
3751 )
3752 client_message: Final = getattr(e, "message", error_msg)
3753 raise ProxyException(
3754 message=redact_internal_details_from_client_message(
3755 strip_bug_report_notice(client_message) if isinstance(client_message, str) else error_msg
3756 ),
3757 type=openai_error_type(e, _code),
3758 param=openai_error_param(e),
3759 openai_code=getattr(e, "code", None),
3760 code=_code,
3761 provider_specific_fields=getattr(e, "provider_specific_fields", None),
3762 headers=safe_headers,
3763 )
3765 #########################################################
3766 # Proxy Level Streaming Data Generator
3767 #########################################################
3769 @staticmethod
3770 def return_sse_chunk(chunk: Any) -> str:
3771 """
3772 Helper function to format streaming chunks for Anthropic API format
3774 Args:
3775 chunk: A string or dictionary to be returned in SSE format
3777 Returns:
3778 str: A properly formatted SSE chunk string
3779 """
3780 if isinstance(chunk, dict):
3781 # Use safe_dumps for proper JSON serialization with circular reference detection
3782 chunk_str: Final = safe_dumps(chunk)
3783 return f"{STREAM_SSE_DATA_PREFIX}{chunk_str}\n\n"
3784 else:
3785 return chunk
3787 @staticmethod
3788 def _sse_chunk_serializer(restamper: AnthropicStreamModelRestamper | None) -> StreamChunkSerializer:
3789 if restamper is None:
3790 return ProxyBaseLLMRequestProcessing.return_sse_chunk
3792 def serialize(chunk: object) -> str:
3793 return ProxyBaseLLMRequestProcessing.return_sse_chunk(restamper.process(chunk))
3795 return serialize
3797 @staticmethod
3798 async def _finalize_streaming_generator_cleanup(
3799 request: Request | None,
3800 request_data: dict,
3801 response: Any,
3802 stream_completed: bool = False,
3803 client_disconnected: bool = False,
3804 user_api_key_dict: UserAPIKeyAuth | None = None,
3805 proxy_logging_obj: ProxyLogging | None = None,
3806 ) -> None:
3807 with anyio.CancelScope(shield=True):
3808 should_record_client_disconnect: Final = client_disconnected or (not stream_completed)
3809 recorded_client_disconnect = False
3810 if should_record_client_disconnect:
3811 recorded_client_disconnect = await _record_streaming_client_disconnect_if_needed(
3812 request,
3813 request_data,
3814 client_disconnected,
3815 )
3816 if recorded_client_disconnect:
3817 deferred_stream_logging_armed: Final = _deferred_stream_logging_is_armed(request_data)
3818 ProxyLogging._fire_deferred_stream_logging(request_data)
3819 # A disconnect-time success event (the deferred-guardrail flush
3820 # above, or the partial-spend billing below) releases the
3821 # request's max_parallel_requests slot through the limiter's
3822 # own success callback. Release the slot explicitly only when
3823 # no such event fires, so exactly one release happens; two
3824 # concurrent releases would race and double-decrement under the
3825 # limiter's in-memory fallback.
3826 success_event_owns_slot_release = deferred_stream_logging_armed
3827 if not deferred_stream_logging_armed:
3828 success_event_owns_slot_release = await _bill_partial_streamed_spend_on_disconnect(
3829 request_data, response
3830 )
3831 if (
3832 not success_event_owns_slot_release
3833 and proxy_logging_obj is not None
3834 and user_api_key_dict is not None
3835 ):
3836 await proxy_logging_obj._arelease_max_parallel_requests_on_disconnect(user_api_key_dict)
3838 if hasattr(response, "aclose"):
3839 try:
3840 await response.aclose()
3841 except BaseException as e: # noqa: BLE001
3842 verbose_proxy_logger.debug(
3843 "async_streaming_data_generator: error closing response stream: %s",
3844 e,
3845 )
3846 logging_obj: Final = request_data.get("litellm_logging_obj")
3847 if (
3848 not stream_completed
3849 and isinstance(logging_obj, LiteLLMLoggingObj)
3850 and logging_obj.baseline_cache_context is not None
3851 and logging_obj.model_call_details.get("prompt_cache_response_complete") is not True
3852 ):
3853 await logging_obj.invalidate_baseline_cache_estimate("incomplete_response", completed=True)
3855 @staticmethod
3856 async def async_streaming_data_generator(
3857 response: object,
3858 user_api_key_dict: UserAPIKeyAuth,
3859 request_data: dict,
3860 proxy_logging_obj: ProxyLogging,
3861 *,
3862 serialize_chunk: StreamChunkSerializer,
3863 serialize_error: StreamErrorSerializer,
3864 request: Request | None = None,
3865 flush_tail: Callable[[], bytes] | None = None,
3866 ) -> AsyncGenerator[str, None]:
3867 """
3868 Shared streaming data generator: runs proxy iterator hook, per-chunk hook,
3869 cost injection, then yields chunks via serialize_chunk; on exception runs
3870 failure hook and yields via serialize_error. Use for SSE or NDJSON.
3872 ``flush_tail`` runs once after the upstream iterator completes cleanly and
3873 its non-empty result is yielded, so a serializer that buffers bytes across
3874 chunks can emit anything still held at end of stream.
3875 """
3876 verbose_proxy_logger.debug("inside generator")
3877 # Resolve per-stream (not per-chunk) whether the heavy per-chunk path
3878 # is needed. When no callback overrides ``async_post_call_streaming_hook``,
3879 # no CustomGuardrail is active, and cost injection is disabled, the
3880 # per-chunk hook returns the chunk unchanged, ``str_so_far`` is never
3881 # consumed, and cost injection is a no-op -- so the per-chunk coroutine
3882 # await, response-string materialization, and cost-injection call are
3883 # pure overhead on the streaming hot path (the default config).
3884 caps: Final = ProxyLogging._callback_capabilities()
3885 cost_injection_enabled: Final = bool(getattr(litellm, "include_cost_in_streaming_usage", False))
3886 fast_path = not caps.has_streaming_chunk_override and not caps.has_guardrail and not cost_injection_enabled
3887 debug_enabled: Final = verbose_proxy_logger.isEnabledFor(logging.DEBUG)
3888 stream_completed = False
3889 client_disconnected = False
3890 delivered_chunk = False
3891 try:
3892 str_so_far = ""
3893 async for chunk in proxy_logging_obj.async_post_call_streaming_iterator_hook(
3894 user_api_key_dict=user_api_key_dict,
3895 response=response,
3896 request_data=request_data,
3897 ):
3898 # ``.format(chunk)`` was previously evaluated for every chunk
3899 # regardless of log level; gate it behind the level check.
3900 if debug_enabled:
3901 verbose_proxy_logger.debug("async_data_generator: received streaming chunk - %s", chunk)
3903 if not fast_path:
3904 chunk = await proxy_logging_obj.async_post_call_streaming_hook(
3905 user_api_key_dict=user_api_key_dict,
3906 response=chunk,
3907 data=request_data,
3908 str_so_far=str_so_far,
3909 )
3911 if isinstance(chunk, (ModelResponse, ModelResponseStream)):
3912 response_str = litellm.get_response_string(response_obj=chunk)
3913 str_so_far += response_str
3914 elif hasattr(chunk, "model_dump"):
3915 try:
3916 d = chunk.model_dump(mode="json", exclude_none=True)
3917 if isinstance(d, dict):
3918 str_so_far += str(d.get("content", ""))
3919 except Exception:
3920 pass
3921 elif isinstance(chunk, dict):
3922 str_so_far += str(chunk.get("content", ""))
3924 model_name = request_data.get("model", "")
3925 chunk = ProxyBaseLLMRequestProcessing._process_chunk_with_cost_injection(
3926 chunk, model_name, request_data.get("litellm_logging_obj")
3927 )
3929 # Set before the yield: an async generator suspends at the yield,
3930 # so a GeneratorExit on client disconnect is raised there and any
3931 # statement after the yield never runs. The slow-path hook is
3932 # awaited above, so a cancellation during it still leaves this
3933 # False and refunds. A keepalive ping carries no provider output,
3934 # so it must not suppress that refund.
3935 delivered_chunk = delivered_chunk or chunk != STREAM_SSE_KEEPALIVE_PING_BYTES
3936 yield serialize_chunk(chunk)
3937 held_tail: Final = flush_tail() if flush_tail is not None else b""
3938 if held_tail:
3939 yield serialize_chunk(held_tail)
3940 stream_completed = True
3941 except (asyncio.CancelledError, GeneratorExit):
3942 # Client disconnected mid-stream. CancelledError / GeneratorExit
3943 # are BaseException and bypass the success/failure logging
3944 # callbacks that release the pre-call max_parallel_requests +1.
3945 # Flag the disconnect; the shielded cleanup in `finally` owns the
3946 # slot release so it can coordinate with disconnect-time success
3947 # billing and release exactly once. This is the outermost generator
3948 # Starlette closes on disconnect, so the nested iterator hook (which
3949 # only sees GeneratorExit on GC) cannot own the refund.
3950 client_disconnected = not stream_completed
3951 if not delivered_chunk and not _withheld_provider_output(response):
3952 from litellm.proxy.spend_tracking.budget_reservation import (
3953 release_budget_reservation_on_cancel,
3954 )
3956 await release_budget_reservation_on_cancel(getattr(user_api_key_dict, "budget_reservation", None))
3957 raise
3958 except Exception as e:
3959 verbose_proxy_logger.exception(
3960 "litellm.proxy.proxy_server.async_data_generator(): Exception occured - %s", e
3961 )
3962 transformed_exception: Final = await proxy_logging_obj.post_call_failure_hook(
3963 user_api_key_dict=user_api_key_dict,
3964 original_exception=e,
3965 request_data=request_data,
3966 )
3967 if transformed_exception is not None:
3968 e = transformed_exception
3969 verbose_proxy_logger.debug(
3970 "\x1b[1;31mAn error occurred: %s\n\n Debug this by setting `--debug`, e.g. `litellm --model gpt-3.5-turbo --debug`",
3971 e,
3972 )
3974 if isinstance(e, HTTPException):
3975 raise e
3976 stream_error_status: Final = error_status_code(e, status.HTTP_500_INTERNAL_SERVER_ERROR)
3977 proxy_exception: Final = ProxyException(
3978 message=redact_internal_details_from_client_message(getattr(e, "message", str(e))),
3979 type=openai_error_type(e, stream_error_status),
3980 param=openai_error_param(e),
3981 code=stream_error_status,
3982 )
3983 stream_completed = True
3984 yield serialize_error(proxy_exception)
3985 finally:
3986 await ProxyBaseLLMRequestProcessing._finalize_streaming_generator_cleanup(
3987 request=request,
3988 request_data=request_data,
3989 response=response,
3990 stream_completed=stream_completed,
3991 client_disconnected=client_disconnected,
3992 user_api_key_dict=user_api_key_dict,
3993 proxy_logging_obj=proxy_logging_obj,
3994 )
3996 @staticmethod
3997 def async_sse_data_generator(
3998 response: object,
3999 user_api_key_dict: UserAPIKeyAuth,
4000 request_data: dict,
4001 proxy_logging_obj: ProxyLogging,
4002 request: Request | None = None,
4003 restamp_model: str | None = None,
4004 ) -> AsyncGenerator[str, None]:
4005 """
4006 Anthropic /messages and Google /generateContent streaming data generator require SSE events.
4008 Returns the underlying ``async_streaming_data_generator`` configured with
4009 SSE serializers directly (rather than re-wrapping it in another
4010 ``async for: yield`` trampoline), so a streamed chunk traverses one
4011 fewer async-generator layer / coroutine resume on the hot path.
4013 ``restamp_model`` publishes that name on the Anthropic ``message_start``
4014 event in place of the provider's model, matching what the non-streaming
4015 response reports.
4016 """
4017 restamper: Final = AnthropicStreamModelRestamper(restamp_model) if restamp_model else None
4018 return ProxyBaseLLMRequestProcessing.async_streaming_data_generator(
4019 response=response,
4020 user_api_key_dict=user_api_key_dict,
4021 request_data=request_data,
4022 proxy_logging_obj=proxy_logging_obj,
4023 serialize_chunk=ProxyBaseLLMRequestProcessing._sse_chunk_serializer(restamper),
4024 serialize_error=lambda proxy_exc: (
4025 f"{STREAM_SSE_DATA_PREFIX}{json.dumps({'error': proxy_exc.to_dict()})}\n\n"
4026 ),
4027 request=request,
4028 flush_tail=None if restamper is None else restamper.flush,
4029 )
4031 @overload
4032 @staticmethod
4033 def _process_chunk_with_cost_injection( 4033 ↛ exitline 4033 didn't return from function '_process_chunk_with_cost_injection' because
4034 chunk: bytes, model_name: str, litellm_logging_obj: LiteLLMLoggingObj | None = None
4035 ) -> bytes: ...
4037 @overload
4038 @staticmethod
4039 def _process_chunk_with_cost_injection( 4039 ↛ exitline 4039 didn't return from function '_process_chunk_with_cost_injection' because
4040 chunk: object, model_name: str, litellm_logging_obj: LiteLLMLoggingObj | None = None
4041 ) -> object: ...
4043 @staticmethod
4044 def _process_chunk_with_cost_injection(
4045 chunk: object, model_name: str, litellm_logging_obj: LiteLLMLoggingObj | None = None
4046 ) -> object:
4047 """
4048 Process a streaming chunk and inject cost information if enabled.
4050 Args:
4051 chunk: The streaming chunk (dict, str, bytes, or bytearray)
4052 model_name: Model name for cost calculation
4053 litellm_logging_obj: The call's logging object, used for pricing
4055 Returns:
4056 The processed chunk with cost information injected if applicable
4057 """
4058 if not getattr(litellm, "include_cost_in_streaming_usage", False):
4059 return chunk
4061 try:
4062 if isinstance(chunk, dict):
4063 maybe_modified: Final = ProxyBaseLLMRequestProcessing._inject_cost_into_usage_dict(
4064 chunk, model_name, litellm_logging_obj
4065 )
4066 if maybe_modified is not None:
4067 return maybe_modified
4068 elif isinstance(chunk, (bytes, bytearray)):
4069 try:
4070 s: Final = chunk.decode("utf-8")
4071 if s.endswith(("\n\n", "\r\n\r\n")):
4072 maybe_mod = ProxyBaseLLMRequestProcessing._inject_cost_into_sse_frame_str(
4073 s, model_name, litellm_logging_obj
4074 )
4075 if maybe_mod is not None:
4076 return maybe_mod.encode("utf-8")
4077 except Exception:
4078 pass
4079 elif isinstance(chunk, str):
4080 # Try to parse SSE frame and inject cost into the data line
4081 maybe_mod = ProxyBaseLLMRequestProcessing._inject_cost_into_sse_frame_str(
4082 chunk, model_name, litellm_logging_obj
4083 )
4084 if maybe_mod is not None:
4085 # Ensure trailing frame separator
4086 return maybe_mod if maybe_mod.endswith("\n\n") else (maybe_mod + "\n\n")
4087 except Exception:
4088 # Never break streaming on optional cost injection
4089 pass
4091 return chunk
4093 @staticmethod
4094 def _inject_cost_into_sse_frame_str(
4095 frame_str: str, model_name: str, litellm_logging_obj: LiteLLMLoggingObj | None = None
4096 ) -> str | None:
4097 """
4098 Inject cost information into an SSE frame string by modifying the JSON in the 'data:' line.
4100 Args:
4101 frame_str: SSE frame string that may contain multiple lines
4102 model_name: Model name for cost calculation
4103 litellm_logging_obj: The call's logging object, forwarded for pricing
4105 Returns:
4106 Modified SSE frame string with cost injected, or None if no modification needed
4107 """
4108 try:
4109 # Split preserving lines
4110 lines: Final = frame_str.split("\n")
4111 for idx, ln in enumerate(lines):
4112 stripped_ln = ln.strip()
4113 if stripped_ln.startswith("data:"):
4114 json_part = stripped_ln.split("data:", 1)[1].strip()
4115 if json_part and json_part != "[DONE]":
4116 obj = json.loads(json_part)
4117 maybe_modified = ProxyBaseLLMRequestProcessing._inject_cost_into_usage_dict(
4118 obj, model_name, litellm_logging_obj
4119 )
4120 if maybe_modified is not None:
4121 lines[idx] = "data: " + safe_dumps(maybe_modified) + ("\r" if ln.endswith("\r") else "")
4122 return "\n".join(lines)
4123 return None
4124 except Exception:
4125 return None
4127 @staticmethod
4128 def _openai_stream_usage_kwargs(usage: Mapping[str, Any]) -> Mapping[str, Any]:
4129 prompt_tokens: Final = int(usage.get("prompt_tokens", 0) or 0)
4130 completion_tokens: Final = int(usage.get("completion_tokens", 0) or 0)
4131 total_tokens: Final = int(
4132 usage.get("total_tokens", prompt_tokens + completion_tokens) or (prompt_tokens + completion_tokens)
4133 )
4134 return MappingProxyType(
4135 {
4136 key: value
4137 for key, value in (
4138 ("prompt_tokens", prompt_tokens),
4139 ("completion_tokens", completion_tokens),
4140 ("total_tokens", total_tokens),
4141 ("completion_tokens_details", usage.get("completion_tokens_details")),
4142 ("prompt_tokens_details", usage.get("prompt_tokens_details")),
4143 )
4144 if value is not None
4145 }
4146 )
4148 @staticmethod
4149 def _stream_usage_for_event(obj: Mapping[str, object], usage: Mapping[str, Any]) -> Usage | None:
4150 # Anthropic reports input_tokens excluding cache tokens, so reuse the non-streaming
4151 # transformation to total the prompt and keep the 5m/1h cache creation split
4152 if obj.get("type") == "message_delta":
4153 return AnthropicConfig().calculate_usage(usage_object=usage, reasoning_content=None)
4154 if obj.get("object") == "chat.completion.chunk":
4155 return Usage(**ProxyBaseLLMRequestProcessing._openai_stream_usage_kwargs(usage))
4156 return None
4158 @staticmethod
4159 def _completion_cost_or_none(
4160 model_response: ModelResponse, model_name: str, service_tier: str | None
4161 ) -> float | None:
4162 try:
4163 return litellm.completion_cost(
4164 completion_response=model_response, model=model_name, service_tier=service_tier
4165 )
4166 except Exception:
4167 return None
4169 @staticmethod
4170 def _logging_obj_cost_or_none(
4171 model_response: ModelResponse, litellm_logging_obj: LiteLLMLoggingObj
4172 ) -> float | None:
4173 # Pricing a frame stamps cost_breakdown and, on failure, the cost-failure debug key onto
4174 # the live logging object. The pass-through handlers never recompute either one, so a
4175 # frame-derived breakdown would outlive the stream and land in the spend log. Snapshot
4176 # both and put them back, so pricing here stays a read as far as the request is concerned
4177 breakdown_before: Final = getattr(litellm_logging_obj, "cost_breakdown", None)
4178 call_details: Final = getattr(litellm_logging_obj, "model_call_details", None)
4179 debug_key: Final = "response_cost_failure_debug_information"
4180 debug_missing: Final = object()
4181 debug_before: Final = call_details.get(debug_key, debug_missing) if isinstance(call_details, dict) else None
4182 try:
4183 cost: Final = litellm_logging_obj._response_cost_calculator(result=model_response) # pyright: ignore[reportPrivateUsage] # reuse the call's own cost calc for pricing parity with the logging callback
4184 except Exception: # noqa: BLE001 # a pricing failure falls back to model-name pricing instead of breaking the stream
4185 return None
4186 finally:
4187 if hasattr(litellm_logging_obj, "cost_breakdown"):
4188 litellm_logging_obj.cost_breakdown = breakdown_before
4189 if isinstance(call_details, dict):
4190 if debug_before is debug_missing:
4191 call_details.pop(debug_key, None)
4192 else:
4193 call_details[debug_key] = debug_before
4194 return float(cost) if isinstance(cost, (int, float)) and not isinstance(cost, bool) else None
4196 @staticmethod
4197 def _streamed_usage_cost(
4198 model_response: ModelResponse,
4199 model_name: str,
4200 service_tier: str | None,
4201 litellm_logging_obj: LiteLLMLoggingObj | None,
4202 ) -> float | None:
4203 # Pricing via the logging object inherits the deployment's custom pricing, so the
4204 # streamed cost matches what the logging callback records instead of sticker price
4205 cost_from_logging_obj: Final = (
4206 ProxyBaseLLMRequestProcessing._logging_obj_cost_or_none(model_response, litellm_logging_obj)
4207 if litellm_logging_obj is not None
4208 else None
4209 )
4210 if cost_from_logging_obj is not None:
4211 return cost_from_logging_obj
4212 return ProxyBaseLLMRequestProcessing._completion_cost_or_none(model_response, model_name, service_tier)
4214 @staticmethod
4215 def _inject_cost_into_usage_dict(
4216 obj: dict, model_name: str, litellm_logging_obj: LiteLLMLoggingObj | None = None
4217 ) -> dict | None:
4218 """
4219 Inject cost information into the usage object of a streamed usage event
4220 (Anthropic ``message_delta`` or OpenAI ``chat.completion.chunk``).
4222 Args:
4223 obj: Dictionary containing the SSE event data
4224 model_name: Model name for cost calculation
4225 litellm_logging_obj: The call's logging object, used for pricing
4227 Returns:
4228 Modified dictionary with cost injected, or None if no modification needed
4229 """
4230 usage: Final = obj.get("usage")
4231 if not isinstance(usage, dict):
4232 return None
4233 stream_usage: Final = ProxyBaseLLMRequestProcessing._stream_usage_for_event(obj, usage)
4234 if stream_usage is None:
4235 return None
4236 service_tier: Final = obj.get("service_tier")
4237 cost_val: Final = ProxyBaseLLMRequestProcessing._streamed_usage_cost(
4238 ModelResponse(usage=stream_usage),
4239 model_name,
4240 service_tier if isinstance(service_tier, str) else None,
4241 litellm_logging_obj,
4242 )
4243 if cost_val is None:
4244 return None
4245 return {**obj, "usage": {**usage, "cost": cost_val}}
4247 def maybe_get_model_id(self, _logging_obj: LiteLLMLoggingObj | None) -> str | None:
4248 """
4249 Get model_id from logging object or request metadata.
4251 The router sets model_info.id when selecting a deployment. This tries multiple locations
4252 where the ID might be stored depending on the request lifecycle stage.
4253 """
4254 model_id = None
4255 if _logging_obj: 4255 ↛ 4257line 4255 didn't jump to line 4257 because the condition on line 4255 was never true
4256 # 1. Try getting from litellm_params (updated during call)
4257 if hasattr(_logging_obj, "litellm_params") and _logging_obj.litellm_params:
4258 # First check direct model_info path (set by router.py with selected deployment)
4259 model_info = _logging_obj.litellm_params.get("model_info") or {}
4260 model_id = model_info.get("id", None)
4262 # Fallback to nested metadata path
4263 if not model_id:
4264 metadata = _logging_obj.litellm_params.get("metadata") or {}
4265 model_info = metadata.get("model_info") or {}
4266 model_id = model_info.get("id", None)
4268 # 2. Fallback to kwargs (initial)
4269 if not model_id:
4270 _kwargs: Final = getattr(_logging_obj, "kwargs", None)
4271 if _kwargs:
4272 litellm_params: Final = _kwargs.get("litellm_params", {})
4273 # First check direct model_info path
4274 model_info = litellm_params.get("model_info") or {}
4275 model_id = model_info.get("id", None)
4277 # Fallback to nested metadata path
4278 if not model_id:
4279 metadata = litellm_params.get("metadata") or {}
4280 model_info = metadata.get("model_info") or {}
4281 model_id = model_info.get("id", None)
4283 # 3. Final fallback to self.data["litellm_metadata"] (for routes like /v1/responses that populate data before error)
4284 if not model_id: 4284 ↛ 4289line 4284 didn't jump to line 4289 because the condition on line 4284 was always true
4285 litellm_metadata: Final = self.data.get("litellm_metadata", {}) or {}
4286 model_info = litellm_metadata.get("model_info", {}) or {}
4287 model_id = model_info.get("id", None)
4289 return model_id