diff --git a/litellm/__init__.py b/litellm/__init__.py index c83e72a78b4..1447e05fdf7 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -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 diff --git a/litellm/llms/bedrock/realtime/handler.py b/litellm/llms/bedrock/realtime/handler.py index 3bda8dd8359..42fe8941443 100644 --- a/litellm/llms/bedrock/realtime/handler.py +++ b/litellm/llms/bedrock/realtime/handler.py @@ -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) diff --git a/litellm/llms/bedrock/realtime/transformation.py b/litellm/llms/bedrock/realtime/transformation.py index 951bf636b2f..28c2e446d10 100644 --- a/litellm/llms/bedrock/realtime/transformation.py +++ b/litellm/llms/bedrock/realtime/transformation.py @@ -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 diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index ff453ba414e..9c13411aaad 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -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, diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index a6115640d78..fcade835cce 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -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] diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index ff453ba414e..9c13411aaad 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -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, diff --git a/osv-scanner.toml b/osv-scanner.toml index 7ab450945f5..5b0339bdcd0 100644 --- a/osv-scanner.toml +++ b/osv-scanner.toml @@ -2,3 +2,8 @@ id = "GHSA-w8v5-vhqr-4h9v" ignoreUntil = 2026-09-09 reason = "diskcache has no fixed release published; remove this entry once one exists" + +[[IgnoredVulns]] +id = "GHSA-h7x2-h6g9-p789" +ignoreUntil = 2026-09-14 +reason = "mlflow has no fixed release published; remove this entry once one exists" diff --git a/tests/logging_callback_tests/gcs_pub_sub_body/spend_logs_payload.json b/tests/logging_callback_tests/gcs_pub_sub_body/spend_logs_payload.json index 1838fb16e91..28912a27501 100644 --- a/tests/logging_callback_tests/gcs_pub_sub_body/spend_logs_payload.json +++ b/tests/logging_callback_tests/gcs_pub_sub_body/spend_logs_payload.json @@ -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, diff --git a/tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_handler.py b/tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_handler.py index 9efcee192b1..0ea5b7ad4a1 100644 --- a/tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_handler.py +++ b/tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_handler.py @@ -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", {}) diff --git a/tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_transformation.py b/tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_transformation.py index ae6b1febd6b..a74f03449a1 100644 --- a/tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_transformation.py +++ b/tests/test_litellm/llms/bedrock/realtime/test_bedrock_realtime_transformation.py @@ -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"]) diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py index 3279c59acd4..598e9276423 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py @@ -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.