PyStator

Configuration-driven, stateless finite state machines for Python: define behavior in YAML, compute transitions, run guards and actions.

Quick start (2 minutes)

Install, define a tiny state machine, and process one event. Copy-paste into a new terminal:

pip install pystator

1. Save this as order_fsm.yaml:

State and trigger names must be lowercase a–z, digits, and underscores only (see Naming).

meta:
  version: "1.0.0"
  machine_name: "order_management"
  strict_mode: true

states:
  - name: pending
    type: initial
  - name: open
    type: stable
  - name: filled
    type: terminal

transitions:
  - trigger: exchange_ack
    source: pending
    dest: open
  - trigger: fill
    source: open
    dest: filled

2. Run this Python:

from pystator import StateMachine

machine = StateMachine.from_yaml("order_fsm.yaml")

# Pure computation: current state + event → next state
result = machine.process("pending", "exchange_ack", {})

print(result.success)        # True
print(result.target_state)   # open

Validate a dict before building (schema + semantics: hierarchy, parallel regions, etc.):

from pystator import StateMachine, validate_config

config = {...}  # loaded YAML/JSON as dict
errors = validate_config(config, strict=False)
if errors:
    print("\n".join(errors))
else:
    machine = StateMachine.from_dict(config)

Or use StateMachine.validate_config(config) for the same return value without importing validate_config.

Then add guards (conditions) and actions (side effects), or use the REST API and UI with pip install pystator[api] (see Concepts and Documentation).

What is PyStator?

PyStator is a stateless finite state machine (FSM) library for Python. You define behavior in YAML or JSON; the engine computes transitions from (current state + event + context) and returns the next state and any actions to run. No internal state is held—ideal for APIs, workers, and distributed systems.

• Configuration-driven: Define states, transitions, guards, and actions in YAML/JSON with schema validation.
• Stateless: Pure computation—pass state in, get state and actions out; you persist state in your database.
• Hierarchical & parallel: Compound states, orthogonal regions, and statechart-style exit/enter semantics.
• Guards & actions: Conditional transitions (sync/async guards) and side effects executed after you persist the transition.
• Declarative checks & effects: YAML-friendly field checks (check:) and context mutations (set, timestamp, …) without Python—see Declarative features.
• Inline code: actions: Optional Python snippets in config (gated by [features] / env); restricted builtins and allowlisted imports when enabled—see Declarative features. The workspace UI edits them in a Monaco editor; execution runs only on orchestrator / entity / worker paths, not on stateless POST /api/v1/process.
• Delayed transitions: Schedule transitions after a delay (asyncio, Redis, or Celery).
• Worker queue: Durable worker_events table + submit_event() + pystator worker for async processing (pip install pystator[worker] with a database).
• Optional API & UI: REST API and web UI for validation, process, machine CRUD, and entity lifecycle (pip install pystator[api]).

Core vs optional: The core is the FSM library (StateMachine, EntitySession, guards, actions, stores, Orchestrator). For database-backed machine definitions, the library also provides machine_parser → machine_store → machine_builder (see Package structure). The API, DB, UI, and schedulers are optional layers on top.

Concepts

A short mental model so you know what to reach for.

Flow from "just compute" to "full app":

  Option A: In-memory (EntitySession or stateless)
  YAML FSM  →  StateMachine.from_yaml()  →  machine.create(context=...)  →  instance.send(trigger, **payload)
  Or: machine.process(state, trigger, context) with state you supply; persist snapshot() yourself if needed.

  Option B: With persistence by entity ID
  State store (DB/Redis)  →  Orchestrator  →  load state  →  process()  →  Persist  →  Execute actions
  (Sandwich pattern). Use a StateStore adapter (docs/guides/state-stores.md).

  Option C: With API & UI
  pystator api  →  Validate configs, stateless POST /api/v1/process, stored machines + entities (POST /api/v1/entities/{id}/events runs actions), REST/UI — see docs/api/rest-api.md

  Option D: Worker queue
  submit_event(...) → worker_events table → pystator worker (docs/guides/worker.md)

Start with Option A (Quick start above); use EntitySession when you want a stateful object per entity; add Orchestrator when you need persistence by entity ID; add API/UI when you want HTTP and a visual builder; add Option D when you want durable queued processing (pystator[worker] + database).

Naming: states, triggers, guards, and actions

