statem

A minimal async state machine engine for Python, built on Pydantic. The state graph is a validated dict; guards and actions are plain Python (sync or async) functions, registered by name.

Full documentation: https://rahuldas-dev.github.io/statem/

Install · Quickstart · Hooks · Visualizing · Why it's built this way · Development · License

Install

pip install statem

The importable package is statem:

from statem import StateMachine, Signal

Quickstart

import asyncio
from statem import Signal, StateMachine

config = {
    "idle": {"on": {"START": {"target": "running", "guard": "can_start"}}},
    "running": {"on": {"STOP": "idle"}},
}


def can_start(ctx, signal) -> bool:
    return True


async def main() -> None:
    machine = StateMachine.from_dict(config, guard_dict={"can_start": can_start})
    state = await machine.run(state_name="idle", events=Signal(event="START"), session={})
    print(state)  # "running"


asyncio.run(main())

See examples/bread.py for a fuller example (guards, actions, error_state, and an always-transition), or examples/bank.py for a richer one that exercises every hook in one conversation, including error_state recovering from a real exception. The guide covers the full config shape.

Hooks

Four lifecycle hooks can carry guards and/or actions on any state:

A fifth field, error_state, isn't a hook itself but a per-state escape hatch: if an on-transition's action raises, the engine catches it and — if error_state is set — transitions there instead of propagating.

Visualizing

to_mermaid(machine, initial=None) renders machine.config as a Mermaid stateDiagram-v2 string — no extra dependency, just text. GitHub, MkDocs, VS Code, and Jupyter all render it natively:

from statem import to_mermaid

print(to_mermaid(machine, initial="idle"))

stateDiagram-v2
    [*] --> idle
    idle --> running: START [can_start]
    running --> idle: STOP

See the guide for guard chains, always, and error_state rendering.

Why it's built this way

• Data and behavior stay apart. config is a plain, JSON/YAML-friendly dict — author it by hand, generate it, or load it from a file/database/LLM. Guards and actions are ordinary functions registered by name, independent of the graph.
• Fails fast. Bad transition targets and unregistered guard/action names raise at construction time, not three hops into a live run.
• Sync or async, your choice. The engine detects and awaits either correctly — no need to wrap trivial sync logic in async def.
• session is yours. Any shape you want (dict, dataclass, ORM row) — the engine never inspects it, just threads it through to your guards/actions via ctx.session.
• No subclassing. StateMachine.from_dict(...) + await machine.run(...). No framework lifecycle to inherit into.
• Every run is traceable. Each guard/action fired is recorded in order; a run_id (yours, or auto-generated) ties every log line from one run together.

Development

uv sync
uv run python -m unittest discover -s tests
uv run coverage run -m unittest discover -s tests && uv run coverage report --fail-under=100
uv run ruff check tests statem

License

MIT — see LICENSE.