🚀 Install

pip install xstate-statemachine

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

xsm info          # verify the install

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={"chargeCard": charge_card},   # you supply the Python side
    guards={"hasStock": 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={"fetchUser": 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.

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={"chargeCard": 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={"callApi": 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 Code Generator

Point xsm at an XState JSON file and get runnable, typed Python scaffolding — every action, guard and service stubbed with the right signature.

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

Templates: class-json, function-json, pythonic-class, pythonic-builder, pythonic-functional.

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

🏭 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)

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

priority=True on send() does the same as send_priority().

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 do not survive a snapshot. Restoring a machine that was mid-timeout will not re-arm it; re-send the triggering event, or model the deadline as data in context and compare against wall-clock on resume.
• 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, and from_snapshot(..., restart_services=True) to re-invoke each of them from scratch through the same path a fresh enter uses.
• Snapshots carry an envelope (version, 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 from_snapshot(..., verify_machine_hash=False) after a deliberate migration.
• 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 five 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

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())