Web Plugin API Reference

Complete API documentation for the litestar_workflows.web module.

Installation

Install with the [web] extra:

pip install litestar-workflows[web]

Module Exports

The following are exported from litestar_workflows.web:

from litestar_workflows.web import (
    # Configuration (for advanced use)
    WorkflowWebConfig,

    # Controllers (auto-registered by plugin)
    WorkflowDefinitionController,
    WorkflowInstanceController,
    HumanTaskController,

    # DTOs
    StartWorkflowDTO,
    WorkflowDefinitionDTO,
    WorkflowInstanceDTO,
    WorkflowInstanceDetailDTO,
    StepExecutionDTO,
    HumanTaskDTO,
    CompleteTaskDTO,
    ReassignTaskDTO,
    GraphDTO,

    # Graph utilities
    generate_mermaid_graph,
    generate_mermaid_graph_with_state,
    parse_graph_to_dict,
)

Plugin Configuration

The REST API is configured through the main WorkflowPlugin using WorkflowPluginConfig. The API is enabled by default when you add the plugin to your Litestar application.

WorkflowPluginConfig API Options

class WorkflowPluginConfig

Configuration dataclass for the WorkflowPlugin, including REST API options.

API Configuration:

Parameters:

• enable_api (bool) – Enable REST API endpoints (default: True)

• api_path_prefix (str) – URL path prefix for all workflow endpoints

• api_guards (list[Guard]) – List of Litestar guards to apply to all routes

• api_tags (list[str]) – OpenAPI tags for route grouping

• include_api_in_schema (bool) – Include routes in OpenAPI documentation

Defaults:

WorkflowPluginConfig(
    enable_api=True,
    api_path_prefix="/workflows",
    api_guards=[],
    api_tags=["Workflows"],
    include_api_in_schema=True,
)

Example:

from litestar import Litestar
from litestar_workflows import WorkflowPlugin, WorkflowPluginConfig

app = Litestar(
    plugins=[
        WorkflowPlugin(
            config=WorkflowPluginConfig(
                enable_api=True,  # Default - API auto-enabled
                api_path_prefix="/api/v1/workflows",
                api_guards=[require_auth],
                api_tags=["Business Processes"],
            )
        ),
    ],
)

Controllers

The plugin registers three controller classes automatically when enable_api=True.

WorkflowDefinitionController

class WorkflowDefinitionController

REST API controller for workflow definitions.

Path: {api_path_prefix}/definitions

Tags: Workflow Definitions

async list_definitions(workflow_registry, active_only=True) → list[WorkflowDefinitionDTO]

List all registered workflow definitions.

Endpoint: GET /definitions

Parameters:

• workflow_registry (WorkflowRegistry) – Injected workflow registry

• active_only (bool) – If True, only return active definitions

Returns:

List of workflow definition DTOs

Example Response:

[
  {
    "name": "expense_approval",
    "version": "1.0.0",
    "description": "Multi-level expense approval",
    "steps": ["submit", "manager_approval", "process"],
    "edges": [
      {"source": "submit", "target": "manager_approval", "condition": null}
    ],
    "initial_step": "submit",
    "terminal_steps": ["process"]
  }
]

async get_definition(name, workflow_registry, version=None) → WorkflowDefinitionDTO

Get a specific workflow definition by name.

Endpoint: GET /definitions/{name}

Parameters:

• name (str) – Workflow name

• workflow_registry (WorkflowRegistry) – Injected workflow registry

• version (str | None) – Optional specific version (defaults to latest)

Returns:

Workflow definition DTO

Raises:

NotFoundException – If workflow not found

async get_definition_graph(name, workflow_registry, graph_format='mermaid') → GraphDTO

Get workflow graph visualization.

Endpoint: GET /definitions/{name}/graph

Parameters:

• name (str) – Workflow name

• workflow_registry (WorkflowRegistry) – Injected workflow registry

• graph_format (str) – Output format (“mermaid” or “json”)

Returns:

Graph DTO with MermaidJS source and node/edge data

Raises:

NotFoundException – If workflow not found or unknown format

Example Response:

