API Reference

Complete class, function, and decorator reference for XState-StateMachine.

Complete reference for every public class, function, decorator, and exception exported by xstate_statemachine. All items listed here are available as top-level imports:

from xstate_statemachine import create_machine, Interpreter, State  # etc.

Factory Functions

create_machine(config, *, context_type=None, logic=None, logic_modules=None, logic_providers=None, strict_targets=True, event_schemas=None, strict_config=None, context_validator=None)

Creates, validates, and assembles a state machine instance from an XState-compatible JSON configuration dictionary. This is the primary entry point when working with JSON/dict-based machine definitions.

The function intelligently resolves business logic (actions, guards, services) from whichever source you provide. If an explicit MachineLogic instance is passed via logic, it takes precedence. Otherwise, the factory delegates to LogicLoader to auto-discover implementations from the specified modules or provider objects.

create_machine() also validates the fully built tree: every transition target must resolve, and no always self-target may be a permanent dead end.

Returns: MachineNode – a fully constructed, validated machine ready for an interpreter.

Raises:

Machine-level policy keys

These keys go at the root of the JSON config (alongside "id" and "states"). Every one preserves 0.7.x behavior at its default; nothing changes on upgrade unless you opt in.

Example 1 – Minimal (no logic)

from xstate_statemachine import create_machine

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

Example 2 – With explicit MachineLogic

from xstate_statemachine import create_machine, MachineLogic


def log_toggle(interpreter, context, event, action_def):
    print(f"Toggled! Count: {context['count']}")


logic = MachineLogic(
    actions={
        "logToggle": log_toggle,
    }
)

config = {
    "id": "toggle",
    "initial": "off",
    "states": {
        "off": {
            "on": {
                "FLIP": {
                    "target": "on",
                    "actions": "logToggle",
                },
            },
        },
        "on": {
            "on": {
                "FLIP": {
                    "target": "off",
                    "actions": "logToggle",
                },
            },
        },
    },
}
machine = create_machine(config, logic=logic)

Example 3 – With module-based auto-discovery

# file: my_logic.py
def log_toggle(interpreter, context, event, action_def):
    context["count"] += 1

# file: main.py
import my_logic
from xstate_statemachine import create_machine

machine = create_machine(config, logic_modules=[my_logic])
# or by string path:
machine = create_machine(config, logic_modules=["my_logic"])

Example 4 – With provider objects

from xstate_statemachine import create_machine

class ToggleLogic:
    def log_toggle(self, interpreter, context, event, action_def):
        context["count"] += 1

machine = create_machine(config, logic_providers=[ToggleLogic()])

build_machine(*, id, states, transitions=None, actions=None, guards=None, services=None, context=None, root=None)

Builds a state machine entirely from Python objects using the functional Pythonic API. This is the simplest style – define State objects, Transition objects, and decorated callables at module level, then call this function to produce a MachineNode.

Internally, build_machine compiles the provided objects into a JSON config dict and a MachineLogic instance, then delegates to create_machine().

Returns: MachineNode

Raises: InvalidConfigError on duplicate state names, missing initial state, or invalid transition sources.

from xstate_statemachine import State, build_machine, action

idle = State("idle", initial=True)
running = State("running")
done = State("done", final=True)


@action
def log_start(interpreter, context, event, action_def):
    print("Machine started!")


machine = build_machine(
    id="workflow",
    states=[idle, running, done],
    transitions=[
        idle.to(running, event="START", actions=["logStart"]),
        running.to(done, event="FINISH"),
    ],
    actions=[log_start],
    context={
        "step": 0,
    },
)

Core Classes

State

State(
    name: str = "",
    *,
    initial: bool = False,
    final: bool = False,
    parallel: bool = False,
    history: Optional[str] = None,
    on: Optional[Dict[str, Any]] = None,
    entry: Optional[List[str]] = None,
    exit: Optional[List[str]] = None,
    after: Optional[Dict[Union[int, str], Any]] = None,
    invoke: Optional[Union[Dict, List]] = None,
    on_done: Optional[Union[str, Dict]] = None,
    always: Optional[Union[str, Dict, List]] = None,
    context: Optional[Dict] = None,
    states: Optional[List[State]] = None,
    tags: Optional[List[str]] = None,
    meta: Optional[Dict[str, Any]] = None,
)

A state definition for use across all three Pythonic API styles (functional, builder, and class-based).

Raises: InvalidConfigError if both final=True and parallel=True.

Key Methods

state.to(target, *, event, guard=None, actions=None, reenter=False)

Creates an external transition from this state to a target state.

state.to(
    target: State,
    *,
    event: Optional[str] = None,    # REQUIRED -- event name
    guard: Optional[str] = None,    # guard function name
    actions: Optional[List[str]] = None,  # action names to run
    reenter: bool = False,          # force exit/re-entry on self-transitions
) -> Transition

Returns: Transition

Raises: InvalidConfigError if event is not provided.

idle = State("idle", initial=True)
active = State("active")

t = idle.to(active, event="ACTIVATE", actions=["logActivation"])

state.internal(event, *, guard=None, actions=None)

Creates an internal transition – actions execute but no state change occurs. Entry and exit actions of the current state are not run.

