Streaming Responses

A long agent run looks broken until it finishes. Streaming lets you show progress instead: which tool the agent just called, what it came back with, and the answer as soon as it exists.

Why Stream?

Immediate feedback: users see each step rather than a spinner
Tool visibility: show what the agent is doing and why it is taking a while
Cancellation: stop early once you have what you need

stream_chat() streams steps, not tokens: whole messages, tool calls, and tool results as each one happens. For token-by-token text, see token-level streaming below.

Quick Start

Enable streaming with the stream_chat method:

import asyncio
from agentor import Agentor

agent = Agentor(
    name="Assistant",
    model="gpt-5-mini",
    instructions="You are a helpful assistant."
)

async def main():
    async for chunk in agent.stream_chat("Explain quantum computing"):
        print(chunk, flush=True)

asyncio.run(main())

Stream Event Types

Agentor emits structured events during streaming:

import asyncio
from agentor import Agentor

agent = Agentor(
    name="Assistant",
    model="gpt-5-mini",
    tools=["get_weather"]
)

async def main():
    async for event in agent.stream_chat(
        "What's the weather in London?",
        serialize=False  # Get AgentOutput objects instead of JSON strings
    ):
        if event.tool_action:
            # type is "tool_called" when the agent asks,
            # "tool_output" when the tool answers
            print(f"[{event.tool_action.type}] {event.tool_action.name}")

        if event.message:
            print(event.message)

asyncio.run(main())

Output for the run above:

[tool_called] get_weather
[tool_output] get_weather
The weather in London is sunny and 22C.
The weather in London is sunny with a temperature of 22°C.

Each AgentOutput carries:

str

Always "run_item_stream_event".

str | None

Text: a tool’s return value on a tool_output event, the agent’s answer on the final one.

ToolAction | None

name and type for a tool call (tool_called) or its result (tool_output).

None

Reserved. stream_chat() never sets it — see token-level streaming.

None

Reserved. Not populated today.

JSON Serialization

Get events as JSON strings for easy transmission:

async def main():
    async for json_event in agent.stream_chat(
        "Write a haiku about Python",
        serialize=True  # Returns JSON strings (default)
    ):
        print(json_event)  # Each event is a JSON string

asyncio.run(main())

Example JSON event:

{
  "type": "run_item_stream_event",
  "message": "Code executes fast",
  "chunk": null,
  "tool_action": null,
  "reasoning": null
}

HTTP Streaming

Serve streaming responses over HTTP:

from agentor import Agentor
import uvicorn
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()
agent = Agentor(name="Assistant", model="gpt-5-mini")

@app.post("/chat")
async def chat(message: str):
    async def event_stream():
        async for chunk in agent.stream_chat(message, serialize=True):
            yield f"data: {chunk}\n\n"
    
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        }
    )

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Built-in Server Streaming

Agentor’s built-in server supports streaming out of the box:

from agentor import Agentor

agent = Agentor(
    name="Streaming Agent",
    model="gpt-5-mini"
)

# Serve with automatic streaming support
agent.serve(port=8000)

Client request with streaming:

import requests

url = "http://localhost:8000/chat"
response = requests.post(
    url,
    json={"input": "Tell me a story", "stream": True},
    stream=True
)

for line in response.iter_lines(decode_unicode=True):
    if line:
        print(line, flush=True)

A2A Protocol Streaming

Stream responses using the A2A protocol:

import requests
import json

url = "http://localhost:8000/"
payload = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "message/stream",
    "params": {
        "message": {
            "parts": [
                {"kind": "text", "text": "Explain machine learning"}
            ]
        }
    }
}

response = requests.post(url, json=payload, stream=True)

for line in response.iter_lines(decode_unicode=True):
    if line.startswith("data: "):
        event = json.loads(line[6:])
        result = event.get("result", {})
        
        # Task creation
        if "id" in result and "status" in result:
            print(f"Task created: {result['id']}")
        
        # Artifact updates (streaming content)
        elif "artifact" in result:
            artifact = result["artifact"]
            if artifact.get("parts"):
                text = artifact["parts"][0].get("text", "")
                print(text, end="", flush=True)
        
        # Status updates
        elif "status" in result:
            status = result["status"]
            if result.get("final"):
                print(f"\n\nCompleted with status: {status['state']}")

Advanced Streaming Patterns

Token-level streaming

stream_chat() gives you steps. For text as the model produces it, use AgentLoop — the engine underneath Agentor — and ask for stream_text=True:

import asyncio

from agentor import AgentLoop, function_tool


@function_tool
def get_weather(city: str) -> str:
    """Get the current weather for a city.

    Args:
        city: The city to look up.
    """
    return f"The weather in {city} is sunny and 22C."


loop = AgentLoop(
    name="Assistant",
    model="gpt-4o-mini",
    instructions="You are helpful.",
    tools=[get_weather],
)

async def main():
    async for event in loop.astream("What is the weather in Tokyo?", stream_text=True):
        if event.type == "text_delta":
            print(event.text, end="", flush=True)
        elif event.type == "tool_call":
            print(f"\n[calling {event.name}({event.args})]")
        elif event.type == "run_end":
            print(f"\n[{event.status}] {event.usage.total_tokens} tokens")

asyncio.run(main())

[calling get_weather({'city': 'Tokyo'})]
The weather in Tokyo is sunny with a temperature of 22°C.
[completed] 188 tokens

