Python StateMachine

https://python-statemachine.readthedocs.io/en/latest/
Admin
Python StateMachine is a library for creating finite state machines and statecharts in Python with a...
Tokens:213,485
Snippets:1,764
Trust Score:9.6
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### Define listeners with setup() and instantiate PersistentMachine

Source: https://python-statemachine.readthedocs.io/en/latest/listeners.html

Shows how to define listeners with a setup() method that receives runtime dependencies via dependency injection, and how to pass those dependencies as keyword arguments when creating the state machine. The example defines DBListener and CacheListener, a PersistentMachine with these listeners, and demonstrates that each listener receives only the kwargs it declares.

```python
>>> class DBListener:
...     def __init__(self):
...         self.session = None
...
...     def setup(self, sm, session=None, **kwargs):
...         self.session = session

>>> class CacheListener:
...     def __init__(self):
...         self.redis = None
...
...     def setup(self, sm, redis=None, **kwargs):
...         self.redis = redis

>>> class PersistentMachine(StateChart):
...     listeners = [DBListener, CacheListener]
...
...     s1 = State(initial=True)
...     s2 = State(final=True)
...     go = s1.to(s2)

>>> sm = PersistentMachine(session="db_conn", redis="redis_conn")
>>> sm.active_listeners[0].session
'db_conn'
>>> sm.active_listeners[1].redis
'redis_conn'
```

--------------------------------

### Install development dependencies

Source: https://python-statemachine.readthedocs.io/en/latest/contributing.html

Use uv to install all development dependencies and create a virtual environment.

```bash
 uv sync --all-extras
```

--------------------------------

### Install and initialize git-flow

Source: https://python-statemachine.readthedocs.io/en/latest/contributing.html

Prerequisites for managing releases using git-flow.

```bash
brew install git-flow   # macOS
git flow init           # use main for production, develop for next release
```

--------------------------------

### Create documents for incremental configuration example

Source: https://python-statemachine.readthedocs.io/en/latest/_downloads/3ff117b988d6a5994fc63b13701a599b/sqlite_persistent_model_machine.ipynb

Creates two Document instances with the ApprovalWorkflow, one for Alice and one for Bob, and verifies they start in the 'draft' state.

```python
db_inc = WorkflowDB()

alice = Document.create(db_inc, ApprovalWorkflow, "RFC-001: API Redesign", "Alice")
bob = Document.create(db_inc, ApprovalWorkflow, "RFC-002: DB Migration", "Bob")

print(f"Created: {alice}")
print(f"Created: {bob}")

assert alice.state == "draft"
assert bob.state == "draft"
```

--------------------------------

### Install with Poetry

Source: https://python-statemachine.readthedocs.io/en/latest/installation.html

Install the latest release of python-statemachine using Poetry.

```bash
poetry add python-statemachine

```

--------------------------------

### Install with pip

Source: https://python-statemachine.readthedocs.io/en/latest/installation.html

Install the latest release of python-statemachine using pip.

```bash
python3 -m pip install python-statemachine

```

--------------------------------

### Install core python-statemachine

Source: https://python-statemachine.readthedocs.io/en/latest?badge=latest

The standard command to install the base library.

```bash
pip install python-statemachine
```

--------------------------------

### Install with uv

Source: https://python-statemachine.readthedocs.io/en/latest/installation.html

Install the latest release of python-statemachine using the uv package manager.

```bash
uv add python-statemachine

```

--------------------------------

### Install from source in editable mode

Source: https://python-statemachine.readthedocs.io/en/latest/installation.html

Install the source code in editable mode after cloning or downloading the repository.

```bash
python3 -m pip install -e .

```

--------------------------------

### Run the Beacons of Gondor example with debug logging

Source: https://python-statemachine.readthedocs.io/en/latest/auto_examples/statechart_delayed_machine.html

Run the example with the -v flag to see the engine's macro/micro step debug log.

```bash
uv run python tests/examples/statechart_delayed_machine.py -v
```

--------------------------------

### Install python-statemachine with diagram support

Source: https://python-statemachine.readthedocs.io/en/latest?badge=latest

Installs the library with the diagrams extra, which requires Graphviz.

```bash
pip install python-statemachine[diagrams]
```

--------------------------------

### Install python-statemachine with IO extras

Source: https://python-statemachine.readthedocs.io/en/latest?badge=latest

Installs the library with IO extras for YAML support and validation.

```bash
pip install python-statemachine[io]
```

--------------------------------

### Instantiate machine and show initial output

Source: https://python-statemachine.readthedocs.io/en/latest/auto_examples/guess_the_number_machine.html

Instantiates the machine with a fixed seed and shows the initial output printed when entering the start state.

```python
sm = GuessTheNumberMachine(seed=103)
```

