Streaming Responses

Agent runs stream results as Server-Sent Events (SSE). Each event is a JSON object prefixed with data: on its own line. This guide walks through parsing every event type, rendering tool calls in a UI, handling errors gracefully, and chaining multi-turn conversations with session IDs.

Prerequisites:

An agent created (see Build an Agent guide)

Basic streaming request

Send a message to an agent and open an SSE stream. Read lines, filter for data: prefixes, and parse each event as JSON.Endpoint: POST /api/agents/{id}/run/stream

The -N flag in curl disables output buffering so events appear in real time.

import requests
import json

response = requests.post(
    f"{BASE_URL}/api/agents/{agent_id}/run/stream",
    headers=headers,
    json={"message": "Hello, what can you help me with?"},
    stream=True,
)

for line in response.iter_lines():
    if not line:
        continue
    text = line.decode("utf-8")
    if text.startswith("data: "):
        event = json.loads(text[6:])
        print(event["event"], "->", event.get("content", "")[:80])

Parse all event types

Handle every event the stream can emit: start, chunk, tool_call, tool_result, step_started, step_completed, complete, and error.Endpoint: POST /api/agents/{id}/run/stream

session_id = None
full_response = []

for line in response.iter_lines():
    if not line:
        continue
    text = line.decode("utf-8")
    if not text.startswith("data: "):
        continue

    event = json.loads(text[6:])
    etype = event["event"]

    if etype == "start":
        session_id = event["session_id"]
        print(f"Run started — session: {session_id}, run: {event['run_id']}")

    elif etype == "step_started":
        print(f"Step {event['step']} started")

    elif etype == "chunk":
        full_response.append(event["content"])
        print(event["content"], end="", flush=True)

    elif etype == "tool_call":
        print(f"\n[Tool call: {event['tool_name']}({json.dumps(event.get('arguments', {}))})]")

    elif etype == "tool_result":
        print(f"[Tool result: {str(event.get('result', ''))[:100]}]")

    elif etype == "step_completed":
        print(f"\nStep {event['step']} completed")

    elif etype == "complete":
        print(f"\n\nRun complete — total tokens: {event.get('usage', {})}")

    elif etype == "error":
        print(f"\nError: {event.get('message', 'Unknown error')}")

Handle tool calls in the UI

Display tool calls and results as collapsible cards in a chat UI. Use the tool_call event to show a loading indicator, then update with the tool_result.Endpoint: POST /api/agents/{id}/run/stream

Tool calls always appear in pairs: a tool_call event followed by a tool_result event. Multiple tool calls can occur in a single step if the agent decides to call several tools.

# Example: collect tool call/result pairs for UI rendering
tool_calls = []

for line in response.iter_lines():
    if not line:
        continue
    text = line.decode("utf-8")
    if not text.startswith("data: "):
        continue

    event = json.loads(text[6:])

    if event["event"] == "tool_call":
        tool_calls.append({
            "name": event["tool_name"],
            "arguments": event.get("arguments", {}),
            "status": "running",
            "result": None,
        })
        # UI: render a loading card for this tool call

    elif event["event"] == "tool_result":
        if tool_calls:
            tool_calls[-1]["status"] = "complete"
            tool_calls[-1]["result"] = event.get("result")
        # UI: update the card with the result

    elif event["event"] == "chunk":
        # UI: append to the assistant message bubble
        pass

Error handling

Handle error events from the stream and connection-level failures. Always wrap your stream reader in a try/catch.Endpoint: POST /api/agents/{id}/run/stream

Error events have an optional code field (e.g., “rate_limited”, “context_length_exceeded”). Always check for HTTP-level errors before reading the stream.

try:
    response = requests.post(
        f"{BASE_URL}/api/agents/{agent_id}/run/stream",
        headers=headers,
        json={"message": "What is the refund policy?"},
        stream=True,
        timeout=60,
    )
    response.raise_for_status()

    for line in response.iter_lines():
        if not line:
            continue
        text = line.decode("utf-8")
        if not text.startswith("data: "):
            continue

        event = json.loads(text[6:])

        if event["event"] == "error":
            print(f"Agent error: {event.get('message', 'Unknown')}")
            print(f"Error code: {event.get('code', 'N/A')}")
            break

        if event["event"] == "chunk":
            print(event["content"], end="")

except requests.exceptions.ConnectionError:
    print("Connection lost — retry with exponential backoff")
except requests.exceptions.Timeout:
    print("Request timed out")
except json.JSONDecodeError as e:
    print(f"Malformed SSE event: {e}")

Multi-turn with session_id

The start event includes a session_id. Pass it in subsequent requests to continue the conversation with full message history.Endpoint: POST /api/agents/{id}/run/stream

Sessions persist the full conversation history. You can also list session messages via GET /api/sessions/{session_id}/messages.

import json

def chat(agent_id: str, message: str, session_id: str | None = None) -> str:
    """Send a message and return (response_text, session_id)."""
    body = {"message": message}
    if session_id:
        body["session_id"] = session_id

    response = requests.post(
        f"{BASE_URL}/api/agents/{agent_id}/run/stream",
        headers=headers,
        json=body,
        stream=True,
    )

    result_text = []
    sid = session_id
    for line in response.iter_lines():
        if not line:
            continue
        text = line.decode("utf-8")
        if not text.startswith("data: "):
            continue
        event = json.loads(text[6:])
        if event["event"] == "start":
            sid = event["session_id"]
        elif event["event"] == "chunk":
            result_text.append(event["content"])
    return "".join(result_text), sid

# First message — no session yet
answer1, session_id = chat(agent_id, "What products do you offer?")
print(answer1)

# Follow-up — same session, agent remembers context
answer2, session_id = chat(agent_id, "Which one is best for small teams?", session_id)
print(answer2)

What’s Next

Streaming Patterns

Understand the SSE protocol and event lifecycle in depth.

Build an Agent

Create an agent from scratch with tools and knowledge bases.

Agents API Reference

Full endpoint documentation for agent runs.

​What’s Next

Streaming Patterns

Build an Agent

Agents API Reference

What’s Next