Tutorial: the golden path

Available · pip install agent-adapter fastapi uvicorn · about 20 minutes

A small support-desk service wired the way we recommend for production: a confirmed refund, approvals that survive a restart, a daily budget per user, guardrails, memory across chats, cited help answers, a crash-safe flow, recorded runs, and evals that gate CI. A scripted function stands in for the model, so it all runs offline.

File	Does
examples/golden_path/tools.py	the tools: orders, a confirm=True refund, help search, memory, tickets
examples/golden_path/agent.py	SupportDesk: one executor, its stores and the flow runner; the chat agent
examples/golden_path/app.py	the FastAPI routes: chat, approvals, tickets, budget
examples/golden_path/requesters.py	who asked for each parked call (four eyes)
examples/golden_path/cases.jsonl	the eval cases CI runs
tests/docs/test_golden_path.py, test_golden_path_app.py	run all of it, over HTTP too

flowchart LR
  C["client<br/>Bearer token"] --> A["app.py<br/>FastAPI"]
  A -- "user id from the token" --> D["SupportDesk.ask"]
  D --> L["run_tool_loop<br/>scripted model"]
  L --> E["Executor"]
  E --> G["PIIGuard<br/>block cards to the model and tools,<br/>mask PII going out"]
  G --> B["BudgetHook<br/>per user per day"]
  B --> P{"confirm?"}
  P -- "refund" --> Q[("approvals.db<br/>DeferredApproval")]
  P -- "no" --> T["tools: orders, search_help,<br/>remember / recall"]
  Q -- "reviewer POST /approvals" --> T
  A -- "POST /tickets" --> F["escalation flow<br/>checkpoints.db"]
  E --> R[("runs/<br/>viewer recorder")]

1. Tools

Plain functions in one registry. The refund is confirm=True, so it never runs without a one-use approval. The order tools check that the order belongs to ctx.user_id, which the host sets and the model never chooses: another user's order looks like no order. retriever_tool returns each passage with its link; memory_tools scope every record to the same user.

def build_registry(memory: SQLiteMemory) -> Registry:
    """Every tool of the desk; the model sees `registry.subset(names=CHAT_TOOLS)`."""
    registry = Registry()

    @registry.tool
    def order_status(order_id: str, ctx: ToolContext) -> str:
        """Where an order is: packed, shipped or unknown."""
        order = own_order(order_id, ctx)
        return order[1] if order else "unknown"

    @registry.tool(confirm=True)  # runs only after a reviewer's one-use approval
    def refund_order(order_id: str, ctx: ToolContext) -> str:
        """Refund an order in full."""
        order = own_order(order_id, ctx)
        if order is None:
            raise ToolUsageError(f"no order {order_id}")
        return f"refunded {order[2]:.2f} EUR on order {order_id}"

    retriever = EmbeddingRetriever(
        CallableEmbedder(embed, provider="local", model="words-v1"), HELP
    )
    registry.add(retriever_tool(retriever, name="search_help", k=2))
    registry.add_many(memory_tools(memory))  # remember, recall, forget: scoped to ctx.user_id

    @registry.tool(idempotent=True)  # the same order and summary give the same ticket
    def open_ticket(order_id: str, summary: str, ctx: ToolContext) -> str:
        """Open a support ticket for a person to follow up."""
        if own_order(order_id, ctx) is None:
            raise ToolUsageError(f"no order {order_id}")
        digest = hashlib.sha256(f"{order_id}:{summary}".encode()).hexdigest()[:8]
        return f"T-{digest}"

    @registry.tool
    def notify_team(ticket_id: str) -> str:
        """Page the support team about a ticket (must not run twice)."""
        return page_team(ticket_id)

    return registry


def own_order(order_id: str, ctx: ToolContext) -> tuple[str, str, float] | None:
    """The order if the signed-in user owns it; another user's order looks like no order."""
    order = ORDERS.get(order_id)
    return order if order is not None and ctx.user_id and order[0] == ctx.user_id else None