```text
(psss.. don't tell anyone the number is 4)
I'm thinking of a number between 1 and 5. Can you guess what it is?
```

--------------------------------

### Start the release branch

Source: https://python-statemachine.readthedocs.io/en/latest/contributing.html

Commands to prepare the release branch from the develop branch.

```bash
git checkout develop
git pull origin develop
git flow release start X.Y.Z
```

--------------------------------

### Expected output of AirConditioner example

Source: https://python-statemachine.readthedocs.io/en/latest/auto_examples/air_conditioner_machine.html

Shows the expected console output when running the example. It prints the initial message and then each transition with the event name, source state, target state, and the temperature value passed as a keyword argument.

```text
Will create AirConditioner machine
Running sensor_updated from Off to Off: {'temperature': 24}
Running sensor_updated from Off to Off: {'temperature': 15}
Running sensor_updated from Off to Off: {'temperature': 20}
Running sensor_updated from Off to Off: {'temperature': 15}
Running sensor_updated from Off to Off: {'temperature': 17}
Running sensor_updated from Off to Off: {'temperature': 16}
Running sensor_updated from Off to Off: {'temperature': 23}
Running sensor_updated from Off to Off: {'temperature': 15}
Running sensor_updated from Off to Off: {'temperature': 18}
Running sensor_updated from Off to Off: {'temperature': 22}
Running sensor_updated from Off to Cooling: {'temperature': 35}
Running sensor_updated from Cooling to Cooling: {'temperature': 30}
Running sensor_updated from Cooling to Cooling: {'temperature': 20}
Running sensor_updated from Cooling to Standby: {'temperature': 15}
Running sensor_updated from Standby to Cooling: {'temperature': 31}
Running sensor_updated from Cooling to Standby: {'temperature': 19}
Running sensor_updated from Standby to Cooling: {'temperature': 27}
Running sensor_updated from Cooling to Cooling: {'temperature': 35}
Running sensor_updated from Cooling to Cooling: {'temperature': 26}
Running sensor_updated from Cooling to Standby: {'temperature': 16}

```

--------------------------------

### Install Graphviz on Debian/Ubuntu

Source: https://python-statemachine.readthedocs.io/en/latest/diagram.html

Installs the `dot` command-line tool from Graphviz, required for diagram generation.

```bash
sudo apt install graphviz
```

--------------------------------

### Install python-statemachine with diagrams extra

Source: https://python-statemachine.readthedocs.io/en/latest/diagram.html

Installs the `diagrams` extra which includes pydot. Requires Graphviz's `dot` command-line tool separately.

```bash
pip install python-statemachine[diagrams]  # installs pydot
```

--------------------------------

### prepare_event callback example

Source: https://python-statemachine.readthedocs.io/en/latest/actions.html

Shows how to use the prepare_event callback to compute and inject additional keyword arguments into the transition. The callback runs before validators and conditions, and its returned dict is merged into the keyword arguments for subsequent callbacks. In this example, the order_total is computed from the order_id and passed to the on_confirm callback.

```python
>>> class OrderFlow(StateChart):
...     pending = State(initial=True)
...     confirmed = State(final=True)
...
...     confirm = pending.to(confirmed)
...
...     def prepare_event(self, order_id=None):
...         if order_id is not None:
...             return {"order_total": order_id * 10}
...         return {}
...
...     def on_confirm(self, order_total=0):
...         return f"confirmed ${order_total}"

>>> sm = OrderFlow()
>>> sm.send("confirm", order_id=5)
'confirmed $50'
```

--------------------------------

### Install pre-commit hooks

Source: https://python-statemachine.readthedocs.io/en/latest/contributing.html

Set up pre-commit validations to ensure code quality before committing.

```bash
 pre-commit install
```

--------------------------------

### Example output

Source: https://python-statemachine.readthedocs.io/en/latest/auto_examples/traffic_light_machine.html

Shows the expected console output when running the traffic light machine.

```text
Running cycle from green to yellow
Running cycle from yellow to red
Running cycle from red to green
Running cycle from green to yellow
Running cycle from yellow to red

```

--------------------------------

### Start QuestMachine and check initial active states

Source: https://python-statemachine.readthedocs.io/en/latest/_downloads/04a1af7738f2b8772cd36ffe0d0eab97/statechart_compound_machine.ipynb

Instantiates the QuestMachine and verifies that both the compound state 'shire' and its initial child 'bag_end' are active at startup.

```python
sm = QuestMachine()
print(f"Active states: {sorted(sm.configuration_values)}")
assert {"shire", "bag_end"} == set(sm.configuration_values)
```

--------------------------------

### Define transitions with from_() method

Source: https://python-statemachine.readthedocs.io/en/latest/how-to/coming_from_transitions.html

Shows the alternative from_() method to define transitions from the target's perspective. This example uses the same workflow but with a different syntax.