state.internal(
    event: str,
    *,
    guard: Optional[str] = None,
    actions: Optional[List[str]] = None,
) -> Transition

Returns: Transition with internal=True.

counter = State("counter", initial=True)
t = counter.internal("INCREMENT", actions=["addOne"])

@state.enter – Entry action decorator

Registers a function as an entry action for this state. Only valid inside a StateMachine class definition. The function name is auto-converted from snake_case to camelCase.

class MyMachine(StateMachine):
    idle = State(initial=True)

    @idle.enter
    def on_idle_entered(self, interpreter, context, event, action_def):
        print("Entered idle!")

Returns: The original function with _xsm_type="action" and _xsm_name markers attached.

@state.exit – Exit action decorator

Registers a function as an exit action for this state. Same semantics as @state.enter.

class MyMachine(StateMachine):
    active = State()

    @active.exit
    def on_active_exited(self, interpreter, context, event, action_def):
        print("Left active!")

Properties

__init_subclass__ keyword behavior

State supports class-inheritance keywords for concise nested-state definitions inside StateMachine subclasses:

class MyMachine(StateMachine):
    class loading(State, initial=True):
        """A compound initial state defined as a class."""
        fetching = State(initial=True)
        parsing  = State()

    class done(State, final=True):
        """A final state."""
        pass

    class regions(State, parallel=True):
        """A parallel state."""
        pass

The keywords initial, final, and parallel are captured by __init_subclass__ and stored as _xsm_initial, _xsm_final, and _xsm_parallel class attributes, respectively.

StateMachine

Base class for defining state machines using Python class syntax. Uses a custom metaclass (_StateMachineMeta) that collects State attributes, Transition/TransitionGroup attributes, and @action/@guard/@service decorated methods at class-definition time.

Class Attributes

Events, hooks and argument injection [0.12.0]

• A transition’s attribute name is its event: cycle = green.to(yellow) is CYCLE, slow_down = ... is SLOW_DOWN. event= overrides. Transitions declared inside nested state classes are exposed as event methods on the machine (order.ship()).
• Naming-convention hooks are bound automatically: on_enter_<state>, on_exit_<state>, on_<event> (runs between exit and entry), before_<event> / after_<event> (run before exit / after entry; instance façade only), and the generic on_enter_state(state), on_exit_state(state), on_transition(event, source, target), before_transition, after_transition. A suffix that matches no state or event logs a warning.
• Every callback declares only the arguments it wants. Injectable names: context/ctx, event, event_type, interpreter/machine, action_def/action, error, data, input, source, target, state, any payload field of the event, and **kwargs for the remaining payload. The legacy positional shapes — (interpreter, context, event, action_def) for actions, (context, event) for guards, (interpreter, context, event) for services — are detected by arity and called unchanged.
• Nested classes accept every State(...) keyword (class paid(State.Compound, on_done="fulfil", after={...}): ...) and are valid transition sources and targets. State.Compound and State.Parallel are aliases of State.
• Reserved attribute names (raise InvalidConfigError if used for a state or transition): send, state, configuration, context, status, can, matches, allowed_events, machine, interpreter, engine, strict, plugins, validate_logic, create_machine, machine_id, initial_context, machine_root.
• Lifecycle names are not reserved, and the lifecycle always wins. A transition may be called start or stop (timers, jobs and media players need those events, and Stately exports use them). The chart keeps the event, but the attribute gets no event-method sugar: sm.start() / sm.stop() always start and stop the machine, and you send the event with sm.send("STOP"). A one-line warning is logged when the class is defined so the collision is visible at import time, never a surprise at run time. Rationale: event methods are new in this release (nothing depended on sm.stop() sending an event), whereas letting the event win would turn a documented shutdown into a machine that silently keeps running.

Instance API [0.12.0]

Instantiating the class builds the machine, binds callbacks to the instance (self is yours to use), and on the sync engine starts it.

StateMachine(context=None, *, engine=None, strict=None, plugins=None)

from xstate_statemachine import State, StateMachine


class Light(StateMachine):
    green = State(initial=True)
    yellow = State()
    cycle = green.to(yellow) | yellow.to(green)

    def on_enter_yellow(self, event):
        print("slow down after", event.type)

    def on_cycle(self, amount=1):
        return amount * 2


with Light() as light:  # stopped on exit; `async with` for the asyncio engine
    print(light.cycle(amount=4))  # slow down after CYCLE / 8
    print(light.state, light.yellow.is_active)  # yellow True

Class Method

StateMachine.create_machine(context=None) -> MachineNode

Compiles the class definition into a JSON config and MachineLogic, then delegates to create_machine().

Returns: MachineNode

from xstate_statemachine import StateMachine, State, action


class TrafficLight(StateMachine):
    machine_id = "trafficLight"
    initial_context = {
        "cycle_count": 0,
    }

    red = State(initial=True)
    yellow = State()
    green = State()

    cycle = (
        red.to(green, event="NEXT")
        | green.to(yellow, event="NEXT")
        | yellow.to(red, event="NEXT", actions=["countCycle"])
    )

    @action
    def count_cycle(self, interpreter, context, event, action_def):
        context["cycle_count"] += 1


machine = TrafficLight.create_machine()

