xstate-statemachine

The event-driven workflow toolkit for Python.

Design the flow. Make the rules explicit. Build systems you can explain.

From a two-state toggle to an event-driven order system, keep the workflow in one place.

Draw it in Stately, load XState-compatible JSON, or define it in Python. Run it with an async or sync engine. Let xsm generate the typed code, tests and docs. Add persistence, broker messaging and framework integrations when your application needs them.

Zero runtime dependencies in the core. MIT licensed. Infrastructure is optional.

Try it in 60 seconds →   ·   Explore event-driven architecture   ·   Compare your options   ·   Read the docs

⚡ The 60-Second Start

pip install xstate-statemachine

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
light.stop()

_{🔗 Open this machine in the Stately editor — public, no account needed.
That dict, as the Stately editor draws it. Paste the JSON into stately.ai/editor and you get this picture; export from the editor and you get that dict.}

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 — because the current state does not accept it. (Prefer it to raise? SyncInterpreter(machine, strict=True) → UnknownEventError at the call site.)

You have a running machine: explicit states, named events and predictable transitions. The same model scales to workflows that cross process restarts, message brokers and teams.

import asyncio
from xstate_statemachine import Interpreter, create_machine


async def main() -> None:
    """Toggle a light with the asynchronous engine."""
    machine = create_machine(
        {
            "id": "toggle",
            "initial": "inactive",
            "states": {
                "inactive": {"on": {"TOGGLE": "active"}},
                "active": {"on": {"TOGGLE": "inactive"}},
            },
        }
    )
    light = await Interpreter(machine).start()
    try:
        receipt = await light.send("TOGGLE", wait=True)
        print(sorted(receipt.state_ids))  # ['toggle.active']
    finally:
        await light.stop()


asyncio.run(main())

Why choose xstate-statemachine?

The advantage is the combination: visual statecharts, native Python execution and operational building blocks around the same definition. See How it compares for specific trade-offs and reproducible benchmarks.

🚀 Install

pip install xstate-statemachine          # the core: zero runtime dependencies, Python 3.9 → 3.15
xsm info                                 # verify the install
xsm update                               # later: upgrade with the installer that installed you

Everything else is an extra you opt into — pip install "xstate-statemachine[fastapi]", [django], [redis], [kafka], [agents], [observability], [testing], … or [all]. The full table is under Integrations.

Match the docs to your version. This README covers the repository's 0.11.0 source version. The PyPI badge shows the published package, which may be older. Run xsm info to check your installation; use the source checkout below when evaluating features that are not in your installed release yet.

— validate machine JSON and keep generated code current in CI. Starting a service from scratch? xsm new my_service --template fastapi scaffolds one from a tested example app.

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

git clone https://github.com/basiltt/xstate-statemachine.git
cd xstate-statemachine
python -m pip install -e .
python -m xstate_statemachine info
# Add only the extras you need, for example:
python -m pip install -e ".[fastapi]"

pip install pypi-attestations
pypi-attestations verify pypi --repository https://github.com/basiltt/xstate-statemachine \
  pypi:xstate_statemachine-0.10.5-py3-none-any.whl   # example published wheel; choose your release

🧰 Real Problems This Solves

Start with a problem that looks like yours. The examples cover customer-facing APIs, background workflows, event-driven services and supervised AI agents. Each row links to a shipped example and its setup instructions; the associated suites exercise failure paths as well as the happy path.

Build an API with FastAPI orders or Django approvals. Connect services with event-driven fulfilment. Control an agent with the offline support bot.

Deployment matters. These are tested scenarios, not a promise of exactly-once external I/O. Use the documented store, locking, inbox and broker configuration; make payment calls and other external effects idempotent. The Guarantees page explains the crash windows explicitly.

🧠 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"},
}

Three bug classes the model helps prevent

Now scale that up. In a distributed system the "late API response" is a redelivered message, the "second click" is a retried webhook, and the "boolean soup" is a status column plus six feature flags plus a cron job that nobody fully understands. The same design addresses the same problems — when paired with durable persistence, input deduplication and reliable output delivery. That is what the rest of this library provides building blocks for.

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 Mental Model

