API Reference

Core

The machine itself, its per-run execution context, and the event that drives a transition.

class statem.StateMachine(**data)

A validated, immutable state graph paired with registries of named guards/actions.

Build one with from_dict, then drive it with await run(…). Instances are frozen and hold no per-run state, so a single StateMachine can safely process many concurrent runs – all mutable state lives in the Context created fresh for each run() call.

Variables:

• config – State name to StateConfig mapping – the validated transition graph.

• guards – Registry of named guard functions, evaluated to decide which transition fires.

• actions – Registry of named action functions, executed on entry/exit/transition.

Parameters:

• config (dict[str, StateConfig])

• guards (GuardRegistry)

• actions (ActionRegistry)

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

config: dict[str, StateConfig]

guards: GuardRegistry

actions: ActionRegistry

classmethod from_dict(config, action_dict=None, guard_dict=None)

Construct, register actions/guards, then validate.

validate_registries() is called automatically when at least one of action_dict / guard_dict is supplied. If you register later via the registries directly, call validate_registries() yourself.

Return type:

StateMachine

Parameters:

• config (dict[str, Any])

• action_dict (dict[str, Callable[[Context, Signal], Awaitable[Any] | Any]] | None)

• guard_dict (dict[str, Callable[[Context, Signal], bool] | Callable[[Context, Signal], Awaitable[bool]]] | None)

validate_registries()

Validate that all guards and actions referenced in config are registered.

Called automatically by from_dict when at least one of action_dict / guard_dict is supplied. If you register actions/guards later via self.actions/self.guards directly, call this yourself to check for typos – raises ValueError listing anything missing.

Return type:

None

async run(*, run_id=None, thread_id=None, state_name, events, session)

Process one or many signals starting from state. All arguments are keyword-only.

Parameters:

• run_id (Optional[str]) – Correlation id for this run, used in log lines. If not provided (left as None), a uuid4().hex string is generated automatically.

• thread_id (Optional[str]) – Correlation id for the broader conversation/session this run belongs to (one thread can span many runs). If not provided, a uuid4().hex string is generated automatically, same as run_id.

• state_name (str) – Current state name (e.g. “idle”).

• events (Union[Signal, list[Signal]]) – Single Signal, list of Signal`s, or `[] to only resolve always transitions for the current state.

• session (Any) – Caller-owned, opaque payload of any shape (mutated in-place by actions, if they choose to).

Return type:

str

Returns:

The final state name after all transitions have settled.

async stream(*, run_id=None, thread_id=None, state_name, events, session, state_accessor=None)

Like run, but yields AG-UI protocol events as the machine executes.

Drives the same engine as run (same guards, actions, transition rules) and has the same effect on session – this is an additive, alternate way to observe a run, not a different execution path. Requires the agui extra: pip install statem[agui].

A step = one state change (one hop), whether triggered by an on transition, an always cascade hop, or an error_state fallback – not one call to stream(). Each step is fully self-contained, emitted in this order:

• STEP_STARTED (step_name is the triggering signal’s event, or “__always__” for an always-cascade hop).

• STATE_SNAPSHOT, taken right before this hop’s actions run.

• ACTIVITY_SNAPSHOT for every guard and action result as it fires during this hop (content carries name, hook source, and result). A guard evaluated but not taken (e.g. an earlier candidate that failed) is reported the same way, just before its step – or the step before it – opens.

• STATE_DELTA (RFC 6902 patch, via jsonpatch.make_patch, against this step’s own STATE_SNAPSHOT) – skipped if this step didn’t actually change the broadcast state.

• STEP_FINISHED.

A signal that matches no transition, or whose guard(s) all fail, produces no events at all (no state change happened). RUN_STARTED / RUN_FINISHED / RUN_ERROR are never emitted. An unhandled exception (e.g. an action error with no error_state, or a bad guard return type) propagates to the caller from the generator itself, exactly as it would from run.

Parameters:

• run_id (Optional[str]) – Correlation id for this run. Auto-generated if not provided.

• thread_id (Optional[str]) – Correlation id for the broader conversation/session this run belongs to (one thread can span many runs). Auto-generated if not provided, same as run_id.

• state_name (str) – Current state name (e.g. “idle”).

• events (Union[Signal, list[Signal]]) – Single Signal, list of Signal`s, or `[] to only resolve always transitions for the current state.

• session (Any) – Caller-owned, opaque payload of any shape.

• state_accessor (Optional[Callable[[Any], dict[str, Any]]]) – Derives the dict broadcast via STATE_SNAPSHOT/STATE_DELTA from session (e.g. lambda session: session.to_dict()). Called at the start and end of every step. Defaults to {“current_state”: <state name>} when omitted.

Yields:

ag_ui.core.BaseEvent instances in execution order.

Return type:

AsyncIterator[BaseEvent]

available_events(state_name)

Return the event names state_name can receive via on, or [] if the state is unknown.

