Actions

Side effects on entry, exit, and transitions — logging, context updates, notifications.

36 min read

Actions are side effects that execute at specific moments in a state machine’s lifecycle. They don’t control flow — they do things: update context, log messages, send notifications, or trigger external systems.

🎬 What are Actions?

An action is a callable that the interpreter invokes at a well-defined point during a transition. Actions are the primary mechanism for making your state machine do something beyond simply switching states.

Key characteristics:

  • Fire-and-forget — actions don’t return meaningful values.
  • Context-mutating — actions are the only place you should modify context.
  • Deterministic ordering — the interpreter runs actions in a predictable, documented order.
  • Synchronous (in SyncInterpreter) — async actions raise NotSupportedError.

✍️ Action Signature

def my_action(interpreter, context, event, action_def) -> None:
    ...
Parameter Type Description
interpreter SyncInterpreter or Interpreter The running interpreter instance
context dict The machine’s mutable context dictionary
event Event The event that triggered this action
action_def ActionDefinition Metadata about the action (name, params)

Note: The action signature (interpreter, context, event, action_def) is different from the guard signature (context, event). Actions get the interpreter and action definition; guards do not.

⏰ When Actions Run

Trigger When It Fires Defined In
Entry actions When a state is entered "entry" on the state
Exit actions When a state is exited "exit" on the state
Transition actions During a transition (between exit and entry) "actions" on the transition

🔢 Execution Order

When a transition fires from state A to state B, actions execute in this strict order:

sequenceDiagram
    participant E as 📨 event
    participant A as state A (source)
    participant T as transition
    participant B as state B (target)
    E->>A: SUBMIT arrives
    A->>A: ① exit actions
    A->>T: leave A
    T->>T: ② transition actions
    T->>B: enter B
    B->>B: ③ entry actions

This order is guaranteed and consistent across all interpreter types.

# Example: editing → submitting
# 1. saveDraft     (exit action on "editing")
# 2. validate      (transition action)
# 3. clearErrors   (transition action)
# 4. showSpinner   (entry action on "submitting")

📄 JSON Actions

Single Action (String)

{
  "on": {
    "SUBMIT": {
      "target": "submitting",
      "actions": "validate"
    }
  }
}

Multiple Actions (Array)

{
  "on": {
    "SUBMIT": {
      "target": "submitting",
      "actions": ["validate", "clearErrors", "logSubmission"]
    }
  }
}

Entry and Exit Actions on States

{
  "states": {
    "editing": {
      "entry": "loadDraft",
      "exit": "saveDraft",
      "on": {
        "SUBMIT": {
          "target": "submitting",
          "actions": ["validate", "clearErrors"]
        }
      }
    },
    "submitting": {
      "entry": "showSpinner"
    }
  }
}

Multiple Entry/Exit Actions

Entry and exit support arrays too:

{
  "states": {
    "editing": {
      "entry": ["loadDraft", "startAutoSave"],
      "exit":  ["saveDraft", "stopAutoSave"]
    }
  }
}

🧠 Action Implementation with MachineLogic

When using JSON configuration, implement actions as methods on a MachineLogic subclass:

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "formMachine",
    "initial": "editing",
    "context": {"draft": "", "errors": [], "isValid": False},
    "states": {
        "editing": {
            "entry": "loadDraft",
            "exit": "saveDraft",
            "on": {
                "SUBMIT": {
                    "target": "submitting",
                    "actions": ["validate", "clearErrors"]
                }
            }
        },
        "submitting": {
            "entry": "showSpinner",
            "on": {
                "SUCCESS": {"target": "done"},
                "FAILURE": {"target": "editing", "actions": "setError"}
            }
        },
        "done": {"type": "final"}
    }
}

