🚀 Install

pip install xstate-statemachine

That's the whole story. Zero runtime dependencies — pure standard library, Python 3.9 → 3.14.

Releases are published from GitHub Actions through PyPI Trusted Publishing (no long-lived token) and carry PEP 740 build provenance attestations binding each wheel and sdist to the exact run, commit and workflow that built it. To verify an artefact instead of trusting a diff you ran yourself:

pip install pypi-attestations
pypi-attestations verify pypi --repository https://github.com/basiltt/xstate-statemachine \
  pypi:xstate_statemachine-0.10.1-py3-none-any.whl   # prints "OK: <file>" on success

xsm info          # verify the install
xsm update        # later: upgrade to the latest release

Windows, xsm.exe blocked by an Application Control policy? That is pip's unsigned launcher stub being refused by WDAC / AppLocker, not the package. Run python -m xstate_statemachine setup once: it parks the blocked launcher and installs a batch shim, after which xsm works normally (re-run after pip install --upgrade; --undo reverts). python -m xstate_statemachine … always works too. Details: CLI → Windows.

Using the code generator and want its output line-wrapped to match your linter? That needs black and isort, which stay optional so the core install keeps its zero-dependency promise:

pip install "xstate-statemachine[format]"

Without them, generated code is still valid and still faithful to your machine — just not reformatted.

uv add xstate-statemachine
poetry add xstate-statemachine
pipx install xstate-statemachine     # if you only want the `xsm` CLI

⚡ The 60-Second Start

Copy, paste, run. No async, no setup, no config files.

from xstate_statemachine import create_machine, SyncInterpreter

machine = create_machine({
    "id": "toggle",
    "initial": "inactive",
    "states": {
        "inactive": {"on": {"TOGGLE": "active"}},
        "active":   {"on": {"TOGGLE": "inactive"}},
    },
})

light = SyncInterpreter(machine).start()

print(light.current_state_ids)      # {'toggle.inactive'}
light.send("TOGGLE")
print(light.current_state_ids)      # {'toggle.active'}
light.send("BANANA")                # not a legal event here
print(light.current_state_ids)      # {'toggle.active'}  ← ignored, not crashed

You just declared the complete set of legal states and the only legal moves between them. TOGGLE advances the machine. BANANA is ignored — not raised, not silently mishandled. Ignored, because the current state does not accept it.

That single property is what kills a whole category of bug.

🧠 Why a Statechart?

Every non-trivial flow starts as a few booleans. Then it grows.

if is_loading and not is_error:
    ...
elif is_error and retry_count < 3:
    ...
elif is_authenticated and not is_loading:
    ...

"states": {
    "idle":    {"on": {"FETCH": "loading"}},
    "loading": {"on": {"OK": "done",
                       "ERR": "failed"}},
    "failed":  {"on": {"RETRY": "loading"}},
    "done":    {"type": "final"},
}

The three bugs this eliminates

The rule — a machine is in exactly one state per region. Parallel states have multiple regions, so multiple states are active at once, which is why current_state_ids returns a set.

🔗 The Part No Other Python Library Does

Your frontend team models a checkout flow in Stately.ai. They export checkout.json and wire it into React with XState.

You take that exact file — unedited — and run it in Python:

import json
from xstate_statemachine import create_machine, MachineLogic, SyncInterpreter

with open("checkout.json") as f:          # ← straight from the frontend repo
    config = json.load(f)

machine = create_machine(config, logic=MachineLogic(
    actions={"charge_card": charge_card},   # you supply the Python side
    guards={"has_stock": has_stock},
))

checkout = SyncInterpreter(machine).start()

One definition. Two runtimes. The UI cannot render a step your backend considers illegal, because there is only one source of truth for what the steps are.

Note — this library implements the SCXML transition-selection algorithm (the W3C standard XState itself follows). That is what makes nested and parallel-region behaviour match XState rather than merely resemble it. It does not import or export .scxml files.

🧩 The Mental Model

Six concepts. That's the entire library.

The split that matters: state is where you are, context is what you know. retries is context. retrying is a state. Getting that boundary right is 90% of good statechart design.

stateDiagram-v2
    direction LR
    [*] --> idle
    idle --> loading: FETCH
    loading --> done: onDone
    loading --> failed: onError
    failed --> loading: RETRY
    done --> [*]

💾 Context — The Machine's Memory

Context is a plain dict. Update it declaratively with assign:

from xstate_statemachine import create_machine, SyncInterpreter, assign

cart = SyncInterpreter(create_machine({
    "id": "cart",
    "initial": "shopping",
    "context": {"items": 0, "total": 0.0},
    "states": {
        "shopping": {
            "on": {
                "ADD_ITEM": {"actions": assign({
                    "items": lambda a: a["context"]["items"] + 1,
                    "total": lambda a: a["context"]["total"] + a["event"].payload["price"],
                })},
                "CLEAR": {"actions": assign(lambda a: {"items": 0, "total": 0.0})},
            }
        }
    },
})).start()

