flying-state-machines 0.3.1

The only Finite State Machine library with a Flying Spaghetti Monster serialization format

Flying State Machines

Ever want to use a finite state machine (FSM) but didn't think it was okay to not also pay homage to the great Flying Spaghetti Monster in the sky whose Noodly abbreviation we use? This is the library for you, fellow Pastafarian, in your pursuits to use deterministic and probabilistic FSMs (aka Markov chains) in a manner befitting an emissary of his Holy Noodlage.

Installation

pip install flying-state-machines

Status

Issues can be tracked here. Changelog can be found here.

Code Structure

The code is organized into four classes: FSM, Transition, AsyncFSM, and AsyncTransition.

Method signatures

• FSM
  • add_event_hook(self, event: Enum|str, hook: Callable[[Enum|str, FSM], bool]) -> None:
  • remove_event_hook(self, event: Enum|str, hook: Callable[[Enum|str, FSM], bool]) -> None:
  • add_transition_hook(self, transition: Transition, hook: Callable[[Transition, dict, Any]]) -> None:
  • remove_transition_hook(self, transition: Transition, hook: Callable[[Transition, dict, Any]]) -> None:
  • would(self, event: Enum|str) -> tuple[Transition]:
  • can(self, event: Enum|str) -> bool:
  • input(self, event: Enum|str, data: Any = None) -> Enum|str:
  • pack(self) -> bytes:
  • @classmethod unpack(data: bytes, inject: dict = {}, event_hooks: dict = {}, transition_hooks: dict = {}, random: Callable[[], float] = random.random) -> FSM:
  • touched(self) -> str:
• Transition
  • add_hook(self, hook: Callable[[Transition, dict, Any]]) -> None:
  • remove_hook(self, hook: Callable[[Transition, dict, Any]]) -> None:
  • trigger(self, context: dict = None, data: Any = None) -> None:
  • pack(self) -> bytes:
  • @classmethod unpack(cls, data: bytes, hooks: list[Callable[[Transition, dict, Any]]] = [], inject: dict = {}) -> Transition:
  • @classmethod from_any(cls, from_states: type[Enum]|list[str], event: Enum|str, to_state: Enum|str, probability = 1.0) -> list[Transition]:
  • @classmethod to_any(cls, from_state: Enum|str, event: Enum|str, to_states: type[Enum]|list[str], total_probability = 1.0) -> list[Transition]:
• AsyncFSM (mirrors FSM; input() and trigger() are async)
  • async input(self, event: Enum|str, data: Any = None) -> Enum|str:
  • (all other methods are synchronous and identical to FSM)
• AsyncTransition (mirrors Transition; trigger() is async)
  • async trigger(self, context: dict = None, data: Any = None) -> None:
  • (all other methods are synchronous and identical to Transition)

To see the full documentation, read the dox.md generated by autodox.

Usage

To use this library to make a Flying State Machine™, import and extend as shown below:

from enum import Enum, auto
from flying_state_machines import Transition, FSM

class State(Enum):
    NORMAL_CLOTHES = auto()
    PIRATE_CLOTHES = auto()

class Event(Enum):
    IS_FRIDAY = auto()
    IS_NOT_FRIDAY = auto()


class Pastafarian(FSM):
    initial_state = State.NORMAL_CLOTHES
    rules = set([
        Transition(State.NORMAL_CLOTHES, Event.IS_FRIDAY, State.PIRATE_CLOTHES),
        Transition(State.PIRATE_CLOTHES, Event.IS_NOT_FRIDAY, State.NORMAL_CLOTHES),
    ])

This will represent the state of a Pastafarian. Events can be passed to the FSM, either to cause a state transition or to see what state transitions are possible.

me = Pastafarian()
would = me.would(Event.IS_FRIDAY) # tuple with the Transition of putting on pirate regalia
print(would)
print('ok' if me.can(Event.IS_FRIDAY) else 'nope') # boolean check if event can be processed
state = me.input(Event.IS_FRIDAY) # state is State.PIRATE_CLOTHES
print(state)

state = me.input('ate a hotdog') # state is still State.PIRATE_CLOTHES
print(state) # State.PIRATE_CLOTHES
print(me.current) # State.PIRATE_CLOTHES
print(me.previous) # State.NORMAL_CLOTHES

It is also possible to use str and list[str] instead of Enums for states and events. As of v0.3.0, subclasses of FSM have per-instance context dicts for storing arbitrary state data in addition to the finite states.

Probabilistic transitions