# With a context override:
machine = TrafficLight.create_machine(
    context={
        "cycle_count": 10,
    }
)

MachineBuilder

MachineBuilder(machine_id: str)

Fluent builder for constructing state machines step-by-step using method chaining. Every mutating method returns self, enabling a fluent pipeline that terminates with .build().

Methods

.state() full parameters

.state(
    name: str,
    *,
    initial: bool = False,
    final: bool = False,
    parallel: bool = False,
    on: Optional[Dict] = None,
    entry: Optional[List] = None,
    exit: Optional[List] = None,
    after: Optional[Dict] = None,
    invoke: Optional[Union[Dict, List]] = None,
    on_done: Optional[Union[str, Dict]] = None,
    always: Optional[Union[str, Dict, List]] = None,
    history: Optional[str] = None,
    tags: Optional[List[str]] = None,
    meta: Optional[Dict[str, Any]] = None,
) -> MachineBuilder

Raises: InvalidConfigError if the state name already exists, or if both final and parallel are True.

.transition() full parameters

.transition(
    source: str,
    event: str,
    target: str,
    *,
    guard: Optional[str] = None,
    actions: Optional[List[str]] = None,
    reenter: bool = False,
    internal: bool = False,
) -> MachineBuilder

.child_states() full parameters

.child_states(
    parent: str,
    *,
    initial: Optional[str] = None,
    states: Optional[Dict[str, Dict]] = None,
    parallel: bool = False,
) -> MachineBuilder

Raises: InvalidConfigError if parent is not an already-defined state.

.build() full parameters

.build(context: Optional[Dict] = None) -> MachineNode

Optional context overrides the context set by .context(). Deep-copies internal state, so repeated .build() calls are safe.

Raises: InvalidConfigError if no initial state is defined (for non-parallel, multi-state machines) or if a transition source is invalid.

Chaining example

from xstate_statemachine import MachineBuilder


def log_it(interpreter, context, event, action_def):
    context["count"] += 1
    print(f"Count: {context['count']}")


machine = (
    MachineBuilder("counter")
    .context(
        {
            "count": 0,
        }
    )
    .state("idle", initial=True)
    .state("counting")
    .state("done", final=True)
    .transition("idle", "START", "counting")
    .transition(
        "counting", "INCREMENT", "counting", actions=["logIt"], reenter=True
    )
    .transition("counting", "FINISH", "done")
    .action("logIt", log_it)
    .build()
)

Transition

Represents a transition between two states, triggered by an event. Created by State.to(), State.internal(), or the standalone transition() function. Not typically instantiated directly by users.

Attributes

| operator

Combine multiple transitions into a TransitionGroup using the pipe operator. The first transition whose guard passes wins (evaluated in order).

idle = State("idle", initial=True)
premium = State("premium")
basic = State("basic")

# Multiple guarded transitions for the same event
signup = (
    idle.to(premium, event="SIGNUP", guard="isPremium")
    | idle.to(basic, event="SIGNUP")
)
# signup is a TransitionGroup with 2 transitions

TransitionGroup

A collection of Transition objects created implicitly by the | operator. Users never instantiate this directly.

Attributes

| operator

TransitionGroup objects can be further combined with other Transition or TransitionGroup objects:

group_a = idle.to(s1, event="GO", guard="isA") | idle.to(s2, event="GO", guard="isB")
group_b = idle.to(s3, event="GO")

# Combine groups
all_transitions = group_a | group_b
# all_transitions is a TransitionGroup with 3 transitions

transition() – Standalone function

transition(
    source: State,
    event: str,
    target: State,
    *,
    guard: Optional[str] = None,
    actions: Optional[List[str]] = None,
    reenter: bool = False,
    internal: bool = False,
) -> Transition

Functional alternative to State.to(). Creates a Transition between two states. Useful when you prefer a function-call style over the method style.

Returns: Transition

from xstate_statemachine import State, transition

idle   = State("idle", initial=True)
active = State("active")

# These two are equivalent:
t1 = idle.to(active, event="START")
t2 = transition(idle, "START", active)

# With all options:
t3 = transition(
    idle, "START", active,
    guard="isReady",
    actions=["logStart", "initProcess"],
    reenter=False,
    internal=False,
)

Interpreters

Interpreter(machine) – Async

Interpreter(
    machine: MachineNode,
    input: Optional[Any] = None,
    clock: Optional[Clock] = None,
    max_queue_size: Optional[int] = None,
    overflow_policy: OverflowPolicy = OverflowPolicy.RAISE,
    strict: Optional[bool] = None,
    service_executor: Optional[concurrent.futures.Executor] = None,
    service_pool_size: int = DEFAULT_SERVICE_POOL_SIZE,
)

The primary asynchronous state machine engine. Processes events from an asyncio.Queue, manages background tasks for after timers and invoked services, and supports spawning child actors. Recommended for I/O-bound applications (web servers, IoT, automation scripts).

Uses a dedicated TaskManager to track all background asyncio.Task objects, ensuring clean cancellation when states are exited.

Constructor parameters

Methods

Class Method

Properties

Full async example

import asyncio
from xstate_statemachine import create_machine, Interpreter, MachineLogic