cart.send("ADD_ITEM", price=9.99)
cart.send("ADD_ITEM", price=5.01)
print(cart.context)          # {'items': 2, 'total': 15.0}

assign takes either a dict of per-key updaters or a single callable returning a partial dict. Each updater receives one mapping with "context" and "event" keys.

Tip — keyword arguments to send() land in event.payload. send("ADD_ITEM", price=9.99) → a["event"].payload["price"].

🛡️ Guards — Conditional Transitions

A guard is a pure function returning bool. List transitions in priority order; the first whose guard passes wins.

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "atm",
    "initial": "idle",
    "context": {"balance": 100, "frozen": False},
    "states": {
        "idle": {
            "on": {
                "WITHDRAW": [
                    {"target": "approved", "guard": {
                        "type": "and",
                        "params": {"guards": [
                            "hasFunds",
                            {"type": "not", "params": {"guards": ["isFrozen"]}},
                        ]},
                    }},
                    {"target": "denied"},          # fallback — no guard
                ]
            }
        },
        "approved": {}, "denied": {},
    },
}

logic = MachineLogic(guards={
    "hasFunds": lambda ctx, e: ctx["balance"] >= e.payload.get("amount", 0),
    "isFrozen": lambda ctx, e: ctx["frozen"],
})

atm = SyncInterpreter(create_machine(config, logic=logic)).start()
atm.send("WITHDRAW", amount=50)
print(atm.current_state_ids)      # {'atm.approved'}

Composite guards — and, or, not nest arbitrarily via params.guards. There's also stateIn for "only if some other region is in state X":

{"guard": {"type": "stateIn", "params": {"state": "auth.loggedIn"}}}

Note — guards must be pure. They can be evaluated more than once, and a guard with side effects will surprise you. Put side effects in actions.

🎬 Actions — Side Effects

Actions fire during a transition, or on entering/leaving a state.

"states": {
    "loading": {
        "entry": ["showSpinner"],          # on the way in
        "exit":  ["hideSpinner"],          # on the way out
        "on": {"CANCEL": {"target": "idle", "actions": ["logCancel"]}},
    }
}

Order is guaranteed: exit actions → transition actions → entry actions.

Built-in action creators

You rarely need to hand-write these — import them and go:

Note — if an action raises, the error is logged and contained. The transition still completes and the interpreter keeps running; one buggy side effect can't take down a long-lived machine. To react to a failure, use the on_action_error plugin hook or record it on context and guard on it.

"on": {"GO": {"target": "b", "actions": [choose([
    {"guard": "isBig",   "actions": [assign({"label": lambda a: "big"})]},
    {"guard": "isSmall", "actions": [assign({"label": lambda a: "small"})]},
    {"actions": [assign({"label": lambda a: "other"})]},   # no guard = default
])]}}

"actions": [pure(lambda a:
    [assign({"n": lambda b: b["context"]["n"] * 2})]
    if a["context"]["n"] < 10 else []
)]

"actions": [raise_("VALIDATE")]

"actions": [send_to("timer", "TICK", delay=1000, send_id="tick")]
# elsewhere
"actions": [cancel("tick")]

"actions": [emit("saved")]                        # or emit({"type": "saved", "id": 7})
interpreter.on("saved", lambda ev: analytics.track(ev.type))

🔌 Services & Invoke

invoke runs an async or sync callable when a state is entered, and routes its result back into the machine as onDone / onError. This is how you do I/O.

import asyncio
from xstate_statemachine import (
    create_machine, Interpreter, MachineLogic, assign, wait_for,
)

config = {
    "id": "fetch",
    "initial": "idle",
    "context": {"user": None, "error": None},
    "states": {
        "idle": {"on": {"FETCH": "loading"}},
        "loading": {
            "invoke": {
                "src": "fetchUser",
                "onDone": {"target": "success",
                           "actions": assign({"user": lambda a: a["event"].data})},
                "onError": {"target": "failure",
                            "actions": assign({"error": lambda a: str(a["event"].data)})},
            }
        },
        "success": {"type": "final"},
        "failure": {"on": {"RETRY": "loading"}},
    },
}

async def fetch_user(interpreter, ctx, event):
    await asyncio.sleep(0.01)
    return {"id": 1, "name": "Ada"}

async def main():
    machine = create_machine(config, logic=MachineLogic(services={"fetch_user": fetch_user}))
    svc = await Interpreter(machine).start()

    await svc.send("FETCH")
    await wait_for(svc, lambda s: s.matches("fetch.success"), timeout=2)

    print(svc.context["user"])        # {'id': 1, 'name': 'Ada'}
    await svc.stop()

asyncio.run(main())

• Success → onDone, with the return value on event.data
• Failure → onError, with the exception object on event.data
• Leaving the state cancels the service automatically — no zombie tasks

Tip — use wait_for (async) or wait_for_sync rather than asyncio.sleep() guesses. It polls a predicate with a real timeout, so tests stay fast and never flake.

⏱️ Timers & Delayed Transitions