Excludes the “*” wildcard entry – use StateConfig.accepts_wildcard to check for a catch-all handler.

Return type:

list[str]

Parameters:

state_name (str)

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

Return type:

None

class statem.Context(*, run_id=None, thread_id=None, current_state, session, _event_q=None, _state_accessor=None, _state_snapshot=None)

Short-lived execution context passed to every action and guard during one run() call.

Created at the start of StateMachine.run and discarded when it returns. The machine updates current_state as transitions fire. All fields are keyword-only.

Generic over T, the type of session. Context used bare (unparameterized) behaves exactly as before – session types as Any. Guards/actions that want typed access to their own session shape can annotate their ctx parameter as Context[BankSession] (or whatever their session type is) to get full type-checking and autocomplete on ctx.session.

Variables:

• run_id – Correlation id for this run() call, used in log lines. None if the caller didn’t supply one – StateMachine.run() is responsible for generating one (a uuid4().hex string) before constructing this object; Context itself just stores whatever it’s given.

• thread_id – Correlation id for the broader conversation/session this run belongs to, distinct from run_id (one thread can span many runs). None if the caller didn’t supply one – StateMachine.run()/StateMachine.stream() generate one the same way they generate run_id; Context itself just stores whatever it’s given.

• current_state – Name of the state the machine is currently in; updated by the engine as each transition fires.

• session – Caller-owned, opaque payload of any shape; the engine never inspects it, only threads it through to actions/guards.

• history – Ordered list of every state entered during this run() call; the first entry is always the initial state.

• results – Ordered list of ResultEntry for every guard and action executed, in the exact order they fired during this run() call.

• _event_q – Internal – set by StateMachine.stream() to an asyncio.Queue it drains to yield AG-UI events; left None by StateMachine.run(). Not meant for guards/actions to use directly; check is_stream instead of reading this field.

• _state_accessor – Internal – set by StateMachine.stream() from its state_accessor argument. Derives the dict broadcast via STATE_SNAPSHOT/STATE_DELTA from session; None means the default {“current_state”: …} view is used instead.

• _state_snapshot – Internal – the last dict broadcast via STATE_SNAPSHOT/STATE_DELTA, kept so StateMachine.stream() can diff against it to build the next STATE_DELTA.

Parameters:

• run_id (str | None)

• thread_id (str | None)

• current_state (str)

• session (T)

• _event_q (Queue[Any] | None)

• _state_accessor (Callable[[T], dict[str, Any]] | None)

• _state_snapshot (dict[str, Any] | None)

run_id: Optional[str]

thread_id: Optional[str]

current_state: str

session: TypeVar(T)

history: list[str]

results: list[ResultEntry]

property is_stream: bool

True when this context was created by StateMachine.stream(), False for run().

class statem.Signal(event, data=<factory>)

A signal that triggers a transition.

Variables:

• event – Event name (upper-case by convention, e.g. “START”).

• data – Arbitrary payload dict; defaults to empty.

Parameters:

• event (str)

• data (dict[str, Any])

event: str

data: dict[str, Any]

Configuration

The (Pydantic-validated) shapes that make up the config dict passed to StateMachine.from_dict.

class statem.StateConfig(**data)

Configuration for one state.

Extra fields (e.g. role, intent, render) are accepted and silently ignored, so raw application config dicts can be passed directly. Each on value is normalized to a list of TransitionConfig at parse time.

Variables:

• on – Maps event names to transition candidates.

• always – Eventless transitions checked after every state entry.

• entry – Action names to fire when entering this state.

• exit – Action names to fire when leaving this state.

• error_state – Fallback state on unhandled action errors.

Parameters:

• on (dict[str, list[TransitionConfig]])

• always (list[TransitionConfig])

• entry (list[str])

• exit (list[str])

• error_state (str | None)