Six concepts give you the foundation. The rest of the library builds on them.

The split that matters: state is where you are, context is what you know. retries is context. retrying is a state. Keeping that boundary clear makes workflows easier to understand and change.

_{🔗 Open this machine in the Stately editor — public, no account needed.}

Two more words you'll meet constantly: an invoke runs a service (an API call, a Celery task, a child machine) while a state is active and routes its result back as onDone / onError; an after is a timer that fires a transition if you are still in the state when it elapses. Everything else — nesting, parallel regions, history, actors, persistence, brokers — is those six things composed.

📨 Event-Driven Architecture: the main event

Events describe what happened. Statecharts decide what is allowed to happen next.

A checkout publishes OrderPaid. The warehouse reacts by packing the order. A shipping service reacts when packing completes. Each service owns its behavior and communicates through an event contract, rather than reaching into another service's internals.

This is why event-driven architecture matters: it lets workflows grow beyond one request, one process and one team while keeping the hand-offs explicit.

Why build around events?

You do not need microservices to benefit. Start with events inside one application; introduce a broker when independent consumers, buffering or cross-process delivery become useful. Event-driven systems introduce their own responsibilities: eventual consistency, duplicate delivery, ordering, schema evolution and operational monitoring. The library provides explicit tools for these concerns; it does not remove the need to design for them.

A statechart is a consumer (it reacts to events) and a producer (its transitions mean something to other services). Here is how the pieces fit together:

Producer → Broker → InboundDispatcher → Statechart
                           │                │
                    validate + dedup   guards + actions
                                            │
                             snapshot + inbox mark + outbox
                                  one shared transaction
                                            │
                                      OutboxRelay
                                            │
                                 Broker → Next consumer

The transaction shown above requires a shared transactional store and the documented locking strategy. The incoming delivery is acknowledged after the state commit; the relay publishes committed outbox records separately and may retry.

Reliability is part of the workflow

Two charts, one broker, no servers

The order-fulfilment chart below shows a richer workflow: payment, packing, shipping, escalation and cancellation, all visible in the model.

_{🔗 Open this machine in the Stately editor — public, no account needed.}

Start with a smaller, runnable version: an order chart and a warehouse chart that never call each other. Every hand-off is an event. The broker, store, inbox and outbox are all in-process, so you can try the full conversation without running a server. The demo's PACKED event completes the order directly; it does not invoke the richer chart's shipOrder service.

_{🔗 Open this machine in the Stately editor — public, no account needed.}

from xstate_statemachine import MachineLogic, assign, create_machine
from xstate_statemachine.eda import (
    Envelope,
    InboundDispatcher,
    MemoryOutboxStore,
    OutboxPlugin,
    OutboxRelay,
    SyncFakeBrokerAdapter,
)
from xstate_statemachine.persistence import MemoryInbox, MemoryStore

ORDER = {
    "id": "order",
    "initial": "placed",
    "context": {"total": 0},
    "states": {
        "placed": {
            "on": {
                "PAY": {
                    "target": "paid",
                    "guard": "hasTotal",
                    "actions": assign(
                        {"total": lambda a: a["event"].payload["total"]}
                    ),
                    "meta": {
                        "publish": {"type": "OrderPaid", "data": ["total"]}
                    },  # ← the contract
                }
            }
        },
        "paid": {"on": {"PACKED": "shipped"}},
        "shipped": {"type": "final"},
    },
}
WAREHOUSE = {
    "id": "warehouse",
    "initial": "waiting",
    "states": {
        "waiting": {
            "on": {
                "PACK": {
                    "target": "packed",
                    "meta": {"publish": "OrderPacked"},
                }
            }
        },
        "packed": {"type": "final"},
    },
}
order = create_machine(
    ORDER,
    logic=MachineLogic(
        guards={"hasTotal": lambda ctx, e: e.payload.get("total", 0) > 0}
    ),
)
warehouse = create_machine(WAREHOUSE)

