Python and FastAPI

API = os.environ.get("AGENTDB_API_URL", "https://app.agentdb.link").rstrip("/")
WORKSPACE = os.environ["AGENTDB_WORKSPACE"]
TOKEN = os.environ["AGENTDB_TOKEN"]  # an agent key with role editor, never an owner key

@asynccontextmanager
async def lifespan(app: FastAPI):
    # One connection pool for the whole app.
    app.state.agentdb = httpx.AsyncClient(
        base_url=f"{API}/v1/workspaces/{WORKSPACE}",
        headers={"Authorization": f"Bearer {TOKEN}"},
        timeout=15,
    )
    yield
    await app.state.agentdb.aclose()


async def agentdb(request: Request, method: str, path: str, body=None, idempotency_key: str | None = None):
    headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {}
    try:
        res = await request.app.state.agentdb.request(method, path, json=body, headers=headers)
    except httpx.HTTPError as e:
        raise AgentDBError(0, None, "Could not reach AgentDB.") from e
    payload = res.json()
    if res.is_error:
        err = payload.get("error", {})
        raise AgentDBError(res.status_code, err.get("code"), err.get("message", f"AgentDB answered {res.status_code}"))
    return payload["data"]

def agentdb(method, path, body=None, idempotency_key=None):
    req = urllib.request.Request(
        f"{API}/v1/workspaces/{WORKSPACE}{path}",
        method=method,
        data=None if body is None else json.dumps(body).encode(),
        headers={"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json",
                 **({"Idempotency-Key": idempotency_key} if idempotency_key else {})},
    )
    try:
        with urllib.request.urlopen(req, timeout=15) as res:
            return json.load(res)["data"]
    except urllib.error.HTTPError as e:
        err = json.load(e).get("error", {})
        raise AgentDBError(e.code, err.get("code"), err.get("message", f"AgentDB answered {e.code}")) from None
    except urllib.error.URLError:
        raise AgentDBError(0, None, "Could not reach AgentDB.") from None

@app.get("/api/contacts")
async def list_contacts(request: Request, q: str = ""):
    q = q.strip()[:100]
    data = await agentdb(request, "POST", "/query", {
        "object": "contact",
        "where": {"all": [{"field": "name", "op": "contains", "value": q}] if q else []},
        "sort": [{"field": "name", "order": "asc"}],
        "limit": 50,
    })
    return {"contacts": [contact_out(r) for r in data["records"]]}


@app.post("/api/contacts", status_code=201)
async def create_contact(request: Request):
    if not request.headers.get("content-type", "").startswith("application/json"):
        raise HTTPException(415, "Send JSON.")
    body = await request.json()
    if not isinstance(body, dict):
        raise HTTPException(400, "Send a JSON object.")
    name = str(body.get("name") or "").strip()[:200]
    email = str(body.get("email") or "").strip()[:254]
    if not name:
        raise HTTPException(400, "A contact needs a name.")
    submit = "".join(ch for ch in request.headers.get("x-submit-id", str(uuid.uuid4())) if ch.isalnum() or ch == "-")[:64]
    rec = await agentdb(request, "POST", "/records",
                        {"object": "contact", "data": {"name": name, **({"email": email} if email else {})}},
                        idempotency_key=f"fastapi.{submit}")
    return {"contact": contact_out(rec)}

async function load() {
  status.textContent = "Loading…";
  try {
    const res = await fetch("/api/contacts?q=" + encodeURIComponent(q.value.trim()));
    const body = await res.json();
    if (!res.ok) throw new Error(body.error ?? "Could not load contacts.");
    // Set text, never HTML: names come from your data and may contain anything.
    list.replaceChildren(
      ...body.contacts.map((c) => {
        const li = document.createElement("li");
        li.textContent = c.email ? `${c.name} (${c.email})` : c.name;
        return li;
      }),
    );
    status.textContent = body.contacts.length ? "" : q.value ? "No contacts match your search." : "No contacts yet. Add the first one below.";
  } catch (e) {
    list.replaceChildren();
    status.textContent = `Error: ${e.message}`;
  }
}

@app.exception_handler(AgentDBError)
async def agentdb_error(_request: Request, e: AgentDBError):
    if e.code == "VERSION_CONFLICT":
        return JSONResponse({"error": "Someone changed it first. Reload and try again."}, status_code=409)
    if e.status in (400, 422):
        return JSONResponse({"error": e.message}, status_code=400)  # validation: safe to show
    print(f"AgentDB call failed: {e.status} {e.code} {e.message}", flush=True)
    return JSONResponse({"error": "The data service is not available. Try again in a moment."}, status_code=502)


@app.exception_handler(HTTPException)
async def http_error(_request: Request, e: HTTPException):
    return JSONResponse({"error": e.detail}, status_code=e.status_code)

cd examples/starters/python-fastapi
cp .env.example .env    # then fill in your workspace and agent key
python3 -m venv .venv && . .venv/bin/activate
pip install -r requirements.txt python-dotenv
uvicorn app:app --env-file .env --port 8000