def page_team(ticket_id: str) -> str:
    """Stands in for your pager or chat-ops client."""
    return f"team notified about {ticket_id}"

2. One desk per process

Build everything once per process, in a factory, never at import: the executor holds the guards, the budget (SQLite, so every worker on the host shares the spend), the approval store, the ledger and the recorder. The model only sees chat_tools; the ticket tools are the flow's.

class SupportDesk:
    """Everything one process needs, built once: tools, stores, the executor and the flow runner."""

    def __init__(self, folder: Path, *, runs: Path, daily_usd: float = 0.50) -> None:
        folder.mkdir(parents=True, exist_ok=True)
        self.memory = SQLiteMemory(folder / "memory.db", retention_days=90)
        self.registry = build_registry(self.memory)
        self.chat_tools = self.registry.subset(names=CHAT_TOOLS)
        self.budget = BudgetHook(
            daily_usd, scope="user", per="day", store=SQLiteBudgetStore(folder / "budgets.db")
        )
        self.executor = Executor(
            self.registry,
            hooks=[
                # Block cards and secrets in what the user sends the model and in tool arguments.
                PIIGuard(kinds=["secret", "card"], on=("llm_request", "args")),
                PIIGuard(on=("result", "llm_response"), action="redact"),  # mask on the way out
                self.budget,  # per user per UTC day, shared by every worker on this file
            ],
            approval=DeferredApproval.sqlite(folder / "approvals.db"),  # survives a restart
            ledger=Ledger(prices=PRICES),
        )
        self.requesters = Requesters(folder / "requesters.db")
        self.recorder = record(self.executor, folder=runs)  # agent-adapter ui
        store = SQLiteCheckpointStore(folder / "checkpoints.db")
        self.runner = FlowRunner(
            self.executor, checkpoints=store, checkpoint_key=checkpoint_key(folder)
        )
        self.model = CallableModel(scripted_model, provider="local", model="desk-v1")

    def context(self, user: str, chat_id: str, run_id: str | None = None) -> ToolContext:
        """A call context; the host sets the user, never the model."""
        return self.executor.context(run_id=run_id, thread_id=chat_id, user_id=user)

    async def ask(self, user: str, chat_id: str, text: str, run_id: str | None = None) -> Reply:
        """One turn of the agent loop for `user` in chat `chat_id`.

        Raises:
            BudgetExceeded: the user's daily budget is spent.
        """
        ctx = self.context(user, chat_id, run_id)
        messages = [Message(role="system", content=SYSTEM), Message(role="user", content=text)]
        result = await run_tool_loop(
            self.model, self.chat_tools, messages, executor=self.executor, ctx=ctx
        )
        if result.finish == "pending":
            waiting = next(r for r in result.tool_results if r.status == "pending")
            approval_id = str(waiting.error.details["approval_id"]) if waiting.error else ""
            text = f"A reviewer must approve {waiting.name} first."
            await self.requesters.add(approval_id, user, ctx.run_id)
            return Reply(text=text, run_id=ctx.run_id, approval_id=approval_id)
        return Reply(text=result.final.text or "", run_id=ctx.run_id)

    async def decide(self, approval_id: str, approved: bool, reviewer: str) -> str:
        """Apply a reviewer's decision; an approved call runs once, here, as its requester.

        Raises:
            ApprovalNotFound: unknown id.
            Forbidden: the reviewer asked for the call (four eyes).
            ApprovalAlreadyDecided: the decision was used already.
        """
        asked = await self.requesters.get(approval_id)
        if asked is None:
            raise ApprovalNotFound(approval_id)
        requester, run_id = asked
        if reviewer == requester:
            raise Forbidden("a reviewer cannot decide their own request")
        ctx = self.executor.context(run_id=run_id, user_id=requester)  # tools check ownership
        try:
            result = await self.executor.resume(approval_id, approved, ctx, decided_by=reviewer)
        finally:
            self.executor.end_run(run_id)
        why = result.error.kind if result.error else result.status
        return str(result.value) if result.ok else f"not run ({why})"

    async def escalate(self, user: str, order_id: str, summary: str) -> Ticket:
        """Run the escalation flow, checkpointed after every step."""
        ctx = self.executor.context(user_id=user)
        result = await self.runner.run(ESCALATION, {"order_id": order_id, "summary": summary}, ctx)
        return result.state

    async def recover(self, *, rerun_in_doubt: bool = False) -> dict[str, str]:
        """Finish escalations a crashed worker left behind (call on startup)."""
        found = await self.runner.recover_all(ESCALATION, rerun_in_doubt=rerun_in_doubt)
        outcome: dict[str, str] = {}
        for run_id, result in found.items():
            if isinstance(result, AdapterError):
                logger.warning("escalation %s not recovered: %s", run_id, result.kind)
                outcome[run_id] = result.kind
            else:
                outcome[run_id] = result.status
        return outcome

    def close(self) -> None:
        """Flush and detach the run recorder."""
        self.recorder.stop()

