From 3d904bf2cafab2408bf55f54962f43b392a80172 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Wed, 28 Oct 2026 12:39:12 -0700 Subject: [PATCH] fix(responses): give proxy stream resumes the deployment stream_timeout and document retrieve query params --- litellm/proxy/_lazy_openapi_snapshot.json | 42 +++++++++++++++++- .../proxy/response_api_endpoints/endpoints.py | 39 +++++++++++++++++ litellm/responses/retrieve_stream_query.py | 13 ++++++ litellm/router.py | 8 ++-- .../responses/test_responses_api_lifecycle.py | 43 ++++++++++++++++--- ui/litellm-dashboard/src/lib/http/schema.d.ts | 42 ++++++++++++++++-- 6 files changed, 173 insertions(+), 14 deletions(-) diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index cc0ec03715d..8600b2dd0a9 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -33564,7 +33564,7 @@ ] }, "get": { - "description": "Get a response by ID.\n\nSupports both:\n- Polling IDs (litellm_poll_*): Returns cumulative cached content from background responses\n- Provider response IDs: Passes through to provider API\n\nFollows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/get\n\n```bash\n# Get polling response\ncurl -X GET http://localhost:4000/v1/responses/litellm_poll_abc123 -H \"Authorization: Bearer $LITELLM_MASTER_KEY\"\n\n# Get provider response\ncurl -X GET http://localhost:4000/v1/responses/resp_abc123 -H \"Authorization: Bearer $LITELLM_MASTER_KEY\"\n```", + "description": "Get a response by ID.\n\nSupports both:\n- Polling IDs (litellm_poll_*): Returns cumulative cached content from background responses\n- Provider response IDs: Passes through to provider API\n\nFollows the OpenAI Responses API spec: https://platform.openai.com/docs/api-reference/responses/get\n\n```bash\n# Get polling response\ncurl -X GET http://localhost:4000/v1/responses/litellm_poll_abc123 -H \"Authorization: Bearer $LITELLM_MASTER_KEY\"\n\n# Get provider response\ncurl -X GET http://localhost:4000/v1/responses/resp_abc123 -H \"Authorization: Bearer $LITELLM_MASTER_KEY\"\n\n# Resume a background response's event stream after sequence number 7\ncurl -N \"http://localhost:4000/v1/responses/resp_abc123?stream=true&starting_after=7\" -H \"Authorization: Bearer $LITELLM_MASTER_KEY\"\n```", "operationId": "get_response_openai_v1_responses__response_id__get", "parameters": [ { @@ -33575,6 +33575,46 @@ "title": "Response Id", "type": "string" } + }, + { + "description": "Replay the response's events as server-sent events (OpenAI, Azure, and Azure AI Foundry)", + "in": "query", + "name": "stream", + "required": false, + "schema": { + "type": "boolean" + } + }, + { + "description": "With stream=true, replay only the events after this sequence number", + "in": "query", + "name": "starting_after", + "required": false, + "schema": { + "minimum": 0, + "type": "integer" + } + }, + { + "description": "Additional output data to include, forwarded to the provider", + "in": "query", + "name": "include", + "required": false, + "schema": { + "items": { + "type": "string" + }, + "type": "array" + } + }, + { + "description": "With stream=true, whether delta events carry obfuscation padding", + "in": "query", + "name": "include_obfuscation", + "required": false, + "schema": { + "type": "boolean" + } } ], "responses": { diff --git a/litellm/proxy/response_api_endpoints/endpoints.py b/litellm/proxy/response_api_endpoints/endpoints.py index 9f1924d7f23..d1d984516d5 100644 --- a/litellm/proxy/response_api_endpoints/endpoints.py +++ b/litellm/proxy/response_api_endpoints/endpoints.py @@ -104,6 +104,38 @@ _RESPONSES_WS_CONFIG_VALUE_ADAPTER: Final[TypeAdapter[object | None]] = TypeAdap _DECODED_JSON: Final = TypeAdapter(object) _PROXY_SERVER_REQUEST_ADAPTER: Final[TypeAdapter[Mapping[str, object]]] = TypeAdapter(Mapping[str, object]) _RETRIEVE_QUERY_PARAMS: Final = tuple(sorted(RETRIEVE_INCLUDE_QUERY_KEYS | RETRIEVE_STREAM_QUERY_KEYS)) +_RETRIEVE_QUERY_OPENAPI: Final = { + "parameters": [ + { + "name": "stream", + "in": "query", + "required": False, + "schema": {"type": "boolean"}, + "description": "Replay the response's events as server-sent events (OpenAI, Azure, and Azure AI Foundry)", + }, + { + "name": "starting_after", + "in": "query", + "required": False, + "schema": {"type": "integer", "minimum": 0}, + "description": "With stream=true, replay only the events after this sequence number", + }, + { + "name": "include", + "in": "query", + "required": False, + "schema": {"type": "array", "items": {"type": "string"}}, + "description": "Additional output data to include, forwarded to the provider", + }, + { + "name": "include_obfuscation", + "in": "query", + "required": False, + "schema": {"type": "boolean"}, + "description": "With stream=true, whether delta events carry obfuscation padding", + }, + ] +} _ResponseDocSchemas: TypeAlias = dict[int | str, dict[str, object]] # fastapi's responses kwarg @@ -743,18 +775,21 @@ async def cursor_chat_completions( dependencies=[Depends(user_api_key_auth)], tags=["responses"], responses=RESPONSES_API_RESPONSE_SCHEMAS, + openapi_extra=_RETRIEVE_QUERY_OPENAPI, ) @router.get( "/responses/{response_id}", dependencies=[Depends(user_api_key_auth)], tags=["responses"], responses=RESPONSES_API_RESPONSE_SCHEMAS, + openapi_extra=_RETRIEVE_QUERY_OPENAPI, ) @router.get( "/openai/v1/responses/{response_id}", dependencies=[Depends(user_api_key_auth)], tags=["responses"], responses=RESPONSES_API_RESPONSE_SCHEMAS, + openapi_extra=_RETRIEVE_QUERY_OPENAPI, ) async def get_response( response_id: str, @@ -779,6 +814,10 @@ async def get_response( # Get provider response curl -X GET http://localhost:4000/v1/responses/resp_abc123 \ -H "Authorization: Bearer $LITELLM_MASTER_KEY" + + # Resume a background response's event stream after sequence number 7 + curl -N "http://localhost:4000/v1/responses/resp_abc123?stream=true&starting_after=7" \ + -H "Authorization: Bearer $LITELLM_MASTER_KEY" ``` """ from litellm.proxy.proxy_server import ( diff --git a/litellm/responses/retrieve_stream_query.py b/litellm/responses/retrieve_stream_query.py index 03b6da116f2..02160985e7b 100644 --- a/litellm/responses/retrieve_stream_query.py +++ b/litellm/responses/retrieve_stream_query.py @@ -1,8 +1,12 @@ +from collections.abc import Mapping from dataclasses import dataclass from typing import Final, Literal +from pydantic import TypeAdapter + RETRIEVE_STREAM_QUERY_KEYS: Final = frozenset({"stream", "starting_after"}) RETRIEVE_INCLUDE_QUERY_KEYS: Final = frozenset({"include", "include[]", "include_obfuscation"}) +_QUERY_ADAPTER: Final = TypeAdapter(Mapping[str, object]) @dataclass(frozen=True, slots=True) @@ -59,3 +63,12 @@ def parse_retrieve_stream_query( if isinstance(parsed_starting_after, InvalidRetrieveStreamQuery): return _FULL_REPLAY if drop_malformed else parsed_starting_after return RetrieveStreamQuery(stream=True, starting_after=parsed_starting_after) + + +def requests_streamed_retrieval(request: Mapping[str, object]) -> bool: + extra_query: Final = request.get("extra_query") + return ( + "response_id" in request + and isinstance(extra_query, Mapping) + and _parsed_stream(_QUERY_ADAPTER.validate_python(extra_query).get("stream")) is True + ) diff --git a/litellm/router.py b/litellm/router.py index c8153b596d6..01ae5175d92 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -148,6 +148,7 @@ from litellm.llms.openai_like.model_info import ( get_openai_compatible_model_info, ) from litellm.responses.bridge_check import keep_registered_mode +from litellm.responses.retrieve_stream_query import requests_streamed_retrieval from litellm.router_strategy.base_routing_strategy import BaseRoutingStrategy from litellm.router_strategy.budget_limiter import RouterBudgetLimiting from litellm.router_strategy.complexity_router.context_compaction import ( @@ -4970,12 +4971,13 @@ class Router: router_timeout=_router_timeout, router_stream_timeout=_router_stream_timeout, ) + timeout_request: Final = {**kwargs, "stream": True} if requests_streamed_retrieval(kwargs) else kwargs kwargs.update( CompletionTimeout.stream_request_kwargs( - passthrough_timeout(kwargs={**kwargs, "stream": False}), passthrough_timeout(kwargs=kwargs) + passthrough_timeout(kwargs={**kwargs, "stream": False}), passthrough_timeout(kwargs=timeout_request) ) - if splits_stream_timeout and kwargs.get("stream") - else {"timeout": passthrough_timeout(kwargs=kwargs)} + if splits_stream_timeout and timeout_request.get("stream") + else {"timeout": passthrough_timeout(kwargs=timeout_request)} ) elif splits_stream_timeout: request_kwargs: Final = _REQUEST_KWARGS_ADAPTER.validate_python(kwargs) diff --git a/tests/unit/responses/test_responses_api_lifecycle.py b/tests/unit/responses/test_responses_api_lifecycle.py index 193ba4c89a1..2197c83e27e 100644 --- a/tests/unit/responses/test_responses_api_lifecycle.py +++ b/tests/unit/responses/test_responses_api_lifecycle.py @@ -1,7 +1,6 @@ import json import queue import threading -import time from collections.abc import AsyncIterator, Iterator from dataclasses import dataclass, field from typing import Final, Literal, TypeAlias @@ -501,6 +500,38 @@ async def test_streamed_retrieve_on_an_unsupported_provider_drops_the_stream_und assert dict(call.request.url.params) == {} +@pytest.mark.asyncio +async def test_a_routed_resume_requested_in_extra_query_gets_the_deployment_stream_timeout() -> None: + router: Final = litellm.Router( + model_list=[ + { + "model_name": "background-model", + "litellm_params": { + "model": "openai/gpt-5.5", + "api_key": "sk-test", + "api_base": "https://api.openai.com/v1", + "timeout": 2, + "stream_timeout": 60, + }, + } + ] + ) + with respx.mock() as mock_router: + route: Final = _retrieve_route(mock_router, "openai").mock( + return_value=httpx.Response(200, headers=_STREAM_HEADERS, content=_REPLAY_AFTER_3) + ) + stream: Final = await router.aget_responses( + model="background-model", + response_id=_STORED_RESPONSE_ID, + extra_query={"stream": "true", "starting_after": "3"}, + ) + event_types: Final = await _event_types(stream) + (call,) = route.calls + + assert event_types == ("response.output_text.delta", "response.completed") + assert call.request.extensions["timeout"]["read"] == 60 + + class _SuccessLog(CustomLogger): def __init__(self) -> None: super().__init__() @@ -510,7 +541,7 @@ class _SuccessLog(CustomLogger): self.logged.put(response_obj) -def test_a_sync_streamed_retrieve_logs_success_once_with_the_finished_response( +def test_a_sync_streamed_retrieve_logs_success_with_the_finished_response( monkeypatch: pytest.MonkeyPatch, ) -> None: success_log: Final = _SuccessLog() @@ -525,10 +556,8 @@ def test_a_sync_streamed_retrieve_logs_success_once_with_the_finished_response( assert isinstance(stream, SyncResponsesAPIStreamingIterator) consumed: Final = tuple(stream) - first_logged: Final = success_log.logged.get(timeout=10) - time.sleep(0.5) + logged: Final = success_log.logged.get(timeout=10) assert len(consumed) == 2 - assert isinstance(first_logged, ResponsesAPIResponse) - assert first_logged.status == "completed" - assert success_log.logged.empty() + assert isinstance(logged, ResponsesAPIResponse) + assert logged.status == "completed" diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 26805e66280..7b9378de516 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -11211,6 +11211,9 @@ export interface paths { * * # Get provider response * curl -X GET http://localhost:4000/v1/responses/resp_abc123 -H "Authorization: Bearer $LITELLM_MASTER_KEY" + * + * # Resume a background response's event stream after sequence number 7 + * curl -N "http://localhost:4000/v1/responses/resp_abc123?stream=true&starting_after=7" -H "Authorization: Bearer $LITELLM_MASTER_KEY" * ``` */ get: operations["get_response_openai_v1_responses__response_id__get"]; @@ -14269,6 +14272,9 @@ export interface paths { * * # Get provider response * curl -X GET http://localhost:4000/v1/responses/resp_abc123 -H "Authorization: Bearer $LITELLM_MASTER_KEY" + * + * # Resume a background response's event stream after sequence number 7 + * curl -N "http://localhost:4000/v1/responses/resp_abc123?stream=true&starting_after=7" -H "Authorization: Bearer $LITELLM_MASTER_KEY" * ``` */ get: operations["get_response_responses__response_id__get"]; @@ -21890,6 +21896,9 @@ export interface paths { * * # Get provider response * curl -X GET http://localhost:4000/v1/responses/resp_abc123 -H "Authorization: Bearer $LITELLM_MASTER_KEY" + * + * # Resume a background response's event stream after sequence number 7 + * curl -N "http://localhost:4000/v1/responses/resp_abc123?stream=true&starting_after=7" -H "Authorization: Bearer $LITELLM_MASTER_KEY" * ``` */ get: operations["get_response_v1_responses__response_id__get"]; @@ -67070,7 +67079,16 @@ export interface operations { }; get_response_openai_v1_responses__response_id__get: { parameters: { - query?: never; + query?: { + /** @description Replay the response's events as server-sent events (OpenAI, Azure, and Azure AI Foundry) */ + stream?: boolean; + /** @description With stream=true, replay only the events after this sequence number */ + starting_after?: number; + /** @description Additional output data to include, forwarded to the provider */ + include?: string[]; + /** @description With stream=true, whether delta events carry obfuscation padding */ + include_obfuscation?: boolean; + }; header?: never; path: { response_id: string; @@ -70535,7 +70553,16 @@ export interface operations { }; get_response_responses__response_id__get: { parameters: { - query?: never; + query?: { + /** @description Replay the response's events as server-sent events (OpenAI, Azure, and Azure AI Foundry) */ + stream?: boolean; + /** @description With stream=true, replay only the events after this sequence number */ + starting_after?: number; + /** @description Additional output data to include, forwarded to the provider */ + include?: string[]; + /** @description With stream=true, whether delta events carry obfuscation padding */ + include_obfuscation?: boolean; + }; header?: never; path: { response_id: string; @@ -81334,7 +81361,16 @@ export interface operations { }; get_response_v1_responses__response_id__get: { parameters: { - query?: never; + query?: { + /** @description Replay the response's events as server-sent events (OpenAI, Azure, and Azure AI Foundry) */ + stream?: boolean; + /** @description With stream=true, replay only the events after this sequence number */ + starting_after?: number; + /** @description Additional output data to include, forwarded to the provider */ + include?: string[]; + /** @description With stream=true, whether delta events carry obfuscation padding */ + include_obfuscation?: boolean; + }; header?: never; path: { response_id: string;