Skip to content

fix(responses): honor caller stream flag when provider forces SSE (internal copy of #34095) #41235

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mateo-berri merged 5 commits into from
Oct 6, 2026
Merged

fix(responses): honor caller stream flag when provider forces SSE (internal copy of #34095) #41235

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5e2e009
fix(responses): honor caller stream flag when provider forces SSE
SunSet0418 Sep 2, 2026
04e164b
test(chatgpt): inject the authenticator into the responses config so …
mateo-berri Sep 15, 2026
20fe356
Merge remote-tracking branch 'origin/main' into litellm_fix_responses…
mateo-berri Oct 5, 2026
78b0bea
fix(responses): treat an extra_body stream flag as the caller's own a…
mateo-berri Oct 5, 2026
f1cdfae
test(integration): audit the chatgpt caller stream flag across respon…
mateo-berri Oct 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions litellm/llms/chatgpt/responses/transformation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,9 @@


class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig):
def __init__(self) -> None:
def __init__(self, authenticator: Authenticator | None = None) -> None:
super().__init__()
self.authenticator = Authenticator()
self.authenticator = authenticator if authenticator is not None else Authenticator()

@property
def custom_llm_provider(self) -> LlmProviders:
Expand Down
59 changes: 32 additions & 27 deletions litellm/llms/custom_httpx/llm_http_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2507,6 +2507,7 @@ def response_api_handler(

# Check if streaming is requested
stream = response_api_optional_request_params.get("stream", False)
caller_requested_stream: Final = bool(stream or (extra_body or {}).get("stream"))

@greptile-apps greptile-apps Bot Oct 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Manus streaming skips fake stream

When a Manus caller sets extra_body={"stream": true} without setting the top-level stream flag, this line selects a real streaming iterator. But should_fake_stream receives only the top-level flag, so Manus does not enable the fake-stream path it needs because it does not support real-time streaming. The request is sent through the SSE iterator instead and fails to produce the expected stream. The async handler has the same mismatch.

@mateo-berri mateo-berri Oct 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pre-existing: fake_stream is decided in responses/main.py from the top-level flag before this handler runs, the same on main, and extra_body.stream never fed it

@greptile-apps greptile-apps Bot Oct 5, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're right — this mismatch is pre-existing and is not introduced by this PR. fake_stream is determined earlier from the top-level stream flag, and this handler has never propagated extra_body.stream into that decision (on main or in this change). The new caller_requested_stream only selects the response shape after that existing decision, so this comment does not identify a regression here. I’m withdrawing it.


api_base: Final = responses_api_provider_config.get_complete_url(
api_base=litellm_params.api_base,
Expand Down Expand Up @@ -2585,8 +2586,20 @@ def response_api_handler(
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,
Expand All @@ -2596,17 +2609,7 @@ def response_api_handler(
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,
Expand Down Expand Up @@ -2699,6 +2702,7 @@ async def async_response_api_handler(

# Check if streaming is requested
stream = response_api_optional_request_params.get("stream", False)
caller_requested_stream: Final = bool(stream or (extra_body or {}).get("stream"))

api_base: Final = responses_api_provider_config.get_complete_url(
api_base=litellm_params.api_base,
Expand Down Expand Up @@ -2778,8 +2782,20 @@ async def async_response_api_handler(
**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 ResponsesAPIStreamingIterator(
response=response,
model=model,
logging_obj=logging_obj,
Expand All @@ -2789,18 +2805,7 @@ async def async_response_api_handler(
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,
Expand Down
50 changes: 50 additions & 0 deletions tests/integration/_support/agentic_probe.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
from __future__ import annotations

import json
import os
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Final

from integration._support import responses_vendor as rv
from pydantic import JsonValue

from litellm.integrations.custom_logger import CustomLogger

OUT_ENVIRONMENT: Final = "AGENTIC_PROBE_OUT"


class AgenticProbe(CustomLogger):
"""Records every agentic-loop hook call the proxy makes, one JSON line per call, and never runs a loop."""

async def async_should_run_agentic_loop(
self,
response: object,
model: str,
messages: Sequence[Mapping[str, object]],
tools: Sequence[Mapping[str, object]] | None,
stream: bool,
custom_llm_provider: str,
kwargs: Mapping[str, object],
) -> tuple[bool, dict[str, object]]: # mutable-ok: the CustomLogger hook contract returns a dict
out: Final = os.environ.get(OUT_ENVIRONMENT)
if out:
line: Final[Mapping[str, JsonValue]] = {
"marker": rv.newest_marker(f"{messages!s} {response!s}"),
"surface": str(kwargs.get("_agentic_loop_api_surface")),
"response_type": type(response).__name__,
"stream": stream,
"model": model,
"provider": custom_llm_provider,
}
with open(out, "a", encoding="utf-8") as sink:
sink.write(json.dumps(line) + "\n")
return False, {}


probe: Final = AgenticProbe()


def lines(path: Path, marker: str) -> tuple[Mapping[str, JsonValue], ...]:
recorded: Final = tuple(rv.JSON_OBJECT.validate_json(line) for line in path.read_text().splitlines() if line)
return tuple(line for line in recorded if line["marker"] == marker)
Loading
Loading