FastAPI Integration

How to integrate Kostrack into a FastAPI application — configure on startup, flush on shutdown, and expose a health endpoint.

Lifespan integration

from contextlib import asynccontextmanager
from fastapi import FastAPI
import kostrack

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    kostrack.configure(
        dsn="postgresql://kostrack:password@localhost/kostrack",
        service_id="openmanagr",
    )
    yield
    # Shutdown — flush remaining queue
    kostrack.shutdown()

app = FastAPI(lifespan=lifespan)

Health endpoint

@app.get("/kostrack/health")
def kostrack_health():
    return kostrack.health()

Response example:

{
  "status": "ok",
  "timescale_available": true,
  "sqlite_backlog": 0,
  "queue_depth": 0,
  "written_timescale": 1247,
  "last_flush": "2026-03-20T08:00:01Z"
}

Per-request cost tracking

from fastapi import Request
from kostrack import Anthropic
import kostrack

client = Anthropic(tags={"project": "openmanagr"})

@app.post("/process-invoice")
async def process_invoice(request: Request, user_id: str):
    with kostrack.trace(
        tags={"feature": "invoice-extraction", "user_id": user_id}
    ) as t:
        response = client.messages.create(...)

    # Log cost to your app metrics if needed
    print(f"Request cost: ${t.total_cost_usd:.6f}")
    return {"result": response.content}