{
  "mermaid_source": "graph TD\n    submit[Submit]\n    ...",
  "nodes": [
    {"id": "submit", "label": "Submit", "type": "machine", "is_initial": true, "is_terminal": false}
  ],
  "edges": [
    {"source": "submit", "target": "manager_approval"}
  ]
}

WorkflowInstanceController

class WorkflowInstanceController

REST API controller for workflow instances.

Path: {api_path_prefix}/instances

Tags: Workflow Instances

async start_workflow(data, workflow_engine, workflow_registry) → WorkflowInstanceDTO

Start a new workflow instance.

Endpoint: POST /instances

Parameters:

• data (DTOData[StartWorkflowDTO]) – Start workflow request data

• workflow_engine (LocalExecutionEngine) – Injected execution engine

• workflow_registry (WorkflowRegistry) – Injected workflow registry

Returns:

Created workflow instance DTO

Raises:

NotFoundException – If workflow definition not found

Request Body:

{
  "definition_name": "expense_approval",
  "input_data": {
    "requester": "alice@example.com",
    "amount": 2500.00
  },
  "user_id": "alice@example.com"
}

async list_instances(workflow_instance_repo, workflow_name=None, status=None, limit=50, offset=0) → list[WorkflowInstanceDTO]

List workflow instances with optional filtering.

Endpoint: GET /instances

Parameters:

• workflow_instance_repo (WorkflowInstanceRepository) – Injected instance repository

• workflow_name (str | None) – Filter by workflow name

• status (str | None) – Filter by status

• limit (int) – Maximum results (1-100)

• offset (int) – Pagination offset

Returns:

List of workflow instance DTOs

async get_instance(instance_id, workflow_instance_repo) → WorkflowInstanceDetailDTO

Get detailed workflow instance information.

Endpoint: GET /instances/{instance_id}

Parameters:

• instance_id (UUID) – The workflow instance UUID

• workflow_instance_repo (WorkflowInstanceRepository) – Injected instance repository

Returns:

Detailed instance DTO with step history

Raises:

NotFoundException – If instance not found

async get_instance_graph(instance_id, workflow_instance_repo, workflow_registry) → GraphDTO

Get instance graph with execution state highlighting.

Endpoint: GET /instances/{instance_id}/graph

Parameters:

• instance_id (UUID) – The workflow instance UUID

• workflow_instance_repo (WorkflowInstanceRepository) – Injected instance repository

• workflow_registry (WorkflowRegistry) – Injected workflow registry

Returns:

Graph DTO with state-based MermaidJS styling

Raises:

NotFoundException – If instance not found

The returned graph uses CSS styling to indicate step states:

• Completed: fill:#90EE90,stroke:#006400 (green)

• Current: fill:#FFD700,stroke:#FFA500 (yellow)

• Failed: fill:#FF6B6B (red)

async cancel_instance(instance_id, workflow_engine, workflow_instance_repo, reason='User canceled') → WorkflowInstanceDTO

Cancel a running workflow instance.

Endpoint: POST /instances/{instance_id}/cancel

Parameters:

• instance_id (UUID) – The workflow instance UUID

• workflow_engine (LocalExecutionEngine) – Injected execution engine

• workflow_instance_repo (WorkflowInstanceRepository) – Injected instance repository

• reason (str) – Cancellation reason

Returns:

Updated instance DTO

Raises:

NotFoundException – If instance not found

async retry_instance(instance_id, workflow_engine, workflow_instance_repo, from_step=None) → WorkflowInstanceDTO

Retry a failed workflow from a specific step.

Endpoint: POST /instances/{instance_id}/retry

Parameters:

• instance_id (UUID) – The workflow instance UUID

• workflow_engine (LocalExecutionEngine) – Injected execution engine

• workflow_instance_repo (WorkflowInstanceRepository) – Injected instance repository

• from_step (str | None) – Step name to retry from (defaults to failed step)

Returns:

Updated instance DTO

Raises:

NotFoundException – If instance not found

HumanTaskController

class HumanTaskController

REST API controller for human tasks.

Path: {api_path_prefix}/tasks

