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.

TypeScript SDK users: the equivalent typed enum is GatewayEventType in praisonai-ts/src/gateway, kept in lockstep with Python.

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.

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.

Typed outcome.error on failed turns

A failed turn adds a machine-readable error object to outcome.

{
  "type": "response",
  "status": "final",
  "content": "Error: ...",
  "outcome": {
    "status": "error",
    "error": {
      "code": "rate_limit",
      "retryable": true,
      "retry_after_seconds": 30,
      "message": "Rate limited by the model provider. Please retry shortly."
    }
  }
}

See Gateway Turn Error Codes for the full code table.

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

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())

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())

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())

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

Related