Chapter 4 · Streaming Output and Real-Time Responses

Learning objectives: Understand the value and mechanism of streaming, master both synchronous and asynchronous streaming patterns and event types, and learn to access final metrics during a stream.

4.1 Why You Need Streaming

Non-streaming requests wait for the model to generate all content before returning — long answers can take tens of seconds, and users face a blank screen, which drives them away. Streaming lets the server push chunks as they're generated: the user sees the first character almost immediately, turning the experience from "waiting for an elevator" to "watching a live broadcast."

Under the hood, streaming uses HTTP chunked transfer: the model provider pushes generated tokens to Agno in segments, and Agno wraps them into event objects for you to iterate.

4.2 Synchronous Streaming: The Basic Typewriter Effect

Add stream=True to run() and the return value changes from RunOutput to an event iterator:

python

# stream_basic.py —— Terminal typewriter effect
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

load_dotenv()
agent = Agent(model=OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
))

# stream=True returns a RunOutputEvent iterator, not a RunOutput
stream = agent.run("Write a 200-word Python learning guide", stream=True)

for chunk in stream:
    # By default only RunContent events flow through (model text fragments)
    print(chunk.content, end="", flush=True)
print()  # Final newline

end="" + flush=True is the fixed pairing for the typewriter effect: no line break, flush the buffer immediately. Note that you cannot access .content to get the full text — each chunk contains only a fragment; you must concatenate yourself.

4.3 Event Types: Not Just Text

By default only model text flows through (RunContent events). Set stream_events=True and every event — tool calls, reasoning steps, memory updates — will stream out. This is powerful for building real-time UIs and debugging:

python

# stream_events.py —— Inspect the full event stream
from agno.agent import Agent, RunEvent
from agno.models.openai import OpenAIChat
from agno.tools.duckduckgo import DuckDuckGoTools
import os
from dotenv import load_dotenv

load_dotenv()
agent = Agent(
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    tools=[DuckDuckGoTools()],
)

stream = agent.run("What framework is Agno?", stream=True, stream_events=True)

for chunk in stream:
    if chunk.event == RunEvent.run_content:
        print(chunk.content or "", end="", flush=True)      # Text fragment
    elif chunk.event == RunEvent.tool_call_started:
        print(f"\n[Starting tool: {chunk.tool.tool_name}]")   # Tool start
    elif chunk.event == RunEvent.tool_call_completed:
        print("[Tool call completed]")
print()

Quick reference for common event types:

Category	Event	Description
Core	`run_started` / `run_content` / `run_completed` / `run_error`	Run lifecycle
Tools	`tool_call_started` / `tool_call_completed` / `tool_call_error`	Three-stage tool invocation
Reasoning	`reasoning_started` / `reasoning_step` / `reasoning_completed`	Thinking process
Memory	`memory_update_started` / `memory_update_completed`	Memory write

Common pitfall

The content of a run_content event may be None during gaps between tool calls. Always guard against null before printing, or you'll pollute the output with the literal string "None".

4.4 Asynchronous Streaming

In async frameworks (FastAPI, WebSocket servers), use arun(stream=True) paired with async for:

python

# stream_async.py —— Async streaming consumer
import asyncio
import os
from dotenv import load_dotenv
from agno.agent import Agent
from agno.models.openai import OpenAIChat

load_dotenv()
agent = Agent(model=OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
))

async def main():
    # arun + stream=True returns an async iterator
    stream = agent.arun("Summarize pros and cons of microservices in three points", stream=True)
    async for chunk in stream:
        print(chunk.content or "", end="", flush=True)

asyncio.run(main())

In FastAPI, push the chunks received in async for to the frontend via StreamingResponse or WebSocket — this is how you build a web-page typewriter. A live preview appears in Chapter 22 (with FastAPI).

4.5 Accessing Metrics After Stream Ends

You cannot get complete metrics during a stream, but the last event (or the RunCompleted event when stream_events=True) carries aggregated info:

python

# stream_metrics.py —— Streaming with final metrics
from agno.agent import Agent, RunEvent
from agno.models.openai import OpenAIChat
import os
from dotenv import load_dotenv

load_dotenv()
agent = Agent(model=OpenAIChat(
    id="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
))

full_text = ""
stream = agent.run("Explain idempotency", stream=True, stream_events=True)

for chunk in stream:
    if chunk.event == RunEvent.run_content:
        full_text += chunk.content or ""
    elif chunk.event == RunEvent.run_completed:
        # RunCompleted event carries full run results and metrics
        m = chunk.metrics
        print("\n--- Run metrics ---")
        print("Input tokens:", m.input_tokens, "Output tokens:", m.output_tokens)

print("Total text length:", len(full_text))

Pattern summary: concatenate text from run_content; read metrics from run_completed. This pattern is standard for billing and log instrumentation.

4.6 Chapter Summary

Streaming turns "wait for everything" into "watch as it generates," significantly improving long-answer UX;
With stream=True, the return type is an event iterator: sync uses for, async uses async for;
By default only text streams; set stream_events=True to see tools/reasoning/memory events too;
run_content's content may be None — always null-guard before printing;
Metrics are read from the run_completed event — text assembly and metrics collection are separate concerns.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What is the return type of agent.run(prompt, stream=True)?

2. To display "Calling search tool..." in real time during streaming, what is the correct approach?

3. Why must you null-guard chunk.content before printing during streaming?

4. Regarding token usage in streaming mode, which statement is correct?

🛠️ Hands-On Exercises

Modify the typewriter effect from 4.2 to count characters as each chunk arrives; print total character count and elapsed time at the end.
Run a search-equipped Agent with stream_events=True; write all events in order to an events.log file and observe the sequence pattern.
Build a FastAPI /chat endpoint (using StreamingResponse) that pushes async stream chunks to a curl client; verify web-level streaming works.

Now that you can "watch it generate," move on to Chapter 5: Structured Output with output_schema.

Chapter 4 · Streaming Output and Real-Time Responses ​

4.1 Why You Need Streaming ​

4.2 Synchronous Streaming: The Basic Typewriter Effect ​

4.3 Event Types: Not Just Text ​

4.4 Asynchronous Streaming ​

4.5 Accessing Metrics After Stream Ends ​

4.6 Chapter Summary ​

🧪 随堂测验

🛠️ Hands-On Exercises ​

Chapter 4 · Streaming Output and Real-Time Responses

4.1 Why You Need Streaming

4.2 Synchronous Streaming: The Basic Typewriter Effect

4.3 Event Types: Not Just Text

4.4 Asynchronous Streaming

4.5 Accessing Metrics After Stream Ends

4.6 Chapter Summary

🛠️ Hands-On Exercises