model_config: ClassVar[ConfigDict] = {'extra': 'ignore', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

on: dict[str, list[TransitionConfig]]

always: list[TransitionConfig]

entry: list[str]

exit: list[str]

error_state: Optional[str]

property available_events: list[str]

Return the event names this state can receive, in declaration order.

Excludes the “*” wildcard entry – use accepts_wildcard to check whether the state has a catch-all handler.

Example:

cfg = StateConfig.model_validate({
    "on": {"START": "running", "CANCEL": "idle", "*": "error"}
})
cfg.available_events  # ["START", "CANCEL"]
cfg.accepts_wildcard  # True

property accepts_wildcard: bool

Return True if this state has a “*” catch-all transition.

class statem.TransitionConfig(**data)

Configuration for a single transition candidate.

Variables:

• target – Name of the state to transition to.

• guard – Registered guard name that must pass for this transition to fire; None means the transition always passes.

• actions – Ordered list of registered action names to execute when this transition fires.

Parameters:

• target (str)

• guard (str | None)

• actions (list[str])

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

target: str

guard: Optional[str]

actions: list[str]

Registries

Where named guard/action callables are registered and looked up by the engine.

class statem.ActionRegistry

Registry of named action functions (sync or async).

register(name, fn)

Register a single named action function (sync or async).

Return type:

None

Parameters:

• name (str)

• fn (Callable[[Context, Signal], Awaitable[Any] | Any])

register_many(actions)

Register multiple named action functions at once.

Return type:

None

Parameters:

actions (dict[str, Callable[[Context, Signal], Awaitable[Any] | Any]])

has(name)

Return True if an action named name is registered.

Return type:

bool

Parameters:

name (str)

async execute(name, ctx, signal, source)

Execute a single named action. Raises KeyError if not registered.

Return type:

None

Parameters:

• name (str)

• ctx (Context)

• signal (Signal)

• source (Literal['on', 'always', 'entry', 'exit'])

async execute_many(names, ctx, signal, source)

Execute actions in order, awaiting each one.

Return type:

None

Parameters:

• names (list[str])

• ctx (Context)

• signal (Signal)

• source (Literal['on', 'always', 'entry', 'exit'])

class statem.GuardRegistry

Registry of named guard functions (sync or async).

Async detection is cached at register time.

register(name, fn)

Register a single named guard function (sync or async).

Return type:

None

Parameters:

• name (str)

• fn (Callable[[Context, Signal], bool] | Callable[[Context, Signal], Awaitable[bool]])

register_many(guards)

Register multiple named guard functions at once.

Return type:

None

Parameters:

guards (dict[str, Callable[[Context, Signal], bool] | Callable[[Context, Signal], Awaitable[bool]]])

has(name)

Return True if a guard named name is registered.

Return type:

bool

Parameters:

name (str)

async evaluate(name, ctx, signal, source)

Evaluate a guard. Returns True if name is None (no guard = pass).

Return type:

bool

Parameters:

• name (str | None)

• ctx (Context)

• signal (Signal)

• source (Literal['on', 'always', 'entry', 'exit'])

Errors

class statem.GuardError

Raised when a guard function returns a non-bool value.

class statem.TransitionError

Raised when an action fails during a transition.

Tracing a run

class statem.ResultEntry(state, source, kind, name, value)

A single recorded guard or action result, in execution order.

Variables:

• state – Name of the state the machine was in when this fired.

• source – Lifecycle hook: “on”, “always”, “entry”, or “exit”.

• kind – “guard” or “action”.

• name – Registered name of the guard or action function.

• value – bool for guards; return value (may be None) for actions.

Parameters:

• state (str)

• source (Literal['on', 'always', 'entry', 'exit'])

• kind (Literal['guard', 'action'])

• name (str)

• value (Any)

state: str

Alias for field number 0

source: Literal['on', 'always', 'entry', 'exit']

Alias for field number 1

kind: Literal['guard', 'action']

Alias for field number 2

name: str

Alias for field number 3

value: Any

Alias for field number 4

statem.show_transitions(ctx)

Return a human-readable summary of every transition in execution order.

Iterates ctx.results (a list[ResultEntry]) which preserves the exact order guards and actions fired. Each hop groups entries by state.

Example output:

Transitions (2 hops):
--------------------------------------------------------------------------------
hop 1: idle -> collecting
    always guard : session_limit_reached      = False
    on     guard : can_start_txn              = True
           action : create_txn                = None
           action : resolve_fields            = None
--------------------------------------------------------------------------------
hop 2: collecting -> confirming
    always guard : is_enquiry_ready           = False
           guard : all_resolved_and_shown     = False
           guard : all_fields_resolved        = True
    exit   action : mark_shown                = None
--------------------------------------------------------------------------------
Final state: confirming
    entry  action : show_confirm              = None
--------------------------------------------------------------------------------

Return type:

str

Parameters:

ctx (Context)

Diagrams

statem.to_mermaid(machine, *, initial=None)

Render machine.config as a Mermaid stateDiagram-v2 diagram source string.

One edge is emitted per transition candidate: on candidates are labeled with the event name (plus [guard_name] if guarded), always candidates are labeled always (plus the guard, if any), and each error_state becomes an error-labeled edge. A guard-chain – multiple candidates for one event – naturally produces multiple labeled edges from the same state, which is the main payoff: it visualizes branching that a flat transition table hides.

Paste the result into a Markdown “mermaid” code fence (GitHub, Sphinx, VS Code, and Jupyter all render it natively) to view the diagram.

Parameters:

• machine (StateMachine) – The StateMachine whose config to render.

• initial (Optional[str]) – If given and present in machine.config, prepends a [*] –> initial edge marking the diagram’s entry point. StateMachine itself has no notion of an “initial” state – that’s chosen fresh by the caller on every run() call – so this is opt-in.

Return type:

str

Returns:

Mermaid stateDiagram-v2 source, ready to paste into a fence.