Identifiers in YAML must match: lowercase letters a–z, digits 0–9, and underscores _ only. No dots, no uppercase (e.g. use pending_new, not PENDING_NEW). This applies to state names, triggers, guard names, and action names. See the FSM config reference.

How PyStator runs (execution model)

PyStator is event-driven and stateless. You do not run a long-lived loop per FSM.

• Events drive transitions: When an event occurs (HTTP request, message from a queue, cron job, or a delayed-transition callback), you call process(current_state, trigger, context) or the Orchestrator’s async_process_event(entity_id, trigger, context) once. State is loaded from your store, the transition is computed, and you persist the new state and run actions.
• Scale by replicas: Run multiple copies of your API or worker (e.g. several pods). They share the same state store (database or Redis). Any replica can process any event for any entity—there is no “one pod per FSM.” See Deployment.
• Delayed transitions (after: "30s" in YAML) need a scheduler so that when the delay expires, something calls the orchestrator again. One process (or Redis/Celery) can track many delays for many entities and FSM types. See Choosing a scheduler below.

Glossary:

Choosing a scheduler

Only needed if your FSM uses delayed transitions (after: "5s" or after: 5000 in a transition).

With AsyncioScheduler, one pod can track delays for all entities and all FSM types; delays are lost if that process exits. With Redis or Celery, delays are stored externally so any healthy worker can fire them. See Schedulers and delayed transitions.

Features

• Configuration-driven: YAML/JSON definitions with schema validation
• Stateless: Pure computation—no internal state
• Hierarchical states: Compound states, parent/child, LCA exit/enter
• Parallel states: Orthogonal regions—multiple active sub-states
• Delayed transitions: after: 5s or after: 5000 with pluggable schedulers (asyncio, Redis, Celery)
• Inline guards: expr: "fill_qty >= order_qty" in YAML (no Python for simple rules)
• Guards & actions: Sync and async; decorator-based registration
• Action parameters: Pass config from YAML into actions via params
• Timeouts: State-level timeout to a destination state
• Type-safe: Full type hints and PEP 561
• Retry & idempotency: Configurable retry, pluggable idempotency backends
• REST API & UI: Optional server and web UI for FSM validation, stateless /process, stored machines, entities (full orchestrator + action_results), dry-run, bulk entity ops, settings/runtime overview, and worker-queue apply-data enqueue—see REST API
• Lint: pystator.lint.lint(config) for static FSM warnings (see Linting)
• Sinks: Pluggable event/metric sinks for actions (pystator.sinks)

Installation

Core library

pip install pystator

With API and UI

pip install pystator[api]

Installs the full API stack (FastAPI, Uvicorn, DB/migrations, PostgreSQL and MongoDB drivers, Pydantic, inline expr: guards, MkDocs for pystator docs serve). The UI is served by the same server when you run pystator ui serve (requires a built UI; see below).

With Worker (queued event processing)

pip install pystator[worker]

Adds SQLAlchemy, Alembic, Redis, Celery, and the PostgreSQL driver used by pystator worker, submit_event, and the worker_events queue. You still configure PYSTATOR_DATABASE_URL (or pystator.cfg) so enqueue and workers hit the same database.

With UI (development)

To build and serve the Next.js UI from source:

pip install pystator[api,ui]
cd src/pystator/ui && npm install && npm run build
pystator ui serve   # Serves UI + proxies API

From the project root you can also run pystator ui dev for hot-reload development.

Inline guard expressions in YAML (expr: "qty > 0") use simpleeval, which is a core dependency (pip install pystator is enough).

Development

pip install -e ".[dev]"

Quick start (extended)

From a YAML file

from pystator import StateMachine

machine = StateMachine.from_yaml("order_fsm.yaml")
result = machine.process("pending", "exchange_ack", {})

From a dict

from pystator import StateMachine

config = {
    "meta": {"version": "1.0.0", "machine_name": "my_fsm", "strict_mode": True},
    "states": [
        {"name": "a", "type": "initial"},
        {"name": "b", "type": "stable"},
        {"name": "c", "type": "terminal"},
    ],
    "transitions": [
        {"trigger": "go", "source": "a", "dest": "b"},
        {"trigger": "done", "source": "b", "dest": "c"},
    ],
}
machine = StateMachine.from_dict(config)
result = machine.process("a", "go", {})