class FormLogic(MachineLogic):
    def load_draft(self, interpreter, context, event, action_def):
        context["draft"] = "Loaded from storage"
        print("Loading saved draft...")

    def save_draft(self, interpreter, context, event, action_def):
        print(f"Auto-saving draft: '{context['draft']}'")

    def validate(self, interpreter, context, event, action_def):
        context["isValid"] = len(context["draft"]) > 0
        print(f"Validating... valid={context['isValid']}")

    def clear_errors(self, interpreter, context, event, action_def):
        context["errors"] = []
        print("Errors cleared.")

    def show_spinner(self, interpreter, context, event, action_def):
        print("Showing loading spinner...")

    def set_error(self, interpreter, context, event, action_def):
        error_msg = event.payload.get("message", "Unknown error")
        context["errors"].append(error_msg)
        print(f"Error: {error_msg}")

machine = create_machine(config, logic=FormLogic())
interp = SyncInterpreter(machine).start()
# Output: Loading saved draft...

interp.send("SUBMIT")
# Output (in order):
#   Auto-saving draft: 'Loaded from storage'   (exit: save_draft)
#   Validating... valid=True                     (transition: validate)
#   Errors cleared.                              (transition: clear_errors)
#   Showing loading spinner...                   (entry: show_spinner)

interp.send("SUCCESS")
interp.stop()

🐍 Pythonic Actions

@action Decorator

The @action decorator marks a function as a state machine action:

from xstate_statemachine import action

@action
def increment_counter(interpreter, context, event, action_def):
    context["count"] += 1
# Registered as "incrementCounter" (auto snake_case → camelCase)

Explicit Naming

Override the auto-generated name:

@action("myCustomAction")
def some_function(interpreter, context, event, action_def):
    print("Custom action executed")
# Registered as "myCustomAction"

@state.enter — Entry Actions (Class-Based)

Register entry actions using the @state.enter decorator:

from xstate_statemachine import State, StateMachine, SyncInterpreter

class FormMachine(StateMachine):
    machine_id = "form"

    editing    = State("editing", initial=True)
    submitting = State("submitting")

    submit = editing.to(submitting, event="SUBMIT")

    @editing.enter
    def on_enter_editing(self, interpreter, context, event, action_def):
        print("Entered editing mode")

    @submitting.enter
    def on_enter_submitting(self, interpreter, context, event, action_def):
        print("Entered submitting mode")

machine = FormMachine.create_machine()
interp = SyncInterpreter(machine).start()
# Output: Entered editing mode

interp.send("SUBMIT")
# Output: Entered submitting mode
interp.stop()

@state.exit — Exit Actions (Class-Based)

Register exit actions using the @state.exit decorator:

class FormMachine(StateMachine):
    machine_id = "form"

    editing    = State("editing", initial=True)
    submitting = State("submitting")

    submit = editing.to(submitting, event="SUBMIT")

    @editing.exit
    def on_exit_editing(self, interpreter, context, event, action_def):
        print("Left editing mode — auto-saving draft...")

    @editing.enter
    def on_enter_editing(self, interpreter, context, event, action_def):
        print("Entered editing mode")

Transition Actions in .to()

Attach actions directly to transitions using the actions parameter:

class FormMachine(StateMachine):
    machine_id = "form"

    editing    = State("editing", initial=True)
    submitting = State("submitting")

    submit = editing.to(
        submitting,
        event="SUBMIT",
        actions=["validate", "clearErrors"]
    )

📝 Modifying Context in Actions

Actions are the designated place to mutate context. Modify the dictionary directly:

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "todoApp",
    "initial": "active",
    "context": {"todos": [], "completedCount": 0},
    "states": {
        "active": {
            "on": {
                "ADD_TODO":      {"actions": "addTodo"},
                "COMPLETE_TODO": {"actions": "completeTodo"}
            }
        }
    }
}

class TodoLogic(MachineLogic):
    def add_todo(self, interpreter, context, event, action_def):
        title = event.payload.get("title", "Untitled")
        context["todos"].append({"title": title, "done": False})

    def complete_todo(self, interpreter, context, event, action_def):
        index = event.payload.get("index", 0)
        if 0 <= index < len(context["todos"]):
            context["todos"][index]["done"] = True
            context["completedCount"] += 1

machine = create_machine(config, logic=TodoLogic())
interp = SyncInterpreter(machine).start()

interp.send("ADD_TODO", title="Buy milk")
interp.send("ADD_TODO", title="Write docs")
interp.send("COMPLETE_TODO", index=0)

