🚀 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

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, record it on context and guard on it.

🔌 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.

🧪 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 create_machine, initial_transition, pure_transition

machine = create_machine(config)

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])  # the actions that 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

Every interpreter answers questions about itself:

interp.matches("checkout.paying")     # bool — supports nested paths
interp.can("SUBMIT")                  # would this event do anything right now?
interp.has_tag("busy")                # state tags, great for UI binding
interp.get_meta()                     # merged `meta` from active states
interp.subscribe(lambda snap: ...)    # observe every transition

Plugins hook the whole lifecycle — one line gives you a complete transition audit trail:

from xstate_statemachine import LoggingInspector

interp.use(LoggingInspector())

Subclass PluginBase for metrics, tracing, or persistence-on-every-transition.

⚖️ 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.

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

For programmatic construction where a dict is the natural shape.

🛠️ 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.

📘 API Reference

❓ FAQ

🗺️ Diagrams

Every machine can draw itself, with no graphviz install:

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