Gateway Handshake Protocol

Versioned WebSocket handshake with capability negotiation and structured errors

from praisonaiagents import Agent

agent = Agent(name="handshake-agent", instructions="Perform gateway connection handshake.")
agent.start("Establish a secure handshake with the remote gateway.")

The gateway handshake lets clients and servers agree on a protocol version, discover supported features, and recover from disconnects — all in one round trip.

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(url="ws://localhost:8765", agent_id="assistant")
    await client.connect()  # sends hello; receives hello_ok with negotiated protocol

asyncio.run(main())

The user connects via GatewayClient; hello and hello_ok negotiate protocol version, features, and session cursor in one round trip.

Quick Start

1. Minimal hello

   {
     "type": "hello",
     "agent_id": "assistant",
     "protocol_min": 1,
     "protocol_max": 1
   }

2. Opt into capabilities

   {
     "type": "hello",
     "agent_id": "assistant",
     "protocol_min": 1,
     "protocol_max": 1,
     "capabilities": ["streaming", "presence", "ack"]
   }

3. Resume a session

   {
     "type": "hello",
     "agent_id": "assistant",
     "protocol_min": 1,
     "protocol_max": 1,
     "session_id": "abc-123",
     "since": 42
   }

   After hello_ok, replayed events arrive as {"type": "replay", "event": …} before normal traffic resumes.

4. Prove device identity

   {
     "type": "hello",
     "agent_id": "assistant",
     "protocol_min": 1,
     "protocol_max": 1,
     "device": {
       "device_id": "c88e80202a4d650841e48649b4c3f553e48131b415db680476fd0d65632ff2b0",
       "public_key": "dGVzdC1wdWJsaWMta2V5",
       "signature": "<base64 signature over DeviceAuth.signing_payload(agent_id)>",
       "signed_at": 1700000000,
       "nonce": "<nonce issued by the gateway>"
     }
   }

   The device block is optional. When present, the gateway validates its shape and checks that device_id matches the public-key fingerprint before any wrapper-level signature check — see Gateway Device Identity.

How It Works

Negotiated protocol version: min(client_max, GATEWAY_PROTOCOL_VERSION) where server version is 1 and minimum accepted client version is 1.

Legacy join → joined still works for existing clients. New clients should prefer hello.

A hello carrying a device block adds two rejection paths:

• Device shape rejected — device is not an object, public_key/device_id/signature/nonce is missing or empty, signed_at is not an integer, or public_key is not padded standard base64 → configuration_error / do_not_retry.
• Spoofed device_id — device.device_id does not match the SHA-256 fingerprint of device.public_key → auth_unauthorized / reauthenticate.

A hello without a device block decodes exactly as before.

Scope requirement

Handshake frames (hello, join, leave) classify as read in the Method → Scope Registry. They must run before a client has negotiated any scope, so requiring anything higher would make the handshake unreachable. Any operator token (viewer, ops, admin) can complete the handshake; scope only starts to matter for the frames that follow.

HelloParams (client → server)

Legacy nested protocol: {min, max} is also accepted; missing values fall back to 1.

HelloResult (server → client, hello_ok)

Example success frame:

{
  "type": "hello_ok",
  "protocol": 1,
  "features": {"methods": ["message", "leave"], "events": ["message", "error", "token_stream", "tool_call_stream", "stream_end", "reasoning_stream", "tool_progress_stream", "model_fallback_stream", "retry_stream", "todo_stream", "tool_result_stream", "stream_error"]},
  "policy": {"max_payload": 1048576, "max_buffered_bytes": 8388608, "max_queued_frames": 1000, "heartbeat_ms": 30000},
  "session_id": "...",
  "resumed": false,
  "cursor": 0
}

With attachments enabled, the policy gains the attachment ceilings a /ws client self-limits against:

{
  "type": "hello_ok",
  "protocol": 1,
  "policy": {
    "max_payload": 1048576,
    "heartbeat_ms": 30000,
    "max_attachment_bytes": 10485760,
    "max_attachments": 10,
    "chunk_bytes": 262144,
    "allowed_attachment_types": ["image/", "application/pdf"]
  },
  "session_id": "...",
  "resumed": false,
  "cursor": 0
}

When attachments are disabled, none of those keys are added and the policy is byte-for-byte the shape above — see Gateway Attachments for the full story.

Base methods: message, leave only (abort is not implemented). Base events: message, error.

HelloError (server → client, hello_error)

Example wire frame (rate-limited):

{
  "type": "hello_error",
  "code": "rate_limited",
  "message": "Too many connection attempts",
  "next_step": "wait_then_retry",
  "retry_after_seconds": 30,
  "next": "wait_then_retry"
}

ConnectErrorCode

ConnectRecoveryStep

Machine-readable recovery hint. Clients branch on (code, next_step) instead of parsing message.

Capability Matrix

Events are advertised only if the client requested the matching capability. The ten streaming events — including their payloads and the accepted vs final response frame — are detailed in Gateway Stream Events.

