litestar-workflows

Workflow automation for Litestar with human approval chains, automated pipelines, and web-based workflow management.

litestar-workflows is a flexible, async-first workflow automation framework built specifically for the Litestar ecosystem. It enables you to define complex business processes as code, combining automated steps with human approval checkpoints.

Getting Started

New to litestar-workflows? Start here for installation and your first workflow.

Getting Started
Core Concepts

Understand the fundamental concepts: workflows, steps, context, and execution.

Core Concepts
How-To Guides

Practical guides for common tasks: approvals, parallel execution, and more.

How-To Guides
API Reference

Complete API documentation for all public classes and functions.

API Reference
Cookbook

Real-world workflow recipes: expense approval, document review, onboarding, and more.

Cookbook

Key Features

  • Async-First Design: Native async/await throughout, leveraging Litestar’s async foundation

  • Human + Machine Tasks: Combine automated processing with human approval checkpoints

  • Composable Workflows: Build complex workflows from simple, reusable primitives

  • Type-Safe: Full typing with Protocol-based interfaces for IDE support

  • Litestar Integration: Deep integration with Litestar’s DI, guards, and plugin system

  • Flexible Execution: Local execution engine with optional distributed backends

  • Database Persistence: SQLAlchemy-backed storage with multi-tenancy support ([db] extra)

  • REST API Plugin: Production-ready web API with OpenAPI docs and MermaidJS visualization ([web] extra)

Quick Example

Here’s a taste of what a workflow looks like:

from litestar_workflows import (
    WorkflowDefinition,
    Edge,
    BaseMachineStep,
    BaseHumanStep,
    LocalExecutionEngine,
    WorkflowRegistry,
    WorkflowContext,
)

class SubmitRequest(BaseMachineStep):
    name = "submit"

    async def execute(self, context: WorkflowContext) -> None:
        context.set("submitted", True)

class ManagerApproval(BaseHumanStep):
    name = "manager_approval"
    title = "Approve Request"
    form_schema = {"type": "object", "properties": {"approved": {"type": "boolean"}}}

# Define and run workflow
definition = WorkflowDefinition(
    name="approval",
    version="1.0.0",
    steps={"submit": SubmitRequest(), "approve": ManagerApproval()},
    edges=[Edge("submit", "approve")],
    initial_step="submit",
    terminal_steps={"approve"},
)

Installation

pip install litestar-workflows

# With optional extras
pip install litestar-workflows[db]  # SQLAlchemy persistence
pip install litestar-workflows[ui]  # Web UI templates

Indices and tables