> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# 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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
User[š¤ User] --> Agent[š§ Agent]
Agent --> GW[š¼ Gateway]
GW -->|token_stream| WS[š WS Client]
GW -->|reasoning_stream| WS
GW -->|tool_call_stream| WS
GW -->|tool_progress_stream| WS
GW -->|model_fallback_stream| WS
GW -->|retry_stream| WS
GW -->|todo_stream| WS
GW -->|tool_result_stream| WS
GW -->|stream_error| WS
GW -->|stream_end| WS
classDef user fill:#6366F1,stroke:#7C90A0,color:#fff
classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
classDef gw fill:#189AB4,stroke:#7C90A0,color:#fff
classDef client fill:#10B981,stroke:#7C90A0,color:#fff
class User user
class Agent agent
class GW gw
class WS client
```
## Quick Start
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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())
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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`.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
Accepted[šØ accepted
provisional ack] --> Final[ā
final
outcome: ok/error/rejected]
classDef ack fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef done fill:#10B981,stroke:#7C90A0,color:#fff
class Accepted ack
class Final done
```
| `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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant User
participant UI
participant Gateway
participant Agent
User->>UI: send message
UI->>Gateway: message
Gateway-->>UI: response (status=accepted)
UI-->>User: show spinner
Agent->>Gateway: reasoning delta
Gateway-->>UI: reasoning_stream
UI-->>User: paint reasoning bubble
Agent->>Gateway: tool progress
Gateway-->>UI: tool_progress_stream
UI-->>User: show tool chip
Gateway-->>UI: stream_end
Gateway-->>UI: response (status=final, outcome)
UI-->>User: render answer, clear chip
```
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant User
participant UI
participant Gateway
participant Agent
Agent->>Gateway: token deltas
Gateway-->>UI: token_stream
UI-->>User: paint tokens
Agent->>Gateway: primary model down
Gateway-->>UI: model_fallback_stream (gpt-4o ā claude)
UI-->>User: "switched to backup model" banner
Agent->>Gateway: rate limited, backing off
Gateway-->>UI: retry_stream (attempt 2/5, delay 3s)
UI-->>User: "retrying in 3sā¦" countdown
Agent->>Gateway: token deltas resume
Gateway-->>UI: token_stream
UI-->>User: tokens resume
```
***
## Common Patterns
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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())
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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())
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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())
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
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.
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`.
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
Negotiate `streaming` and read the advertised event set
Reconnecting client that streams these events
Validate inbound frames at the WebSocket boundary
How sessions carry these events