feat(ui): add Interactions API endpoint to playground with SSE streaming (#28156)

* feat(ui): add Interactions API support to playground with streaming

Adds /v1beta/interactions as a selectable endpoint in the UI playground.
Uses SSE streaming (stream=true) and parses content.delta events for real-time output.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(interactions): remove forced gemini provider so all providers work via interactions API

Proxy endpoint was hardcoding custom_llm_provider="gemini" before routing,
preventing non-Gemini models from using the litellm_responses bridge.
Also reverts the UI Gemini-only model filter.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(interactions): fix streaming for non-gemini providers via bridge

Two bugs in LiteLLMResponsesInteractionsStreamingIterator:
1. content.delta was emitted without "type":"text" in delta dict, so the
   UI type-check always failed and no tokens were displayed
2. First OutputTextDeltaEvent was silently dropped (used to emit content.start
   with empty text); fixed by handling ResponsePartAddedEvent for content.start
   so text deltas go directly to content.delta

Co-authored-by: Cursor <cursoragent@cursor.com>

* undo unrelated changes

* fix(ui): extract model from top-level field in interactions bridge events

Co-authored-by: Yassin Kortam <yassin@berri.ai>

* test(interactions): remove tautological gemini-provider assertion

The test_no_forced_gemini_provider_in_request_data check only asserted
against dict literals it had just constructed, so it always passed and
did not exercise the create_interaction endpoint. The endpoint
deliberately defaults custom_llm_provider to gemini, so the assertion
was also factually incorrect. Drop the misleading test.

Co-authored-by: Yassin Kortam <yassin@berri.ai>

* fix(interactions): use ContentPartAddedEvent and guard interaction.start ordering

- ResponsePartAddedEvent corresponds to reasoning summary parts, not text
  content parts. Use ContentPartAddedEvent which is the event emitted before
  text output deltas (type response.content_part.added).
- Mirror the OutputTextDeltaEvent ordering guard: if interaction.start has
  not been sent yet, emit it first before content.start to honor the
  documented event ordering contract.

Co-authored-by: Yassin Kortam <yassin@berri.ai>

* test(interactions): cover ContentPartAddedEvent ordering and no-op paths

* fix(tests): treat corrupt VCR cassette payloads as cache miss + use gpt-realtime in OpenAI realtime guardrails test

VCR redis persister was raising UnicodeDecodeError on cached payloads that
fail to UTF-8 decode (e.g. legacy entries written by another version of
the persister), failing tests at fixture setup instead of degrading to a
cache miss. Wrap decode+deserialize in a try/except so corrupt cache
entries are treated as CassetteNotFoundError, surfacing the failure via
the existing _record_cache_failure / VCRCassetteCacheWarning path.

OpenAI shut down gpt-4o-realtime-preview-2024-12-17 (and the entire
gpt-4o-realtime-preview family) on 2026-05-07. The live realtime
guardrails integration test now fails with model_not_found instead of
receiving session.created. Point OPENAI_REALTIME_URL at the current GA
model gpt-realtime, and relax the assertion in
test_text_message_blocked_by_guardrail_no_ai_response to also accept the
model's refusal-to-repeat the block message (gpt-realtime declines
verbatim-repeat instructions, which is still a safe outcome since the
original user message was blocked before reaching OpenAI). The
BLOCKED_PHRASE leak check is preserved as a hard invariant.

* fix(tests): migrate realtime + nvidia_nim rerank tests off shut-down upstream models

OpenAI shut down the entire gpt-4o-realtime-preview family (including the
undated alias) on 2026-05-07. The live realtime tests still connected
with that dead alias and failed with messages_received=1 (an error event
'The model gpt-4o-realtime-preview does not exist' instead of
session.created). Point the live OpenAI realtime tests at gpt-realtime,
the current GA realtime model:

- test_openai_realtime_simple.py: get_model() -> gpt-realtime
- test_openai_realtime.py: test_openai_realtime_direct_call_no_intent and
  test_openai_realtime_direct_call_with_intent -> openai/gpt-realtime

Mocked unit tests (test_realtime_query_params_construction,
test_realtime_query_params_use_normalized_model_name) are left as-is:
they never hit the network and assert string plumbing only.

NVIDIA reached end-of-life for the hosted
nvidia/llama-3.2-nv-rerankqa-1b-v2 rerank API on 2026-05-18 with no
published replacement, so the live BaseLLMRerankTest.test_basic_rerank
for nvidia_nim now returns HTTP 410 ('Gone'). NVIDIA's hosted catalog
rotates on a schedule, so swapping in another live model would only
defer the failure. Override test_basic_rerank in TestNvidiaNim to mock
the sync/async HTTP transport (same pattern as
test_nvidia_nim_rerank_ranking_endpoint in this file) and inject a fake
NVIDIA_NIM_API_KEY via monkeypatch. The request/response transformation
and cost calculation stay covered offline.

* test(callbacks): harden flaky proxy callback-leak detector

The proxy callback-leak detector (test_check_num_callbacks_on_lowest_latency)
was failing on this PR with 'abs(85 - 95) <= 4' — a bounded one-time
registration jump caused by switching to latency-based-routing
(+LowestLatencyLoggingHandler, +SlackAlerting). The count then plateaus
under load, so this is pollution from the test's own config update, not a
leak.

Replace the brittle two-sample diff threshold with a sampler that settles
past the deliberate config switch and only flags sustained monotonic
per-type growth, with a terminal-burst confirmation pass for leaks that
would otherwise escape the >=2-interval guard. Normalizes instance
addresses so identical callbacks at different memory locations collapse,
and names the leaking type on failure.

* fix(interactions): preserve first text token when both start events are missing

When OutputTextDeltaEvent arrived before any ResponseCreatedEvent or
ContentPartAddedEvent, the double-fallback path emitted interaction.start
and silently dropped the first delta's text — the second delta's
content.start carried only that chunk's delta, and the first token never
made it to any content.delta event consumed by the UI.

Queue a content.start that carries the first delta's text alongside the
interaction.start emission, and drain pending events before pulling the
next upstream chunk.

* chore(ui): remove unused InteractionOutput/InteractionResponse interfaces

Co-authored-by: Yassin Kortam <yassin@berri.ai>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Yassin Kortam <yassin@berri.ai>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
Sameer Kankute 2026-05-19 08:25:40 +05:30 committed by GitHub
parent 761c280a6e
commit 581882879d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 388 additions and 16 deletions

View file

@ -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

View file

@ -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_<id(self)> 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."""

View file

@ -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<ChatUIProps> = ({
EndpointType.ANTHROPIC_MESSAGES,
EndpointType.EMBEDDINGS,
EndpointType.TRANSCRIPTION,
EndpointType.INTERACTIONS,
];
if (modelRequiredEndpoints.includes(endpointType as EndpointType) && !selectedModel) {
@ -914,6 +916,16 @@ const ChatUI: React.FC<ChatUIProps> = ({
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<ChatUIProps> = ({
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<ChatUIProps> = ({
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..."

View file

@ -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" },
];

View file

@ -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

View file

@ -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<void> {
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<string, string> = {
"Content-Type": "application/json",
[getGlobalLitellmHeaderName()]: `Bearer ${accessToken}`,
};
if (tags && tags.length > 0) {
headers["x-litellm-tags"] = tags.join(",");
}
const body: Record<string, unknown> = {
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<string, unknown>;
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<string, unknown> | 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<string, unknown> | 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;
}
}