```python
>>> class Workflow2(StateChart):
...     draft = State(initial=True)
...     producing = State()
...     closed = State(final=True)
...     cancelled = State(final=True)
...
...     produce = draft.to(producing)
...     deliver = producing.to(closed)
...     cancel = cancelled.from_(draft, producing)

>>> sm = Workflow2()
>>> sm.send("produce")
>>> sm.send("cancel")
>>> sm.cancelled.is_active
True
```

--------------------------------

### Parallel states StateChart example

Source: https://python-statemachine.readthedocs.io/en/latest/diagram.html

Defines a parallel state (both) with two compound regions (left and right), each with initial and final child states. The machine starts in start, enters the parallel region, and ends in end.

```python
class ParallelSC(StateChart):
    class both(State.Parallel, name="Both"):
        class left(State.Compound, name="Left"):
            l1 = State(initial=True)
            l2 = State(final=True)
            go_l = l1.to(l2)

        class right(State.Compound, name="Right"):
            r1 = State(initial=True)
            r2 = State(final=True)
            go_r = r1.to(r2)

    start = State(initial=True)
    end = State(final=True)

    enter = start.to(both)
    done_state_both = both.to(end)

```

--------------------------------

### Serve documentation locally

Source: https://python-statemachine.readthedocs.io/en/latest/contributing.html

Serve the built documentation files using a local web server.

```bash
 python -m http.server --directory docs/_build/html
```

--------------------------------

### States.from_enum default change: After (3.0)

Source: https://python-statemachine.readthedocs.io/en/latest/releases/upgrade_2x_to_3.html

In 3.0, the default is use_enum_instance=True, so state values are enum instances. This example shows the new behavior where states.start.value equals MyEnum.start. To get raw values, pass use_enum_instance=False explicitly.

```python
states = States.from_enum(MyEnum, initial=MyEnum.start)
# states.start.value == MyEnum.start  (enum instance)

```

--------------------------------

### Initialize machine and show initial states

Source: https://python-statemachine.readthedocs.io/en/latest/auto_examples/statechart_in_condition_machine.html

Instantiates the machine and prints the initial configuration values. The output shows both initial states (shire_f and shire_s) are active along with the parallel and compound region states.

```python
sm = FellowshipMachine()
vals = set(sm.configuration_values)
print(f"Start: {sorted(vals)}")
assert "shire_f" in vals
assert "shire_s" in vals

```

```text
Start: ['frodo_path', 'quest', 'sam_path', 'shire_f', 'shire_s']

```

--------------------------------

### Chaining transitions with raise_()

Source: https://python-statemachine.readthedocs.io/en/latest/processing_model.html

Uses raise_() inside callbacks to enqueue events, allowing multiple transitions to fire from a single send() call. The example defines a Pipeline StateChart with states start, step1, step2, and done, and prints progress messages. All steps execute within one macrostep before control returns to the caller.

```python
>>> from statemachine import State, StateChart

>>> class Pipeline(StateChart):
...     start = State("Start", initial=True)
...     step1 = State("Step 1")
...     step2 = State("Step 2")
...     done = State("Done", final=True)
...
...     begin = start.to(step1)
...     advance_1 = step1.to(step2)
...     advance_2 = step2.to(done)
...
...     def on_enter_step1(self):
...         print("  step 1: extract")
...         self.raise_("advance_1")
...
...     def on_enter_step2(self):
...         print("  step 2: transform")
...         self.raise_("advance_2")
...
...     def on_enter_done(self):
...         print("  done: load complete")

>>> sm = Pipeline()
>>> sm.send("begin")
  step 1: extract
  step 2: transform
  done: load complete

>>> [s.id for s in sm.configuration]
['done']

```

--------------------------------

### Usage instructions for AI Shell

Source: https://python-statemachine.readthedocs.io/en/latest/_downloads/4ce333a6c7118347cd9e1fec05ae0439/ai_shell_machine.ipynb

Shows how to run the AI Shell example, including standalone execution, from the repo, and debug mode. Requires an OPENAI_API_KEY environment variable.

```bash
# Standalone (installs deps from PyPI)
OPENAI_API_KEY=sk-... uv run examples/ai_shell.py

# From the repo (uses local statemachine)
OPENAI_API_KEY=sk-... uv run --with openai python examples/ai_shell.py

# Debug mode — shows engine macro/micro step log on stderr
OPENAI_API_KEY=sk-... uv run --with openai python examples/ai_shell.py -v
```

--------------------------------

### Pass invoke class to State in StateChart

Source: https://python-statemachine.readthedocs.io/en/latest/invoke.html

Passes a class to the invoke parameter of a State, so each state machine instance gets a fresh handler. The example defines a CSVLoader state machine that uses FileReader to read a CSV file, stores the content in on_enter_ready, and verifies the machine reaches the ready state and the content is correct. It also cleans up the temporary file.

