Gateway Stream Events

Live progress events the gateway forwards over WebSocket after a client negotiates streaming

The gateway relays ten structured progress events to WebSocket clients once they negotiate the streaming capability, so UIs can paint reasoning, tool progress, model fallbacks, retries, live todos, tool results, and the final answer as they arrive.

from praisonaiagents import Agent

agent = Agent(name="stream-agent", instructions="Answer while streaming reasoning and tool progress.")
agent.start("Look up the weather in Paris and explain your reasoning.")

The agent emits token, reasoning, and tool events; the gateway forwards each to the WS client, which distinguishes a provisional accepted ack from the final answer.

Quick Start

1. Open a streaming client

   import asyncio
   from praisonai.gateway import GatewayClient
   
   async def main():
       client = GatewayClient(
           url="ws://localhost:8765",
           agent_id="assistant",
           capabilities=["streaming"],   # advertise streaming in hello
       )
       await client.connect()
   
   asyncio.run(main())
   
2. Print each event type

   import asyncio
   from praisonai.gateway import GatewayClient
   
   async def main():
       client = GatewayClient(
           url="ws://localhost:8765",
           agent_id="assistant",
           capabilities=["streaming"],
       )
       await client.connect()
   
       async for event in client.events():
           if event.type == "reasoning_stream":
               print("[reasoning]", event.data)
           elif event.type == "tool_progress_stream":
               print("[tool]", event.data)
           elif event.type == "model_fallback_stream":
               md = event.data["metadata"]
               print(f"[fallback] {md['from_model']} → {md['to_model']} ({md['reason_category']})")
           elif event.type == "retry_stream":
               md = event.data["metadata"]
               print(f"[retry] attempt {md['attempt']}/{md['max_attempts']} in {md['delay']}s")
           elif event.type == "todo_stream":
               print("[todos]", event.data["metadata"]["todos"])
           elif event.type == "tool_result_stream":
               print("[result]", event.data["tool_call"])
           elif event.type == "stream_error":
               print("[error]", event.data)
           elif event.type == "stream_end":
               print("[done]")
           else:
               print(event.type, event.data)
   
   asyncio.run(main())
   

---

Event Vocabulary

Each streaming event maps to an EventType in praisonaiagents.gateway.protocols and is advertised in hello_ok.features["events"] when the client negotiates streaming.

Event	Wire type	Payload	When it fires
TOKEN_STREAM	token_stream	text delta	LLM token deltas
TOOL_CALL_STREAM	tool_call_stream	tool call frame	Tool call streamed to client
REASONING_STREAM	reasoning_stream	DELTA_TEXT with is_reasoning=True	Model emits reasoning/thinking
TOOL_PROGRESS_STREAM	tool_progress_stream	tool progress frame	Tool reports in-flight progress
MODEL_FALLBACK_STREAM	model_fallback_stream	metadata: { from_model, to_model, reason_category, fallback_index }	Primary model failed; agent switched to a backup
RETRY_STREAM	retry_stream	metadata: { attempt, max_attempts, delay, reason }	Backing off before retrying a rate-limited / transient failure
TODO_STREAM	todo_stream	metadata: { todos: [...] } — full ordered list	Agent updates its plan / todo list
TOOL_RESULT_STREAM	tool_result_stream	tool_call: {...}, metadata: {...}	A tool execution completed with a result
STREAM_ERROR	stream_error	error frame	Streaming pipeline hits an error
STREAM_END	stream_end	terminal marker	Stream completes

What each new event means to a live UI

• model_fallback_stream — the primary model went down and the agent switched to a backup mid-turn. Render a “switched to backup model” banner instead of leaving the token stream frozen. metadata.from_model is the model that failed, metadata.to_model is now serving the turn, metadata.reason_category classifies the failure (rate_limit, unavailable, error), and metadata.fallback_index is the 0-based position in the fallback chain.
• retry_stream — the agent is backing off before retrying a rate-limited or transient failure. Render a countdown (“retrying in 3s… attempt 2/5”) so the pause reads as progress, not a freeze. metadata.attempt / metadata.max_attempts drive the counter, metadata.delay is seconds until the next attempt, and metadata.reason is a short human string.
• todo_stream — the agent mutated its plan. metadata.todos is the full ordered list (each item carries its status and text), so re-render the whole checklist from it.
• tool_result_stream — a tool finished with a result. tool_call carries the name, args, id, and result — render a tool result card; metadata holds any extra context.

