From 5e2e009d494e99f3e33f14ea04c36660e9e82bab Mon Sep 17 00:00:00 2001 From: SeongWoon Cho Date: Thu, 3 Sep 2026 00:16:34 +0900 Subject: [PATCH] fix(responses): honor caller stream flag when provider forces SSE The Responses handlers decided whether to hand back a streaming iterator from the provider payload's `stream` field, which chatgpt sets unconditionally because the Codex backend only serves SSE. A caller that sent `stream: false` therefore received a raw SSE stream on /v1/responses, and the chat-completions bridge failed with "Unknown items in responses API response: []" once its recovery path lost the raw SSE it reads from Transport streaming still follows the provider payload; only the caller's own `stream` value now decides the response shape. When the provider forces SSE for a non-streaming caller the body is read and aggregated through the existing path --- litellm/llms/custom_httpx/llm_http_handler.py | 60 ++++---- .../custom_httpx/test_llm_http_handler.py | 138 +++++++++++++++++- 2 files changed, 170 insertions(+), 28 deletions(-) diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index 2fe4130a310..70329e600f7 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -2701,6 +2701,7 @@ class BaseLLMHTTPHandler: # Check if streaming is requested stream = response_api_optional_request_params.get("stream", False) + caller_requested_stream: Final = bool(stream) api_base: Final = responses_api_provider_config.get_complete_url( api_base=litellm_params.api_base, @@ -2775,8 +2776,20 @@ class BaseLLMHTTPHandler: stream=stream, **body_kwargs, ) - if fake_stream is True: - return MockResponsesAPIStreamingIterator( + if caller_requested_stream: + if fake_stream is True: + return MockResponsesAPIStreamingIterator( + response=response, + model=model, + logging_obj=logging_obj, + responses_api_provider_config=responses_api_provider_config, + litellm_metadata=litellm_metadata, + custom_llm_provider=custom_llm_provider, + request_data=request_context, + call_type=CallTypes.responses.value, + ) + + return SyncResponsesAPIStreamingIterator( response=response, model=model, logging_obj=logging_obj, @@ -2786,17 +2799,7 @@ class BaseLLMHTTPHandler: request_data=request_context, call_type=CallTypes.responses.value, ) - - return SyncResponsesAPIStreamingIterator( - response=response, - model=model, - logging_obj=logging_obj, - responses_api_provider_config=responses_api_provider_config, - litellm_metadata=litellm_metadata, - custom_llm_provider=custom_llm_provider, - request_data=request_context, - call_type=CallTypes.responses.value, - ) + response.read() else: response = sync_httpx_client.post( url=api_base, @@ -2889,6 +2892,7 @@ class BaseLLMHTTPHandler: # Check if streaming is requested stream = response_api_optional_request_params.get("stream", False) + caller_requested_stream: Final = bool(stream) api_base: Final = responses_api_provider_config.get_complete_url( api_base=litellm_params.api_base, @@ -2964,8 +2968,21 @@ class BaseLLMHTTPHandler: **body_kwargs, ) - if fake_stream is True: - return MockResponsesAPIStreamingIterator( + if caller_requested_stream: + if fake_stream is True: + return MockResponsesAPIStreamingIterator( + response=response, + model=model, + logging_obj=logging_obj, + responses_api_provider_config=responses_api_provider_config, + litellm_metadata=litellm_metadata, + custom_llm_provider=custom_llm_provider, + request_data=request_context, + call_type=CallTypes.responses.value, + ) + + # Return the streaming iterator + return ResponsesAPIStreamingIterator( response=response, model=model, logging_obj=logging_obj, @@ -2975,18 +2992,7 @@ class BaseLLMHTTPHandler: request_data=request_context, call_type=CallTypes.responses.value, ) - - # Return the streaming iterator - return ResponsesAPIStreamingIterator( - response=response, - model=model, - logging_obj=logging_obj, - responses_api_provider_config=responses_api_provider_config, - litellm_metadata=litellm_metadata, - custom_llm_provider=custom_llm_provider, - request_data=request_context, - call_type=CallTypes.responses.value, - ) + await response.aread() else: response = await async_httpx_client.post( url=api_base, diff --git a/tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py b/tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py index c39779972c0..6f2bcaa15f5 100644 --- a/tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py +++ b/tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py @@ -25,6 +25,7 @@ from litellm.llms.base_llm.search.transformation import BaseSearchConfig, Search from litellm.llms.bedrock.base_aws_llm import SignsRequestsWithAWS from litellm.llms.brave.search.transformation import BraveSearchConfig from litellm.llms.base_llm.image_edit.transformation import BaseImageEditConfig +from litellm.llms.chatgpt.responses.transformation import ChatGPTResponsesAPIConfig from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler from litellm.llms.custom_httpx.llm_http_handler import ( BaseLLMHTTPHandler, @@ -40,6 +41,10 @@ from litellm.llms.bedrock.messages.invoke_transformations.anthropic_claude3_tran from litellm.llms.mistral.ocr.transformation import MistralOCRConfig from litellm.llms.openai.videos.transformation import OpenAIVideoConfig from litellm.llms.tinyfish.search.transformation import TinyfishSearchConfig +from litellm.responses.streaming_iterator import ( + BaseResponsesAPIStreamingIterator, + MockResponsesAPIStreamingIterator, +) from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.router import GenericLiteLLMParams from litellm.types.utils import ImageObject, ImageResponse, ModelResponse, TranscriptionResponse @@ -420,6 +425,7 @@ async def test_async_response_api_handler_streams_when_provider_transform_adds_s ) ) logging_obj = Mock() + logging_obj.dynamic_success_callbacks = None await handler.async_response_api_handler( model="gpt-5.3-codex", @@ -436,6 +442,136 @@ async def test_async_response_api_handler_streams_when_provider_transform_adds_s assert client.post.call_args.kwargs["json"]["stream"] is True +_CHATGPT_SSE_BODY = ( + "event: response.output_item.done\n" + 'data: {"type": "response.output_item.done", "output_index": 0, "item": {"type": "message", ' + '"id": "msg_1", "status": "completed", "role": "assistant", "content": [{"type": "output_text", ' + '"text": "aggregated", "annotations": []}]}}\n' + "\n" + "event: response.completed\n" + 'data: {"type": "response.completed", "response": {"id": "resp_1", "object": "response", ' + '"created_at": 1, "model": "gpt-5.3-codex", "status": "completed", "output": [], ' + '"parallel_tool_calls": false, "tool_choice": "auto", "tools": [], ' + '"usage": {"input_tokens": 1, "output_tokens": 2, "total_tokens": 3}}}\n' + "\n" +) + + +def _chatgpt_sse_response(): + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=_CHATGPT_SSE_BODY.encode(), + request=httpx.Request("POST", "https://chatgpt.example.com/responses"), + ) + + +def _chatgpt_responses_logging_obj(): + logging_obj = Mock() + logging_obj.dynamic_success_callbacks = None + logging_obj.async_success_handler = AsyncMock() + return logging_obj + + +def _chatgpt_handler_kwargs(caller_params, client): + return { + "model": "gpt-5.3-codex", + "input": "hi", + "responses_api_provider_config": ChatGPTResponsesAPIConfig(), + "response_api_optional_request_params": caller_params, + "custom_llm_provider": "chatgpt", + "litellm_params": GenericLiteLLMParams(api_key="sk-test", api_base="https://chatgpt.example.com"), + "logging_obj": _chatgpt_responses_logging_obj(), + "client": client, + } + + +def _assert_aggregated_chatgpt_response(result): + assert not isinstance(result, BaseResponsesAPIStreamingIterator) + assert isinstance(result, ResponsesAPIResponse) + assert result.id == "resp_1" + assert result.output[0].content[0].text == "aggregated" + + +@pytest.mark.parametrize("caller_params", [{}, {"stream": False}]) +def test_response_api_handler_aggregates_chatgpt_sse_for_a_non_streaming_caller(caller_params): + """chatgpt forces `stream: true` on every request because the Codex backend only serves SSE. + A caller that did not ask for streaming must still get one aggregated ResponsesAPIResponse.""" + handler = BaseLLMHTTPHandler() + client = HTTPHandler(client=httpx.Client()) + client.post = Mock(return_value=_chatgpt_sse_response()) + + result = handler.response_api_handler(**_chatgpt_handler_kwargs(caller_params, client)) + + assert client.post.call_args.kwargs["json"]["stream"] is True + _assert_aggregated_chatgpt_response(result) + + +def test_response_api_handler_streams_chatgpt_sse_for_a_streaming_caller(): + handler = BaseLLMHTTPHandler() + client = HTTPHandler(client=httpx.Client()) + client.post = Mock(return_value=_chatgpt_sse_response()) + + result = handler.response_api_handler(**_chatgpt_handler_kwargs({"stream": True}, client)) + + assert isinstance(result, BaseResponsesAPIStreamingIterator) + assert [event.type for event in result][-1] == "response.completed" + + +def test_response_api_handler_fake_streams_only_for_a_streaming_caller(): + handler = BaseLLMHTTPHandler() + client = HTTPHandler(client=httpx.Client()) + client.post = Mock(return_value=_chatgpt_sse_response()) + + streamed = handler.response_api_handler(fake_stream=True, **_chatgpt_handler_kwargs({"stream": True}, client)) + assert isinstance(streamed, MockResponsesAPIStreamingIterator) + + client.post = Mock(return_value=_chatgpt_sse_response()) + aggregated = handler.response_api_handler(fake_stream=True, **_chatgpt_handler_kwargs({}, client)) + _assert_aggregated_chatgpt_response(aggregated) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("caller_params", [{}, {"stream": False}]) +async def test_async_response_api_handler_aggregates_chatgpt_sse_for_a_non_streaming_caller(caller_params): + handler = BaseLLMHTTPHandler() + client = AsyncHTTPHandler() + client.post = AsyncMock(return_value=_chatgpt_sse_response()) + + result = await handler.async_response_api_handler(**_chatgpt_handler_kwargs(caller_params, client)) + + assert client.post.call_args.kwargs["json"]["stream"] is True + _assert_aggregated_chatgpt_response(result) + + +@pytest.mark.asyncio +async def test_async_response_api_handler_streams_chatgpt_sse_for_a_streaming_caller(): + handler = BaseLLMHTTPHandler() + client = AsyncHTTPHandler() + client.post = AsyncMock(return_value=_chatgpt_sse_response()) + + result = await handler.async_response_api_handler(**_chatgpt_handler_kwargs({"stream": True}, client)) + + assert isinstance(result, BaseResponsesAPIStreamingIterator) + assert [event.type async for event in result][-1] == "response.completed" + + +@pytest.mark.asyncio +async def test_async_response_api_handler_fake_streams_only_for_a_streaming_caller(): + handler = BaseLLMHTTPHandler() + client = AsyncHTTPHandler() + client.post = AsyncMock(return_value=_chatgpt_sse_response()) + + streamed = await handler.async_response_api_handler( + fake_stream=True, **_chatgpt_handler_kwargs({"stream": True}, client) + ) + assert isinstance(streamed, MockResponsesAPIStreamingIterator) + + client.post = AsyncMock(return_value=_chatgpt_sse_response()) + aggregated = await handler.async_response_api_handler(fake_stream=True, **_chatgpt_handler_kwargs({}, client)) + _assert_aggregated_chatgpt_response(aggregated) + + @pytest.mark.asyncio async def test_async_response_api_handler_streaming_passes_logging_obj_to_post(): """LIT-5466: @track_llm_api_timing only records llm_api_duration_ms when the POST @@ -461,7 +597,7 @@ async def test_async_response_api_handler_streaming_passes_logging_obj_to_post() model="gpt-5", input="hi", responses_api_provider_config=config, - response_api_optional_request_params={}, + response_api_optional_request_params={"stream": True}, custom_llm_provider="chatgpt", litellm_params=GenericLiteLLMParams(), logging_obj=logging_obj,