```python
>>> config_file = Path(tempfile.mktemp(suffix=".csv"))
>>> _ = config_file.write_text("name,age\nAlice,30\nBob,25\n")

>>> class CSVLoader(StateChart):
...     loading = State(initial=True, invoke=FileReader)
...     ready = State(final=True)
...     done_invoke_loading = loading.to(ready)
...
...     def __init__(self, file_path, **kwargs):
...         self.file_path = file_path
...         super().__init__(**kwargs)
...
...     def on_enter_ready(self, data=None, **kwargs):
...         self.content = data

>>> sm = CSVLoader(file_path=str(config_file))
>>> time.sleep(0.2)

>>> "ready" in sm.configuration_values
True
>>> sm.content
'name,age\nAlice,30\nBob,25\n'

>>> config_file.unlink()
```

--------------------------------

### Handle error.execution with error_ naming convention

Source: https://python-statemachine.readthedocs.io/en/latest/error_handling.html

Shows the error_ naming convention: an event attribute starting with error_ (here error_execution) matches the dotted event "error.execution". The example defines a state chart with a self-transition that raises RuntimeError and an error_execution transition to a final state, then verifies the machine ends in the broken state after sending "do_work".

```python
>>> from statemachine import State, StateChart

>>> class ResilientChart(StateChart):
...     operational = State(initial=True)
...     broken = State(final=True)
...
...     do_work = operational.to(operational, on="risky_action")
...     error_execution = operational.to(broken)
...
...     def risky_action(self):
...         raise RuntimeError("something went wrong")

>>> sm = ResilientChart()
>>> sm.send("do_work")
>>> "broken" in sm.configuration_values
True
```

--------------------------------

### Execute user workflow with event binding

Source: https://python-statemachine.readthedocs.io/en/latest/_downloads/fe2bb12be4d57fbaa17389d66a564851/user_machine.ipynb

Defines a main function that creates a User, attempts signup with an empty token (expecting an exception), then signs up with a valid token and upgrades the experience. It asserts the expected state changes and prints the experience values.

```python
def main():  # type: ignore[attr-defined]
    # By binding the events to the User model, the events can be fired directly from the model
    user = User(name="Frodo", email="frodo@lor.com")

    try:
        # Trying to signup with an empty token should raise an exception
        user.signup("")
    except Exception as e:
        print(e)

    assert user.verified is False

    user.signup("1234")

    assert user.status == UserStatus.signup_complete
    assert user.verified is True

    print(user.experience)
    user.upgrade()
    print(user.experience)


if __name__ == "__main__":
    main()
```

--------------------------------

### Run Scenario 1: All beacons lit before siege

Source: https://python-statemachine.readthedocs.io/en/latest/_downloads/802f165c8160509392435452bcaa8e39/statechart_delayed_machine.ipynb

Executes the machine's 'start' event, which runs the full beacon chain and results in the rohan_rides state. Prints scenario header and final configuration values.

```python
print("=== Scenario 1: Beacons lit in time ===")
sm = BeaconsMachine()
sm.send("start")
print(f"  Result: {sorted(sm.configuration_values)}")
assert "rohan_rides" in sm.configuration_values
```

--------------------------------

### Scenario 1: Beacons lit in time - BeaconsMachine

Source: https://python-statemachine.readthedocs.io/en/latest/auto_examples/statechart_delayed_machine.html

Runs the BeaconsMachine with a single 'start' event. The expected output shows all seven beacons lighting in sequence, the signal reaching Rohan, and the final configuration containing 'rohan_rides'.

```python
print("=== Scenario 1: Beacons lit in time ===")
sm = BeaconsMachine()
sm.send("start")
print(f"  Result: {sorted(sm.configuration_values)}")
assert "rohan_rides" in sm.configuration_values

```

```text
=== Scenario 1: Beacons lit in time ===
  Minas Tirith -- The beacon is lit!
  Amon Din -- The beacon is lit!
  Eilenach -- The beacon is lit!
  Nardol -- The beacon is lit!
  Erelas -- The beacon is lit!
  Min-Rimmon -- The beacon is lit!
  Calenhad -- The beacon is lit!
  Signal reaches Rohan!
  The beacons are answered! Rohan rides to aid!
  Result: ['rohan_rides']

```

--------------------------------

### Execute ApprovalMachine

Source: https://python-statemachine.readthedocs.io/en/latest/api.html

Demonstrates initializing the ApprovalMachine and transitioning through states.

```python
>>> sm = ApprovalMachine()
```

```python
>>> sm.pending.is_active
True
```

```python
>>> sm.send("finish")
Completed!
```

```python
>>> sm.completed.is_active
True
```

```python
>>> sm.current_state_value
<Status.completed: 2>
```

