Appearance

FastAPI — full setup ​

Complete FastAPI integration: middleware, OpenAI, lifespan shutdown.

Install ​

bash
pip install "amlexia[fastapi]" openai

Environment ​

env
AMLEXIA_SDK_KEY=am_...
AMLEXIA_INGEST_URL=https://ingest.amlexia.com
AMLEXIA_ENVIRONMENT=production
OPENAI_API_KEY=sk-...

Application ​

python
import os
import time
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from openai import OpenAI
from amlexia import AmlexiaClient
from amlexia.fastapi_integration import AmlexiaMiddleware
from amlexia.openai_helper import track_openai_completion
from amlexia.http_wrap import wrap_requests_session
import requests

client = AmlexiaClient.from_env()
http = wrap_requests_session(client, requests.Session())
openai = OpenAI(http_client=http)  # optional: route OpenAI via wrapped session

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    client.shutdown()

app = FastAPI(lifespan=lifespan)
app.add_middleware(AmlexiaMiddleware, client=client, service_name="api")

@app.get("/api/health")
async def health():
    return {"status": "ok"}

@app.post("/api/chat")
async def chat(request: Request):
    body = await request.json()
    start = time.time()
    try:
        completion = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=body["messages"],
        )
        track_openai_completion(
            client,
            model=completion.model,
            status_code=200,
            latency_ms=int((time.time() - start) * 1000),
            usage=completion.usage,
        )
        return completion.model_dump()
    except Exception as e:
        client.track(
            endpoint="POST /api/chat",
            method="POST",
            status_code=500,
            latency_ms=int((time.time() - start) * 1000),
            provider="openai",
            error_message=str(e)[:500],
        )
        raise

@app.post("/webhooks/stripe")
async def stripe_webhook():
    return {"received": True}

Manual track for background tasks ​

python
client.track(
    endpoint="task/nightly-sync",
    method="JOB",
    status_code=200,
    latency_ms=120000,
    environment=os.environ.get("AMLEXIA_ENVIRONMENT"),
)

Verify ​

bash
amlexia health
uvicorn main:app --reload