config = {
    "id": "lightSwitch",
    "initial": "off",
    "context": {
        "flips": 0,
    },
    "states": {
        "off": {
            "on": {
                "TOGGLE": {
                    "target": "on",
                    "actions": "countFlip",
                },
            },
        },
        "on": {
            "on": {
                "TOGGLE": {
                    "target": "off",
                    "actions": "countFlip",
                },
            },
        },
    },
}


def count_flip(interpreter, context, event, action_def):
    context["flips"] += 1


logic = MachineLogic(
    actions={
        "countFlip": count_flip,
    }
)
machine = create_machine(config, logic=logic)


async def main():
    service = await Interpreter(machine).start()

    await service.send("TOGGLE")  # off -> on
    await asyncio.sleep(0.05)  # let event loop process
    print(service.current_state_ids)  # {"lightSwitch.on"}
    print(service.context)  # {"flips": 1}

    await service.send("TOGGLE")  # on -> off
    await asyncio.sleep(0.05)
    print(service.context)  # {"flips": 2}

    await service.stop()


asyncio.run(main())

SyncInterpreter(machine) – Synchronous

SyncInterpreter(
    machine: MachineNode,
    input: Optional[Any] = None,
    clock: Optional[Clock] = None,
    strict: Optional[bool] = None,
    max_queue_size: Optional[int] = None,   # parity only; non-None raises ValueError (#245)
    overflow_policy: Optional[OverflowPolicy] = None,
)

A fully synchronous interpreter that processes events immediately within the send() call. Suitable for CLI tools, desktop GUI event loops, simple workflows, and predictable testing scenarios.

Events are handled one at a time from an internal collections.deque, ensuring sequential, blocking execution. after timers are scheduled on the injected Clock (thread-free; #49/#50) rather than a background thread.

Constructor parameters

Methods

Class Method

Properties

Same as Interpreter:

Sync limitations

Full sync example

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "door",
    "initial": "closed",
    "states": {
        "closed": {
            "on": {
                "OPEN": "opened",
            },
        },
        "opened": {
            "on": {
                "CLOSE": "closed",
            },
        },
    },
}

machine = create_machine(config)
interpreter = SyncInterpreter(machine).start()

print(interpreter.current_state_ids)  # {"door.closed"}
interpreter.send("OPEN")
print(interpreter.current_state_ids)  # {"door.opened"}
interpreter.send("CLOSE")
print(interpreter.current_state_ids)  # {"door.closed"}

# Snapshot round-trip
snapshot = interpreter.get_snapshot()
restored = SyncInterpreter.from_snapshot(snapshot, machine)
print(restored.current_state_ids)  # {"door.closed"}

interpreter.stop()

Using plugins with .use()

from xstate_statemachine import SyncInterpreter, LoggingInspector

interpreter = (
    SyncInterpreter(machine)
    .use(LoggingInspector())
    .start()
)
# All events, transitions, and actions will be logged

Clock

