Durable DAG Workflows (DynamicWorkflow & DurableWorkflowEngine)¶

litestar-mcp provides a declarative Directed Acyclic Graph (DAG) workflow abstraction (DynamicWorkflow and WorkflowNode) that can run either in-process via WorkflowEngine or across distributed litestar-queues worker replicas via DurableWorkflowEngine and WorkflowStore.

Defining DAGs (DynamicWorkflow & WorkflowNode)¶

A DynamicWorkflow connects discrete Python functions (handler) and Agent turns (agent) into an acyclic dependency graph. At registration time, WorkflowEngine.validate_dependencies(workflow) verifies that all depends_on references exist and that the graph contains no cycles.

"""Snippet: durable workflow execution on litestar-queues."""

from litestar import Litestar, post
from litestar_queues import QueuePlugin

from litestar_mcp.agent.workflow import DynamicWorkflow, WorkflowContext
from litestar_mcp.queues import DurableWorkflowEngine


def step_extract(context: WorkflowContext) -> dict[str, int]:
    """Extract raw metrics."""
    return {"raw_count": 42}


def step_transform(context: WorkflowContext) -> dict[str, int]:
    """Transform extracted metrics."""
    raw = int(context.node_outputs["extract"]["raw_count"])
    return {"processed_count": raw * 2}


metrics_pipeline = DynamicWorkflow(name="metrics_pipeline")
metrics_pipeline.add_node("extract", handler=step_extract)
metrics_pipeline.add_node("transform", handler=step_transform, depends_on=["extract"])


@post("/pipelines/metrics")
async def start_metrics_pipeline(workflow_engine: DurableWorkflowEngine) -> dict[str, str]:
    """Start a durable run; any worker sharing the queue backend executes it."""
    return {"run_id": await workflow_engine.start("metrics_pipeline")}


def build() -> Litestar:
    """Build a Litestar application configured for durable workflows."""
    return Litestar(
        route_handlers=[start_metrics_pipeline],
        plugins=[QueuePlugin(), DurableWorkflowEngine([metrics_pipeline])],
    )

WorkflowNode Parameters (workflow.add_node(...))¶

In-Process WorkflowEngine vs Distributed DurableWorkflowEngine¶

The exact same DynamicWorkflow instance can be executed in-process or distributed across litestar-queues:

Running In-Process with WorkflowEngine¶

from litestar_mcp.agent.workflow import DynamicWorkflow, WorkflowContext, WorkflowEngine


async def fetch_items(ctx: WorkflowContext) -> list[str]:
    """Return initial items from workflow state."""
    return list(ctx.state.get("items", []))


async def count_items(ctx: WorkflowContext) -> int:
    """Count items produced by fetch_items."""
    return len(ctx.node_outputs["fetch"])


pipeline = DynamicWorkflow(name="local_pipeline")
pipeline.add_node("fetch", handler=fetch_items)
pipeline.add_node("count", handler=count_items, depends_on=["fetch"])


async def run_local() -> int:
    """Execute the DAG in-process and return the count output."""
    engine = WorkflowEngine()
    ctx = await engine.execute_run(pipeline, initial_state={"items": ["a", "b", "c"]})
    return int(ctx.node_outputs["count"])

DurableWorkflowEngine Plugin Architecture¶

Register DurableWorkflowEngine(workflows, ...) in Litestar(plugins=[QueuePlugin(...), DurableWorkflowEngine([...])]):

Automatic DI & TaskDependencyChain Wiring¶

During on_app_init, DurableWorkflowEngine performs three setup steps automatically:

1. TaskDependencyChain.attach: Appends DurableWorkflowEngine.resolve_task_dependencies to QueuePlugin.config.task_dependency_resolver so the workflow_engine argument is injected into advance_workflow (litestar_mcp.workflow.advance), execute_workflow_node (litestar_mcp.workflow.execute_node), and watch_workflow (litestar_mcp.workflow.watchdog). Because TaskDependencyChain merges resolvers in order, QueueTaskExecutor and DurableWorkflowEngine coexist on the same QueuePlugin without manual resolver wiring.
2. Default Store Resolution: Resolves "litestar_mcp_workflows" from app_config.stores (StoreRegistry) when store=None.
3. Route Handler Dependency (workflow_engine): Registers "workflow_engine": Provide(self._provide_engine, sync_to_thread=False) so any Litestar route handler can declare workflow_engine: DurableWorkflowEngine to call await workflow_engine.start(...) or await workflow_engine.snapshot(run_id).

WorkflowStore Single-Writer Key Separation¶

When parallel sibling nodes in a DAG wave finish simultaneously on different worker replicas, updating a single shared run document would cause read-modify-write races and lost node outputs. WorkflowStore eliminates sibling write contention by separating run metadata from write-once node records:

To inspect a running or completed workflow, call await workflow_engine.snapshot(run_id), which loads the WorkflowRun and all persisted NodeRecord entries in parallel (store.get_nodes) and reconstructs a WorkflowContext via store.to_context(run, nodes).

Note: Node outputs must be JSON-serializable. If a node handler returns an unserializable object (TypeError, msgspec.EncodeError, or SerializationException), DurableWorkflowEngine.execute_node catches the encoding failure and persists a NodeRecord(node=node, status="failed", output=None, error="TypeError: node output is not JSON serializable") so the workflow terminates deterministically instead of stalling.

Forever-Unique Node Scheduling & reset_task_identity¶

Multiple completed upstream nodes can trigger advance_workflow(run_id) concurrently. To guarantee each node in a run executes at most once:

1. execute_workflow_node is decorated with @task(EXECUTE_NODE_TASK, queue="workflow", retries=0, unique_until="forever", unique_by="arguments").
2. DurableWorkflowEngine.advance(run_id) enqueues ready nodes with key=f"wf-node:{run_id}:{name}". The queue backend deduplicates concurrent enqueue calls for the same wf-node:{run_id}:{name} key.
3. Inside execute_node(run_id, node), WorkflowStore.get_node(run_id, node) performs an additional idempotency check before executing the node and enqueuing the next advance_workflow.
4. When all nodes in wf.nodes have a persisted NodeRecord (or when the run times out in watch()), DurableWorkflowEngine finalizes WorkflowRun.status ("completed" or "failed") and calls await service.reset_task_identity(f"wf-node:{run_id}:{name}") for every node in the workflow to release the unique key reservations cleanly.

Per-Run Watchdog Recovery¶

If a transient broker or worker crash drops an advance_workflow enqueue after a node writes its NodeRecord, the per-run watchdog heals the run automatically:

1. Initial Watchdog Scheduling: When watchdog_interval is not None (default 30.0 seconds), workflow_engine.start() enqueues watch_workflow with tick=1, key=f"wf-watchdog:{run_id}:1", and run_after=self.watchdog_interval.
2. Backend Validation: Delayed run_after tasks require an asynchronous queue execution backend ("local", "redis", "sqlspec", etc.). If QueuePlugin is configured with execution_backend="immediate" while watchdog_interval is enabled, workflow_engine.start() raises ImproperlyConfiguredException.
3. Timeout & Re-Advance Loop: On each tick, DurableWorkflowEngine.watch(run_id, tick):
4. Returns immediately if the run is already "completed" or "failed".
5. Checks whether time.time() - run.created_at >= self.run_timeout_seconds. If exceeded, marks the run "failed" with error="workflow timed out" and resets all wf-node:{run_id}:{name} identities.
6. Calls await self.advance(run_id) to schedule any unscheduled ready nodes or finalize the run.
7. If the run is still "running", schedules tick tick + 1 under key=f"wf-watchdog:{run_id}:{tick + 1}" with run_after=self.watchdog_interval.