Gateway Client - PraisonAI

GatewayClient is a reconnecting WebSocket client that handles version negotiation, exponential backoff with jitter, and event-sequence gap detection — so your integration stays connected without you writing the socket loop.

Quick Start

Simplest Usage

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="my-agent",
        reconnect=True
    )

    await client.connect()

    async for event in client.events():
        print(f"Event: {event.type}, Data: {event.data}")

asyncio.run(main())

With Backoff Tuning

import asyncio
from praisonai.gateway import GatewayClient, BackoffConfig

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="my-agent",
        reconnect=True,
        backoff=BackoffConfig(
            initial=1.0,
            max=30.0,
            multiplier=2.0,
            jitter=0.2
        )
    )

    await client.connect()

    async for event in client.events():
        print(f"Event: {event.type}")

asyncio.run(main())

With Gap and State Callbacks

import asyncio
from praisonai.gateway import GatewayClient, BackoffConfig

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="my-agent",
        reconnect=True,
    )

    def on_gap(expected: int, received: int):
        print(f"Gap detected: expected {expected}, got {received}")
        asyncio.create_task(client.resync())

    def on_state_change(state: str):
        print(f"Connection state changed: {state}")

    client.on_gap = on_gap
    client.on_state_change = on_state_change

    await client.connect()

    async for event in client.events():
        print(f"Event: {event.type}")

asyncio.run(main())

How It Works

Stage	What happens
Connect	TCP/WS opened; `join` includes `min_version`, `max_version`, optional `agent_id`, `token`
Negotiate	Server picks `min(client_max, MAX_PROTOCOL_VERSION)` or rejects with `version_unsupported`
Stream	Each event carries a monotonic `sequence`; client tracks `_expected_sequence`
Drop	`on_state_change("reconnecting")` fires; backoff with jitter (`initial * multiplier^attempts`, clamped to `max`)
Resume	Re-`join` with `session_id` + `since=cursor`; server replays from cursor; `presence` + `health` returned
Gap	If received `sequence` ≠ expected, `on_gap(expected, received)` fires; caller can `await client.resync()`

Configuration Options

GatewayClient constructor:

Option	Type	Default	Description
`url`	`str`	—	WebSocket URL (e.g. `ws://localhost:8765`)
`agent_id`	`str`	—	Agent ID to join as
`token`	`Optional[str]`	`None`	Auth token, appended as `?token=` query param
`reconnect`	`bool`	`True`	Auto-reconnect on disconnect
`backoff`	`Optional[BackoffConfig]`	`BackoffConfig()`	Backoff configuration
`max_reconnect_attempts`	`Optional[int]`	`None` (infinite)	Max reconnect attempts before giving up

BackoffConfig fields:

Option	Type	Default	Description
`initial`	`float`	`1.0`	Initial delay in seconds
`max`	`float`	`30.0`	Maximum delay in seconds
`multiplier`	`float`	`2.0`	Backoff multiplier per attempt
`jitter`	`float`	`0.2`	Random jitter factor (0–1)

Connection states (from ConnectionState):

State	Meaning
`disconnected`	No active connection
`connecting`	Attempting initial connection
`connected`	Joined and streaming events
`reconnecting`	Waiting before next retry

Callbacks (set as attributes on the client instance):

Attribute	Signature	When it fires
`on_gap`	`Callable[[int, int], None]`	A monotonic sequence gap is detected (`expected`, `received`)
`on_state_change`	`Callable[[str], None]`	Connection state changes

Protocol Version Negotiation

The client and server negotiate a protocol version during the join handshake.

Constants in protocols.py: PROTOCOL_VERSION = 1, MIN_PROTOCOL_VERSION = 1, MAX_PROTOCOL_VERSION = 1.
Client sends min_version and max_version in the join message.
Server replies in joined with protocol_version, server_min_version, server_max_version.

version_unsupported is a permanent error. When the server returns {"type": "error", "code": "version_unsupported"}, GatewayClient.connect() raises ValueError and does not retry. Wrap your connect() call in a try/except ValueError and do not loop on this error.

Invalid min_version/max_version fields (non-integer, or min > max) produce code: "invalid_protocol_hello" and raise ConnectionError.

Gap Detection

Every event carries a monotonic sequence field; the client tracks _expected_sequence and fires on_gap when there’s a mismatch.

from praisonaiagents.gateway import GatewayEvent

# Events from the server carry sequence numbers:
# event.sequence = 1, 2, 3, ...
# If the client receives sequence=5 when expecting 4, on_gap(4, 5) fires.

on_gap is a synchronous Callable[[int, int], None]. You cannot await inside it. To call client.resync() from within the callback, schedule it as a task:

def on_gap(expected: int, received: int):
    asyncio.create_task(client.resync())

client.on_gap = on_gap

client.resync() resets the cursor to 0 and reconnects, triggering a full state reload from the server.

Resume Snapshot

When reconnecting with a stored session_id and since=cursor, the server replies with a single joined payload that restores full client state in one round trip. The joined payload includes:

Field	Description
`session_id`	Session identifier
`cursor`	Current event cursor position
`resumed`	`True` if this is a reconnection
`sequence`	Current sequence number (aligned with replay)
`protocol_version`	Negotiated protocol version
`server_min_version`	Server’s minimum supported version
`server_max_version`	Server’s maximum supported version
`presence`	List of presence dicts for all connected clients
`health`	Gateway health dict

This means a reconnecting client learns current presence and health without extra requests — one round trip restores all state.

Common Patterns

Basic Reconnecting Consumer
Gap Handler with Resync
Capped Retries
Authenticated

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="consumer",
        reconnect=True,
    )

    await client.connect()

    async for event in client.events():
        print(f"{event.type}: {event.data}")

asyncio.run(main())

import asyncio
from praisonai.gateway import GatewayClient

THRESHOLD = 5  # Force resync if more than 5 events are missed

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="consumer",
        reconnect=True,
    )

    def on_gap(expected: int, received: int):
        missed = received - expected
        if missed > THRESHOLD:
            print(f"Large gap ({missed} events), forcing resync")
            asyncio.create_task(client.resync())
        else:
            print(f"Small gap ({missed} events), continuing")

    client.on_gap = on_gap

    await client.connect()

    async for event in client.events():
        print(f"{event.type}: {event.data}")

asyncio.run(main())

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="batch-job",
        reconnect=True,
        max_reconnect_attempts=5,
    )

    try:
        await client.connect()
        async for event in client.events():
            print(f"{event.type}: {event.data}")
    except Exception as e:
        print(f"Connection failed after 5 attempts: {e}")
    finally:
        await client.disconnect()

asyncio.run(main())

import asyncio
import os
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="secure-agent",
        token=os.environ["GATEWAY_TOKEN"],
        reconnect=True,
    )

    await client.connect()

    async for event in client.events():
        print(f"{event.type}: {event.data}")

asyncio.run(main())

Network Blip: What Users See

When a network blip occurs, the client handles reconnection silently — users see no errors and events resume from the cursor automatically. The bot keeps responding because:

Events are buffered until the cursor is confirmed
The since=cursor on reconnect replays any events delivered while offline
Users see continuous responses with no error messages

Best Practices

Pick a backoff that matches your network

The defaults (initial=1.0, max=30.0, multiplier=2.0) suit residential and cellular networks. For LAN-only deployments with fast recovery, tighten initial to 0.1 and max to 5.0.

from praisonai.gateway import BackoffConfig

lan_backoff = BackoffConfig(initial=0.1, max=5.0, multiplier=2.0, jitter=0.1)

Treat version_unsupported as a permanent failure

connect() raises ValueError on version_unsupported and stops retrying. Wrap it and alert your ops team — this means the server and client are incompatible and need a coordinated upgrade.

try:
    await client.connect()
except ValueError as e:
    # Permanent — do not loop, alert immediately
    alert_ops(f"Protocol mismatch: {e}")
    raise

Wire on_gap to your replay or snapshot path

If your app already has its own snapshot mechanism, prefer it over resync(). A targeted snapshot is faster than a full cursor reset.

def on_gap(expected: int, received: int):
    if received - expected > 100:
        # Large gap — use app-level snapshot
        asyncio.create_task(app.load_snapshot())
    else:
        # Small gap — let cursor replay handle it
        pass

Pin max_reconnect_attempts in batch jobs

Long-running batch jobs should not retry forever on a dead gateway. Set max_reconnect_attempts so the job fails fast and can be requeued.

client = GatewayClient(
    url="ws://localhost:8765",
    agent_id="batch-processor",
    max_reconnect_attempts=3,
)

Gateway

WebSocket control plane for multi-agent coordination

Gateway Overview

Architecture and deployment patterns

Session Persistence

How sessions survive across processes

Push Notifications

Channel-based pub/sub and delivery guarantees

​Quick Start

​How It Works

​Configuration Options

​Protocol Version Negotiation

​Gap Detection

​Resume Snapshot

​Common Patterns

​Network Blip: What Users See

​Best Practices

​Related

Gateway

Gateway Overview

Session Persistence

Push Notifications

Quick Start

How It Works

Configuration Options

Protocol Version Negotiation

Gap Detection

Resume Snapshot

Common Patterns

Network Blip: What Users See

Best Practices

Related