It is possible to encode probabilistic transitions by supplying multiple Transitions with identical from_state and on_event. The cumulative probability of all such Transitions should be <= 1.0, but the result of a probabilistic transition choice will be normalized to the cumulative probability of all valid Transitions for the event.

from flying_state_machines import FSM, Transition

class RussianRoulette(FSM):
    initial_state = 'safe'
    rules = set([
        Transition('safe', 'spin', 'safe', 5.0/6.0),
        Transition('safe', 'spin', 'dead', 1.0/6.0),
    ])

gun = RussianRoulette()
state = gun.input('spin') # 1/6 chance of getting shot
print(state)

Support for context-aware dynamic probabilities was added in v0.3.0. This should be useful for dynamic simulations, video game NPCs, etc.

Dynamic probabilities example

from flying_state_machines import FSM, Transition

def p_dead(context: dict) -> float:
    return context.get('loaded_chambers', 0.0) / context.get('capacity', 6.0)

def p_safe(context: dict) -> float:
    return 1.0 - p_dead(context)

class RussianRoulette(FSM):
    initial_state = 'safe'
    rules = set([
        Transition('safe', 'spin', 'safe', p_safe),
        Transition('safe', 'spin', 'dead', p_dead),
    ])

gun = RussianRoulette()
state = gun.input('spin') # 0/6 chance of getting shot with an empty gun
print(state)
# load the gun
gun.context['loaded_chambers'] = 6.0
state = gun.input('spin') # guaranteed blam
print(state)

Custom Random Functions

As of v0.3.0, you can specify a custom random function for probabilistic transitions. This is useful for deterministic testing or implementing custom randomization strategies.

Custom randomizer example

from flying_state_machines import FSM, Transition
from hashlib import sha256
import struct

class Randomizer:
    def __init__(self, seed: bytes = b'test'):
        self.seed = seed
        self.nonce = 0
    def next_float(self) -> float:
        self.nonce += 1
        return struct.unpack('!d', sha256(
            self.seed + self.nonce.to_bytes(4, 'big')
        ).digest()[:8])[0]
    def reset(self):
        self.nonce = 0

class Machine(FSM):
    initial_state = 'safe'
    rules = set([
        Transition('safe', 'spin', 'safe', 5/6),
        Transition('safe', 'spin', 'dead', 1/6),
    ])

# Deterministic testing with custom randomizer
randomizer = Randomizer(b'deterministic-seed-69420')
machine = Machine(random=randomizer.next_float)
result = machine.input('spin')
print(result)  # Always 'safe' with this seed

Custom random functions are not serialized in pack() data and must be re-supplied when calling unpack():

randomizer = Randomizer(b'my-seed')
machine = Machine(random=randomizer.next_float)
packed = machine.pack()
unpacked = Machine.unpack(
    packed,
    inject={'State': State, 'Event': Event},
    random=randomizer.next_float
)

Transition.to_any and Transition.from_any

There are helper class methods available for generating lists of Transitions. The .to_any method will return a list of Transitions that represents a probabilistic transition from a specific state to a valid state on the given event, which will be useful in creating Markov chains. The .from_any method will return a list of Transitions that represents a probabilistic transition from a valid state to a specific state on a given event, which is useful for example in aborting to an error state. They can be used as follows:

Code example of `Transition.to_any` and `Transition.from_any`

from enum import Enum, auto
from flying_state_machines import FSM, Transition


class State(Enum):
    WAITING = auto()
    GOING = auto()
    NEITHER = auto()
    SUPERPOSITION = auto()

class Event(Enum):
    START = auto()
    STOP = auto()
    CONTINUE = auto()
    QUANTUM_FOAM = auto()
    NORMALIZE = auto()

class Machine(FSM):
    initial_state = State.WAITING
    rules = set([
        Transition(State.WAITING, Event.CONTINUE, State.WAITING),
        Transition(State.WAITING, Event.START, State.GOING),
        Transition(State.GOING, Event.CONTINUE, State.GOING),
        Transition(State.GOING, Event.STOP, State.WAITING),
        *Transition.from_any(
            State, Event.QUANTUM_FOAM, State.SUPERPOSITION, 0.5
        ),
        *Transition.from_any(
            State, Event.QUANTUM_FOAM, State.NEITHER, 0.5
        ),
        *Transition.to_any(
            State.NEITHER, Event.NORMALIZE, [State.WAITING, State.GOING]
        ),
        *Transition.to_any(
            State.SUPERPOSITION, Event.NORMALIZE, [State.WAITING, State.GOING]
        ),
    ])

The above will create a FSM that will transition to either SUPERPOSITION or NEITHER probabilistically upon the QUANTUM_FOAM event, and it will transition to either WAITING or GOING probabilistically upon the NORMALIZE event.

