FastAPI¶

Available · pip install agent-adapter fastapi · add 'agent-adapter[server]' to mount the HTTP tool server

FastAPI runs on an event loop, so routes await the executor directly. No run_sync.

File	Does
`examples/backend_common.py`	the service layer every backend guide shares: tools, `build_backend`, `invoke`, `decide`, `resume`, `ask`
`examples/backend_fastapi.py`	the FastAPI app: three routes, auth dependency, the tool server under `/agent`
`tests/docs/test_backend_examples.py`	runs both offline with FastAPI's `TestClient`

Data enters as a JSON body, goes through invoke / decide / ask to the executor, and leaves as a JSON reply with an HTTP status (an RFC 9457 problem on errors).

One Executor per process¶

@dataclass(frozen=True)
class Backend:
    """What one process shares across requests: the executor, its approval store, the ledger."""

    executor: Executor
    store: ApprovalStore
    ledger: Ledger


def build_backend(*, store: ApprovalStore | None = None, max_usd: float = 0.01) -> Backend:
    """Build once per process; pass a database `ApprovalStore` when workers share approvals."""
    store = store if store is not None else MemoryApprovalStore()
    ledger = Ledger(prices=PRICES)
    executor = Executor(
        registry,
        approval=DeferredApproval(store),  # confirm tools park in the store
        ledger=ledger,
        hooks=[LoggingHook(), BudgetHook(max_usd)],  # the budget is per run: one per request
        limits=RunLimits(max_calls=20),
    )
    return Backend(executor, store, ledger)

Build it once, in make_app(), not per request. Run guards, the ledger, the approval store and the budget hook live on it. Each uvicorn or gunicorn worker is its own process, so each builds its own; anything workers must share goes in a shared store.

def make_app(backend: Backend | None = None) -> FastAPI:
    """The app; `backend` defaults to one built here, once per worker process."""
    service = backend if backend is not None else build_backend()

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        yield
        await LoopLocal.aclose_all()  # close loop-bound clients before the server's loop ends

    app = FastAPI(lifespan=lifespan)

    @app.post("/tools/{name}")
    async def call_tool(name: str, body: InvokeBody, user: User) -> JSONResponse:
        status, reply = await invoke(service, name, body.arguments)  # await: same loop
        return JSONResponse(reply, status_code=status)

    @app.post("/approvals/{approval_id}/decide")
    async def decide_call(approval_id: str, body: DecideBody, user: User) -> JSONResponse:
        status, reply = await decide(
            service,
            approval_id,
            reviewer=user,
            approved=body.approved,
            edited_args=body.edited_arguments,
        )
        return JSONResponse(reply, status_code=status)

    @app.post("/ask")
    async def ask_model(body: AskBody, user: User) -> JSONResponse:
        status, reply = await ask(service, body.question)
        return JSONResponse(reply, status_code=status)

    tools = create_app(registry, mcp=False, **tool_server_options(service))
    app.mount("/agent", tools)  # REST tool server: /agent/tools, /agent/approvals, ...
    return app

The service layer¶

Views stay thin. All tool work is in backend_common.py:

async def call(
    backend: Backend, tool: str, arguments: dict[str, Any], *, run_id: str, call_id: str
) -> Any:
    """Run one tool call as the run `run_id`; errors raise (`ApprovalPending` included)."""
    async with backend.executor.run_scope(run_id=run_id) as ctx:
        return await backend.executor.run(tool, arguments, ctx, call_id=call_id)


async def invoke(
    backend: Backend,
    tool: str,
    arguments: dict[str, Any],
    *,
    run_id: str | None = None,
    call_id: str | None = None,
) -> Reply:
    """200 with the result, 202 when the call waits for a reviewer, else a problem."""
    run_id, call_id = run_id or new_id("run_"), call_id or new_id("call_")
    try:
        value = await call(backend, tool, arguments, run_id=run_id, call_id=call_id)
    except ApprovalPending as pending:
        return 202, {"status": "pending_approval", "approval_id": pending.approval_id}
    except AdapterError as exc:
        return error_reply(exc)
    return 200, {"status": "ok", "run_id": run_id, "result": value}


async def record_decision(
    backend: Backend,
    approval_id: str,
    *,
    reviewer: str,
    approved: bool,
    edited_args: dict[str, Any] | None = None,
) -> Reply:
    """Check the reviewer, validate edits, store the decision; nothing runs yet."""
    try:
        if reviewer not in REVIEWERS:
            raise Forbidden("only reviewers decide approvals")
        record = await backend.store.get(approval_id)  # ApprovalNotFound -> 404
        if edited_args is not None:  # validate edits before the decision is stored
            registry.get(record.request.tool).validate_args(edited_args)
        decision = ApprovalDecision(approved=approved, edited_args=edited_args, decided_by=reviewer)
        await backend.store.decide(approval_id, decision)  # a second decision -> 409
    except AdapterError as exc:
        return error_reply(exc)
    return 200, {"status": "approved" if approved else "denied", "approval_id": approval_id}


async def decide(
    backend: Backend,
    approval_id: str,
    *,
    reviewer: str,
    approved: bool,
    edited_args: dict[str, Any] | None = None,
) -> Reply:
    """Record a reviewer's decision; an approved call then runs once, in this request."""
    reply = await record_decision(
        backend, approval_id, reviewer=reviewer, approved=approved, edited_args=edited_args
    )
    if reply[0] != 200 or not approved:
        return reply
    return await resume(backend, approval_id)