store, inbox = (
    MemoryStore(),
    MemoryInbox(),
)  # in-memory demo, not crash-durable
outbox, broker = MemoryOutboxStore(), SyncFakeBrokerAdapter()
publisher = OutboxPlugin(
    outbox, topic="events"
)  # meta.publish → staged outbox record
relay = OutboxRelay(outbox, broker)  # committed outbox records → broker

ROUTES = {  # envelope type → (machine, the event it means to it)
    "xsm.order.PAY": (order, "PAY"),
    "OrderPaid": (
        warehouse,
        "PACK",
    ),  # the warehouse reacts to the order's event…
    "OrderPacked": (
        order,
        "PACKED",
    ),  # …and the order reacts to the warehouse's
}
dispatcher = InboundDispatcher(
    store,
    lambda t: ROUTES[t][0] if t in ROUTES else None,
    event_type=lambda env: ROUTES[env.type][1],
    key_for=lambda env, m: f"{m.id}:{env.subject}",  # one order id, two machines
    plugins=[publisher],
    inbox=inbox,  # dedup on envelope.id
)

# checkout publishes a command; nothing else is called directly
broker.publish(
    "events",
    Envelope.new(
        type="xsm.order.PAY",
        subject="order-1",
        data={"total": 42},
        source="checkout",
    ),
)

for _ in range(4):  # PAY → OrderPaid → PACK → OrderPacked → PACKED
    dispatcher.run_once_sync(broker, "events")
    relay.relay_once_sync()

import json

print(json.loads(store.load("order:order-1").snapshot)["value"])  # shipped
print([e.type for e in broker.published_on("events")])
# ['xsm.order.PAY', 'OrderPaid', 'OrderPacked']

Take the same design to production in three steps:

1. Make state and outgoing events durable. Use SQLiteStore("app.db"), SQLiteInbox(store) and SQLiteOutboxStore(store) with PessimisticLock() passed to InboundDispatcher. Sharing the store and transaction is what makes snapshot, inbox mark and outbox commit together. Replacing only MemoryStore is not enough.
2. Choose the broker and execution model. The sync loop can use SyncRedisStreamsBroker(...) or SyncSqsBroker(...). For RedisStreamsBroker(...), KafkaBroker(...), RabbitMQBroker(...), NatsBroker(...) or SqsBroker(...), use the async dispatcher and relay APIs. The machine definitions and event contracts stay the same; connection settings, queue setup and ordering requirements are broker-specific.
3. Exercise redelivery and recovery. Publish the same envelope twice: with the inbox configured, the second delivery is a duplicate and runs no actions. Configure a dead-letter store and max_attempts=3, then inject a delivery-processing failure: an exhausted delivery becomes a DLQ record. An invoked service such as shipOrder instead reports its failure through onError; that alone does not make a broker delivery fail.

Delivery contract: outbox delivery is at least once, not exactly once. Keep external effects idempotent. Per-subject order needs the broker topology described in the guide; SQS Standard does not preserve it. Kafka has no broker-side delivery count, so the dispatcher's poison-attempt counter does not survive a consumer restart there.

What's in the box

The complete worked example — two charts, five brokers behind one flag, Celery in eager mode, Prometheus + OpenTelemetry + the live inspector, a poison message, a crashed consumer — is examples/integrations/eda_fulfilment:

pip install "xstate-statemachine[redis,celery,observability]" "fakeredis[lua]" opentelemetry-sdk
cd examples/integrations && python -m eda_fulfilment --broker redis-streams

Guides → Event-driven architecture · Broker adapters · Celery · Guarantees (the exact order of save, outbox write, inbox mark and publish, and what a crash between any two of them means).

🏭 Automatic Code Generation

The chart already declares every action, guard, service and delay it needs, every event it accepts and every state it can be in. Typing that twice is where the bugs come from. So don't:

xsm generate-template subscription.json -t pythonic-class --with-tests --with-types -o ./billing

OK Generated logic file: billing/subscription_logic.py
OK Generated runner file: billing/subscription_runner.py
OK Generated pytest file: billing/test_subscription.py
OK Generated typed file: billing/subscription_types.py

_{🔗 Open this machine in the Stately editor — public, no account needed.
The input: the Stripe-subscription chart from the recipes.}