Tags: Human Tasks

async list_tasks(human_task_repo, assignee_id=None, assignee_group=None, status='pending') → list[HumanTaskDTO]

List human tasks with optional filtering.

Endpoint: GET /tasks

Parameters:

• human_task_repo (HumanTaskRepository) – Injected human task repository

• assignee_id (str | None) – Filter by assignee user ID

• assignee_group (str | None) – Filter by assignee group

• status (str) – Filter by task status

Returns:

List of human task DTOs

async get_task(task_id, human_task_repo) → HumanTaskDTO

Get detailed information about a human task.

Endpoint: GET /tasks/{task_id}

Parameters:

• task_id (UUID) – The task UUID

• human_task_repo (HumanTaskRepository) – Injected task repository

Returns:

Human task DTO with form schema

Raises:

NotFoundException – If task not found

async complete_task(task_id, data, workflow_engine, human_task_repo) → WorkflowInstanceDTO

Complete a human task with form data.

Endpoint: POST /tasks/{task_id}/complete

Parameters:

• task_id (UUID) – The task UUID

• data (DTOData[CompleteTaskDTO]) – Task completion data

• workflow_engine (LocalExecutionEngine) – Injected execution engine

• human_task_repo (HumanTaskRepository) – Injected task repository

Returns:

Updated workflow instance DTO

Raises:

NotFoundException – If task not found

Request Body:

{
  "output_data": {
    "approved": true,
    "comments": "Looks good!"
  },
  "completed_by": "manager@example.com"
}

async reassign_task(task_id, data, human_task_repo) → HumanTaskDTO

Reassign a task to another user.

Endpoint: POST /tasks/{task_id}/reassign

Parameters:

• task_id (UUID) – The task UUID

• data (DTOData[ReassignTaskDTO]) – Reassignment data

• human_task_repo (HumanTaskRepository) – Injected task repository

Returns:

Updated task DTO

Raises:

NotFoundException – If task not found

Data Transfer Objects

Request DTOs

class StartWorkflowDTO

DTO for starting a workflow instance.

Parameters:

• definition_name (str) – Name of the workflow definition to instantiate

• input_data (dict[str, Any] | None) – Initial data to pass to the workflow context

• correlation_id (str | None) – Optional correlation ID for tracking related workflows

• user_id (str | None) – Optional user ID who started the workflow

• tenant_id (str | None) – Optional tenant ID for multi-tenancy

class CompleteTaskDTO

DTO for completing a human task.

Parameters:

• output_data (dict[str, Any]) – Data submitted by the user completing the task

• completed_by (str) – User ID who completed the task

• comment (str | None) – Optional comment about the completion

class ReassignTaskDTO

DTO for reassigning a human task.

Parameters:

• new_assignee (str) – User ID to assign the task to

• reason (str | None) – Optional reason for reassignment

Response DTOs

class WorkflowDefinitionDTO

DTO for workflow definition metadata.

Parameters:

• name (str) – Workflow name

• version (str) – Workflow version

• description (str) – Human-readable description

• steps (list[str]) – List of step names in the workflow

• edges (list[dict[str, Any]]) – List of edge definitions (source, target, condition)

• initial_step (str) – Name of the starting step

• terminal_steps (list[str]) – List of terminal step names

class WorkflowInstanceDTO

DTO for workflow instance summary.

Parameters:

• id (UUID) – Instance UUID

• definition_name (str) – Name of the workflow definition

• status (str) – Current execution status

• current_step (str | None) – Currently executing step (if applicable)

• started_at (datetime) – When the workflow started

• completed_at (datetime | None) – When the workflow completed (if finished)

• created_by (str | None) – User who started the workflow

class WorkflowInstanceDetailDTO

DTO for detailed workflow instance information.

Extends WorkflowInstanceDTO with context and execution history.

Parameters:

• id (UUID) – Instance UUID

• definition_name (str) – Name of the workflow definition

• status (str) – Current execution status

• current_step (str | None) – Currently executing step

• started_at (datetime) – When the workflow started

• completed_at (datetime | None) – When the workflow completed