--------------------------------

### Embed statechart in docstring with {statechart:rst}

Source: https://python-statemachine.readthedocs.io/en/latest/diagram.html

Use this snippet to embed a live state machine representation in a class docstring using the 'rst' format, which is recommended for Sphinx with RST (autodoc default). The placeholder is expanded at class definition time, so Sphinx autodoc sees the final rendered text without extra configuration. The example shows a simple three-state machine with start and finish transitions.

```python
class SimpleSC(StateChart):
    """A simple three-state machine.

    {statechart:rst}
    """

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

    start = idle.to(running)
    finish = running.to(done)
```

--------------------------------

### Run user workflow and show expected output

Source: https://python-statemachine.readthedocs.io/en/latest/auto_examples/user_machine.html

Executes the workflow: creates a User, attempts signup with an empty token (raises ValueError), then signs up with a valid token and upgrades the experience. The output shows the listener messages and the final experience values.

```python
def main():  # type: ignore[attr-defined]
    # By binding the events to the User model, the events can be fired directly from the model
    user = User(name="Frodo", email="frodo@lor.com")

    try:
        # Trying to signup with an empty token should raise an exception
        user.signup("")
    except Exception as e:
        print(e)

    assert user.verified is False

    user.signup("1234")

    assert user.status == UserStatus.signup_complete
    assert user.verified is True

    print(user.experience)
    user.upgrade()
    print(user.experience)


if __name__ == "__main__":
    main()

```

```text
Before signup in Signup incomplete
Token is required
Before signup in Signup incomplete
Entering Signup complete from signup
UserExperience.basic
Before upgrade in Basic
Entering Premium from upgrade
UserExperience.premium

```

--------------------------------

### Build documentation

Source: https://python-statemachine.readthedocs.io/en/latest/contributing.html

Build the project documentation using sphinx-build.

```bash
 uv run sphinx-build docs docs/_build/html
```

--------------------------------

### Run the looping state machine

Source: https://python-statemachine.readthedocs.io/en/latest/auto_examples/recursive_event_machine.html

Create an instance of MyStateMachine and call the 'do_startup' event. The event is processed, and the loop continues until StopIteration is raised, which is caught and ignored. The output shows the sequence of state entries.

```python
sm = MyStateMachine()

try:
    sm.do_startup()
except StopIteration:
    pass

```

--------------------------------

### Optional dependencies for YAML and validation support

Source: https://python-statemachine.readthedocs.io/en/latest/releases/3.2.0.html

Install the optional extras to enable YAML format support (PyYAML) and validation with jsonschema. The 'io' extra installs both.

```text
python-statemachine[yaml]        # PyYAML, for the YAML format
python-statemachine[validation]  # jsonschema, for validate=True
python-statemachine[io]          # both of the above
```

--------------------------------

### YAML with before/on/after and Python lifecycle

Source: https://python-statemachine.readthedocs.io/en/latest/io/formats.html

Shows a YAML state machine defining before, on, and after callbacks for a transition, and a Python model with matching methods. The example loads the machine, sends the 'save' event, and verifies the callbacks run in order: validate, persist, notify.

```python
>>> sc = load(
...     """
...     states:
...       editing:
...         initial: true
...         transitions:
...           - event: save
...             target: saved
...             before: validate
...             on: persist
...             after: notify
...       saved:
...         final: true
...     """,
...     format="yaml",
...     validate=True,
... )

>>> class Doc:
...     def __init__(self):
...         self.steps = []
...     def validate(self):
...         self.steps.append("validate")
...     def persist(self):
...         self.steps.append("persist")
...     def notify(self):
...         self.steps.append("notify")

>>> doc = Doc()
>>> sm = sc(model=doc)
>>> _ = sm.send("save")
>>> doc.steps
['validate', 'persist', 'notify']
```

--------------------------------

### AI Shell: Full Python script for an interactive coding assistant

Source: https://python-statemachine.readthedocs.io/en/latest/_downloads/4ce333a6c7118347cd9e1fec05ae0439/ai_shell_machine.ipynb

This is the complete example script from the notebook. It defines an AIShell state machine with states for idle, responding, recovering, deep, and conversation_ended, and uses OpenAI's chat completions with tool calling. The main() function checks for the openai package, prints usage info, and runs an interactive loop that sends user messages to the state machine. It includes retry logic for API errors and a warning when the conversation context gets long.