What came out — abridged, but these are the real files:

class SubscriptionMachine(StateMachine):
    machine_id = "subscription"
    initial_context = {"failures": 0, "last_invoice": None}

    incomplete = State(
        "incomplete",
        initial=True,
        on={
            "PAYMENT_SUCCEEDED": {
                "target": "active",
                "actions": ["recordPayment"],
            },
            "CANCELED": "canceled",
        },
    )
    active = State("active", on={...}, tags=["billable"])
    past_due = State("past_due", on={...}, tags=["billable", "dunning"])
    canceled = State("canceled", tags=["terminal"])

    @action
    def record_payment(self, interpreter, context, event, action_def) -> None:
        # TODO: implement action logic
        ...

def test_step_01_PAYMENT_SUCCEEDED(interp, clock):
    interp.send("PAYMENT_SUCCEEDED")
    assert sorted(interp.current_state_ids) == ["subscription.active"]


def test_every_declared_event_is_known(machine): ...
def test_snapshot_round_trip(interp, clock, machine): ...

EventType = Literal["CANCELED", "PAYMENT_FAILED", "PAYMENT_SUCCEEDED"]
StateId = Literal["subscription.active", "subscription.canceled", ...]


class Context(TypedDict): ...

The generator checks structural equivalence before writing. 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.

This verifies the generated machine's structure, not the business correctness of the implementations you add. Recorded tests start with stub logic; extend them with your own guard, action and service scenarios.

Three more things fall out of the same idea — the chart knows itself:

xsm generate-template subscription.json -t pythonic-class -o ./billing --check   # CI: has the code drifted from the JSON?
xsm docs machines/*.json -o docs/           # a Markdown reference page per machine
xsm diagram subscription.json -f mermaid    # or plantuml / ascii — no graphviz
xsm paths subscription.json                 # one path to every reachable configuration

…and the [testing] pytest plugin turns that last list into a parametrised xsm_path fixture, with pytest --xsm-coverage telling you which states and transitions your suite never reached. Django projects get all of this as manage.py xsm_*.

Full reference → CLI · Templates deep dive · Hierarchical generation (parent + child charts in one go).

🔗 One JSON, Two Runtimes

Your frontend team models a checkout flow in Stately. 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 and backend can share the same workflow contract instead of reimplementing the state graph independently. Keep implementations and versions aligned, and continue to enforce authorization and business rules on the backend.

The statechart screenshots in this README come from the Stately editor — the source configs live in docs/assets/images/machines/machines.json. Examples show the corresponding workflow or an explicitly simplified subset. xsm diagram and machine.to_mermaid() render a machine without leaving the terminal. The round trip, the version key, meta conventions and the editor JSON Schema are in Stately editor → Python.

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.

💾 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"].

Keys that start with _xsm_ (PRIVATE_CONTEXT_PREFIX) are private context: they round-trip through snapshots but are stripped by public_context() and never leave the process through the web adapters. Want typed context? [pydantic] validates it on every assign — see Integrations.

🛡️ Guards — Conditional Transitions

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

_{🔗 Open this machine in the Stately editor — public, no account needed.}

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. A guard that raises is treated as False by default (guardErrorPolicy lets you change that), and a guard that is declared but never implemented fails loudly at create_machine().

🎬 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 by default. The transition still completes and the interpreter keeps running; one buggy side effect can't take down a long-lived machine. actionErrorPolicy: "rollback" or "fail" changes that per machine — see Failure semantics.

"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 invocation and discards stale completion events. Cancellation is cooperative; it cannot forcibly stop a blocking thread or undo an external request
• A plain-def service under the async engine runs on a private thread pool (service_pool_size, or your service_executor) so a blocking call cannot stall the loop

Beyond callables, src can be actor logic: from_callback (push events in from a socket or a thread), from_iterator / from_async_iterator (each item becomes a StreamEvent), from_coroutine, from_interpreter (a whole child machine), or a Celery task via celery_service.

Tip — use wait_for (async) or wait_for_sync rather than asyncio.sleep() guesses. It waits on a predicate with a real timeout, so tests do not depend on guessed delays.

⏱️ 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"}}

Three things make timers production-grade here rather than a threading.Timer per state:

• No OS thread per timer. The async engine fires after through a priority lane the run loop checks ahead of its inbox; the sync engine fires a due timer on the caller's thread inside send() or tick().
• Virtual time in tests. SimulatedClock().increment(30_000) fires a 30-second after instantly.
• Durable across restarts. Deadlines persist as wall-clock instants; persisted() resumes the remaining time and DueTimerScanner wakes a machine whose deadline passed while nothing was running — a 7-day dunning timer survives every worker dying. See Persistence.

🌳 Nested & Parallel States

Nested (compound) states

Group related substates so shared transitions live in one place:

_{🔗 Open this machine in the Stately editor — public, no account needed.}

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

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:

_{🔗 Open this machine in the Stately editor — public, no account needed.}

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":

_{🔗 Open this machine in the Stately editor — public, no account needed.}

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

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. Await it with to_promise(interp) or interp.wait_done().

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

_{🔗 Open this machine in the Stately editor — public, no account needed.}

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": "pool",
                        "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. Spawned actors inherit the parent's plugins and are captured in its snapshot.

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

💾 Persistence — Snapshots, Stores and Durable Timers

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, pending events (with their lane), deferred events, armed delayed sends, history, child actors, after deadlines and systemId registrations all round-trip. Every snapshot carries an envelope (version — layout 4 — machine_id, machine_hash, machine_version), so from_snapshot() refuses a structurally different machine with SnapshotDriftError instead of silently resuming into it; a newer layout than this release understands is SnapshotVersionError, never a guess.

The safe loop: load → act → persist → discard

Most apps never call from_snapshot by hand. xstate_statemachine.persistence ships the whole loop with built-in stores — MemoryStore, FileStore (atomic writes, 0600), SQLiteStore (WAL, stdlib sqlite3) — and RedisStore, SQLAlchemyStore / AsyncSQLAlchemyStore and DjangoStore in their extras:

from xstate_statemachine import MachineLogic, create_machine
from xstate_statemachine.persistence import MemoryStore, persisted

cfg = {
    "id": "order",
    "initial": "cart",
    "context": {"items": 0},
    "states": {
        "cart": {"on": {"ADD": {"actions": "add"}, "PAY": "paid"}},
        "paid": {"type": "final"},
    },
}
machine = create_machine(
    cfg,
    logic=MachineLogic(
        actions={
            "add": lambda i, ctx, e, a: ctx.__setitem__(
                "items", ctx["items"] + 1
            )
        }
    ),
)
store = MemoryStore()  # or SQLiteStore("orders.db"), RedisStore(...)

with persisted(
    store, "order:42", machine
) as order:  # loaded (or created), started
    order.send("ADD")
    order.send("ADD")
# saved with expected_version on clean exit; a concurrent writer gets ConflictError

with persisted(store, "order:42", machine) as order:
    assert order.context["items"] == 2
    assert order.send("PAY", wait=True).changed
assert store.load("order:42").version == 2

apersisted() is the async twin; persisted_retry() reloads and re-applies on ConflictError. What the loop gives you, all documented on the Guarantees page:

The honest boundary: the persistence layer's exactly-once transition guarantee applies to idempotency-keyed events while deduplication evidence is retained, using the documented store and locking setup. Actions' external side effects and outbox delivery are at-least-once, and a memory-only inbox is not durable retention. Use invoke to model I/O, use the transactional outbox for outgoing messages, and make external effects idempotent. Full guide → Persistence & Durability.

🔌 Integrations — Optional Extras

The core stays zero-dependency; every integration is an extra under xstate_statemachine.contrib that you install explicitly (pip install "xstate-statemachine[fastapi]"). Each has a guide page with a Guarantees box and a Threat model box — CI refuses a page without them — and every framework version it claims is proven by a CI cell on the Compatibility page.

from fastapi import FastAPI
from xstate_statemachine.contrib.fastapi import (
    StatechartRouter,
    instrument_app,
)
from xstate_statemachine.contrib.starlette import StatechartRegistry
from xstate_statemachine.persistence import SQLiteStore

# … order_machine, AddItem/Pay/Cancel event models and my_authorizer defined elsewhere
registry = StatechartRegistry(
    SQLiteStore(DB_PATH), run_timers=True
)  # or RedisStore for many hosts
registry.register(
    "orders", order_machine, authorize=my_authorizer
)  # closed by default

app = FastAPI()
instrument_app(app, registry)  # lifespan, /_xsm/health, /_xsm/ready
app.include_router(
    StatechartRouter(registry, "orders", event_models=[AddItem, Pay, Cancel])
)
# POST /orders/42/events/PAY  →  200 {state, state_ids, changed, available_events, …}
#                            →  409 when a business rule (guard) refuses, 422 on a bad body
#                            →  503 when the store is unavailable

Every web integration is closed by default — authorize= is required, GET returns state only unless you opt into a context_serializer, and error bodies never carry exception text. The multi-worker model (why an interpreter cannot live in a uvicorn worker, where timers run, how 4 workers × 200 concurrent PAY yields exactly one success) is the FastAPI guide's first section.

Six runnable, README-driven apps live under examples/integrations/ — fastapi_orders (N uvicorn workers, a 200-payment stress test, gateway-outage and rolling-upgrade scenarios), django_approvals (parallel Legal/Finance review, roles, admin, DRF, Channels), flask_wizard, sqlalchemy_orders, eda_fulfilment and agents_support_bot. CI installs each one into a fresh venv from its own pip install line.

🤖 For LLM agents

An agent loop is a statechart the model does not get to rewrite. With pip install "xstate-statemachine[agents]" the model becomes one invoked service that proposes text or tool calls, and the chart decides what happens next. This is the reference TOOL_LOOP chart the extra ships — every budget check, timeout, denial and human-approval wait is a state you can see, test, persist and diagram:

_{🔗 Open this machine in the Stately editor — public, no account needed.}

• Per-state tool allow-lists (meta.tools) are re-checked inside run_tool — editing the guard out of the chart does not let a forbidden tool through.
• Budgets are guards: turns, tokens and cost are context the chart refuses to exceed. BudgetPlugin rolls them up across a tree of spawned agents.
• Timeouts are after deadlines, retried with backoff through timed_out.
• Human approval is a durable state. awaiting_human persists with the exact pending call ids and an escalation deadline; the approval event must name the call it approves.
• Structured output is validated (pydantic) with bounded retries; a bad answer is a state, not an exception in your handler.
• Multi-agent: spawn_agent runs children as actors with per-child and global budgets; supervisor, pipeline and debate reference charts ship alongside TOOL_LOOP.
• Offline by default: FakeModel scripts the model's answers, so the whole loop runs in CI with no key. OpenAI and Anthropic providers are soft imports; it interoperates with LangGraph and pydantic-ai.

# pip install "xstate-statemachine[agents]"   # … the extra's only hard dependency is pydantic
import asyncio
from xstate_statemachine.contrib.agents import (
    FakeModel,
    run_agent,
    tool_registry,
)


def get_weather(city: str) -> str:
    """Current weather for a city."""
    return f"sunny in {city}"


model = FakeModel(
    [{"tool": "get_weather", "args": {"city": "Kochi"}}, {"text": "Sunny."}]
)
res = asyncio.run(
    run_agent(
        model, tools=tool_registry(get_weather), prompt="Weather?", max_turns=5
    )
)
assert res.final_state == "toolLoop.done" and res.usage["turns"] == 2

AgentTracePlugin writes JSONL traces with gen_ai.* attributes and secret scrubbing. The two-minute offline walkthrough is examples/integrations/agents_support_bot (python run.py --fake --prompt "refund order 42" — one refund, never two). Honest comparisons with LangGraph, Burr and @statelyai/agent include checked-version notes and scheduled drift checks.

🔭 Observability & Inspection

Three ways to see inside a running machine, from zero-dependency to full APM.

Built in — LoggingInspector prints every event, transition and action; InspectorPlugin speaks the @statelyai/inspect protocol so xsm inspect --live (or an SseSink in your app) streams any machine into the Stately Inspector in your browser; JsonLinesSink records a session and xsm replay plays it back, live or as a transcript.

[observability] — one line:

from xstate_statemachine.contrib.observability import instrument_all

# OpenTelemetry spans per transition/service, Prometheus counters and histograms,
# structlog / loguru context vars, Sentry breadcrumbs — whichever SDKs are installed
instrument_all(interpreter)  # … any running Interpreter / SyncInterpreter

Metric labels are an allow-list by default (state and event names, never payloads), and every sink runs the same redact() denylist as the stores. For agents, AgentTracePlugin writes JSONL traces with gen_ai.* attributes and secret scrubbing.

Programmatic — subscribe(), on(), send(wait=True) → Receipt, plugin hooks for every lifecycle moment, and the pure API below. Guide → Observability · Live inspector.

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

The graph module goes further: reachable_states(), shortest_paths(), simple_paths() and transition_coverage_targets() enumerate a machine — which is what xsm paths prints and what the [testing] plugin's xsm_path fixture walks, one engine-verified path per test.

🔍 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 named event outward. Subscribe to it by name:

ed.on("saved", lambda ev: print("saved!", ev.type))

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

register_global(plugin) attaches a plugin to every interpreter constructed afterwards — including spawned children and from_snapshot() restores. Third-party packages can publish plugins, stores and brokers under the xstate_statemachine.plugins / .brokers entry-point groups; xsm plugins lists what is installed and attach_discovered(interp, allow=[...]) loads only the distributions you name (it never runs an unmarked callable — see Security).

⚖️ How It Compares

Choose for the workflow you need, not the longest feature list. The distinctive combination here is XState JSON portability, Python statechart execution and optional persistence, web and event-driven integrations around the same model.

The seven detailed comparisons include source notes and checked versions. The repository also runs scheduled release/API drift checks; those are not a claim to test every feature of every competing product on every release: transitions · python-statemachine · django-fsm · LangGraph · Burr · @statelyai/agent · AWS Step Functions.

Python state-machine libraries

The feature comparison is scoped to the APIs recorded in the comparison source notes, including transitions 0.9.3 and python-statemachine 3.2.1. Application code and third-party extensions can add capabilities beyond those listed. See the projects' own documentation for their current surface: transitions and python-statemachine.

Pick transitions if you want a mature, flexible FSM attached to an existing class, with extensions for hierarchy, async execution and diagrams.

Pick python-statemachine if you want a beautiful, pythonic declarative API and don't need XState JSON interop. Its compound, parallel and history states, invoked services and diagram tooling make it a substantive alternative.

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

Beyond a state-machine library

Speed

Same machine shape, each library through its own idiomatic API, all measured in one session on 0.11.0 (October 11, 2026, Python 3.14, median of 7 runs, GC disabled, setup excluded except for the construction scenario). These are a recorded benchmark snapshot, not live measurements or a throughput guarantee. Events per second; bold is fastest in the row.

In this recorded run, transitions leads the flat scenario by approximately 1.8×. xstate-statemachine leads the two measured nested/parallel scenarios by 3.7–4.5× over the next library. Construction and the 1,000-instance scenario are 14% and 10% ahead of transitions, while retaining build-time validation.

These are workload-specific microbenchmarks. They do not measure database writes, broker delivery, network calls or model latency, and they do not establish a universal ranking. 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. Around five states, a sketch may be enough; at twenty interacting states, an explicit model can be much easier to reason about. State count is only a heuristic: concurrency, timeouts, cancellation and recovery paths are better reasons to adopt a statechart. Use a managed workflow service when managed operations are the requirement, and a task queue when all you need is independent background jobs.

📚 Cookbook

Real problems, small solutions.

🍳 Full recipes, each a runnable folder under examples/recipes/ and tested in CI — hostile inputs included: Stripe webhooks (HMAC verification, event.id as the idempotency key, 1,000 deliveries with forged / stale / tampered ones among them), APScheduler durable timers, RQ / arq / Dramatiq workers with ConflictError retry, a Streamlit / Gradio wizard, chatbot slot filling, feature-flag rollout on a SimulatedClock, WebSocket reconnect with RetryPolicy backoff, and circuit breaker + retry for an HTTP client: Recipes →

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:

🧾 A whole machine, end to end

_{🔗 Open this machine in the Stately editor — public, no account needed.}

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

Note what is absent: no try/except around the charge, no is_charging flag, no "did we already ship?" check. A declined card is an onError edge and "cart is empty" is a guard. A repeated CHECKOUT cannot start another local charge: shipped is final and charging has no CHECKOUT handler. For a real payment provider, also use durable persistence and provider-side idempotency to handle retries across failures.

🔁 Retry with exponential backoff and a give-up limit

_{🔗 Open this machine in the Stately editor — public, no account needed.}

The pattern that turns into unreadable nested loops when hand-written:

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},
)

Attempt counting, backoff math, and the give-up condition are each in exactly one place. xstate_statemachine.patterns.RetryPolicy packages this (with full / equal jitter and a max_ms cap) as ready-made guards and delays.

"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, a validator, an inspector, a live simulator, a diagram/docs exporter and an operations console 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 (--plain, --json, NO_COLOR).

xsm                                                           # interactive launcher
xsm new my_service --template fastapi                         # scaffold a project from an example app
xsm gt checkout.json -t pythonic-class --with-tests --with-types -o ./app
xsm validate machines/*.json                                  # build each file with the real library
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 inspect checkout.json --live                              # stream to the Stately Inspector
xsm diagram checkout.json -f mermaid -o docs/
xsm docs machines/*.json -o docs/
xsm snapshots sqlite:///orders.db --stale --fail-if-stale     # written by another machine version?
xsm dlq --dlq sqlite:///dlq.db list                           # dead letters; `replay` is a dry run by default
xsm asyncapi checkout.json -o asyncapi.json

Code generation (generate-template, the companion templates and the proof-before-write check) has its own section.

Full reference: CLI Tool · Templates deep dive · Hierarchical generation. Django projects get the same commands as manage.py xsm_*.

🏭 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

Async interpreters attached to the same event loop share its thread and execution budget. Adding more machines does not create more CPU capacity. The published reference measurements report roughly 30k trivial events/s on a laptop, and approximately 45 ms after lateness at 500 busy machines, compared with approximately 180 ms before 0.8.0. These are historical workload measurements, not a deployment SLA.

after timers use a priority lane ahead of the async inbox. A sync timer runs on the owner's thread when it calls send() or tick(). Neither engine creates an OS thread per timer. Other work can use threads: blocking def services in the async engine run in its service executor, and non-blocking spawned work has engine-specific delivery.

Pass Interpreter(machine, clock=SimulatedClock()) in tests to advance time without sleeping. For measured tables, the benchmark runner and a sizing rule, read Production Characteristics before deploying, then measure your own machine shapes, handlers and persistence layer.

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. Blocking services on the async engine use its service executor. A SyncInterpreter remains caller-driven; use send_threadsafe() for cross-thread producers and drain its mailbox with send() or tick().

Long-running machines

• Persist at settled transition boundaries. Prefer the persisted() / apersisted() loop and a durable store for the documented locking and crash-recovery behavior. get_persisted_snapshot() in a subscribe() callback gives you a snapshot for custom storage; the callback alone does not make the write atomic or durable.
• after timers across a restore. A snapshot carries each pending timer's wall-clock deadline. The persisted() loop and the web registries resume the remaining time (restart_timers="resume"), and DueTimerScanner fires a deadline that matured while nothing was running. A bare from_snapshot() is a static rebuild: has_dormant_timers is True until you pass restart_timers="resume" (or True to re-arm from zero). 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 4 — machine_id, machine_hash, machine_version) 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=4, 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")

🔐 Security & Trust Model

The short version, from SECURITY.md: installed packages are trusted, your machine definition is trusted, events and snapshots are not. Everything that crosses that line is validated.

The numbered baseline every integration is built against, each item mapped to the test that proves it: Security.

📘 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