Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_gigachat_passthrough_25886

# Conflicts:
#	osv-scanner.toml
This commit is contained in:
mateo-berri 2026-08-31 15:25:10 -07:00
commit eb00986f18
57 changed files with 2481 additions and 212 deletions

View file

@ -1,10 +1,10 @@
# syntax=docker/dockerfile:1.7
# Base image for building
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
# Runtime image
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
# Pinned by digest like the other base images; bump explicitly on Node upgrades.
ARG UI_BUILD_IMAGE=node:24.19-alpine3.24@sha256:d32cdf619f63fe0471182d08996dd516c6275bb5fd31ae06e55a570bd9e1ad43
@ -40,8 +40,8 @@ COPY --from=uvbin /uvx /usr/local/bin/uvx
RUN apk add --no-cache \
bash \
gcc \
python3 \
python3-dev \
python-3.13 \
python-3.13-dev \
rust \
openssl \
openssl-dev \
@ -51,6 +51,7 @@ RUN apk add --no-cache \
ENV UV_PROJECT_ENVIRONMENT=/app/.venv \
UV_LINK_MODE=copy \
UV_PYTHON_DOWNLOADS=0 \
PATH="/app/.venv/bin:${PATH}"
# Copy dependency metadata first for layer caching
@ -65,7 +66,7 @@ RUN uv sync --frozen --no-install-project --no-install-workspace --no-default-gr
--extra extra_proxy \
--extra semantic-router \
--extra saml \
--python python3
--python python3.13
# Copy full source tree
COPY . .
@ -86,7 +87,7 @@ RUN uv sync --frozen --no-default-groups --no-editable \
--extra extra_proxy \
--extra semantic-router \
--extra saml \
--python python3
--python python3.13
RUN HOME=/opt/prisma XDG_CACHE_HOME=/opt/prisma/.cache PRISMA_BINARY_CACHE_DIR=/opt/prisma/binaries \
npm_config_cache=/root/.npm \
@ -101,7 +102,7 @@ FROM $LITELLM_RUNTIME_IMAGE AS runtime
USER root
# node (without npm) is required by the prisma CLI at runtime
RUN apk add --no-cache bash openssl tzdata nodejs python3 libsndfile
RUN apk add --no-cache bash openssl tzdata nodejs python-3.13 libsndfile
WORKDIR /app
ENV PATH="/app/.venv/bin:${PATH}" \

View file

@ -1,5 +1,5 @@
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
FROM $UV_IMAGE AS uvbin
@ -16,7 +16,7 @@ COPY --from=uvbin /uv /uvx /usr/local/bin/
# instead of nodeenv downloading one whose dynamic deps may not be in Wolfi
# (e.g. Node 26.2.0 needs libatomic). Retry for transient apk.cgr.dev flakes.
RUN for i in 1 2 3; do \
apk add --no-cache bash gcc python3 python3-dev openssl openssl-dev libsndfile nodejs npm && break; \
apk add --no-cache bash gcc python-3.13 python-3.13-dev openssl openssl-dev libsndfile nodejs npm && break; \
[ $i = 3 ] && { echo "apk add failed after 3 retries" >&2; exit 1; }; \
sleep 5; \
done
@ -46,7 +46,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \
--python python3
--python python3.13
# Stage 2 — copy source and install the project + workspace members.
COPY . .
@ -57,7 +57,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
--extra proxy-runtime \
--extra extra_proxy \
--extra semantic-router \
--python python3
--python python3.13
RUN HOME=/opt/prisma XDG_CACHE_HOME=/opt/prisma/.cache PRISMA_BINARY_CACHE_DIR=/opt/prisma/binaries \
npm_config_cache=/root/.npm \
@ -71,7 +71,7 @@ FROM $LITELLM_RUNTIME_IMAGE AS runtime
USER root
RUN for i in 1 2 3; do \
apk add --no-cache bash openssl tzdata python3 libsndfile libatomic && break; \
apk add --no-cache bash openssl tzdata python-3.13 libsndfile libatomic && break; \
[ $i = 3 ] && { echo "apk add failed after 3 retries" >&2; exit 1; }; \
sleep 5; \
done

View file