Server-side state after handshake

After hello_ok is sent, the gateway records the negotiated protocol version and the client’s advertised capabilities on the session itself. Both are read-only and survive resume.

session = gateway.get_session(session_id)
session.protocol_version    # → 1   (negotiated min(client_max, GATEWAY_PROTOCOL_VERSION))
session.capabilities        # → ["streaming", "presence", "ack"]   (defensive copy)

Both properties are populated on the hello path and the legacy join path, so server code can branch on session.capabilities without checking which handshake the client used.

Client-side wiring (praisonai-bot)

The bundled praisonai-bot[gateway] client speaks the modern handshake for you — construct it with the capabilities you handle, connect(), then read the negotiated manifest off the instance.

import asyncio
from praisonai_bot.gateway import GatewayClient, PayloadTooLarge

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="assistant",
        capabilities=["streaming", "ack"],   # what to advertise in `hello`
    )
    await client.connect()

    # Read the negotiated manifest — empty on legacy gateways.
    print("protocol   :", client.features)               # {"methods": [...], "events": [...]}
    print("policy     :", client.policy)                 # {"max_payload": ..., "heartbeat_ms": ...}
    print("heartbeat  :", client.heartbeat_ms, "ms")     # convenience getter

    # Gate optional behaviour on the negotiated event set instead of probing.
    if client.supports_event("token_stream"):
        # enable streaming UI
        pass
    if client.supports_event("reasoning_stream"):
        # render live reasoning bubble
        pass
    if client.supports_event("tool_progress_stream"):
        # render tool progress indicator
        pass

    # Self-guarded send — oversized frame fails locally, never hits the wire.
    try:
        await client.send({"type": "message", "content": "hello"})
    except PayloadTooLarge as e:
        print("dropped:", e)

asyncio.run(main())

The client always sends hello first — capabilities= controls what it advertises, not whether the modern handshake is used.

Client construction

Negotiated accessors

connect() populates these from the server’s hello_ok. Against a legacy gateway they stay at permissive defaults so existing code keeps working.

supports_event(event)

Gate optional behaviour on the negotiated event set instead of try: … except: probing.

from praisonaiagents.gateway import EventType

client.supports_event("token_stream")            # str lookup
client.supports_event("reasoning_stream")        # new streaming event
client.supports_event("tool_progress_stream")    # new streaming event
client.supports_event(EventType.MESSAGE)         # EventType — uses .value

• Accepts a str (e.g. "token_stream") or an EventType member; the enum’s .value is used for lookup.
• Returns True for legacy gateways (empty features["events"]) so code that assumed everything worked keeps working — this is the intentional permissive default.
• Otherwise returns True iff event in client.features["events"].

Handshake flow — client branch on the server’s first reply

The wire is documented server-side above; this is what the client does with each reply.

PayloadTooLarge

from praisonai_bot.gateway import PayloadTooLarge

Raised inside client.send(...), before the WebSocket write, when len(payload.encode("utf-8")) > client.policy["max_payload"].

• The frame never leaves the client — nothing reaches the server and no round-trip is spent.
• Size is measured in UTF-8 bytes, not len(str), matching how the server counts. Size payloads by encoded length, not string length.
• Catch it and shrink/split the payload or accept the loss — retrying the same frame fails identically.
• A max_payload of 0 is honoured as “reject everything”; None/missing means unbounded. On a legacy gateway (policy == {}) this exception is unreachable and the previous server-side reject still applies.

Policy Limits

Clients should self-configure from the policy object in hello_ok. See Gateway Flow Control for tuning max_buffered_bytes and max_queued_frames.

Persisted Session State

After hello_ok, the gateway records both the negotiated protocol version and the client-advertised capabilities on the GatewaySession. These values survive disconnect and resume.

import asyncio
from praisonai.gateway import GatewayClient

async def main():
    client = GatewayClient(
        url="ws://localhost:8765",
        agent_id="assistant",
        capabilities=["streaming", "ack"],
    )
    await client.connect()
    # After hello_ok, the server-side session exposes:
    # session.protocol_version  → "1"  (negotiated version string)
    # session.capabilities      → ["streaming", "ack"]  (client-advertised)

asyncio.run(main())

Both properties are written into session.to_dict() and restored by from_dict(), so they survive server restarts and session persistence backends.

Server-side hooks and custom routing code can read these off any live session:

def on_message(session, payload):
    if "streaming" in session.capabilities:
        # send token_stream events
        ...
    else:
        # send complete message at once
        ...

Best Practices

if frame.get("type") == "hello_error":
    step = frame.get("next_step")
    if step == "wait_then_retry":
        await asyncio.sleep(frame.get("retry_after_seconds", 1))
        await reconnect()
    elif step == "reauthenticate":
        await refresh_credentials()
        await reconnect()
    elif step in ("do_not_retry", "upgrade_client", "downgrade_client", "repair"):
        surface_to_user(frame["message"])
    else:
        await reconnect()  # default: try again

Related