Adapter tutorial: Python

The goal

By the end of this tutorial you’ll have a single runnable file, terminal_adapter.py, that connects to the mobrule bridge and prints every Invocation chat triggers to your terminal, like this:

mobrule terminal adapter — connected as streamer_login (session 42)
▶ play_sound      {"sound_id": "airhorn"}        → applied → done
▶ spawn_enemy     {"kind": "zubat", "count": 3}  → applied → done

No game, no overlay — just text. That keeps the focus on the MPP protocol itself: once you can see Invocations land in a terminal, swapping the print for “actually do something in my game” is the easy part. We’ll build the file up one piece at a time, then run it.

What you’ll need

Python 3 — stdlib only, no pip install.
The mobrule binary on PATH — the adapter shells out to it to compute the Manifest hash (see below). Install it from the platform workspace with cargo install --path cli, or add the workspace’s target/release/ directory to PATH.
A Pack to attach to — this tutorial assumes the reference Pack pack-hello-world (pack_id org.mobrule.hello-world). Any Pack works; keep its pack.toml directory handy.
A bridge token — issued when you pair the Pack; the bridge address and token land in your local mobrule runtime.

Keep terminal_adapter.py open as we go — each step adds to it, and the full file is at the bottom.

How an Adapter works

An Adapter is a process you write. It opens a TCP socket to the bridge, identifies itself with a hello frame, grants dispatch credits with pull frames, then receives invocation frames the bridge has already validated against the Pack’s Manifest. For each Invocation the Adapter does its thing — here, print a line — and reports back with applied + done (or failed). MPP is plain JSON over NDJSON: one JSON object per line, any language with a socket and a JSON encoder can speak it.

Our terminal adapter touches every step except “change a game”. Here’s the path: handshake → compute hash → prime credits → receive → print → reply → repeat.

Step 1: compute the Manifest hash

The hello frame must carry a manifest_hash the bridge agrees with. It’s a SHA-256 over the canonicalised Manifest — not sha256 of the raw pack.toml bytes — so shell out to mobrule pack print-hash rather than hashing the file yourself:

import subprocess
from pathlib import Path

PACK_DIR = Path("path/to/pack-hello-world")  # directory containing pack.toml

def manifest_hash():
    result = subprocess.run(
        ["mobrule", "pack", "print-hash", str(PACK_DIR)],
        check=True, capture_output=True, text=True,
    )
    h = result.stdout.strip()
    assert h.startswith("sha256:") and len(h) == 71
    return h

If mobrule is missing, subprocess.run raises FileNotFoundError and the adapter exits before it ever opens the bridge connection — that’s intentional. A stale or wrong hash would just be rejected by the bridge, so failing loudly up front is better. Cosmetic edits to pack.toml (whitespace, comments, key reordering) don’t change the hash; semantic edits (adding/renaming events, changing schemas, bumping version) do. See the Manifest reference.

Step 2: connect and handshake

Open the socket, wrap it in line-buffered file objects (NDJSON is line-oriented), and send hello as the very first frame:

import json, socket

BRIDGE_HOST, BRIDGE_PORT = "127.0.0.1", 7777  # bridge default port (protocol §2)
PACK_ID = "org.mobrule.hello-world"
BRIDGE_TOKEN = "<BRIDGE_TOKEN>"

sock = socket.create_connection((BRIDGE_HOST, BRIDGE_PORT))
rfile = sock.makefile("r", encoding="utf-8")

def send(frame):
    sock.sendall((json.dumps(frame) + "\n").encode("utf-8"))

send({"mpp": 2, "type": "hello",
      "pack_id": PACK_ID,
      "manifest_hash": manifest_hash(),
      "bridge_token": BRIDGE_TOKEN})

ack = json.loads(rfile.readline())
if not ack.get("ok"):
    raise SystemExit(f"handshake rejected: {ack.get('reason')}")
print(f"mobrule terminal adapter — connected as "
      f"{ack['broadcaster_login']} (session {ack['session_id']})")

The bridge replies with hello_ack. If ok is false it supplies a reason and closes the connection — the most common one is a manifest_hash mismatch, meaning the Pack on disk changed and your hash is stale. The full frame catalog is in the MPP protocol reference.

Step 3: prime the pump with `pull`

MPP v2 is pull-driven: after the handshake the bridge sends nothing until you grant it a credit. Send one pull per queue your Pack declares:

DECLARED_QUEUES = ["default"]  # keep in lockstep with pack.toml [queues.<name>]

for q in DECLARED_QUEUES:
    send({"mpp": 2, "type": "pull", "queue": q})

Each pull{queue} lets the bridge dispatch at most one Invocation on that queue. You re-pull after handling each one (Step 5). Forgetting to re-pull is the single most common Adapter bug: everything works once, then goes silent.

Step 4: receive and print

Now loop over incoming lines and print each Invocation. The invocation frame carries id, queue, event, and params — exactly what we want on screen:

for line in rfile:
    if not line.strip():
        continue
    msg = json.loads(line)
    if msg["type"] == "invocation":
        handle(msg)

def handle(msg):
    inv_id, event, params = msg["id"], msg["event"], msg["params"]
    print(f"▶ {event:<15} {json.dumps(params)}", end="", flush=True)
    # ... Step 5 replies here ...

This is the line you’d later replace with real game control. For the tutorial, the terminal is the game.

Step 5: reply with the lifecycle

Every Invocation moves pending → dispatched → applied → done | failed. The Adapter owns the last two transitions, and must re-pull to keep credits flowing. Fill in handle:

def handle(msg):
    inv_id, event, params = msg["id"], msg["event"], msg["params"]
    print(f"▶ {event:<15} {json.dumps(params)}", end="", flush=True)

    send({"mpp": 2, "type": "ack", "id": inv_id})      # received it
    send({"mpp": 2, "type": "applied", "id": inv_id, "result": None})
    send({"mpp": 2, "type": "done", "id": inv_id})
    print("  → applied → done")

    send({"mpp": 2, "type": "pull", "queue": msg["queue"]})  # replenish credit

ack — confirms you received the invocation. Informational: it doesn’t change the Invocation’s state or release the queue slot, so it’s optional — but the reference Adapter sends it, and it makes the bridge logs readable.
applied — the change took (here, instantly). Carries a result payload or result: null. With the default ready_after = "applied" this is what frees the queue slot for your next pull.
done — the effect fully landed (e.g. an animation finished).
failed — the game refused it; include a reason, and "refund": true if the viewer’s Cue should be refunded.

If the Adapter disconnects mid-Invocation, the bridge requeues it back to pending — see the protocol state machine.

The complete adapter

Everything above, assembled into one runnable terminal_adapter.py:

import json, socket, subprocess
from pathlib import Path

PACK_DIR = Path("path/to/pack-hello-world")
PACK_ID = "org.mobrule.hello-world"
BRIDGE_HOST, BRIDGE_PORT = "127.0.0.1", 7777
BRIDGE_TOKEN = "<BRIDGE_TOKEN>"
DECLARED_QUEUES = ["default"]

def manifest_hash():
    result = subprocess.run(
        ["mobrule", "pack", "print-hash", str(PACK_DIR)],
        check=True, capture_output=True, text=True,
    )
    h = result.stdout.strip()
    assert h.startswith("sha256:") and len(h) == 71
    return h

sock = socket.create_connection((BRIDGE_HOST, BRIDGE_PORT))
rfile = sock.makefile("r", encoding="utf-8")

def send(frame):
    sock.sendall((json.dumps(frame) + "\n").encode("utf-8"))

def handle(msg):
    inv_id, event, params = msg["id"], msg["event"], msg["params"]
    print(f"▶ {event:<15} {json.dumps(params)}", end="", flush=True)
    send({"mpp": 2, "type": "ack", "id": inv_id})
    send({"mpp": 2, "type": "applied", "id": inv_id, "result": None})
    send({"mpp": 2, "type": "done", "id": inv_id})
    print("  → applied → done")
    send({"mpp": 2, "type": "pull", "queue": msg["queue"]})

send({"mpp": 2, "type": "hello", "pack_id": PACK_ID,
      "manifest_hash": manifest_hash(), "bridge_token": BRIDGE_TOKEN})
ack = json.loads(rfile.readline())
if not ack.get("ok"):
    raise SystemExit(f"handshake rejected: {ack.get('reason')}")
print(f"mobrule terminal adapter — connected as "
      f"{ack['broadcaster_login']} (session {ack['session_id']})")

for q in DECLARED_QUEUES:
    send({"mpp": 2, "type": "pull", "queue": q})

for line in rfile:
    if not line.strip():
        continue
    msg = json.loads(line)
    if msg["type"] == "invocation":
        handle(msg)

Run it

python3 terminal_adapter.py

Start your mobrule runtime and Pack, then trigger an Event from chat (or the dashboard’s test button). Each redemption prints a line as it flows through the lifecycle. That’s a complete, conformant Adapter — the only thing separating it from a real one is what handle does between applied and done.

Where to go next

Two protocol features the terminal adapter skips, both ready when you need them:

Publishing Pack state. If your Pack declares [state.<key>] entries with writer = "adapter", push values with fire-and-forget state_write frames, independent of the Invocation lifecycle. The bridge validates against the key’s schema and fans it out to overlay subscribers; an accepted write is silent.
```
send({"mpp": 2, "type": "state_write", "key": "play_count", "value": 7})
```
Heartbeats. Either side may send {"mpp": 2, "type": "heartbeat"} any time after the handshake; no response required. The reference Adapter sends one every 15 seconds from a background thread to keep the connection warm.

The reference Adapter at pack-hello-world/adapter.py is the canonical, runnable version with both of these wired in, in about 150 lines of stdlib-only Python.