reasoning_stream, tool_progress_stream, stream_error, and the four newer events — model_fallback_stream, retry_stream, todo_stream, and tool_result_stream — are additive. Existing token_stream consumers keep working unchanged — gate each new event on client.supports_event(...) so a legacy gateway that never advertises them degrades silently.

reasoning_stream now fires for async agent runs on OpenAI Responses-API reasoning models too, matching the sync path. Gateway consumers of async runs on these models will start seeing this signal where they previously did not.

---

Response Frame States

Every WebSocket response frame carries a status: a provisional accepted ack arrives first, then the real answer as final with a structured outcome.

status	outcome present?	Meaning	Client action
accepted	no	Provisional ack (“queued”)	Show spinner / typing indicator
final	yes (ok | error | rejected)	Real answer with completion state	Render text + reflect outcome

Never treat type: "response" alone as “done”. Check status: "final" and read outcome.status before clearing the spinner.

---

User Interaction Flow

A streaming UI paints each event as it arrives — reasoning bubble, then a tool progress chip, then the final answer.

When the primary model fails mid-turn, model_fallback_stream and retry_stream fill the visible-progress gap so the token stream never looks frozen.

---

Common Patterns

Reasoning Inline with Answer

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(url="ws://localhost:8765", agent_id="assistant", capabilities=["streaming"])
    await client.connect()

    reasoning, answer = [], []
    async for event in client.events():
        if event.type == "reasoning_stream":
            reasoning.append(event.data.get("text", ""))
        elif event.type == "token_stream":
            answer.append(event.data.get("text", ""))
        elif event.type == "stream_end":
            print("Reasoning:", "".join(reasoning))
            print("Answer:", "".join(answer))
            break

asyncio.run(main())

Tool Progress Chip

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(url="ws://localhost:8765", agent_id="assistant", capabilities=["streaming"])
    await client.connect()

    async for event in client.events():
        if event.type == "tool_progress_stream":
            show_chip(event.data)          # display in-flight tool progress
        elif event.type == "stream_end":
            clear_chip()                   # remove the chip when the stream ends
            break

asyncio.run(main())

Backup-Model Banner

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(url="ws://localhost:8765", agent_id="assistant", capabilities=["streaming"])
    await client.connect()

    async for event in client.events():
        if event.type == "model_fallback_stream":
            md = event.data["metadata"]
            show_banner(f"Switched to {md['to_model']} ({md['reason_category']})")
        elif event.type == "stream_end":
            break

asyncio.run(main())

Live Todo / Plan Progress

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(url="ws://localhost:8765", agent_id="assistant", capabilities=["streaming"])
    await client.connect()

    async for event in client.events():
        if event.type == "todo_stream":
            render_checklist(event.data["metadata"]["todos"])  # full ordered list
        elif event.type == "stream_end":
            break

asyncio.run(main())

---

Best Practices

Gate rendering on client.supports_event(...)

Wrap new-event rendering in client.supports_event("reasoning_stream") and client.supports_event("tool_progress_stream") so legacy gateways — which never advertise these events — degrade silently instead of breaking.

Treat stream_error as a soft signal

A stream_error frame does not end the run. The gateway may still deliver a final response with outcome.status="error". Surface the error but keep listening until stream_end / final.

Never assume type: response means done

Both the provisional ack and the real answer share type: "response". Branch on status: show a spinner on accepted, render only on final, and read outcome.status for the completion state.

---

Related

Gateway Handshake Protocol

Negotiate streaming and read the advertised event set

Gateway Client

Reconnecting client that streams these events

Frame Codec

Validate inbound frames at the WebSocket boundary

Session Protocol

How sessions carry these events