Streaming: chat_stream + SSE endpoint - #7
Merged
Merged
Conversation
…e}/chat Runner.run_streamed's async iterator of raw response events replaces the blocking Runner.run for UX targets that need first output before the whole turn completes. HeadlessRunner.run_streamed mirrors run's trace/sandbox/ run_config lifecycle, App.chat_stream threads the same session as chat(), and the SSE endpoint (?stream=true) reassembles deltas into the done event's full output. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Review fixes for the streaming PR: - run_streamed wraps consumption in contextlib.aclosing and cancels the SDK RunResultStreaming in a finally, so an abandoned generator (client disconnect, which Starlette never aclose()s) stops the detached run loop before the sandbox unwinds instead of deferring to GC-time finalizers. - run_streamed now ends with a StreamDone(final_output, usage) sentinel; the done SSE event carries the SDK's final_output instead of the re-joined deltas, which disagreed with chat() for tool-using agents and returned raw JSON for output_type agents. usage rides along for token accounting. - serve.py reads session_id/message before the response starts (missing field -> 422, not a 200 that streams nothing) and emits an `event: error` frame, with no internal detail, when the turn fails mid-stream. - Failed or abandoned streamed turns record tr.set_output(error=...) so they show up as ERROR in Langfuse. - SSE responses send Cache-Control: no-cache and X-Accel-Buffering: no. - BaseRunner declares run_streamed (opt-in, not abstract); stale headless module docstring fixed; redundant asyncio markers dropped. - New endpoint tests over FastAPI's TestClient: frame sequence, 4xx on missing session_id, mid-stream error frame; plus a cancel-on-abandonment test. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Resolve app.py (chat_stream + open/aclose lifecycle), serve.py (streaming chat endpoint over dev's deck() 503 helper and error handlers), CHANGELOG. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
… 503s pre-startup) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
App.chat_stream(name, session_id, message, **runner_options)— async iterator of text deltas, wrapping the Agents SDKRunner.run_streamed. Same session semantics aschat().HeadlessRunner.run_streamed— streaming counterpart toHeadlessRunner.run, mirroring its trace_run span, sandbox attach/detach, andrun_config/max_turnshandling; the trace's output is set from the streamed result'sfinal_outputonce the generator is fully drained.POST /agents/{name}/chat?stream=true—text/event-streamresponse: incrementaldata: {"delta": ...}events, then oneevent: donecarrying the reassembled full output.chat()/POST /agents/{name}/chatpath is unchanged.Why
Middle's UX targets (first acknowledgement p95 < 5s) need incremental output; today
chat()blocks until the whole turn completes.Decisions / deviations
doneevent'soutputis the deltas rejoined, not a separate RunResult field — that's exactly the final output for a plain-text agent (the streaming use case), and keepschat_stream's contract to "async iterator of text deltas" per the issue, without holding the SDK'sRunResultStreamingopen past that.Runner.run_streamed) and stubHeadlessRunner.from_agentrather than mocking a live model — no live key needed. They assert: deltas arrive incrementally (not batched), non-text-delta events are filtered,run_config/max_turns/sessionare threaded through identically torun, andchat()/chat_stream()resolve the same session object for a givensession_id(session history stays identical whether a turn is streamed or not).Test plan
.venv/bin/ruff check agentdeck/ tests/.venv/bin/ty check agentdeck.venv/bin/pytest tests/ -q(7 passed)Closes #2
🤖 Generated with Claude Code