Skip to content

Streaming

The toolkit uses streaming internally for all provider API calls, preventing timeouts and enabling real-time output.

Why streaming

  1. Timeout prevention — Each streaming chunk resets the HTTP read timer. Long-running generations that would timeout with a single blocking call succeed because the connection stays active.
  2. Partial output capture — If an error occurs mid-generation, tokens received before the failure are still available for cost accounting.
  3. Retry cost tracking — Failed attempts accumulate cost data through RetryAccumulator, so retried requests no longer silently discard billing information.
  4. Real-time output — Applications can display output as it generates via Agent.stream().

Internal streaming vs Agent.stream()

Internal streaming (transparent)

All provider API calls use streaming under the hood. When you call Agent.run() or Agent.run_sync(), the streaming response is fully consumed internally and returned as a complete AgentResult. No code changes are needed — existing callers work identically.

# Streaming happens transparently — this works the same as before
result = await agent.run("Explain quantum computing")
print(result.output)

Agent.stream() (explicit)

For applications that want real-time output, Agent.stream() exposes an async iterator of AgentStreamEvent objects. The underlying execution is the same, but text deltas and tool events are yielded incrementally.

async for event in agent.stream("Explain quantum computing"):
    if event.event_type == "text_delta":
        print(event.content, end="", flush=True)

To stop reading before the stream ends, see Stopping a stream early. A bare break does not reliably cancel the request.

How streaming works

Provider transport layer

Each provider has a dedicated streaming implementation:

Provider Streaming API Notes
OpenAI client.responses.stream() → get_final_response() SDK helper with Pydantic parsing support
Anthropic client.messages.stream() → get_final_message() Returns identical Message object to non-streaming
Google client.models.generate_content_stream() Chunk iterator; usage_metadata on the final chunk
Fireworks client.chat.completions.create(stream=True), or the OpenAI SDK against the Fireworks Responses API on the remote-MCP path OpenAI-compatible streaming; usage in final chunk

All providers return the same result dict structure, so Agent._convert_result() needs no provider-specific streaming logic.

The OpenAI tool loop can read the same events from a WebSocket instead of one HTTP stream per request (Agent(openai_transport="websocket"), opt-in). The events a run emits are the same on both. See OpenAI WebSocket Transport.

Fireworks usage chunks across SDK eras

Some Fireworks models omit usage entirely unless include_usage is requested, which would produce silent zero-token (and therefore zero-cost) turns. The toolkit always requests it, but how it is sent depends on the installed SDK: fireworks-ai < 1.x takes stream_options as a direct keyword, while >= 1.x rejects that keyword and receives the same payload through extra_body. The request on the wire is identical, so cost tracking is accurate on both. See Providers.

Stream callback bridge

When Agent.stream() is called, it passes a stream_callback function into the internal execution pipeline. Provider-level code invokes this callback with (event_type, content, metadata) on each chunk. The callback enqueues AgentStreamEvent objects that the async generator yields to the caller.

Stream callback signature

The stream callback passed through the provider pipeline has the signature:

def stream_callback(
    event_type: str,
    content: str | None = None,
    metadata: dict | None = None,
) -> None:
    ...
  • text_delta uses content, plus metadata when the delta belongs to a structured payload (see Marking the structured payload). thinking_delta uses content only.
  • tool_call_start, tool_call_result, iteration_complete and request_start use the metadata parameter.

A callback that declares only (event_type, content) still receives every text_delta in the historical two-argument form: the toolkit inspects the callback's signature once per provider call and omits the metadata rather than raising. Note that tool_call_start, tool_call_result and iteration_complete have always been invoked with three positional arguments, so a strictly two-parameter callback has never been able to receive those — write the three-parameter form.

Stream event types

AgentStreamEvent is a dataclass with event_type, content, and metadata fields:

Event type Description content metadata
text_delta Partial text from the model The text chunk None for ordinary prose; {"structured_output": True} on a schema-constrained request
thinking_delta Partial reasoning/thinking output The thinking chunk None
tool_call_start A tool call is beginning None {"tool_name": str, "tool_call_id": str\|None, "arguments": dict, "iteration": int}
tool_call_result A tool call completed None {"tool_name": str, "tool_call_id": str\|None, "result": str, "result_truncated": bool, "success": bool, "duration_ms": int, "iteration": int}; result_data too when tool_result_preview_chars=0 (see Tool results in events), and tool_reported_cost: float / tool_reported_cost_estimated: bool when the tool reported spend
iteration_complete One tool-calling iteration finished None {"iteration": int, "input_tokens": int, "output_tokens": int, "cost": float, "tools_called": list, "stop_reason": str\|None}; the usage covers every model request of the tool loop billed since the previous event, so the events sum to the run's requests up to its last event (0.5.1; see Hooks)
turn_complete A non-final turn ended under a "new_turn" session None The same metadata complete would have carried, plus turn_index and auto_follow_up
turn_start A follow-up turn is starting None {"turn_index": int, "auto_follow_up": True, "trigger": "late_injection", "message": str}
error An error occurred, or a hook aborted the run Error message See error event metadata
complete Stream finished None Full AgentResult as dict, including injected_messages and undelivered_injections
request_start Opt-in: a real provider request is about to be sent None {"request_index": int, "iteration": int, "attempt": int, "purpose": str, "model": str, "provider": str}

Accessing the final result

The complete event's metadata contains the same data as AgentResult.to_dict(). You can inspect total tokens, cost, and tool usage from it.

Tool results in events

