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 0)
▶ play_sound      {"sound_id": "airhorn"}        → applied → done
▶ spawn_enemy     {"kind": "zubat", "count": 3}  → applied → done

(The session number is always 0 today; it’s a placeholder field for a future numbered-session scheme, not something you need to track.)

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 tv.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 = "tv.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. The reference Adapter sends it anyway, and it makes the bridge logs readable.
applied means 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 means the effect fully landed (e.g. an animation finished).
failed means 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 = "tv.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 200 lines of stdlib-only Python.