after fires a transition if the machine is still in that state when the timer elapses. Leave early and the timer is cancelled for you.

"connecting": {
    "after": {5000: "timedOut"},          # 5000 ms
    "on": {"OPEN": "online"},             # ...unless we connect first
}

Name your delays to keep magic numbers out of the config — and to compute them at runtime, which is exactly how you express exponential backoff:

logic = MachineLogic(delays={
    "TIMEOUT": 60_000,
    "BACKOFF": lambda ctx, e: 2 ** ctx["attempt"] * 1000,   # 1s, 2s, 4s, 8s…
})

"retrying": {"after": {"BACKOFF": "loading"}}

🌳 Nested & Parallel States

Nested (compound) states

Group related substates so shared transitions live in one place:

"states": {
    "authenticated": {
        "initial": "browsing",
        "on": {"LOGOUT": "loggedOut"},     # ← applies to EVERY substate
        "states": {
            "browsing":  {"on": {"CHECKOUT": "paying"}},
            "paying":    {"on": {"DONE": "confirmed"}},
            "confirmed": {},
        },
    },
    "loggedOut": {},
}

LOGOUT works from browsing, paying, and confirmed. Write it once.

Parallel states — concurrent regions

Regions run independently. onDone fires exactly once, when all of them reach a final state — fan-out and fan-in with no bookkeeping:

from xstate_statemachine import create_machine, SyncInterpreter

ci = SyncInterpreter(create_machine({
    "id": "ci",
    "initial": "running",
    "states": {
        "running": {
            "type": "parallel",
            "onDone": "deployed",
            "states": {
                "build": {"initial": "b", "states": {
                    "b": {"on": {"BUILD_OK": "done"}}, "done": {"type": "final"}}},
                "lint":  {"initial": "l", "states": {
                    "l": {"on": {"LINT_OK": "done"}},  "done": {"type": "final"}}},
            },
        },
        "deployed": {},
    },
})).start()

print(sorted(ci.current_state_ids))   # ['ci.running.build.b', 'ci.running.lint.l']
ci.send("BUILD_OK")
print(sorted(ci.current_state_ids))   # ['ci.running.build.done', 'ci.running.lint.l']
ci.send("LINT_OK")
print(sorted(ci.current_state_ids))   # ['ci.deployed']   ← fan-in fired

This is where current_state_ids returning a set finally makes sense.

🕰️ History & Final States

History remembers where you were, so an interruption doesn't lose progress — the classic "resume the wizard where the user left off":

"states": {
    "wizard": {
        "initial": "step1",
        "states": {
            "step1": {}, "step2": {}, "step3": {},
            "hist": {"type": "history", "history": "shallow"},   # or "deep"
        },
    },
    "helpModal": {"on": {"CLOSE": "wizard.hist"}},    # ← back to the exact step
}

Final states mark completion. A final state in a compound state fires its parent's onDone; a top-level final state stops the machine and can produce output.

🤖 The Actor Model

Machines can spawn other machines. Each child gets its own state, context and lifecycle — a supervision tree, not a callback pile. Register a child under a systemId and any machine in the system can address it by name.

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

# The child machine — an independent actor with its own context.
worker = {
    "id": "worker",
    "initial": "idle",
    "context": {"jobs": 0},
    "states": {"idle": {"on": {"JOB": {"target": "idle", "actions": ["count"]}}}},
}
worker_logic = MachineLogic(actions={
    "count": lambda i, ctx, e, a: ctx.__setitem__("jobs", ctx["jobs"] + 1),
})

parent = {
    "id": "super",
    "initial": "up",
    "context": {},
    "states": {
        "up": {
            "entry": [{"type": "spawnChild",
                       "params": {"src": "worker", "id": "w1", "systemId": "pool"}}],
            "on": {"DISPATCH": {"actions": [
                {"type": "sendTo", "params": {"to": "pool", "event": {"type": "JOB"}}}
            ]}},
        }
    },
}

logic = MachineLogic(services={
    "worker": lambda i, ctx, e: create_machine(worker, logic=worker_logic),
})

sup = SyncInterpreter(create_machine(parent, logic=logic)).start()
print(list(sup.system.get_all()))          # ['pool']

sup.send("DISPATCH")
sup.send("DISPATCH")
print(sup.system.get("pool").context["jobs"])   # 2

Children talk back with send_parent, escalate failures with escalate, and are torn down with stop_child — or automatically when the parent stops.

Good fit for: LLM agent orchestration (each tool call a supervised child), connection pools, per-user session machines, job workers.

💾 Persistence — Snapshots

Serialize a running machine to JSON, store it anywhere, rebuild it later. Long-running flows survive deploys and restarts.

from xstate_statemachine import create_machine, SyncInterpreter

job = SyncInterpreter(create_machine(config)).start()
job.send("NEXT")

snapshot = job.get_snapshot()      # a JSON string → Redis, Postgres, a file…
job.stop()