```python
import json
import sys
import time
from threading import Event

from openai import OpenAI

from statemachine import State, StateMachine
from statemachine.contrib.diagram import DotGraphMachine


MAX_RETRIES = 3

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "run_shell",
            "description": "Run a shell command and return its output.",
            "parameters": {
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "The shell command to run.",
                    }
                },
                "required": ["command"],
            },
        },
    }
]

SM_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "sm_get_state",
            "description": "Get the current state of the state machine.",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "sm_send",
            "description": "Send an event to the state machine.",
            "parameters": {
                "type": "object",
                "properties": {
                    "event": {
                        "type": "string",
                        "description": "The event name.",
                    },
                    "data": {
                        "type": "object",
                        "description": "Optional data to pass to the event.",
                    },
                },
                "required": ["event"],
            },
        },
    },
]


def execute_tool(name, args, sm=None):
    """Execute a tool call and return the result as a string."""
    if name == "run_shell":
        import subprocess

        result = subprocess.run(
            args["command"],
            shell=True,
            capture_output=True,
            text=True,
        )
        return result.stdout or result.stderr
    elif name == "sm_get_state":
        return sm.current_state.id if sm else "unknown"
    elif name == "sm_send":
        if sm:
            sm.send(args["event"], **args.get("data", {}))
            return "ok"
        return "error: no state machine"
    return f"Unknown tool: {name}"


class AIShell(StateMachine):
    """A state machine that represents the AI assistant's conversation flow."""

    idle = State(initial=True)
    responding = State()
    recovering = State()
    deep = State()
    conversation_ended = State(final=True)

    start = idle.to(responding, cond="has_user_message")
    respond = responding.to(responding, cond="has_user_message")
    respond = responding.to(idle, cond="no_user_message")
    error = responding.to(recovering, on="on_enter_recovering")
    retry = recovering.to(responding, after="on_enter_responding")
    give_up = recovering.to(conversation_ended)
    deepen = responding.to(deep, cond="is_deep")
    back = deep.to(responding)
    end = responding.to(conversation_ended, cond="is_final")

    def __init__(self):
        self.client = OpenAI()
        self.messages = [
            {
                "role": "system",
                "content": "You are a helpful coding assistant. You can run shell commands and inspect the state machine.",
            }
        ]
        self._last_text = ""
        self._retries = 0
        self._ready = Event()
        self._ready.set()
        super().__init__()

    def has_user_message(self, text=None, **kwargs):
        return bool(text)

    def no_user_message(self, text=None, **kwargs):
        return not text

    def is_deep(self, **kwargs):
        return len(self.messages) > 10

    def is_final(self, **kwargs):
        return self._last_text and "goodbye" in self._last_text.lower()

    def on_enter_responding(self, **kwargs):
        """Send the conversation to the API and handle the response."""
        self._ready.clear()
        try:
            with Spinner():
                response = self.client.chat.completions.create(
                    model="gpt-4o-mini",
                    messages=self.messages,
                    tools=TOOLS + SM_TOOLS,
                )
        except Exception as e:
            self.send("error", error=str(e))
            return

        message = response.choices[0].message
        self.messages.append(message)

        if not message.tool_calls:
            self._last_text = message.content or ""

        # Process tool calls if any
        if message.tool_calls:
            for call in message.tool_calls:
                args = json.loads(call.function.arguments)
                print(f"  [tool] {call.function.name}({json.dumps(args)})")
                result = execute_tool(call.function.name, args, sm=self)
                self.messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": call.id,
                        "content": result,
                    }
                )
            # After tool calls, continue the conversation
            self.send("respond")
        else:
            self.send("respond")

    def on_enter_idle(self, **kwargs):
        """Reset retry counter and signal readiness when returning to idle."""
        self._retries = 0
        self._ready.set()

    def on_enter_recovering(self, **kwargs):
        """Handle API errors with retry logic (via error.execution)."""
        self._retries += 1
        if self._retries < MAX_RETRIES:
            print(f"\n  [error] API call failed, retrying ({self._retries}/{MAX_RETRIES})...")
            self.send("retry")
        else:
            print(f"\n  [error] API call failed after {MAX_RETRIES} attempts. Giving up.")
            self.send("give_up")

    def on_enter_deep(self, **kwargs):
        """Warn when conversation context is getting long."""
        print("  [context] Conversation is getting long — responses may degrade.")

    def on_enter_conversation_ended(self, **kwargs):
        print("\nGoodbye!")


# ---------------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------------


def _check_openai():
    """Return True if the openai package is available."""
    try:
        import openai  # noqa: F401

        return True
    except ImportError:
        return False


def main():
    if not _check_openai():
        print("This example requires the 'openai' package.")
        print("Install it with: pip install openai")
        return

    print("AI Shell")
    print("A coding assistant powered by python-statemachine + OpenAI.")
    print("Type 'bye', 'exit', or 'quit' to end. Ctrl+C to interrupt.")
    if "-v" in sys.argv or "--verbose" in sys.argv:
        print("Debug mode enabled — engine log is written to stderr.\n")
    else:
        print("Tip: run with -v to see engine macro/micro step debug log.\n")

    try:
        sm = AIShell()
    except Exception as e:
        sys.exit(f"Error initializing: {e}")

    while not sm.is_terminated:
        sm._ready.wait()
        sm._ready.clear()
        try:
            text = input("> ")
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if text.strip():
            sm.send("user_message", text=text)


if __name__ == "__main__" and "sphinx" not in sys.modules:  # pragma: no cover
    main()
```