Note that the first argument for Transition.from_any can be a list of specific states rather than a state enum.

Async Usage

AsyncFSM and AsyncTransition provide async versions of the sync classes. Only input() and trigger() are async def and must be awaited; all other methods (would, can, hooks, pack/unpack, touched) remain synchronous and work identically.

Hooks on async classes accept both sync and async callables — coroutines are detected and awaited automatically.

import asyncio
from flying_state_machines import AsyncFSM, AsyncTransition

class AsyncMachine(AsyncFSM):
    initial_state = 'hungry'
    rules = set([
        AsyncTransition('hungry', 'get food', 'eating'),
        AsyncTransition('eating', 'food gone', 'sad'),
    ])

async def main():
    machine = AsyncMachine()
    state = await machine.input('get food')
    print(state)  # 'eating'

asyncio.run(main())

Hooks

What good is a pirate without a hook? Hooks can be specified for events and for transitions. The hooks for an event get called when the event is being processed and before any transition occurs, and if an event hook returns False, the state transition will be cancelled. For example:

Event Hooks Example

from flying_state_machines import Transition, FSM


class PastaMachine(FSM):
    rules = set([
        Transition('in a box', 'pour into pot', 'is cooking'),
        Transition('is cooking', '7 minutes pass', 'al dente'),
        Transition('is cooking', '10 minutes pass', 'done'),
        Transition('is cooking', '15 minutes pass', 'mush'),
    ])
    initial_state = 'in a box'

def status_hook(event, fsm, data):
    print([event, fsm.current, fsm.next, data])

machine = PastaMachine()
machine.add_event_hook('pour into pot', status_hook)
state = machine.input('pour into pot', 'optional event data of Any type goes here')
# console will show 'pour into pot', 'in a box', and 'is cooking'
print(state, '==', machine.current)
# state and machine.current will be 'is cooking'
print(machine.next)
# machine.next will be None


def the_box_was_not_open(event, fsm):
    print('you forgot to open the box')
    return False

machine = PastaMachine()
machine.add_event_hook('pour into pot', the_box_was_not_open)
state = machine.input('pour into pot')
# console will show 'you forgot to open the box'
print(state, '==', machine.current)
# state and machine.current will be 'in a box'
print(machine.next)
# machine.next will be 'is cooking', indicating an aborted state transition

Transition hooks are set on the individual Transitions and are called whenever the Transition is triggered (i.e. after the state has changed). FSM has an add_transition_hook method for convenience; it is semantically identical to calling the add_hook method on the Transition. Since the Transition has already occurred by the time the hooks are called, they do not have any chance to interact with the process.

Transition Hooks Example

machine = PastaMachine()
transition = machine.would('pour into pot')[0]

def transition_hook(transition, context, data):
    print(
        f'{transition.from_state} => {transition.to_state} '
        f'with {context=} and {data=}'
    )

machine.add_transition_hook(transition, transition_hook)
# semantically identical to transition.add_hook(transition_hook)

One thing to note is that FSM.add_transition_hook will perform an additional check to ensure that the Transition supplied is within the FSM rules. Also note that transition hooks will be called with the Transition, the FSM context and the same event data as the event hooks, the latter of which is passed in as the optional second argument for FSM.input.

Serialization

As of v0.2.0, Transition can be serialized and deserialized to and from bytes. This uses the packify library, so deserialization when using enums can be accomplished by passing in an inject dict mapping enum class names to their classes. Hooks cannot be serialized and deserialized, so they also must be supplied as an argument to Transition.unpack. E.g.

Transition Serialization Example

class State(Enum):
    SAFE = 'safe'
    DEAD = 'RIP'

class Event(Enum):
    SPIN = 'spin'

def p_dead(context: dict) -> float:
    return context.get('loaded_chambers', 1.0) / context.get('capacity', 6.0)

# dynamic, context-based probabilities added in v0.3.0
transition = Transition(State.SAFE, Event.SPIN, State.DEAD, p_dead)
hook = lambda *_: print('BANG')
transition.add_hook(hook)
packed = transition.pack()
unpacked = Transition.unpack(
    packed, hooks=[hook], inject={
        'State': State, 'Event': Event,
        'p_dead': p_dead,
    }
)
unpacked.trigger() # prints 'BANG'

Also as of v0.2.0, FSM now can be serialized and deserialized to and from bytes. This also uses packify, and it has similar syntax to above.

FSM Serialization Example

