Examples¶
Three runnable, fully-tested example scripts live in
examples/ in the repository. All
are shown in full below, kept in sync automatically with the actual source files.
Bakery process (examples/bread.py)¶
A small bakery process (idle → mixing → baking → cooling → done) that exercises guards,
actions, an error_state fallback, and an always-transition that auto-advances
cooling → done once the oven has cooled — all triggered by a single TIMER_DONE signal.
Its graph, rendered with to_mermaid:
stateDiagram-v2
[*] --> idle
idle --> mixing: START
mixing --> baking: MIXED [ingredients_ready]
mixing --> failed: error
baking --> cooling: TIMER_DONE
cooling --> done: always [oven_is_cool]
uv run python examples/bread.py
# ruff: noqa: ARG001
"""Runnable example: a bakery process modeled as a StateMachine.
Demonstrates:
- declaring config as a plain dict (on / always / entry / error_state)
- registering guards and actions by name
- an arbitrary, library-agnostic ``session`` object (here: a plain dataclass)
- the ``always``-transition auto-advance mechanism
"""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass, field
from statem import Context, Signal, StateMachine
logging.basicConfig(level=logging.INFO, format="%(message)s")
@dataclass
class BakingSession:
oven_temp_c: int = 0
ingredients_checked: bool = False
log: list[str] = field(default_factory=list)
def check_ingredients(ctx: Context[BakingSession], signal: Signal) -> None:
ctx.session.ingredients_checked = True
ctx.session.log.append("ingredients checked")
def ingredients_ready(ctx: Context[BakingSession], signal: Signal) -> bool:
return ctx.session.ingredients_checked
def preheat_oven(ctx: Context[BakingSession], signal: Signal) -> None:
ctx.session.oven_temp_c = 180
ctx.session.log.append("oven preheated to 180C")
def start_cooling(ctx: Context[BakingSession], signal: Signal) -> None:
ctx.session.oven_temp_c = 25
ctx.session.log.append("cake pulled, cooling started")
def oven_is_cool(ctx: Context[BakingSession], signal: Signal) -> bool:
return ctx.session.oven_temp_c <= 30 # noqa: PLR2004
def plate_cake(ctx: Context[BakingSession], signal: Signal) -> None:
ctx.session.log.append("cake plated")
CONFIG = {
"idle": {
"on": {"START": {"target": "mixing", "actions": ["check_ingredients"]}},
},
"mixing": {
"on": {"MIXED": {"target": "baking", "guard": "ingredients_ready", "actions": ["preheat_oven"]}},
"error_state": "failed",
},
"baking": {
"on": {"TIMER_DONE": {"target": "cooling", "actions": ["start_cooling"]}},
},
"cooling": {
"always": [{"target": "done", "guard": "oven_is_cool"}],
},
"done": {
"entry": ["plate_cake"],
},
"failed": {},
}
async def main() -> None:
machine = StateMachine.from_dict(
CONFIG,
action_dict={
"check_ingredients": check_ingredients,
"preheat_oven": preheat_oven,
"start_cooling": start_cooling,
"plate_cake": plate_cake,
},
guard_dict={
"ingredients_ready": ingredients_ready,
"oven_is_cool": oven_is_cool,
},
)
session = BakingSession()
state = "idle"
for event in ("START", "MIXED", "TIMER_DONE"):
state = await machine.run(
run_id="bake-001",
state_name=state,
events=Signal(event=event),
session=session,
)
print(f"final state: {state}")
print("session log:")
for line in session.log:
print(f" - {line}")
if __name__ == "__main__":
asyncio.run(main())
Bank teller bot (examples/bank.py)¶
A richer example: a bank teller’s transaction-posting bot that exercises every hook in one
conversation — on guard chains (a two-candidate check for supported vs. unsupported
transaction types), a multi-hop always cascade (missing fields loop back to ask the teller,
then re-resolve), and error_state recovering from a real exception raised inside an async
action (a simulated ledger call), followed by a correction and retry.
Its graph shows both failure mechanisms side by side – POST’s guard chain rejecting
over-the-limit amounts up front, and the separate error-labeled edge from error_state
catching the ledger call’s exception:
stateDiagram-v2
[*] --> idle
idle --> txn_identify: START_TXN
txn_identify --> resolve_fields: IDENTIFY [txn_type_supported]
txn_identify --> resolution_failed: IDENTIFY [txn_type_unsupported]
resolve_fields --> posting: always [all_fields_resolved]
resolve_fields --> resolution_failed: always [has_missing_fields]
resolution_failed --> collect_data: PROVIDE_DATA
collect_data --> resolve_fields: always
posting --> posting_failed: POST [exceeds_daily_limit]
posting --> posting_pass: POST [within_daily_limit]
posting --> posting_failed: error
posting_failed --> collect_data: CORRECT
uv run python examples/bank.py
# ruff: noqa: ARG001
"""Runnable example: a bank teller's transaction-posting bot modeled as a StateMachine.
A teller walks a wire transfer through: identifying the transaction, resolving its required
fields (looping back to collect missing data from the teller when something's absent), and
posting it to the ledger -- including a real posting failure that gets corrected and retried.
Demonstrates every hook the engine has:
- `on` guard chains: `txn_identify` tries two candidates in order (supported vs. unsupported type).
- `always` auto-advance, including a *chain* of several always-transitions firing within a single
turn (`resolve_fields` -> `resolution_failed`, or `resolve_fields` -> `posting`).
- `error_state`: a real `RuntimeError` raised deep inside an async action (`call_ledger_api`) is
caught by the engine and routed to `posting_failed` automatically.
- `entry` actions used as teller-facing prompts (`prompt_confirm_post`, `ask_teller_for_missing_fields`).
- Both sync and async guards/actions, and a state (`collect_data`) re-entered from two different
places in the graph (`resolution_failed` and `posting_failed`), showing this is a real graph,
not a linear pipeline.
"""
from __future__ import annotations
import asyncio
import uuid
from dataclasses import dataclass, field
from statem import Context, Signal, StateMachine
SUPPORTED_TXN_TYPES = {"TRANSFER", "WITHDRAWAL", "DEPOSIT"}
REQUIRED_FIELDS = ("txn_type", "from_account", "to_account", "amount")
DAILY_LIMIT = 1000.0
FROZEN_ACCOUNTS = {"ACC-2002"}
@dataclass
class BankSession:
txn_id: str | None = None
txn_type: str | None = None
from_account: str | None = None
to_account: str | None = None
amount: float | None = None
missing_fields: list[str] = field(default_factory=list)
last_error: str | None = None
receipt_id: str | None = None
log: list[str] = field(default_factory=list)
def create_txn(ctx: Context[BankSession], signal: Signal) -> None:
ctx.session.txn_id = f"TXN-{uuid.uuid4().hex[:8].upper()}"
ctx.session.log.append(f"transaction {ctx.session.txn_id} opened")
def capture_identify_fields(ctx: Context[BankSession], signal: Signal) -> None:
ctx.session.txn_type = signal.data["txn_type"]
ctx.session.from_account = signal.data["from_account"]
ctx.session.log.append(f"identified as {ctx.session.txn_type} from {ctx.session.from_account}")
def log_unsupported_type(ctx: Context[BankSession], signal: Signal) -> None:
ctx.session.log.append(f"bot: sorry, {signal.data.get('txn_type')!r} is not a supported transaction type")
def run_field_resolution(ctx: Context[BankSession], signal: Signal) -> None:
session = ctx.session
session.missing_fields = [name for name in REQUIRED_FIELDS if getattr(session, name) is None]
def log_missing_fields(ctx: Context[BankSession], signal: Signal) -> None:
ctx.session.log.append(f"resolution incomplete, missing: {', '.join(ctx.session.missing_fields)}")
def ask_teller_for_missing_fields(ctx: Context[BankSession], signal: Signal) -> None:
if ctx.session.missing_fields:
ctx.session.log.append(f"bot: please provide {', '.join(ctx.session.missing_fields)}")
else:
ctx.session.log.append("bot: please provide a supported transaction type and try again")
def apply_teller_data(ctx: Context[BankSession], signal: Signal) -> None:
for key, value in signal.data.items():
setattr(ctx.session, key, value)
ctx.session.log.append(f"teller provided: {signal.data}")
def prompt_confirm_post(ctx: Context[BankSession], signal: Signal) -> None:
session = ctx.session
ctx.session.log.append(
f"bot: ready to post {session.txn_type} of {session.amount} from "
f"{session.from_account} to {session.to_account} -- confirm?"
)
def reject_over_limit(ctx: Context[BankSession], signal: Signal) -> None:
ctx.session.last_error = f"amount {ctx.session.amount} exceeds daily limit {DAILY_LIMIT}"
ctx.session.log.append(f"bot: rejected -- {ctx.session.last_error}")
async def call_ledger_api(ctx: Context[BankSession], signal: Signal) -> None:
await asyncio.sleep(0) # simulated network hop
if ctx.session.to_account in FROZEN_ACCOUNTS:
ctx.session.last_error = f"ledger rejected posting to {ctx.session.to_account} (account frozen)"
raise RuntimeError(ctx.session.last_error)
ctx.session.receipt_id = f"RCPT-{uuid.uuid4().hex[:8].upper()}"
def notify_failure(ctx: Context[BankSession], signal: Signal) -> None:
ctx.session.log.append(f"bot: posting failed -- {ctx.session.last_error}")
def print_receipt(ctx: Context[BankSession], signal: Signal) -> None:
session = ctx.session
ctx.session.log.append(
f"bot: posted! receipt {session.receipt_id} -- "
f"{session.txn_type} {session.amount} {session.from_account} -> {session.to_account}"
)
def txn_type_supported(ctx: Context[BankSession], signal: Signal) -> bool:
return signal.data.get("txn_type") in SUPPORTED_TXN_TYPES
def txn_type_unsupported(ctx: Context[BankSession], signal: Signal) -> bool:
return not txn_type_supported(ctx, signal)
def all_fields_resolved(ctx: Context[BankSession], signal: Signal) -> bool:
return not ctx.session.missing_fields
def has_missing_fields(ctx: Context[BankSession], signal: Signal) -> bool:
return bool(ctx.session.missing_fields)
def exceeds_daily_limit(ctx: Context[BankSession], signal: Signal) -> bool:
return ctx.session.amount is not None and ctx.session.amount > DAILY_LIMIT
async def within_daily_limit(ctx: Context[BankSession], signal: Signal) -> bool:
await asyncio.sleep(0) # simulated fraud/limits-service lookup
return not exceeds_daily_limit(ctx, signal)
CONFIG = {
"idle": {
"on": {"START_TXN": {"target": "txn_identify", "actions": ["create_txn"]}},
},
"txn_identify": {
"on": {
"IDENTIFY": [
{"target": "resolve_fields", "guard": "txn_type_supported", "actions": ["capture_identify_fields"]},
{"target": "resolution_failed", "guard": "txn_type_unsupported", "actions": ["log_unsupported_type"]},
]
},
},
"resolve_fields": {
"entry": ["run_field_resolution"],
"always": [
{"target": "posting", "guard": "all_fields_resolved"},
{"target": "resolution_failed", "guard": "has_missing_fields", "actions": ["log_missing_fields"]},
],
},
"resolution_failed": {
"entry": ["ask_teller_for_missing_fields"],
"on": {"PROVIDE_DATA": {"target": "collect_data", "actions": ["apply_teller_data"]}},
},
"collect_data": {
"always": [{"target": "resolve_fields"}],
},
"posting": {
"entry": ["prompt_confirm_post"],
"on": {
"POST": [
{"target": "posting_failed", "guard": "exceeds_daily_limit", "actions": ["reject_over_limit"]},
{"target": "posting_pass", "guard": "within_daily_limit", "actions": ["call_ledger_api"]},
]
},
"error_state": "posting_failed",
},
"posting_failed": {
"entry": ["notify_failure"],
"on": {"CORRECT": {"target": "collect_data", "actions": ["apply_teller_data"]}},
},
"posting_pass": {
"entry": ["print_receipt"],
},
}
ACTIONS = {
"create_txn": create_txn,
"capture_identify_fields": capture_identify_fields,
"log_unsupported_type": log_unsupported_type,
"run_field_resolution": run_field_resolution,
"log_missing_fields": log_missing_fields,
"ask_teller_for_missing_fields": ask_teller_for_missing_fields,
"apply_teller_data": apply_teller_data,
"prompt_confirm_post": prompt_confirm_post,
"reject_over_limit": reject_over_limit,
"call_ledger_api": call_ledger_api,
"notify_failure": notify_failure,
"print_receipt": print_receipt,
}
GUARDS = {
"txn_type_supported": txn_type_supported,
"txn_type_unsupported": txn_type_unsupported,
"all_fields_resolved": all_fields_resolved,
"has_missing_fields": has_missing_fields,
"exceeds_daily_limit": exceeds_daily_limit,
"within_daily_limit": within_daily_limit,
}
# (event, data, what the teller says this turn)
CONVERSATION = [
("START_TXN", {}, "I'd like to start a new transaction."),
("IDENTIFY", {"txn_type": "TRANSFER", "from_account": "ACC-1001"}, "It's a transfer from ACC-1001."),
("PROVIDE_DATA", {"amount": 500.0, "to_account": "ACC-2002"}, "Send $500 to ACC-2002."),
("POST", {}, "Go ahead and post it."),
("CORRECT", {"to_account": "ACC-3003"}, "Oh -- use ACC-3003 instead."),
("POST", {}, "Post it now."),
]
async def main() -> None:
machine = StateMachine.from_dict(CONFIG, action_dict=ACTIONS, guard_dict=GUARDS)
session = BankSession()
state = "idle"
cursor = 0
for event, data, teller_says in CONVERSATION:
print(f"\nTeller: {teller_says}")
state = await machine.run(
run_id="bank-demo-001",
state_name=state,
events=Signal(event=event, data=data),
session=session,
)
for line in session.log[cursor:]:
print(f" {line}")
cursor = len(session.log)
print(f" -> state: {state}")
print(f"\nFinal state: {state}")
print(f"Receipt: {session.receipt_id}")
if __name__ == "__main__":
asyncio.run(main())
Pizza order bot (examples/pizza.py, streaming)¶
The largest example: an order-to-delivery pizza bot (payment, quality-check retries, delivery
retries, CANCEL escape hatches at multiple points) that also demonstrates
stream() — run_streaming_demo() drives one signal through the machine and
prints the raw AG-UI event sequence (STEP_STARTED, STATE_SNAPSHOT, ACTIVITY_SNAPSHOT,
STATE_DELTA, STEP_FINISHED) as it happens, alongside two plain run() demos for comparison.
Requires the agui extra (pip install statem[agui]) for the streaming portion only.
stateDiagram-v2
[*] --> order_received
order_received --> payment_processing: CONFIRM
payment_processing --> preparing: always [payment_approved]
payment_processing --> payment_failed: always [payment_declined]
payment_processing --> payment_failed: error
payment_failed --> payment_processing: RETRY_PAYMENT
payment_failed --> cancelled: CANCEL
preparing --> baking: always [prep_complete]
baking --> quality_check: BAKING_DONE
quality_check --> ready: always [quality_passed]
quality_check --> remaking: always [quality_failed]
remaking --> preparing: always
ready --> assigning_driver: DISPATCH
assigning_driver --> out_for_delivery: always [driver_found]
out_for_delivery --> delivered: DELIVERED
out_for_delivery --> delivery_failed: FAILED
out_for_delivery --> delivery_failed: error
delivery_failed --> assigning_driver: RETRY_DELIVERY
delivery_failed --> cancelled: CANCEL
uv run python examples/pizza.py
# ruff: noqa: ARG001
"""Pizza Order Tracker - a Domino's-style order lifecycle modelled as a StateMachine
Demonstrates every engine feature on a problem everyone instantly recognises:
State graph (12 states)
-----------------------
stateDiagram-v2
[*] --> order_received
order_received --> payment_processing: CONFIRM
payment_processing --> preparing: payment_approved
payment_processing --> payment_failed: payment_declined
payment_processing --> payment_failed: error
payment_failed --> payment_processing: RETRY_PAYMENT
payment_failed --> cancelled: CANCEL
preparing --> baking: prep_complete
baking --> quality_check: BAKING_DONE
quality_check --> ready: quality_passed
quality_check --> remaking: quality_failed
remaking --> preparing: retry
ready --> assigning_driver: DISPATCH
assigning_driver --> out_for_delivery: driver_found
out_for_delivery --> delivered: DELIVERED
out_for_delivery --> delivery_failed: FAILED
out_for_delivery --> delivery_failed: error
delivery_failed --> assigning_driver: RETRY_DELIVERY
delivery_failed --> cancelled: CANCEL
delivered --> [*]
cancelled --> [*]
Interesting paths in the demo
-----------------------------
Run 1 (happy path with quality-fail loop):
CONFIRM -> [payment ok] -> preparing -> baking
BAKING_DONE -> quality_check [score 5 -> fail] -> remaking -> preparing -> baking
BAKING_DONE -> quality_check [score 9 -> pass] -> ready
DISPATCH -> [driver found] -> out_for_delivery
DELIVERED -> delivered ✓
Run 2 (payment declined -> retry -> success):
CONFIRM(cash) -> [payment declined] -> payment_failed
RETRY_PAYMENT(card) -> [payment ok] -> preparing -> ... -> delivered ✓
"""
from __future__ import annotations
import asyncio
import sys
import uuid
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
from statem import Context, Signal, StateMachine
from statem.diagram import to_mermaid
if TYPE_CHECKING:
from ag_ui.core import BaseEvent
if sys.stdout.encoding.lower() != "utf-8": # notes below use emoji; avoid UnicodeEncodeError on cp1252 consoles
sys.stdout.reconfigure(encoding="utf-8")
QUALITY_PASS_THRESHOLD = 7
# --- Session ---
@dataclass
class PizzaSession:
order_id: str | None = None
customer_name: str = ""
pizza: str = ""
address: str = ""
payment_method: str = "" # "card" | "cash"
payment_result: str | None = None # "approved" | "declined"
prep_status: str | None = None # "done" | None
quality_score: int = 0 # 0-10; ≥7 passes
quality_attempts: int = 0
driver_id: str | None = None
delivery_result: str | None = None
notes: list[str] = field(default_factory=list)
# --- Actions ---
def create_order(ctx: Context[PizzaSession], sig: Signal) -> str:
ctx.session.order_id = f"ORD-{uuid.uuid4().hex[:6].upper()}"
for k, v in sig.data.items():
setattr(ctx.session, k, v)
ctx.session.notes.append(f"order {ctx.session.order_id} created for {ctx.session.customer_name}")
return ctx.session.order_id
def notify_received(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append(
f"SMS -> {ctx.session.customer_name}: we got your {ctx.session.pizza} order! "
f"Delivering to {ctx.session.address}"
)
async def charge_card(ctx: Context[PizzaSession], sig: Signal) -> str:
await asyncio.sleep(0) # simulated gateway round-trip
if ctx.session.payment_method == "cash":
ctx.session.payment_result = "declined"
return "declined"
ctx.session.payment_result = "approved"
ctx.session.notes.append(f"gateway: {ctx.session.payment_method} charged ✓")
return "approved"
def notify_payment_success(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append("SMS - payment confirmed, kitchen notified")
def notify_payment_failure(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append("SMS - payment failed, please retry or cancel")
def reset_payment(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.payment_result = None
new_method = sig.data.get("payment_method", ctx.session.payment_method)
ctx.session.payment_method = new_method
ctx.session.notes.append(f"payment method updated to {new_method}")
def start_preparation(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.prep_status = "done" # kitchen is fast in this restaurant
ctx.session.notes.append(f"kitchen: started {ctx.session.pizza} (attempt {ctx.session.quality_attempts + 1})")
def log_prep_complete(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append("kitchen: prep done, going into oven")
def start_baking(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append("oven: baking started 🔥")
def run_quality_check(ctx: Context[PizzaSession], sig: Signal) -> int:
ctx.session.quality_attempts += 1
# First attempt: underbaked (score 5). Subsequent: perfect (score 9).
ctx.session.quality_score = 9 if ctx.session.quality_attempts > 1 else 5
ctx.session.notes.append(
f"QC attempt {ctx.session.quality_attempts}: score {ctx.session.quality_score}/10 - "
f"{'pass' if ctx.session.quality_score >= QUALITY_PASS_THRESHOLD else 'fail - remaking'}"
)
return ctx.session.quality_score
def log_quality_fail(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append("QC: sending back to kitchen for remake")
def reset_for_remake(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.prep_status = None # will be set again by start_preparation
ctx.session.quality_score = 0
def notify_ready(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append(f"SMS - {ctx.session.customer_name}: your {ctx.session.pizza} is ready! 🍕")
async def find_driver(ctx: Context[PizzaSession], sig: Signal) -> str:
await asyncio.sleep(0) # simulated dispatch API
ctx.session.driver_id = f"DRV-{uuid.uuid4().hex[:4].upper()}"
ctx.session.notes.append(f"dispatch: driver {ctx.session.driver_id} assigned")
return ctx.session.driver_id
def notify_dispatch(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append(f"SMS - {ctx.session.customer_name}: {ctx.session.driver_id} is on the way! 🛵")
def complete_delivery(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.delivery_result = "delivered"
ctx.session.notes.append(f"✓ delivered to {ctx.session.address} - enjoy your {ctx.session.pizza}!")
def handle_delivery_failure(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.delivery_result = "failed"
ctx.session.driver_id = None
ctx.session.notes.append(f"✗ delivery failed at {ctx.session.address} - notifying customer")
def reset_driver(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.driver_id = None
ctx.session.delivery_result = None
ctx.session.notes.append("retrying driver assignment")
def notify_cancellation(ctx: Context[PizzaSession], sig: Signal) -> None:
ctx.session.notes.append(f"SMS - {ctx.session.customer_name}: order cancelled, refund issued")
# --- Guards ---
def payment_approved(ctx: Context[PizzaSession], sig: Signal) -> bool:
return ctx.session.payment_result == "approved"
def payment_declined(ctx: Context[PizzaSession], sig: Signal) -> bool:
return ctx.session.payment_result == "declined"
def prep_complete(ctx: Context[PizzaSession], sig: Signal) -> bool:
return ctx.session.prep_status == "done"
def quality_passed(ctx: Context[PizzaSession], sig: Signal) -> bool:
return ctx.session.quality_score >= QUALITY_PASS_THRESHOLD
def quality_failed(ctx: Context[PizzaSession], sig: Signal) -> bool:
return 0 < ctx.session.quality_score < QUALITY_PASS_THRESHOLD
def driver_found(ctx: Context[PizzaSession], sig: Signal) -> bool:
return ctx.session.driver_id is not None
# --- Config ---
CONFIG: dict = {
"order_received": {
"on": {
"CONFIRM": {
"target": "payment_processing",
"actions": ["create_order", "notify_received"],
}
}
},
"payment_processing": {
"entry": ["charge_card"],
"always": [
{"target": "preparing", "guard": "payment_approved", "actions": ["notify_payment_success"]},
{"target": "payment_failed", "guard": "payment_declined", "actions": ["notify_payment_failure"]},
],
"error_state": "payment_failed",
},
"payment_failed": {
"on": {
"RETRY_PAYMENT": {"target": "payment_processing", "actions": ["reset_payment"]},
"CANCEL": {"target": "cancelled", "actions": ["notify_cancellation"]},
}
},
"preparing": {
"entry": ["start_preparation"],
"always": [
{"target": "baking", "guard": "prep_complete", "actions": ["log_prep_complete"]},
],
},
"baking": {
"entry": ["start_baking"],
"on": {"BAKING_DONE": {"target": "quality_check"}},
},
"quality_check": {
"entry": ["run_quality_check"],
"always": [
{"target": "ready", "guard": "quality_passed"},
{"target": "remaking", "guard": "quality_failed", "actions": ["log_quality_fail"]},
],
},
"remaking": {
"entry": ["reset_for_remake"],
"always": [{"target": "preparing"}],
},
"ready": {
"entry": ["notify_ready"],
"on": {"DISPATCH": {"target": "assigning_driver"}},
},
"assigning_driver": {
"entry": ["find_driver"],
"always": [
{"target": "out_for_delivery", "guard": "driver_found", "actions": ["notify_dispatch"]},
],
},
"out_for_delivery": {
"on": {
"DELIVERED": {"target": "delivered", "actions": ["complete_delivery"]},
"FAILED": {"target": "delivery_failed", "actions": ["handle_delivery_failure"]},
},
"error_state": "delivery_failed",
},
"delivered": {},
"delivery_failed": {
"on": {
"RETRY_DELIVERY": {"target": "assigning_driver", "actions": ["reset_driver"]},
"CANCEL": {"target": "cancelled", "actions": ["notify_cancellation"]},
}
},
"cancelled": {},
}
ACTIONS = {
"create_order": create_order,
"notify_received": notify_received,
"charge_card": charge_card,
"notify_payment_success": notify_payment_success,
"notify_payment_failure": notify_payment_failure,
"reset_payment": reset_payment,
"start_preparation": start_preparation,
"log_prep_complete": log_prep_complete,
"start_baking": start_baking,
"run_quality_check": run_quality_check,
"log_quality_fail": log_quality_fail,
"reset_for_remake": reset_for_remake,
"notify_ready": notify_ready,
"find_driver": find_driver,
"notify_dispatch": notify_dispatch,
"complete_delivery": complete_delivery,
"handle_delivery_failure": handle_delivery_failure,
"reset_driver": reset_driver,
"notify_cancellation": notify_cancellation,
}
GUARDS = {
"payment_approved": payment_approved,
"payment_declined": payment_declined,
"prep_complete": prep_complete,
"quality_passed": quality_passed,
"quality_failed": quality_failed,
"driver_found": driver_found,
}
# --- Demo runs ---
async def run_happy_path_with_quality_fail() -> None:
"""Happy path: Card payment, quality fails once -> remake -> pass -> delivered."""
print("\n" + "=" * 60)
print("RUN 1 - Happy path (quality fail -> remake loop)")
print("=" * 60)
machine = StateMachine.from_dict(CONFIG, action_dict=ACTIONS, guard_dict=GUARDS)
session = PizzaSession(customer_name="Alice")
state = "order_received"
cursor = 0
signals = [
(
"CONFIRM",
{"customer_name": "Alice", "pizza": "Pepperoni", "address": "42 Baker St", "payment_method": "card"},
),
("BAKING_DONE", {}), # first bake -> quality fail -> remake
("BAKING_DONE", {}), # second bake -> quality pass -> ready
("DISPATCH", {}),
("DELIVERED", {}),
]
for event, data in signals:
state = await machine.run(
state_name=state,
events=Signal(event, data),
session=session,
)
new_notes = session.notes[cursor:]
cursor = len(session.notes)
print(f"\n[{event}] -> {state}")
for note in new_notes:
print(f" {note}")
print(f"\n Final: {state} | order {session.order_id} | receipt via driver {session.driver_id}")
async def run_payment_declined_retry() -> None:
"""Payment declined on cash, retried with card, then delivered."""
print("\n" + "=" * 60)
print("RUN 2 - Payment declined -> retry with card")
print("=" * 60)
machine = StateMachine.from_dict(CONFIG, action_dict=ACTIONS, guard_dict=GUARDS)
session = PizzaSession(customer_name="Bob")
state = "order_received"
cursor = 0
signals = [
("CONFIRM", {"customer_name": "Bob", "pizza": "Margherita", "address": "7 Elm Rd", "payment_method": "cash"}),
("RETRY_PAYMENT", {"payment_method": "card"}),
("BAKING_DONE", {}),
("BAKING_DONE", {}), # quality fail -> remake path again
("DISPATCH", {}),
("DELIVERED", {}),
]
for event, data in signals:
state = await machine.run(
state_name=state,
events=Signal(event, data),
session=session,
)
new_notes = session.notes[cursor:]
cursor = len(session.notes)
print(f"\n[{event}] -> {state}")
for note in new_notes:
print(f" {note}")
print(f"\n Final: {state} | order {session.order_id}")
async def run_streaming_demo() -> None:
"""Show raw AG-UI event stream for one signal turn."""
print("\n" + "=" * 60)
print("RUN 3 - AG-UI stream() events for CONFIRM signal")
print("=" * 60)
machine = StateMachine.from_dict(CONFIG, action_dict=ACTIONS, guard_dict=GUARDS)
session = PizzaSession(customer_name="Carol")
async for event in machine.stream(
state_name="order_received",
events=Signal(
"CONFIRM",
{"customer_name": "Carol", "pizza": "BBQ Chicken", "address": "99 Pine Ave", "payment_method": "card"},
),
session=session,
run_id="pizza-stream-demo",
thread_id="thread-carol-001",
):
print(f" {event.type.value:<28} {_event_summary(event)}")
def _event_summary(event: BaseEvent) -> str:
from ag_ui.core import EventType # noqa: PLC0415 -- lazy so plain `import statem` never needs ag-ui-protocol
if event.type == EventType.STATE_SNAPSHOT:
return f"snapshot={event.snapshot}"
if event.type == EventType.STEP_STARTED:
return f"step={event.step_name}"
if event.type == EventType.STEP_FINISHED:
return f"step={event.step_name}"
if event.type == EventType.ACTIVITY_SNAPSHOT:
return event.content
if event.type == EventType.STATE_DELTA:
return str([f"{p['op']} {p['path']}={p['value']}" for p in event.delta])
return ""
async def main() -> None:
print(to_mermaid(StateMachine.from_dict(CONFIG, action_dict=ACTIONS, guard_dict=GUARDS), initial="order_received"))
await run_happy_path_with_quality_fail()
await run_payment_declined_retry()
await run_streaming_demo()
if __name__ == "__main__":
asyncio.run(main())
