diff --git a/litellm/interactions/litellm_responses_transformation/streaming_iterator.py b/litellm/interactions/litellm_responses_transformation/streaming_iterator.py index 72a3afbc3c5..567e9b523e8 100644 --- a/litellm/interactions/litellm_responses_transformation/streaming_iterator.py +++ b/litellm/interactions/litellm_responses_transformation/streaming_iterator.py @@ -2,7 +2,7 @@ Streaming iterator for transforming Responses API stream to Interactions API stream. """ -from typing import Any, AsyncIterator, Dict, Iterator, Optional, cast +from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, cast from litellm.responses.streaming_iterator import ( BaseResponsesAPIStreamingIterator, @@ -15,6 +15,7 @@ from litellm.types.interactions import ( InteractionsAPIStreamingResponse, ) from litellm.types.llms.openai import ( + ContentPartAddedEvent, OutputTextDeltaEvent, ResponseCompletedEvent, ResponseCreatedEvent, @@ -51,6 +52,7 @@ class LiteLLMResponsesInteractionsStreamingIterator: self.collected_text = "" self.sent_interaction_start = False self.sent_content_start = False + self._pending_events: List[InteractionsAPIStreamingResponse] = [] def _transform_responses_chunk_to_interactions_chunk( self, @@ -80,7 +82,49 @@ class LiteLLMResponsesInteractionsStreamingIterator: ) self.collected_text += delta_text - # Send interaction.start if not sent + # Fallback: emit interaction.start, and queue content.start carrying this + # delta so the first token is preserved in the stream. + if not self.sent_interaction_start: + self.sent_interaction_start = True + self.sent_content_start = True + self._pending_events.append( + InteractionsAPIStreamingResponse( + event_type="content.start", + id=getattr(responses_chunk, "item_id", None), + object="content", + delta={"type": "text", "text": delta_text}, + ) + ) + return InteractionsAPIStreamingResponse( + event_type="interaction.start", + id=getattr(responses_chunk, "item_id", None) + or f"interaction_{id(self)}", + object="interaction", + status="in_progress", + model=self.model, + ) + + # Fallback: emit content.start if ContentPartAddedEvent never arrived + if not self.sent_content_start: + self.sent_content_start = True + return InteractionsAPIStreamingResponse( + event_type="content.start", + id=getattr(responses_chunk, "item_id", None), + object="content", + delta={"type": "text", "text": delta_text}, + ) + + # Normal path: emit content.delta with type field + return InteractionsAPIStreamingResponse( + event_type="content.delta", + id=getattr(responses_chunk, "item_id", None), + object="content", + delta={"type": "text", "text": delta_text}, + ) + + # Handle ContentPartAddedEvent -> content.start (arrives before text deltas) + if isinstance(responses_chunk, ContentPartAddedEvent): + # Fallback: emit interaction.start if ResponseCreatedEvent never arrived if not self.sent_interaction_start: self.sent_interaction_start = True return InteractionsAPIStreamingResponse( @@ -91,8 +135,6 @@ class LiteLLMResponsesInteractionsStreamingIterator: status="in_progress", model=self.model, ) - - # Send content.start if not sent if not self.sent_content_start: self.sent_content_start = True return InteractionsAPIStreamingResponse( @@ -101,14 +143,7 @@ class LiteLLMResponsesInteractionsStreamingIterator: object="content", delta={"type": "text", "text": ""}, ) - - # Send content.delta - return InteractionsAPIStreamingResponse( - event_type="content.delta", - id=getattr(responses_chunk, "item_id", None), - object="content", - delta={"text": delta_text}, - ) + return None # Handle ResponseCreatedEvent or ResponseInProgressEvent -> interaction.start if isinstance(responses_chunk, (ResponseCreatedEvent, ResponseInProgressEvent)): @@ -172,6 +207,10 @@ class LiteLLMResponsesInteractionsStreamingIterator: delattr(self, "_pending_interaction_complete") return pending + # Drain events queued from a prior chunk (e.g. content.start emitted alongside + # the interaction.start fallback for the first OutputTextDeltaEvent). + if self._pending_events: + return self._pending_events.pop(0) # Use a loop instead of recursion to avoid stack overflow sync_iterator = cast( SyncResponsesAPIStreamingIterator, self.responses_stream_iterator @@ -237,6 +276,10 @@ class LiteLLMResponsesInteractionsStreamingIterator: delattr(self, "_pending_interaction_complete") return pending + # Drain events queued from a prior chunk (e.g. content.start emitted alongside + # the interaction.start fallback for the first OutputTextDeltaEvent). + if self._pending_events: + return self._pending_events.pop(0) # Use a loop instead of recursion to avoid stack overflow async_iterator = cast( ResponsesAPIStreamingIterator, self.responses_stream_iterator diff --git a/tests/test_litellm/interactions/test_gemini_interactions_transformation.py b/tests/test_litellm/interactions/test_gemini_interactions_transformation.py index 758ff3ea38e..0ef97e25e44 100644 --- a/tests/test_litellm/interactions/test_gemini_interactions_transformation.py +++ b/tests/test_litellm/interactions/test_gemini_interactions_transformation.py @@ -9,15 +9,24 @@ Covers credential leak prevention changes: import os import sys -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest sys.path.insert(0, os.path.abspath("../../..")) +from litellm.interactions.litellm_responses_transformation.streaming_iterator import ( + LiteLLMResponsesInteractionsStreamingIterator, +) from litellm.llms.gemini.interactions.transformation import ( GoogleAIStudioInteractionsConfig, ) +from litellm.types.llms.openai import ( + ContentPartAddedEvent, + OutputTextDeltaEvent, + ResponseCompletedEvent, + ResponseCreatedEvent, +) from litellm.types.router import GenericLiteLLMParams _PATCH_GET_API_KEY = "litellm.llms.gemini.common_utils.GeminiModelInfo.get_api_key" @@ -113,6 +122,186 @@ class TestGetCompleteUrl: ) +class TestStreamingIterator: + def _make_iterator(self) -> LiteLLMResponsesInteractionsStreamingIterator: + return LiteLLMResponsesInteractionsStreamingIterator( + model="gpt-5.4", + litellm_custom_stream_wrapper=MagicMock(), + request_input="hi", + optional_params={}, + ) + + def _make_text_delta( + self, text: str, item_id: str = "item_1" + ) -> OutputTextDeltaEvent: + event = MagicMock(spec=OutputTextDeltaEvent) + event.delta = text + event.item_id = item_id + return event + + def _make_part_added(self, item_id: str = "item_1") -> ContentPartAddedEvent: + event = MagicMock(spec=ContentPartAddedEvent) + event.item_id = item_id + return event + + def _make_response_created(self) -> ResponseCreatedEvent: + event = MagicMock(spec=ResponseCreatedEvent) + event.response = MagicMock(id="resp_123") + return event + + def test_content_delta_includes_type_field(self): + """content.delta events must carry delta.type='text' so the UI can display them.""" + it = self._make_iterator() + it.sent_interaction_start = True + it.sent_content_start = True + + chunk = it._transform_responses_chunk_to_interactions_chunk( + self._make_text_delta("Hello") + ) + + assert chunk is not None + assert chunk.event_type == "content.delta" + assert chunk.delta == {"type": "text", "text": "Hello"} + + def test_response_part_added_emits_content_start(self): + """ContentPartAddedEvent (arrives before text deltas) should emit content.start + so the first OutputTextDeltaEvent immediately emits content.delta without dropping text. + """ + it = self._make_iterator() + it.sent_interaction_start = True + + chunk = it._transform_responses_chunk_to_interactions_chunk( + self._make_part_added() + ) + + assert chunk is not None + assert chunk.event_type == "content.start" + assert it.sent_content_start is True + + def test_first_text_delta_not_dropped_when_part_added_seen(self): + """After ContentPartAddedEvent, the first text delta must yield content.delta + (not content.start), preserving the token text.""" + it = self._make_iterator() + it.sent_interaction_start = True + it._transform_responses_chunk_to_interactions_chunk(self._make_part_added()) + + chunk = it._transform_responses_chunk_to_interactions_chunk( + self._make_text_delta("Hello") + ) + + assert chunk is not None + assert chunk.event_type == "content.delta" + assert chunk.delta is not None + assert chunk.delta.get("text") == "Hello" + + def test_part_added_emits_interaction_start_fallback_when_not_sent(self): + """If ContentPartAddedEvent arrives before any ResponseCreatedEvent, + the iterator must emit interaction.start before content.start to honor + the documented event ordering contract.""" + it = self._make_iterator() + + chunk = it._transform_responses_chunk_to_interactions_chunk( + self._make_part_added(item_id="item_42") + ) + + assert chunk is not None + assert chunk.event_type == "interaction.start" + assert chunk.id == "item_42" + assert chunk.status == "in_progress" + assert chunk.model == "gpt-5.4" + assert it.sent_interaction_start is True + assert it.sent_content_start is False + + def test_part_added_returns_none_when_already_started(self): + """A second ContentPartAddedEvent (after content.start was already emitted) + should be a no-op so we don't re-emit content.start.""" + it = self._make_iterator() + it.sent_interaction_start = True + it.sent_content_start = True + + chunk = it._transform_responses_chunk_to_interactions_chunk( + self._make_part_added() + ) + + assert chunk is None + + def test_part_added_without_item_id_falls_back_to_self_id(self): + """When ContentPartAddedEvent has no item_id and we emit the interaction.start + fallback, the id must default to an interaction_ string.""" + it = self._make_iterator() + event = MagicMock(spec=ContentPartAddedEvent) + event.item_id = None + + chunk = it._transform_responses_chunk_to_interactions_chunk(event) + + assert chunk is not None + assert chunk.event_type == "interaction.start" + assert chunk.id == f"interaction_{id(it)}" + + def test_first_text_delta_not_dropped_when_no_prior_start_events(self): + """When OutputTextDeltaEvent arrives before any ResponseCreatedEvent or + ContentPartAddedEvent, the iterator must emit interaction.start *and* + immediately follow with a content.start that carries this delta's text, + so the first token is never silently dropped from the stream.""" + events = [ + self._make_text_delta("Hello"), + self._make_text_delta(" World"), + ] + wrapper = MagicMock() + wrapper.__iter__ = lambda self: iter(events) + wrapper.__next__ = lambda self, _it=iter(events): next(_it) + it = LiteLLMResponsesInteractionsStreamingIterator( + model="gpt-5.4", + litellm_custom_stream_wrapper=wrapper, + request_input="hi", + optional_params={}, + ) + + first = it._transform_responses_chunk_to_interactions_chunk(events[0]) + assert first is not None + assert first.event_type == "interaction.start" + assert it.sent_interaction_start is True + assert it.sent_content_start is True + assert len(it._pending_events) == 1 + pending = it._pending_events[0] + assert pending.event_type == "content.start" + assert pending.delta == {"type": "text", "text": "Hello"} + + second = it._transform_responses_chunk_to_interactions_chunk(events[1]) + assert second is not None + assert second.event_type == "content.delta" + assert second.delta == {"type": "text", "text": " World"} + + +class TestTransformRequest: + def test_stream_param_included_in_request_body(self, config): + """When stream=True is in optional_params, the request body must include it + so the proxy forwards the SSE streaming flag to Google's backend.""" + body = config.transform_request( + model="gemini-2.5-flash", + agent=None, + input="Hello", + optional_params={"stream": True}, + litellm_params=GenericLiteLLMParams(api_key="test-key"), + headers={}, + ) + + assert body.get("stream") is True + assert body.get("input") == "Hello" + + def test_stream_false_not_included_when_absent(self, config): + body = config.transform_request( + model="gemini-2.5-flash", + agent=None, + input="Hello", + optional_params={}, + litellm_params=GenericLiteLLMParams(api_key="test-key"), + headers={}, + ) + + assert "stream" not in body + + class TestInteractionOperationUrls: """Test that get/delete/cancel interaction URLs exclude API key.""" diff --git a/ui/litellm-dashboard/src/components/playground/chat_ui/ChatUI.tsx b/ui/litellm-dashboard/src/components/playground/chat_ui/ChatUI.tsx index 06a09ca2c37..a80cf935c4b 100644 --- a/ui/litellm-dashboard/src/components/playground/chat_ui/ChatUI.tsx +++ b/ui/litellm-dashboard/src/components/playground/chat_ui/ChatUI.tsx @@ -49,6 +49,7 @@ import { fetchAvailableModels, ModelGroup } from "../llm_calls/fetch_models"; import { makeOpenAIImageEditsRequest } from "../llm_calls/image_edits"; import { makeOpenAIImageGenerationRequest } from "../llm_calls/image_generation"; import { makeOpenAIResponsesRequest } from "../llm_calls/responses_api"; +import { makeInteractionsRequest } from "../llm_calls/interactions_api"; import A2AMetrics from "./A2AMetrics"; import AdditionalModelSettings from "./AdditionalModelSettings"; import AudioRenderer from "./AudioRenderer"; @@ -649,6 +650,7 @@ const ChatUI: React.FC = ({ EndpointType.ANTHROPIC_MESSAGES, EndpointType.EMBEDDINGS, EndpointType.TRANSCRIPTION, + EndpointType.INTERACTIONS, ]; if (modelRequiredEndpoints.includes(endpointType as EndpointType) && !selectedModel) { @@ -914,6 +916,16 @@ const ChatUI: React.FC = ({ customProxyBaseUrl || undefined, ); } + } else if (endpointType === EndpointType.INTERACTIONS) { + await makeInteractionsRequest( + inputMessage, + (text, model) => updateTextUI("assistant", text, model), + selectedModel, + effectiveApiKey, + selectedTags, + signal, + customProxyBaseUrl || undefined, + ); } } @@ -1241,10 +1253,11 @@ const ChatUI: React.FC = ({ return true; } const optionEndpoint = getEndpointType(option.mode); - // Show chat models for responses/anthropic_messages endpoints as they are compatible + // Show chat models for responses/anthropic_messages/interactions endpoints as they are compatible if ( endpointType === EndpointType.RESPONSES || - endpointType === EndpointType.ANTHROPIC_MESSAGES + endpointType === EndpointType.ANTHROPIC_MESSAGES || + endpointType === EndpointType.INTERACTIONS ) { return optionEndpoint === endpointType || optionEndpoint === EndpointType.CHAT; } @@ -2089,7 +2102,8 @@ const ChatUI: React.FC = ({ endpointType === EndpointType.CHAT || endpointType === EndpointType.EMBEDDINGS || endpointType === EndpointType.RESPONSES || - endpointType === EndpointType.ANTHROPIC_MESSAGES + endpointType === EndpointType.ANTHROPIC_MESSAGES || + endpointType === EndpointType.INTERACTIONS ? "Type your message... (Shift+Enter for new line)" : endpointType === EndpointType.A2A_AGENTS ? "Send a message to the A2A agent..." diff --git a/ui/litellm-dashboard/src/components/playground/chat_ui/chatConstants.ts b/ui/litellm-dashboard/src/components/playground/chat_ui/chatConstants.ts index 919e3bc1c65..9592a521250 100644 --- a/ui/litellm-dashboard/src/components/playground/chat_ui/chatConstants.ts +++ b/ui/litellm-dashboard/src/components/playground/chat_ui/chatConstants.ts @@ -45,4 +45,5 @@ export const ENDPOINT_OPTIONS = [ { value: EndpointType.A2A_AGENTS, label: "/v1/a2a/message/send" }, { value: EndpointType.MCP, label: "/mcp-rest/tools/call" }, { value: EndpointType.REALTIME, label: "/v1/realtime" }, + { value: EndpointType.INTERACTIONS, label: "/v1beta/interactions" }, ]; diff --git a/ui/litellm-dashboard/src/components/playground/chat_ui/mode_endpoint_mapping.tsx b/ui/litellm-dashboard/src/components/playground/chat_ui/mode_endpoint_mapping.tsx index f354efe641e..3b2a5e0fd60 100644 --- a/ui/litellm-dashboard/src/components/playground/chat_ui/mode_endpoint_mapping.tsx +++ b/ui/litellm-dashboard/src/components/playground/chat_ui/mode_endpoint_mapping.tsx @@ -28,6 +28,7 @@ export enum EndpointType { A2A_AGENTS = "a2a_agents", MCP = "mcp", REALTIME = "realtime", + INTERACTIONS = "interactions", } // Create a mapping between the model mode and the corresponding endpoint type diff --git a/ui/litellm-dashboard/src/components/playground/llm_calls/interactions_api.tsx b/ui/litellm-dashboard/src/components/playground/llm_calls/interactions_api.tsx new file mode 100644 index 00000000000..6a6e4a2cfa0 --- /dev/null +++ b/ui/litellm-dashboard/src/components/playground/llm_calls/interactions_api.tsx @@ -0,0 +1,124 @@ +import NotificationManager from "@/components/molecules/notifications_manager"; +import { getGlobalLitellmHeaderName, getProxyBaseUrl } from "@/components/networking"; + +export async function makeInteractionsRequest( + input: string, + updateUI: (text: string, model?: string) => void, + selectedModel: string, + accessToken: string, + tags?: string[], + signal?: AbortSignal, + customBaseUrl?: string, + previousInteractionId?: string, +): Promise { + if (!accessToken) { + throw new Error("Virtual Key is required"); + } + + const isLocal = process.env.NODE_ENV === "development"; + if (isLocal !== true) { + console.log = function () {}; + } + + const proxyBaseUrl = customBaseUrl || getProxyBaseUrl(); + const normalizedBaseUrl = proxyBaseUrl.endsWith("/") ? proxyBaseUrl.slice(0, -1) : proxyBaseUrl; + const requestUrl = `${normalizedBaseUrl}/v1beta/interactions`; + + const headers: Record = { + "Content-Type": "application/json", + [getGlobalLitellmHeaderName()]: `Bearer ${accessToken}`, + }; + if (tags && tags.length > 0) { + headers["x-litellm-tags"] = tags.join(","); + } + + const body: Record = { + model: selectedModel, + input, + stream: true, + }; + if (previousInteractionId) { + body.previous_interaction_id = previousInteractionId; + } + + try { + const response = await fetch(requestUrl, { + method: "POST", + headers, + body: JSON.stringify(body), + signal, + }); + + if (!response.ok) { + const errorText = await response.text(); + throw new Error(errorText || `Request failed with status ${response.status}`); + } + + if (!response.body) { + throw new Error("No response body received"); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let responseModel: string | undefined; + let buffer = ""; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + + // SSE lines are separated by double newlines; split on single newlines and + // look for "data: " prefixed lines. + const lines = buffer.split("\n"); + // Keep the last (potentially incomplete) line in the buffer + buffer = lines.pop() ?? ""; + + for (const line of lines) { + const trimmed = line.trim(); + if (!trimmed.startsWith("data:")) continue; + + const jsonStr = trimmed.slice("data:".length).trim(); + if (!jsonStr || jsonStr === "[DONE]") continue; + + let event: Record; + try { + event = JSON.parse(jsonStr); + } catch { + continue; + } + + const eventType = event.event_type as string | undefined; + + if (eventType === "interaction.start" || eventType === "interaction.complete") { + // Capture model from either the native Gemini shape (nested under + // `interaction`) or the bridge shape (top-level `model` field). + const interaction = event.interaction as Record | undefined; + if (typeof interaction?.model === "string" && interaction.model) { + responseModel = interaction.model; + } else if (typeof event.model === "string" && event.model) { + responseModel = event.model; + } + } else if (eventType === "content.delta" || eventType === "content.start") { + const delta = event.delta as Record | undefined; + // Accept both native Gemini format {"type":"text","text":"..."} and bridge + // format {"text":"..."} (no type discriminator) + if (typeof delta?.text === "string" && delta.text) { + updateUI(delta.text, responseModel ?? selectedModel); + } + } + // content.start, content.stop, interaction.status_update — no UI action needed + } + } + } catch (error: unknown) { + if (signal?.aborted) { + console.log("Interactions request was cancelled"); + throw error; + } + NotificationManager.fromBackend( + `Error occurred while making Interactions API request. Error: ${error}`, + ); + throw error; + } +}