How to wire Tidebase into a FastAPI app

Enqueue durable agent runs from your FastAPI routes, execute them in a worker process, and verify recovery webhooks with verify_webhook_signature from the zero-dependency Python SDK. The route returns a runId immediately; the workflow’s checkpoints, state, gates, and costs live in your own Postgres.

from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request
from tidebase import Tidebase, verify_webhook_signature
import json, os

from workflows import generate_report  # def generate_report(run, input): ...

tide = Tidebase()  # reads TIDEBASE_URL, default http://localhost:7373
app = FastAPI()

@app.post("/reports")
def create_report(body: dict):
    result = tide.enqueue(
        "generate-report",
        input={"topic": body["topic"]},
        dedupe_key=f"report-{body['topic']}",
    )
    return {"runId": result["run"]["id"]}

@app.get("/reports/{run_id}")
def report_status(run_id: str):
    detail = tide.runs.get(run_id)
    return {"status": detail["run"]["status"], "state": detail["state"]}

@app.post("/tidebase")
async def recovery_webhook(
    request: Request,
    background: BackgroundTasks,
    x_tidebase_signature: str | None = Header(default=None),
):
    raw = await request.body()  # exact bytes — verification needs them unparsed
    if not verify_webhook_signature(raw, x_tidebase_signature, os.environ["TIDEBASE_WEBHOOK_SECRET"]):
        raise HTTPException(status_code=401, detail="invalid signature")
    payload = json.loads(raw)
    background.add_task(
        tide.run, payload["workflowName"], generate_report, run_id=payload["runId"]
    )
    return {"accepted": True, "runId": payload["runId"]}

Tidebase is an open-source checkpoint layer for AI agents: wrap your steps, and failed runs resume from the last safe point — in your own Postgres, without moving execution into a new runtime.

The shape

Routes enqueue, never execute. tide.enqueue writes a durable queued run with dedupe — a double-submitted form can’t create two reports — and the request returns inside any timeout.
A worker executes. The honest tradeoff, stated plainly: Tidebase does not execute your code. A separate process claims and runs jobs:

from tidebase import Tidebase
from workflows import generate_report

tide = Tidebase()
tide.workflow("generate-report", generate_report)
tide.work(queues=["default"])  # blocks; retries/backoff/leases are Tidebase's problem

If the worker dies mid-run, the lease expires and the reconciler requeues the run; when it’s claimed again, completed steps replay from checkpoints per the replay contract. Async workflows (calling httpx, async LLM SDKs) use tidebase.aio.AsyncTidebase — same protocol, async def steps awaited on the event loop.

The recovery webhook re-invokes. For non-queue runs, Tidebase POSTs a signed run.resume payload when a run stalls or fails. Verify over the raw body (as above — don’t let a parsed model regenerate the JSON), then re-invoke with the same run_id. BackgroundTasks keeps the webhook response fast; for long tails, enqueue instead.

Gates in your product UI

A run paused at run.gate("approve-send", "Send this?") shows a pending gate in tide.runs.get(run_id) — render it in your app, and resolve it using the resolveUrl + resolveToken from the gate’s webhook channel payload. Durable, exactly-once, recorded with the actor. See human approval gates.

What Tidebase does not do here

It is not Celery. There’s overlap (queues, retries) but the center of gravity is the checkpointed run record: replay semantics, gates, live state, and the cost ledger — in Postgres rows you can join against your own tables.
It does not proxy your LLM calls. Record usage explicitly with run.usage.record(...).
Alpha, opt-in auth. Set TIDEBASE_API_KEY before exposing the server beyond localhost.

Repo: https://github.com/BlueprintLabIO/tidebase · See also: How to checkpoint OpenAI Agents SDK runs