# …new process, hours later…
resumed = SyncInterpreter.from_snapshot(snapshot, create_machine(config))
print(resumed.current_state_ids)   # {'job.step2'}   ← exactly where it left off
resumed.send("NEXT")

State, context and systemId registrations all round-trip. get_persisted_snapshot() gives you the dict form if you'd rather store structured data.

Note — pending after timers are not resumed by a restore. A machine saved while waiting on a 30-minute timeout will wait indefinitely after restore. If a deadline must survive a restart, store it in context and re-arm it yourself on resume.

Every snapshot carries an envelope (version, machine_id, machine_hash) so from_snapshot() refuses a structurally different machine with SnapshotDriftError instead of silently resuming into it; pass verify_machine_hash=False after a deliberate migration. Invokes are not restarted either — pass from_snapshot(..., restart_services=True) to re-invoke every service pending_invocations() reports, or leave them stopped and re-trigger manually.

🧪 The Pure API — No Interpreter

Sometimes you want to ask "what would happen if…" without running anything. The pure API is a set of side-effect-free functions over immutable snapshots — ideal for tests, planning, and "preview the next step" UI.

from xstate_statemachine import (
    MachineLogic, create_machine, initial_transition, pure_transition,
)

machine = create_machine({
    "id": "fetch",
    "initial": "idle",
    "states": {
        "idle":    {"on": {"FETCH": {"target": "loading", "actions": "logStart"}}},
        "loading": {"on": {"OK": "done"}},
        "done":    {"type": "final"},
    },
}, logic=MachineLogic())      # 📝 no implementations needed — nothing runs

snapshot, entry_actions = initial_transition(machine)
next_snapshot, actions = pure_transition(machine, snapshot, "FETCH")

print(snapshot.state_ids)         # {'fetch.idle'}
print(next_snapshot.state_ids)    # {'fetch.loading'}
print([a.type for a in actions])  # ['logStart'] — what WOULD have run

Both functions return (snapshot, actions). If you only want the next state, get_next_snapshot(machine, snapshot, "FETCH") returns the snapshot alone.

A PureSnapshot exposes state_ids, context, status, output, configuration and matches(). No timers start. No services fire. Nothing mutates.

🔍 Introspection & Plugins

A running machine can answer questions about itself — which is what lets you drive a UI from it without duplicating its logic in your view layer.

from xstate_statemachine import (
    create_machine, MachineLogic, SyncInterpreter, assign, emit,
)

editor = create_machine({
    "id": "editor",
    "initial": "clean",
    "context": {"saves": 0},
    "states": {
        "clean":  {"tags": ["idle"],
                   "meta": {"hint": "Nothing to save"},
                   "on": {"EDIT": "dirty"}},
        "dirty":  {"tags": ["unsaved"], "on": {"SAVE": "saving"}},
        "saving": {"tags": ["unsaved", "busy"],
                   "on": {"OK": {"target": "clean", "actions": [
                       assign({"saves": lambda a: a["context"]["saves"] + 1}),
                       emit("saved"),
                   ]}}},
    },
}, logic=MachineLogic())

ed = SyncInterpreter(editor).start()

ed.matches("editor.clean")   # True  — nested paths work: "a.b.c"
ed.can("EDIT")               # True  — would this event do anything *right now*?
ed.can("SAVE")               # False — not handled in `clean`
ed.has_tag("idle")           # True
ed.tags                      # {'idle'}
ed.get_meta()                # {'editor.clean': {'hint': 'Nothing to save'}}
ed.context                   # {'saves': 0}
ed.is_running                # True

can() — disable buttons without duplicating logic

The machine already knows which events are legal. Ask it, instead of re-deriving the rule in your template:

save_button.disabled = not ed.can("SAVE")

Tags — style many states with one check

saving and dirty are different states but share the unsaved tag, so a spinner needs one condition rather than a growing or chain:

if ed.has_tag("busy"):
    show_spinner()

subscribe() — react to every settled transition

The listener receives the interpreter, so read whatever you need from it:

unsubscribe = ed.subscribe(
    lambda i: print(sorted(i.current_state_ids), i.context)
)
# … later
unsubscribe()

on() — listen for emitted events

emit publishes a domain event without coupling the machine to your transport:

ed.on("saved", lambda event: analytics.track(event.type))
ed.on("*", lambda event: audit_log.append(event))   # every emitted event

ed.send("EDIT"); ed.send("SAVE"); ed.send("OK")
ed.context     # {'saves': 1}   ← assign ran
ed.tags        # {'idle'}       ← back in `clean`

Plugins — the whole lifecycle, one line

from xstate_statemachine import LoggingInspector

ed.use(LoggingInspector())    # complete transition audit trail

Subclass PluginBase for metrics, tracing, or persistence-on-every-transition. Every hook is optional:

from xstate_statemachine import PluginBase

class Metrics(PluginBase):
    def on_transition(self, interpreter, from_states, to_states, transition):
        statsd.increment(f"fsm.{transition.event}")

    def on_action_error(self, interpreter, action, error):
        sentry.capture_exception(error)   # otherwise silently contained