The toolkit keeps two copies of every tool result:

  • the model's copy, sent back in the next request and stored in the run's messages and the session history. max_tool_result_chars cuts it (by default at a quarter of the model's context window);
  • the client's copy, carried by the tool_call_result stream event and the hook event. Agent(tool_result_preview_chars=500) sets its length.

The two settings are independent: a longer client copy never changes what the model is sent, its tokens or its cost. result is the first 500 characters of the result's str() by default, as it always was, and result_truncated says whether that is all of it. With tool_result_preview_chars=0 the event carries the whole result, and adds result_data: the result as JSON data, which a client can use directly where result is only Python's str():

The tool returned result_data
A dict (a local @tool), or an MCP result that is one JSON-object text item That object, e.g. {"output_path": "...", "duration_seconds": 3.1}
Plain text The text
Several MCP content items (text and images) The list of items, each text item's JSON decoded in place; images keep their base64

result_data is always valid JSON: a NaN or infinite number becomes null, a value JSON has no type for becomes its str(), and a result JSON cannot represent at all (a tuple dict key, a circular reference) arrives as the result's str(). Building the event never fails the run. The fields are built by utils.tool_result_event_fields, and result_data by utils.tool_result_data.

agent = Agent(model="claude-sonnet-5", tools=[fetch_report], tool_result_preview_chars=0)

async for event in agent.stream("Summarise the Q3 report."):
    if event.event_type == "tool_call_result":
        report = event.metadata["result_data"]      # the tool's dict, whole
        assert event.metadata["result_truncated"] is False

Keep the default where events are logged or stored

A full result can be megabytes (an image tool's base64, a long transcript). DEBUG_AGENT_STREAMING mirrors every event into the log, and a hook may store the events it sees. Turn the full copy on only where the client needs it.

The same key works in YAML (tool_result_preview_chars: 0) and as an Agent.from_config override.

request_start is opt-in

request_start is emitted only when the Agent was built with Agent(emit_request_start=True) (there is a matching emit_request_start: YAML key). By default the stream carries exactly the other events listed above, so a consumer that enumerates event types is unaffected, and with the flag off no emitter object is constructed at all.

When it is enabled, one event fires immediately before every actual provider send — the first request, each tool-calling iteration, each retried attempt, each resend the WebSocket transport makes by itself, each max_tokens / truncation / empty-turn re-request, each Anthropic pause_turn continuation, each drop_block recovery and each compaction summarisation call — on all four providers. On a clean run the event count therefore equals len(result.request_cost_records).

Key Meaning
request_index 1-based and strictly monotonic across the whole run — the field to correlate on
iteration The same numbering the iteration_start hook reports, so a tool roundtrip's request and its follow-up share an iteration number
attempt The retry attempt number, so a retried request is distinguishable from a fresh one
purpose One of generation, tool_followup, max_tokens_continuation, truncation_retry, empty_turn_retry, pause_turn_continuation, prefix_binding_retry, compaction_summary (agent.hooks.REQUEST_PURPOSES)
model / provider Echoed on every payload

The event also reaches hooks on the tool-calling path, where it is observation-only. The structured-output and plain-text path has no hooks parameter on any provider, so there it reaches stream_callback alone. run_start is unchanged and still fires once per run.

A request that is sent again streams again

Text and thinking deltas are delivered as they arrive, before the toolkit knows how the request ends. When a request is then sent again, its deltas start over, and a consumer that appends every delta shows the abandoned text followed by the new text. That happens for a retried attempt and for an OpenAI or Anthropic reply cut off at max_tokens that the tool loop re-requests with a higher limit (0.5.2 on OpenAI). A pause_turn_continuation is different: the paused turn is kept and the next request continues it.

With emit_request_start=True, the request_start event before the second send is the marker: a purpose of truncation_retry or max_tokens_continuation means the text streamed since the previous request_start is not part of the answer, and an attempt higher than the previous event's means the whole earlier attempt is not. AgentResult.output never contains the abandoned text.

Marking the structured payload

When output_schema is active and the model supports native structured output, the schema is enforced on the wire, so every text_delta of that request belongs to the terminal JSON payload and is tagged {"structured_output": True}. A UI can suppress those deltas and render the rest as interim prose without guessing from a JSON head.

When the toolkit falls back to the tool-based path instead, the payload comes back as tool input and never reaches text_delta at all — so nothing is tagged there, and the absence of a tag is not evidence of prose. The attribution is never a mix: a request is either schema-constrained or it is not.

Hook injections on the complete event

Two keys of the complete metadata describe what hooks injected during the stream:

Key Description
injected_messages The hook-injected texts this turn actually delivered to the model, flat and in the order the model saw them. Empty without hooks.
undelivered_injections Texts that arrived after the last turn boundary and were therefore never shown to the model — deliver them as the next turn's user message. Empty without hooks, and on most hooked runs.

The two lists are disjoint: a text is in exactly one of them, so a UI can render the delivered ones as part of the turn and re-send the undelivered ones without counting anything twice.

async for event in session.stream("Summarise the contract"):
    if event.event_type == "complete":
        pending = event.metadata["undelivered_injections"]

if pending:
    async for event in session.stream("\n\n".join(pending)):
        ...

undelivered_injections is also on the error event and on the terminal complete of a stream that failed, so an aborted stream still hands back what the user typed. See Hooks.

A plain-text streamed turn's stored history is [user, injected…, assistant]: the opening user message (with its attachments, in the provider's own shape, when the turn carried files=), each delivered injection in order, then the final answer. Unlike session.run(), that path does not store the provider-native transcript, so tool calls and tool results from a plain-text streamed turn are not visible to later turns. A streamed turn with output_schema is stored exactly as session.run() stores it — provider-native transcript and, on the preserved-thinking rows, the retained final turn — instead of the stringified structured output it used to keep.

Follow-up turns under a "new_turn" session

complete is emitted exactly once per stream() call, and that contract is unchanged by follow-up turns. Under AgentSession(late_injection_policy="new_turn") a non-final turn ends with turn_complete instead — the same metadata complete would have carried, plus turn_index and auto_follow_up — and the next turn opens with turn_start, carrying turn_index, auto_follow_up, trigger="late_injection" and the message being sent.

Both events appear only when a follow-up actually starts, so a stream under the default "surface" policy is event-for-event unchanged. When one did run, the final complete also carries turns, turn_costs and total_cost; on a stream that never followed up those keys are absent.

Agent.run_handle is per-run and is None between turns, so a host that drives a run through the handle must re-capture it on each turn_start.

error event metadata

stream() never raises. Every exception exit — retry exhaustion, a non-retryable API error, or a hook aborting the run — emits an error event and is always followed by a terminal complete event, so consumers that short-circuit on error see the failure immediately while teardown bound to complete still runs exactly once per stream.

Key Description
input_tokens / output_tokens / cost Usage recovered from the failure — the provider's partial-stream data for a broken stream, or AgentAborted.partial for a hook abort. cost includes any spend the run's tools reported before the failure, counted once
tool_reported_cost / tool_reported_costs Present only when a tool that ran before the failure reported spend: the sum and one record per call, as on AgentResult
iterations Tool-calling iterations completed before the failure (0 when none finished)
token_source "stream_partial" when partial usage was recovered, "unknown" when the failure happened before any tokens flowed
aborted True when the run ended via HookDirective(action="abort") or RunHandle.abort(); False for any other failure
reason The hook-supplied abort rationale, or None
undelivered_injections Hook-injected texts still queued when the run failed — never shown to the model, and yours to deliver as the next turn

The trailing complete event's metadata is an AgentResult dict with error set and the same partial usage, which is what lets AgentSession.stream() bank the spend of an aborted turn.

Aborted streams used to report zeros

Before 0.3.0 the error event read only the provider's partial-stream attribute, so a hook abort — which carries its usage on AgentAborted.partial instead — reported zero tokens and zero cost, and no complete event followed at all. Both are fixed; see Hooks.

Stopping a stream early

Agent.stream() cancels the request behind it when the generator is closed. Once it is closed, the request stops within one chunk. What the request had already cost is reported, so nothing is billed silently.

A bare break does not close the generator by itself. Pair the break with contextlib.aclosing (or call aclose() on the generator yourself) and the request is cancelled as soon as the block exits:

import contextlib

async with contextlib.aclosing(agent.stream("Write a long essay")) as stream:
    async for event in stream:
        if event.event_type == "text_delta":
            print(event.content, end="", flush=True)
            if should_stop():
                break
# The request has been cancelled here.

Without aclosing, whether the request stops depends on the interpreter finalizing the abandoned generator. On CPython an inline async for event in agent.stream(...) loses its last reference at the break, and asyncio closes the generator a moment later. If anything still references it — a variable, a wrapper, a list — the request runs to its full length and is billed. The same applies to AgentSession.stream(). aclosing is therefore still the recommended pattern.

What a closed stream reports

A generator cannot yield after it has been closed, so the consumer receives no further events. When it is closed before the complete event, the toolkit waits at most STREAM_CANCEL_WAIT_SECONDS (5 s) for the run to stop. It then builds a complete-style record: an AgentResult dict with metadata["abandoned"] = True, the tokens and cost the run had already spent (including the spend of tools that had already run, read from their tool_call_result events), and error set to a "stream closed" message. The record goes to four places:

  • Agent.last_abandoned_stream: the agent's last-run cost surface. It is reset to None when a stream starts.
  • AgentSession.stream(): the record is banked into the session's total_cost, token totals and run_count, and appended to last_turn_results. The turn stores no answer in history, and its question is moved to session.interrupted_turn (error_type "StreamAbandoned") rather than left dangling (History and Checkpointing).
  • hooks: an observation-only complete event. A directive the hook returns is ignored, and the call is bounded by the same 5 s.
  • The stream's callback bridge: DEBUG_AGENT_STREAMING mirrors the record to the log.

What the record contains depends on how far the run got:

When the stream was closed The record carries
After the run finished (events were still queued) The run's own result, billed in full, with error unset
On the error event of a failed run The same partial usage the error event carried
During a run without local tools The provider's partial-stream figures: the pre-flight input estimate, the output counted from the text received so far, priced like any failed stream (token_source="stream_partial")
During a run with local tools The usage of the tool-loop iterations that had completed, summed from their iteration_complete events: since 0.5.1 every request up to the last event, including the one that asked for the round's tools. The request in flight is not included.

Nothing is reported twice. A stream consumed through its complete event reports nothing here, because the consumer already holds that record, and a session banks it from the event. When a session consumer stops at complete, the session still folds that turn into its history and totals.

A request that cannot stop within the bound is left to finish on its own. This happens, for example, when a worker thread is sleeping in a retry back-off. A WARNING is logged, and the thread's eventual cost is logged when it finishes but not reported on any result.

DPT_STREAM_CANCEL_ON_CLOSE=off restores the previous behaviour: an unbounded wait for the cancelled run, and no record. See Configuration.

Agents without local tools stream from a worker thread

A run with no local tools takes the provider's synchronous structured-output function. For a streamed run, that function now runs in a worker thread (asyncio.to_thread) instead of on the event-loop thread:

  • Every text_delta, thinking_delta and request_start event is handed back to the loop with loop.call_soon_threadsafe, in order, so the consumer receives deltas while the response is still being generated.
  • Other tasks on the same event loop keep running for the duration of the request.
  • When the stream is closed, a stop flag (a threading.Event) is set. The provider's sync stream loop checks it at the next chunk it passes to the stream callback, then stops. The SDK stream is closed, and the provider's own failure handling prices the partial output. The stop is never retried.

The sync SDK clients are safe to use from a worker thread for sequential requests, and Python's logging is thread-safe. Agent.run() is unchanged: it passes no stream callback and still calls the provider on the loop thread. This applies equally to a stream_callback you pass directly to a provider function. The provider logs a stopped request once, at INFO, as "request stopped", with no traceback and no console output.

DPT_STREAM_STRUCTURED_OFFLOOP=off restores the on-loop call. The loop is then blocked for the whole request, every delta arrives after the response has completed, and a close has nothing left to cancel. See Configuration.

Debugging grouped stream output

Set DEBUG_AGENT_STREAMING=true to mirror streamed activity to the logger and console while keeping normal event delivery unchanged.

  • Consecutive text_delta and thinking_delta chunks are grouped into readable blocks before logging.
  • Metadata-driven events such as tool_call_start, tool_call_result, and iteration_complete are flushed immediately with pretty-printed metadata.
  • This is an opt-in debugging aid for local development; it does not change the Agent.stream() API or the events yielded to callers.

Thinking/reasoning events

The thinking_delta event is emitted when a model produces reasoning tokens during streaming. Not all providers and models support thinking — the event is only emitted when applicable:

Provider Model requirement Config needed Event source
Anthropic Claude 3.5+, Claude 4+ (Sonnet, Opus) thinking={"budget_tokens": N} or thinking={"type": "adaptive"} content_block_delta with thinking_delta type
OpenAI GPT-5, o-series models thinking={"effort": ...}; the reasoning summary defaults to "auto" response.reasoning_summary_text.delta
Google Gemini models with thinking support thinking={"budget_tokens": N} or thinking={"level": ...}; the toolkit sets include_thoughts=True itself part.thought == True on streaming chunks
Fireworks DeepSeek-R1 and similar reasoning models Automatic (model-dependent) delta.reasoning_content in Chat Completions stream

On Claude Fable 5, Fable 5.1, Opus 5.5 and Sonnet 5.5 the same event can also carry the model's progress notes. Progress notes has a worked example and what it returned live.

Backward compatibility

Streaming is completely transparent to existing code:

  • Agent.run() and Agent.run_sync() return AgentResult with the same fields and values as before.
  • The result dict structure from provider functions is unchanged.
  • Token counts, cost calculations, and iteration metrics remain identical.
  • All existing tests pass without modification.

The only new surface is Agent.stream() and AgentSession.stream(), which are additive APIs.