Streaming

ask_stream() yields events as they happen — tool calls, results, and the final answer. Use it to build real-time UIs.

Python

async for event in agent.ask_stream("Who are the top 10 users by order count?"):
    match event["type"]:
        case "tool_call":
            print(f"→ Calling {event['tool_name']}({event['tool_args']})")

        case "tool_result":
            status = "✓" if event["success"] else "✗"
            print(f"  {status} {event['tool_name']}: {event['preview']}")

        case "answer":
            print(f"\n{event['text']}")

        case "done":
            print(f"\nTokens: {event['input_tokens']} in / {event['output_tokens']} out")
            print(f"Memory hits: {event['memory_hits']}")

Event reference

tool_call

Emitted before a tool is executed.

{
    "type": "tool_call",
    "tool_name": "run_mql",
    "tool_args": {"operation": "aggregate", "collection": "orders", ...}
}

tool_result

Emitted after a tool completes.

{
    "type": "tool_result",
    "tool_name": "run_mql",
    "success": True,
    "preview": "{\"rows\": [{\"_id\": \"user_1\", \"count\": 42}, ...], \"row_count\": 10}"
}

preview is the full text handed to the LLM, not a truncated excerpt. On failures it includes the wrapper the agent adds — [RETRY n/max], [MAX RETRIES EXCEEDED], [FATAL] or [ACCESS DENIED] — followed by the error.

answer

Emitted when the LLM produces a final text response.

{
    "type": "answer",
    "text": "The top 10 users by order count are..."
}

done

Emitted after answer. Contains full metadata for the turn.

{
    "type": "done",
    "iterations": 2,
    "input_tokens": 1820,
    "output_tokens": 94,
    "memory_hits": 1,
    "tool_calls_made": ["list_collections", "run_mql"],
    "retries_made": 0
}

FastAPI + SSE

The built-in server already exposes this at POST /api/v1/ask/stream, adding a session event ({session_id}) first and an error event ({code, message, request_id}) on failure — see FastAPI Server. A minimal hand-rolled equivalent:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

@app.post("/ask")
async def ask(body: dict):
    async def event_stream():
        async for event in agent.ask_stream(body["question"]):
            yield f"data: {json.dumps(event)}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

JavaScript client

const res = await fetch('/ask', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ question: 'How many orders last week?' }),
})

for await (const chunk of res.body) {
  const lines = new TextDecoder().decode(chunk).split('\n')
  for (const line of lines) {
    if (!line.startsWith('data: ')) continue
    const event = JSON.parse(line.slice(6))

    if (event.type === 'tool_call') showSpinner(event.tool_name)
    if (event.type === 'answer') showAnswer(event.text)
    if (event.type === 'done') hideSpinner()
  }
}