ed.use(Metrics())

⚖️ How It Compares

Python has good state machine libraries. Here's an honest read on when to pick which.

Pick transitions if you want the most battle-tested option and a simple FSM bolted onto an existing class. It's mature, widely deployed, and excellent at that job.

Pick python-statemachine if you want a beautiful, pythonic declarative API and don't need JS interop. It genuinely supports compound, parallel and history states too — this is a real alternative, not a strawman.

Pick this library when you want XState/Stately JSON to run in Python unchanged, or you want invoke, after, actors and snapshots as first-class primitives instead of patterns you assemble yourself.

Speed

Same machine shape, each library through its own idiomatic API, all measured in one session on 0.9.0 (Python 3.14, median of 7 runs, GC disabled, setup excluded). Events per second; bold is fastest in the row.

transitions is a transition table, not a statechart engine, and wins the flat scenario by ~2.1×. The moment states nest or run in parallel it has to emulate the SCXML algorithm and this library is 3.2–7.5× faster than everything else. It is also the fastest of the four to construct (1.19× transitions) and to fan out to 1,000 instances (1.10×), while still running the full build-time validator on every create_machine(). Full table, method and caveats: benchmarks/competitors/. The production numbers (throughput budget, after lateness under load) come from benchmarks/production_characteristics.py; run it with --json on your own hardware to gate CI on your figures.

When not to use this

For a three-state toggle with no I/O, a plain enum and an if is less machinery and easier to read. Statecharts start paying for themselves when you have concurrency, timeouts, cancellation, or more than ~5 states — and they pay enormously at 20.

📚 Cookbook

Real problems, small solutions.

Every recipe below is a fragment for readability. Here is one complete, runnable program first — a checkout that guards an empty cart, retries a declining card, and records the failure reason, in 40 lines:

from xstate_statemachine import (
    MachineLogic, SyncInterpreter, assign, create_machine,
)

ORDER = {
    "id": "order",
    "initial": "cart",
    "context": {"items": 0, "attempts": 0, "error": None},
    "states": {
        "cart": {
            "on": {
                "ADD": {"actions": assign(
                    {"items": lambda a: a["context"]["items"] + 1})},
                "CHECKOUT": {"target": "charging", "guard": "hasItems"},
            },
        },
        "charging": {
            "entry": assign({"attempts": lambda a: a["context"]["attempts"] + 1}),
            "invoke": {
                "src": "chargeCard",
                "onDone": "shipped",
                "onError": {
                    "target": "failed",
                    "actions": assign({"error": lambda a: str(a["event"].data)}),
                },
            },
        },
        "failed": {"on": {"RETRY": {"target": "charging", "guard": "canRetry"}}},
        "shipped": {"type": "final"},
    },
}


def charge_card(interpreter, context, event):
    """Fails the first time, succeeds on the retry."""
    if context["attempts"] < 2:
        raise RuntimeError("card declined")
    return {"receipt": "r-123"}


logic = MachineLogic(
    guards={
        "hasItems": lambda ctx, e: ctx["items"] > 0,
        "canRetry": lambda ctx, e: ctx["attempts"] < 3,
    },
    services={"charge_card": charge_card},
)

order = SyncInterpreter(create_machine(ORDER, logic=logic)).start()

order.send("CHECKOUT")                    # guard blocks — the cart is empty
print(sorted(order.current_state_ids))    # ['order.cart']

order.send("ADD")
order.send("CHECKOUT")                    # charges; the service raises
print(sorted(order.current_state_ids))    # ['order.failed']
print(order.context["error"])             # card declined

order.send("RETRY")                       # second attempt succeeds
print(sorted(order.current_state_ids))    # ['order.shipped']
print(order.context["attempts"])          # 2

config = {
    "id": "api",
    "initial": "idle",
    "context": {"attempt": 0},
    "states": {
        "idle": {"on": {"CALL": "loading"}},
        "loading": {
            "invoke": {
                "src": "callApi",
                "onDone": "success",
                "onError": [
                    {"target": "waiting", "guard": "canRetry"},
                    {"target": "failed"},               # out of retries
                ],
            }
        },
        "waiting": {
            "entry": assign({"attempt": lambda a: a["context"]["attempt"] + 1}),
            "after": {"BACKOFF": "loading"},
        },
        "success": {"type": "final"},
        "failed":  {"type": "final"},
    },
}

logic = MachineLogic(
    services={"call_api": call_api},
    guards={"canRetry": lambda ctx, e: ctx["attempt"] < 5},
    delays={"BACKOFF": lambda ctx, e: 2 ** ctx["attempt"] * 1000},
)

"states": {
    "reviewing":  {"on": {"SUBMIT": "charging"}},
    "charging":   {                                  # ← no SUBMIT handler here
        "invoke": {"src": "chargeCard",
                   "onDone": "confirmed", "onError": "declined"},
    },
    "confirmed":  {"type": "final"},
    "declined":   {"on": {"SUBMIT": "charging"}},
}