EntitySession — per-entity stateful wrapper

For per-entity stateful processing in memory (one instance per order, user, trade, etc.), use EntitySession:

from pystator import StateMachine

machine = StateMachine.from_yaml("order_fsm.yaml")

# Create a stateful instance for one entity (e.g. order-123)
order = machine.create(context={"order_id": "123", "order_qty": 100})

print(order.current_state)   # pending
print(order.is_pending)      # True  — auto-generated per-state property

# Send events; payload merges into context
order.send("exchange_ack")
print(order.current_state)   # open

order.send("fill", fill_qty=100)
print(order.current_state)   # filled
print(order.is_final)        # True — entity has reached a terminal state

# Auto-generated trigger methods (same as send):
# order.exchange_ack()
# order.fill(fill_qty=100)

# Serialize/restore (e.g. to Redis or a DB)
snapshot = order.snapshot()
restored = EntitySession.from_snapshot(machine, snapshot)

EntitySession auto-generates instance.is_<state> properties and instance.<trigger>() methods for every state and trigger in your FSM. Import EntitySession from pystator to use from_snapshot.

With guards and actions

You can bind guards and actions in two ways: decorators on the machine, or registries plus bind_guards / bind_actions.

Option 1 — decorators:

from pystator import StateMachine

machine = StateMachine.from_yaml("order_fsm.yaml")

@machine.guard("is_full_fill")
def is_full_fill(ctx):
    return ctx.get("fill_qty", 0) >= ctx.get("order_qty", 1)

@machine.action("update_positions")
def update_positions(ctx):
    print("Positions updated")

# Then: machine.process(...) and ActionExecutor(machine.action_registry).execute(result, context) to run actions

Option 2 — registries:

from pystator import StateMachine, GuardRegistry, ActionRegistry
from pystator.actions import ActionExecutor

machine = StateMachine.from_yaml("order_fsm.yaml")

guards = GuardRegistry()
guards.register("is_full_fill", lambda ctx: ctx.get("fill_qty", 0) >= ctx.get("order_qty", 1))
machine.bind_guards(guards)

actions = ActionRegistry()
actions.register("update_positions", lambda ctx: print("Positions updated"))
executor = ActionExecutor(actions)

result = machine.process("open", "execution_report", {"fill_qty": 100, "order_qty": 100})
if result.success:
    # 1. Persist state change to your DB
    # 2. Then run actions
    executor.execute(result, {"fill_qty": 100, "order_qty": 100})

Orchestrator and delayed transitions

For persistence plus delayed transitions (after: "5s" in YAML), use the Orchestrator with a state store and a scheduler. The orchestrator runs the full loop: load state → process → persist → schedule delayed transitions → execute actions.

import asyncio
from pystator import StateMachine, Orchestrator, GuardRegistry, ActionRegistry
from pystator import InMemoryStateStore
from pystator.scheduler import AsyncioScheduler

machine = StateMachine.from_yaml("my_fsm.yaml")  # has a transition with after: "2s"
store = InMemoryStateStore()
guards = GuardRegistry()
actions = ActionRegistry()
scheduler = AsyncioScheduler()

orchestrator = Orchestrator(
    machine=machine, state_store=store, guards=guards, actions=actions, scheduler=scheduler
)

async def main():
    await orchestrator.async_process_event("entity-1", "start", {})
    await asyncio.sleep(2.5)  # delayed transition fires
    await orchestrator.close()

asyncio.run(main())

No extra infrastructure: AsyncioScheduler keeps delays in memory. For multiple replicas or restarts, use RedisScheduler or CeleryScheduler.

Context validation (e.g. with PyCharter)

Pass a context_validator to the Orchestrator to validate context before any guards run. The canonical use case is integrating PyCharter data contracts:

from pystator import Orchestrator, ContextValidatorFn
from pycharter import Validator

validator = Validator.from_dir("contracts/order")

def order_context_validator(entity_id: str, state: str, trigger: str, ctx: dict):
    result = validator.validate_for_state(ctx, state)
    if result.is_valid:
        return True, []
    return False, [str(e) for e in result.errors]

orchestrator = Orchestrator(
    machine=machine,
    state_store=store,
    context_validator=order_context_validator,   # (entity_id, state, trigger, ctx) -> (ok, errors)
)

