Gateway Stream Events

The gateway relays six structured progress events to WebSocket clients once they negotiate the streaming capability, so UIs can paint reasoning, tool progress, 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

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

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 == "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
`STREAM_ERROR`	`stream_error`	error frame	Streaming pipeline hits an error
`STREAM_END`	`stream_end`	terminal marker	Stream completes

reasoning_stream, tool_progress_stream, and stream_error are additive. Existing token_stream consumers keep working unchanged — gate the new events on client.supports_event(...).

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.

Common Patterns

Reasoning Inline with Answer
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()

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

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.

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

​Quick Start

​Event Vocabulary

​Response Frame States

​User Interaction Flow

​Common Patterns

​Best Practices

​Related

Gateway Handshake Protocol

Gateway Client

Frame Codec

Session Protocol

Quick Start

Event Vocabulary

Response Frame States

User Interaction Flow

Common Patterns

Best Practices

Related