The blocking PIIGuard checks what the user sends the model (llm_request) and every tool's arguments: a card number in the chat ends the turn with guardrail-blocked (422) before any model sees it. A request cannot be rewritten, only blocked; drop llm_request if your model may see such data. The redacting guard masks emails and cards in tool results and model replies.

3. A model you can swap

scripted_model picks one tool and answers from its result. Replace CallableModel(...) with OpenAIModel, AnthropicModel or your own port; nothing else changes (Bring your own model).

def scripted_model(request: LLMRequest) -> LLMResponse:
    """Stands in for a real model: picks one tool for the question, then answers from its result."""
    last = request.messages[-1]
    if last.role == "tool":
        return _reply(Message(role="assistant", content=_answer(last.name, last.content)))
    call = ToolCall(id="call_1", **_pick_tool(last.content))
    return _reply(Message(role="assistant", tool_calls=(call,)))

4. Approvals that survive a restart

Please refund order 1043 parks the call in approvals.db and the reply carries its id. A reviewer decides with POST /approvals/{id}, on any worker that opens the same files; the call runs there, once, as the user who asked, so the ownership check holds. A second decision gets 409, a customer token 403, and a reviewer deciding their own request 403 (forbidden): requesters.db keeps who asked for each parked call.

5. A crash-safe flow

POST /tickets opens a ticket and pages the team. With a checkpoint store, every step is signed and every tool call is logged, so a worker that dies mid-flow is finished by the next one: the app's startup calls desk.recover().

class Ticket(BaseModel):
    order_id: str
    summary: str
    ticket_id: str = ""
    note: str = ""


def ticket_args(ticket: Ticket) -> dict[str, str]:
    """The arguments of `open_ticket`."""
    return {"order_id": ticket.order_id, "summary": ticket.summary}


ESCALATION = (
    Flow("escalation", state=Ticket)
    .node("open", Call("open_ticket", args=ticket_args, into="ticket_id"))
    .node("notify", Call("notify_team", args=lambda t: {"ticket_id": t.ticket_id}, into="note"))
    .edge("open", "notify")
    .end("notify")
)

open_ticket is idempotent=True, so it replays. notify_team is not: if the crash hit while it ran, recovery stops with recovery-needs-decision until a person checks the page went out and recovers with rerun_in_doubt=True (Crash-safe runs). The key that signs checkpoints comes from SUPPORT_DESK_KEY; set it from your vault and use the same one on every worker.

6. The HTTP routes