ContextValidatorFn is a type alias: (entity_id, current_state, trigger, context) -> tuple[bool, list[str]]. Sync or async callables are supported. The validator runs before guards; on failure the orchestrator returns a TransitionResult with success=False and metadata["reason"] == "context_validation".

Three context validation mechanisms exist and can be combined:

Hooks and metrics

Attach lifecycle hooks to the TransitionObserver to add logging, tracing, or custom metrics:

from pystator import LoggingHook, MetricsCollector, StateMachine, TransitionObserver

machine = StateMachine.from_yaml("order_fsm.yaml")

# Optional: attach hooks to a standalone observer (use at orchestrator/process level)
observer = TransitionObserver()
observer.add_hook(LoggingHook())
metrics = MetricsCollector()
observer.add_hook(metrics)

# ... process events ...
machine.process("pending", "exchange_ack", {})

summary = metrics.get_summary()
print(summary["total_transitions"])   # 1
print(summary["success_rate"])        # 1.0
print(summary["duration"])            # {"min_ms": ..., "avg_ms": ..., "p95_ms": ...}

To write a custom hook implement the TransitionHook protocol (four methods: on_before_process, on_transition_start, on_transition_complete, on_transition_error).

Error handling

All exceptions inherit from FSMError. Catch the specific subclass you care about:

from pystator import (
    FSMError,             # base — catch everything
    GuardRejectedError,   # guard returned False
    InvalidTransitionError, UndefinedTriggerError,
    TerminalStateError,   # trying to leave a terminal state
    StaleVersionError,    # optimistic locking conflict
    ErrorCode,            # enum: GUARD_REJECTED, VALIDATION_FAILED, TERMINAL_STATE, …
)

try:
    result = machine.process(current_state, trigger, context)
except GuardRejectedError as e:
    print(e.guard_name, e.current_state)   # which guard failed and in which state
except TerminalStateError:
    print("Entity already in a terminal state")
except FSMError as e:
    print(e.code)   # ErrorCode member, e.g. ErrorCode.INVALID_TRIGGER

TransitionResult (the non-exception path) carries the same information for soft failures:

result = machine.process(current_state, trigger, context)
if not result.success:
    print(result.error)          # FSMError or subclass
    print(result.metadata)       # {"reason": "guard_rejected", "guard_name": "is_full_fill", …}

Common pitfalls

• Guards vs actions: Use guards for pure logic (can this transition run?). Use actions for side effects (notify, persist to another system). Don’t put side effects in guards.
• AsyncioScheduler: Delays are in-memory; they are lost if the process exits. Use Redis or Celery for production or multiple replicas.
• State store: With Option B, you must implement a StateStore and persist before running actions; the library does not persist for you.

REST API

With pip install pystator[api]:

# Start API (default: http://localhost:8004 — see pystator.config.ports.API_PORT)
pystator api
# or: uvicorn pystator.api.main:app --reload --port 8004
# Optional: use pystator.cfg for database and auth (copy pystator.cfg.example to pystator.cfg)

Full paths and bodies: /docs (OpenAPI) is the source of truth when the server is running.

Documentation

• Full documentation (Python API, guides, tutorials): https://optophi.github.io/pystator/ — built from this repo with MkDocs (pystator docs serve or pystator docs build).
• CLI reference — pystator api, ui, db, machines, discovery, schema, worker, docs
• Quick start (detailed) — Step-by-step first FSM and first API call
• Concepts — States, transitions, guards, actions, hierarchical and parallel
• Architecture — Design goals, core flow, sandwich pattern, components
• Configuration — Config file, environment, database (for API)
• Worker — Continuous event-processing service: pystator worker, submit_event, worker_events
• Declarative features — YAML check: / effects (set, timestamp, …), builtins, sinks
• Tutorials — Order workflow, API & UI, delayed transitions
• Examples & Notebooks — Runnable examples/ scripts and flat notebooks/ (Marimo)
• FSM config reference — Full YAML/JSON schema (meta, states, transitions, validation)
• API reference — StateMachine, Orchestrator, Worker, schedulers, execution modes (also in the full docs site with docstrings).

Examples and tutorials

Runnable examples live in the examples/ directory:

See examples/README.md for how to run each. Tutorials in docs/tutorials/ walk through building an order workflow and using the API and UI.

Marimo notebooks (optional, pip install -e ".[dev]"): plain-Python .py files in the repo’s flat notebooks/ directory — catalog and YAML assets are listed there.

