Skip to content

Asynchronous Execution, Queues & Durable Workflows

litestar-mcp supports two complementary execution models for background MCP Tasks (io.modelcontextprotocol/tasks), multi-step DAG workflows (DynamicWorkflow), and streaming agent turns (AgentRuntime):

  1. Single-Process In-Memory Execution: Built into the base litestar-mcp package with zero external infrastructure dependencies. Ideal for local development, litestar mcp stdio servers, and single-replica deployments.
  2. Distributed Queue Execution (litestar-mcp[queues]): Powered by litestar-queues, Litestar's StoreRegistry, and ChannelsPlugin. Scales tool tasks, DAG nodes, and long-running agent turns across horizontally autoscaled worker replicas.

Single-Process vs Distributed Execution Matrix

Capability Single-Process (Base Package) Distributed (litestar-mcp[queues]) Shared State & Coordination
MCP Tasks (io.modelcontextprotocol/tasks) InProcessTaskExecutor (process-local asyncio.Task runners in MCPTaskStore) QueueTaskExecutor (auto-wired when QueuePlugin is registered; dispatches litestar_mcp.mcp.run_tool_task) MCPTaskStore backed by the "litestar_mcp_tasks" Litestar Store + mcp-task:{id}:job tracking
Principal & DI Context Clones the active HTTP Request scope in-memory (stripping request-scoped Dishka containers for fresh resolution) Captures PrincipalSnapshot (user, auth, litestar-security SecurityContext) and rehydrates Provide + Dishka Scope.REQUEST in workers ToolTaskCall persisted in TaskRecord.dispatch (dispatchInternal)
Cooperative Cancellation & input_required Resume Local asyncio.Event / asyncio.Queue signalling within the originating process Cross-replica QueueService.cancel_task(include_running=True) and stateless executor.resume(call) re-enqueueing on tasks/update Shared Store record + litestar-queues durable cancellation signal
DAG Workflows (DynamicWorkflow) WorkflowEngine (topological wave execution via in-process AnyIO TaskGroup and memory streams) DurableWorkflowEngine (plugin injecting workflow_engine, scheduling advance_workflow, execute_workflow_node, and watch_workflow) WorkflowStore backed by "litestar_mcp_workflows" with single-writer run and write-once node keys
Agent Turn Execution & SSE Streaming AgentChatController (POST /agent/chat and POST /agent/turns running inline inside the HTTP request) QueuedAgentChatController + QueuedTurnManager (POST /agent/turns returns 202 Accepted; workers run litestar_mcp.agent.run_turn) LitestarStoreSessionStore + ChannelsPlugin history replay and live AgentStreamFrame fan-out

Architecture Overview

The interactive diagram below illustrates how task-augmented tools/call requests, principal snapshots, cooperative cancellation, and multi-round MCPInputRequiredResult resumptions flow through QueueTaskExecutor and InProcessTaskExecutor:

Distributed QueueTaskExecutor on litestar-queues tools/call with params.task persists a TaskRecord in MCPTaskStore (litestar_mcp_tasks), snapshots PrincipalSnapshot and SecurityContext, enqueues to litestar-queues, rehydrates Dishka Scope.REQUEST on any worker replica, and supports cooperative tasks/cancel and multi-round MCPInputRequiredResult via tasks/update. 1. Task Submission tools/call (params.task) Captures PrincipalSnapshot request.user & request.auth SecurityContext (scopes/roles) Owner ID: user:<subject> Returns CreateTaskResult 2. Durable Store & Broker MCPTaskStore + QueuePlugin Store: litestar_mcp_tasks QueueTaskExecutor.submit() Per-route queue="..." override Postgres / Redis / SQLSpec status: "working" 3. Worker Rehydration Replica Worker Execution Rehydrates user & auth in scope Restores SecurityContext Enters Dishka Scope.REQUEST Polls store for cancel_requested ToolContext.report_progress 4. Pause / Complete tasks/result & update MCPInputRequiredResult -> status: "input_required" tasks/update re-enqueues tasks/cancel sets flag Cross-Replica Safe Multi-Round Human-in-the-Loop & Cooperative Cancellation Across Replicas Replica A: tools/call -> Worker B: MCPInputRequiredResult -> Replica C: tasks/update(input) -> Worker A: tasks/result
Single-Process InProcessTaskExecutor Lifecycle Default single-process task executor running background AnyIO tasks inside the Litestar process while persisting state in MCPTaskStore. 1. Client Request tools/call (params.task) Validates task="optional"|"required" Binds owner_id from authenticated user Applies MCPTaskConfig.default_ttl Enforces max_tasks capacity limit Immediate taskId handle 2. InProcessTaskExecutor AnyIO Background TaskGroup Spawns local async task in same process Holds live CancelScope & input Event Executes tool via internal ASGI dispatch Persists TaskRecord to MCPTaskStore Zero external broker required 3. Polling, Resume & Cancel tasks/get • list • result • cancel Owner-isolated lookup (non-owners masked) tasks/result awaits terminal status tasks/update wakes local input waiter tasks/cancel triggers AnyIO CancelScope notifications/tasks/status SSE When to Use InProcessTaskExecutor vs QueueTaskExecutor InProcessTaskExecutor: Local dev, stdio servers & single-worker apps | QueueTaskExecutor: Multi-replica HTTP deployments

Installing litestar-mcp[queues]

Install the queues extra to unlock litestar_mcp.queues (QueueTaskExecutor, DurableWorkflowEngine, WorkflowStore, TaskDependencyChain, QueuedAgentChatController, QueuedTurnManager, AgentRegistry, and PrincipalSnapshot):

uv add "litestar-mcp[queues]"

pip install "litestar-mcp[queues]"

pdm add "litestar-mcp[queues]"

poetry add "litestar-mcp[queues]"

Guides in This Section

MCP Tasks Extension

Configure MCPTaskConfig, declare task-capable tools, drive the working / input_required state machine with MCPInputRequiredResult, and handle SEP-2663 tool errors.

MCP Tasks Guide →

Distributed Task Queues

Auto-wire QueueTaskExecutor with QueuePlugin, override queues per controller or tool, rehydrate SecurityContext and Dishka dependencies in workers, and resume across replicas.

Distributed Queues →

Durable DAG Workflows

Build DynamicWorkflow and WorkflowNode pipelines, compare WorkflowEngine with DurableWorkflowEngine, and inspect single-writer WorkflowStore keys and watchdog recovery.

Durable Workflows →

Queued Agent Turns

Offload multi-step AgentRuntime turns to background workers via QueuedAgentChatController and QueuedTurnManager with deterministic SSE frame replay over ChannelsPlugin.

Queued Agent Turns →

Multi-Replica Deployment

Production checklist for horizontal scaling: stateless Streamable HTTP routing, shared StoreRegistry, cross-worker ChannelsPlugin, external worker placement, and conformance testing.

Deployment Checklist →