Pydantic integration

Typed, validated context and events for statecharts with Pydantic v2 — plus static config validation and JSON Schema for your frontend.

Modern Python expects typed, validated data at every boundary. python-statemachine and transitions are stringly typed; XState v5’s setup() gives TypeScript users typed context and events. The Python equivalent is Pydantic v2 — and it is the validation layer the FastAPI, DRF and agents integrations reuse. Silent acceptance is a bug: a typo’d context key or a malformed payload should fail at the boundary, not deep inside an action. This extra makes that the default, through seams core already has — core never imports pydantic.

Install

pip install "xstate-statemachine[pydantic]"

Requires pydantic>=2.5. Tested versions are in the compatibility table.

Runnable example. examples/integrations/fastapi_orders uses this extra: its models.py imports EventModel and events_union from xstate_statemachine.contrib.pydantic, and its app.py imports context_model. The agents_support_bot example does not import the extra itself; its lookup_order tool calls the fastapi_orders app in-process, so it exercises these validated events through that app’s HTTP surface.

Quick start

from decimal import Decimal
from typing import Literal
from pydantic import BaseModel
from xstate_statemachine import MachineLogic, SyncInterpreter, create_machine
from xstate_statemachine.contrib.pydantic import (
    ContextValidationError,
    EventModel,
    TypedContextPlugin,
    context_model,
    context_of,
    events_union,
    typed_context,
)


class OrderContext(BaseModel):
    total: Decimal = Decimal(0)
    currency: Literal["USD", "EUR"] = "USD"


class Pay(EventModel):  # an instance IS an event
    type: Literal["PAY"] = "PAY"
    amount: Decimal


cfg = {
    "id": "order",
    "initial": "open",
    "actionErrorPolicy": "rollback",
    "states": {
        "open": {
            "on": {
                "PAY": {
                    "actions": "add",
                },
                "BREAK": {
                    "actions": "bad",
                },
            },
        },
    },
}
logic = MachineLogic(
    actions={
        "add": lambda i, ctx, e, a: ctx.__setitem__(
            "total", ctx["total"] + e.payload["amount"]
        ),
        "bad": lambda i, ctx, e, a: ctx.__setitem__(
            "currency", "GBP"
        ),  # not in the Literal
    }
)
machine = create_machine(
    typed_context(OrderContext, cfg),  # initial context validated + completed
    logic=logic,
    context_validator=context_model(
        OrderContext
    ),  # re-validated after every mutating action
    event_schemas=events_union(Pay),  # payloads validated at send()
)
order = SyncInterpreter(machine).use(TypedContextPlugin(OrderContext)).start()

order.send(Pay(amount=Decimal("9.50")))  # typed event, no dict
assert context_of(order, OrderContext) == OrderContext(total=Decimal("9.50"))

receipt = order.send("BREAK", wait=True)  # the action breaks the model...
assert isinstance(
    receipt.error, ContextValidationError
)  # ...reported as an action error...
assert (
    order.context["currency"] == "USD"
)  # ...and rolled back: never an invalid context

Reference

Typed context

Typed events

Config validation

JSON Schema

Guarantees

What this does: the machine never holds a context that violates the model — an action that breaks it is an action error and rollback restores the previous context; a payload that does not fit its EventModel never enters the machine; a chart that fails validate_machine_json never reaches create_machine.

What this does not do: validate context on every read (only after a mutating action, and on start with TypedContextPlugin); type-check guards or services; replace create_machine’s own tree validation (unknown targets, dead always loops). With actionErrorPolicy: "continue" (the pre-1.0 default) an invalid context IS kept and only reported — set "rollback".

See the programme-wide Guarantees and Security pages (#303).

Threat model

Who can call this: whoever can send() — the same surface as any event. Validation runs in the sender’s call, before queueing, so a malformed payload costs the sender, not the machine.

What it exposes: ContextValidationError / InvalidEventPayloadError messages include pydantic’s field paths and messages (not values by default — pydantic omits input from str()); machine_json_schema publishes your event and context shapes — review it before serving it publicly.

You must configure: actionErrorPolicy: "rollback" for the never-invalid guarantee; extra="forbid" on your context model if unknown keys should be refused; the store codec= if Decimal / datetime precision at rest matters.

Compatibility

Troubleshooting