Actions

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

45 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")

Nested and parallel states

When a transition leaves or enters several states at once, the engine follows XState v5 (and SCXML) exactly:

Exit actions run in reverse document order: innermost states first, ancestors after their descendants, and in a parallel state one whole region at a time, the region declared last first.
Transition actions run next, in the order they are listed.
Entry actions run in document order: ancestors before their descendants, regions one at a time in the order they are declared — including the region that holds the target.
With "reenter": true the source state itself exits and re-enters, even when the target is one of its own children.
When several regions of a parallel state take a transition for the same event, all of them form one step: every exit first, then each transition’s actions (regions in document order), then every entry. Two transitions whose exits overlap conflict, and only one of them runs (Parallel States → When Several Regions Handle the Same Event).
When the machine reaches a top-level final state, the exit actions of every state still active — the final state and the machine root included — run after the final state’s entry actions, innermost first, and then status becomes "done".

Parallel States → Entry and Exit Order has a worked example.

📄 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)

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)

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 — with `delay=` it is a timer like `after`, never counted by `maxIterations`; without, a zero-delay self-cycle is cut at the budget
`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

`emit` and `interpreter.on()` — notify external listeners

emit publishes an event to code outside the machine. In JSON it is {"type": "emit", "params": {"event": {...}}}; the event is not sent to the machine itself. Subscribe with interpreter.on(event_type, listener) — use "*" to receive every emitted event. The listener gets an Event (.type, .payload), and on() returns a callable that unsubscribes:

from xstate_statemachine import SyncInterpreter, create_machine

config = {
    "id": "m",
    "initial": "idle",
    "states": {
        "idle": {
            "on": {
                "GO": {
                    "target": "done",
                    "actions": [
                        {
                            "type": "emit",
                            "params": {
                                "event": {
                                    "type": "TOAST",
                                    "text": "hi",
                                },
                            },
                        }
                    ],
                },
            },
        },
        "done": {},
    },
}

interp = SyncInterpreter(create_machine(config)).start()
toasts = []
unsubscribe = interp.on("TOAST", toasts.append)
interp.send("GO")
print(toasts)
# [Event(type='TOAST', payload={'text': 'hi'})]
unsubscribe()
interp.stop()

The log built-in writes params.expr (a string, or a callable of {context, event}) to the library logger at INFO, prefixed with [label] when params.label is given. cancel ({"type": "cancel", "params": {"sendId": "<id>"}}) disarms a delayed raise / sendTo that was given that id, before it fires.

`assign` — update context

assign({...}) maps context keys to new values; a value may be a callable of {context, event}. As in XState v5, every property of one assign is computed against the context as it was before that action, and the results are merged afterwards — a property never sees what another property of the same assign just wrote, so two keys can swap without a temporary:

from xstate_statemachine import SyncInterpreter, assign, create_machine

config = {
    "id": "m",
    "initial": "a",
    "context": {
        "left": "x",
        "right": "y",
    },
    "states": {
        "a": {
            "on": {
                "SWAP": {
                    "actions": assign(
                        {
                            "left": lambda args: args["context"]["right"],
                            "right": lambda args: args["context"]["left"],
                        }
                    ),
                },
            },
        },
    },
}

interp = SyncInterpreter(create_machine(config)).start()
interp.send("SWAP")
assert interp.context == {
    "left": "y",
    "right": "x",
}

Separate assign actions run in order, each seeing what the previous ones wrote (exit actions, then the transition’s actions, then entry actions). To derive one key from another key’s new value, use two assigns or a single callable assign(lambda args: {...}) that returns the whole update.

In JSON, assign is {"type": "xstate.assign", "params": {"assignment": {...}}} (what assign() builds) or, equivalently, the property mapping directly as params: {"type": "xstate.assign", "params": {"count": 1}}. "assign" works as the type too. When params has an assignment key, that key is the mapping.

Changed in 0.12.0: the properties of one assign used to be written one at a time, so a later property’s callable saw an earlier property’s new value; and the JSON form without the assignment wrapper was silently ignored.

`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

The same action in plain JSON (the form a Stately export or a hand-written config uses) is xstate.choose with params.branches. The branches are tried in order and the first whose guard passes runs; a branch without a guard always passes. Branch guards may be composite (and / or / not) or the built-in stateIn:

{
  "type": "xstate.choose",
  "params": {
    "branches": [
      {
        "guard": {
          "type": "stateIn",
          "params": {
            "stateValue": "#m.p.mode.night"
          }
        },
        "actions": "atNight"
      },
      {
        "guard": {
          "type": "and",
          "params": {
            "guards": [
              "isAdmin",
              {
                "type": "not",
                "params": {
                  "guard": "isBanned"
                }
              }
            ]
          }
        },
        "actions": "adminByDay"
      },
      {
        "actions": "otherwise"
      }
    ]
  }
}

params.conditions (what the choose() helper builds) is accepted as a synonym of params.branches.

`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.

Do not await your own receipt inside an action. await interp.send(..., wait=True) issued on the action’s own task raises ReentrantWaitError (0.9.0, #219) — the run loop cannot advance until the action returns, so the receipt could never resolve. Send without wait, or hand the receipt to another task (asyncio.ensure_future(...)); a helper the action spawns may await freely (#225). A plain def action that drops a wait=True result gets a RuntimeWarning (#232). Details in Interpreters.

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.9.0 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 `invoke`d 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 `invoke`s)	✅ restored, timers re-armed
Actor created by `spawn_*`	✅ stopped and unregistered
Event queued by the `raise` built-in	✅ withdrawn (0.9.0) — 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:

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.
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.9.0 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()

What a provider contributes (0.12.0). Discovery scans the instance’s public bound methods and @staticmethods (a pure guard needs no self). A method named like a built-in action – log, cancel, stop, send_to, … – is bound and runs instead of the built-in, exactly as a same-named entry in an explicit MachineLogic does; for a logic_modules module the same holds for a function defined in that module, not for one merely imported into it (from logging import log cannot hijack log). Passing logic= together with logic_modules / logic_providers emits a UserWarning: the explicit logic is used and the other sources are not scanned.

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.

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