> ## Documentation Index
> Fetch the complete documentation index at: https://docs.celesto.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming Responses

> Stream Agentor agent progress with stream_chat to show tool calls live, or drop to AgentLoop for token-by-token output.

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

<Note>
  `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](#token-level-streaming) below.
</Note>

## Quick Start

Enable streaming with the `stream_chat` method:

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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:

<ResponseField name="type" type="str">
  Always `"run_item_stream_event"`.
</ResponseField>

<ResponseField name="message" type="str | None">
  Text: a tool's return value on a `tool_output` event, the agent's answer on the final one.
</ResponseField>

<ResponseField name="tool_action" type="ToolAction | None">
  `name` and `type` for a tool call (`tool_called`) or its result (`tool_output`).
</ResponseField>

<ResponseField name="chunk" type="None">
  Reserved. `stream_chat()` never sets it — see [token-level streaming](#token-level-streaming).
</ResponseField>

<ResponseField name="reasoning" type="None">
  Reserved. Not populated today.
</ResponseField>

## JSON Serialization

Get events as JSON strings for easy transmission:

```python theme={null} theme={null}
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:

```json theme={null} theme={null}
{
  "type": "run_item_stream_event",
  "message": "Code executes fast",
  "chunk": null,
  "tool_action": null,
  "reasoning": null
}
```

## HTTP Streaming

Serve streaming responses over HTTP:

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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`:

```python theme={null} theme={null}
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](/agentor/tracing) and [durable runs](/agentor/durable-runs).

<Warning>
  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.
</Warning>

### Filter to just the answer

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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`:

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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:

```python theme={null} theme={null}
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

<Steps>
  ### Use Async/Await

  Streaming requires async execution:

  ```python theme={null} theme={null}
  # Good
  async def main():
      async for event in agent.stream_chat("Hello"):
          print(event)

  # Won't work
  for event in agent.stream_chat("Hello"):  # Error!
      print(event)
  ```

  ### Flush Output

  Flush stdout so text appears as it arrives:

  ```python theme={null} theme={null}
  async for event in loop.astream("Hello", stream_text=True):
      if event.type == "text_delta":
          print(event.text, end="", flush=True)  # flush=True is important
  ```

  ### Handle Partial Content

  Token deltas do not align with word boundaries — `'Wh'`, `'ispers'`, `' in'`. Buffer until you see a separator:

  ```python theme={null} theme={null}
  buffer = ""
  async for event in loop.astream("Hello", stream_text=True):
      if event.type == "text_delta":
          buffer += event.text
          # Process complete words only
          if " " in buffer:
              words = buffer.split(" ")
              for word in words[:-1]:
                  process_word(word)
              buffer = words[-1]
  ```

  ### Set Timeouts

  Prevent hanging streams:

  ```python theme={null} theme={null}
  import asyncio

  async def stream_with_timeout():
      try:
          async with asyncio.timeout(30):  # 30 second timeout
              async for event in agent.stream_chat("Long task"):
                  print(event)
      except asyncio.TimeoutError:
          print("Stream timeout")
  ```

  ### Monitor Connection Health

  For HTTP streaming:

  ```python theme={null} theme={null}
  from fastapi.responses import StreamingResponse

  @app.post("/chat")
  async def chat(message: str):
      async def event_stream():
          try:
              async for chunk in agent.stream_chat(message):
                  yield f"data: {chunk}\n\n"
          except Exception as e:
              yield f"event: error\ndata: {str(e)}\n\n"
      
      return StreamingResponse(
          event_stream(),
          media_type="text/event-stream"
      )
  ```
</Steps>

## 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](/agentor/deploy) behind your own infrastructure
* Enable [observability](/agentor/guides/observability) to monitor stream performance
* Learn about [agent communication](/agentor/guides/agent-communication) with streaming A2A