print(interp.context["todos"])
# [{"title": "Buy milk", "done": True}, {"title": "Write docs", "done": False}]
print(interp.context["completedCount"])  # 1
interp.stop()

📨 Accessing Event Data in Actions

The event parameter carries the payload that was sent with the event:

class Logic(MachineLogic):
    def store_user(self, interpreter, context, event, action_def):
        # Access payload sent via: interp.send("LOGIN", username="alice", role="admin")
        context["username"] = event.payload.get("username", "unknown")
        context["role"] = event.payload.get("role", "guest")

For DoneEvent from services, the result is on event.data:

class Logic(MachineLogic):
    def save_result(self, interpreter, context, event, action_def):
        # event.data holds the service's return value
        context["result"] = event.data

🔗 Multiple Actions on One Transition

When multiple actions are defined on a single transition, they execute in order, left-to-right:

"SUBMIT": {
  "target": "submitted",
  "actions": ["validate", "sanitize", "log", "submit"]
}

class Logic(MachineLogic):
    def validate(self, interpreter, context, event, action_def):
        print("1. Validating...")

    def sanitize(self, interpreter, context, event, action_def):
        print("2. Sanitizing...")

    def log(self, interpreter, context, event, action_def):
        print("3. Logging...")

    def submit(self, interpreter, context, event, action_def):
        print("4. Submitting...")

# Output when SUBMIT fires:
# 1. Validating...
# 2. Sanitizing...
# 3. Logging...
# 4. Submitting...

🧾 Action Definition (action_def Parameter)

The action_def parameter is an ActionDefinition object that carries metadata about the action:

Attribute Type Description
action_def.type str The action’s registered name (e.g., "addOne")
action_def.params dict or None Static parameters from the JSON config

Using action_def.params

You can define static parameters in the JSON config:

{
  "actions": {
    "type": "showNotification",
    "params": {
      "message": "Form submitted successfully!",
      "level": "success"
    }
  }
}

class Logic(MachineLogic):
    def show_notification(self, interpreter, context, event, action_def):
        msg = action_def.params.get("message", "")
        level = action_def.params.get("level", "info")
        print(f"[{level.upper()}] {msg}")

🧰 Built-in Action Creators (v0.6.0)

You rarely need to hand-write these. Import them and use them directly in a config — each returns a plain action definition, so they also work as raw JSON.

Creator Does
assign Update context
log Emit a structured log line
raise_ Send an event to this machine
send_to Send to another actor by id or systemId
send_parent Send to the machine that spawned this one
choose Run the first action list whose guard passes
pure Compute actions from context at runtime
enqueue_actions Queue actions imperatively in a callback
spawn_child / stop_child Start / stop a child actor
cancel Cancel a delayed send_to
emit Emit an event to external subscribers
escalate Raise an error to the parent
forward_to Forward the current event to another actor

choose — branch on guards

from xstate_statemachine import (
    create_machine, SyncInterpreter, MachineLogic, assign, choose,
)

config = {
    "id": "m",
    "initial": "a",
    "context": {"vip": True, "tier": ""},
    "states": {
        "a": {"on": {"CLASSIFY": {"actions": choose([
            {"guard": "isVip", "actions": assign({"tier": lambda x: "gold"})},
            {"actions": assign({"tier": lambda x: "standard"})},
        ])}}}
    },
}

logic = MachineLogic(guards={"isVip": lambda ctx, e: ctx["vip"]})
interp = SyncInterpreter(create_machine(config, logic=logic)).start()

interp.send("CLASSIFY")
print(interp.context["tier"])      # gold

enqueue_actions — imperative queueing

The most flexible creator: it subsumes both pure and choose. Your callback receives one mapping with context, event, enqueue, check and self:

from xstate_statemachine import create_machine, SyncInterpreter, enqueue_actions

def build_queue(args):
    enqueue = args["enqueue"]
    enqueue.assign({"n": lambda x: x["context"]["n"] + 10})
    if args["context"]["n"] == 0:
        enqueue.raise_({"type": "NEXT"})

