scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-19 00:01:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

fix(responses): honor caller stream flag when provider forces SSE

Browse source 
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
This commit is contained in:
SeongWoon Cho 2026-09-03 00:16:34 +09:00
parent 3ed6c19b8d
commit 5e2e009d49
No known key found for this signature in database
2 changed files with 170 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
litellm/llms/custom_httpx/llm_http_handler.py
View file

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

138
tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py
View file

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