def create_app(desk: SupportDesk | None = None) -> FastAPI:
    """The HTTP routes; startup builds the default desk (unless given), recovers crashed flows."""

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        support = desk if desk is not None else default_desk()
        app.state.desk = support
        await support.recover()  # in-doubt runs are logged and wait for a person
        yield
        support.close()

    app = FastAPI(title="Support desk", lifespan=lifespan)

    @app.exception_handler(AdapterError)
    async def adapter_error(request: Request, error: AdapterError) -> JSONResponse:
        detail = {"kind": error.kind, "message": str(error)}
        return JSONResponse({"detail": detail}, status_code=STATUS.get(error.kind, 400))

    @app.post("/chat")
    async def chat(body: Ask, who: Caller, support: Desk) -> dict[str, str]:
        """One agent turn; the chat id is the caller's own (prefixed with the user)."""
        user = who[0]
        reply = await support.ask(user, f"{user}:{body.chat_id}", body.text)
        return {"text": reply.text, "run_id": reply.run_id, "approval_id": reply.approval_id}

    @app.post("/approvals/{approval_id}")
    async def decide(
        approval_id: str, body: Decision, who: Caller, support: Desk
    ) -> dict[str, str]:
        """A reviewer approves or denies a parked call; it runs at most once."""
        user, role = who
        if role != "reviewer":
            raise HTTPException(403, "only reviewers decide approvals")
        return {"result": await support.decide(approval_id, body.approved, user)}

    @app.post("/tickets")
    async def escalate(body: Escalation, who: Caller, support: Desk) -> dict[str, Any]:
        """Open a ticket and page the team, as a crash-safe flow."""
        ticket = await support.escalate(who[0], body.order_id, body.summary)
        return ticket.model_dump()

    @app.get("/budget")
    async def budget(who: Caller, support: Desk) -> dict[str, Any]:
        """What the caller has spent today, and what is left."""
        status = await support.budget.status(support.context(who[0], ""))
        return {"spent_usd": status.spent_usd, "remaining_usd": status.remaining_usd}

    return app

Route	Who	Does
POST /chat	any user	one agent turn; the chat id is prefixed with the user
POST /approvals/{id}	reviewers	approve or deny a parked call
POST /tickets	any user	the escalation flow
GET /budget	any user	today's spend and what is left

Adapter errors become JSON problems with their kind: budget-exceeded is 429, guardrail-blocked and flow-error 422. The demo USERS tokens stand in for your IdP.

cd examples
uvicorn golden_path.app:app                    # data in .support_desk/
curl -X POST localhost:8000/chat -H "Authorization: Bearer ann-token" \
  -H "Content-Type: application/json" -d '{"chat_id": "c1", "text": "Where is order 1042?"}'
agent-adapter ui                               # recorded turns, flows, approvals, cost

7. Evals as the CI gate

create_app() builds the live desk at startup with default_desk() (data in SUPPORT_DESK_DATA, default .support_desk); chat is the same agent for the chat page, where an anonymous chat is its own user. An eval must never park a live refund, spend a user's budget or read their memory, so it runs on eval_desk(): fresh stores in a temp folder, deleted at exit, with the same guards, budget and approval. Its runs go to .agent_adapter/runs, the eval's default, so it keeps one recorder.

EVAL_USER = "ann"  # the demo customer whose orders the eval cases ask about


@functools.cache
def eval_desk() -> SupportDesk:
    """A desk on fresh temporary stores: an eval never parks a live refund or spends a budget."""
    folder = Path(tempfile.mkdtemp(prefix="support-desk-eval-"))
    desk = build_desk(folder)  # runs still go to the viewer's folder, to inspect failures
    atexit.register(_discard, desk, folder)
    return desk


async def eval_chat(message: ChatMessage, session: ChatSession) -> str:
    """The eval agent: the chat agent's turn, on the eval desk, as the demo customer."""
    reply = await eval_desk().ask(EVAL_USER, session.id, message.text, session.run_id)
    return reply.text


def __getattr__(name: str) -> Executor:
    """`eval_executor`, built on first access (for `agent-adapter eval --executor`)."""
    if name == "eval_executor":
        return eval_desk().executor
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

agent-adapter eval golden_path.agent:eval_chat golden_path/cases.jsonl \
  --executor golden_path.agent:eval_executor --grader contains --grader max-cost=0.01 \
  --junit evals.xml

It exits 1 when a case fails, so a CI step fails with it. test_golden_path_app.py runs exactly this command, and a wrong answer, to show the gate bites (Evaluate your agent).

Next

• Threats this setup covers, and what is still yours: Threat model.
• Before you ship: the Production checklist.
• What the executor costs per call: Performance.