• created_by (str | None) – User who started the workflow

• context_data (dict[str, Any]) – Current workflow context data

• metadata (dict[str, Any]) – Workflow metadata

• step_history (list[StepExecutionDTO]) – List of step executions

• error (str | None) – Error message if workflow failed

class StepExecutionDTO

DTO for step execution record.

Parameters:

• id (UUID) – Step execution UUID

• step_name (str) – Name of the executed step

• status (str) – Execution status

• started_at (datetime) – When execution started

• completed_at (datetime | None) – When execution completed (if finished)

• error (str | None) – Error message if execution failed

class HumanTaskDTO

DTO for human task summary.

Parameters:

• id (UUID) – Task UUID

• instance_id (UUID) – Workflow instance UUID

• step_name (str) – Name of the human task step

• title (str) – Display title for the task

• description (str | None) – Detailed task description

• assignee (str | None) – User ID assigned to complete the task

• status (str) – Task status (pending, completed, canceled)

• due_date (datetime | None) – Optional due date for task completion

• created_at (datetime) – When the task was created

• form_schema (dict[str, Any] | None) – Optional JSON Schema for task form

class GraphDTO

DTO for workflow graph visualization.

Parameters:

• mermaid_source (str) – MermaidJS graph definition

• nodes (list[dict[str, Any]]) – List of node definitions

• edges (list[dict[str, Any]]) – List of edge definitions

Graph Utilities

generate_mermaid_graph(definition: WorkflowDefinition) → str

Generate a MermaidJS graph representation of a workflow definition.

Parameters:

definition (WorkflowDefinition) – The workflow definition to visualize

Returns:

MermaidJS flowchart definition string

Example:

from litestar_workflows.web import generate_mermaid_graph

mermaid = generate_mermaid_graph(definition)
# Returns:
# graph TD
#     submit[Submit]
#     review{{Review}}
#     submit --> review

generate_mermaid_graph_with_state(definition, current_step=None, completed_steps=None, failed_steps=None) → str

Generate a MermaidJS graph with execution state highlighting.

Parameters:

• definition (WorkflowDefinition) – The workflow definition to visualize

• current_step (str | None) – Name of the currently executing step

• completed_steps (list[str] | None) – List of successfully completed step names

• failed_steps (list[str] | None) – List of failed step names

Returns:

MermaidJS flowchart with state styling

Example:

mermaid = generate_mermaid_graph_with_state(
    definition,
    current_step="review",
    completed_steps=["submit"],
    failed_steps=[],
)

parse_graph_to_dict(definition: WorkflowDefinition) → dict[str, Any]

Parse a workflow definition into a dictionary representation.

Parameters:

definition (WorkflowDefinition) – The workflow definition to parse

Returns:

Dictionary with nodes and edges lists

Example:

graph_dict = parse_graph_to_dict(definition)
# Returns:
# {
#     "nodes": [
#         {"id": "submit", "label": "Submit", "type": "machine", "is_initial": True, "is_terminal": False}
#     ],
#     "edges": [
#         {"source": "submit", "target": "review"}
#     ]
# }

Route Summary

Complete route listing for the REST API (when enable_api=True):

Method	Path	Description
GET	/definitions	List workflow definitions
GET	/definitions/{name}	Get workflow definition
GET	/definitions/{name}/graph	Get workflow graph
POST	/instances	Start new workflow
GET	/instances	List workflow instances
GET	/instances/{id}	Get instance details
GET	/instances/{id}/graph	Get instance graph with state
POST	/instances/{id}/cancel	Cancel workflow
POST	/instances/{id}/retry	Retry failed workflow
GET	/tasks	List tasks
GET	/tasks/{id}	Get task details
POST	/tasks/{id}/complete	Complete task
POST	/tasks/{id}/reassign	Reassign task

All paths are relative to the configured api_path_prefix (default: /workflows).

See Also

• Web Plugin (REST API) - Web Plugin integration guide

• Database Persistence - Database persistence setup

• Working with Human Tasks - Human task patterns

• Phase 3: Web Plugin Architecture - Web plugin architecture details