Streaming

Streaming delivers LLM tokens to your application as they are generated, rather than waiting for the full response. This enables real-time chat interfaces, lower perceived latency, and progressive output rendering.

Quick Start

import asyncio
from fastaiagent import Agent, LLMClient
from fastaiagent.llm.stream import TextDelta

agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    llm=LLMClient(provider="openai", model="gpt-4.1"),
)

async def main():
    async for event in agent.astream("Explain quantum computing in 3 sentences."):
        if isinstance(event, TextDelta):
            print(event.text, end="", flush=True)
    print()

asyncio.run(main())

StreamEvent Types

Every streaming method yields StreamEvent objects. There are seven event types:

Event	Fields	When Emitted
TextDelta	text: str	Each token/chunk of text from the LLM
ToolCallStart	call_id: str, tool_name: str	LLM initiates a tool call
ToolCallEnd	call_id: str, tool_name: str, arguments: dict	Tool call arguments fully parsed
Usage	prompt_tokens: int, completion_tokens: int	Token counts (typically at end of response)
HandoffEvent	from_agent: str, to_agent: str, reason: str = ""	Control passes from one agent to another. Emitted by Swarm.astream() only, tagged in ahead of the target agent's first TextDelta
Paused	reason: str, context: dict, execution_id: str, node_id: str, agent_path: str \| None	The run paused — a managed approval policy or an interrupt() in a tool. Always the last event; resume with agent.aresume(execution_id, resume_value=Resume(...)). Since 1.77.0
StreamDone	(none)	End-of-stream marker

from fastaiagent.llm.stream import (
    StreamEvent, TextDelta, ToolCallStart, ToolCallEnd, Usage, HandoffEvent, Paused, StreamDone
)

A stream that pauses ends with one Paused event — the same pause arun() returns as status="paused", with the same context (for a policy pause: tool, tool_input, pending_id). There is no output yet, so output guardrails and the memory write do not run for it.

A single-agent astream() never yields a HandoffEvent — there is nothing to hand off to — so a loop that only handles the other five is still correct for LLMClient.astream() and Agent.astream(). Handle it when you stream a Swarm, or an agent switch renders as a gap in the text.

These types align with the FastAIAgent Platform's streaming protocol for seamless compatibility.

Streaming Layers

Streaming is available at three layers, from low-level to high-level:

LLMClient.astream()

Stream directly from the LLM provider. No tool execution — just raw tokens.

from fastaiagent import LLMClient
from fastaiagent.llm.message import UserMessage
from fastaiagent.llm.stream import TextDelta, Usage

llm = LLMClient(provider="openai", model="gpt-4.1")

async for event in llm.astream([UserMessage("Hello!")]):
    if isinstance(event, TextDelta):
        print(event.text, end="", flush=True)
    elif isinstance(event, Usage):
        print(f"\nTokens: {event.prompt_tokens} in, {event.completion_tokens} out")

Supported streaming providers:

Provider	Streaming	Protocol
OpenAI	Yes	SSE (data: {json} lines)
Anthropic	Yes	SSE (Anthropic event format)
Ollama	Yes	Newline-delimited JSON
Azure	Yes	SSE (OpenAI-compatible)
Custom	Yes	SSE (OpenAI-compatible)
Bedrock	No	Use acomplete() instead

stream_tool_loop()

Streaming with tool execution. The loop streams LLM tokens, detects tool calls, executes tools, and continues until the LLM produces a final text response.

from fastaiagent.agent.executor import stream_tool_loop
from fastaiagent.llm.message import SystemMessage, UserMessage
from fastaiagent.llm.stream import TextDelta, ToolCallStart, ToolCallEnd

messages = [SystemMessage("You are helpful."), UserMessage("What's the weather?")]

async for event in stream_tool_loop(llm=llm, messages=messages, tools=[weather_tool]):
    if isinstance(event, TextDelta):
        print(event.text, end="", flush=True)
    elif isinstance(event, ToolCallStart):
        print(f"\n[Calling {event.tool_name}...]")
    elif isinstance(event, ToolCallEnd):
        print(f"[{event.tool_name} done]")

Agent.astream()

Full agent streaming with guardrails, memory, and tool execution.

from fastaiagent import Agent
from fastaiagent.guardrail import no_pii
from fastaiagent.llm.stream import TextDelta, ToolCallStart

agent = Agent(
    name="assistant",
    system_prompt="You are helpful.",
    llm=llm,
    tools=[search_tool],
    guardrails=[no_pii()],
)

async for event in agent.astream("Find me a restaurant"):
    if isinstance(event, TextDelta):
        print(event.text, end="", flush=True)
    elif isinstance(event, ToolCallStart):
        print(f"\n  -> Using {event.tool_name}...")

Execution order: 1. Input guardrails run before streaming begins 2. Stream events are yielded during the tool-calling loop 3. Output guardrails run after streaming completes 4. Memory is updated at the end

Handling Tool Calls

When an agent uses tools, the stream emits events in this order for each tool-calling iteration:

TextDelta         (optional — LLM may emit text before tool calls)
ToolCallStart     (call_id, tool_name)
ToolCallEnd       (call_id, tool_name, arguments)
Usage             (token counts for this iteration)
--- tool executes, result appended to messages ---
TextDelta ...     (next iteration's tokens)

Using Python 3.10+ pattern matching:

async for event in agent.astream("Search for Python tutorials"):
    match event:
        case TextDelta(text=text):
            print(text, end="", flush=True)
        case ToolCallStart(tool_name=name):
            print(f"\n[Tool: {name}]", end="")
        case ToolCallEnd(tool_name=name):
            print(f" [done]")
        case Usage(prompt_tokens=p, completion_tokens=c):
            print(f"\n({p}+{c} tokens)", end="")
        case _:
            pass

Sync vs Async

Every streaming method has both sync and async versions:

# Async — yields events in real time
async for event in agent.astream("Hello"):
    ...

# Sync — collects all events into a single result
result = agent.stream("Hello")   # returns AgentResult
print(result.output)
print(result.trace_id, result.tokens_used, result.execution_id)

A streamed run is a traced run (1.67.0)

astream() opens an agent.<name> root span, exactly as arun() does, so the llm.* and tool.* spans of a streamed run nest under one root and the AgentResult from stream() carries trace_id, tokens_used, cost and execution_id. Before 1.67.0 astream() accepted a trace parameter and ignored it: streamed spans were emitted as orphan roots and all four fields came back empty. Pass trace=False to stream inside a workflow that already owns the root span. Swarm.stream() and Supervisor.stream() open their own swarm.* / supervisor.* roots the same way.

A streamed Swarm / Supervisor now reports tokens and cost too (1.68.0)

Swarm.stream() and Supervisor.stream() assemble their AgentResult by hand, and until 1.68.0 that hand-assembly never mentioned tokens_used, cost or cost_known — all three came back at their defaults, so a streamed multi-agent run looked free while the identical run() / arun() call reported real numbers. Both now sum Usage events for tokens and read the run-scoped accumulator for cost.

If you have been treating a streamed swarm or supervisor result as having no spend data, that assumption no longer holds — and if you were reading 0.0 as "free", it will now read as whatever the run actually cost.

# Async — yields events from LLM
async for event in llm.astream(messages):
    ...

# Sync — collects into LLMResponse
response = llm.stream(messages)  # returns LLMResponse
print(response.content)

The sync stream() safely handles being called from within an async context (e.g., Jupyter notebooks, async frameworks).

Building a Chat UI

Streaming is ideal for chat interfaces. Here is a pattern for a streaming chat loop:

import asyncio
from fastaiagent import Agent, LLMClient
from fastaiagent.agent import AgentMemory
from fastaiagent.llm.stream import TextDelta

agent = Agent(
    name="chatbot",
    system_prompt="You are a friendly assistant.",
    llm=LLMClient(provider="openai", model="gpt-4.1"),
    memory=AgentMemory(),
)

async def chat():
    while True:
        user_input = input("\nYou: ")
        if user_input.lower() in ("quit", "exit"):
            break
        print("Assistant: ", end="", flush=True)
        async for event in agent.astream(user_input):
            if isinstance(event, TextDelta):
                print(event.text, end="", flush=True)
        print()

asyncio.run(chat())

Error Handling

Streaming errors are raised as exceptions, same as non-streaming:

from fastaiagent._internal.errors import (
    LLMProviderError,        # LLM API error (auth, rate limit, etc.)
    MaxIterationsError,      # Tool loop exceeded max_iterations
    GuardrailBlockedError,   # Guardrail rejected input/output
    LLMError,                # Streaming not supported for provider
)

try:
    async for event in agent.astream("Do something"):
        if isinstance(event, TextDelta):
            print(event.text, end="")
except GuardrailBlockedError as e:
    print(f"\nBlocked by guardrail: {e}")
except MaxIterationsError:
    print("\nAgent hit iteration limit")
except LLMProviderError as e:
    print(f"\nLLM error: {e}")

Note: Input guardrails raise GuardrailBlockedError before any streaming begins. Output guardrails raise after streaming completes. In both cases, your except block handles it normally.

Platform Compatibility

The SDK's StreamEvent types (TextDelta, ToolCallStart, ToolCallEnd, Usage, StreamDone) align with the FastAIAgent Platform's streaming protocol. The same event types work in both local SDK streaming and platform streaming, making it straightforward to build clients that work with both.

Middleware, Durability, and HITL Parity (1.5.0+)

As of 1.5.0, Agent.astream() is at full feature parity with Agent.run() / arun():

• Middleware hooks fire during streaming. before_model, after_model, and wrap_tool are invoked at the same logical points they're invoked during a non-streaming run. A configured ToolBudget, TrimLongMessages, or custom AgentMiddleware works identically for both modes.
• Checkpoints are written during streaming. When the agent has a Checkpointer configured, turn-boundary and pre-tool checkpoints are persisted as the loop runs — so a process crash mid-stream can resume from the last checkpoint with chain.aresume(...).
• InterruptSignal works inside streamed tool calls. Calling interrupt(...) from within a tool that runs during astream() pauses and checkpoints the run identically to arun(), and the execution resumes via the standard aresume flow. Since 1.77.0 the stream says so with a final Paused event; before, a private exception escaped the generator at that point (a managed approval policy's pause did the same).

Before 1.5.0, all three were silently bypassed during streaming — middleware was ignored, no checkpoints were written, and interrupt() raised an unhandled exception. If you upgrade and your existing streaming code starts seeing middleware applied for the first time, that is the intended behavior.

---

Next Steps

• Agents — Agent construction, tools, guardrails, and configuration
• Tools — Deep dive into FunctionTool, RESTTool, and MCPTool
• Tracing — Trace streaming executions with OTel spans
• Guardrails — Input/output validation during streaming
• Replay — Debug streaming executions with fork-and-rerun