async def resume(backend: Backend, approval_id: str) -> Reply:
    """Re-run a decided call with its own run and call ids: the store answers it, once."""
    try:
        record = await backend.store.get(approval_id)
        if record.consumed:
            raise ApprovalAlreadyDecided(approval_id, status=record.status)
    except AdapterError as exc:
        return error_reply(exc)
    request = record.request  # the original arguments: the decision is bound to them
    return await invoke(
        backend, request.tool, request.arguments, run_id=request.run_id, call_id=request.call_id
    )

Function	Returns
`invoke(backend, tool, arguments)`	200 `{"status": "ok", "result"}`, 202 `{"status": "pending_approval", "approval_id"}`, or a problem
`record_decision(...)`	checks the reviewer, validates edits, stores the decision; runs nothing
`decide(...)`	`record_decision`, then `resume` when approved
`resume(backend, approval_id)`	re-runs the call with its original run id, call id and arguments; 409 when already used
`ask(backend, question)`	the tool loop under a per-run budget, with the run's cost

Errors map to statuses with problem_for, the same table the HTTP tool server uses: tool-args-invalid 422, forbidden 403, approval-already-decided 409, budget-exceeded 429, tool-timeout 504.

Approvals in a request/response cycle¶

A confirm=True tool never runs inside the request that asked for it. The executor's DeferredApproval parks it in the store and raises ApprovalPending; the route answers 202. A reviewer's later request decides it, and the call runs then, once.

sequenceDiagram
  participant U as Client
  participant A as FastAPI route
  participant S as service layer
  participant E as Executor
  participant St as ApprovalStore
  U->>A: POST /tools/send_invoice
  A->>S: await invoke(...)
  S->>E: run(tool, args, call_id)
  E->>St: create pending record
  E-->>S: ApprovalPending(approval_id)
  A-->>U: 202 pending_approval
  participant R as Reviewer
  R->>A: POST /approvals/{id}/decide
  A->>S: await decide(...)
  S->>St: decide (reviewer checked, edits validated)
  S->>E: run again: same run id, call id, arguments
  E->>St: consume the decision (once)
  E-->>S: result
  A-->>R: 200 result

Rule	Why
resume with the record's `run_id`, `call_id` and original `arguments`	the decision is bound to them; edits come from the decision
only a reviewer may decide (`REVIEWERS`; use your own roles)	the invoker should not approve itself; see the four-eyes rule in HTTP server
never set `ToolContext(approval=...)` from a request body or a header	it skips the port; only adapter-issued one-use tokens may set it
never call `spec.invoke(...)` for a tool with `confirm`	it bypasses the executor, so no approval, budget or ledger

Budget per request¶

BudgetHook(max_usd) checks the ledger total of the current run before each tool and model call. Each request is its own run (run_scope()), so the cap is per request. Over it, ask answers 429.

async def ask(backend: Backend, question: str) -> Reply:
    """Answer with the tool loop; the run's budget caps it and the reply carries its cost."""
    async with backend.executor.run_scope() as ctx:
        messages = (Message(role="user", content=question),)
        try:
            result = await run_tool_loop(
                MODEL, registry, messages, executor=backend.executor, ctx=ctx
            )
        except AdapterError as exc:  # BudgetExceeded -> 429
            return error_reply(exc)
        cost = backend.ledger.total_usd(ctx.run_id)
    return 200, {"status": result.finish, "answer": result.final.text, "cost_usd": round(cost, 6)}

RunLimits(max_calls=20) also caps tool calls per run. For a cap across one user's requests, pass the user as thread_id (run_scope(thread_id=...)) and use BudgetHook(max_usd, scope="thread"); the ledger keeps its latest max_runs runs.

Mount the HTTP tool server¶

tools = create_app(registry, mcp=False, **tool_server_options(service))
app.mount("/agent", tools)  # REST tool server: /agent/tools, /agent/approvals, ...

create_app gives you the REST routes (/agent/tools, /agent/approvals, events, cost) with authentication, run ownership and approval rules built in. It shares the executor (hooks, budget, ledger) but keeps its own approval store, decided only through its own /agent/approvals/{id}/decide.

Logging and tracing¶

Need	Use
a log line per tool call	`LoggingHook()` (arguments redacted; `confirm` tools' arguments never logged)
events as logs	`Executor(..., sink=LoggingSink())`
OpenTelemetry spans	`Executor(..., sink=OTelSink())` (`agent-adapter[otel]`)
correlated logs across workers of the tool server	`set_log_key(secret)` in every worker

Pitfalls¶

Pitfall	Fix
`run_sync(...)` inside an `async def` route	`await` instead; `run_sync` would block the loop with a worker thread
a sync `def` route	FastAPI runs it in a thread: there `run_sync` is right
a loop-bound client (httpx, Redis) used from async routes and from `run_sync` code or tests	keep it in a `LoopLocal`; the lifespan closes them with `LoopLocal.aclose_all()`
`MemoryApprovalStore` with several workers	a decision lands in one worker only: use `SQLiteApprovalStore` (one host), or `RedisApprovalStore`, `SQLAlchemyApprovalStore` or `DynamoDBApprovalStore` (many hosts); see Approval stores, Database stores
`Idempotency-Key` on the mounted server with several workers	`idempotency=RedisIdempotencyStore(...)`
`TestClient(app)` gets 421 on `/agent/...`	by default the tool server allows local `Host` names only; set `allowed_hosts=[...]` for a public host, or test with `base_url="http://localhost"`
`Location` from the mounted server lacks `/agent`	it is relative to the mount; prefix it in the client
`mcp=True` on a mounted app	Starlette does not run a mounted app's lifespan: keep `mcp=False`, or enter `tools.router.lifespan_context(tools)` in your lifespan

See also: Flask, Django, Celery, scripts, Runtime, Approval.