@ -3,7 +3,7 @@
"limit": 16171
},
"reportArgumentType": {
"limit": 2226
"limit": 2224
},
"reportAssignmentType": {
"limit": 319

View file

@ -1,10 +1,10 @@
# syntax=docker/dockerfile:1.7
# Base image for building
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
# Runtime image
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
# Pinned by digest like the other base images; bump explicitly on Node upgrades.
ARG UI_BUILD_IMAGE=node:24.19-alpine3.24@sha256:d32cdf619f63fe0471182d08996dd516c6275bb5fd31ae06e55a570bd9e1ad43
@ -39,8 +39,8 @@ COPY --from=uvbin /uvx /usr/local/bin/uvx
RUN apk add --no-cache \
bash \
gcc \
python3 \
python3-dev \
python-3.13 \
python-3.13-dev \
openssl \
openssl-dev \
nodejs \
@ -49,6 +49,7 @@ RUN apk add --no-cache \
ENV UV_PROJECT_ENVIRONMENT=/app/.venv \
UV_LINK_MODE=copy \
UV_PYTHON_DOWNLOADS=0 \
PATH="/app/.venv/bin:${PATH}"
# Copy dependency metadata first for layer caching
@ -63,7 +64,7 @@ RUN uv sync --frozen --no-install-project --no-install-workspace --no-default-gr
--extra extra_proxy \
--extra semantic-router \
--extra saml \
--python python3
--python python3.13
# Copy full source tree
COPY . .
@ -84,7 +85,7 @@ RUN uv sync --frozen --no-default-groups --no-editable \
--extra extra_proxy \
--extra semantic-router \
--extra saml \
--python python3
--python python3.13
RUN HOME=/opt/prisma XDG_CACHE_HOME=/opt/prisma/.cache PRISMA_BINARY_CACHE_DIR=/opt/prisma/binaries \
npm_config_cache=/root/.npm \
@ -98,7 +99,7 @@ FROM $LITELLM_RUNTIME_IMAGE AS runtime
USER root
# node (without npm) is required by the prisma CLI at runtime
RUN apk add --no-cache bash openssl tzdata nodejs python3 libsndfile
RUN apk add --no-cache bash openssl tzdata nodejs python-3.13 libsndfile
WORKDIR /app
ENV PATH="/app/.venv/bin:${PATH}" \

View file

@ -1,8 +1,8 @@
# syntax=docker/dockerfile:1.7
# Base images
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG PROXY_EXTRAS_SOURCE=published
ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
# Pinned by digest like the other base images; bump explicitly on Node upgrades.
@ -37,8 +37,8 @@ COPY --from=uvbin /uvx /usr/local/bin/uvx
RUN for i in 1 2 3; do \
apk add --no-cache \
python3 \
python3-dev \
python-3.13 \
python-3.13-dev \
gcc \
rust \
bash \
@ -52,6 +52,7 @@ RUN for i in 1 2 3; do \
ENV UV_PROJECT_ENVIRONMENT=/app/.venv \
UV_LINK_MODE=copy \
UV_PYTHON_DOWNLOADS=0 \
PATH="/app/.venv/bin:${PATH}" \
LITELLM_NON_ROOT=true \
XDG_CACHE_HOME=/app/.cache
@ -69,7 +70,7 @@ RUN --mount=type=cache,target=/app/.cache/uv,id=litellm-uv-cache \
--extra extra_proxy \
--extra semantic-router \
--extra saml \
--python python3
--python python3.13
# Copy full source tree
COPY . .
@ -96,7 +97,7 @@ RUN --mount=type=cache,target=/app/.cache/uv,id=litellm-uv-cache \
--extra extra_proxy \
--extra semantic-router \
--extra saml \
--python python3 \
--python python3.13 \
--no-sources-package litellm-proxy-extras; \
else \
uv sync --frozen --no-default-groups --no-editable \
@ -105,7 +106,7 @@ RUN --mount=type=cache,target=/app/.cache/uv,id=litellm-uv-cache \
--extra extra_proxy \
--extra semantic-router \
--extra saml \
--python python3; \
--python python3.13; \
fi
RUN HOME=/opt/prisma XDG_CACHE_HOME=/opt/prisma/.cache PRISMA_BINARY_CACHE_DIR=/opt/prisma/binaries \
@ -124,7 +125,7 @@ RUN for i in 1 2 3; do \
apk upgrade --no-cache && break || sleep 5; \
done && \
for i in 1 2 3; do \
apk add --no-cache python3 bash openssl tzdata libsndfile nodejs && break || sleep 5; \
apk add --no-cache python-3.13 bash openssl tzdata libsndfile nodejs && break || sleep 5; \
done
# Copy only what runtime needs. The application is installed inside the venv;

View file

@ -1,5 +1,5 @@
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
FROM $UV_IMAGE AS uvbin
@ -16,7 +16,7 @@ COPY --from=uvbin /uv /uvx /usr/local/bin/
# instead of nodeenv downloading one whose dynamic deps may not be in Wolfi
# (e.g. Node 26.2.0 needs libatomic). Retry for transient apk.cgr.dev flakes.
RUN for i in 1 2 3; do \
apk add --no-cache bash gcc python3 python3-dev openssl openssl-dev libsndfile nodejs npm && break; \
apk add --no-cache bash gcc python-3.13 python-3.13-dev openssl openssl-dev libsndfile nodejs npm && break; \
[ $i = 3 ] && { echo "apk add failed after 3 retries" >&2; exit 1; }; \
sleep 5; \
done
@ -47,7 +47,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
--extra extra_proxy \
--extra semantic-router \
--extra bedrock-realtime \
--python python3
--python python3.13
# Stage 2 — copy source and install the project + workspace members.
COPY . .
@ -59,7 +59,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
--extra extra_proxy \
--extra semantic-router \
--extra bedrock-realtime \
--python python3
--python python3.13
RUN HOME=/opt/prisma XDG_CACHE_HOME=/opt/prisma/.cache PRISMA_BINARY_CACHE_DIR=/opt/prisma/binaries \
npm_config_cache=/root/.npm \
@ -73,7 +73,7 @@ FROM $LITELLM_RUNTIME_IMAGE AS runtime
USER root
RUN for i in 1 2 3; do \
apk add --no-cache bash openssl tzdata python3 libsndfile libatomic && break; \
apk add --no-cache bash openssl tzdata python-3.13 libsndfile libatomic && break; \
[ $i = 3 ] && { echo "apk add failed after 3 retries" >&2; exit 1; }; \
sleep 5; \
done

View file

@ -7,6 +7,9 @@ warnings.filterwarnings("ignore", message=".*conflict with protected namespace.*
# Suppress Pydantic 2.11+ deprecation warning about accessing model_fields on instances
# This warning can accumulate during streaming and cause memory leaks
warnings.filterwarnings("ignore", message=".*Accessing the.*attribute on the instance is deprecated.*")
# ReadOnly on TypedDict fields is repo-wide static discipline (LIT012); pydantic warns it
# cannot enforce it at runtime, which floods proxy boot once such a type is schema-walked
warnings.filterwarnings("ignore", message=".*`ReadOnly` qualifier.*")
### INIT VARIABLES #########################
import threading
import os

View file

@ -108,7 +108,6 @@ class GoogleGenAIStreamWrapper(AdapterCompletionStreamWrapper):
def __init__(self, completion_stream: object):
self.sent_first_chunk = False
# State tracking for accumulating partial tool calls
self.accumulated_tool_calls = dict[int, _ToolCallAccumulator]()
self._returned_response = False
super().__init__(completion_stream)

View file

@ -62,6 +62,8 @@ class GenAIMapper:
GenAI.RESPONSE_TIME_TO_FIRST_CHUNK: lambda d: d.time_to_first_chunk_seconds,
GenAI.USAGE_INPUT_TOKENS: lambda d: d.usage.input_tokens,
GenAI.USAGE_OUTPUT_TOKENS: lambda d: d.usage.output_tokens,
GenAI.USAGE_CACHE_CREATION_INPUT_TOKENS: lambda d: d.usage.cache_creation_input_tokens,
GenAI.USAGE_CACHE_READ_INPUT_TOKENS: lambda d: d.usage.cache_read_input_tokens,
Error.TYPE: lambda d: d.error.error_type if d.error else None,
Server.ADDRESS: lambda d: d.server.address if d.server else None,
Server.PORT: lambda d: d.server.port if d.server else None,

View file

@ -95,6 +95,22 @@ class LLMUsage:
input_tokens: int | None = None
output_tokens: int | None = None
total_tokens: int | None = None
cache_creation_input_tokens: int | None = None
cache_read_input_tokens: int | None = None
@classmethod
def from_standard_logging_payload(cls, payload: StandardLoggingPayload) -> LLMUsage:
# Cache token counts only exist on the raw provider usage object under metadata
metadata: Final[Mapping[str, object]] = payload.get("metadata") or {}
raw_usage: Final = metadata.get("usage_object")
usage_object: Final[Mapping[str, object]] = raw_usage if isinstance(raw_usage, Mapping) else {}
return cls(
input_tokens=as_int(payload.get("prompt_tokens")),
output_tokens=as_int(payload.get("completion_tokens")),
total_tokens=as_int(payload.get("total_tokens")),
cache_creation_input_tokens=as_int(usage_object.get("cache_creation_input_tokens")),
cache_read_input_tokens=as_int(usage_object.get("cache_read_input_tokens")),
)
@dataclass(frozen=True)
@ -363,11 +379,7 @@ class LLMCallSpanData:
response_model=context.response_model,
response_id=as_str(response.get("id")),
request_params=LLMRequestParams.from_model_parameters(params),
usage=LLMUsage(
input_tokens=as_int(payload.get("prompt_tokens")),
output_tokens=as_int(payload.get("completion_tokens")),
total_tokens=as_int(payload.get("total_tokens")),
),
usage=LLMUsage.from_standard_logging_payload(payload),
finish_reasons=finish_reasons,
error=_parse_error(payload),
response_cost=as_float(payload.get("response_cost")),

View file

@ -110,6 +110,8 @@ class GenAI:
# usage
USAGE_INPUT_TOKENS: Final = "gen_ai.usage.input_tokens"
USAGE_OUTPUT_TOKENS: Final = "gen_ai.usage.output_tokens"
USAGE_CACHE_CREATION_INPUT_TOKENS: Final = "gen_ai.usage.cache_creation.input_tokens"
USAGE_CACHE_READ_INPUT_TOKENS: Final = "gen_ai.usage.cache_read.input_tokens"
# content (opt-in, gated by capture mode)
INPUT_MESSAGES: Final = "gen_ai.input.messages"
OUTPUT_MESSAGES: Final = "gen_ai.output.messages"

View file

@ -144,7 +144,6 @@ def _is_choice_non_empty(choice: StreamingChoices) -> bool:
# Check model_extra for dynamically added fields on the choice
choice_extra_fields: Final[Mapping[str, object]] = choice.model_extra or {}
for extra_field_name, extra_field_value in choice_extra_fields.items():
# Skip certain structural fields that are just default/None placeholders
if extra_field_name == "index" and extra_field_value == 0:
continue
if extra_field_name in {"finish_reason", "logprobs"} and extra_field_value is None:
@ -192,7 +191,6 @@ def _is_delta_non_empty(delta: Delta) -> bool:
# Check model_extra for dynamically added fields (this is where Pydantic stores them)
delta_extra_fields: Final[Mapping[str, object]] = delta.model_extra or {}
for extra_field_value in delta_extra_fields.values():
# Even structural fields are meaningful if they have actual content
if _has_meaningful_content(extra_field_value):
return True

View file

@ -693,6 +693,26 @@ def _count_document_tokens(
)
def _count_file_tokens(
file_value: object,
count_function: TokenCounterFunction,
use_default_image_token_count: bool,
) -> int:
"""An OpenAI `file` block is the chat-completions spelling of a document, so it prices like one."""
if not isinstance(file_value, Mapping):
return 0
filename: Final = file_value.get("filename")
file_data: Final = file_value.get("file_data")
name_tokens: Final = count_function(filename) if isinstance(filename, str) and filename else 0
if not isinstance(file_data, str) or not file_data:
return name_tokens
return name_tokens + calculate_img_tokens(
data=file_data,
mode="auto",
use_default_image_token_count=use_default_image_token_count,
)
def _count_anthropic_content(
content: Mapping[str, Any],
count_function: TokenCounterFunction,
@ -778,6 +798,12 @@ def _count_content_list(
use_default_image_token_count,
default_token_count,
)
elif c["type"] == "file":
num_tokens += _count_file_tokens(
c.get("file"),
count_function,
use_default_image_token_count,
)
elif c["type"] in ("tool_use", "tool_result"):
num_tokens += _count_anthropic_content(
c,
@ -807,7 +833,7 @@ def _count_content_list(
raise ValueError(
f"Invalid content item type: {content_type}. "
f"Expected str or dict with 'type' field "
f"(text, image_url, image, document, tool_use, tool_result, thinking, tool_reference)."
f"(text, image_url, image, document, file, tool_use, tool_result, thinking, tool_reference)."
)
return num_tokens
except Exception as e:

View file

@ -859,12 +859,28 @@ class AnthropicMessagesHandler(BaseTranslation):
@staticmethod
def _image_sources(block: Mapping[str, object]) -> tuple[str, ...]:
"""Normalize an Anthropic image block into strings a guardrail can read.
base64 becomes a data URI so the format travels with the payload, which is what
the OpenAI path already puts in this field. A file source yields nothing: those
bytes live behind the Files API and this extractor has no client to fetch them.
"""
source: Final = block.get("source")
if not isinstance(source, Mapping):
return ()
# Could be base64 or url
source_type: Final = source.get("type")
if source_type == "url":
url: Final = source.get("url")
return (url,) if isinstance(url, str) and url else ()
data: Final = source.get("data")
return (data,) if data else ()
if not isinstance(data, str) or not data:
return ()
media_type: Final = source.get("media_type")
if isinstance(media_type, str) and media_type:
return (f"data:{media_type};base64,{data}",)
return (data,)
async def _apply_guardrail_responses_to_input(
self,

View file

@ -7,13 +7,18 @@ This uses aws_sdk_bedrock_runtime for bidirectional streaming with Nova Sonic.
import asyncio
import contextlib
import json
from collections.abc import AsyncIterator, Mapping
from typing import Final, Protocol
from pydantic import JsonValue, TypeAdapter
import litellm
from litellm._logging import _redact_string, verbose_proxy_logger
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.litellm_core_utils.realtime_streaming import DefaultLoggedRealTimeEventTypes
from litellm.types.llms.openai import OpenAIRealtimeEvents
from litellm.types.realtime import RealtimeResponseTransformInput
from ..base_aws_llm import BaseAWSLLM
@ -32,6 +37,17 @@ def _json_str(value: JsonValue) -> str | None:
return value if isinstance(value, str) else None
def _should_log_event(openai_message: Mapping[str, object]) -> bool:
logged_types: Final = (
litellm.logged_real_time_event_types
if litellm.logged_real_time_event_types is not None
else DefaultLoggedRealTimeEventTypes
)
if logged_types == "*":
return True
return openai_message.get("type") in logged_types
class RealtimeClientWebSocket(Protocol):
"""The client-facing websocket surface the realtime bridge talks to."""
@ -205,16 +221,22 @@ class BedrockRealtime(BaseAWSLLM):
)
)
bedrock_to_client_task: Final = asyncio.create_task(
self._forward_bedrock_to_client(
bedrock_stream,
websocket,
transformation_config,
model,
logging_obj,
session_state,
async def forward_bedrock_and_collect_logged_events() -> tuple[OpenAIRealtimeEvents, ...]:
return tuple(
[
event
async for event in self._forward_bedrock_to_client(
bedrock_stream,
websocket,
transformation_config,
model,
logging_obj,
session_state,
)
]
)
)
bedrock_to_client_task: Final = asyncio.create_task(forward_bedrock_and_collect_logged_events())
# Wait for both tasks to complete
await asyncio.gather(
@ -223,6 +245,27 @@ class BedrockRealtime(BaseAWSLLM):
return_exceptions=True,
)
forwarded_logged_events: Final = (
bedrock_to_client_task.result()
if not bedrock_to_client_task.cancelled() and bedrock_to_client_task.exception() is None
else ()
)
logged_events: Final = (
*forwarded_logged_events,
*(
leftover_event
for leftover_event in transformation_config.leftover_usage_done_events()
if _should_log_event(leftover_event)
),
)
if logged_events:
GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
logging_obj.dispatch_success_handlers(
list(logged_events), # mutable-ok: realtime spend logging requires a list result
prefer_async_handlers=True,
)
)
except Exception as e:
verbose_proxy_logger.exception("Error in BedrockRealtime.async_realtime: %s", e)
try:
@ -304,8 +347,8 @@ class BedrockRealtime(BaseAWSLLM):
model: str,
logging_obj: LiteLLMLogging,
session_state: RealtimeResponseTransformInput,
):
"""Forward messages from Bedrock stream to client WebSocket."""
) -> AsyncIterator[OpenAIRealtimeEvents]:
"""Forward messages from Bedrock to the client, yielding the ones to record for spend logging."""
try:
while True:
# Receive from Bedrock
@ -353,11 +396,14 @@ class BedrockRealtime(BaseAWSLLM):
)
# Send transformed messages to client
openai_messages = transformed_response.get("response", [])
response_value = transformed_response["response"]
openai_messages = response_value if isinstance(response_value, list) else (response_value,)
for openai_message in openai_messages:
message_json = json.dumps(openai_message)
await client_ws.send_text(message_json)
verbose_proxy_logger.debug("Bedrock Realtime: Sent to client: %s", message_json[:200])
if _should_log_event(openai_message):
yield openai_message
except Exception as e:
verbose_proxy_logger.debug("Bedrock to client forwarding ended: %s", e, exc_info=True)

View file

@ -7,7 +7,7 @@ Transforms between OpenAI Realtime API format and Bedrock Nova Sonic format.
import base64
import json
import uuid as uuid_lib
from typing import Any, Final
from typing import Any, Final, cast
from pydantic import BaseModel
@ -20,29 +20,54 @@ from litellm.types.llms.openai import (
OpenAIRealtimeContentPartDone,
OpenAIRealtimeDoneEvent,
OpenAIRealtimeEvents,
OpenAIRealtimeInputAudioBufferSpeechEvent,
OpenAIRealtimeInputAudioTranscriptionCompleted,
OpenAIRealtimeInputAudioTranscriptionDelta,
OpenAIRealtimeOutputItemDone,
OpenAIRealtimeResponseAudioDone,
OpenAIRealtimeResponseContentPartAdded,
OpenAIRealtimeResponseDelta,
OpenAIRealtimeResponseDoneObject,
OpenAIRealtimeResponseTextDone,
OpenAIRealtimeResponseUsage,
OpenAIRealtimeStreamResponseBaseObject,
OpenAIRealtimeStreamResponseOutputItemAdded,
OpenAIRealtimeStreamSession,
OpenAIRealtimeStreamSessionEvents,
OpenAIRealtimeUsageTokenDetails,
)
from litellm.types.realtime import (
ALL_DELTA_TYPES,
RealtimeResponseTransformInput,
RealtimeResponseTypedDict,
)
from litellm.utils import get_empty_usage
class BedrockContentEnd(BaseModel):
stopReason: str | None = None
class BedrockUsageTokenDetails(BaseModel):
speechTokens: int = 0
textTokens: int = 0
class BedrockUsageDetailsTotal(BaseModel):
input: BedrockUsageTokenDetails = BedrockUsageTokenDetails()
output: BedrockUsageTokenDetails = BedrockUsageTokenDetails()
class BedrockUsageDetails(BaseModel):
total: BedrockUsageDetailsTotal = BedrockUsageDetailsTotal()
class BedrockUsageEvent(BaseModel):
totalInputTokens: int = 0
totalOutputTokens: int = 0
totalTokens: int = 0
details: BedrockUsageDetails = BedrockUsageDetails()
TRIGGER_AUDIO_SAMPLE_RATE_HERTZ: Final = 16000
TRIGGER_AUDIO_BYTES_PER_SECOND: Final = TRIGGER_AUDIO_SAMPLE_RATE_HERTZ * 2
TRIGGER_LEADING_SILENCE: Final = bytes(TRIGGER_AUDIO_BYTES_PER_SECOND // 2)
@ -87,6 +112,15 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
# Text configuration
self.text_media_type = "text/plain"
# Response-stream state (Bedrock events carry no role on textOutput,
# so the USER/ASSISTANT split from contentStart is tracked here)
self._user_transcript_active = False
self._user_transcript_generation_stage: str | None = None
self._user_item_id: str | None = None
self._user_transcript_buffer = ""
self._cumulative_usage = BedrockUsageEvent()
self._reported_usage = BedrockUsageEvent()
def validate_environment(self, headers: dict, model: str, api_key: str | None = None) -> dict:
"""Validate environment - no special validation needed for Bedrock."""
return headers
@ -691,6 +725,11 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
role: Final = content_start.get("role")
if role != "ASSISTANT":
if role == "USER" and content_start.get("type") == "TEXT":
self._user_transcript_active = True
self._user_transcript_generation_stage = self._parse_generation_stage(
content_start.get("additionalModelFields")
)
return (
[],
current_response_id,
@ -700,6 +739,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
)
verbose_logger.debug("Handling ASSISTANT contentStart")
is_new_response: Final = current_response_id is None
# Initialize IDs if needed
if not current_response_id:
@ -715,7 +755,8 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
returned_messages: Final[list[OpenAIRealtimeEvents]] = []
# Send response.created
# Send response.created only once per response (a response can contain
# multiple content blocks, e.g. TEXT then AUDIO)
response_created: Final = OpenAIRealtimeStreamResponseBaseObject(
type="response.created",
event_id=f"event_{uuid.uuid4()}",
@ -727,7 +768,8 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
"conversation_id": current_conversation_id,
},
)
returned_messages.append(response_created)
if is_new_response:
returned_messages.append(response_created)
# Send response.output_item.added
output_item_added: Final = OpenAIRealtimeStreamResponseOutputItemAdded(
@ -767,6 +809,108 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
current_delta_type,
)
@staticmethod
def _parse_generation_stage(additional_model_fields: object) -> str | None:
if not isinstance(additional_model_fields, str):
return None
try:
parsed: Final = json.loads(additional_model_fields)
except json.JSONDecodeError:
return None
stage: Final = parsed.get("generationStage") if isinstance(parsed, dict) else None
return stage if isinstance(stage, str) else None
def _current_user_item_id(self, new_utterance: bool = False) -> str:
"""Item id shared by all events of one user utterance (speech boundaries and transcript)."""
if new_utterance or self._user_item_id is None:
self._user_item_id = f"item_{uuid.uuid4()}"
return self._user_item_id
def transform_user_speech_event(self, is_speech_start: bool) -> tuple[OpenAIRealtimeEvents, ...]:
"""Transform Bedrock userSpeechStart/userSpeechEnd to OpenAI speech boundary events."""
verbose_logger.debug("Handling userSpeech%s", "Start" if is_speech_start else "End")
speech_event: Final[OpenAIRealtimeInputAudioBufferSpeechEvent] = {
"type": "input_audio_buffer.speech_started" if is_speech_start else "input_audio_buffer.speech_stopped",
"event_id": f"event_{uuid.uuid4()}",
"item_id": self._current_user_item_id(new_utterance=is_speech_start),
}
return (speech_event,)
def transform_usage_event(self, usage_event: BedrockUsageEvent) -> None:
"""Record Bedrock's session-cumulative usage totals for the next response.done."""
verbose_logger.debug("Handling usageEvent")
self._cumulative_usage = usage_event
def _take_usage_delta(self) -> OpenAIRealtimeResponseUsage:
"""Usage for the response now completing: cumulative totals minus what prior response.done events reported."""
prior: Final = self._reported_usage
latest: Final = self._cumulative_usage
self._reported_usage = latest
input_details: Final[OpenAIRealtimeUsageTokenDetails] = {
"audio_tokens": latest.details.total.input.speechTokens - prior.details.total.input.speechTokens,
"text_tokens": latest.details.total.input.textTokens - prior.details.total.input.textTokens,
"cached_tokens": 0,
}
output_details: Final[OpenAIRealtimeUsageTokenDetails] = {
"audio_tokens": latest.details.total.output.speechTokens - prior.details.total.output.speechTokens,
"text_tokens": latest.details.total.output.textTokens - prior.details.total.output.textTokens,
}
usage_delta: Final[OpenAIRealtimeResponseUsage] = {
"input_tokens": latest.totalInputTokens - prior.totalInputTokens,
"output_tokens": latest.totalOutputTokens - prior.totalOutputTokens,
"total_tokens": latest.totalTokens - prior.totalTokens,
"input_token_details": input_details,
"output_token_details": output_details,
}
return usage_delta
def leftover_usage_done_events(self) -> tuple[OpenAIRealtimeEvents, ...]:
"""Logged-only response.done for usage Bedrock reports after the final turn's contentEnd."""
if self._cumulative_usage == self._reported_usage:
return ()
usage: Final = self._take_usage_delta()
leftover_done: Final = OpenAIRealtimeDoneEvent(
type="response.done",
event_id=f"event_{uuid.uuid4()}",
response=OpenAIRealtimeResponseDoneObject(
object="realtime.response",
id=f"resp_{uuid.uuid4()}",
status="completed",
conversation_id=f"conv_{uuid.uuid4()}",
usage=dict(usage), # mutable-ok: OpenAIRealtimeResponseDoneObject types usage as plain dict
),
)
return (leftover_done,)
def transform_user_transcript_event(self, transcript: str) -> tuple[OpenAIRealtimeEvents, ...]:
"""Transform a USER-role Bedrock textOutput (ASR transcript) to an OpenAI transcription delta."""
verbose_logger.debug("Handling USER textOutput (ASR transcript)")
delta_event: Final[OpenAIRealtimeInputAudioTranscriptionDelta] = {
"type": "conversation.item.input_audio_transcription.delta",
"event_id": f"event_{uuid.uuid4()}",
"item_id": self._current_user_item_id(),
"content_index": 0,
"delta": transcript,
}
if self._user_transcript_generation_stage != "SPECULATIVE":
self._user_transcript_buffer += transcript
return (delta_event,)
def user_transcript_completed_events(self) -> tuple[OpenAIRealtimeEvents, ...]:
"""One completed event with the full transcript once the FINAL user content block ends."""
transcript: Final = self._user_transcript_buffer
if not transcript:
return ()
self._user_transcript_buffer = ""
completed_event: Final[OpenAIRealtimeInputAudioTranscriptionCompleted] = {
"type": "conversation.item.input_audio_transcription.completed",
"event_id": f"event_{uuid.uuid4()}",
"item_id": self._current_user_item_id(),
"content_index": 0,
"transcript": transcript,
}
return (completed_event,)
def transform_text_output_event(
self,
event: dict,
@ -985,7 +1129,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
if not current_response_id or not current_conversation_id:
return [], None, None, None
usage_obj: Final = get_empty_usage()
usage: Final = self._take_usage_delta()
response_done: Final = OpenAIRealtimeDoneEvent(
type="response.done",
event_id=f"event_{uuid.uuid4()}",
@ -995,11 +1139,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
status="completed",
output=[],
conversation_id=current_conversation_id,
usage={
"prompt_tokens": usage_obj.prompt_tokens,
"completion_tokens": usage_obj.completion_tokens,
"total_tokens": usage_obj.total_tokens,
},
usage=dict(usage),
),
)
@ -1042,8 +1182,6 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
# Create a function call arguments done event
# This is a custom event format that matches what clients expect
from typing import cast
function_call_event: Final[dict[str, Any]] = {
"type": "response.function_call_arguments.done",
"event_id": f"event_{uuid.uuid4()}",
@ -1194,18 +1332,26 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
returned_messages.extend(events)
elif "textOutput" in event:
events, current_delta_chunks = self.transform_text_output_event(
event,
current_output_item_id,
current_response_id,
current_delta_chunks,
)
returned_messages.extend(events)
if self._user_transcript_active:
returned_messages.extend(self.transform_user_transcript_event(event["textOutput"].get("content", "")))
else:
events, current_delta_chunks = self.transform_text_output_event(
event,
current_output_item_id,
current_response_id,
current_delta_chunks,
)
returned_messages.extend(events)
elif "audioOutput" in event:
events = self.transform_audio_output_event(event, current_output_item_id, current_response_id)
returned_messages.extend(events)
elif "contentEnd" in event and self._user_transcript_active:
self._user_transcript_active = False
self._user_transcript_generation_stage = None
returned_messages.extend(self.user_transcript_completed_events())
elif "contentEnd" in event:
events, current_delta_chunks = self.transform_content_end_event(
event,
@ -1224,6 +1370,12 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
) = self._response_done_events(current_response_id, current_conversation_id)
returned_messages.extend(done_events)
elif "userSpeechStart" in event or "userSpeechEnd" in event:
returned_messages.extend(self.transform_user_speech_event("userSpeechStart" in event))
elif "usageEvent" in event:
self.transform_usage_event(BedrockUsageEvent.model_validate(event["usageEvent"]))
elif "toolUse" in event:
events, tool_call_id, tool_name = self.transform_tool_use_event(
event, current_output_item_id, current_response_id

View file

@ -9,7 +9,7 @@ response parsing, and streaming chunk parsing for models served with
import datetime
import json
from collections.abc import Iterable, Mapping, Sequence
from typing import Any, Final
from typing import Final
import httpx
from pydantic import JsonValue, TypeAdapter, ValidationError
@ -76,7 +76,7 @@ def _content_text(content: str | Iterable[Mapping[str, object]] | None) -> str:
return str(content)
def _extract_text_content(content: Any) -> str:
def _extract_text_content(content: str | Iterable[Mapping[str, object]] | None) -> str:
"""Return the plain-text representation of a message content value."""
return _content_text(content)

View file

@ -268,6 +268,7 @@ class BaseOpenAILLM:
"max_retries",
"organization",
"api_base",
"workload_identity_config",
)
openai_client_fields: Final = (
BaseOpenAILLM.get_openai_client_initialization_param_fields(client_type=client_type)

View file

@ -51,6 +51,7 @@ from .common_utils import (
drop_params_from_unprocessable_entity_error,
is_output_token_limit_error,
)
from .workload_identity import resolve_openai_workload_identity_config
openaiOSeriesConfig: Final = OpenAIOSeriesConfig()
openAIGPT5Config: Final = OpenAIGPT5Config()
@ -349,6 +350,7 @@ class OpenAIChatCompletion(BaseLLM, BaseOpenAILLM):
client: OpenAI | AsyncOpenAI | None = None,
shared_session: Optional["ClientSession"] = None,
) -> OpenAI | AsyncOpenAI | None:
workload_identity_config: Final = resolve_openai_workload_identity_config(api_key=api_key, api_base=api_base)
client_initialization_params: Final[dict] = locals()
if client is None:
if not isinstance(max_retries, int):
@ -364,28 +366,49 @@ class OpenAIChatCompletion(BaseLLM, BaseOpenAILLM):
if cached_client:
if isinstance(cached_client, OpenAI) or isinstance(cached_client, AsyncOpenAI):
return cached_client
http_client: Final[httpx.Client | httpx.AsyncClient | None] = (
OpenAIChatCompletion._get_async_http_client(shared_session=shared_session)
if is_async
else OpenAIChatCompletion._get_sync_http_client()
)
if is_async:
_new_client: OpenAI | AsyncOpenAI = AsyncOpenAI(
api_key=api_key,
base_url=api_base,
http_client=http_client,
timeout=timeout,
max_retries=max_retries,
organization=organization,
async_http_client: Final = OpenAIChatCompletion._get_async_http_client(shared_session=shared_session)
http_client: httpx.Client | httpx.AsyncClient | None = async_http_client
_new_client: OpenAI | AsyncOpenAI = (
AsyncOpenAI(
workload_identity=workload_identity_config.to_sdk_workload_identity(),
base_url=api_base,
http_client=async_http_client,
timeout=timeout,
max_retries=max_retries,
organization=organization,
)
if workload_identity_config is not None
else AsyncOpenAI(
api_key=api_key,
base_url=api_base,
http_client=async_http_client,
timeout=timeout,
max_retries=max_retries,
organization=organization,
)
)
else:
_new_client = OpenAI(
api_key=api_key,
base_url=api_base,
http_client=http_client,
timeout=timeout,
max_retries=max_retries,
organization=organization,
sync_http_client: Final = OpenAIChatCompletion._get_sync_http_client()
http_client = sync_http_client
_new_client = (
OpenAI(
workload_identity=workload_identity_config.to_sdk_workload_identity(),
base_url=api_base,
http_client=sync_http_client,
timeout=timeout,
max_retries=max_retries,
organization=organization,
)
if workload_identity_config is not None
else OpenAI(
api_key=api_key,
base_url=api_base,
http_client=sync_http_client,
timeout=timeout,
max_retries=max_retries,
organization=organization,
)
)
## SAVE CACHE KEY

View file

@ -4,7 +4,100 @@ OpenAI Responses API token counting transformation logic.
This module handles the transformation of requests to OpenAI's /v1/responses/input_tokens endpoint.
"""
from typing import Any, Final
from collections.abc import Mapping, Sequence
from typing import Any, Final, Literal
from typing_extensions import ReadOnly, TypedDict
class ResponsesInputTextPart(TypedDict):
type: ReadOnly[Literal["input_text"]]
text: ReadOnly[str]
class ResponsesInputImagePart(TypedDict):
type: ReadOnly[Literal["input_image"]]
image_url: ReadOnly[str]
detail: ReadOnly[str]
class ResponsesInputFilePart(TypedDict):
type: ReadOnly[Literal["input_file"]]
filename: ReadOnly[str]
file_data: ReadOnly[str]
ResponsesInputPart = ResponsesInputTextPart | ResponsesInputImagePart | ResponsesInputFilePart
ResponsesContentRole = Literal["user", "assistant"]
def _chat_image_block_to_responses_part(image_url: object) -> ResponsesInputImagePart | None:
url: Final = image_url.get("url") if isinstance(image_url, Mapping) else image_url
if not isinstance(url, str) or not url:
return None
detail: Final = image_url.get("detail") if isinstance(image_url, Mapping) else None
part: Final[ResponsesInputImagePart] = {
"type": "input_image",
"image_url": url,
"detail": detail if isinstance(detail, str) and detail else "auto",
}
return part
def _chat_file_block_to_responses_part(file_value: object) -> ResponsesInputFilePart | None:
"""Only an inline file round trips: OpenAI rejects `file_data` without the `filename` beside it."""
if not isinstance(file_value, Mapping):
return None
filename: Final = file_value.get("filename")
file_data: Final = file_value.get("file_data")
if not isinstance(filename, str) or not filename or not isinstance(file_data, str) or not file_data:
return None
part: Final[ResponsesInputFilePart] = {
"type": "input_file",
"filename": filename,
"file_data": file_data,
}
return part
def _chat_block_to_responses_part(block: object, role: ResponsesContentRole) -> ResponsesInputPart | None:
if isinstance(block, str):
bare: Final[ResponsesInputTextPart] = {"type": "input_text", "text": block}
return bare
if not isinstance(block, Mapping):
return None
match block.get("type"):
case "text":
text_value: Final = block.get("text")
text: Final[ResponsesInputTextPart] = {
"type": "input_text",
"text": text_value if isinstance(text_value, str) else "",
}
return text
case "image_url" if role == "user":
return _chat_image_block_to_responses_part(block.get("image_url"))
case "file" if role == "user":
return _chat_file_block_to_responses_part(block.get("file"))
case _:
return None
def chat_content_blocks_to_responses_content(
content: Sequence[object],
role: ResponsesContentRole,
) -> str | tuple[ResponsesInputPart, ...]:
"""Text-only content collapses to a joined string, which every role accepts and counts identically.
Only a user turn may carry an image or file part: the Responses API rejects any part but
output_text and refusal inside an assistant turn.
"""
parts: Final = tuple(
part for part in (_chat_block_to_responses_part(block, role) for block in content) if part is not None
)
if any(part["type"] != "input_text" for part in parts):
return parts
return "\n".join(part["text"] for part in parts if part["type"] == "input_text")
class OpenAICountTokensConfig:
@ -120,18 +213,13 @@ class OpenAICountTokensConfig:
instructions_parts.append("\n".join(text_parts))
elif role == "user":
if isinstance(content, list):
# Extract text from content blocks for Responses API
text_parts = []
for block in content:
if isinstance(block, dict) and block.get("type") == "text":
text_parts.append(block.get("text", ""))
elif isinstance(block, str):
text_parts.append(block)
content = "\n".join(text_parts)
content = chat_content_blocks_to_responses_content(content, "user")
input_items.append({"role": "user", "content": content})
elif role == "assistant":
# Map tool_calls to Responses API function_call items
tool_calls = msg.get("tool_calls")
if isinstance(content, list):
content = chat_content_blocks_to_responses_content(content, "assistant")
if content:
input_items.append({"role": "assistant", "content": content})
if tool_calls:

View file

@ -21,6 +21,7 @@ from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
from ..common_utils import OpenAIError
from ..workload_identity import get_workload_identity_bearer_token, resolve_openai_workload_identity_config
OPENAI_RESPONSES_API_MIN_MAX_OUTPUT_TOKENS: Final = 16
@ -392,6 +393,14 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
litellm_params = litellm_params or GenericLiteLLMParams()
api_key = litellm_params.api_key or litellm.api_key or litellm.openai_key or get_secret_str("OPENAI_API_KEY")
headers.setdefault("Content-Type", "application/json")
workload_identity_config: Final = (
resolve_openai_workload_identity_config(api_key=api_key, api_base=litellm_params.api_base)
if self.custom_llm_provider is LlmProviders.OPENAI
else None
)
if workload_identity_config is not None:
headers["Authorization"] = f"Bearer {get_workload_identity_bearer_token(workload_identity_config)}"
return headers
headers["Authorization"] = f"Bearer {api_key}"
return headers

View file

@ -0,0 +1,100 @@
from __future__ import annotations
from dataclasses import dataclass
from functools import lru_cache
from typing import TYPE_CHECKING, Final
from urllib.parse import urlparse
import litellm
from litellm.secret_managers.main import get_secret_str, normalize_nonempty_secret_str
from .common_utils import OpenAIError
if TYPE_CHECKING:
from collections.abc import Callable
from openai.auth import SubjectTokenProvider, WorkloadIdentity, WorkloadIdentityAuth
OPENAI_WIF_CLIENT_ID: Final = "litellm"
_OPENAI_API_HOST: Final = "api.openai.com"
_SDK_UPGRADE_MESSAGE: Final = (
"OpenAI workload identity federation requires openai>=2.32.0. "
"Upgrade the installed openai package to use OPENAI_IDENTITY_PROVIDER_ID / "
"OPENAI_SERVICE_ACCOUNT_ID / OPENAI_IDENTITY_TOKEN_FILE."
)
@dataclass(frozen=True, slots=True)
class OpenAIWorkloadIdentityConfig:
identity_provider_id: str
service_account_id: str
token_file: str
def to_sdk_workload_identity(self) -> WorkloadIdentity:
k8s_token_provider: Final = _load_sdk_k8s_token_provider()
workload_identity: Final[WorkloadIdentity] = {
"client_id": OPENAI_WIF_CLIENT_ID,
"identity_provider_id": self.identity_provider_id,
"service_account_id": self.service_account_id,
"provider": k8s_token_provider(self.token_file),
}
return workload_identity
def resolve_openai_workload_identity_config(
api_key: str | None,
api_base: str | None,
) -> OpenAIWorkloadIdentityConfig | None:
static_api_key: Final = normalize_nonempty_secret_str(api_key) or normalize_nonempty_secret_str(
get_secret_str("OPENAI_API_KEY")
)
if static_api_key is not None:
return None
effective_api_base: Final = (
api_base or litellm.api_base or get_secret_str("OPENAI_BASE_URL") or get_secret_str("OPENAI_API_BASE")
)
if not _targets_openai_api(effective_api_base):
return None
identity_provider_id: Final = get_secret_str("OPENAI_IDENTITY_PROVIDER_ID")
service_account_id: Final = get_secret_str("OPENAI_SERVICE_ACCOUNT_ID")
token_file: Final = get_secret_str("OPENAI_IDENTITY_TOKEN_FILE")
if not identity_provider_id or not service_account_id or not token_file:
return None
return OpenAIWorkloadIdentityConfig(
identity_provider_id=identity_provider_id,
service_account_id=service_account_id,
token_file=token_file,
)
def get_workload_identity_bearer_token(config: OpenAIWorkloadIdentityConfig) -> str:
return _workload_identity_auth(config).get_token()
def _targets_openai_api(api_base: str | None) -> bool:
if api_base is None:
return True
parsed: Final = urlparse(api_base)
return parsed.scheme == "https" and parsed.hostname == _OPENAI_API_HOST
@lru_cache(maxsize=16)
def _workload_identity_auth(config: OpenAIWorkloadIdentityConfig) -> WorkloadIdentityAuth:
sdk_workload_identity_auth: Final = _load_sdk_workload_identity_auth()
return sdk_workload_identity_auth(workload_identity=config.to_sdk_workload_identity())
def _load_sdk_workload_identity_auth() -> type[WorkloadIdentityAuth]:
try:
from openai.auth import WorkloadIdentityAuth as sdk_workload_identity_auth
except ImportError as e:
raise OpenAIError(status_code=500, message=_SDK_UPGRADE_MESSAGE) from e
return sdk_workload_identity_auth
def _load_sdk_k8s_token_provider() -> Callable[[str], SubjectTokenProvider]:
try:
from openai.auth import k8s_service_account_token_provider
except ImportError as e:
raise OpenAIError(status_code=500, message=_SDK_UPGRADE_MESSAGE) from e
return k8s_service_account_token_provider

View file

@ -160,14 +160,12 @@ class RunwayMLVideoConfig(BaseVideoConfig):
**self._prompt_image_param(video_create_optional_params),
**self._ratio_param(video_create_optional_params),
**self._duration_param(video_create_optional_params),
# Pass through other parameters that aren't OpenAI-specific
**{key: value for key, value in video_create_optional_params.items() if key not in supported_openai_params},
}
@staticmethod
def _prompt_image_param(video_create_optional_params: VideoCreateOptionalRequestParams) -> Mapping[str, object]:
# Handle input_reference parameter - map to promptImage
# RunwayML supports URLs and data URIs directly
if "input_reference" in video_create_optional_params:
return {"promptImage": video_create_optional_params["input_reference"]}
return {}

View file

@ -182,7 +182,6 @@ class VertexAIGeminiImageEditConfig(BaseImageEditConfig, VertexLLM):
else None
)
# Generation config with proper structure for image editing
generation_config: Final[dict[str, object]] = {
key: value for key, value in (("response_modalities", ["IMAGE"]), ("image_config", image_config)) if value
}

View file

@ -203,7 +203,6 @@ class VertexVectorStoreConfig(BaseVectorStoreConfig, VertexBase):
if value is not None
}
# Build the request body for Vertex AI RAG API
query_body: Final[Mapping[str, object]] = {
key: value
for key, value in (("text", query), ("rag_retrieval_config", rag_retrieval_config or None))
@ -294,7 +293,6 @@ class VertexVectorStoreConfig(BaseVectorStoreConfig, VertexBase):
# Add metadata if provided
metadata: Final = vector_store_create_optional_params.get("metadata")
# Build the request body for Vertex AI RAG Corpus creation
request_body: Final[dict[str, object]] = {
key: value
for key, value in (

View file

@ -553,6 +553,26 @@
"supports_response_schema": true,
"supports_vision": true
},
"amazon.nova-sonic-v1:0": {
"input_cost_per_audio_token": 3.4e-06,
"input_cost_per_token": 6e-08,
"litellm_provider": "bedrock",
"mode": "realtime",
"output_cost_per_audio_token": 1.36e-05,
"output_cost_per_token": 2.4e-07,
"supports_audio_input": true,
"supports_audio_output": true
},
"amazon.nova-2-sonic-v1:0": {
"input_cost_per_audio_token": 3e-06,
"input_cost_per_token": 3.3e-07,
"litellm_provider": "bedrock",
"mode": "realtime",
"output_cost_per_audio_token": 1.2e-05,
"output_cost_per_token": 2.75e-06,
"supports_audio_input": true,
"supports_audio_output": true
},
"amazon.rerank-v1:0": {
"input_cost_per_query": 0.001,
"input_cost_per_token": 0.0,
@ -19566,6 +19586,34 @@
"supports_system_messages": true,
"supports_tool_choice": true
},
"friendliai/zai-org/GLM-5.3-Flash": {
"litellm_provider": "friendliai",
"supports_reasoning": true,
"supports_function_calling": true,
"max_input_tokens": 1048576,
"max_tokens": 1048576,
"max_output_tokens": 1048576,
"input_cost_per_token": 1.5e-07,
"output_cost_per_token": 5e-07,
"cache_read_input_token_cost": 3e-08,
"supports_prompt_caching": true,
"reasoning_effort_levels": [
"low",
"high",
"max"
],
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_native_structured_output": true,
"supports_system_messages": true,
"supports_tool_choice": true,
"mode": "chat",
"comment": "Native multimodal GLM model for efficient coding and long-horizon agent tasks",
"source": "https://api.friendli.ai/serverless/v1/models",
"supports_vision": true,
"supports_image_input": true,
"supports_video_input": true
},
"ft:babbage-002": {
"deprecation_date": "2026-10-23",
"input_cost_per_token": 1.6e-06,

View file

@ -422,6 +422,9 @@ class LiteLLMRoutes(enum.Enum):
"/responses/{response_id}/cancel",
"/v1/responses/{response_id}/cancel",
"/openai/v1/responses/{response_id}/cancel",
"/responses/input_tokens",
"/v1/responses/input_tokens",
"/openai/v1/responses/input_tokens",
# vector stores
"/vector_stores",
"/v1/vector_stores",

View file

@ -433,7 +433,7 @@ class HeadroomGuardrail(CustomGuardrail):
payload["model"] = model
try:
raw_response: HttpxResponse = await self.async_handler.post( # pyright: ignore[reportUnknownMemberType]
raw_response: HttpxResponse = await self.async_handler.post( # pyright: ignore[reportUnknownMemberType] # AsyncHTTPHandler.post is untyped
url=f"{self.headroom_api_base}/v1/compress",
json=payload,
headers=self._request_headers(),
@ -570,7 +570,7 @@ class HeadroomGuardrail(CustomGuardrail):
params["query"] = query
try:
raw_response: HttpxResponse = await self.async_handler.get( # pyright: ignore[reportUnknownMemberType]
raw_response: HttpxResponse = await self.async_handler.get( # pyright: ignore[reportUnknownMemberType] # AsyncHTTPHandler.get is untyped
url=f"{self.headroom_api_base}/v1/retrieve/{hash_value}",
params=params,
headers=self._request_headers(),

View file

@ -197,7 +197,7 @@ class RepelloAIGuardrail(CustomGuardrail):
repelloai_response: RepelloAIAnalyzeResponse | None = None
try:
verbose_proxy_logger.debug("RepelloAI Argus request: %s", request)
response: Final[HttpxResponse] = await self.async_handler.post( # pyright: ignore[reportUnknownMemberType]
response: Final[HttpxResponse] = await self.async_handler.post( # pyright: ignore[reportUnknownMemberType] # AsyncHTTPHandler.post is untyped
url=endpoint,
headers={"X-API-Key": self.repelloai_api_key},
json=request,

View file

@ -626,11 +626,7 @@ async def update_end_user(
# get non default values for key
non_default_values: Final = dict[str, object]()
for k, v in data_json.items():
if v is not None and v not in (
[],
{},
0,
): # models default to [], spend defaults to 0, we should not reset these values
if v is not None and ((isinstance(v, bool) and k in data.fields_set()) or v not in ([], {}, 0)):
non_default_values[k] = v
## Get end user table data ##

View file

@ -2820,7 +2820,7 @@ async def gigachat_proxy_route(
request: Request,
fastapi_response: Response,
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
):
) -> Response:
"""
[Docs](https://docs.litellm.ai/docs/pass_through/gigachat)
"""

View file

@ -1,14 +1,18 @@
import asyncio
import json
import time
from collections.abc import AsyncIterator, Mapping
from collections.abc import AsyncIterator, Awaitable, Mapping
from enum import Enum
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, NamedTuple, cast, get_args
from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, cast, get_args
from uuid import uuid4
import fastapi
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi.responses import JSONResponse
from openai.types.responses.response_create_params import ResponseInputParam
from starlette.websockets import WebSocket, WebSocketDisconnect
from typing_extensions import ReadOnly, TypedDict
from litellm._logging import verbose_proxy_logger
from litellm.integrations.custom_guardrail import ModifyResponseException
@ -26,8 +30,13 @@ from litellm.proxy.common_utils.http_parsing_utils import (
_read_request_body,
_safe_set_request_parsed_body,
)
from litellm.types.llms.openai import REASONING_EFFORT, ResponsesAPIResponse
from litellm.types.llms.openai import (
REASONING_EFFORT,
ResponsesAPIOptionalRequestParams,
ResponsesAPIResponse,
)
from litellm.types.responses.main import DeleteResponseResult
from litellm.types.utils import TokenCountResponse
if TYPE_CHECKING:
from litellm.router import Router
@ -35,7 +44,7 @@ if TYPE_CHECKING:
router: Final = APIRouter()
_user_api_key_auth_dep: Final = Depends(user_api_key_auth)
_RESPONSES_TAGS: Final = ["responses"] # mutable-ok: fastapi's route signature requires List[str] tags
_RESPONSES_TAGS: Final[list[str | Enum]] = ["responses"] # mutable-ok: fastapi's route signature requires list tags
_TOOL_PAYLOAD_KEYS: Final[Mapping[str, tuple[str, ...]]] = MappingProxyType(
{
@ -1017,6 +1026,152 @@ async def compact_response(
)
class _ResponsesApiErrorDetail(TypedDict):
message: ReadOnly[str]
type: ReadOnly[str]
param: ReadOnly[str | None]
code: ReadOnly[str | None]
class _ResponsesApiErrorBody(TypedDict):
error: ReadOnly[_ResponsesApiErrorDetail]
class _ResponsesInputTokensResult(TypedDict):
object: ReadOnly[str]
input_tokens: ReadOnly[int]
class _TokenCountPayload(TypedDict):
model: ReadOnly[str]
messages: ReadOnly[tuple[Mapping[str, object], ...]]
tools: ReadOnly[object]
class _TokenCounter(Protocol):
def __call__(self, request: TokenCountRequest, call_endpoint: bool) -> Awaitable[TokenCountResponse]: ...
def _proxy_token_counter() -> _TokenCounter:
from litellm.proxy.proxy_server import token_counter
return token_counter
_token_counter_dep: Final = Depends(_proxy_token_counter)
def _responses_invalid_request_response(message: str, param: str | None, code: str | None) -> JSONResponse:
body: Final[_ResponsesApiErrorBody] = {
"error": {
"message": message,
"type": "invalid_request_error",
"param": param,
"code": code,
}
}
return JSONResponse(status_code=400, content=body)
def _missing_responses_param_response(param: str) -> JSONResponse:
return _responses_invalid_request_response(
message=f"Missing required parameter: '{param}'.",
param=param,
code="missing_required_parameter",
)
def _responses_input_as_token_count_messages(
input_value: str | ResponseInputParam,
instructions: str | None,
) -> tuple[Mapping[str, object], ...]:
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
)
request_params: Final[ResponsesAPIOptionalRequestParams] = {"instructions": instructions}
transformed: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=input_value,
responses_api_request=request_params,
)
return tuple(
message if isinstance(message, dict) else message.model_dump(exclude_none=True) for message in transformed
)
@router.post(
"/v1/responses/input_tokens",
dependencies=(_user_api_key_auth_dep,),
tags=_RESPONSES_TAGS,
)
@router.post(
"/responses/input_tokens",
dependencies=(_user_api_key_auth_dep,),
tags=_RESPONSES_TAGS,
)
@router.post(
"/openai/v1/responses/input_tokens",
dependencies=(_user_api_key_auth_dep,),
tags=_RESPONSES_TAGS,
)
async def responses_input_tokens(
request: Request,
token_counter: _TokenCounter = _token_counter_dep,
):
"""
Count the input tokens of a Responses API request without calling the model.
Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/input-tokens
```bash
curl -X POST http://localhost:4000/v1/responses/input_tokens \
-H "Content-Type: application/json" \
-H "Authorization: Bearer sk-1234" \
-d '{
"model": "gpt-4o",
"input": "Hello, how are you?"
}'
```
Returns: `{"object": "response.input_tokens", "input_tokens": <count>}`
"""
data: Final = await _read_request_body(request=request)
model_name: Final = data.get("model")
input_value: Final = data.get("input")
if not isinstance(model_name, str) or not model_name:
return _missing_responses_param_response("model")
if input_value is None:
return _missing_responses_param_response("input")
if isinstance(input_value, (str, list)) and not input_value:
return _responses_invalid_request_response(
message="""One of "input" or "previous_response_id" or 'prompt' or 'conversation' must be provided.""",
param=None,
code="missing_required_parameter",
)
try:
payload: Final[_TokenCountPayload] = {
"model": model_name,
"messages": _responses_input_as_token_count_messages(
input_value=input_value,
instructions=data.get("instructions"),
),
"tools": data.get("tools"),
}
token_request: Final = TokenCountRequest.model_validate(payload)
except Exception as e:
return _responses_invalid_request_response(
message=f"Invalid request for token counting: {e}", param=None, code=None
)
token_response: Final = await token_counter(request=token_request, call_endpoint=True)
result: Final[_ResponsesInputTokensResult] = {
"object": "response.input_tokens",
"input_tokens": token_response.total_tokens,
}
return result
@router.post(
"/v1/responses/{response_id}/cancel",
dependencies=[Depends(user_api_key_auth)],

View file

@ -144,10 +144,8 @@ async def background_streaming_task(
# Process streaming response following OpenAI events format
# https://platform.openai.com/docs/api-reference/responses-streaming
output_items: Final = dict[str, _OutputItem]() # Track output items by ID
accumulated_text: Final = dict[
tuple[str, int], str
]() # Track accumulated text deltas by (item_id, content_index)
output_items: Final = dict[str, _OutputItem]()
accumulated_text: Final = dict[tuple[str, int], str]()
# ResponsesAPIResponse fields to extract from response.completed
usage_data = None
@ -262,7 +260,6 @@ async def background_streaming_task(
if "content" in delta_item:
content_list = delta_item["content"]
if content_index < len(content_list):
# Update existing content part with accumulated text
content_entry = content_list[content_index]
if isinstance(content_entry, dict):
content_entry["text"] = accumulated_text[key]

View file

@ -172,7 +172,14 @@ async def reserve_budget_for_request(
) -> dict | None:
if valid_token is None or not RouteChecks.is_llm_api_route(route=route):
return None
if route in {"/models", "/v1/models", "/utils/token_counter"}:
if route in {
"/models",
"/v1/models",
"/utils/token_counter",
"/responses/input_tokens",
"/v1/responses/input_tokens",
"/openai/v1/responses/input_tokens",
}:
return None
if get_model_from_request(request_body, route, llm_router=llm_router) is None:
return None

View file

@ -186,7 +186,6 @@ class VertexAIRAGIngestion(BaseRAGIngestion, VertexBase):
base_url: Final = get_vertex_base_url(self.location)
url: Final = f"{base_url}/v1beta1/projects/{self.project_id}/locations/{self.location}/ragCorpora"
# Build request body with camelCase keys (Vertex AI API format)
vector_db_config: Final = self.vector_store_config.get("vector_db_config")
embedding_model: Final = self.vector_store_config.get("embedding_model")
embedding_model_config: Final = (
@ -447,7 +446,6 @@ class VertexAIRAGIngestion(BaseRAGIngestion, VertexBase):
# Add max embedding requests per minute if specified
max_embedding_qpm: Final = self.vector_store_config.get("max_embedding_requests_per_min")
# Build request body with camelCase keys (Vertex AI API format)
chunking_config: Final = (
{"chunkSize": chunk_size or 1024, "chunkOverlap": chunk_overlap or 200}
if chunk_size or chunk_overlap

View file

@ -1629,6 +1629,8 @@ class LiteLLMCompletionResponsesConfig:
file_dict["file_id"] = file_id
if item.get("file_data"):
file_dict["file_data"] = item["file_data"]
if item.get("filename"):
file_dict["filename"] = item["filename"]
new_item: Final[dict[str, object]] = {"type": "file", "file": file_dict}
if "cache_control" in item:

View file

@ -2162,6 +2162,42 @@ class OpenAIRealtimeDoneEvent(TypedDict):
type: Literal["response.done"]
class OpenAIRealtimeInputAudioBufferSpeechEvent(TypedDict):
type: ReadOnly[Literal["input_audio_buffer.speech_started", "input_audio_buffer.speech_stopped"]]
event_id: ReadOnly[str]
item_id: ReadOnly[str]
class OpenAIRealtimeInputAudioTranscriptionDelta(TypedDict):
type: ReadOnly[Literal["conversation.item.input_audio_transcription.delta"]]
event_id: ReadOnly[str]
item_id: ReadOnly[str]
content_index: ReadOnly[int]
delta: ReadOnly[str]
class OpenAIRealtimeInputAudioTranscriptionCompleted(TypedDict):
type: ReadOnly[Literal["conversation.item.input_audio_transcription.completed"]]
event_id: ReadOnly[str]
item_id: ReadOnly[str]
content_index: ReadOnly[int]
transcript: ReadOnly[str]
class OpenAIRealtimeUsageTokenDetails(TypedDict):
audio_tokens: ReadOnly[int]
text_tokens: ReadOnly[int]
cached_tokens: NotRequired[ReadOnly[int]]
class OpenAIRealtimeResponseUsage(TypedDict):
input_tokens: ReadOnly[int]
output_tokens: ReadOnly[int]
total_tokens: ReadOnly[int]
input_token_details: NotRequired[ReadOnly[OpenAIRealtimeUsageTokenDetails]]
output_token_details: NotRequired[ReadOnly[OpenAIRealtimeUsageTokenDetails]]
class OpenAIRealtimeEventTypes(Enum):
SESSION_CREATED = "session.created"
# Beta delta event names
@ -2199,6 +2235,9 @@ OpenAIRealtimeEvents = (
| OpenAIRealtimeOutputItemDone
| OpenAIRealtimeFunctionCallArgumentsDone
| OpenAIRealtimeDoneEvent
| OpenAIRealtimeInputAudioBufferSpeechEvent
| OpenAIRealtimeInputAudioTranscriptionDelta
| OpenAIRealtimeInputAudioTranscriptionCompleted
)
OpenAIRealtimeStreamList = list[OpenAIRealtimeEvents]

View file

@ -1,5 +1,5 @@
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:a31344ab2cb8618db84f535eec56f76f6178b142cb92cb2e48676cc2dcebea72
ARG LITELLM_BUILD_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG LITELLM_RUNTIME_IMAGE=cgr.dev/chainguard/wolfi-base@sha256:e624c5d5e42382ce7165ddafcbbf8e6769a24cbd02ea6114b880b05ae5ba2a8d
ARG UV_IMAGE=ghcr.io/astral-sh/uv:0.11.7@sha256:240fb85ab0f263ef12f492d8476aa3a2e4e1e333f7d67fbdd923d00a506a516a
FROM $UV_IMAGE AS uvbin
@ -35,7 +35,7 @@ COPY --from=uvbin /uv /uvx /usr/local/bin/
# instead of nodeenv downloading one whose dynamic deps may not be in Wolfi
# (e.g. Node 26.2.0 needs libatomic). Retry for transient apk.cgr.dev flakes.
RUN for i in 1 2 3; do \
apk add --no-cache bash gcc python3 python3-dev openssl openssl-dev libsndfile nodejs npm && break; \
apk add --no-cache bash gcc python-3.13 python-3.13-dev openssl openssl-dev libsndfile nodejs npm && break; \
[ $i = 3 ] && { echo "apk add failed after 3 retries" >&2; exit 1; }; \
sleep 5; \
done
@ -56,7 +56,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-install-project --no-install-workspace --no-default-groups --no-editable \
--extra proxy \
--extra extra_proxy \
--python python3
--python python3.13
# Stage 2 — copy source and install the project + workspace members.
COPY . .
@ -65,7 +65,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-default-groups --no-editable \
--extra proxy \
--extra extra_proxy \
--python python3
--python python3.13
COPY migrations/run.py /app/run.py
@ -87,7 +87,7 @@ FROM $LITELLM_RUNTIME_IMAGE AS runtime
USER root
RUN for i in 1 2 3; do \
apk add --no-cache bash openssl tzdata python3 nodejs libsndfile libatomic && break; \
apk add --no-cache bash openssl tzdata python-3.13 nodejs libsndfile libatomic && break; \
[ $i = 3 ] && { echo "apk add failed after 3 retries" >&2; exit 1; }; \
sleep 5; \
done

View file

@ -553,6 +553,26 @@
"supports_response_schema": true,
"supports_vision": true
},
"amazon.nova-sonic-v1:0": {
"input_cost_per_audio_token": 3.4e-06,
"input_cost_per_token": 6e-08,
"litellm_provider": "bedrock",
"mode": "realtime",
"output_cost_per_audio_token": 1.36e-05,
"output_cost_per_token": 2.4e-07,
"supports_audio_input": true,
"supports_audio_output": true
},
"amazon.nova-2-sonic-v1:0": {
"input_cost_per_audio_token": 3e-06,
"input_cost_per_token": 3.3e-07,
"litellm_provider": "bedrock",
"mode": "realtime",
"output_cost_per_audio_token": 1.2e-05,
"output_cost_per_token": 2.75e-06,
"supports_audio_input": true,
"supports_audio_output": true
},
"amazon.rerank-v1:0": {
"input_cost_per_query": 0.001,
"input_cost_per_token": 0.0,
@ -19566,6 +19586,34 @@
"supports_system_messages": true,
"supports_tool_choice": true
},
"friendliai/zai-org/GLM-5.3-Flash": {
"litellm_provider": "friendliai",
"supports_reasoning": true,
"supports_function_calling": true,
"max_input_tokens": 1048576,
"max_tokens": 1048576,
"max_output_tokens": 1048576,
"input_cost_per_token": 1.5e-07,
"output_cost_per_token": 5e-07,
"cache_read_input_token_cost": 3e-08,
"supports_prompt_caching": true,
"reasoning_effort_levels": [
"low",
"high",
"max"
],
"supports_parallel_function_calling": true,
"supports_response_schema": true,
"supports_native_structured_output": true,
"supports_system_messages": true,
"supports_tool_choice": true,
"mode": "chat",
"comment": "Native multimodal GLM model for efficient coding and long-horizon agent tasks",
"source": "https://api.friendli.ai/serverless/v1/models",
"supports_vision": true,
"supports_image_input": true,
"supports_video_input": true
},
"ft:babbage-002": {
"deprecation_date": "2026-10-23",
"input_cost_per_token": 1.6e-06,

View file

@ -6,4 +6,4 @@ reason = "diskcache has no fixed release published; remove this entry once one e
[[IgnoredVulns]]
id = "GHSA-h7x2-h6g9-p789"
ignoreUntil = 2026-09-14
reason = "mlflow 3.15.0 has no fixed release published; remove this entry once one exists"
reason = "mlflow has no fixed release published; remove this entry once one exists"

View file

@ -11,7 +11,7 @@
"user": "",
"team_id": "",
"organization_id": "",
"metadata": "{\"applied_guardrails\": [], \"attempted_fallbacks\": null, \"original_model_group\": null, \"batch_models\": null, \"batch_successful_requests\": null, \"batch_failed_requests\": null, \"mcp_tool_call_metadata\": null, \"vector_store_request_metadata\": null, \"routing_decision\": null, \"internal_call_origin\": null, \"guardrail_information\": null, \"compression_savings\": null, \"litellm_gateway_injected_cache\": null, \"usage_object\": {\"completion_tokens\": 20, \"prompt_tokens\": 10, \"total_tokens\": 30, \"completion_tokens_details\": null, \"prompt_tokens_details\": null}, \"model_map_information\": {\"model_map_key\": \"gpt-4o\", \"model_map_value\": {\"key\": \"gpt-4o\", \"max_tokens\": 16384, \"max_input_tokens\": 128000, \"max_output_tokens\": 16384, \"input_cost_per_token\": 2.5e-06, \"cache_creation_input_token_cost\": null, \"cache_read_input_token_cost\": 1.25e-06, \"input_cost_per_character\": null, \"input_cost_per_token_above_128k_tokens\": null, \"input_cost_per_token_above_200k_tokens\": null, \"input_cost_per_query\": null, \"input_cost_per_second\": null, \"input_cost_per_audio_token\": null, \"input_cost_per_token_batches\": 1.25e-06, \"output_cost_per_token_batches\": 5e-06, \"output_cost_per_token\": 1e-05, \"output_cost_per_audio_token\": null, \"output_cost_per_character\": null, \"output_cost_per_token_above_128k_tokens\": null, \"output_cost_per_character_above_128k_tokens\": null, \"output_cost_per_token_above_200k_tokens\": null, \"output_cost_per_second\": null, \"output_cost_per_image\": null, \"output_vector_size\": null, \"litellm_provider\": \"openai\", \"mode\": \"chat\", \"supports_system_messages\": true, \"supports_response_schema\": true, \"supports_vision\": true, \"supports_function_calling\": true, \"supports_tool_choice\": true, \"supports_assistant_prefill\": false, \"supports_prompt_caching\": true, \"supports_audio_input\": false, \"supports_audio_output\": false, \"supports_pdf_input\": false, \"supports_embedding_image_input\": false, \"supports_native_streaming\": null, \"supports_web_search\": true, \"supports_reasoning\": false, \"search_context_cost_per_query\": {\"search_context_size_low\": 0.03, \"search_context_size_medium\": 0.035, \"search_context_size_high\": 0.05}, \"tpm\": null, \"rpm\": null, \"supported_openai_params\": [\"frequency_penalty\", \"logit_bias\", \"logprobs\", \"top_logprobs\", \"max_tokens\", \"max_completion_tokens\", \"modalities\", \"prediction\", \"n\", \"presence_penalty\", \"seed\", \"stop\", \"stream\", \"stream_options\", \"temperature\", \"top_p\", \"tools\", \"tool_choice\", \"function_call\", \"functions\", \"max_retries\", \"extra_headers\", \"parallel_tool_calls\", \"audio\", \"response_format\", \"user\"]}}, \"additional_usage_values\": {\"completion_tokens_details\": null, \"prompt_tokens_details\": null}, \"user_api_key\": null, \"user_api_key_alias\": null, \"user_api_key_team_id\": null, \"user_api_key_project_id\": null, \"user_api_key_project_alias\": null, \"user_api_key_org_id\": null, \"user_api_key_user_id\": null, \"user_api_key_team_alias\": null, \"spend_logs_metadata\": null, \"requester_ip_address\": null, \"status\": null, \"proxy_server_request\": null, \"error_information\": null, \"attempted_retries\": null, \"max_retries\": null}",
"metadata": "{\"applied_guardrails\": [], \"attempted_fallbacks\": null, \"original_model_group\": null, \"batch_models\": null, \"batch_successful_requests\": null, \"batch_failed_requests\": null, \"mcp_tool_call_metadata\": null, \"vector_store_request_metadata\": null, \"routing_decision\": null, \"internal_call_origin\": null, \"router_metadata\": null, \"guardrail_information\": null, \"compression_savings\": null, \"litellm_gateway_injected_cache\": null, \"usage_object\": {\"completion_tokens\": 20, \"prompt_tokens\": 10, \"total_tokens\": 30, \"completion_tokens_details\": null, \"prompt_tokens_details\": null}, \"model_map_information\": {\"model_map_key\": \"gpt-4o\", \"model_map_value\": {\"key\": \"gpt-4o\", \"max_tokens\": 16384, \"max_input_tokens\": 128000, \"max_output_tokens\": 16384, \"input_cost_per_token\": 2.5e-06, \"cache_creation_input_token_cost\": null, \"cache_read_input_token_cost\": 1.25e-06, \"input_cost_per_character\": null, \"input_cost_per_token_above_128k_tokens\": null, \"input_cost_per_token_above_200k_tokens\": null, \"input_cost_per_query\": null, \"input_cost_per_second\": null, \"input_cost_per_audio_token\": null, \"input_cost_per_token_batches\": 1.25e-06, \"output_cost_per_token_batches\": 5e-06, \"output_cost_per_token\": 1e-05, \"output_cost_per_audio_token\": null, \"output_cost_per_character\": null, \"output_cost_per_token_above_128k_tokens\": null, \"output_cost_per_character_above_128k_tokens\": null, \"output_cost_per_token_above_200k_tokens\": null, \"output_cost_per_second\": null, \"output_cost_per_image\": null, \"output_vector_size\": null, \"litellm_provider\": \"openai\", \"mode\": \"chat\", \"supports_system_messages\": true, \"supports_response_schema\": true, \"supports_vision\": true, \"supports_function_calling\": true, \"supports_tool_choice\": true, \"supports_assistant_prefill\": false, \"supports_prompt_caching\": true, \"supports_audio_input\": false, \"supports_audio_output\": false, \"supports_pdf_input\": false, \"supports_embedding_image_input\": false, \"supports_native_streaming\": null, \"supports_web_search\": true, \"supports_reasoning\": false, \"search_context_cost_per_query\": {\"search_context_size_low\": 0.03, \"search_context_size_medium\": 0.035, \"search_context_size_high\": 0.05}, \"tpm\": null, \"rpm\": null, \"supported_openai_params\": [\"frequency_penalty\", \"logit_bias\", \"logprobs\", \"top_logprobs\", \"max_tokens\", \"max_completion_tokens\", \"modalities\", \"prediction\", \"n\", \"presence_penalty\", \"seed\", \"stop\", \"stream\", \"stream_options\", \"temperature\", \"top_p\", \"tools\", \"tool_choice\", \"function_call\", \"functions\", \"max_retries\", \"extra_headers\", \"parallel_tool_calls\", \"audio\", \"response_format\", \"user\"]}}, \"additional_usage_values\": {\"completion_tokens_details\": null, \"prompt_tokens_details\": null}, \"user_api_key\": null, \"user_api_key_alias\": null, \"user_api_key_team_id\": null, \"user_api_key_project_id\": null, \"user_api_key_project_alias\": null, \"user_api_key_org_id\": null, \"user_api_key_user_id\": null, \"user_api_key_team_alias\": null, \"spend_logs_metadata\": null, \"requester_ip_address\": null, \"status\": null, \"proxy_server_request\": null, \"error_information\": null, \"attempted_retries\": null, \"max_retries\": null}",
"cache_key": "Cache OFF",
"spend": 0.00022500000000000002,
"total_tokens": 30,

View file

@ -3,6 +3,7 @@ baggage helpers, metrics, the typed coercion helpers, mapper branches, span-name
builders, and the registry validator's failure paths. Needs the OTel SDK."""
import json
from dataclasses import replace
import pytest
@ -215,6 +216,27 @@ def test_genai_mapper_all_request_params():
assert attrs["server.port"] == 443
def test_genai_mapper_cache_token_attrs():
cached = replace(
_full_llm_call(),
usage=LLMUsage(
input_tokens=10,
output_tokens=5,
total_tokens=15,
cache_creation_input_tokens=7,
cache_read_input_tokens=3,
),
)
attrs = GenAIMapper().map(cached)
assert attrs[GenAI.USAGE_CACHE_CREATION_INPUT_TOKENS] == 7
assert attrs[GenAI.USAGE_CACHE_READ_INPUT_TOKENS] == 3
# No cache usage keeps the span sparse: neither key present.
uncached = GenAIMapper().map(_full_llm_call())
assert GenAI.USAGE_CACHE_CREATION_INPUT_TOKENS not in uncached
assert GenAI.USAGE_CACHE_READ_INPUT_TOKENS not in uncached
def test_genai_mapper_stamps_input_output_messages():
data = LLMCallSpanData(
operation=GenAIOperation.CHAT,

View file

@ -525,6 +525,28 @@ def test_llm_call_adapter_extracts_all_fields():
assert data.identity.key_hash == "hsh"
def test_llm_call_adapter_extracts_cache_tokens_from_usage_object():
payload = _sample_payload()
payload["metadata"] = {
**payload["metadata"],
"usage_object": {
"prompt_tokens": 10,
"completion_tokens": 5,
"cache_creation_input_tokens": 7,
"cache_read_input_tokens": 3,
},
}
data = LLMCallSpanData.from_standard_logging_payload(payload)
assert data.usage.cache_creation_input_tokens == 7
assert data.usage.cache_read_input_tokens == 3
def test_llm_call_adapter_cache_tokens_none_without_usage_object():
data = LLMCallSpanData.from_standard_logging_payload(_sample_payload())
assert data.usage.cache_creation_input_tokens is None
assert data.usage.cache_read_input_tokens is None
def test_llm_call_adapter_failure_path():
payload = _sample_payload(
status="failure",

View file

@ -1377,3 +1377,38 @@ def test_anthropic_document_title_and_context_add_their_tokens():
{"type": "document", "source": source},
]
)
def test_openai_file_block_prices_like_the_equivalent_anthropic_document():
"""An inline `file` is a `document` in the chat-completions dialect, so it must price identically, not raise.
Before the fix `file` was missing from the content-block match even though `ChatCompletionFileObject`
is in the union this counter accepts, so every local count of a Responses `input_file` raised
`Invalid content item type: file` and surfaced as a 500 on /v1/responses/input_tokens.
"""
prompt = {"type": "text", "text": "Summarize this file."}
inline_file = {
"type": "file",
"file": {"filename": "report.pdf", "file_data": "data:application/pdf;base64,JVBERi0xLjQK"},
}
document = {
"type": "document",
"title": "report.pdf",
"source": {"type": "base64", "media_type": "application/pdf", "data": "JVBERi0xLjQK"},
}
assert _count_user_content([prompt, inline_file]) == _count_user_content([prompt, document])
assert _count_user_content([prompt, inline_file]) > _count_user_content([prompt])
def test_openai_file_block_without_inline_bytes_counts_what_it_carries():
"""A `file` block naming an uploaded file has no bytes to price, so it adds only the filename's tokens."""
prompt = {"type": "text", "text": "Summarize this file."}
by_id = {"type": "file", "file": {"file_id": "file-abc123"}}
assert _count_user_content([prompt, by_id]) == _count_user_content([prompt])
named = {"type": "file", "file": {"file_id": "file-abc123", "filename": "report.pdf"}}
assert _count_user_content([prompt, named]) == _count_user_content(
[prompt, {"type": "text", "text": "report.pdf"}]
)

View file

@ -1490,6 +1490,92 @@ class MockCanaryMaskingGuardrail(CustomGuardrail):
return inputs
class TestAnthropicMessagesImageSources:
"""An Anthropic image block has three source shapes (`AnthropicMessagesImageParam.source`).
Only the base64 one carries "data", so reading that key alone drops url images
entirely -- for every guardrail consuming GenericGuardrailAPIInputs["images"],
not just Bedrock.
"""
def _data(self, messages):
return {"model": "claude-sonnet-4-5", "messages": messages}
async def _images_seen(self, content) -> list[str]:
handler = AnthropicMessagesHandler()
class ImageRecordingGuardrail(MockCanaryMaskingGuardrail):
def __init__(self):
super().__init__()
self.seen_images: list[str] = [] # mutable-ok: accumulator for the assertion
async def apply_guardrail(self, inputs, request_data, input_type, logging_obj=None):
self.seen_images.extend(inputs.get("images") or [])
return await super().apply_guardrail(inputs, request_data, input_type, logging_obj)
guardrail = ImageRecordingGuardrail()
# The text block is what gets the guardrail invoked at all: a message with
# no text gives the handler nothing to scan, so it never reaches the
# guardrail and every source shape would look equally "dropped".
await handler.process_input_messages(
data=self._data([{"role": "user", "content": [{"type": "text", "text": "describe it"}, *content]}]),
guardrail_to_apply=guardrail,
)
return guardrail.seen_images
@pytest.mark.asyncio
async def test_url_source_reaches_the_guardrail(self):
"""A url source has no "data" key, so it used to yield nothing at all."""
seen = await self._images_seen(
[{"type": "image", "source": {"type": "url", "url": "https://example.com/a.png"}}]
)
assert seen == ["https://example.com/a.png"]
@pytest.mark.asyncio
async def test_base64_source_carries_its_media_type(self):
"""Bare base64 leaves the consumer no way to recover the format.
An API like Bedrock's ApplyGuardrail needs it to build the request, so the
media_type travels with the payload as a data URI.
"""
seen = await self._images_seen(
[{"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "AAAA"}}]
)
assert seen == ["data:image/png;base64,AAAA"]
@pytest.mark.asyncio
async def test_base64_source_without_a_media_type_is_passed_through(self):
"""There is no format to attach, so the payload goes through unchanged."""
seen = await self._images_seen([{"type": "image", "source": {"type": "base64", "data": "AAAA"}}])
assert seen == ["AAAA"]
@pytest.mark.asyncio
async def test_file_source_yields_nothing(self):
"""The bytes live behind the Files API and this extractor has no client.
Documented as a known gap rather than silently handed on as a file_id string,
which a consumer would try to decode as an image.
"""
seen = await self._images_seen([{"type": "image", "source": {"type": "file", "file_id": "file_abc"}}])
assert seen == []
@pytest.mark.asyncio
async def test_a_malformed_source_is_dropped_rather_than_passed_on(self):
seen = await self._images_seen(
[
{"type": "image", "source": {"type": "base64"}},
{"type": "image", "source": {"type": "url"}},
{"type": "image", "source": {"type": "base64", "data": ""}},
]
)
assert seen == []
class TestAnthropicMessagesToolResultScanning:
"""LIT-5251: tool_result blocks carry whatever a client's local tool fetched, so
they are the request-path payload an indirect prompt injection actually arrives in.

View file

@ -6,7 +6,7 @@ from unittest.mock import MagicMock
import pytest
import litellm
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.realtime.handler import BedrockRealtime
from litellm.llms.bedrock.realtime.transformation import BedrockRealtimeConfig
@ -104,12 +104,24 @@ class RealtimeClientWS:
self.closed = True
class ImmediatelyEndingBedrockStream:
def __init__(self):
class ScriptedBedrockReceiver:
def __init__(self, payloads):
self._payloads = list(payloads)
async def receive(self):
if not self._payloads:
return None
payload = self._payloads.pop(0)
return SimpleNamespace(value=SimpleNamespace(bytes_=payload.encode("utf-8")))
class ScriptedBedrockStream:
def __init__(self, payloads):
self.input_stream = FakeInputStream()
self._receiver = ScriptedBedrockReceiver(payloads)
async def await_output(self):
return (None, EndedBedrockReceiver())
return (None, self._receiver)
class FakeStaticCredentialsResolver:
@ -151,7 +163,7 @@ def stub_aws_sdk_client(monkeypatch):
async def invoke_model_with_bidirectional_stream(self, operation_input):
captured["operation_input"] = operation_input
return ImmediatelyEndingBedrockStream()
return ScriptedBedrockStream(captured.get("scripted_payloads", []))
package = types.ModuleType("aws_sdk_bedrock_runtime")
client_module = types.ModuleType("aws_sdk_bedrock_runtime.client")
@ -271,19 +283,132 @@ class TestBedrockRealtimeHandler:
assert "sessionEnd" in event_names
assert stream.input_stream.closed
@pytest.mark.asyncio
async def test_forwarded_events_are_filtered_to_logged_types_for_spend_logging(self):
handler = BedrockRealtime()
stream = ScriptedBedrockStream(
[
json.dumps({"event": {"userSpeechStart": {}}}),
json.dumps({"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}}),
json.dumps({"event": {"textOutput": {"content": "Hi"}}}),
json.dumps({"event": {"contentEnd": {"stopReason": "END_TURN"}}}),
]
)
client_ws = RealtimeClientWS()
logged_events = [
event
async for event in handler._forward_bedrock_to_client(
stream,
client_ws,
BedrockRealtimeConfig(),
"amazon.nova-sonic-v1:0",
FakeLogging(),
{},
)
]
assert [event["type"] for event in logged_events] == ["response.done"]
sent_types = [json.loads(message)["type"] for message in client_ws.sent_to_client]
assert "input_audio_buffer.speech_started" in sent_types
assert "response.text.delta" in sent_types
assert "response.done" in sent_types
assert client_ws.closed
@pytest.mark.asyncio
async def test_logged_event_types_star_collects_every_forwarded_event(self, monkeypatch):
monkeypatch.setattr(litellm, "logged_real_time_event_types", "*")
handler = BedrockRealtime()
stream = ScriptedBedrockStream(
[
json.dumps({"event": {"userSpeechStart": {}}}),
json.dumps({"event": {"userSpeechEnd": {}}}),
]
)
client_ws = RealtimeClientWS()
logged_events = [
event
async for event in handler._forward_bedrock_to_client(
stream,
client_ws,
BedrockRealtimeConfig(),
"amazon.nova-sonic-v1:0",
FakeLogging(),
{},
)
]
assert [event["type"] for event in logged_events] == [
"input_audio_buffer.speech_started",
"input_audio_buffer.speech_stopped",
]
@pytest.mark.asyncio
async def test_trailing_usage_after_last_done_is_dispatched_for_spend(self, stub_aws_sdk_client, monkeypatch):
import litellm.llms.bedrock.realtime.handler as handler_module
dispatched = {}
class RecordingLogging(FakeLogging):
async def dispatch_success_handlers(self, result=None, prefer_async_handlers=False, **kwargs):
dispatched["events"] = result
class RecordingLoggingWorker:
def ensure_initialized_and_enqueue(self, coro):
dispatched["coro"] = coro
monkeypatch.setattr(handler_module, "GLOBAL_LOGGING_WORKER", RecordingLoggingWorker())
stub_aws_sdk_client["scripted_payloads"] = [
json.dumps(
{
"event": {
"usageEvent": {
"totalInputTokens": 3,
"totalOutputTokens": 6,
"totalTokens": 9,
"details": {
"total": {
"input": {"speechTokens": 3, "textTokens": 0},
"output": {"speechTokens": 0, "textTokens": 6},
}
},
}
}
}
)
]
await BedrockRealtime().async_realtime(
model="amazon.nova-sonic-v1:0",
websocket=RealtimeClientWS(),
logging_obj=RecordingLogging(),
aws_region_name="us-east-1",
aws_access_key_id="k",
aws_secret_access_key="s",
)
await dispatched["coro"]
assert [event["type"] for event in dispatched["events"]] == ["response.done"]
usage = dispatched["events"][0]["response"]["usage"]
assert (usage["input_tokens"], usage["output_tokens"], usage["total_tokens"]) == (3, 6, 9)
assert usage["input_token_details"] == {"audio_tokens": 3, "text_tokens": 0, "cached_tokens": 0}
assert usage["output_token_details"] == {"audio_tokens": 0, "text_tokens": 6}
@pytest.mark.asyncio
async def test_bedrock_stream_end_closes_client_websocket(self):
handler = BedrockRealtime()
client_ws = ClosableClientWS()
await handler._forward_bedrock_to_client(
async for _ in handler._forward_bedrock_to_client(
EndedBedrockStream(),
client_ws,
BedrockRealtimeConfig(),
"amazon.nova-sonic-v1:0",
MagicMock(),
{},
)
):
pass
assert client_ws.closed
@ -320,9 +445,7 @@ class TestBedrockRealtimeSessionLifecycle:
[json.dumps({"type": "session.update", "session": {"instructions": "hi", "modalities": ["text"]}})]
)
await handler._forward_client_to_bedrock(
client_ws, stream, config, "amazon.nova-sonic-v1:0", {}, FakeLogging()
)
await handler._forward_client_to_bedrock(client_ws, stream, config, "amazon.nova-sonic-v1:0", {}, FakeLogging())
acked = [json.loads(message) for message in client_ws.sent_to_client]
updated = [event for event in acked if event["type"] == "session.updated"]
@ -334,9 +457,7 @@ class TestBedrockRealtimeSessionLifecycle:
handler = BedrockRealtime()
config = BedrockRealtimeConfig()
stream = FakeBedrockStream()
client_ws = DisconnectingClientWS(
[json.dumps({"type": "session.update", "session": {"instructions": "hi"}})]
)
client_ws = DisconnectingClientWS([json.dumps({"type": "session.update", "session": {"instructions": "hi"}})])
await handler._forward_client_to_bedrock(client_ws, stream, config, "amazon.nova-sonic-v1:0", {})

View file

@ -827,5 +827,310 @@ class TestBedrockRealtimeSessionEvents:
assert event["session"]["modalities"] == ["text", "audio"]
class TestBedrockRealtimeUserEventsAndUsage:
"""Regression tests for #38346: USER ASR transcripts, speech boundary events,
usage propagation, and duplicate response.created"""
@staticmethod
def _run(config, messages):
logging_obj = MagicMock()
logging_obj.litellm_trace_id = "trace_123"
state = {
"session_configuration_request": json.dumps({"configured": True}),
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": [],
"current_item_chunks": [],
"current_delta_type": None,
}
all_events = []
for msg in messages:
result = config.transform_realtime_response(
json.dumps(msg),
"amazon.nova-2-sonic-v1:0",
logging_obj,
realtime_response_transform_input=dict(state),
)
all_events.extend(result["response"])
state.update(
{
"current_output_item_id": result["current_output_item_id"],
"current_response_id": result["current_response_id"],
"current_conversation_id": result["current_conversation_id"],
"current_delta_chunks": result["current_delta_chunks"],
"current_delta_type": result["current_delta_type"],
}
)
return all_events
def test_user_speech_start_and_stop_events(self):
events = self._run(
BedrockRealtimeConfig(),
[{"event": {"userSpeechStart": {}}}, {"event": {"userSpeechEnd": {}}}],
)
assert [e["type"] for e in events] == [
"input_audio_buffer.speech_started",
"input_audio_buffer.speech_stopped",
]
assert all(e["event_id"] and e["item_id"] for e in events)
assert events[0]["item_id"] == events[1]["item_id"]
def test_utterance_lifecycle_shares_one_item_id(self):
events = self._run(
BedrockRealtimeConfig(),
[
{"event": {"userSpeechStart": {}}},
{"event": {"userSpeechEnd": {}}},
{
"event": {
"contentStart": {
"role": "USER",
"type": "TEXT",
"additionalModelFields": json.dumps({"generationStage": "FINAL"}),
}
}
},
{"event": {"textOutput": {"content": "ready"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
],
)
item_ids = {e["item_id"] for e in events if "item_id" in e}
assert len(item_ids) == 1
def test_new_utterance_gets_new_item_id(self):
config = BedrockRealtimeConfig()
first = self._run(config, [{"event": {"userSpeechStart": {}}}, {"event": {"userSpeechEnd": {}}}])
second = self._run(config, [{"event": {"userSpeechStart": {}}}, {"event": {"userSpeechEnd": {}}}])
assert first[0]["item_id"] == first[1]["item_id"]
assert second[0]["item_id"] == second[1]["item_id"]
assert first[0]["item_id"] != second[0]["item_id"]
def test_user_transcript_emits_input_audio_transcription_events(self):
events = self._run(
BedrockRealtimeConfig(),
[
{
"event": {
"contentStart": {
"role": "USER",
"type": "TEXT",
"additionalModelFields": json.dumps({"generationStage": "FINAL"}),
}
}
},
{"event": {"textOutput": {"content": "ready"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
],
)
deltas = [e for e in events if e["type"] == "conversation.item.input_audio_transcription.delta"]
completed = [e for e in events if e["type"] == "conversation.item.input_audio_transcription.completed"]
assert len(deltas) == 1 and deltas[0]["delta"] == "ready"
assert len(completed) == 1 and completed[0]["transcript"] == "ready"
assert deltas[0]["item_id"] == completed[0]["item_id"]
assert not any(e["type"] == "response.text.delta" for e in events)
def test_speculative_user_transcript_emits_delta_only(self):
events = self._run(
BedrockRealtimeConfig(),
[
{
"event": {
"contentStart": {
"role": "USER",
"type": "TEXT",
"additionalModelFields": json.dumps({"generationStage": "SPECULATIVE"}),
}
}
},
{"event": {"textOutput": {"content": "rea"}}},
],
)
assert [e["type"] for e in events] == ["conversation.item.input_audio_transcription.delta"]
def test_user_transcript_state_resets_on_content_end(self):
events = self._run(
BedrockRealtimeConfig(),
[
{"event": {"contentStart": {"role": "USER", "type": "TEXT"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi there"}}},
],
)
text_deltas = [e for e in events if e["type"] == "response.text.delta"]
assert len(text_deltas) == 1 and text_deltas[0]["delta"] == "Hi there"
assert not any(e["type"].startswith("conversation.item.input_audio_transcription") for e in events)
def test_response_created_emitted_once_per_response(self):
events = self._run(
BedrockRealtimeConfig(),
[
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
{"event": {"contentStart": {"role": "ASSISTANT", "type": "AUDIO"}}},
],
)
assert sum(1 for e in events if e["type"] == "response.created") == 1
def test_usage_event_propagates_to_response_done(self):
events = self._run(
BedrockRealtimeConfig(),
[
{
"event": {
"usageEvent": {
"totalInputTokens": 25,
"totalOutputTokens": 40,
"totalTokens": 65,
"details": {
"total": {
"input": {"speechTokens": 20, "textTokens": 5},
"output": {"speechTokens": 30, "textTokens": 10},
}
},
}
}
},
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi"}}},
{"event": {"contentEnd": {"stopReason": "END_TURN"}}},
],
)
done_events = [e for e in events if e["type"] == "response.done"]
assert len(done_events) == 1
usage = done_events[0]["response"]["usage"]
assert usage["input_tokens"] == 25
assert usage["output_tokens"] == 40
assert usage["total_tokens"] == 65
assert usage["input_token_details"]["audio_tokens"] == 20
assert usage["input_token_details"]["text_tokens"] == 5
assert usage["output_token_details"]["audio_tokens"] == 30
assert usage["output_token_details"]["text_tokens"] == 10
def test_response_done_without_usage_event_reports_zero_usage(self):
events = self._run(
BedrockRealtimeConfig(),
[
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi"}}},
{"event": {"contentEnd": {"stopReason": "END_TURN"}}},
],
)
done_events = [e for e in events if e["type"] == "response.done"]
assert len(done_events) == 1
usage = done_events[0]["response"]["usage"]
assert usage["input_tokens"] == 0
assert usage["output_tokens"] == 0
assert usage["total_tokens"] == 0
@staticmethod
def _usage_event(total_input, total_output, in_speech, in_text, out_speech, out_text):
return {
"event": {
"usageEvent": {
"totalInputTokens": total_input,
"totalOutputTokens": total_output,
"totalTokens": total_input + total_output,
"details": {
"total": {
"input": {"speechTokens": in_speech, "textTokens": in_text},
"output": {"speechTokens": out_speech, "textTokens": out_text},
}
},
}
}
}
_ASSISTANT_TURN = (
{"event": {"contentStart": {"role": "ASSISTANT", "type": "TEXT"}}},
{"event": {"textOutput": {"content": "Hi"}}},
{"event": {"contentEnd": {"stopReason": "END_TURN"}}},
)
def test_multi_turn_usage_reports_per_response_deltas_not_cumulative_totals(self):
events = self._run(
BedrockRealtimeConfig(),
[
self._usage_event(25, 40, in_speech=20, in_text=5, out_speech=30, out_text=10),
*self._ASSISTANT_TURN,
self._usage_event(40, 100, in_speech=30, in_text=10, out_speech=75, out_text=25),
*self._ASSISTANT_TURN,
],
)
usages = [e["response"]["usage"] for e in events if e["type"] == "response.done"]
assert len(usages) == 2
assert (usages[0]["input_tokens"], usages[0]["output_tokens"], usages[0]["total_tokens"]) == (25, 40, 65)
assert (usages[1]["input_tokens"], usages[1]["output_tokens"], usages[1]["total_tokens"]) == (15, 60, 75)
assert usages[1]["input_token_details"] == {"audio_tokens": 10, "text_tokens": 5, "cached_tokens": 0}
assert usages[1]["output_token_details"] == {"audio_tokens": 45, "text_tokens": 15}
assert sum(u["total_tokens"] for u in usages) == 140
def test_usage_reported_after_last_response_done_flushes_as_logged_only_done(self):
config = BedrockRealtimeConfig()
self._run(
config,
[
self._usage_event(25, 40, in_speech=20, in_text=5, out_speech=30, out_text=10),
*self._ASSISTANT_TURN,
],
)
assert config.leftover_usage_done_events() == ()
self._run(config, [self._usage_event(25, 46, in_speech=20, in_text=5, out_speech=30, out_text=16)])
leftover = config.leftover_usage_done_events()
assert len(leftover) == 1
assert leftover[0]["type"] == "response.done"
usage = leftover[0]["response"]["usage"]
assert (usage["input_tokens"], usage["output_tokens"], usage["total_tokens"]) == (0, 6, 6)
assert usage["output_token_details"] == {"audio_tokens": 0, "text_tokens": 6}
assert config.leftover_usage_done_events() == ()
def test_final_transcript_fragments_emit_one_completed_with_full_transcript(self):
events = self._run(
BedrockRealtimeConfig(),
[
{
"event": {
"contentStart": {
"role": "USER",
"type": "TEXT",
"additionalModelFields": json.dumps({"generationStage": "FINAL"}),
}
}
},
{"event": {"textOutput": {"content": "What is the "}}},
{"event": {"textOutput": {"content": "capital of France?"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
],
)
deltas = [e for e in events if e["type"] == "conversation.item.input_audio_transcription.delta"]
completed = [e for e in events if e["type"] == "conversation.item.input_audio_transcription.completed"]
assert [d["delta"] for d in deltas] == ["What is the ", "capital of France?"]
assert len(completed) == 1
assert completed[0]["transcript"] == "What is the capital of France?"
assert {e["item_id"] for e in deltas + completed} == {completed[0]["item_id"]}
def test_speculative_transcript_block_end_emits_no_completed(self):
events = self._run(
BedrockRealtimeConfig(),
[
{
"event": {
"contentStart": {
"role": "USER",
"type": "TEXT",
"additionalModelFields": json.dumps({"generationStage": "SPECULATIVE"}),
}
}
},
{"event": {"textOutput": {"content": "rea"}}},
{"event": {"contentEnd": {"stopReason": "PARTIAL_TURN"}}},
],
)
assert [e["type"] for e in events] == ["conversation.item.input_audio_transcription.delta"]
if __name__ == "__main__":
pytest.main([__file__, "-v"])

View file

@ -163,6 +163,240 @@ def test_messages_to_responses_input_with_tool():
}
def test_messages_to_responses_input_preserves_images():
"""An image block must survive the round trip, or OpenAI counts only the text.
A 256x256 image is worth 255 tokens to OpenAI's counting API; dropping it
turned a 268-token request into a 13-token one.
"""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "What is in this image?"},
{
"type": "image_url",
"image_url": {"url": "data:image/png;base64,iVBORw0KGgo=", "detail": "high"},
},
],
}
]
input_items, instructions = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert instructions is None
assert input_items == [
{
"role": "user",
"content": (
{"type": "input_text", "text": "What is in this image?"},
{
"type": "input_image",
"image_url": "data:image/png;base64,iVBORw0KGgo=",
"detail": "high",
},
),
}
]
def test_messages_to_responses_input_image_without_detail_defaults_to_auto():
messages = [
{
"role": "user",
"content": [{"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}}],
}
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items[0]["content"] == (
{"type": "input_image", "image_url": "https://example.com/cat.png", "detail": "auto"},
)
def test_messages_to_responses_input_bare_string_image_url_is_preserved():
messages = [{"role": "user", "content": [{"type": "image_url", "image_url": "https://example.com/cat.png"}]}]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items[0]["content"] == (
{"type": "input_image", "image_url": "https://example.com/cat.png", "detail": "auto"},
)
def test_messages_to_responses_input_text_only_blocks_stay_a_joined_string():
"""Text-only content must keep collapsing to a string so existing counts do not shift."""
messages = [
{
"role": "user",
"content": [{"type": "text", "text": "first"}, {"type": "text", "text": "second"}],
}
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items == [{"role": "user", "content": "first\nsecond"}]
def test_messages_to_responses_input_drops_unmappable_blocks():
"""A block with no Responses API equivalent is skipped, never forwarded verbatim."""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "hi"},
{"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}},
{"type": "input_audio", "input_audio": {"data": "AAAA", "format": "wav"}},
],
}
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items[0]["content"] == (
{"type": "input_text", "text": "hi"},
{"type": "input_image", "image_url": "https://example.com/cat.png", "detail": "auto"},
)
def test_messages_to_responses_input_assistant_blocks_collapse_to_a_string():
"""An assistant turn must never forward chat `text` blocks.
The Responses API only accepts output_text and refusal inside an assistant turn, so
forwarding them 400s the whole request and silently drops the count back to the local
tokenizer, which is exactly what defeats the image fix above.
"""
messages = [
{"role": "user", "content": [{"type": "text", "text": "What is the capital of France?"}]},
{"role": "assistant", "content": [{"type": "text", "text": "Paris."}]},
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items == [
{"role": "user", "content": "What is the capital of France?"},
{"role": "assistant", "content": "Paris."},
]
def test_messages_to_responses_input_assistant_image_block_is_dropped():
"""An image part is illegal inside an assistant turn, so it must not reach the provider."""
messages = [
{
"role": "assistant",
"content": [
{"type": "text", "text": "Here it is"},
{"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}},
],
}
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items == [{"role": "assistant", "content": "Here it is"}]
def test_messages_to_responses_input_keeps_user_image_alongside_an_assistant_turn():
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "What is in this image?"},
{"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}},
],
},
{"role": "assistant", "content": [{"type": "text", "text": "A cat."}]},
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items == [
{
"role": "user",
"content": (
{"type": "input_text", "text": "What is in this image?"},
{"type": "input_image", "image_url": "https://example.com/cat.png", "detail": "auto"},
),
},
{"role": "assistant", "content": "A cat."},
]
def test_messages_to_responses_input_preserves_inline_files():
"""An inline file must survive the round trip, or the count silently drops the file.
A small PDF is worth 36 tokens to OpenAI's counting API; dropping it left the same
request counting 13, the text-only total.
"""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "Summarize this file."},
{
"type": "file",
"file": {"filename": "report.pdf", "file_data": "data:application/pdf;base64,JVBERi0="},
},
],
}
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items == [
{
"role": "user",
"content": (
{"type": "input_text", "text": "Summarize this file."},
{
"type": "input_file",
"filename": "report.pdf",
"file_data": "data:application/pdf;base64,JVBERi0=",
},
),
}
]
def test_messages_to_responses_input_drops_a_file_with_no_inline_data():
"""OpenAI rejects `file_data` without a `filename`, and a rejected request loses the whole count."""
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "Summarize this file."},
{"type": "file", "file": {"file_data": "data:application/pdf;base64,JVBERi0="}},
{"type": "file", "file": {"file_id": "file-abc123"}},
],
}
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items == [{"role": "user", "content": "Summarize this file."}]
def test_messages_to_responses_input_assistant_file_block_is_dropped():
"""A file part is illegal inside an assistant turn, so it must not reach the provider."""
messages = [
{
"role": "assistant",
"content": [
{"type": "text", "text": "Here it is"},
{
"type": "file",
"file": {"filename": "report.pdf", "file_data": "data:application/pdf;base64,JVBERi0="},
},
],
}
]
input_items, _ = OpenAICountTokensConfig.messages_to_responses_input(messages)
assert input_items == [{"role": "assistant", "content": "Here it is"}]
def test_validate_request_valid():
"""Test that valid requests pass validation."""
config = OpenAICountTokensConfig()

View file

@ -0,0 +1,238 @@
import json
import sys
from pathlib import Path
from typing import Final
import httpx
import pytest
import respx
from openai import AsyncOpenAI, OpenAI
import litellm
from litellm.llms.litellm_proxy.responses.transformation import LiteLLMProxyResponsesAPIConfig
from litellm.llms.openai.common_utils import BaseOpenAILLM, OpenAIError
from litellm.llms.openai.openai import OpenAIChatCompletion
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.llms.openai.workload_identity import (
OpenAIWorkloadIdentityConfig,
_workload_identity_auth,
get_workload_identity_bearer_token,
resolve_openai_workload_identity_config,
)
from litellm.types.router import GenericLiteLLMParams
TOKEN_EXCHANGE_URL: Final = "https://auth.openai.com/oauth/token"
@pytest.fixture
def wif_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> OpenAIWorkloadIdentityConfig:
token_file: Final = tmp_path / "subject_token.jwt"
token_file.write_text("subject-token-from-file")
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
monkeypatch.delenv("OPENAI_API_BASE", raising=False)
monkeypatch.setattr(litellm, "api_base", None)
monkeypatch.setenv("OPENAI_IDENTITY_PROVIDER_ID", "idp_test123")
monkeypatch.setenv("OPENAI_SERVICE_ACCOUNT_ID", "user-test456")
monkeypatch.setenv("OPENAI_IDENTITY_TOKEN_FILE", str(token_file))
_workload_identity_auth.cache_clear()
litellm.in_memory_llm_clients_cache.flush_cache()
return OpenAIWorkloadIdentityConfig(
identity_provider_id="idp_test123",
service_account_id="user-test456",
token_file=str(token_file),
)
def mock_token_exchange(access_token: str = "exchanged-bearer-token") -> respx.Route:
return respx.post(TOKEN_EXCHANGE_URL).mock(
return_value=httpx.Response(200, json={"access_token": access_token, "expires_in": 3600})
)
class TestResolveConfig:
def test_resolves_from_env(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
assert resolve_openai_workload_identity_config(api_key=None, api_base=None) == wif_env
def test_static_api_key_wins(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
assert resolve_openai_workload_identity_config(api_key="sk-static", api_base=None) is None
def test_env_openai_api_key_wins(
self, wif_env: OpenAIWorkloadIdentityConfig, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OPENAI_API_KEY", "sk-from-env")
assert resolve_openai_workload_identity_config(api_key=None, api_base=None) is None
@pytest.mark.parametrize("empty_key", ["", " "])
def test_empty_api_key_arg_does_not_disable_wif(
self, wif_env: OpenAIWorkloadIdentityConfig, empty_key: str
) -> None:
assert resolve_openai_workload_identity_config(api_key=empty_key, api_base=None) == wif_env
@pytest.mark.parametrize("empty_key", ["", " "])
def test_empty_env_openai_api_key_does_not_disable_wif(
self, wif_env: OpenAIWorkloadIdentityConfig, monkeypatch: pytest.MonkeyPatch, empty_key: str
) -> None:
monkeypatch.setenv("OPENAI_API_KEY", empty_key)
assert resolve_openai_workload_identity_config(api_key=None, api_base=None) == wif_env
def test_foreign_api_base_disables(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
assert resolve_openai_workload_identity_config(api_key=None, api_base="https://my-vllm.internal/v1") is None
def test_openai_api_base_allows(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
assert resolve_openai_workload_identity_config(api_key=None, api_base="https://api.openai.com/v1") == wif_env
def test_plaintext_http_api_base_disables(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
assert resolve_openai_workload_identity_config(api_key=None, api_base="http://api.openai.com/v1") is None
def test_foreign_env_base_url_disables(
self, wif_env: OpenAIWorkloadIdentityConfig, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OPENAI_BASE_URL", "https://my-vllm.internal/v1")
assert resolve_openai_workload_identity_config(api_key=None, api_base=None) is None
def test_openai_env_base_url_allows(
self, wif_env: OpenAIWorkloadIdentityConfig, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
assert resolve_openai_workload_identity_config(api_key=None, api_base=None) == wif_env
def test_foreign_litellm_api_base_disables(
self, wif_env: OpenAIWorkloadIdentityConfig, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "api_base", "https://my-vllm.internal/v1")
assert resolve_openai_workload_identity_config(api_key=None, api_base=None) is None
@pytest.mark.parametrize(
"missing_var",
["OPENAI_IDENTITY_PROVIDER_ID", "OPENAI_SERVICE_ACCOUNT_ID", "OPENAI_IDENTITY_TOKEN_FILE"],
)
def test_partial_env_disables(
self, wif_env: OpenAIWorkloadIdentityConfig, monkeypatch: pytest.MonkeyPatch, missing_var: str
) -> None:
monkeypatch.delenv(missing_var)
assert resolve_openai_workload_identity_config(api_key=None, api_base=None) is None
class TestTokenExchange:
@respx.mock
def test_exchanges_subject_token_for_bearer(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
route: Final = mock_token_exchange()
assert get_workload_identity_bearer_token(wif_env) == "exchanged-bearer-token"
request_body: Final = json.loads(route.calls.last.request.content)
assert request_body["grant_type"] == "urn:ietf:params:oauth:grant-type:token-exchange"
assert request_body["subject_token"] == "subject-token-from-file"
assert request_body["subject_token_type"] == "urn:ietf:params:oauth:token-type:jwt"
assert request_body["identity_provider_id"] == "idp_test123"
assert request_body["service_account_id"] == "user-test456"
@respx.mock
def test_token_cached_across_mints(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
route: Final = mock_token_exchange()
first: Final = get_workload_identity_bearer_token(wif_env)
second: Final = get_workload_identity_bearer_token(wif_env)
assert first == second == "exchanged-bearer-token"
assert route.call_count == 1
def test_old_sdk_raises_upgrade_error(
self, wif_env: OpenAIWorkloadIdentityConfig, monkeypatch: pytest.MonkeyPatch
) -> None:
import openai as openai_module
monkeypatch.delattr(openai_module, "auth", raising=False)
monkeypatch.setitem(sys.modules, "openai.auth", None)
with pytest.raises(OpenAIError, match=r"openai>=2\.32\.0"):
wif_env.to_sdk_workload_identity()
class TestClientConstruction:
def test_sync_client_uses_workload_identity(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
client: Final = OpenAIChatCompletion()._get_openai_client(is_async=False, api_key=None, api_base=None)
assert isinstance(client, OpenAI)
assert client.api_key == "workload-identity-auth"
assert client._workload_identity_auth is not None
def test_async_client_uses_workload_identity(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
client: Final = OpenAIChatCompletion()._get_openai_client(is_async=True, api_key=None, api_base=None)
assert isinstance(client, AsyncOpenAI)
assert client.api_key == "workload-identity-auth"
assert client._workload_identity_auth is not None
def test_static_key_client_unaffected(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
client: Final = OpenAIChatCompletion()._get_openai_client(is_async=False, api_key="sk-static", api_base=None)
assert isinstance(client, OpenAI)
assert client.api_key == "sk-static"
assert client._workload_identity_auth is None
def test_cache_key_separates_wif_identities(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
other_config: Final = OpenAIWorkloadIdentityConfig(
identity_provider_id="idp_other",
service_account_id="user-other",
token_file=wif_env.token_file,
)
keys: Final = tuple(
BaseOpenAILLM.get_openai_client_cache_key(
client_initialization_params={"api_key": None, "is_async": False, "workload_identity_config": config},
client_type="openai",
)
for config in (wif_env, other_config, None)
)
assert len(set(keys)) == 3
@respx.mock
def test_request_carries_exchanged_bearer(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
mock_token_exchange()
completion_route: Final = respx.post("https://api.openai.com/v1/chat/completions").mock(
return_value=httpx.Response(
200,
json={
"id": "chatcmpl-wif",
"object": "chat.completion",
"created": 1,
"model": "gpt-4o-mini",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "ok"},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
},
)
)
client = OpenAIChatCompletion()._get_openai_client(is_async=False, api_key=None, api_base=None)
assert isinstance(client, OpenAI)
client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": "hi"}])
auth_header: Final = completion_route.calls.last.request.headers["Authorization"]
assert auth_header == "Bearer exchanged-bearer-token"
class TestResponsesValidateEnvironment:
@respx.mock
def test_mints_bearer_when_wif_configured(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
mock_token_exchange()
headers: Final = OpenAIResponsesAPIConfig().validate_environment(
headers={}, model="gpt-4o-mini", litellm_params=GenericLiteLLMParams()
)
assert headers["Authorization"] == "Bearer exchanged-bearer-token"
def test_static_key_wins(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
headers: Final = OpenAIResponsesAPIConfig().validate_environment(
headers={}, model="gpt-4o-mini", litellm_params=GenericLiteLLMParams(api_key="sk-responses")
)
assert headers["Authorization"] == "Bearer sk-responses"
def test_foreign_api_base_skips_wif(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
headers: Final = OpenAIResponsesAPIConfig().validate_environment(
headers={},
model="gpt-4o-mini",
litellm_params=GenericLiteLLMParams(api_base="https://my-vllm.internal/v1"),
)
assert headers["Authorization"] == "Bearer None"
def test_litellm_proxy_subclass_never_mints_wif(self, wif_env: OpenAIWorkloadIdentityConfig) -> None:
headers: Final = LiteLLMProxyResponsesAPIConfig().validate_environment(
headers={}, model="gpt-4o-mini", litellm_params=GenericLiteLLMParams()
)
assert headers["Authorization"] == "Bearer None"

View file

@ -35,6 +35,22 @@ def mock_mcp_client_ip():
yield
@pytest.fixture(autouse=True)
def isolate_global_mcp_registry():
"""Restore the module-global MCP server registry after each test.
Tests here register servers on ``global_mcp_server_manager`` directly; without a
restore, entries leak into other test modules sharing the same worker and break
assertions over the full registry contents.
"""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import global_mcp_server_manager
snapshot = dict(global_mcp_server_manager.registry)
yield
global_mcp_server_manager.registry.clear()
global_mcp_server_manager.registry.update(snapshot)
def _mock_callback_request(base_url: str = "http://localhost:3000/"):
"""Return a MagicMock Request for callback/authorize same-origin tests.

View file

@ -83,6 +83,50 @@ def test_update_customer_success(mock_prisma_client, mock_user_api_key_auth):
assert response.json()["alias"] == "Updated Test User"
def test_update_customer_unblock(mock_prisma_client, mock_user_api_key_auth):
mock_end_user = LiteLLM_EndUserTable(user_id="test-user-1", blocked=True)
updated_mock_end_user = LiteLLM_EndUserTable(user_id="test-user-1", blocked=False)
mock_prisma_client.db.litellm_endusertable.find_first = AsyncMock(return_value=mock_end_user)
mock_prisma_client.db.litellm_endusertable.update = AsyncMock(return_value=updated_mock_end_user)
response = client.post(
"/customer/update",
json={"user_id": "test-user-1", "blocked": False},
headers={"Authorization": "Bearer test-key"},
)
assert response.status_code == 200
assert response.json()["blocked"] is False
update_mock = mock_prisma_client.db.litellm_endusertable.update
update_mock.assert_called_once()
assert update_mock.call_args.kwargs["data"]["blocked"] is False
def test_update_customer_keeps_blocked_when_omitted(mock_prisma_client, mock_user_api_key_auth):
"""
Regression test: updating a blocked customer without supplying `blocked`
must NOT reset it to unblocked. `blocked=False` is the model default and
should only be applied when explicitly provided by the caller.
"""
mock_end_user = LiteLLM_EndUserTable(user_id="test-user-1", blocked=True)
updated_mock_end_user = LiteLLM_EndUserTable(user_id="test-user-1", blocked=True)
mock_prisma_client.db.litellm_endusertable.find_first = AsyncMock(return_value=mock_end_user)
mock_prisma_client.db.litellm_endusertable.update = AsyncMock(return_value=updated_mock_end_user)
response = client.post(
"/customer/update",
json={"user_id": "test-user-1", "alias": "Updated Test User"},
headers={"Authorization": "Bearer test-key"},
)
assert response.status_code == 200
update_mock = mock_prisma_client.db.litellm_endusertable.update
update_mock.assert_called_once()
assert "blocked" not in update_mock.call_args.kwargs["data"]
def test_update_customer_not_found(mock_prisma_client, mock_user_api_key_auth):
"""
Test that update_end_user raises a 404 ProxyException when user_id does not exist.

View file

@ -3,10 +3,12 @@ Test for response_api_endpoints/endpoints.py
"""
import unittest
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from fastapi.testclient import TestClient
from httpx import Response
import litellm
from litellm.proxy.proxy_server import app
@ -82,11 +84,7 @@ class TestResponsesAPIEndpoints(unittest.TestCase):
ResponseOutputMessage(
type="message",
role="assistant",
content=[
ResponseOutputText(
type="output_text", text="Hello from Cursor!"
)
],
content=[ResponseOutputText(type="output_text", text="Hello from Cursor!")],
)
],
)
@ -121,9 +119,7 @@ class TestResponsesAPIEndpoints(unittest.TestCase):
@pytest.mark.asyncio
@patch("litellm.proxy.proxy_server.llm_router")
@patch("litellm.proxy.proxy_server.user_api_key_auth")
async def test_responses_api_key_spend_header_includes_response_cost(
self, mock_auth, mock_router
):
async def test_responses_api_key_spend_header_includes_response_cost(self, mock_auth, mock_router):
"""
Test that x-litellm-key-spend header includes the current request's response_cost
for /v1/responses endpoint.
@ -159,9 +155,7 @@ class TestResponsesAPIEndpoints(unittest.TestCase):
ResponseOutputMessage(
type="message",
role="assistant",
content=[
ResponseOutputText(type="output_text", text="Test response")
],
content=[ResponseOutputText(type="output_text", text="Test response")],
)
],
)
@ -356,6 +350,7 @@ class TestWSModelExtraction:
from litellm.proxy.response_api_endpoints.endpoints import (
_extract_model_from_first_ws_event,
)
event = {"type": "response.create", "model": "gpt-4o", "input": "hello"}
assert _extract_model_from_first_ws_event(event) == "gpt-4o"
@ -363,6 +358,7 @@ class TestWSModelExtraction:
from litellm.proxy.response_api_endpoints.endpoints import (
_extract_model_from_first_ws_event,
)
event = {"type": "response.create", "response": {"model": "gpt-4o", "input": "hello"}}
assert _extract_model_from_first_ws_event(event) == "gpt-4o"
@ -370,6 +366,7 @@ class TestWSModelExtraction:
from litellm.proxy.response_api_endpoints.endpoints import (
_extract_model_from_first_ws_event,
)
event = {
"type": "response.create",
"model": "flat-model",
@ -381,6 +378,7 @@ class TestWSModelExtraction:
from litellm.proxy.response_api_endpoints.endpoints import (
_extract_model_from_first_ws_event,
)
event = {"type": "response.create", "input": "hello"}
assert _extract_model_from_first_ws_event(event) is None
@ -400,9 +398,7 @@ class TestResponsesWSFirstFrameValidation:
)
ws = MagicMock()
ws.receive_text = AsyncMock(
return_value=json.dumps({"type": "session.update", "model": "gpt-4o"})
)
ws.receive_text = AsyncMock(return_value=json.dumps({"type": "session.update", "model": "gpt-4o"}))
ws.send_text = AsyncMock()
ws.close = AsyncMock()
@ -412,10 +408,7 @@ class TestResponsesWSFirstFrameValidation:
ws.send_text.assert_awaited_once()
ws.close.assert_awaited_once_with(code=1008, reason="Invalid first message")
error_payload = json.loads(ws.send_text.await_args.args[0])
assert (
error_payload["error"]["message"]
== "First message must be a response.create JSON object."
)
assert error_payload["error"]["message"] == "First message must be a response.create JSON object."
@pytest.mark.asyncio
async def test_rejects_non_object_json_first_frame(self):
@ -484,16 +477,12 @@ class TestResponsesWSFirstFrameModelAuth:
ws.url = "ws://testserver/v1/responses"
ws.accept = AsyncMock()
ws.receive_text = AsyncMock(
return_value=json.dumps(
{"type": "response.create", "model": "gpt-4o-mini", "input": []}
)
return_value=json.dumps({"type": "response.create", "model": "gpt-4o-mini", "input": []})
)
ws.close = AsyncMock()
processor = MagicMock()
processor.common_processing_pre_call_logic = AsyncMock(
return_value=({"model": "gpt-4o-mini"}, MagicMock())
)
processor.common_processing_pre_call_logic = AsyncMock(return_value=({"model": "gpt-4o-mini"}, MagicMock()))
async def fake_llm_call():
return None
@ -529,9 +518,7 @@ class TestResponsesWSFirstFrameModelAuth:
_enforce_responses_ws_first_frame_model_auth,
)
request = Request(
{"type": "http", "method": "POST", "path": "/v1/responses", "headers": []}
)
request = Request({"type": "http", "method": "POST", "path": "/v1/responses", "headers": []})
user_api_key_dict = MagicMock()
llm_router = MagicMock()
@ -593,9 +580,7 @@ class TestReadWSModelFromFirstFrameErrors:
assert result is None
ws.send_text.assert_not_awaited()
ws.close.assert_awaited_once_with(
code=1008, reason="Timed out waiting for first message"
)
ws.close.assert_awaited_once_with(code=1008, reason="Timed out waiting for first message")
@pytest.mark.asyncio
async def test_invalid_json_sends_error_and_closes(self):
@ -613,9 +598,7 @@ class TestReadWSModelFromFirstFrameErrors:
assert result is None
payload = json.loads(ws.send_text.await_args.args[0])
assert payload["error"]["message"] == "First message is not valid JSON."
ws.close.assert_awaited_once_with(
code=1008, reason="Invalid JSON in first message"
)
ws.close.assert_awaited_once_with(code=1008, reason="Invalid JSON in first message")
@pytest.mark.asyncio
async def test_missing_model_sends_error_and_closes(self):
@ -624,9 +607,7 @@ class TestReadWSModelFromFirstFrameErrors:
)
ws = MagicMock()
ws.receive_text = AsyncMock(
return_value=json.dumps({"type": "response.create", "input": []})
)
ws.receive_text = AsyncMock(return_value=json.dumps({"type": "response.create", "input": []}))
ws.send_text = AsyncMock()
ws.close = AsyncMock()
@ -679,10 +660,7 @@ class TestManagedResponsesSameProvider:
assert self._handler("gpt-4o")._same_provider("gpt-4o-mini") is True
def test_different_provider_is_not_same(self):
assert (
self._handler("gpt-4o")._same_provider("vertex_ai/gemini-2.0-flash")
is False
)
assert self._handler("gpt-4o")._same_provider("vertex_ai/gemini-2.0-flash") is False
def test_inject_credentials_keeps_provider_for_same_provider_model(self):
handler = self._handler("gpt-4o", custom_llm_provider="openai")
@ -697,18 +675,14 @@ class TestManagedResponsesSameProvider:
assert "custom_llm_provider" not in call_kwargs
def test_unresolvable_connection_model_falls_back_to_custom_provider(self):
handler = self._handler(
"my-custom-deployment", custom_llm_provider="openai"
)
handler = self._handler("my-custom-deployment", custom_llm_provider="openai")
assert handler._same_provider("gpt-4o-mini") is True
call_kwargs: dict = {}
handler._inject_credentials(call_kwargs, model="gpt-4o-mini")
assert call_kwargs["custom_llm_provider"] == "openai"
def test_unresolvable_connection_model_still_drops_cross_provider(self):
handler = self._handler(
"my-custom-deployment", custom_llm_provider="openai"
)
handler = self._handler("my-custom-deployment", custom_llm_provider="openai")
call_kwargs: dict = {}
handler._inject_credentials(call_kwargs, model="vertex_ai/gemini-2.0-flash")
assert "custom_llm_provider" not in call_kwargs
@ -840,9 +814,7 @@ def test_cursor_chat_completions_input_body_uses_responses_pipeline_and_strips_s
type="message",
role="assistant",
status="completed",
content=[
ResponseOutputText(type="output_text", text="agent reply", annotations=[])
],
content=[ResponseOutputText(type="output_text", text="agent reply", annotations=[])],
)
],
)
@ -851,9 +823,12 @@ def test_cursor_chat_completions_input_body_uses_responses_pipeline_and_strips_s
app.dependency_overrides[user_api_key_auth] = _auth_override
try:
with patch.object(ps, "llm_router", mock_router), patch(
"litellm.proxy.response_api_endpoints.endpoints._read_request_body",
side_effect=capturing_read_request_body,
with (
patch.object(ps, "llm_router", mock_router),
patch(
"litellm.proxy.response_api_endpoints.endpoints._read_request_body",
side_effect=capturing_read_request_body,
),
):
client = TestClient(app)
response = client.post(
@ -1488,8 +1463,8 @@ def _router_serving_only(base_model: str) -> MagicMock:
mock_router.router_general_settings.pass_through_all_models = False
mock_router.default_deployment = None
mock_router.pattern_router.patterns = {base_model: ["anthropic/*"]}
mock_router.pattern_router.get_pattern.side_effect = (
lambda model: [{"model_name": "anthropic/*"}] if model == base_model else None
mock_router.pattern_router.get_pattern.side_effect = lambda model: (
[{"model_name": "anthropic/*"}] if model == base_model else None
)
return mock_router
@ -1739,9 +1714,7 @@ class TestCursorGateRecognizesRoutingGroups:
from litellm.proxy.response_api_endpoints.endpoints import _resolve_cursor_model_variant
router = Router(
model_list=[
{"model_name": "member-fast", "litellm_params": {"model": "openai/gpt-4o", "api_key": "fake"}}
],
model_list=[{"model_name": "member-fast", "litellm_params": {"model": "openai/gpt-4o", "api_key": "fake"}}],
routing_groups=[
{"group_name": "grouped-thinking-high", "models": ["member-fast"], "routing_strategy": "simple-shuffle"}
],
@ -1836,3 +1809,153 @@ class TestGuardrailBlockedResponsesUsage:
assert usage["input_tokens"] == 0
assert usage["output_tokens"] == 0
assert usage["total_tokens"] == 0
class TestResponsesInputTokens:
"""Regression tests for POST /v1/responses/input_tokens.
The docs promise OpenAI-format token counting on the proxy, but the route was
never registered, so the POST fell through to the GET/DELETE-only
/v1/responses/{response_id} route and returned 405."""
def _post_input_tokens(
self,
body: dict[str, Any],
path: str = "/v1/responses/input_tokens",
counter: AsyncMock | None = None,
) -> tuple[Response, AsyncMock]:
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.response_api_endpoints.endpoints import _proxy_token_counter
from litellm.types.utils import TokenCountResponse
token_counter_mock = (
counter
if counter is not None
else AsyncMock(
return_value=TokenCountResponse(
total_tokens=13,
request_model=body.get("model", ""),
model_used=body.get("model", ""),
tokenizer_type="openai_api",
)
)
)
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(api_key="sk-test", request_route=path)
app.dependency_overrides[_proxy_token_counter] = lambda: token_counter_mock
try:
client = TestClient(app)
response = client.post(path, json=body, headers={"Authorization": "Bearer sk-1234"})
return response, token_counter_mock
finally:
app.dependency_overrides.pop(user_api_key_auth, None)
app.dependency_overrides.pop(_proxy_token_counter, None)
def test_string_input_returns_openai_input_tokens_shape(self):
response, counter = self._post_input_tokens({"model": "gpt-4o", "input": "Hello, how are you?"})
assert response.status_code == 200, response.text
assert response.json() == {"object": "response.input_tokens", "input_tokens": 13}
counter.assert_awaited_once()
assert counter.call_args.kwargs["call_endpoint"] is True
token_request = counter.call_args.kwargs["request"]
assert token_request.model == "gpt-4o"
assert token_request.messages == [{"role": "user", "content": "Hello, how are you?"}]
def test_every_route_alias_is_registered(self):
for path in ("/v1/responses/input_tokens", "/responses/input_tokens", "/openai/v1/responses/input_tokens"):
response, _ = self._post_input_tokens({"model": "gpt-4o", "input": "hi"}, path=path)
assert response.status_code == 200, f"{path}: {response.status_code} {response.text}"
def test_input_items_instructions_and_tools_are_forwarded(self):
tools = [
{
"type": "function",
"name": "get_weather",
"description": "Get weather for a city",
"parameters": {"type": "object", "properties": {"city": {"type": "string"}}},
}
]
response, counter = self._post_input_tokens(
{
"model": "gpt-4o",
"input": [{"role": "user", "content": "What is the weather in Paris?"}],
"instructions": "You are terse.",
"tools": tools,
}
)
assert response.status_code == 200, response.text
token_request = counter.call_args.kwargs["request"]
assert token_request.messages == [
{"role": "system", "content": "You are terse."},
{"role": "user", "content": "What is the weather in Paris?"},
]
assert token_request.tools == tools
def test_missing_model_returns_openai_400(self):
response, counter = self._post_input_tokens({"input": "Hello"})
assert response.status_code == 400, response.text
assert response.json() == {
"error": {
"message": "Missing required parameter: 'model'.",
"type": "invalid_request_error",
"param": "model",
"code": "missing_required_parameter",
}
}
counter.assert_not_awaited()
def test_missing_input_returns_openai_400(self):
response, counter = self._post_input_tokens({"model": "gpt-4o"})
assert response.status_code == 400, response.text
assert response.json() == {
"error": {
"message": "Missing required parameter: 'input'.",
"type": "invalid_request_error",
"param": "input",
"code": "missing_required_parameter",
}
}
counter.assert_not_awaited()
@pytest.mark.parametrize("empty_input", ["", []])
def test_empty_input_returns_openai_400(self, empty_input):
response, counter = self._post_input_tokens({"model": "gpt-4o", "input": empty_input})
assert response.status_code == 400, response.text
assert response.json() == {
"error": {
"message": """One of "input" or "previous_response_id" or 'prompt' or 'conversation' must be provided.""",
"type": "invalid_request_error",
"param": None,
"code": "missing_required_parameter",
}
}
counter.assert_not_awaited()
def test_invalid_tools_returns_openai_400(self):
response, counter = self._post_input_tokens({"model": "gpt-4o", "input": "hi", "tools": "not-a-list"})
assert response.status_code == 400, response.text
error = response.json()["error"]
assert error["type"] == "invalid_request_error"
counter.assert_not_awaited()
def test_provider_error_maps_status_code(self):
from litellm.proxy._types import ProxyException
failing_counter = AsyncMock(
side_effect=ProxyException(
message="rate limited",
type="token_counting_error",
param="model",
code="429",
)
)
response, _ = self._post_input_tokens({"model": "gpt-4o", "input": "hi"}, counter=failing_counter)
assert response.status_code == 429, response.text
assert response.json()["error"]["message"] == "rate limited"

View file

@ -0,0 +1,48 @@
from typing import Final
import pytest
from litellm.caching import DualCache
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.spend_tracking.budget_reservation import reserve_budget_for_request
from litellm.proxy.utils import ProxyLogging
TOKEN_COUNTING_ROUTES: Final = (
"/responses/input_tokens",
"/v1/responses/input_tokens",
"/openai/v1/responses/input_tokens",
"/utils/token_counter",
)
def _budgeted_token() -> UserAPIKeyAuth:
return UserAPIKeyAuth(api_key="sk-test", token="hashed-token", max_budget=100.0, spend=0.0)
async def _reserve(route: str) -> dict | None:
return await reserve_budget_for_request(
request_body={"model": "gpt-4o", "input": "hello"},
route=route,
llm_router=None,
valid_token=_budgeted_token(),
team_object=None,
user_object=None,
prisma_client=None,
user_api_key_cache=UserApiKeyCache(),
proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache()),
)
@pytest.mark.asyncio
@pytest.mark.parametrize("route", TOKEN_COUNTING_ROUTES)
async def test_token_counting_routes_are_exempt_from_budget_reservation(route):
assert await _reserve(route) is None
@pytest.mark.asyncio
async def test_non_exempt_llm_route_still_reserves_budget():
reservation: Final = await _reserve("/v1/responses")
assert reservation is not None
assert reservation["reserved_cost"] > 0

View file

@ -124,6 +124,25 @@ class TestLiteLLMCompletionResponsesConfig:
assert "extra_field" not in result["file"]
assert "another_field" not in result["file"]
def test_transform_input_file_item_to_file_item_keeps_filename(self):
"""OpenAI rejects file_data with no filename beside it, so dropping it 400s the request"""
result = (
LiteLLMCompletionResponsesConfig._transform_input_file_item_to_file_item(
{
"type": "input_file",
"filename": "report.pdf",
"file_data": "data:application/pdf;base64,JVBERi0=",
}
)
)
assert result == {
"type": "file",
"file": {
"file_data": "data:application/pdf;base64,JVBERi0=",
"filename": "report.pdf",
},
}
def test_transform_input_file_item_to_file_item_with_file_url(self):
"""file_url should be mapped to file_id for downstream URL handling"""
result = (

View file

@ -0,0 +1,35 @@
import json
from pathlib import Path
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
def test_friendli_glm_5_3_flash_model_info():
model = "friendliai/zai-org/GLM-5.3-Flash"
json_path = Path(__file__).parents[2] / "model_prices_and_context_window.json"
with open(json_path) as f:
model_cost = json.load(f)
info = model_cost.get(model)
assert (
info is not None
), f"{model} not found in model_prices_and_context_window.json"
assert info["litellm_provider"] == "friendliai"
assert info["mode"] == "chat"
assert info["input_cost_per_token"] == 1.5e-07
assert info["output_cost_per_token"] == 5e-07
assert info["cache_read_input_token_cost"] == 3e-08
assert info["max_input_tokens"] == 1048576
assert info["max_output_tokens"] == 1048576
assert info["supports_function_calling"] is True
assert info["supports_reasoning"] is True
assert info["reasoning_effort_levels"] == ["low", "high", "max"]
assert info["supports_tool_choice"] is True
assert info["supports_prompt_caching"] is True
assert info["supports_vision"] is True
assert info["supports_image_input"] is True
assert info["supports_video_input"] is True
routed_model, provider, _, _ = get_llm_provider(model=model)
assert routed_model == "zai-org/GLM-5.3-Flash"
assert provider == "friendliai"

View file

@ -9,7 +9,7 @@
"limit": 269
},
"LIT004": {
"limit": 43
"limit": 40
},
"LIT005": {
"limit": 0

View file

@ -9736,6 +9736,37 @@ export interface paths {
patch?: never;
trace?: never;
};
"/openai/v1/responses/input_tokens": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Responses Input Tokens
* @description Count the input tokens of a Responses API request without calling the model.
*
* Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/input-tokens
*
* ```bash
* curl -X POST http://localhost:4000/v1/responses/input_tokens -H "Content-Type: application/json" -H "Authorization: Bearer sk-1234" -d '{
* "model": "gpt-4o",
* "input": "Hello, how are you?"
* }'
* ```
*
* Returns: `{"object": "response.input_tokens", "input_tokens": <count>}`
*/
post: operations["responses_input_tokens_openai_v1_responses_input_tokens_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/openai/v1/responses/{response_id}": {
parameters: {
query?: never;
@ -12655,6 +12686,37 @@ export interface paths {
patch?: never;
trace?: never;
};
"/responses/input_tokens": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Responses Input Tokens
* @description Count the input tokens of a Responses API request without calling the model.
*
* Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/input-tokens
*
* ```bash
* curl -X POST http://localhost:4000/v1/responses/input_tokens -H "Content-Type: application/json" -H "Authorization: Bearer sk-1234" -d '{
* "model": "gpt-4o",
* "input": "Hello, how are you?"
* }'
* ```
*
* Returns: `{"object": "response.input_tokens", "input_tokens": <count>}`
*/
post: operations["responses_input_tokens_responses_input_tokens_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/responses/{response_id}": {
parameters: {
query?: never;
@ -19230,6 +19292,37 @@ export interface paths {
patch?: never;
trace?: never;
};
"/v1/responses/input_tokens": {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
get?: never;
put?: never;
/**
* Responses Input Tokens
* @description Count the input tokens of a Responses API request without calling the model.
*
* Follows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/input-tokens
*
* ```bash
* curl -X POST http://localhost:4000/v1/responses/input_tokens -H "Content-Type: application/json" -H "Authorization: Bearer sk-1234" -d '{
* "model": "gpt-4o",
* "input": "Hello, how are you?"
* }'
* ```
*
* Returns: `{"object": "response.input_tokens", "input_tokens": <count>}`
*/
post: operations["responses_input_tokens_v1_responses_input_tokens_post"];
delete?: never;
options?: never;
head?: never;
patch?: never;
trace?: never;
};
"/v1/responses/{response_id}": {
parameters: {
query?: never;
@ -51687,6 +51780,26 @@ export interface operations {
};
};
};
responses_input_tokens_openai_v1_responses_input_tokens_post: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
};
};
get_response_openai_v1_responses__response_id__get: {
parameters: {
query?: never;
@ -54651,6 +54764,26 @@ export interface operations {
};
};
};
responses_input_tokens_responses_input_tokens_post: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
};
};
get_response_responses__response_id__get: {
parameters: {
query?: never;
@ -63073,6 +63206,26 @@ export interface operations {
};
};
};
responses_input_tokens_v1_responses_input_tokens_post: {
parameters: {
query?: never;
header?: never;
path?: never;
cookie?: never;
};
requestBody?: never;
responses: {
/** @description Successful Response */
200: {
headers: {
[name: string]: unknown;
};
content: {
"application/json": unknown;
};
};
};
};
get_response_v1_responses__response_id__get: {
parameters: {
query?: never;