Recipes

Task-shaped answers to "how do I actually wire this into my application". Every Chassis API here is the real one; anything belonging to another library is marked as such.

• Behind a web service
• Per-tenant composition and identity
• Swap a provider while runs are in flight
• Budget per request, per tenant, per child run
• Durable runs: checkpointing and resume

Behind a web service

One harness per process, not per request: composition is expensive, and async with harness is what makes it start and shut down safely.

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException

from chassis import MODEL, Harness

harness = Harness("harness.yaml")                     # declarative composition
harness.register_plugin_type("openai-model", OpenAIModelPlugin)


@asynccontextmanager
async def lifespan(app: FastAPI):
    async with harness:                               # reconcile on start, drain on stop
        yield


app = FastAPI(lifespan=lifespan)


@app.post("/chat")
async def chat(body: ChatRequest) -> dict[str, str]:
    try:
        result = await harness.agents.invoke(
            "research-agent",
            {"messages": body.messages},
            thread_id=body.thread_id,
            user_id=body.user_id,
            tenant_id=body.tenant_id,
            metadata={"request_id": body.request_id},
        )
    except ToolExecutionError as error:
        raise HTTPException(status_code=502, detail=str(error)) from error

    return {"text": result.text or "", "generation": result.generation_id}

Points that matter:

• enter the harness before invoking it: async with harness (or an explicit await harness.start()) is what creates the first generation. Once running, pending desired-state changes are applied automatically before a run, so install/provide can be called synchronously at any time and the next invocation picks them up;
• shutdown is not optional - harness.stop() waits for in-flight runs (shutdown_grace_seconds), then disposes what nothing can reach any more;
• the run's thread_id maps to LangGraph checkpointing; user_id/tenant_id travel in the immutable run context, so a graph node can read them from runtime.context;
• result.generation_id is the composition the answer came from - log it, it is what makes an incident reproducible.

Expose health from the same authoritative state:

@app.get("/healthz")
async def healthz() -> dict:
    return harness.diagnostics.status()

Per-tenant composition and identity

Identity is per run; composition is per harness. Three workable shapes, from cheapest to most expressive:

# 1. Per-request identity: same composition, different tenant.
await harness.agents.invoke("support-agent", {"messages": [...]}, tenant_id=tenant.id)

# 2. Per-tenant wiring through the flat configuration: one entry per tenant.
config = {
    "version": 1,
    "plugins": [
        {"id": f"db-{tenant.id}", "plugin": "postgres-pool", "config": {"dsn_ref": tenant.dsn_ref}},
        {"id": f"agent-{tenant.id}", "plugin": "support-agent", "requires": {"database": ">=1,<2"}},
    ],
    "provider_preferences": {"database": f"db-{tenant.id}"},
}

# 3. A composition scope per tenant: shared providers inherited from the root,
#    tenant-local providers, and isolation between tenants for free.
model = harness.install(OpenAIModelPlugin(), entry_id="model")     # shared, root scope
tenant_scope = harness.composition.child(f"tenant:{tenant.id}")
tenant_scope.install(PostgresPoolPlugin(dsn_ref=tenant.dsn_ref), entry_id="db")
tenant_scope.install(SupportAgentPlugin(), entry_id="support-agent")
await harness.reconcile()

Shape 3 is what 0.3 is for. Each tenant sees the shared model and its own database, never another tenant's; a plugin that needs the database resolves it without any per-consumer preference, because siblings are not candidates (scopes.md). Add always-on shared infrastructure at the root and narrow what a tenant may observe when it should not see everything:

tenant_scope.restrict(MODEL, DATABASE, TOOLS)

Entries carry stable ids, so a provider change is a REPLACE that publishes a new generation, while runs already in flight keep the one they acquired (configuration.md).

When one capability genuinely has several visible providers, decide explicitly instead of relying on resolution order:

harness.prefer_provider("database", "postgres")                     # globally
harness.prefer_provider("database", "postgres", consumer="agent")   # per consumer entry
harness.prefer_provider("database", "postgres", scope="/tenant:acme")  # per scope

Ambiguity is never resolved by shadowing: a valid local provider and a valid inherited provider both stay candidates until a preference selects one.

Swap a provider while runs are in flight

The whole point of generations: reconfiguration never mutates a running composition.

harness.install(OpenAIModelPlugin(model="gpt-x"), entry_id="model", replace=True)
await harness.reconcile()          # publishes generation N+1

# Runs that already acquired generation N keep its model object, its policy, and
# its secret provider until they finish; nothing they hold is disposed meanwhile.

examples/safe_provider_replacement.py runs exactly this and asserts it, including that the old provider is disposed after the last run releases it. Two rules of thumb:

• swap by replacing the entry (replace=True) rather than by mutating an object you already handed to the harness;
• if a capability is runtime-bound (model, database, policy, secrets, tenant), a swap reuses the compiled graph - it does not recompile it (langgraph.md).

Budget per request, per tenant, per child run

Three levels, from coarsest to finest:

# 1. harness default for every run
harness = Harness(default_budget_limits=BudgetLimits(wall_clock_seconds=60, tool_calls=20))

# 2. this run only
await harness.agents.invoke(
    "research-agent",
    {"messages": [...]},
    limits=BudgetLimits(tool_calls=5, child_runs=2),
)

# 3. a nested run started from inside a run inherits what is left
#    (a graph node calling harness.agents.invoke(...) is a child run)

Enforced by Chassis: wall clock, tool calls, child runs. Accounted: model_calls, tokens, and estimated_cost, because graphs call models and not the harness — they hold only when the code that owns the call reports usage:

run_context.budget.record(model_calls=1, tokens=usage.total_tokens, estimated_cost=cost)

BudgetDimension.TOKENS.enforcement and harness.diagnostics.budgets() tell you which is which at runtime, so a token limit is never mistaken for a guarantee.

A BudgetExceeded carries dimension, limit, used, and requested, so a service can map it to a 429 rather than a 500. Children can never exceed what the parent has left, so tenant-wide ceilings hold even through nested agents.

Durable runs: checkpointing and resume

LangGraph owns durability; Chassis only carries what a run needs to be attributed. Pass the checkpointer you want (any LangGraph checkpointer; install langgraph-checkpoint-postgres for AsyncPostgresSaver):

harness.register_agent(
    LangGraphAgent(definition, checkpointer=AsyncPostgresSaver(pool), store=store)
)

Then a run and its resume are two ordinary calls:

first = await harness.agents.invoke("research-agent", {"messages": [...]}, thread_id="t-1")

if first.interrupted:
    approved = await harness.agents.invoke(
        "research-agent", {"messages": []}, thread_id="t-1", resume=first.resume_values()
    )

examples/basic_agent.py does this end to end with InMemorySaver, including streaming and the interrupt payload a UI would render.

Two caveats worth knowing before you scale it: a thread_id is a concurrency key - two concurrent runs on the same thread are a LangGraph problem, not something Chassis serializes - and an abandoned agents.stream(...) generator holds its generation lease until it is closed.