Workflows

A workflow is a Pydantic state class plus a graph. You build the graph with LangGraph’s StateGraph; AgentDeck compiles it and plays it on the Runtime the same way it plays an agent.

# .agentdeck/workflows/new_booking/workflow.py
from langgraph.graph import END, StateGraph
from pydantic import BaseModel
 
from agentdeck import Workflow
 
 
class BookingState(BaseModel):
    request: str
    quote: str = ""
 
 
def _quote(state: BookingState) -> dict:
    return {"quote": f"€420 for {state.request}"}
 
 
def _build_graph() -> StateGraph:
    graph = StateGraph(BookingState)
    graph.add_node("quote", _quote)
    graph.set_entry_point("quote")
    graph.add_edge("quote", END)
    return graph
 
 
new_booking = Workflow(name="NewBooking", state=BookingState, graph=_build_graph)

Run it by name; the return value is the final state:

import asyncio
 
from agentdeck import Deck
 
 
async def main() -> None:
    async with Deck.from_project() as deck:
        state = await deck.run("NewBooking", {"request": "Berlin -> Munich"})
        print(state["quote"])
 
 
asyncio.run(main())

Nodes that do the interesting work

from agentdeck.authoring import AgentNode
 
graph.add_node("draft", AgentNode(drafting_agent, input_key="request", output_key="draft"))

An agent used inside a workflow node compiles standalone, with no Deck catalog in view: its own MCP servers resolve the same way a root agent’s do, but handoffs=/skills= naming another catalog entry do not — put those on a root agent instead.

Durability and human approval

Set durable=True and the graph compiles with a checkpointer (AGENTDECK_CHECKPOINT_*), so a run is identified by session_id (its thread_id) and survives the process dying. Only then can a node pause for a human:

from langgraph.types import interrupt
 
 
def _confirm(state) -> dict:
    decision = interrupt({"question": "Send this quote?"})
    return {"approved": decision == "yes"}

A paused run returns an InterruptResult — {"type": "interrupt", "payload": …, "thread_id": …} — instead of a final state. List it in the inbox and answer it by run_id:

import asyncio
 
from agentdeck import Deck
 
 
async def main() -> None:
    async with Deck.from_project() as deck:
        paused = await deck.run("Approval", {}, session_id="quote-42")
        if paused["type"] == "interrupt":
            [mine] = [p for p in await deck.pending() if p.thread_id == "quote-42"]
            await deck.answer(mine.run_id, "yes")
 
 
asyncio.run(main())

Timers

sleep_until(when) pauses a durable run until a timezone-aware moment, using the same interrupt machinery with a {"type": "timer", "wake_at": …} payload — so a timer-paused thread is distinguishable from a human-paused one in the inbox. Naive datetimes are rejected.

AgentDeck runs no daemon: you own the cadence. Deck.due_resumes() lists timer threads whose moment has passed, and Deck.tick() resumes every one of them.

Known limit: the listing reads the checkpointer, not the event log. Both calls find due threads through each workflow’s own LangGraph checkpointer rather than the event log — deliberately, since the checkpoint backend defaults to durable (sqlite) while the event store defaults to memory, and a due timer surviving a process restart depends on the checkpointer’s own durability. tick() still resumes a due thread through the Runtime when it matches a run Deck.pending() already knows about, so that resume itself is recorded in the log like any other; only a thread with no logged run at all falls back to resuming straight off the checkpointer.

Streaming and composition

Deck.stream() yields a node.updated event per completed node and a custom event per get_stream_writer() call, then one terminal run.completed carrying the final state — or a run.interrupted event in its place when the run pauses.

Workflow.as_tool() turns the whole workflow into a tool an agent can call (pass it in Agent(tools=[...])), which is how a conversation reaches deterministic multi-step work.

A workflow exposed as a tool must be durable=False. A tool call carries no thread, and a durable workflow needs one to load and persist its checkpoint — so build() rejects the combination rather than letting it fail the first time a model reaches for the tool:

agent 'booking' uses workflow 'Onboarding' as a tool, but it is durable=True.

If the work genuinely needs to survive a restart or pause for a human, it is a root invocable, not an ability: call it with deck.run("Onboarding", state, session_id=...), where you control the thread.