API reference (condensed)

StateMachine

# Create
machine = StateMachine.from_yaml("config.yaml")
machine = StateMachine.from_dict(config_dict)

# Process (sync)
result = machine.process(current_state, trigger, context)

# Process (async, for async guards)
result = await machine.aprocess(current_state, trigger, context)

# Parallel states
config = machine.enter_parallel_state("parallel_state_name")
config, results = machine.process_parallel(config, event, context)

# Queries
machine.get_initial_state()
machine.get_available_transitions("pending")

# Target state without a trigger (e.g. validate an incoming status string)
check = machine.evaluate_destination("pending", "open", context)
check.allowed                 # True if at least one transition reaches "open"
check.candidates            # per-trigger outcomes (guards applied)
check.is_ambiguous          # True if multiple triggers can reach the destination
machine.can_transition_to("pending", "open", context)  # bool shortcut

See Declarative features for StateMachineBuilder, YAML check: / effects, and lint.

EntitySession

instance = machine.create(context={...}, initial_state=None)
instance.current_state       # str
instance.context             # dict
instance.is_<state>          # bool — auto-generated (e.g. instance.is_pending)
instance.<trigger>(**payload) # auto-generated method (e.g. instance.confirm())
instance.send(trigger, **payload)
await instance.asend(trigger, **payload)  # async
instance.allowed_triggers    # list[str]
instance.is_final            # bool — True if in a terminal state
instance.evaluate_destination("filled", context_overrides={...})  # DestinationCheckResult
instance.can_transition_to("filled")  # bool
snapshot = instance.snapshot()
instance2 = EntitySession.from_snapshot(machine, snapshot)

TransitionResult

result.success          # bool
result.source_state     # str
result.target_state     # str | None
result.trigger          # str
result.all_actions      # tuple[str, ...]  (exit + transition + enter)
result.error            # FSMError | None
result.metadata         # dict — e.g. {"reason": "guard_rejected", "guard_name": "..."}

Guards and actions

guards = GuardRegistry()
guards.register("name", lambda ctx: bool)
@guards.decorator("name")
def my_guard(ctx: dict) -> bool: ...

actions = ActionRegistry()
actions.register("name", lambda ctx: None)
@actions.decorator()
def my_action(ctx: dict) -> None: ...

machine.bind_guards(guards)
executor = ActionExecutor(actions)
executor.execute(transition_result, context)
# Async: await executor.async_execute_parallel(result, context)

Orchestrator

from pystator import Orchestrator, ContextValidatorFn

orchestrator = Orchestrator(
    machine=machine,
    state_store=store,
    guards=guards,
    actions=actions,
    context_validator=fn,    # ContextValidatorFn: (entity_id, state, trigger, ctx) -> (ok, errors)
    scheduler=scheduler,
    use_initial_state_when_missing=True,
)
result = orchestrator.process_event(entity_id, trigger, context)
result = await orchestrator.async_process_event(entity_id, trigger, context)
await orchestrator.close()  # shut down scheduler

Hooks

from pystator import LoggingHook, MetricsCollector, TransitionHook, TransitionObserver

observer = TransitionObserver()
observer.add_hook(LoggingHook())
metrics = MetricsCollector()
observer.add_hook(metrics)
# Use observer at your process/orchestrator level; then metrics.get_summary()
# returns {"total_transitions", "success_rate", "duration": {"avg_ms", "p95_ms", ...}}

Error codes

from pystator import ErrorCode
# ErrorCode.GUARD_REJECTED, VALIDATION_FAILED, INVALID_TRIGGER, INVALID_STATE,
# TERMINAL_STATE, TIMEOUT, CONFIG, ACTION_FAILED, STALE_VERSION,
# GUARD_NOT_FOUND, ACTION_NOT_FOUND, UNDEFINED_STATE

Development

pip install -e ".[dev]"
# Optional: ./scripts/setup.sh   # venv, editable install, pre-commit / pre-push hooks

pytest
make check          # format (Ruff) + lint + mypy src/pystator + tests
# or: make format && make lint && make type-check && pytest

License

MIT — see LICENSE.

Links

• Repository: GitHub
• Issues: GitHub Issues
• Contributing: CONTRIBUTING.md · Security · Code of conduct
• Documentation: docs/ — quick start, guides, tutorials, examples