--------------------------------

### Get dot representation of state machine

Source: https://python-statemachine.readthedocs.io/en/latest/diagram.html

Shows how to get the dot representation of a state machine instance using the _graph().to_string() method. The output is a digraph in DOT language.

```python
>>> from tests.examples.order_control_machine import OrderControl
>>> sm = OrderControl()
>>> print(sm._graph().to_string())
digraph OrderControl {
...
}

```

--------------------------------

### Validate JSON document at load time

Source: https://python-statemachine.readthedocs.io/en/latest/io/json_schema.html

Pass validate=True to load() to validate a document against the schema before building it. Requires the optional [validation] extra (pip install "python-statemachine[validation]", which installs jsonschema).

```python
>>> from statemachine.io import load

>>> sc = load(
...     '{"name": "M", "states": {"a": {"initial": true}}}',
...     format="json",
...     validate=True,
... )
>>> sc.__name__
'M'
```

--------------------------------

### Transition.__init__

Source: https://python-statemachine.readthedocs.io/en/latest/api.html

Creates a Transition instance with the given source and target states, event triggers, and optional callbacks for validation, conditions, and execution hooks.

```APIDOC
## Transition.__init__

### Description
Creates a Transition instance that holds a reference to the source and target states, and configures event triggers and callbacks.

### Method
Constructor

### Signature
`Transition(source: State, target: State | list[State] | None = None, event=None, internal=False, initial=False, validators=None, cond=None, unless=None, on=None, before=None, after=None)`

### Parameters
- **source** (State) - Required - The origin state of the transition.
- **target** (State | list[State] | None) - Optional - The target state(s) of the transition. Can be a single State, a list of states (for multi-target transitions), or None (targetless transition).
- **event** (Optional[Union[str, List[str]]]) - Optional - List of designators of events that trigger this transition. Can be either a list of strings, or a space-separated string list of event descriptors.
- **internal** (bool) - Optional - Is the transition internal or external? Internal transitions don't execute the state entry/exit actions. Default `False`.
- **validators** (Optional[Union[str, Callable, List[Callable]]]) - Optional - The validation callbacks to be invoked before the transition is executed.
- **cond** (Optional[Union[str, Callable, List[Callable]]]) - Optional - The condition callbacks to be invoked before the transition is executed that should evaluate to True.
- **unless** (Optional[Union[str, Callable, List[Callable]]]) - Optional - The condition callbacks to be invoked if the cond is False before the transition is executed.
- **on** (Optional[Union[str, Callable, List[Callable]]]) - Optional - The callbacks to be invoked when the transition is executed.
- **before** (Optional[Union[str, Callable, List[Callable]]]) - Optional - The callbacks to be invoked before the transition is executed.
- **after** (Optional[Union[str, Callable, List[Callable]]]) - Optional - The callbacks to be invoked after the transition is executed.

### Returns
A new Transition instance.
```

--------------------------------

### Typed model example with StateChart generic

Source: https://python-statemachine.readthedocs.io/en/latest/models.html

Shows how to declare a typed model and pass it as a type parameter to StateChart, enabling type inference for sm.model. The example defines an OrderModel and OrderWorkflow, then sends a 'confirm' event.

```python
>>> from statemachine import State, StateChart

>>> class OrderModel:
...     order_id: str = ""
...     total: float = 0.0
...     def confirm(self):
...         return f"Order {self.order_id} confirmed: ${self.total}"

>>> class OrderWorkflow(StateChart["OrderModel"]):
...     draft = State(initial=True)
...     confirmed = State(final=True)
...     confirm = draft.to(confirmed, on="on_confirm")
...     def on_confirm(self):
...         return self.model.confirm()

>>> model = OrderModel()
>>> model.order_id = "A-123"
>>> model.total = 49.90
>>> sm = OrderWorkflow(model=model)

>>> sm.send("confirm")
'Order A-123 confirmed: $49.9'
```

--------------------------------

### Install optional extras with pip

Source: https://python-statemachine.readthedocs.io/en/latest/installation.html

Install optional extras for python-statemachine. The [diagrams] extra requires the Graphviz system package. The [yaml], [validation], and [io] extras support declarative loaders; loading JSON needs no extra.

```bash
python3 -m pip install "python-statemachine[diagrams]"    # pydot, to generate diagrams from your machines
python3 -m pip install "python-statemachine[yaml]"        # PyYAML, to load YAML statechart documents
python3 -m pip install "python-statemachine[validation]"  # jsonschema, to validate documents (validate=True)
python3 -m pip install "python-statemachine[io]"          # PyYAML + jsonschema, the full JSON/YAML IO stack

```