config = {
    "id": "m",
    "initial": "a",
    "context": {"n": 0},
    "states": {
        "a": {"on": {"E": {"actions": enqueue_actions(build_queue)},
                     "NEXT": "b"}},
        "b": {},
    },
}

interp = SyncInterpreter(create_machine(config)).start()
interp.send("E")
print(interp.context, interp.current_state_ids)   # {'n': 10} {'m.b'}

The enqueue object exposes assign, raise_, send_to, send_parent, spawn_child, stop_child, emit, log and cancel.

Note: the callback takes a single argument (the mapping), not separate positional parameters.

🚨 Error Handling in Actions

If an action raises, the interpreter contains the error: it is logged, the transition still completes, and the machine keeps running. A single buggy side effect cannot take down a long-lived interpreter or its run loop.

This means .send() does not re-raise your action’s exception:

class Logic(MachineLogic):
    def risky_action(self, interpreter, context, event, action_def):
        raise ValueError("Something went wrong!")

interp.send("GO")
print(interp.current_state_ids)   # the transition completed anyway
print(interp.status)              # "running"

Because the exception never reaches your caller, the way to react to a failure is to catch it inside the action and record it on context, then branch on that with a guard:

class Logic(MachineLogic):
    def save_to_database(self, interpreter, context, event, action_def):
        try:
            # Simulate database save
            data = context.get("formData", {})
            if not data:
                raise ValueError("No form data to save")
            # ... perform save ...
            context["saveStatus"] = "success"
            print("Data saved successfully")
        except Exception as e:
            context["saveStatus"] = "error"
            context["lastError"] = str(e)
            print(f"Save failed: {e}")

class Logic(MachineLogic):
    def has_error(self, context, event):
        return context.get("saveStatus") == "error"

"saving": {
    "on": {"DONE": [
        {"target": "failed", "guard": "hasError"},
        {"target": "saved"},
    ]}
}

Observing contained failures

Because errors are contained, an uncaught exception is invisible to your machine — it is logged, but the flow carries on as if the action succeeded. When you need to know, register a plugin:

from xstate_statemachine import PluginBase

class ActionErrorReporter(PluginBase):
    def on_action_error(self, interpreter, action, error):
        sentry_sdk.capture_exception(error)      # or a metric, or a DLQ

interp.use(ActionErrorReporter())

on_action_error fires on both engines, for user actions and built-in action creators alike.

Note: Always record failures on context as well when the machine needs to react to them — a plugin observes, but it cannot change the flow.

Invoked services behave differently: their failures are routed back into the machine as onError, which is the idiomatic way to model expected errors. See Services & Invoke.

💥 When an Action Raises

The behavior above is one of three policies, controlled by the machine-config key actionErrorPolicy:

Value Behavior
"continue" (default) The error is contained: it is logged, the transition still completes, and the machine keeps running. Emits a one-shot DeprecationWarning — the default flips to "rollback" in 1.0.
"rollback" The transition’s configuration and context changes are rolled back; the machine stays in its pre-transition state.
"fail" Same rollback, plus the interpreter stops: status becomes "stopped", the configuration is cleared (a stopped machine has no active leaf — before 0.8.1 it kept reporting the source leaf under status="error", #145), children and timers are torn down, and the TransitionFailedError is retained on interp.error (__cause__ is the action’s exception). The sync send() caller and the async wait=True receipt both receive it. A parent that invoked this machine sees the failure on its onError. 
{
  "id": "m",
  "actionErrorPolicy": "rollback",
  "initial": "a",
  "states": { "a": {} }
}

Guards have an analogous guardErrorPolicy ("false"/"true"/"raise", see Guards) and unhandled events have their own onUnhandled ("ignore"/"defer"/"error", see JSON Configuration) — these are siblings of actionErrorPolicy, not alternatives to it.

Whichever policy is set, interpreter.last_transition_ok reports whether the most recent transition’s actions all ran to completion, and the on_transition_failed(interpreter, transition, failed_actions) plugin hook fires with the list of (action_def, exception) pairs that failed. See Plugins.

What rollback does — and does not — undo

rollback is a configuration + context transaction. It is not an effect transaction:

Effect of an earlier action in the same list After rollback
Context mutation ✅ restored
State configuration (entered / exited states, their after timers and invokes) ✅ restored, timers re-armed
Actor created by spawn_* ✅ stopped and unregistered
Event queued by the raise built-in ✅ withdrawn (0.8.1) — it was queued for the machine itself and not yet processed
sendTo / send_to to another actor ❌ delivered — the event has already left this machine
Anything your own code did (HTTP call, database write, log line) ❌ happened

A rolled-back machine can therefore leave a remote side effect behind: if a failing entry action runs after a sendTo that told a risk actor “order is live”, this machine returns to idle while the risk actor believes an order exists. Two design rules keep that from mattering:

  1. Put the outward-facing action last in its list — or on the entry of the state the transition commits to — so it only runs once everything that could fail has succeeded.
  2. Treat any cross-actor message as at-least-once and make the receiver idempotent, exactly as you would with a network.

Under "continue" the on_transition_failed hook fires once per action slot that failed — a transition whose own action and the target’s entry action both raise reports two calls. Under "rollback" / "fail" the first failure aborts the transition, so there is exactly one.

Cost of arming rollback

rollback and fail checkpoint the context (a deepcopy) before a transition that can run actions. Since 0.8.1 the checkpoint is skipped when no action can run — the transition has no actions, no exited state has exit, and no entered subtree has entry/exit — so an idle machine on rollback runs at ≈ 0.98× of the default. For action-bearing transitions the cost scales with the size of your context; see Production Characteristics. Keep large, immutable reference data out of context (pass it via input or close over it in your logic) if this matters to you.

Best Practices

Keep Actions Simple

Each action should do one thing. If an action is getting complex, split it into multiple smaller actions:

# Instead of one monolithic action:
# def processOrder(self, interpreter, context, event, action_def):
#     validate + calculate + save + notify + log

# Split into focused actions:
"actions": ["validateOrder", "calculateTotal", "saveOrder", "notifyUser", "logOrder"]

Use Context for Data Flow

Pass data between actions through context, not through side channels:

class Logic(MachineLogic):
    def validate_form(self, interpreter, context, event, action_def):
        # Store validation result in context
        context["validationResult"] = {
            "isValid": True,
            "errors": []
        }

    def submit_form(self, interpreter, context, event, action_def):
        # Read from context — don't recompute
        if context["validationResult"]["isValid"]:
            print("Submitting valid form...")

Log Important Actions

Use the interpreter ID for traceable logs in multi-machine systems:

import logging

class Logic(MachineLogic):
    def process_payment(self, interpreter, context, event, action_def):
        logging.info(
            "[%s] Processing payment of $%.2f",
            interpreter.id,
            event.payload.get("amount", 0)
        )

Complete Example: Form Machine with Full Action Lifecycle

A comprehensive form machine demonstrating entry, exit, and transition actions working together:

Note: As of 0.8.0, a MachineLogic subclass method with ambiguous arity — e.g. a 2-arg method that could be a guard or a 2-arg service — is still registered by arity, but now emits a UserWarning recommending explicit @action/@guard/@service decoration. The example below decorates hasRequiredFields with @guard for exactly this reason.

from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic, guard

config = {
    "id": "contactForm",
    "initial": "editing",
    "context": {
        "formData": {"name": "", "email": "", "message": ""},
        "errors": [],
        "draft": None,
        "submissionId": None,
        "isLoading": False
    },
    "states": {
        "editing": {
            "entry": ["loadDraft", "clearErrors"],
            "exit": "saveDraft",
            "on": {
                "UPDATE_FIELD": {"actions": "updateField"},
                "SUBMIT": [
                    {"target": "validating", "guard": "hasRequiredFields"},
                    {"target": "editing", "actions": "showFieldErrors"}
                ]
            }
        },
        "validating": {
            "entry": "validateAll",
            "on": {
                "VALIDATION_PASS": {"target": "submitting"},
                "VALIDATION_FAIL": {"target": "editing", "actions": "setErrors"}
            }
        },
        "submitting": {
            "entry": "showSpinner",
            "exit": "hideSpinner",
            "on": {
                "SUCCESS": {
                    "target": "success",
                    "actions": ["storeSubmissionId", "clearDraft"]
                },
                "FAILURE": {
                    "target": "editing",
                    "actions": "setErrors"
                }
            }
        },
        "success": {
            "entry": "showConfirmation",
            "type": "final"
        }
    }
}

class ContactFormLogic(MachineLogic):
    # ---- Entry Actions ----
    def load_draft(self, interpreter, context, event, action_def):
        if context["draft"]:
            context["formData"] = dict(context["draft"])
            print("Draft restored from auto-save.")
        else:
            print("Starting with empty form.")

    def show_spinner(self, interpreter, context, event, action_def):
        context["isLoading"] = True
        print("Loading...")

    def show_confirmation(self, interpreter, context, event, action_def):
        sid = context["submissionId"]
        print(f"Thank you! Your submission ID is: {sid}")

    def validate_all(self, interpreter, context, event, action_def):
        errors = []
        fd = context["formData"]
        if not fd.get("name"):
            errors.append("Name is required")
        if "@" not in fd.get("email", ""):
            errors.append("Valid email is required")
        if not fd.get("message"):
            errors.append("Message is required")

        if errors:
            context["errors"] = errors
            interpreter.send("VALIDATION_FAIL")
        else:
            interpreter.send("VALIDATION_PASS")

    # ---- Exit Actions ----
    def save_draft(self, interpreter, context, event, action_def):
        context["draft"] = dict(context["formData"])
        print("Draft auto-saved.")

    def hide_spinner(self, interpreter, context, event, action_def):
        context["isLoading"] = False

    # ---- Transition Actions ----
    def update_field(self, interpreter, context, event, action_def):
        field = event.payload.get("field")
        value = event.payload.get("value", "")
        if field and field in context["formData"]:
            context["formData"][field] = value

    def clear_errors(self, interpreter, context, event, action_def):
        context["errors"] = []

    def set_errors(self, interpreter, context, event, action_def):
        if event.payload.get("errors"):
            context["errors"] = event.payload["errors"]

    def show_field_errors(self, interpreter, context, event, action_def):
        context["errors"] = ["Please fill in all required fields"]
        print(f"Errors: {context['errors']}")

    def store_submission_id(self, interpreter, context, event, action_def):
        context["submissionId"] = event.payload.get("id", "UNKNOWN")

    def clear_draft(self, interpreter, context, event, action_def):
        context["draft"] = None

    # ---- Guards ----
    @guard
    def has_required_fields(self, context, event):
        fd = context["formData"]
        return bool(fd.get("name") and fd.get("email") and fd.get("message"))

# Run the form machine
machine = create_machine(config, logic=ContactFormLogic())
interp = SyncInterpreter(machine).start()
# Output: Starting with empty form.

# Fill in the form
interp.send("UPDATE_FIELD", field="name", value="Alice")
interp.send("UPDATE_FIELD", field="email", value="alice@example.com")
interp.send("UPDATE_FIELD", field="message", value="Hello!")

# Submit the form
interp.send("SUBMIT")
# Output: Draft auto-saved.  (exit action on editing)
# The machine transitions: editing → validating → submitting

# Simulate server response
interp.send("SUCCESS", id="FORM-12345")
# Output: Thank you! Your submission ID is: FORM-12345

print(interp.context["submissionId"])  # FORM-12345
interp.stop()

Auto-Discovery with LogicLoader

The LogicLoader is a singleton that automatically discovers actions, guards, and services from Python modules or class instances. It maps snake_case Python function names to camelCase JSON names.

Module-Based Discovery

# my_logic.py — separate file with all your logic
from xstate_statemachine import MachineLogic

def validate_input(interpreter, context, event, action_def):
    """Auto-maps to 'validateInput' in JSON config."""
    data = event.data if hasattr(event, 'data') else {}
    context["is_valid"] = bool(data.get("name"))

def is_valid(context, event):
    """Auto-maps to 'isValid' guard in JSON config."""
    return context.get("is_valid", False)

def fetch_data(interpreter, context, event):
    """Auto-maps to 'fetchData' service in JSON config."""
    return {"items": [1, 2, 3]}

# main.py — use the module for auto-discovery
import my_logic
from xstate_statemachine import create_machine, SyncInterpreter

config = {
    "id": "autoDiscover",
    "initial": "input",
    "context": {"is_valid": False},
    "states": {
        "input": {
            "on": {
                "SUBMIT": {
                    "target": "validating",
                    "actions": "validateInput"
                }
            }
        },
        "validating": {
            "always": [
                {"target": "fetching", "guard": "isValid"},
                {"target": "input"}
            ]
        },
        "fetching": {
            "invoke": {
                "src": "fetchData",
                "onDone": "done"
            }
        },
        "done": {"type": "final"}
    }
}

# Pass the module — LogicLoader discovers all matching functions
machine = create_machine(config, logic_modules=[my_logic])
interp = SyncInterpreter(machine).start()

interp.send("SUBMIT", name="Alice")
print(interp.active_state_ids)
# {'autoDiscover.done'}

interp.stop()

Class-Based Discovery (logic_providers)

from xstate_statemachine import create_machine, SyncInterpreter

class OrderLogic:
    """Class with methods that auto-map to JSON names."""

    def calculate_total(self, interpreter, context, event, action_def):
        """Maps to 'calculateTotal' in JSON."""
        items = context.get("items", [])
        context["total"] = sum(item["price"] for item in items)

    def has_items(self, context, event):
        """Maps to 'hasItems' guard in JSON."""
        return len(context.get("items", [])) > 0

config = {
    "id": "order",
    "initial": "cart",
    "context": {"items": [{"name": "Widget", "price": 9.99}], "total": 0},
    "states": {
        "cart": {
            "on": {
                "CHECKOUT": {
                    "target": "checkout",
                    "guard": "hasItems",
                    "actions": "calculateTotal"
                }
            }
        },
        "checkout": {"type": "final"}
    }
}

# Pass instances — LogicLoader discovers methods by snake_case matching
machine = create_machine(config, logic_providers=[OrderLogic()])
interp = SyncInterpreter(machine).start()

interp.send("CHECKOUT")
print(interp.context["total"])
# 9.99

interp.stop()

Global Registration

For large applications, register modules globally so all machines can discover them:

from xstate_statemachine import LogicLoader

import my_actions
import my_guards
import my_services

# Register once at startup
loader = LogicLoader.get_instance()
loader.register_logic_module(my_actions)
loader.register_logic_module(my_guards)
loader.register_logic_module(my_services)

# All subsequent create_machine() calls can find these functions
# without passing logic_modules every time

Naming Convention: Write Python in PEP 8 snake_case; keep the JSON in XState’s camelCase. Matching is case- and separator-insensitive on both sides — every entry point (MachineLogic dicts, subclass methods, logic_modules, logic_providers, the Pythonic decorators) resolves the same way:

JSON name Python implementation
validateInput def validate_input(...)
logHTTPStatus def log_http_status(...) — acronyms are fine
fetchUserV2 def fetch_user_v2(...) — digits are fine
fetch-data, inline:machine.state#entry[0] def fetch_data(...), def inline_machine_state_entry_0(...) — Stately’s non-identifier names need no decorator

An exact-name entry always wins over an alias, and registering two different callables that differ only by case/separators (fetch_data and fetchData) is rejected at create_machine() as ambiguous. This is exactly what xsm gt generates, so hand-written and generated logic look the same.

See Also

  • Guards — conditional transitions that decide whether actions fire
  • Context — the data that actions typically mutate
  • Services & Invoke — actions triggered by onDone / onError service callbacks
  • Pythonic API — @action decorator and State.enter()/State.exit() decorators
  • Core Concepts — action execution order (entry → transition → exit)
  • Interpreters — how actions interact with sync vs async interpreters
Edit this page on GitHub

XState-StateMachine for Python — MIT License

3,183 tests across Python 3.9–3.14 on Linux, macOS and Windows.

GitHub PyPI Changelog