# continuing with the Pastafarian example
hook = lambda event, *args: print(f'celebrate {event.name}') or True
me.add_event_hook(Event.IS_FRIDAY, hook)
packed = me.pack()
assert type(packed) is bytes
unpacked = Pastafarian.unpack(
    packed, event_hooks={Event.IS_FRIDAY: [hook]},
    inject={'State':State, 'Event': Event}
)
assert unpacked.current == me.current
unpacked.input(Event.IS_FRIDAY) # prints 'celebrate IS_FRIDAY'

FSMs have a unique serialization format that can be accessed by using the touched method. print(machine.touched()) will result in something like the following:

    [State.WAITING]        [None]
           \                /
          (((State.GOING)))
{<State.GOING: 2>: {<Event.QUANTUM_FOAM: 4>: [(<State.GOING: 2>, <Event.QUANTUM_FOAM: 4>, <State.SUPERPOSITION: 4>, 0.5), (<State.GOING: 2>, <Event.QUANTUM_FOAM: 4>, <State.NEITHER: 3>, 0.5)], <Event.STOP: 2>: [(<State.GOING: 2>, <Event.STOP: 2>, <State.WAITING: 1>, 1.0)], <Event.CONTINUE: 3>: [(<State.GOING: 2>, <Event.CONTINUE: 3>, <State.GOING: 2>, 1.0)]}, <State.SUPERPOSITION: 4>: {<Event.NORMALIZE: 5>: [(<State.SUPERPOSITION: 4>, <Event.NORMALIZE: 5>, <State.WAITING: 1>, 0.5), (<State.SUPERPOSITION: 4>, <Event.NORMALIZE: 5>, <State.GOING: 2>, 0.5)], <Event.QUANTUM_FOAM: 4>: [(<State.SUPERPOSITION: 4>, <Event.QUANTUM_FOAM: 4>, <State.SUPERPOSITION: 4>, 0.5), (<State.SUPERPOSITION: 4>, <Event.QUANTUM_FOAM: 4>, <State.NEITHER: 3>, 0.5)]}, <State.WAITING: 1>: {<Event.QUANTUM_FOAM: 4>: [(<State.WAITING: 1>, <Event.QUANTUM_FOAM: 4>, <State.SUPERPOSITION: 4>, 0.5), (<State.WAITING: 1>, <Event.QUANTUM_FOAM: 4>, <State.NEITHER: 3>, 0.5)], <Event.START: 1>: [(<State.WAITING: 1>, <Event.START: 1>, <State.GOING: 2>, 1.0)], <Event.CONTINUE: 3>: [(<State.WAITING: 1>, <Event.CONTINUE: 3>, <State.WAITING: 1>, 1.0)]}, <State.NEITHER: 3>: {<Event.NORMALIZE: 5>: [(<State.NEITHER: 3>, <Event.NORMALIZE: 5>, <State.GOING: 2>, 0.5), (<State.NEITHER: 3>, <Event.NORMALIZE: 5>, <State.WAITING: 1>, 0.5)], <Event.QUANTUM_FOAM: 4>: [(<State.NEITHER: 3>, <Event.QUANTUM_FOAM: 4>, <State.NEITHER: 3>, 0.5), (<State.NEITHER: 3>, <Event.QUANTUM_FOAM: 4>, <State.SUPERPOSITION: 4>, 0.5)]}}
        s     s        s         s
       s        s     s            s
      s        s                  s
       s                            s

~Touched by His Noodly Appendage~

To the author's knowledge, this is the only FSM library that serializes FSMs as FSMs.

AI Agent Skills

As of v0.3.0, the library includes a CLI for exporting agent skills to AI coding environments. This enables AI agents to assist with FSM implementation using the library.

# Export skill to stdout or file
fsm skill [--output OUTPUT]
# Export for specific AI environments
fsm opencode   # .opencode/skills/flying-state-machine/SKILL.md
fsm claude     # .claude/skills/flying-state-machines/SKILL.md
fsm cursor     # .cursor/skills/flying-state-machine/SKILL.md
fsm codex      # .agents/skills/flying-state-machine/SKILL.md

The agent skill provides comprehensive documentation and examples for AI-assisted FSM development.

Testing

This is a simple library with 52 tests across two test files. To run the tests, clone the repo and then run the following:

python tests/test_classes.py
python tests/test_asynclasses.py

Each file has a visual output test (touched()), which I suggest inspecting. The async variant additionally validates that async event hooks and transition hooks are handled correctly.

License

ISC License

Copyright (c) 2026 k98kurz

Permission to use, copy, modify, and/or distribute this software for any purpose with or without fee is hereby granted, provided that the above copyright notice and this permission notice appear in all copies.

THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.