Documentation Index
Fetch the complete documentation index at: https://docs.powabase.ai/llms.txt
Use this file to discover all available pages before exploring further.
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/streamThe -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/streamsession_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_number']} 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_number']} 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/streamTool 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/streamError 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/streamSessions 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.
