Protocol Client Guide

Protocol Client Guide

WebSocket Protocol v1 is engine-agnostic. Unity implements it via the C# SDK. This guide walks you through implementing it yourself — for Godot, Bevy, Unreal, a browser client, a Python test harness, or any other environment with a WebSocket library.

The full wire format spec is in Transport. This guide is the step-by-step implementation walkthrough.

What you need

A WebSocket library for your target language, JSON parsing, and UUID generation. No other dependencies.

How the protocol works

All communication is JSON text frames over a single WebSocket connection. The flow is:

  1. Connect to ws://host:port
  2. Wait for the server's hlo (hello) frame
  3. Branch on tick_mode — different patterns for host-driven vs autonomous
  4. Send req (request) frames; receive res (response) and evt (event) frames

There are three frame types:

TypeDirectionPurpose
hloServer → ClientConnection handshake. Sent once on connect. Client waits for this before sending anything.
reqClient → ServerMethod call. Each request has a UUID id for response correlation.
resServer → ClientMethod result. Contains the same id as the request. Can be ok: true or ok: false.
evtServer → ClientAsync event (tick_start, brain_decided, action_completed, etc.). Not correlated to any request.

Step 1 — Connect and handle the hello

The server sends hlo immediately on connection. Do not send any requests until you receive it. Use the hello to configure your client for the session.

python
import asyncio, json, uuid
import websockets

async def connect(host="localhost", port=8765):
    uri = f"ws://{host}:{port}"
    async with websockets.connect(uri) as ws:
        # Wait for hello — do NOT send anything before this
        raw = await ws.recv()
        hello = json.loads(raw)

        assert hello["type"] == "hlo", f"Expected hlo, got {hello['type']}"

        tick_mode   = hello["tick_mode"]      # "host_driven" or "autonomous"
        session_id  = hello["session_id"]
        capabilities = hello["capabilities"]

        print(f"Connected. Mode: {tick_mode}, session: {session_id}")

        if tick_mode == "host_driven":
            await run_host_driven(ws)
        else:
            await run_autonomous(ws)

Step 2 — Sending requests

Every request is a JSON object with type: "req", v: 1, a UUID id, a method string, and a params object. Responses carry the same id, so you can correlate them.

python
async def send_request(ws, method: str, params: dict) -> dict:
    """Send a request and wait for its response."""
    req_id = str(uuid.uuid4())
    frame = {"type": "req", "v": 1, "id": req_id, "method": method, "params": params}
    await ws.send(json.dumps(frame))

    # Read frames until we get the matching response
    # (events can arrive between request and response)
    while True:
        raw = await ws.recv()
        msg = json.loads(raw)

        if msg["type"] == "res" and msg["id"] == req_id:
            if not msg["ok"]:
                err = msg["error"]
                raise RuntimeError(f"[{err['code']}] {err['name']}: {err['message']}")
            return msg["result"]

        elif msg["type"] == "evt":
            handle_event(msg)   # dispatch to your event handler

        # hlo after initial connect is unexpected — log and ignore
        else:
            print(f"Unexpected frame type: {msg['type']}")
Serialise your sends

WebSocket sends are not concurrent-safe on most platforms. Use a single async task or mutex to serialize outgoing frames. The example above handles one request at a time; for a concurrent client, use a send queue.

Step 3 — Register agents

Before ticking, register each agent the simulation should cognize. The engine imports the brain_class by dotted Python path at registration time.

python
result = await send_request(ws, "register_agent", {
    "agent_id":     "npc_guard_01",
    "agent_name":   "Guard",
    "brain_class":  "src.plugins.builtin.idle_brain.brain.IdleBrain",
    "brain_config": {},
    "memory_class": "",
    "memory_config": {}
})
print(f"Registered: {result['agent_id']}")

For LLM cognition, swap brain_class to src.plugins.builtin.ollama.brain.OllamaLLMBrain and pass personality config in brain_config.

Step 4a — Host-driven tick loop

In host_driven mode, you control when cognition runs. Call tick once per game frame (or physics step), passing the current world observations. The server returns agent decisions synchronously.

python
async def run_host_driven(ws):
    for frame_number in range(10):
        # Build observations for each agent from your world state
        agent_observations = [
            {
                "agent_id":    "npc_guard_01",
                "observation": {
                    "position_x": 3.5,
                    "position_z": -2.0,
                    "health":     100,
                    "nearby_npcs": []
                }
            }
        ]

        result = await send_request(ws, "tick", {
            "agent_observations": agent_observations,
            "world_metadata":     {"time_of_day": "noon", "weather": "clear"}
        })

        print(f"Tick {result['tick']}")
        for decision in result["decisions"]:
            print(f"  {decision['agent_name']}: {decision['action']} → {decision['outcome_text']}")
            # Apply engine_commands back to your scene
            for cmd in decision.get("engine_commands", []):
                apply_engine_command(cmd)

Step 4b — Autonomous tick loop

In autonomous mode, the engine drives its own tick loop. Subscribe to events and react to them. Use pause/resume to control the loop — do not call tick.

python
async def run_autonomous(ws):
    # Subscribe to the events you care about
    await send_request(ws, "subscribe_events", {
        "event_types": ["tick_end", "action_completed"]
    })

    # The engine is already running — just listen
    async for raw in ws:
        msg = json.loads(raw)
        if msg["type"] == "evt":
            handle_event(msg)

