statebus

Typed, versioned state for multi-agent workflows.

Every multi-agent team rebuilds the same three things: shared state, schema validation between steps, and handoff tracing. statebus is that layer, framework-agnostic: agents read/write named slots, the bus validates every write against a versioned JSON Schema, keeps full version history, and traces every handoff.

"Agent B got v3 of customer_record (schema v2.1) from Agent A at 14:03."

Replay any handoff. Library-first, with MCP tools and a REST adapter for non-Python agents.

Install

pip install statebus        # PyPI: statebus (free, unreleased — v0.1.0)

npm (for the adapters' JS clients): @decnutz/statebus — the unscoped statebus name is taken on npm by an unrelated package.

Requires Python 3.10+ and jsonschema.

Quick start

from statebus import StateBus, ConflictError

bus = StateBus("team.db")  # durable SQLite; StateBus() is in-memory

bus.register_slot(
    "customer_record",
    schema={
        "type": "object",
        "properties": {
            "name": {"type": "string"},
            "email": {"type": "string"},
            "tier": {"type": "string", "enum": ["free", "pro"]},
        },
        "required": ["name", "email"],
    },
    schema_version="2.1",
)

# Agent A writes; the value is validated, versioned, traced.
receipt = bus.write("customer_record", {"name": "Ada", "email": "a@x.com", "tier": "pro"},
                    agent="agent_a")
print(receipt.version)  # 1

# Agent B reads — plain JSON, no wrapper.
customer = bus.read("customer_record")

# What did Agent B actually see? Replay the exact handoff.
old = bus.replay("customer_record", version=1)

# Who wrote what, when:
for e in bus.trace(slot="customer_record"):
    print(e.seq, e.ts, e.kind, e.agent, e.version)

Safe concurrent writes

Default is last-write-wins with full history preserved (nothing is ever lost — both versions stay replayable). For contested slots, use optimistic concurrency:

try:
    bus.write("customer_record", new_value, agent="agent_b", expected_version=3)
except ConflictError as e:
    # e.current_version / e.current_agent tell you who moved it — re-read and retry.
    ...

Schema evolution without silent breakage

# Compatible change (new optional field): registered, becomes active.
bus.register_schema_version("customer_record", new_schema, "2.2")

# Breaking change (new required field): rejected unless you mean it.
bus.register_schema_version("customer_record", breaking_schema, "3.0",
                            allow_breaking=True)  # explicit, and traced

Every write stamps the schema version it was validated against, so readers always know what shape they're holding.

CLI

statebus --db team.db register customer_record --file schema.json
statebus --db team.db write customer_record --json '{"name":"Ada","email":"a@x.com"}'
statebus --db team.db read customer_record
statebus --db team.db history customer_record --limit 5
statebus --db team.db replay customer_record 2
statebus --db team.db trace --slot customer_record --limit 20
statebus --db team.db watch            # follow the trace live
statebus --db team.db prune            # enforce max_versions retention
statebus --db team.db prune-trace      # bound trace storage

MCP adapter (for Claude Desktop, agents, etc.)

statebus-mcp --db team.db

Speaks MCP over stdio (newline-delimited JSON-RPC): initialize → tools/list → tools/call. Nine tools: statebus_register, statebus_write, statebus_read, statebus_inspect, statebus_history, statebus_replay, statebus_slots, statebus_trace, statebus_register_schema_version. Validation and conflict errors are returned as tool-level isError results with structured detail — the agent can repair its own output.

REST adapter

statebus-serve --db team.db --port 8765 --token SECRET

JSON over HTTP for non-Python agents: GET/POST /slots, POST /slots/{name}/write, GET /slots/{name} (envelope), GET /slots/{name}/value (plain value), GET /slots/{name}/history, GET /slots/{name}/versions/{v}, GET /trace, … Status codes are meaningful: 422 validation (with structured issues), 409 conflict or duplicate, 404 unknown slot/version, 413 oversized, 400 bad argument.

Design decisions

• JSON Schema only (Draft 2020-12) — agents speak JSON, not Python classes; the schema registry stays portable across languages.
• Append-only versions — history is never mutated; archive, never delete.
• No auto-merge — the bus never invents data. If two agents disagree, a reducer agent merges explicitly, in the trace.
• Explicit schemas only — no inference from first write (that's the "small formatting differences" bug factory).
• Genesis pinning — pruning keeps version 1 plus the newest N.
• Reads traced only on demand (trace_reads) — polling agents would otherwise flood the trace.

Limitations (honest)

1. Single-host. Durability is SQLite+WAL: crash-safe and serialized across processes on one machine. Multi-host deployments should run one bus behind statebus-serve — there is no consensus, replication, or HA.
2. Compat checker is heuristic, not a proof. It catches the common breakages (added required fields, narrowed types/enums, tightened bounds, removed properties under closed schemas). Exotic constructs (allOf/oneOf/not, conditionals, patternProperties) are skipped conservatively — review those schema changes by hand.
3. Hand-rolled MCP server. No mcp SDK dependency keeps installs light, but the stdio implementation covers the protocol core (initialize/tools/list/tools/call/ping) and could drift from future MCP spec revisions.
4. No pub/sub. Agents poll trace(since=last_seq) or use statebus watch. Webhooks are a future addition.
5. No encryption at rest. The SQLite file holds plaintext values; rely on filesystem permissions. Values are capped (default 1 MiB per write) and NaN/Infinity are rejected to keep the JSON valid for every language.
6. Trace retention is manual. prune() bounds versions; call prune_trace() (or schedule it) to bound the trace table on long-running buses.
7. In-process trust boundary. A caller with a reference to the StateBus object can reach bus._store; the guarantee "every write is validated" holds for all public APIs (library, CLI, MCP, REST), not against deliberate in-process sabotage.

API reference (essentials)

Errors: ValidationError (structured .issues → [{path, message, validator, expected}]), ConflictError, BreakingSchemaChangeError, InvalidSchemaError, SlotNotFoundError, SlotArchivedError, ValueTooLargeError, NotJsonSerializableError.

License

MIT — Calas Automations Inc.