"states": {
    "disconnected": {"on": {"CONNECT": "connecting"}},
    "connecting": {
        "invoke": {"src": "openSocket", "onDone": "connected", "onError": "backoff"},
        "after": {"CONNECT_TIMEOUT": "backoff"},
    },
    "connected": {
        "on": {"PONG": "connected", "CLOSE": "disconnected"},   # self-transition resets timer
        "after": {"HEARTBEAT": "reconnecting"},
    },
    "backoff": {"after": {"RETRY_DELAY": "connecting"}},
    "reconnecting": {"on": {"CONNECT": "connecting"}},
}

"states": {
    "planning": {"invoke": {"src": "askModel",
                            "onDone": [{"target": "callingTool", "guard": "wantsTool"},
                                       {"target": "answering"}]}},
    "callingTool": {
        "entry": [{"type": "spawnChild",
                   "params": {"src": "toolRunner", "id": "tool", "systemId": "tool"}}],
        "on": {"TOOL_RESULT": "reflecting", "TOOL_FAILED": "recovering"},
        "after": {"TOOL_TIMEOUT": "recovering"},
    },
    "reflecting": {"always": [{"target": "planning", "guard": "needsMoreWork"},
                              {"target": "answering"}]},
    "recovering": {"always": [{"target": "planning", "guard": "canRetry"},
                              {"target": "givingUp"}]},
    "answering": {"type": "final"},
    "givingUp":  {"type": "final"},
}

def test_declined_card_allows_retry():
    checkout = SyncInterpreter(create_machine(config, logic=test_logic)).start()

    checkout.send("SUBMIT")
    assert checkout.matches("checkout.charging")

    checkout.send("SUBMIT")                       # double click
    assert checkout.matches("checkout.charging")  # …ignored

🐍 Prefer Pure Python? Three More Ways to Define a Machine

JSON is the interop format, not an obligation. If you're not sharing definitions with a frontend, define machines in Python instead.

Class-based — declarative and readable

from xstate_statemachine import State, StateMachine, SyncInterpreter, action

class Checkout(StateMachine):
    machine_id = "checkout"
    initial_context = {"attempts": 0}

    reviewing = State(initial=True)
    charging  = State()
    confirmed = State()

    submit = reviewing.to(charging, event="SUBMIT", actions=["recordAttempt"])
    ok     = charging.to(confirmed, event="PAID")

    @action
    def record_attempt(self, interpreter, ctx, evt, action_def):
        ctx["attempts"] += 1

c = SyncInterpreter(Checkout.create_machine()).start()
c.send("SUBMIT")
print(c.current_state_ids, c.context)   # {'checkout.charging'} {'attempts': 1}
c.send("SUBMIT")                        # double click → ignored
print(c.context)                        # {'attempts': 1}

Watch out — @action, @guard and @service convert snake_case method names to camelCase keys. The method record_attempt is referenced as "recordAttempt".

Compose multiple transitions for one event with |:

flip = off.to(on, event="TOGGLE") | on.to(off, event="TOGGLE")

Builder — fluent

from xstate_statemachine import MachineBuilder

machine = (MachineBuilder("toggle")
           .state("off", initial=True)
           .state("on")
           .transition("off", "TOGGLE", "on")
           .transition("on", "TOGGLE", "off")
           .build())

transition() takes (source, event, target), so states and transitions can be declared in any order — handy when you're generating a machine from data.

Functional — build_machine()

Plain objects and explicit wiring. The style to reach for when the machine is data you are assembling, not a shape you are declaring:

from xstate_statemachine import (
    State, SyncInterpreter, action, build_machine,
)

@action
def record_attempt(interpreter, ctx, evt, action_def):
    ctx["attempts"] += 1

reviewing = State("reviewing", initial=True,
                  on={"SUBMIT": {"target": "charging",
                                 "actions": ["recordAttempt"]}})
charging  = State("charging", on={"PAID": "confirmed"})
confirmed = State("confirmed", final=True, tags=["done"])

machine = build_machine(
    id="checkout",
    states=[reviewing, charging, confirmed],
    context={"attempts": 0},
    actions=[record_attempt],
)

c = SyncInterpreter(machine).start()
c.send("SUBMIT")
print(sorted(c.current_state_ids), c.context)   # ['checkout.charging'] {'attempts': 1}
c.send("PAID")
print(sorted(c.current_state_ids), sorted(c.tags))  # ['checkout.confirmed'] ['done']

Everything the JSON format supports

All three styles compile to the same MachineNode, so none of them is a reduced subset. Nesting, parallel regions, history, timers, tags and metadata are all expressible:

State("online", initial=True, states=[configuring, running, resume],
      on={"DISCONNECT": "offline"}, tags=["connected"])

State("resume", history="deep")           // remembers the last active child
State("failed", meta={"alert": True})     // arbitrary data for your UI
State("regions", parallel=True, states=[...])

Machine-level properties — a global escape transition, root entry/exit, or a parallel root — go on root=:

from xstate_statemachine import State, SyncInterpreter, build_machine

root = State("", on={"EMERGENCY": "halted"}, tags=["v2"])
machine = build_machine(
    id="press",
    states=[State("idle", initial=True), State("running"), State("halted")],
    root=root,
)

p = SyncInterpreter(machine).start()
p.send("EMERGENCY")                       # works from ANY state
print(sorted(p.current_state_ids))        # ['press.halted']

MachineBuilder.root(...) and a machine_root class attribute do the same for the other two styles.

Runnable examples for all three styles — building the same machine, with invoke, timers, guards, tags and meta — live in examples/sync/easy/pythonic_approach/.

🛠️ CLI Tool

xsm is the terminal companion to the library — a code generator, an inspector, a live simulator and a diagram/docs exporter in one zero-dependency command. Run it bare on a terminal for an interactive launcher with a menu, recent files and a generate wizard that previews before it writes; pipe it and every command degrades to clean plain text.

xsm                                                           # interactive launcher
xsm gt checkout.json -t pythonic-class --with-tests --with-types -o ./app
xsm inspect checkout.json                                     # tree, transitions, logic, policies
xsm simulate checkout.json                                    # live: pick events, +clock, undo
xsm sim checkout.json --events SUBMIT,+2001 --json            # scripted, for CI
xsm diagram checkout.json -f mermaid -o docs/
xsm docs machines/*.json -o docs/

Primary templates: class-json, function-json, pythonic-class, pythonic-builder, pythonic-functional. Companion templates: pytest (a test module recorded from the engine — one test per reachable step, green on day one), typed (TypedDict context, Literal events, typed stubs), plugin (a PluginBase wired for exactly the hooks the chart can fire).

The generator proves its output before writing it. For templates that build the machine in Python, xsm compiles the generated code, runs it, and compares the resulting machine against create_machine(your.json). If anything diverges it prints what and exits non-zero — nothing is written. Nesting, parallel regions, history, timers (numeric and named delays), composite guards, invoke, tags and meta all round-trip exactly.

Add --check in CI to catch generated code that has drifted from its source JSON:

xsm generate-template checkout.json --template pythonic-class -o ./app --check

--plain, --no-color (also NO_COLOR) and --no-anim control presentation; --json on validate, inspect, simulate, list-templates and info gives scripts the same facts. Full reference: CLI Tool.

🏭 Running It in Production

Everything above is the happy path. Here is what matters once real traffic arrives.

Throughput, timers and threads — read this before sizing

All async interpreters in a process share one event loop on one thread: throughput is a per-process budget (~30k trivial ev/s on a laptop), divided among your machines. after timers now fire through a priority lane the run loop checks ahead of its inbox — ~45 ms late at 500 busy machines, down from ~180 ms before 0.8.0 — and a SyncInterpreter timer only fires when someone calls send() or tick(), on the caller's thread. In fact, neither engine spawns an OS thread per timer anymore; the only thread work either one does is running a non-blocking spawn_* child. Pass Interpreter(clock=SimulatedClock()) in tests to fire an after timer without sleeping — see Testing below. The measured tables and a sizing rule are in Production Characteristics — the one page to read before deploying.

Failure semantics — know what is contained

Each is a per-machine policy. The default preserves the historical behaviour; production machines should opt in explicitly.

Containment by default is deliberate: a long-lived machine should not die because one side effect had a bad day. The cost is that failures are invisible unless you look, so wire up the hooks early — every one of them fires whatever policy you choose:

from xstate_statemachine import PluginBase

class ErrorReporter(PluginBase):
    def on_action_error(self, interpreter, action, error):
        sentry.capture_exception(error)

interp.use(ErrorReporter())

Note — actionErrorPolicy defaults to "continue" today (with a one-shot DeprecationWarning); it flips to "rollback" in 1.0. Pin it explicitly if you need today's behaviour to survive the upgrade.

Asking the machine a question

send() normally fires and forgets. Pass wait=True to get a Receipt once that exact event's macrostep has run — no polling, no wait_for():

receipt = await interp.send("SUBMIT", wait=True)
# Receipt(state_ids=frozenset({'checkout.paying'}), changed=True, error=None,
#         deferred=False, denied=False)

await interp.send_priority("CANCEL")   # ahead of the inbox, exempt from its bound

priority=True on send() does the same as send_priority(). A Receipt has five fields — read them by attribute; deferred says the event was parked under onUnhandled: "defer", denied that a handler existed but every guard said no.

Two rules for wait=True inside an action: don't await your own receipt on the action's own task (the run loop can't advance until the action returns, so it raises ReentrantWaitError rather than deadlocking), and don't drop it from a plain def action (it warns). Hand it out — asyncio.ensure_future(i.send("GO", wait=True)) — or send without wait; a helper task the action spawns may await freely.

Waiting for a machine to settle

Do not poll by hand or sleep() and hope:

from xstate_statemachine import wait_for, wait_for_sync, to_promise

# async
await wait_for(interp, lambda i: i.matches("job.done"), timeout=30)
result = await to_promise(interp)          # resolves when the machine reaches a final state

# sync
wait_for_sync(interp, lambda i: i.matches("job.done"), timeout=30)

Choosing an interpreter

Same machine JSON, same semantics, same guarantees. Timers, services and actors all work on both engines; neither spawns an OS thread per timer — the sync engine delivers a due after timer on the caller's own thread inside send() or tick(), and the async engine delivers it through the priority lane described above. A spawn_* (non-blocking) child is the one thing either engine runs off-thread.

Long-running machines

• Persist on transition, not on a timer — get_persisted_snapshot() in a subscribe() callback gives you crash-safe resume points.
• after timers are not re-armed by default. A snapshot records that a timer was pending, not how far along it was; after a static restore has_dormant_timers is True. Pass from_snapshot(..., restart_timers=True) to re-arm each from zero, or model the deadline as data in context. A timer that had already fired is in the snapshot and replays. A delayed self-send (raise/send with delay) is persisted with its remaining time and resumes where it would have been.
• Invokes do not restart on restore either — from_snapshot() is a static rebuild that starts nothing by default. Call pending_invocations() on the restored interpreter to see every PendingInvocation(state_id, invoke_id, src) with no live service (has_dormant_invocations is the boolean), and from_snapshot(..., restart_services=True) to re-invoke each of them from scratch.
• Snapshots carry an envelope (version — layout 3 — machine_id, machine_hash) so a restore against a machine that no longer matches the one that produced the snapshot fails loud with SnapshotDriftError instead of resuming into undefined behaviour. Pass verify_machine_hash=False after a deliberate migration. The hash is a drift check, not an authentication tag: if a blob crosses a trust boundary, sign it outside and pin from_snapshot(..., minimum_version=3, expected_machine_hash=...) so the payload cannot pick its own level of checking.
• What else round-trips: pending events with their lane (priority events restore ahead of the inbox on both engines), deferred events, armed delayed sends, history, child actors, the error, and the chain-trip latch (chain_trips / last_chain_error) — a restart is not an acknowledgement. strict and event_schemas are applied to every restored event; a refusal is reported, not silently admitted.
• Always stop() — it cancels timers and stops spawned actors. In a web app, tie it to request teardown; in a worker, to the task's finally.

Testing

The pure API is the simplest way to test machine logic — no event loop, no mocks, no sleeping:

from xstate_statemachine import get_initial_snapshot, get_next_snapshot

snap = get_initial_snapshot(machine)
snap = get_next_snapshot(machine, snap, "SUBMIT")
assert snap.matches("checkout.paying")

Use a real interpreter for integration tests, where you want the actions to actually run. For an after timer, don't sleep — inject a SimulatedClock and jump virtual time:

from xstate_statemachine import SyncInterpreter, SimulatedClock

clock = SimulatedClock()
interp = SyncInterpreter(machine, clock=clock).start()
clock.increment(30_000)             # fires a 30 s `after` with no real delay
assert interp.matches("job.timedout")

📘 API Reference

🚨 Troubleshooting

The errors you are most likely to meet, and what each actually means.

create_machine({"id": "a", "initial": "s",
                "states": {"s": {"entry": "logStart"}}})
# ImplementationMissingError: Action 'logStart' is defined in the machine
# but no implementation was found …

create_machine(config, logic=MachineLogic(actions={"logStart": my_fn}))
create_machine(config, logic=MachineLogic())   # accept the stubs; nothing runs

{"s": {"on": {"GO": "ghost"}}}     # 'ghost' is not a sibling of 's'

create_machine({"id": "c"})        # InvalidConfigError: 'states' key is missing
create_machine({"id": "c", "initial": "a", "states": {"a": {"entyr": ["x"]}}},
               strict_config=True)
# InvalidConfigError: Machine 'c' has unknown config key(s) -- c.a: 'entyr'
#   (did you mean 'entry'?) ...

interp.chain_trips          # monotonic, survives a snapshot
interp.last_chain_error     # the RunawayChainError, latched until clear_chain_error()
receipt.error               # on the send(wait=True) that tripped it

async def act(i, ctx, event, action):
    i.send("GO")                                         # fine
    fut = asyncio.ensure_future(i.send("GO", wait=True))  # fine: awaited elsewhere
    await i.send("GO", wait=True)                        # ReentrantWaitError

class ErrorReporter(PluginBase):
    def on_action_error(self, interpreter, action, error):
        raise error          # or log it, or ship it to Sentry

interp.use(ErrorReporter())

interp.can("SUBMIT")          # False → not handled here at all
interp.current_state_ids      # are you in the state you think you are?
interp.use(LoggingInspector())  # shows guards evaluating and rejecting

"entry": assign({"deadline": lambda a: time.time() + 30}),

❓ FAQ

🗺️ Diagrams

Every machine can draw itself, with no graphviz install:

print(machine.to_mermaid())      # paste into GitHub, Notion, Obsidian…
print(machine.to_plantuml())