--------------------------------

### Main game loop with async and sync variants

Source: https://python-statemachine.readthedocs.io/en/latest/_downloads/8e5c9e1579aecf121966d9d7048ed4da/async_guess_the_number_machine.ipynb

Defines main_async() and main_sync() to run the game. The async version uses connect_stdin_stdout() and manually activates the initial state with await sm.activate_initial_state(). The sync version uses print and sys.stdin. The entry point checks for '-i' in sys.argv and runs the appropriate variant based on '-s'.

```python
async def main_async():
    reader, writer = await connect_stdin_stdout()
    sm = GuessTheNumberMachine(
        lambda s: writer.write(b"\n" + s.encode("utf-8")), seed=random.randint(1, 1000)
    )
    await sm.activate_initial_state()
    while not sm.is_terminated:
        res = await reader.read(100)
        if not res:
            break
        await sm.send("guess", int(res))
        await writer.drain()
    writer.close()


def main_sync():
    sm = GuessTheNumberMachine(print, seed=random.randint(1, 1000))
    sm.activate_initial_state()
    while not sm.is_terminated:
        res = sys.stdin.readline()
        if not res:
            break
        sm.send("guess", int(res))


if __name__ == "__main__" and "-i" in sys.argv:
    if "-s" in sys.argv:
        main_sync()
    else:
        asyncio.run(main_async())
```

--------------------------------

### In(state) condition check example

Source: https://python-statemachine.readthedocs.io/en/latest/releases/3.0.0.html

Shows how to use the 'In('<state>')' syntax in conditions to check if a state is in the current configuration, useful in parallel state machines. The example demonstrates a spaceship launch that requires both engine on and hatch sealed.

```python
>>> from statemachine import State, StateChart

>>> class Spaceship(StateChart):
...     class systems(State.Parallel):
...         class engine(State.Compound):
...             off = State(initial=True)
...             on = State()
...             ignite = off.to(on)
...         class hatch(State.Compound):
...             open = State(initial=True)
...             sealed = State()
...             seal = open.to(sealed)
...     orbit = State(final=True)
...     launch = systems.to(orbit, cond="In('on') and In('sealed')")

>>> sm = Spaceship()
>>> sm.send("launch")  # engine off, hatch open — guard fails
>>> "off" in sm.configuration_values
True

>>> sm.send("ignite")
>>> sm.send("launch")  # engine on, hatch open — guard fails
>>> "on" in sm.configuration_values and "open" in sm.configuration_values
True

>>> sm.send("seal")
>>> sm.send("launch")  # both conditions met — launches!
>>> sm.is_terminated
True


```

--------------------------------

### Deep History State Machine Example

Source: https://python-statemachine.readthedocs.io/en/latest/_downloads/22f129a947492736994a36e04064178c/statechart_history_machine.ipynb

Defines a state machine with nested compounds and deep history. Use this to see how to create a deep history state and restore the inner state after escaping. The example prints the configuration values after each transition and asserts the expected states.

```python
class DeepPersonalityMachine(StateChart):
    """A machine with nested compounds and deep history."""

    class realm(State.Compound):
        class inner(State.Compound):
            entrance = State("Entrance", initial=True)
            chamber = State("Chamber")

            explore = entrance.to(chamber)

        assert isinstance(inner, State)
        h = HistoryState(type="deep")  # type: ignore[has-type]
        bridge = State("Bridge", final=True)
        flee = inner.to(bridge)

    outside = State("Outside")
    escape = realm.to(outside)
    return_deep = outside.to(realm.h)  # type: ignore[has-type]


sm2 = DeepPersonalityMachine()
print(f"\nDeep history initial: {sorted(sm2.configuration_values)}")
assert "entrance" in sm2.configuration_values

# Move to the inner leaf state
sm2.send("explore")
print(f"Explored chamber: {sorted(sm2.configuration_values)}")
assert "chamber" in sm2.configuration_values

# Exit and return via deep history
sm2.send("escape")
print(f"Escaped: {sorted(sm2.configuration_values)}")
assert {"outside"} == set(sm2.configuration_values)

sm2.send("return_deep")
print(f"Deep history restored: {sorted(sm2.configuration_values)}")
assert "chamber" in sm2.configuration_values
assert "inner" in sm2.configuration_values
assert "realm" in sm2.configuration_values
```

--------------------------------

### Build documentation with live reloading

Source: https://python-statemachine.readthedocs.io/en/latest/contributing.html

Use sphinx-autobuild for a better documentation writing experience with live reloading.

```bash
 uv run sphinx-autobuild docs docs/_build/html --re-ignore "auto_examples/.*"
```
Tokens:
Raw