Workflow State Store

Server-side workflow state storage implementations

WorkflowStateStore Protocol

The WorkflowStateStore protocol defines the interface for persisting workflow state. This allows patterns like StepFlow to work with different storage backends (in-memory, file-based, database) without being coupled to any specific implementation.

Implementations receive the sess (session) object to extract a session identifier for scoping state to individual users.

---

source

WorkflowStateStore

def WorkflowStateStore(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Protocol for workflow state storage backends.

Session ID Extraction

The state store needs a way to identify individual users/sessions. We store a small UUID in the session object itself (the only thing we store there) and use it as a key for server-side storage.

---

source

get_session_id

def get_session_id(
    sess:Any, # FastHTML session object
    key:str='_workflow_session_id', # Session key for storing the ID
)->str: # Stable session identifier

Get or create a stable session identifier.

InMemoryWorkflowStateStore

A simple in-memory implementation of WorkflowStateStore suitable for development and testing. State is lost when the server restarts.

For production use, this can be replaced with a file-based or database-backed implementation that follows the same protocol.

---

source

InMemoryWorkflowStateStore

def InMemoryWorkflowStateStore(
    
):

In-memory workflow state storage for development and testing.

Usage Example

# Create a store instance
store = InMemoryWorkflowStateStore()

# Mock session object (in real usage, this comes from FastHTML)
mock_sess = {}

# Store and retrieve state
store.set_current_step("my_workflow", mock_sess, "step_1")
print(f"Current step: {store.get_current_step('my_workflow', mock_sess)}")

store.update_state("my_workflow", mock_sess, {"name": "Alice", "email": "alice@example.com"})
print(f"State: {store.get_state('my_workflow', mock_sess)}")

# Session only contains the session ID
print(f"Session contents: {mock_sess}")

Current step: step_1
State: {'name': 'Alice', 'email': 'alice@example.com'}
Session contents: {'_workflow_session_id': 'c51f6c6f-f753-403c-a92a-3ae4e4275ba1'}

# Verify it satisfies the protocol
assert isinstance(store, WorkflowStateStore), "InMemoryWorkflowStateStore must implement WorkflowStateStore protocol"
print("Protocol check passed!")

Protocol check passed!