Time as an injectable dependency (#48, #49, #50). A Clock schedules callbacks; the interpreter decides what a fired callback means. The same Clock object may serve both engines at once, so a parent and its invoked children (which may be either engine) share one timeline.

Clock (Protocol)

@runtime_checkable
class Clock(Protocol):
    def now(self) -> float: ...
    def set_timeout(self, fn, delay_sec: float, *, owner: Any = None, sync: bool | None = None) -> Any: ...
    def clear_timeout(self, handle: Any) -> None: ...
    def pump(self) -> int: ...

RealClock()

Wall-clock time and the default for both engines when no clock is passed to Interpreter/SyncInterpreter. Inside a running asyncio loop, a timeout is loop.call_later (the same primitive asyncio.sleep uses). Outside a loop (the sync engine), a timeout is a record in a deadline heap and no thread is started – due callbacks run when SyncInterpreter.send() / .tick() calls .pump().

SimulatedClock(*, wall_start=None)

Virtual time for deterministic tests, mirroring XState’s SimulatedClock. Time does not pass on its own – .increment() advances virtual time and fires every timer that became due, in due order.

from xstate_statemachine import Interpreter, SyncInterpreter, SimulatedClock

# Sync engine
clock = SimulatedClock()
interp = SyncInterpreter(machine, clock=clock).start()
clock.increment(30_000)   # fires a 30s `after` transition synchronously

# Async engine
async def main():
    clock = SimulatedClock()
    service = await Interpreter(machine, clock=clock).start()
    await clock.increment(30_000)   # fires the 30s `after`, settles the loop

Data Classes

Event(type, payload={})

@dataclass(frozen=True)
class Event:
    type: str
    payload: Dict[str, Any] = field(default_factory=dict)

The standard event sent to the state machine. This is a frozen dataclass – instances are immutable after creation. Each instance gets its own payload dict (no shared mutable default).

Properties

Type-checking behavior

Because Event is frozen, attempts to mutate fields after creation raise FrozenInstanceError:

e = Event(
    type="CLICK",
    payload={
        "x": 10,
    },
)
e.type = "OTHER"  # Raises dataclasses.FrozenInstanceError

However, the payload dict itself is mutable (only the reference is frozen):

e.payload["y"] = 20  # This works -- dict mutation, not field reassignment

Creating events

from xstate_statemachine import Event

# Simple event
click = Event(type="CLICK")

# Event with payload
login = Event(
    type="LOGIN",
    payload={
        "username": "alice",
        "role": "admin",
    },
)

# Access payload via .data alias
print(login.data["username"])  # "alice"

The send() method on interpreters also accepts strings and dicts, which are internally converted to Event objects:

interpreter.send("CLICK")  # Event(type="CLICK")
interpreter.send(
    "LOGIN", username="alice"
)  # Event(type="LOGIN", payload={"username": "alice"})
interpreter.send(
    {
        "type": "CLICK",
        "x": 10,
    }
)  # Event(type="CLICK", payload={"x": 10})

ActionDefinition(config)

class ActionDefinition:
    def __init__(self, config: Union[str, Dict[str, Any]]): ...

Represents a single action to be executed, standardizing both shorthand string definitions and detailed object definitions from the JSON config.

Construction from string:

ad = ActionDefinition("myAction")
# ad.type == "myAction", ad.params == None

Construction from dict:

ad = ActionDefinition(
    {
        "type": "myAction",
        "params": {
            "delay": 100,
        },
    }
)
# ad.type == "myAction", ad.params == {"delay": 100}

Raises: InvalidConfigError if config is not a string or dictionary.

The __xstate_event__ adapter protocol [0.11.0]

Any object can say how it becomes an event by implementing __xstate_event__(self) -> str | dict | Event — the way __fspath__ lets any object be a path (#305). Every send path on both engines (send, send_events, send_threadsafe, sendTo specs, can()) normalises through one function, so the adapter works everywhere a str / dict / Event does. The result is normalised by the same rules as a direct argument (one level: an adapter returning another adapter is InvalidEventError, not a recursion). Keyword payload merges over an adapter’s dict, so a call site can annotate: interp.send(order, source="retry"). Native Event / DoneEvent / AfterEvent / ErrorEvent instances are never adapted, even if a subclass defines the method.

from xstate_statemachine import SyncInterpreter, create_machine, MachineLogic


class OrderPlaced:
    def __init__(self, order_id: int) -> None:
        self.order_id = order_id

    def __xstate_event__(self) -> dict:
        return {
            "type": "PLACE",
            "order_id": self.order_id,
        }


cfg = {
    "id": "shop",
    "initial": "idle",
    "context": {},
    "states": {
        "idle": {
            "on": {
                "PLACE": {
                    "target": "placed",
                    "actions": "remember",
                },
            },
        },
        "placed": {},
    },
}


def remember(i, ctx, e, a):
    ctx["order_id"] = e.payload["order_id"]


interp = SyncInterpreter(
    create_machine(
        cfg,
        logic=MachineLogic(
            actions={
                "remember": remember,
            }
        ),
    )
).start()
interp.send(OrderPlaced(42))  # no translation at the call site
assert interp.context["order_id"] == 42 and interp.matches("shop.placed")

DoneEvent(type, data, src)

class DoneEvent(NamedTuple):
    type: str
    data: Any
    src: str

Provenance (#195). Only a DoneEvent the engine minted is a completion. One you construct yourself is an ordinary user event with an engine-shaped name: under strict=True it is refused (UnknownEventError, “engine-generated name”), otherwise it reaches onUnhandled like any undeclared event. It never drives onDone. The same holds for ErrorEvent and AfterEvent.

An internal success event generated by the interpreter when:

1. An invoked service completes successfully (done.invoke.<service_id>).
2. A compound/parallel state reaches its final state (done.state.<state_id>).

Failures are delivered as an ErrorEvent since 0.9.0 (#80).

Accessing data in handlers

def save_result(interpreter, context, event, action_def):
    # event is a DoneEvent when handling onDone
    context["result"] = event.data

ErrorEvent(type, error, src) [0.9.0]

class ErrorEvent(NamedTuple):
    type: str
    error: BaseException
    src: str

Delivered when an invoked service raises, a child actor’s bring-up fails, or a child machine ends in the error status (#80). A distinct type from DoneEvent, so onError handlers branch on isinstance(event, ErrorEvent) or read event.error instead of string-prefixing event.type — the shape XState v5 uses for xstate.error.actor.*.

event.data still returns error with a DeprecationWarning so 0.8.0-era onError actions keep working; it is removed in 1.0.

from xstate_statemachine import ErrorEvent

def store_error(interpreter, context, event, action_def):
    assert isinstance(event, ErrorEvent)
    context["error"] = str(event.error)

AfterEvent(type, scheduled_for=0.0, fired_at=0.0)

class AfterEvent(NamedTuple):
    type: str
    scheduled_for: float = 0.0   # clock time the deadline was set for
    fired_at: float = 0.0        # clock time the timer actually fired

    @property
    def lateness_ms(self) -> float: ...

An internal event for delayed (after) transitions. Created and sent automatically by the interpreter when entering a state with "after" configuration. Users never create this event manually.

Format examples

Engine provenance: is_system_event, system_event, ENGINE_EVENT_SHAPES [0.9.0]

def is_system_event(event: Any) -> bool: ...
def system_event(event_type: str, **payload: Any) -> Event: ...

ENGINE_EVENT_SHAPES: Tuple[str, ...]   # ("done.invoke.", "done.state.", "error.platform.", "after.", "xstate.", "___xstate")
SYSTEM_EVENT_PREFIXES: Tuple[str, ...]  # coarser legacy prefixes, kept for compatibility

Three behaviours — wildcard matching ("*" / "prefix.*"), onUnhandled: "error" / "defer", and strict=True — treat events minted by the engine differently from yours. Since 0.9.0 (#79, #137) that decision is made by provenance, not by name — and, since #195, not by type either: is_system_event is the single predicate all three consult, and system_event(...) is the only way to mint a plain Event that satisfies it. A DoneEvent / ErrorEvent / AfterEvent you construct by hand is user traffic: it does not drive onDone / onError / after, strict refuses it as an engine-generated name, and onUnhandled applies. The engine mints its own completions through private subclasses (identical to the public class under isinstance, equality, _replace, pickle and deepcopy); a persisted engine completion round-trips through persist_event / restore_event with its provenance intact ("engine": true on the record), while a hand-written record restores as user traffic. An event you name done.review is user traffic. The marker survives deepcopy, pickle, send(wait=True) and a snapshot round-trip (#111, #138). ENGINE_EVENT_SHAPES lists the name shapes the engine produces, for build-time checks and documentation only.

from xstate_statemachine import Event, is_system_event, system_event

is_system_event(Event("done.review"))                            # False
is_system_event(system_event("___xstate_statemachine_init___"))  # True

re_mint(original, **fields) [0.9.1]

def re_mint(original: DoneEvent | ErrorEvent | AfterEvent, **fields: Any) -> DoneEvent | ErrorEvent | AfterEvent: ...

The sanctioned way to change a field of an engine-minted event and keep its provenance (#248). _replace() on an engine event deliberately returns the public class (#235) — a caller-chosen variant is user traffic — which left no route for the legitimate case: a plugin that redacts data before re-emitting a completion. re_mint is safe because it is gated on the input: original must already satisfy is_system_event, so provenance can only be carried forward from an event the engine produced, never created. A hand-built event, a plain Event, or a _replace()-demoted value raises TypeError. The original is untouched.

from xstate_statemachine import re_mint, is_system_event


def on_event_received(self, interp, event):
    if event.type.startswith("done.invoke."):
        safe = re_mint(
            event,
            data={
                "redacted": True,
            },
        )
        assert is_system_event(safe)

event_kind, persist_event, restore_event (module xstate_statemachine.events)

The record format a snapshot uses for one pending / deferred / scheduled event. Importable for tooling that reads or rewrites journals; the engine’s own get_persisted_snapshot() / from_snapshot() go through them.

Receipt(state_ids, changed, error=None, deferred=False, denied=False, duplicate=False) [0.9.0]

class Receipt(NamedTuple):
    state_ids: FrozenSet[str]
    changed: bool
    error: Optional[BaseException] = None
    deferred: bool = False
    denied: bool = False
    duplicate: bool = False

What send(..., wait=True) resolves to once the event’s macrostep has run to completion (#39).

⚠️ Arity changes. Receipt grew from three fields to five in 0.9.0 (deferred, then denied) and to six in 0.11.0 (duplicate). A positional destructure written for an older width raises ValueError; read fields by attribute (#119, #153, #304).

receipt = await interpreter.send("SUBMIT", wait=True)
if receipt.error is not None:
    print(f"SUBMIT did not run cleanly: {receipt.error}")

OverflowPolicy

class OverflowPolicy(str, Enum):
    RAISE = "raise"
    BLOCK = "block"
    DROP_NEWEST = "drop_newest"

What send() does when a bounded inbox (max_queue_size on Interpreter) is full (#38).

OverflowPolicy is a str subclass, so plain strings ("raise", "block", "drop_newest") are also accepted anywhere it is expected.

from xstate_statemachine import Interpreter, OverflowPolicy

service = await Interpreter(
    machine, max_queue_size=1000, overflow_policy=OverflowPolicy.DROP_NEWEST
).start()

PendingInvocation(state_id, invoke_id, src)

class PendingInvocation(NamedTuple):
    state_id: str
    invoke_id: str
    src: str

An invoke that is part of the active configuration but has no live task – what .pending_invocations() returns (#44).

Decorators

@action / @action("name")

Marks a function as a state machine action. Actions are side-effect functions that run during transitions, on state entry, or on state exit.

@action
def my_action_name(interpreter, context, event, action_def):
    ...

@action("customActionName")
def whatever(interpreter, context, event, action_def):
    ...

Signature target

(interpreter: BaseInterpreter, context: Dict, event: Event, action_def: ActionDefinition) -> None

Auto-naming

When used without arguments (@action), the function name is auto-converted from snake_case to camelCase:

When used with an explicit name (@action("myName")), the provided string is used exactly as-is.

Examples

from xstate_statemachine import action

# Auto-named: registered as "incrementCounter"
@action
def increment_counter(interpreter, context, event, action_def):
    context["count"] += 1

# Explicitly named: registered as "logIt"
@action("logIt")
def my_logger(interpreter, context, event, action_def):
    print(f"Event: {event.type}, Context: {context}")

@guard / @guard("name")

Marks a function as a state machine guard. Guards are boolean predicate functions that control whether a transition is taken. Guards MUST be synchronous – async guards raise NotSupportedError at decoration time.

@guard
def is_valid(context, event):
    return context.get("valid", False)

@guard("canProceed")
def check_proceed(context, event):
    return context["step"] > 0

Signature target

(context: Dict, event: Event) -> bool

Must return: bool – True to allow the transition, False to block it.

Auto-naming

Same snake_case to camelCase conversion as @action.

Examples

from xstate_statemachine import guard

# Auto-named: registered as "hasBalance"
@guard
def has_balance(context, event):
    return context["balance"] > 0

# Explicitly named
@guard("isAdmin")
def check_admin(context, event):
    return event.payload.get("role") == "admin"

# INVALID -- raises NotSupportedError immediately:
# @guard
# async def async_guard(context, event):
#     return True

@service / @service("name")

Marks a function as a state machine service. Services are long-running operations invoked when a state is entered (via the invoke config). They can be synchronous or asynchronous.

@service
def fetch_user(interpreter, context, event):
    return {
        "name": "Alice",
        "id": 42,
    }


@service("loadData")
async def load(interpreter, context, event):
    # async services only work with the async Interpreter
    data = await some_api_call()
    return data

Signature target

(interpreter: BaseInterpreter, context: Dict, event: Event) -> Any

Returns: Any value. The return value becomes the data field on the resulting DoneEvent.

Auto-naming

Same snake_case to camelCase conversion as @action.

Examples

from xstate_statemachine import service


# Auto-named: registered as "fetchData"
@service
def fetch_data(interpreter, context, event):
    import requests

    return requests.get("https://api.example.com/data").json()


# Explicitly named, async
@service("processOrder")
async def process(interpreter, context, event):
    await asyncio.sleep(1)  # simulate work
    return {
        "status": "processed",
        "order_id": context["order_id"],
    }

Logic Binding

MachineLogic(actions=None, guards=None, services=None, delays=None, *, strict=False)

class MachineLogic(Generic[TContext]):
    def __init__(
        self,
        actions: Optional[Mapping[str, ActionCallable]] = None,
        guards: Optional[Mapping[str, GuardCallable]] = None,
        services: Optional[Mapping[str, Union[ServiceCallable, MachineNode]]] = None,
        delays: Optional[Mapping[str, Union[int, float, DelayCallable]]] = None,
    ) -> None: ...

# The callable blueprints pin ARITY and the guard's bool return:
ActionCallable  = Callable[[Any, Any, Any, ActionDefinition], Union[None, Awaitable[None]]]
GuardCallable   = Callable[[Any, Any], bool]
ServiceCallable = Callable[[Any, Any, Any], Any]
DelayCallable   = Callable[[Any, Any], Union[int, float]]

A two-argument action or a guard returning str is now a type error at the MachineLogic(...) call. The interpreter/context/event slots are Any on purpose: annotate them as narrowly as you like on your own functions (interp: SyncInterpreter[MyCtx]) and they still fit.

A container (“registry”) for the implementation logic of a state machine. Separates the declarative machine definition (JSON) from the imperative implementation (Python functions).

Subclass example

from xstate_statemachine import MachineLogic


class MyLogic(MachineLogic):
    def __init__(self):
        super().__init__(
            actions={
                "logEvent": self._log_event,
                "increment": self._increment,
            },
            guards={
                "isReady": self._is_ready,
            },
        )

    def _log_event(self, interpreter, context, event, action_def):
        print(f"Event: {event.type}")

    def _increment(self, interpreter, context, event, action_def):
        context["count"] += 1

    @staticmethod
    def _is_ready(context, event):
        return context.get("ready", False)

Dict-based example

from xstate_statemachine import MachineLogic

logic = MachineLogic(
    actions={
        "increment": lambda i, ctx, e, a: ctx.update(
            {
                "count": ctx["count"] + 1,
            }
        ),
    },
    guards={
        "isPositive": lambda ctx, e: ctx["count"] > 0,
    },
    services={
        "fetchData": my_fetch_function,
    },
)

LogicLoader

Singleton class that manages automatic discovery of actions, guards, and services from Python modules and provider class instances. Implements the Convention over Configuration principle.

Singleton access

LogicLoader.get_instance() -> LogicLoader

Returns the single shared instance of LogicLoader. Creates it on first call.

register_logic_module(module)

loader = LogicLoader.get_instance()
loader.register_logic_module(my_module)

Registers a Python module for global logic discovery. Registered modules are automatically included in every subsequent create_machine() call that uses auto-discovery. Useful for large applications where logic is spread across many files.

Discovery rules

1. Underscore prefix ignored: Functions/methods starting with _ are skipped during discovery.
2. Auto camelCase conversion: Every discovered function is registered under both its original snake_case name and its camelCase equivalent. For example, increment_counter is available as both "increment_counter" and "incrementCounter".
3. Provider methods override modules: When the same name appears in both a module and a provider instance, the provider’s method takes precedence.
4. Fail-fast: If a required implementation cannot be found, ImplementationMissingError is raised immediately.

Usage with create_machine()

from xstate_statemachine import create_machine, LogicLoader
import my_actions_module

# Option A: Register globally (affects all future create_machine calls)
loader = LogicLoader.get_instance()
loader.register_logic_module(my_actions_module)
machine = create_machine(config)  # auto-discovers from registered modules

# Option B: Pass modules per-call
machine = create_machine(config, logic_modules=[my_actions_module])

# Option C: Pass by import string
machine = create_machine(config, logic_modules=["my_app.actions"])

# Option D: Use provider instances
class MyProvider:
    def increment_counter(self, interpreter, context, event, action_def):
        context["count"] += 1

machine = create_machine(config, logic_providers=[MyProvider()])

Machine Inspection

MachineNode.get_state_by_id(state_id) -> StateNode | None

machine.get_state_by_id(state_id: str) -> Optional[StateNode]

Finds a state node by its fully qualified ID by traversing the state tree.

Returns: The StateNode if found, otherwise None.

machine = create_machine(config)
node = machine.get_state_by_id("myMachine.active.loading")
if node:
    print(f"Found: {node.id}, type: {node.type}")

MachineNode.get_next_state(from_state_id, event) -> Set[str] | None

machine.get_next_state(from_state_id: str, event: Event) -> Optional[Set[str]]

Calculates the target state(s) for an event without side effects. A pure function intended for testing your machine’s flow logic. Finds the first valid transition by bubbling up the state hierarchy.

Note: This utility does not evaluate guards. It assumes any guard would pass to show the potential transition target.

Returns: A Set[str] containing the target state ID(s), or None if no transition is found.

from xstate_statemachine import Event

targets = machine.get_next_state("myMachine.idle", Event(type="START"))
assert targets == {"myMachine.running"}

MachineNode.version [0.11.0]

Optional[str] — the chart’s own "version" label from the root of the config, coerced to str; None when the chart declares none. Before 0.11.0 the key was accepted and silently ignored (#305). It is not part of structure_hash — re-labelling a chart does not invalidate stored snapshots — and is written to every snapshot as machine_version. xsm inspect shows it in the header.

machine = create_machine(
    {
        "id": "m",
        "version": 7,
        "initial": "a",
        "states": {
            "a": {},
        },
    }
)
assert machine.version == "7"

MachineNode.structure_hash

machine.structure_hash -> str

A 16-hex-character structural fingerprint of the machine’s behavior: states, transitions, guard/action names, invokes, and after delays. Stable across meta/description edits and key reordering; changes when a state, transition, guard, action, invoke, or delay is added, removed, or renamed. Lazily computed and cached on first access. Written into every snapshot as machine_hash and checked on restore — see Snapshots — Snapshot Envelope.

print(machine.structure_hash)  # e.g. "f21b173044383a6d"

MachineNode.to_json(*, indent=None) -> str

Serialises the XState configuration the machine was built from, exactly as create_machine() received it (so a StateMachine class or a build_machine() chart can be handed to Stately, to the xsm CLI, or to another engine). Pass indent=2 for a readable file. A value JSON cannot represent (a callable context factory, a Python object in meta) raises TypeError rather than being stringified. [0.12.0]

text = machine.to_json(indent=2)

MachineNode.to_plantuml() -> str

Generates a PlantUML state diagram string from the machine definition. Useful for auto-generating documentation diagrams.

puml = machine.to_plantuml()
print(puml)
# @startuml
# hide empty description
# state "off" as toggle_off
# state "on" as toggle_on
# [*] --> toggle_off
# toggle_off --> toggle_on : TOGGLE
# toggle_on --> toggle_off : TOGGLE
# @enduml

MachineNode.to_mermaid() -> str

Generates a Mermaid.js state diagram string. Renders directly in GitHub markdown, MkDocs, and other tools that support Mermaid.

mmd = machine.to_mermaid()
print(mmd)
# stateDiagram-v2
# [*] --> off
# off --> on : TOGGLE
# on --> off : TOGGLE

MachineNode.is_known_event(event_type) -> bool

machine.is_known_event(event_type: str) -> bool

True if event_type matches a descriptor declared anywhere in the machine (#51) – the check strict mode runs against a sent event. Honours the same matching rules as dispatch: an exact on key, a partial "prefix.*" whose prefix matches by dot-segment, or the bare "*" wildcard, which makes every event known. Engine-synthesised events (done.*, error.*, after.*, xstate.*, the init sentinel) are always known.

Backed by the machine.known_events property (FrozenSet[str]), which is built once, lazily, from every state’s on keys, every after delay’s generated type, and every invoke’s generated done.invoke.<id> / error.platform.<id>.

machine = create_machine(config)
assert machine.is_known_event("OPEN")
assert not machine.is_known_event("TYPO_EVENT")

xstate_statemachine.persistence

The package that owns the snapshot format contract — used internally by get_snapshot() / from_snapshot(), and importable directly for tooling that needs to inspect or migrate snapshots. Since 0.11 it is a package (persistence/snapshot.py holds the envelope; stores, locking, idempotency and durable timers are being added under it — see Integrations). Every name below is re-exported from xstate_statemachine.persistence, so imports written against the old module keep working.

from xstate_statemachine.persistence import SNAPSHOT_VERSION, structure_hash

Stores [0.11.0]

The create → act → persist → discard contract (#259). Guide: Persistence Stores. Every backend passes tests/persistence/test_store_contract.py.