astream emits the engine’s raw events: run_start, generation, text_delta, tool_call, tool_result, message, and run_end. The same events power tracing and durable runs.

Drain the generator. If you break out of an astream loop early, the run never finishes — its trace is not exported, and its saved log has no terminal event.

Filter to just the answer

async def stream_only_messages():
    async for event in agent.stream_chat("Hello", serialize=False):
        if event.message and not event.tool_action:
            print(event.message)

asyncio.run(stream_only_messages())

Progress Tracking

Count the steps as they go by:

import asyncio

async def stream_with_progress():
    tool_calls = 0
    last_message = None

    async for event in agent.stream_chat(
        "Research quantum computing and summarize",
        serialize=False
    ):
        if event.tool_action and event.tool_action.type == "tool_called":
            tool_calls += 1
            print(f"Tool call #{tool_calls}: {event.tool_action.name}")

        if event.message:
            last_message = event.message

    print(f"\n{last_message}")
    print(f"Complete! Tool calls: {tool_calls}")

asyncio.run(stream_with_progress())

Buffered Streaming

Batch tokens before writing them, to cut down on network round trips. This needs text_delta events, so it uses AgentLoop:

import asyncio

from agentor import AgentLoop

loop = AgentLoop(name="Assistant", model="gpt-4o-mini")

async def buffered_stream(buffer_size=5):
    buffer = []

    async for event in loop.astream("Write a story", stream_text=True):
        if event.type != "text_delta":
            continue

        buffer.append(event.text)
        if len(buffer) == buffer_size:
            print("".join(buffer), end="", flush=True)
            buffer.clear()

    if buffer:
        print("".join(buffer), flush=True)

asyncio.run(buffered_stream())

Multi-Agent Streaming

Stream from multiple agents concurrently:

import asyncio

agent1 = Agentor(name="Agent 1", model="gpt-5-mini")
agent2 = Agentor(name="Agent 2", model="gpt-5-mini")

async def stream_multiple():
    async def stream_agent(agent, prompt, prefix):
        async for event in agent.stream_chat(prompt, serialize=False):
            if event.message:
                print(f"[{prefix}] {event.message}")
    
    await asyncio.gather(
        stream_agent(agent1, "What is Python?", "A1"),
        stream_agent(agent2, "What is JavaScript?", "A2")
    )

asyncio.run(stream_multiple())

WebSocket Streaming

For bidirectional streaming, use WebSockets:

from fastapi import FastAPI, WebSocket
from agentor import Agentor
import uvicorn
import json

app = FastAPI()
agent = Agentor(name="Assistant", model="gpt-5-mini")

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    
    try:
        while True:
            # Receive message
            data = await websocket.receive_text()
            message = json.loads(data)
            
            # Stream response
            async for event in agent.stream_chat(
                message["text"],
                serialize=True
            ):
                await websocket.send_text(event)
    
    except Exception as e:
        print(f"WebSocket error: {e}")
    finally:
        await websocket.close()

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

WebSocket client:

import asyncio
import websockets
import json

async def chat():
    uri = "ws://localhost:8000/ws"
    async with websockets.connect(uri) as websocket:
        # Send message
        await websocket.send(json.dumps({"text": "Hello!"}))
        
        # Receive stream
        async for message in websocket:
            event = json.loads(message)
            if event.get("message"):
                print(event["message"])

asyncio.run(chat())

Error Handling

Handle streaming errors gracefully:

import asyncio

async def safe_stream():
    try:
        async for event in agent.stream_chat("Your query", serialize=False):
            if event.message:
                print(event.message)
    
    except asyncio.TimeoutError:
        print("Streaming timeout")
    except Exception as e:
        print(f"Streaming error: {e}")
    finally:
        print("Stream closed")

asyncio.run(safe_stream())

Best Practices

Performance Tips

Use serialize=True (default) when sending over network
Use serialize=False for local processing to avoid JSON overhead
Buffer small chunks for better network efficiency
Set appropriate timeouts based on expected response time
Close streams properly to free resources

Next Steps

Serve streaming agents behind your own infrastructure
Enable observability to monitor stream performance
Learn about agent communication with streaming A2A

Start Here

Build Agents

Concepts

Tool Use & MCP

Guides

SuperAuth

Why Stream?

Quick Start

Stream Event Types

JSON Serialization

HTTP Streaming

Built-in Server Streaming

A2A Protocol Streaming

Advanced Streaming Patterns

Token-level streaming

Filter to just the answer

Progress Tracking

Buffered Streaming

Multi-Agent Streaming

WebSocket Streaming

Error Handling

Best Practices

Performance Tips

Next Steps

​Why Stream?

​Quick Start

​Stream Event Types

​JSON Serialization

​HTTP Streaming

​Built-in Server Streaming

​A2A Protocol Streaming

​Advanced Streaming Patterns

​Token-level streaming

​Filter to just the answer

​Progress Tracking

​Buffered Streaming

​Multi-Agent Streaming

​WebSocket Streaming

​Error Handling

​Best Practices

​Performance Tips

​Next Steps

Why Stream?

Quick Start

Stream Event Types

JSON Serialization

HTTP Streaming

Built-in Server Streaming

A2A Protocol Streaming

Advanced Streaming Patterns

Token-level streaming

Filter to just the answer

Progress Tracking

Buffered Streaming

Multi-Agent Streaming

WebSocket Streaming

Error Handling

Best Practices

Performance Tips

Next Steps