Skip to content

FastAPI

This guide adds Bunkhouse to a FastAPI application that renders its pages with Jinja2 templates. Every file shown below is part of a runnable example app that is tested end to end against the real Bunkhouse API and embed.

Framework FastAPI 0.162.0 (Starlette 1.10.0), Jinja2 3.1 templates, Starlette SessionMiddleware, served by Uvicorn 0.54.0
Runtime Python 3.14 (3.14.10 in the tested image)
Example apps/docs/integrations/fastapi/ in the Bunkhouse repository

You need:

  • Your application’s exact origin registered on the site (for example https://app.example.com).
  • The site’s public key (pk_...).
  • For identity tokens (recommended): a signing secret (ss_...) and its key ID from the site’s Keys page.
  • For server mint instead: a mint key (sk_...).
  • Existing sign-in. The example uses Starlette’s signed-cookie SessionMiddleware with two demo users; your own authentication takes its place. Bunkhouse only needs one thing from it: the signed-in user’s stable ID on the server.
File Purpose
app/config.py Validates the public Bunkhouse configuration at startup
app/dependencies.py current_user, the dependency that connects your authentication
app/agent.py APIRouter with GET /api/agent-identity and POST /api/agent-session
app/csp.py Your Content-Security-Policy with the Bunkhouse sources added
templates/agent-embed-*.html The script tags for one installation mode
static/agent-identity.js, static/agent-auth-headers.js Page hooks, loaded before the loader

Set the server environment. Only AGENT_PUBLIC_KEY and the origins ever reach the browser:

Terminal window
AGENT_PUBLIC_KEY=pk_...
AGENT_SIGNING_SECRET=ss_... # server only
AGENT_SIGNING_KID=... # key ID shown with the signing secret
AGENT_ALLOWED_ORIGINS='["https://app.example.com"]'
# AGENT_API_BASE and AGENT_EMBED_ORIGIN default to https://api.bunkhouse.ai and https://embed.bunkhouse.ai
app/config.py
# Bunkhouse public configuration, validated once at startup. Signing and mint credentials are read
# only by the endpoints that use them, so each mode needs only its own secret.
import json
import os
import re
from dataclasses import dataclass
from urllib.parse import urlsplit
DEFAULT_PORTS = {"http": 80, "https": 443}
def exact_origin(name: str, value: str) -> str:
"""value must be its own origin, as `new URL(value).origin === value` checks in JavaScript."""
try:
url = urlsplit(value)
port = url.port # ValueError for a non-numeric or out-of-range port
except ValueError:
url = None
if url is not None and url.scheme in DEFAULT_PORTS and url.hostname:
host = f"[{url.hostname}]" if ":" in url.hostname else url.hostname
suffix = f":{port}" if port is not None and port != DEFAULT_PORTS[url.scheme] else ""
# Lowercase host, no userinfo, path, trailing ':' or explicit default port.
if re.fullmatch(r"[a-z0-9.-]+|\[[0-9a-f:.]+\]", host) and value == f"{url.scheme}://{host}{suffix}":
return value
raise RuntimeError(f"{name} must be an exact origin like https://app.example.com")
@dataclass(frozen=True)
class AgentConfig:
public_key: str
allowed_origins: tuple[str, ...]
api_base: str
embed_origin: str
def load() -> AgentConfig:
public_key = os.environ.get("AGENT_PUBLIC_KEY", "")
if not public_key.startswith("pk_"):
raise RuntimeError("AGENT_PUBLIC_KEY must be the site's public key (pk_...)")
origins = json.loads(os.environ.get("AGENT_ALLOWED_ORIGINS", "[]"))
if not isinstance(origins, list) or not origins:
raise RuntimeError('AGENT_ALLOWED_ORIGINS must be a JSON array such as ["https://app.example.com"]')
return AgentConfig(
public_key=public_key,
allowed_origins=tuple(exact_origin("AGENT_ALLOWED_ORIGINS", str(o)) for o in origins),
api_base=exact_origin("AGENT_API_BASE", os.environ.get("AGENT_API_BASE", "https://api.bunkhouse.ai")),
embed_origin=exact_origin("AGENT_EMBED_ORIGIN", os.environ.get("AGENT_EMBED_ORIGIN", "https://embed.bunkhouse.ai")),
)
agent_config = load()

The router signs a ten-minute HS256 token with the standard library (hmac, hashlib) for the user your session already authenticated. It rejects other methods (405 with Allow), requests whose Sec-Fetch-Site is not same-origin (403) and signed-out users (401, never a login redirect), and marks every response Cache-Control: no-store. The subject comes only from the server session, never from the query, body or headers. The same file contains the server mint endpoint used by the alternative below.

app/agent.py
# Bunkhouse endpoints: GET /api/agent-identity (identity token mode) and POST /api/agent-session
# (server mint mode). Both vouch only for the user your existing session already authenticated.
import base64
import hashlib
import hmac
import json
import logging
import os
import time
from http.client import HTTPException as UpstreamProtocolError
from typing import Annotated, Any
from urllib.error import HTTPError
from urllib.request import Request as UpstreamRequest, urlopen
from fastapi import APIRouter, Depends, Header
from fastapi.responses import JSONResponse, Response
from .config import agent_config
from .dependencies import current_user
router = APIRouter()
log = logging.getLogger("bunkhouse")
User = Annotated[dict[str, str] | None, Depends(current_user)]
METHODS = ("GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS")
def b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def b64json(value: dict[str, Any]) -> str:
return b64url(json.dumps(value, separators=(",", ":")).encode())
def sign_identity_token(user_id: str, secret: str, kid: str) -> str:
"""HS256 identity token: kid header, sub = stable opaque user ID, aud = public key, 10 minutes."""
iat = int(time.time())
signing_input = f"{b64json({'alg': 'HS256', 'typ': 'JWT', 'kid': kid})}.{b64json({'sub': user_id, 'aud': agent_config.public_key, 'iat': iat, 'exp': iat + 600})}"
signature = hmac.new(secret.encode(), signing_input.encode(), hashlib.sha256).digest()
return f"{signing_input}.{b64url(signature)}"
def no_store(status: int, body: dict[str, Any], headers: dict[str, str] | None = None) -> JSONResponse:
return JSONResponse(body, status_code=status, headers={"Cache-Control": "no-store", **(headers or {})})
def not_configured(variable: str) -> JSONResponse:
log.error("Bunkhouse: %s is not set", variable)
return no_store(503, {"error": "not_configured"})
@router.get("/api/agent-identity")
def agent_identity(user: User, sec_fetch_site: Annotated[str | None, Header()] = None) -> Response:
if sec_fetch_site != "same-origin":
return no_store(403, {"error": "forbidden"})
if user is None:
return no_store(401, {"error": "unauthorized"})
secret, kid = os.environ.get("AGENT_SIGNING_SECRET"), os.environ.get("AGENT_SIGNING_KID")
if not secret:
return not_configured("AGENT_SIGNING_SECRET")
if not kid:
return not_configured("AGENT_SIGNING_KID")
return no_store(200, {"identityToken": sign_identity_token(str(user["id"]), secret, kid)})
# A sync endpoint: FastAPI runs it in its threadpool, so the blocking upstream call is fine.
@router.post("/api/agent-session")
def agent_session(
user: User,
origin: Annotated[str | None, Header()] = None,
x_agent_embed: Annotated[str | None, Header()] = None,
sec_fetch_site: Annotated[str | None, Header()] = None,
) -> Response:
# The embed's POST carries no CSRF token; these three checks are this route's CSRF defense.
if origin not in agent_config.allowed_origins or x_agent_embed != "1" or sec_fetch_site != "same-origin":
return no_store(403, {"error": "forbidden"})
if user is None:
return no_store(401, {"error": "unauthorized"})
key = os.environ.get("AGENT_EMBED_SECRET_KEY")
if not key:
return not_configured("AGENT_EMBED_SECRET_KEY")
upstream = UpstreamRequest(
f"{agent_config.api_base}/v1/embed/sessions",
method="POST",
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
data=json.dumps({"publicKey": agent_config.public_key, "mode": "signed", "userId": str(user["id"]), "origin": origin}).encode(),
)
try:
try:
res = urlopen(upstream, timeout=15)
except HTTPError as error: # a 4xx/5xx answer is forwarded too
res = error
with res:
status, body, content_type = res.status, res.read(), res.headers.get("Content-Type", "application/json")
except (OSError, UpstreamProtocolError):
return no_store(502, {"error": "session_service_unavailable"})
return Response(body, status_code=status, media_type=content_type, headers={"Cache-Control": "no-store"})
# Other methods: 405 with Allow, also no-store.
@router.api_route("/api/agent-identity", methods=[m for m in METHODS if m != "GET"], include_in_schema=False)
def agent_identity_method() -> Response:
return no_store(405, {"error": "method_not_allowed"}, {"Allow": "GET"})
@router.api_route("/api/agent-session", methods=[m for m in METHODS if m != "POST"], include_in_schema=False)
def agent_session_method() -> Response:
return no_store(405, {"error": "method_not_allowed"}, {"Allow": "POST"})

Both endpoints get the user through the current_user dependency. It is the seam to your authentication: return your signed-in user, or None. Return None rather than raising HTTPException, so the endpoints can reject cross-site requests (403) before they report a missing session (401). If your user objects are not dictionaries, change the two user["id"] lookups in app/agent.py to your stable ID attribute.

app/dependencies.py
from fastapi import Request
def current_user(request: Request) -> dict[str, str] | None:
"""The signed-in user of this request (your existing authentication), or None.
The example reads Starlette's signed session cookie; use your own auth dependency here. Return None
when signed out instead of raising, so the Bunkhouse endpoints can answer 403 before 401.
"""
return request.session.get("user")

Include the router in your application:

app/main.py (excerpt)
# Bunkhouse endpoints; they read the signed-in user through the current_user dependency.
app.include_router(agent_router)

In the base template’s head, load the page hooks and then the loader, before your application’s own scripts: window.AgentEmbedConfig must be defined before the loader runs. The hooks are external same-origin scripts, so no inline script needs a CSP exception. The base template loads the auth-headers hook (see Auth headers) and then the mode partial, which here is the identity hook followed by the loader:

templates/agent-embed-identity.html
{# Identity token mode (recommended): the hook first, then the loader, before the app's scripts. #}
<script src="/static/agent-identity.js" nonce="{{ nonce }}"></script>
<script src="{{ agent.embed_origin }}/embed.js" nonce="{{ nonce }}"
data-public-key="{{ agent.public_key }}" data-hints-url="/agent-hints.json"></script>
static/agent-identity.js
// Identity token mode: the embed calls this hook whenever it starts a session for the signed-in user.
window.AgentEmbedConfig = {
...window.AgentEmbedConfig,
getIdentityToken: async () => {
const res = await fetch("/api/agent-identity", { credentials: "same-origin", cache: "no-store" });
if (!res.ok) throw new Error("Not signed in");
return (await res.json()).identityToken;
},
};
templates/base.html (excerpt)
{% if csrf_token %}<meta name="csrf-token" content="{{ csrf_token }}" />{% endif %}
<script src="/static/agent-auth-headers.js" nonce="{{ nonce }}"></script>
{% include "agent-embed-" ~ mode ~ ".html" %}
<!-- The application's own JavaScript comes after the loader. -->
<script src="/static/app.js" nonce="{{ nonce }}" defer></script>

The example’s base template includes one of the three agent-embed-* partials, selected by a test-only ?mode= switch. Your application uses exactly one.

Instead of the identity hook, add data-session-endpoint to the loader. Do not keep agent-identity.js on the page: when getIdentityToken is defined it takes precedence.

templates/agent-embed-mint.html
{# Server mint mode: no identity hook; the embed posts to the same-origin session endpoint. #}
<script src="{{ agent.embed_origin }}/embed.js" nonce="{{ nonce }}"
data-public-key="{{ agent.public_key }}" data-hints-url="/agent-hints.json"
data-session-endpoint="/api/agent-session"></script>

POST /api/agent-session in app/agent.py above accepts only requests with an Origin from AGENT_ALLOWED_ORIGINS, X-Agent-Embed: 1 and Sec-Fetch-Site: same-origin (403 otherwise), authenticates the session (401), then creates the session with the mint key and returns the API’s status and body unchanged with Cache-Control: no-store. If the API is unreachable it returns 502 without details. It is a plain def endpoint, so FastAPI runs the blocking upstream request in its threadpool. Set AGENT_EMBED_SECRET_KEY=sk_... on the server; the identity endpoint does not need it.

FastAPI has no built-in CSRF protection. If you add a CSRF middleware, exempt only /api/agent-session: the embed’s POST carries no CSRF token, and the three header checks are that route’s protection.

Anonymous sessions need only the loader. Enable anonymous access for the site in the dashboard first and set its monthly token and spend limits; usage is charged to your account’s credit and stops at a limit. Anyone who can load your page can start one, so it is not proof of identity.

templates/agent-embed-anonymous.html
{# Anonymous mode (enable it for the site first): the loader alone. #}
<script src="{{ agent.embed_origin }}/embed.js" nonce="{{ nonce }}"
data-public-key="{{ agent.public_key }}" data-hints-url="/agent-hints.json"></script>

The example signs in and out with ordinary form posts that redirect back to the same page. Every account change therefore loads a new document, and a new document starts a new embed with no session, so the previous user’s session can never carry over. The session cookie is HttpOnly and SameSite=Lax, Secure outside the loopback test origin, and login clears the previous session before storing the new user and a fresh CSRF token:

app/main.py (excerpt)
# Starlette's signed-cookie session: HttpOnly, SameSite=Lax, Secure except on the loopback HTTP test origin.
app.add_middleware(
SessionMiddleware,
secret_key=SESSION_SECRET,
session_cookie="example_session",
same_site="lax",
https_only=APP_ORIGIN.startswith("https://"),
)

If your application changes the signed-in user without a full page load (a fetch login, a single-page frontend calling your FastAPI backend), call window.AgentEmbed?.resetSession() right after the login, logout or account switch succeeds and before the new user’s requests. Render the loader once in the base template, not in per-page templates.

The assistant calls your API with the page’s cookies; supply any extra header your API requires with getAuthHeaders. The example’s session owns a random CSRF token, rendered into a <meta> tag for signed-in users and checked by a require_csrf dependency on every write (see app/main.py):

static/agent-auth-headers.js
// The app's own CSRF token (rendered by the server for the signed-in session) for the assistant's calls
// to this app's API. Values never reach the model or the sandbox.
window.AgentEmbedConfig = {
...window.AgentEmbedConfig,
getAuthHeaders: () => {
const token = document.querySelector('meta[name="csrf-token"]')?.content;
return token ? { "X-CSRF-Token": token } : {};
},
};

Both hooks merge into window.AgentEmbedConfig and load before the loader, so neither overwrites the other. The header values never reach the model or the sandbox.

Send a stable, opaque user ID (the example’s fixture-alice), not a name or email address.

Add the Bunkhouse sources to your existing policy rather than replacing it. The example’s HTTP middleware creates a fresh nonce per response in request.state.csp_nonce, and the templates put it on their script tags; no 'unsafe-inline' or 'unsafe-eval' is needed:

app/csp.py
# The app's existing Content-Security-Policy with the Bunkhouse sources added: the embed origin to
# script-src and frame-src, the API to connect-src. A fresh nonce per response (request.state.csp_nonce)
# goes on the page's script tags.
import secrets
from collections.abc import Awaitable, Callable
from fastapi import Request, Response
from .config import agent_config
async def content_security_policy(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
nonce = request.state.csp_nonce = secrets.token_urlsafe(16)
response = await call_next(request)
response.headers["Content-Security-Policy"] = "; ".join(
[
"default-src 'self'",
f"script-src 'self' 'nonce-{nonce}' {agent_config.embed_origin}",
f"style-src 'self' 'nonce-{nonce}'",
f"frame-src {agent_config.embed_origin}",
f"connect-src 'self' {agent_config.api_base}",
"img-src 'self' data:",
"object-src 'none'",
"base-uri 'self'",
"form-action 'self'",
"frame-ancestors 'self'",
]
)
return response

From a checkout of the Bunkhouse repository, with Docker and local Supabase running:

Terminal window
node tests/framework-integrations/run.ts fastapi

The run builds the example image, starts it against a fresh local site and drives the real pane in Chromium: identity, mint and anonymous sessions, endpoint guards, sign-out and account switches, discovery of the page’s startup request, a CSRF-protected write through getAuthHeaders, CSP and secret checks. In your own app, sign in and open the Assistant launcher: the first message creates a signed session.

Symptom Check
/api/agent-identity returns 401 current_user does not find the user: the session cookie was not sent (for example https_only=True on an http origin) or your auth dependency returns None. A missing SessionMiddleware shows up as a 500 instead.
403 from either endpoint The request is cross-site or missing Sec-Fetch-Site; for mint, Origin must exactly match AGENT_ALLOWED_ORIGINS and the loader must carry data-session-endpoint. A CSRF middleware must exempt only /api/agent-session.
405 Identity is GET, mint is POST.
503 The signing secret, key ID or mint key for the mode you use is not set.
The pane reports a CSP block Add the embed origin to script-src and frame-src and the API to connect-src, and put the response’s nonce on the script tags.
Startup requests are not discovered The loader must come before your application’s scripts in the head.
Two launchers Render the loader once, in the base template.
A secret in page source Only AGENT_PUBLIC_KEY and origins belong in templates; keep ss_ and sk_ values in server environment variables.