def handle_event(evt):
    match evt["event_type"]:
        case "action_completed":
            agent_id = evt["agent_id"]
            action   = evt["data"]["action"]
            outcome  = evt["data"]["outcome"]
            print(f"{agent_id}: {action} → {outcome}")
            apply_engine_command_from_event(evt)

        case "tick_end":
            tick = evt["tick"]
            print(f"--- Tick {tick} complete ---")

Sending observations without ticking

Use send_observation to pre-load an agent's observation before the next tick without triggering cognition. Useful when your world state updates on a different cadence from your tick rate.

json
{
  "type":   "req",
  "v":      1,
  "id":     "<uuid>",
  "method": "send_observation",
  "params": {
    "agent_id":    "npc_guard_01",
    "observation": {"location": "gatehouse", "health": 100}
  }
}

Error handling

When a request fails, the response has "ok": false and an error object:

json
{
  "type":  "res",
  "v":     1,
  "id":    "550e8400-...",
  "ok":    false,
  "error": {
    "code":    -2,
    "name":    "AGENT_EXISTS",
    "message": "agent 'npc_guard_01' is already registered"
  }
}

Handle the codes you care about and surface the rest as fatal errors:

python
match error["code"]:
    case -2:   # AGENT_EXISTS — safe to ignore on reconnect
        pass
    case -3:   # AGENT_NOT_FOUND — agent was removed, re-register
        await register_agent(ws, agent_id)
    case -4:   # IMPORT_ERROR — brain_class path is wrong
        raise ValueError(f"Bad brain class: {error['message']}")
    case _:
        raise RuntimeError(f"Server error: {error}")

Full error code table: Transport — Error codes.

Snapshot and restore

Snapshot captures the full simulation state as a signed blob. Restore replays from that point. Use this for checkpointing long runs or testing specific states reproducibly.

python
import json, pathlib

# Save a checkpoint
snap = await send_request(ws, "snapshot", {})
pathlib.Path("checkpoint.json").write_text(json.dumps(snap))

# Restore it (must be the same engine process — HMAC key is ephemeral)
snap = json.loads(pathlib.Path("checkpoint.json").read_text())
result = await send_request(ws, "restore", {
    "data_b64":    snap["data_b64"],
    "hmac_sha256": snap["hmac_sha256"]
})
print(f"Restored to tick {result['tick']}")
HMAC is ephemeral

The signing key is generated fresh each time the server process starts. Snapshots cannot be restored across process restarts. Pass data_b64 and hmac_sha256 verbatim — do not modify them.

Complete minimal client

This is a self-contained Python client that connects, registers an agent, runs 5 ticks, and prints decisions. Copy it as a starting point.

python
import asyncio, json, uuid
import websockets

async def send(ws, method, params):
    req_id = str(uuid.uuid4())
    await ws.send(json.dumps({"type":"req","v":1,"id":req_id,"method":method,"params":params}))
    while True:
        msg = json.loads(await ws.recv())
        if msg.get("type") == "res" and msg.get("id") == req_id:
            if not msg["ok"]:
                raise RuntimeError(msg["error"])
            return msg["result"]

async def main():
    async with websockets.connect("ws://localhost:8765") as ws:
        hello = json.loads(await ws.recv())
        print(f"tick_mode: {hello['tick_mode']}")

        await send(ws, "register_agent", {
            "agent_id":    "agent_001",
            "agent_name":  "Scout",
            "brain_class": "src.plugins.builtin.idle_brain.brain.IdleBrain",
            "brain_config": {},
            "memory_class": "",
            "memory_config": {}
        })

        for _ in range(5):
            result = await send(ws, "tick", {
                "agent_observations": [
                    {"agent_id": "agent_001", "observation": {"location": "field"}}
                ],
                "world_metadata": {}
            })
            for d in result["decisions"]:
                print(f"Tick {result['tick']} | {d['agent_name']} | {d['action']} | {d['outcome_text']}")

asyncio.run(main())

Start the server first:

bash
biomata-ws --config examples/engine_owned/sim.yaml --port 8765

Implementation checklist

When implementing a client in a new language or engine:

  • Connect via WebSocket (RFC 6455), UTF-8 JSON text frames only — ignore binary frames
  • Wait for "type":"hlo" before sending any requests
  • Record tick_mode from hello and branch behavior accordingly
  • Send "v": 1 in every req frame
  • Generate a UUID v4 id per request; correlate responses by matching id
  • Handle "type":"evt" frames independently of pending requests — they can arrive at any time
  • Track evt.seq — a gap means dropped events (backpressure at 2048 events)
  • Handle "ok": false with error.code switching, not string matching
  • Implement a per-call timeout; fail pending requests on WebSocket disconnect
  • Serialise sends — concurrent WebSocket writes are unsafe on most platforms
  • Log a warning (do not disconnect) if hello v differs from 1
  • Ignore unknown top-level fields in all frame types — forward compatibility

Debugging tips

Because the protocol is plain JSON, you can drive it from the command line:

bash
# Install wscat
npm install -g wscat

# Connect and see the hello frame
wscat -c ws://localhost:8765

# Then paste a raw request:
{"type":"req","v":1,"id":"test-001","method":"health_check","params":{}}

Browser DevTools work too — open the Network tab, filter by WS, and inspect frames in real time.

Next steps