Deck

Deck is the one object a project talks to. Build it directly from Agent/Workflow objects, or discover the same catalog from a project directory with Deck.from_project() — either way it is the composition root: every turn started through it plays on the same Runtime the HTTP surface runs chats on, and is recorded the same way.

from agentdeck import Agent
 
greeter = Agent(name="Greeter", instructions="You are a friendly scheduling assistant. Keep replies to one short sentence.")

import asyncio
 
from agentdeck import Deck
 
 
async def main() -> None:
    async with Deck.from_project() as deck:
        result = await deck.run("Greeter", "hello")
        print(result.output)
        print(result.usage)
        print(result.run_id)
 
 
asyncio.run(main())

Construction and lifecycle

Deck(
    agents=[...],                   # Agent instances
    workflows=[...],                # Workflow instances
    skills="./skills",              # a path, a sequence of paths, or a Skills(...) object
    mcp=".mcp.json",                # a path, or an MCP(...) object
    context=Calendar,               # the *type* of the per-run context, if this catalog wants one
    observers=[Langfuse()],         # taps on the event stream; None reads settings, () means none
)

Deck.from_project(path=".agentdeck") discovers the same four catalog arguments from today’s directory layout (agents/<bundle>/agent.py, workflows/<bundle>/workflow.py, skills/*/SKILL.md, .mcp.json next to .agentdeck/) and hands them to the same constructor — there is one catalog mechanism underneath either front door.

The lifecycle is NEW -> build() -> BUILT -> (async with) -> OPEN -> CLOSED:

• build() validates every name the catalog references (an unknown skill, MCP server, or workflow-as-tool name; an agent and a workflow sharing a root name) and compiles every agent/workflow to an InvocableSpec. It reads local files only — no network call, no MCP server started — and is idempotent, so it doubles as a CI check. Calling a turn-starting method without an explicit build() still works — async with deck: calls it for you.
• async with deck: starts everything build() left alone: the real engines, the event store, the session factory, the observers, and every configured MCP server. This is what a turn-starting method actually needs; calling one before opening raises.
• CLOSED is terminal. aclose() (or the async with block’s exit) closes the Runtime’s observers and tears down what this Deck itself opened — an MCP(...) it holds, always, and the event store it built from settings, but never one passed in through construction. Reopening a closed Deck raises; construct a new one instead.

One Deck per process. Constructing a second one while the first is still live raises ConfigError, naming both projects. The restriction is real rather than stylistic: every project is mounted under a single module alias and MCP servers are registered process-wide, so two decks side by side would read each other’s bundles and share each other’s servers. Decks one after another are fine — close the first, construct the next:

async def validate_every_project() -> None:
    async with Deck.from_project("./alpha/.agentdeck"):
        ...
 
    # the first deck is closed by the time this one is constructed
    async with Deck.from_project("./beta/.agentdeck"):
        ...

A script that validates several projects in a loop needs that aclose() between them; a service that mounts one deck’s asgi() inside an existing app is unaffected, since that is still one deck. Running two side by side is a capability we intend to add, tracked in issue #213  — until then, a deck per tenant is a process per tenant.

deck.agents/deck.workflows are read-only mappings once built — mutating either raises.

Declaring the context type

context= on the constructor is the type of the application context this catalog’s callables receive. The value itself never goes here — it arrives per run, on run/stream/answer/resume below. Declaring the type is what lets build() check it against every Context[...] parameter in the catalog before anything runs:

from agentdeck import Agent, Context, ContextTypeError, Deck
 
 
class Calendar:
    def find(self, day: str) -> str:
        return f"{day} 09:00"
 
 
class Warehouse:
    """A different application environment entirely."""
 
 
def find_slots(day: str, environment: Context[Calendar]) -> str:
    """Find free appointment slots on a given day."""
    return environment.data.find(day)
 
 
booking = Agent(name="Booking", instructions="Book things.", tools=[find_slots])
 
try:
    Deck(agents=[booking], context=Warehouse).build()
except ContextTypeError as refused:
    print(refused)

Every injection site is walked: an agent’s tools, its instructions= callable, its hooks= methods, and every node of every workflow. The message names the callable (or the node, or the hook method), what it requires, and what the deck provides.

What counts as compatible is what the runtime can decide, and nothing more:

build() is not a partial type checker and does not try to become one. Where no runtime answer exists it accepts and lets the requirement stand or fall when the callable actually runs — refusing on a guess would reject builds a static checker would pass. ContextTypeError is a subclass of ConfigError, so a caller already catching ConfigError around build() keeps catching it.

Declaring nothing is the third case, and the default: no context= on the constructor means no build-time check at all. run(context=...) still works exactly the same — the declaration buys the check, not the capability. Passing an instance where the type belongs (Deck(context=my_calendar)) is refused at construction, since every check would then silently defer and the parameter would promise something it never delivered.

Starting a turn

run/stream resolve name against whichever catalog holds it — an agent or a workflow — and play it on the Runtime, recorded exactly as POST /agents/{name}/chat or POST /workflows/{name} would record it. Pass session_id= for a conversational agent turn (same id, same history across calls) or a durable workflow’s thread_id (required when that workflow is durable = True, so a later call with the same id resumes it). A node that calls interrupt() makes a workflow’s run return an InterruptResult — {"type": "interrupt", "payload": ..., "thread_id": ...} — instead of a final state; see Workflows for what that means for the node that paused, and answer it with answer() below.

What input accepts

For an agent turn: a str — the common case, and unchanged — or a list of content blocks, which is how a turn carries anything that is not text. A string is coerced to one text block on the way in, so the two forms meet as the same thing before any engine sees them.

For a workflow, input is its initial state — the mapping its first node reads, e.g. deck.run("RefundApproval", {"order_id": "A-1003"}). It is JSON data, not content, and the log records it as one data block; see Workflows for what a node does with it.

import base64
 
from agentdeck.core import ImageBlock, TextBlock
 
photo = base64.b64encode(open("receipt.png", "rb").read()).decode()
result = await deck.run(
    "Intake",
    [TextBlock(text="What is the total on this receipt?"), ImageBlock(media_type="image/png", data_b64=photo)],
)

Inline blocks are capped at 1 MB decoded, enforced at construction rather than documented and hoped for: base64 in an event lands in an append-only log and replays down every SSE connection for the life of that run. Anything larger belongs in a ResourceBlock.

An engine that cannot express a block raises ConfigError naming the block type, rather than dropping it — a turn that silently loses its image is worse than one that refuses. Two known limits on the openai-agents engine at the pinned openai-agents==0.17.0: AudioBlock needs the Chat-Completions path (OPENAI_USE_RESPONSES=false), and ResourceBlock/DataBlock are not sent to the model at all.

stream yields the same Event objects a run’s log would hand back after the fact — text.delta per token for an agent, node.updated per completed node for a workflow, a terminal event last — not a rendered wire format. This is the one method here that does not return a TurnResult, because a caller that wants the final answer out of a stream has to fold it from the events, the same way any other consumer of the log would.

context= supplies the application’s own environment for one run — a database handle, a client, whatever the code the run reaches needs. A tool, a dynamic-instructions callable, an agent hook or a workflow node that declares an agentdeck.Context parameter receives it as ctx.data, by reference; see Definitions for how each declares one. Both engines carry it, through their own native runtime-context channel rather than anything agentdeck invented. Three things it deliberately is not: the model never sees it (the context parameter is absent from the tool schema), it is never written to the event log, and it cannot cross the HTTP surface — a live Python object has no wire form, so a context-requiring root is reachable from an embedded Python caller and not from asgi(). That boundary and three others are set out under Where a context does not reach below; read it before you build one into a served or timer-driven path.

type(event) is always Event — it is the envelope, not the discriminator. Switch on event.payload (a match narrows it to TextDelta, MessageCompleted, RunCompleted, …, see Agents for a worked example) or on event.kind if a string is more convenient; either way, include a default case, since an unfamiliar kind still parses as UnknownEvent rather than raising.

TurnResult

An agent’s run assembles a TurnResult from the run’s own run.completed event — never the SDK’s own result object, so a caller depends on agentdeck’s event schema rather than on whichever engine ran the turn. stream does not return one; it yields the events themselves, as above.

Calling Agent.run() directly — the class’s own headless runner, bypassing Deck and the Runtime entirely — still returns the SDK’s own RunResult; see Definitions for that distinction.

Reading a run back

Deck has no public event-log reader — store is deliberately not one of its properties. For a quick check, status(run_id) folds a run’s current RunStatus from its own events:

current = await deck.status(run_id)  # None if the log has never heard of this run_id

A caller that needs the full log — every event, or every run in a namespace — reaches for the same EventStorePort the Runtime itself uses, e.g. agentdeck.composition.resolve_event_store(), or the HTTP surface’s own endpoints once a project is served.

Controlling a run in flight

await deck.pause(run_id, reason="operator stepped away")
events = await deck.resume(run_id)
await deck.cancel(run_id, reason="user closed the tab")

pause and cancel record a request and return immediately — not when the run actually stops, which nobody can know at the moment of asking. resume plays a paused run’s continuation and returns the events it produced, or an empty list if there was nothing to resume. Run Control is the full contract: what a safe point is, why a request is not a status change, and what a resume replays.

Workflow bookkeeping

pending = await deck.pending()  # every run currently waiting on a human, across the catalog
result = await deck.answer(pending[0].run_id, "yes")

pending() lists every run currently WAITING_HUMAN, across every workflow the catalog holds — the approval inbox, read off the event log. answer(run_id, value) answers the interrupt that run is paused on and returns the final state (or the next InterruptResult); it looks the run up itself (which workflow, which thread, which session) from the same source pending() reads, so a caller supplies only the id pending() named and the value.

answer(run_id, value, context=...) and resume(run_id, context=...) take the same context= run does, and it has to be supplied again: the value is never serialized, so the run’s own copy is gone by the time anybody picks it up. Omitting it is not “keep what the run had” — there is nothing kept, and the resumed run reads None. An interrupted node re-runs from its start, so that is the pass where a missing context shows up.

due_resumes(now=None) and tick(now=None) are a separate, timer-only inbox: due_resumes lists threads whose sleep_until has passed, reading each workflow’s own checkpointer rather than the Runtime’s log — deliberately, since the checkpoint backend defaults to durable (sqlite) while the event store defaults to memory, and listing off the log alone would stop surviving a process restart under that default pairing. now defaults to the current UTC time and must be timezone-aware if given. AgentDeck runs no daemon of its own — a cron job, a systemd timer, or a loop calling tick() owns the cadence.

tick’s own resume, unlike the listing, goes through the Runtime whenever a due thread matches a run pending() already knows about — closing that run’s log entry and freeing its session claim, the same as answer() does. Only a thread with no logged run (parked by calling a durable workflow’s own run/resume directly, a deliberately log-free path) falls back to resuming straight off the checkpointer. Neither route carries a context — see below for what a timer-driven resume does to a graph that needed one.

Serving over HTTP

asgi() returns an ASGI application — a FastAPI app whose lifespan opens the deck on startup and closes it on shutdown, so a served deck needs no separate async with:

from agentdeck import Deck
 
deck = Deck.from_project("./.agentdeck")
app = deck.asgi()                          # uvicorn yourmodule:app

That is all agentdeck serve is. It also means you can mount a deck inside an existing service and have it open and close with the host app.

Two details worth knowing. The FastAPI import is deferred inside the method, so agentdeck.deck stays importable without the [serve] extra — you only need it to actually serve. And requests arriving before the lifespan has run get 503 {"status": "starting"} rather than touching a deck that is not open yet.

A third, if any of this catalog’s callables declare a context: it does not reach a served run. A context is a live Python object with no wire form, so an HTTP-started run always carries None — see below.

Observers

The event log is the hub. An observer is a read-only tap on it — telemetry, cost accounting, audit — and a deck can have as many as it likes. observers= is where they are declared, and they start with the deck.

from agentdeck import Deck
from agentdeck.observers import Langfuse
 
deck = Deck(agents=[booking], observers=[Langfuse()])
 
async with deck:                      # every observer starts here, once, before any run
    await deck.run("booking", "hi")   # never mid-run

Langfuse() is configured entirely by AGENTDECK_LANGFUSE_*, so there is one place to set the endpoint, environment, sample rate and service name rather than two. Naming it with no keys configured raises ConfigError at open, rather than tracing nothing quietly.

Two layers, and only one is on by default

deck = Deck(agents=[booking], observers=[Langfuse(sdk_spans=True)])

The raw layer arrives as a second, separate trace per run — it is not nested under the first. Nesting would require the engine to establish an OTel context, and the engines are barred from the Langfuse SDK by design (.importlinter’s langfuse-is-telemetry-private). So a turn with sdk_spans=True shows up in Langfuse as two traces: the agentdeck one, and the SDK’s own.

That is why it is opt-in. Reach for it when you are debugging how a turn ran — latency, retries, what a tool actually received — and leave it off when you want one clean trace per run.

Writing your own

An observer implements EventSinkPort — one required method, two optional lifecycle hooks:

from agentdeck.core.ports import EventSinkPort
 
 
class CostObserver(EventSinkPort):
    async def start(self) -> None:
        ...   # open a client or a file; called once, at deck open, before any run
 
    async def emit(self, event) -> None:
        ...   # in-memory work only; never awaits a round trip
 
    async def close(self) -> None:
        ...   # the stream has ended: write out whatever is buffered

start() and close() both default to no-ops, so an observer that only needs emit defines only emit. Raising from start() refuses the deck’s open — better than a deck that runs with an observer which silently never worked.

The lifecycle is the one the rest of the deck follows:

• build() checks that everything in observers= is an EventSinkPort and does nothing else. Nothing is started, no telemetry client is constructed, no exporter contacted — so a deck with Langfuse configured still validates in CI with nothing reachable.
• Opening calls each start() in order and registers every observer before the Runtime exists, so no run can be the thing that turns observability on.
• Closing tells every observer its stream has ended, which is what makes it write out whatever it buffered. What the deck constructs it owns; naming observers= means the deck builds no Langfuse client of its own.

An observer is fire-and-forget by contract: each has its own bounded queue, one that is slow or raises costs its own backlog and never a run, and one that keeps failing is disabled and later retried. An observer that cannot afford to lose an event reads the store instead.

One run is one trace. Traces are rendered from the canonical event stream, which is why an agent turn and a workflow run are traced by the same code and both carry their session_id — nothing instruments the OpenAI Agents SDK, and nothing opens a span outside an observer. A direct Workflow.run(), which bypasses the Runtime, therefore bypasses the observers too.

There is no deck.observers property, for the same reason there is no runtime or store: nothing needs one, and adding a property later is additive while removing one is not.

Sessions

session_for(session_id) returns the SDK conversation-memory session run/stream use for that id: Redis-backed when AGENTDECK_SESSION is set, an in-process SQLite session otherwise (lost on process exit). Pass session_factory= to Deck(...) to inject one — the seam tests use to swap in a fake wrapping fakeredis without a real Redis server.

Where a context does not reach

Four boundaries, all of them consequences of the same fact: a context is a live Python object that is never serialized, so it exists only for as long as some caller is holding it.

A context cannot cross the HTTP surface at all. There is no wire form for a live object, and asgi() invents none. An agent or workflow whose callables require a context is therefore reachable from an embedded Python caller — deck.run(...), deck.answer(...) — and not from a served deck. POST /agents/{name}/chat will run it with ctx.data set to None. If a root needs a context, do not serve it; drive it from the process that holds the object.

tick() cannot resume a context-requiring workflow. The unsupported combination is specifically durable = True + a node that pauses on sleep_until + any node in that graph declaring Context[T]. tick() takes no context= — an autonomous resume has nobody present to supply one, and the original value was never written to the log or the checkpoint, so there is nothing to recover either. What actually happens is not an error: the thread resumes and the graph replays with ctx.data set to None. That surfaces wherever the node first touches it — an AttributeError on None if it reaches for an attribute, or, for a node written defensively as if ctx.data:, a plausible wrong answer and no failure at all. This is a real limitation of v3.0.0, stated rather than worked around: there is no deck-level context provider and no way to supply one to the timer inbox. answer(run_id, value, context=...) and resume(run_id, context=...) are the resumes that do take a context, because a caller is present.

The headless runners pass no context. Agent.run() calls the SDK’s own Runner.run with no context object, so a tool that declares one fails when the model calls it — the compiled tool refuses rather than running short an argument, saying the run “carries NoneType rather than an AgentDeck run context”. Workflow.run(), Workflow.resume() and as_tool() are further out still: the bridge that injects a context into a node is installed when a Deck builds its catalog, and those paths never install it, so a Context[...] node reached that way dies with LangGraph’s own TypeError: <node>() missing 1 required positional argument. Both are the price of leaving a deliberately log-free convenience alone; Deck.run() is the path with the contract.

Skills never receive a context. A skill is progressive disclosure of prose — a SKILL.md an agent reads through the generated load_skill tool — not a program agentdeck runs. There is no callable to inject into, so there is nothing here that a future release would “add”: a skill that needs application state is an ordinary tool with a Context[...] parameter.

What else this does not cover

Run Control reaches agent runs today; a workflow (LangGraph) run has no safe point yet, so pause/cancel record the request but nothing acts on it.