Writing a plugin

This guide builds the smallest useful plugin: it answers the handshake, routes “open youtube” to a single fully resolved step, and invokes it. The tested references are plugins/fake_py/ (Python, standard library only) and plugins/fake_rs/ (Rust). Read the protocol summary first if the terms are new.

Python

Copy the framing helpers from plugins/fake_py/plugin.py verbatim; the skeleton below shows the handshake and the three methods the vertical slice needs.

#!/usr/bin/env python3
"""hello_py — minimal stdio plugin."""
import json, sys

PROTOCOL, ABI = "1.0", 1
PROVIDES = ["action.open_url@1", "context.live@1", "fs.tmp@1"]
REQUIRES = ["fs.tmp@1"]

def read_message(stream):
    header = b""
    while not header.endswith(b"\r\n\r\n"):
        chunk = stream.read(1)
        if not chunk:
            return None
        header += chunk
    length = int(header.split(b"content-length:")[1].split(b"\r\n")[0])
    body = stream.read(length)
    return json.loads(body.decode("utf-8")) if body else None

def write_message(stream, obj):
    body = json.dumps(obj, separators=(",", ":")).encode("utf-8")
    stream.write(b"Content-Length: %d\r\n\r\n" % len(body) + body)
    stream.flush()

def handle(method, params):
    if method == "protocol.hello":
        return {"protocol": PROTOCOL, "abi": ABI,
                "plugin": {"name": "hello_py", "version": "0.1.0", "kind": "bundle"},
                "transport": "stdio",
                "provides": PROVIDES, "requires": REQUIRES, "permissions": ["network"]}
    if method == "plugin.describe":
        return {"methods": ["protocol.hello", "plugin.describe", "plugin.health",
                            "router.plan", "action.capabilities", "action.invoke",
                            "context.snapshot"], "streams": []}
    if method == "plugin.health":
        return {"status": "ok", "detail": "hello_py ready"}
    if method == "router.plan":
        # "open youtube" -> one fully-resolved step
        utt = str(params.get("utterance", "")).strip().lower()
        if utt in ("open youtube", "go to youtube"):
            return {"steps": [{"op": "ensure_url",
                               "args": {"url": "https://www.youtube.com"},
                               "provenance": "user"}]}
        return {"steps": []}
    if method == "action.capabilities":
        return {"ops": [{"op": "ensure_url", "side_effect": "open_url",
                         "needs_confirm": False}]}
    if method == "action.invoke":
        args = params.get("args") or {}
        return {"ok": True, "op": params.get("op"),
                "detail": f"opened {args.get('url', '')}"}
    if method == "context.snapshot":
        return {"focused": None, "windows": [], "clipboard": "", "timestamp": 0}
    raise ValueError(f"method not found: {method}")

def main():
    stdin, stdout = sys.stdin.buffer, sys.stdout.buffer
    while True:
        msg = read_message(stdin)
        if msg is None:
            return 0
        if not isinstance(msg, dict) or msg.get("id") is None:
            continue
        try:
            result = handle(str(msg.get("method") or ""), msg.get("params") or {})
            write_message(stdout, {"jsonrpc": "2.0", "id": msg["id"], "result": result})
        except Exception as exc:                       # noqa: BLE001
            write_message(stdout, {"jsonrpc": "2.0", "id": msg["id"],
                                   "error": {"code": -32000, "message": str(exc)}})

if __name__ == "__main__":
    raise SystemExit(main())

You do not need to check provenance yourself for policy: the runner stamps and enforces it. A screen-provenance request that authors a concrete argument is rejected before your plugin is called.

Wire it into a runner config

[[plugin]]
id = "hello"
kind = "bundle"
runtime = "subprocess"
transport = "stdio"
entrypoint = ["python", "plugins/hello_py/plugin.py"]
provides = ["action.open_url@1", "context.live@1", "fs.tmp@1"]
requires = ["fs.tmp@1"]

A relative entrypoint is resolved against the runner’s working directory, and the runner always prepends the repository root to PYTHONPATH so plugins can import utter. Use cwd for a plugin that needs its own directory as the working directory.

Validate a config without starting the runner:

python3 -m runner --check-config runner/config.example.toml

Drive a plugin by hand with the independent client:

python3 tests/conformance/framing_client.py python3 plugins/hello_py/plugin.py

Rust

The tested reference is plugins/fake_rs/src/main.rs. It hand-rolls the Content-Length framing and uses serde_json only for parsing.

cargo build --release --manifest-path plugins/fake_rs/Cargo.toml
# -> plugins/fake_rs/target/release/fake_rs

Its handshake and action replies:

"protocol.hello" => json!({
    "protocol": PROTOCOL, "abi": ABI,
    "plugin": {"name": PLUGIN_NAME, "version": PLUGIN_VERSION, "kind": PLUGIN_KIND},
    "transport": "stdio",
    "provides": PROVIDES, "requires": REQUIRES, "permissions": PERMISSIONS,
}),
"plugin.describe" => json!({ "methods": [...], "streams": [] }),
"plugin.health"   => json!({"status": "ok", "detail": "fake_rs ready"}),
"action.capabilities" => json!({"ops": [
    {"op": "ensure_url", "side_effect": "open_url", "needs_confirm": false}
]}),
"action.invoke" => json!({"ok": true, "op": "ensure_url",
                          "detail": format!("opened {}", url)}),

Terminate on stdin EOF: the runner closes stdin to drain a stdio plugin.

The manifest for the Rust fake, plugins/fake_rs/utter-plugin.toml, is the minimal valid example of a package manifest; see the manifest.

Streams

A plugin that produces audio, deltas or events implements stream.subscribe and stream.stop, plus stream.ack for reliable mode. The emitter in plugins/fake_py/plugin.py is a working example of both lossy (drop-oldest with seq gaps) and reliable (credit window) streams.

Testing with the conformance suite

tests/conformance/run.py exercises the protocol against the runner from a separate client. It needs a Python interpreter with the repo on its path and a built Rust fake.

# build the Rust fake once
cargo build --release --manifest-path plugins/fake_rs/Cargo.toml

# run the suite
python3 tests/conformance/run.py

It generates a runner config, starts python -m runner --config <that file> from the repository root, waits for $XDG_RUNTIME_DIR/utter/runner.sock, and asserts:

Handshake and negotiation, provides ∩ requires, unknown capability tolerated, missing requirement rejected fail-closed, the cross-language Rust fake, the runner.command "open youtube" vertical slice, and policy (terminal with screen provenance gives -32006; with user provenance it gives -32003).
End-to-end lossy backpressure (drop-oldest, seq gaps, responsive sibling RPCs, idempotent stream.stop), reliable flow (pause at the credit window, resume on stream.ack), runner.invoke including -32006, runner.validate_plugin, handles and fd.pass (-32007, fd round trip), socket default-deny, and memfd.
Features the runner has not implemented are reported NOT-YET-SUPPORTED and counted separately, never as failures.

Exit codes: 0 all assertions passed (or only skips), 1 a check failed, 2 blocked (runner absent or not ready). Flags: --keep, --timeout 30.

Run the repository’s full verification with:

scripts/verify.sh

Checklist before you ship

Answer protocol.hello, plugin.describe and plugin.health.
Declare only the capabilities you really provide; declare everything you require.
Advertise your real ops in action.capabilities.
Never turn screen text into arguments. Let the runner’s provenance policy do its job.
Declare permissions honestly, and check what doctor reports as enforced versus advisory.
Exit cleanly on stdin EOF.
Run the conformance suite.