How to add agent analytics to a FastAPI backend

An async FastAPI route needs an async model client; a synchronous call blocks the worker’s event loop. I also keep streaming sessions inside the generator and wire agent analytics shutdown into the app’s lifespan.

I’ll add an async JSON endpoint and a synchronous streaming endpoint to a FastAPI app served by Uvicorn. Each request uses the same conversation ID and browser identity, and the full reference includes the lifespan setup.

Prerequisites

• A FastAPI app served by Uvicorn that calls a model. The examples use OpenAI; the SDK also wraps Anthropic, Azure OpenAI, Google Gen AI, Bedrock, and Mistral.
• An Amplitude project with agent analytics enabled, and its project API key. In Amplitude, go to Settings > Projects > your project > General, then Manage next to API Key. Agent data can live in the same project as your product analytics, which is the simplest setup.
• The View Agent Analytics Objects permission on your role, which you need to open agent analytics at all.
• Optional, for step 6: browser analytics with @amplitude/unified, as in How to add product analytics to a React app.

No Amplitude account yet? A free account is enough for this tutorial; every plan includes a free allotment of agent sessions. Sign up for free to get your project API key.

These examples use a US-region Amplitude project. For an EU project, supply an EU-configured analytics client instead of initializing with the API key alone.

1. Install the SDK

The Python package is amplitude-ai. It installs provider support as extras, so add the openai extra for the OpenAI wrapper. For every option, see the agent analytics SDK reference.

uv add "amplitude-ai[openai]"

The docs name the variable AMPLITUDE_AI_API_KEY. It holds your standard project API key. Set both keys in the environment Uvicorn runs in: your process manager, your host's settings, or the .env loader you already use.

# .env
AMPLITUDE_AI_API_KEY=your-project-api-key
OPENAI_API_KEY=your-openai-key

Prefer to let a coding agent do the instrumentation? Install the package, then tell your agent to run amplitude-ai --print-guide and follow the guide it prints. The steps below are what it should produce on FastAPI.

2. Initialize once in a module of its own

Create one module in your package that constructs the client, the wrapped OpenAI clients, and the agent at import time. Routes import from it and never construct their own. The async client serves async def routes; the sync client serves the streaming generator in step 5.

# app/agent_analytics.py
import os

from amplitude_ai import AIConfig, AmplitudeAI, AsyncOpenAI, OpenAI

ai = AmplitudeAI(
    api_key=os.environ["AMPLITUDE_AI_API_KEY"],
    config=AIConfig(
        content_mode="full",  # prompt and response text, with PII redaction
        redact_pii=True,
        debug=os.environ.get("AMPLITUDE_AI_DEBUG") == "true",
    ),
)

# Use these instead of the clients from the openai package. They forward
# every call and record model, tokens, latency, and cost on [Agent] AI Response.
async_openai_client = AsyncOpenAI(amplitude=ai, api_key=os.environ["OPENAI_API_KEY"])
openai_client = OpenAI(amplitude=ai, api_key=os.environ["OPENAI_API_KEY"])

support_agent = ai.agent("support-chat")

Add one helper for the user ID, and use the same format your product analytics already sends.

# app/identity.py
def amplitude_user_id(user_id):
    # No account, no user ID. Never send placeholders like "anonymous".
    if user_id is None:
        return None
    # Prefix integer primary keys: Amplitude drops user IDs shorter than
    # 5 characters, and an ID of 42 is two.
    return f"user-{user_id}"

content_mode="full" is the default and sends prompt and response text after scrubbing emails, phone numbers, SSNs, card numbers, and IP addresses. redact_pii defaults to True from version 1.17.0 and to False before it, so set it explicitly when upgrading an older installation. If message text must stay in your infrastructure, use "metadata_only": you still get tokens, latency, model, cost, and session grouping, with no text. With AMPLITUDE_AI_DEBUG=true, every tracked event prints to stderr.

Redaction matches patterns; it does not remove every kind of personal data. Review representative payloads before production capture. In metadata_only, built-in quality signals that need message text are unavailable.

3. Wrap an async route in a session

In an async def route, use the async wrapper and async with. The wrapper sees the call happen inside the session and records the user's latest message from messages and the AI response against it, so the route doesn't call track_user_message() as well. A sync OpenAI call inside an async route blocks the event loop for the length of the model call, which stalls every other request on that worker.

# app/main.py
from fastapi import FastAPI, Request
from pydantic import BaseModel

from .agent_analytics import ai, async_openai_client, support_agent
from .identity import amplitude_user_id

app = FastAPI()  # step 4 adds the lifespan


class ChatMessage(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    conversationId: str
    messages: list[ChatMessage]


def current_user_id(request: Request):
    # Replace with your auth dependency. Return None when signed out.
    user = getattr(request.state, "user", None)
    return getattr(user, "id", None)


def session_for(request: Request, conversation_id: str):
    return support_agent.session(
        user_id=amplitude_user_id(current_user_id(request)),
        # Sent by the browser in step 6, from the Browser SDK.
        device_id=request.headers.get("x-amplitude-device-id"),
        browser_session_id=request.headers.get("x-amplitude-session-id"),
        session_id=conversation_id,
    )


@app.post("/api/chat")
async def chat(body: ChatRequest, request: Request):
    messages = [m.model_dump() for m in body.messages]

    async with session_for(request, body.conversationId):
        completion = await async_openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
        )

    return {"reply": completion.choices[0].message.content or ""}

Run it with uvicorn app.main:app. The relative imports assume app/ is a package with an __init__.py.

The SDK also ships an AmplitudeAIMiddleware for FastAPI and Starlette, installed with the fastapi extra, that opens a session for every request. The explicit session above is the version where you choose the session ID.

Reuse conversationId across every turn of the same conversation, and never across users. Every turn then carries the same [Agent] Session ID, so don't generate a fresh one per request. When the same user starts a new conversation, send a new ID.

The async with block closes the session when it exits, and Amplitude enriches a session once, at its first close. Later turns with the same ID are stored, but they don't reach that session's [Agent] Session Record. Pass a user ID, a device ID, or both on every session; a session with neither has nothing to attach to.

Use the canonical model ID, gpt-4o-mini, not a gateway alias. Cost comes from token counts multiplied against a public model-price catalog, and a name the catalog can't price gets no cost.

4. Flush from the lifespan on shutdown

A Uvicorn worker is a long-running process, so the SDK skips the per-session flush and lets the analytics client batch events normally. That's efficient until the worker stops with a batch still in memory. Close the client from FastAPI's lifespan, which runs once per worker on shutdown.

# app/main.py (replace the app = FastAPI() line from step 3)
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    # The worker is stopping: send everything still buffered, then close.
    ai.shutdown()


app = FastAPI(lifespan=lifespan)

5. Keep streaming sessions open until the last chunk

A streamed reply finishes after your route returns. If the session opens in the route, it closes before the model finishes and the [Agent] AI Response event is dropped. Open the session inside the generator instead, so it stays open until the last chunk goes out.

# app/main.py (add to the file)
from fastapi.responses import StreamingResponse

from .agent_analytics import openai_client


def stream_answer(request, conversation_id, messages):
    with session_for(request, conversation_id):
        stream = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            stream=True,
            # Streams only report token usage when asked; tokens and cost need it.
            stream_options={"include_usage": True},
        )
        for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content


@app.post("/api/chat/stream")
def chat_stream(body: ChatRequest, request: Request):
    return StreamingResponse(
        stream_answer(
            request,
            body.conversationId,
            [m.model_dump() for m in body.messages],
        ),
        media_type="text/plain",
    )

The route and generator are plain def, the same shape as the SDK reference's FastAPI streaming example. FastAPI runs a plain def route in its thread pool instead of on the event loop. The session opens when the generator starts, on the first chunk, and closes after the last one.

6. Send the browser IDs with each request

To tie a conversation to the product session around it, and to get a session replay link on it, the route needs the Browser SDK's session ID and device ID. Steps 3 and 5 already read them from two headers; send them from the browser.

// src/chat.js (browser)
import * as amplitude from '@amplitude/unified';

export async function sendChat(conversationId, messages) {
  const headers = { 'Content-Type': 'application/json' };
  const deviceId = amplitude.getDeviceId();
  const sessionId = amplitude.getSessionId();
  if (deviceId) headers['x-amplitude-device-id'] = deviceId;
  if (sessionId) headers['x-amplitude-session-id'] = String(sessionId);

  const res = await fetch('/api/chat', {
    method: 'POST',
    headers,
    body: JSON.stringify({ conversationId, messages }),
  });
  if (!res.ok) throw new Error(`/api/chat failed with ${res.status}`);
  return (await res.json()).reply;
}

Create conversationId with crypto.randomUUID() when a conversation starts and keep it for that conversation only. If the frontend is on another origin, configure CORSMiddleware to allow POST in allow_methods, which defaults to GET only, and the two x-amplitude-* headers in allow_headers.

Verify it worked

Run AMPLITUDE_AI_DEBUG=true uvicorn app.main:app --reload, send two messages in the same conversation, then check in this order:

1. Terminal. Every tracked event prints to stderr as a one-line summary, for example [amplitude-ai] [Agent] AI Response | user=user-42 session=... agent=support-chat model=gpt-4o-mini .... If nothing prints, check instrumentation, including an LLM call outside the session or an unwrapped client. You can also run amplitude-ai doctor, or amplitude-ai verify for local checks.
2. Network tab. Filter for /api/chat. The request headers should include x-amplitude-session-id and x-amplitude-device-id. The server's own upload to Amplitude doesn't show here, since it leaves from Python.
3. Ingestion Debugger. In Amplitude, click Data, select Sources, then open the Ingestion Debugger tab. The successful requests chart should show your server's uploads. It shows counts, not individual events.
4. User lookup. Go to Users > User Profiles and search your user ID or device ID. Within ten seconds to a minute the event stream shows [Agent] User Message and [Agent] AI Response for each turn. Open an AI Response and confirm [Agent] Model Name, [Agent] Provider, [Agent] Input Tokens, [Agent] Output Tokens, [Agent] Cost USD, [Agent] Latency Ms, [Agent] Session ID (your conversationId), and [Agent] Agent ID (support-chat) are all set.
5. Agent analytics. Open agent analytics in the left nav and find the session by your conversationId. Quality signals typically arrive on [Agent] Session Record 15 to 20 minutes after the first close, and cover the events received before it. An enrichment backlog can delay them.

Troubleshooting

ImportError: cannot import name 'AmplitudeAI' from partially initialized module

A file in your project is named amplitude_ai.py and shadows the SDK package. Rename it, for example to agent_analytics.py, and delete any stale __pycache__ copy.

Requests stall or time out under load

A sync model call runs inside an async def route and blocks the event loop. Use async_openai_client with await in async routes, or make the route a plain def so FastAPI runs it in the thread pool.

Nothing prints in the terminal and no AI Response arrives

The model call used an unwrapped client or ran outside the session. Import the clients from app/agent_analytics.py, not from the openai package, and make the call inside the with or async with block.

Every turn has two User Message events

The route calls s.track_user_message() for a message it also sends through the wrapped client, which records it already. Remove the manual call.

The last events before a deploy never arrive

The lifespan isn't wired in, so ai.shutdown() never ran. Confirm the app is built with FastAPI(lifespan=lifespan) and that your host sends SIGTERM and waits for Uvicorn to finish shutting down.

The browser IDs never reach the route

A cross-origin preflight rejected the request. In CORSMiddleware, add POST to allow_methods and x-amplitude-device-id and x-amplitude-session-id to allow_headers.

Streamed replies have no AI Response

The session closed before the stream finished. Open the session inside the generator, as in step 5, not in the route that returns the StreamingResponse.

[Agent] Cost USD is empty

The model name isn't one the price catalog knows, such as a gateway alias or a fine-tuned ft: model. Pass the canonical provider model ID. On streamed replies, also check that stream_options asks for usage.

FAQ

Can I keep conversation text out of Amplitude?

Yes. Set the privacy mode to metadata_only. Tokens, cost, latency, models, and session grouping remain available; message text and tool payloads stay out. Built-in quality signals that need message text will not run.

Should every message get a new agent session ID?

Use one stable ID for the same conversation or task, and a new ID for a new job. Never reuse an ID across users. These scope helpers close at the end of each request or task; later events with the same ID are stored but do not update the first Session Record.

Why are events visible before quality signals?

Events normally appear within about a minute. The Session Record typically follows 15 to 20 minutes after an explicit close, and can take longer during an enrichment backlog.

Full code reference

Every file from this tutorial in one block, the version to hand to an agent.

# -- .env --
# AMPLITUDE_AI_API_KEY=your-project-api-key
# OPENAI_API_KEY=your-openai-key

# -- app/__init__.py --
# (empty)

# -- app/agent_analytics.py --
import os

from amplitude_ai import AIConfig, AmplitudeAI, AsyncOpenAI, OpenAI

ai = AmplitudeAI(
    api_key=os.environ["AMPLITUDE_AI_API_KEY"],
    config=AIConfig(
        content_mode="full",
        redact_pii=True,
        debug=os.environ.get("AMPLITUDE_AI_DEBUG") == "true",
    ),
)

async_openai_client = AsyncOpenAI(amplitude=ai, api_key=os.environ["OPENAI_API_KEY"])
openai_client = OpenAI(amplitude=ai, api_key=os.environ["OPENAI_API_KEY"])

support_agent = ai.agent("support-chat")

# -- app/identity.py --
def amplitude_user_id(user_id):
    if user_id is None:
        return None
    return f"user-{user_id}"

# -- app/main.py --
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from .agent_analytics import ai, async_openai_client, openai_client, support_agent
from .identity import amplitude_user_id


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


app = FastAPI(lifespan=lifespan)


class ChatMessage(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    conversationId: str
    messages: list[ChatMessage]


def current_user_id(request: Request):
    user = getattr(request.state, "user", None)
    return getattr(user, "id", None)


def session_for(request: Request, conversation_id: str):
    return support_agent.session(
        user_id=amplitude_user_id(current_user_id(request)),
        device_id=request.headers.get("x-amplitude-device-id"),
        browser_session_id=request.headers.get("x-amplitude-session-id"),
        session_id=conversation_id,
    )


@app.post("/api/chat")
async def chat(body: ChatRequest, request: Request):
    messages = [m.model_dump() for m in body.messages]
    async with session_for(request, body.conversationId):
        completion = await async_openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
        )
    return {"reply": completion.choices[0].message.content or ""}


def stream_answer(request, conversation_id, messages):
    with session_for(request, conversation_id):
        stream = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            stream=True,
            stream_options={"include_usage": True},
        )
        for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content


@app.post("/api/chat/stream")
def chat_stream(body: ChatRequest, request: Request):
    return StreamingResponse(
        stream_answer(
            request,
            body.conversationId,
            [m.model_dump() for m in body.messages],
        ),
        media_type="text/plain",
    )

// -- src/chat.js (browser) --
import * as amplitude from '@amplitude/unified';

export async function sendChat(conversationId, messages) {
  const headers = { 'Content-Type': 'application/json' };
  const deviceId = amplitude.getDeviceId();
  const sessionId = amplitude.getSessionId();
  if (deviceId) headers['x-amplitude-device-id'] = deviceId;
  if (sessionId) headers['x-amplitude-session-id'] = String(sessionId);

  const res = await fetch('/api/chat', {
    method: 'POST',
    headers,
    body: JSON.stringify({ conversationId, messages }),
  });
  if (!res.ok) throw new Error(`/api/chat failed with ${res.status}`);
  return (await res.json()).reply;
}

Hand it to your agent

Prefer to have Cursor, Claude Code, or Codex CLI do the wiring? This prompt points the agent at this page and pins the decisions that matter.

Add Amplitude agent analytics to this FastAPI app by following
https://amplitude.com/build/fastapi-agent-analytics exactly. Install
"amplitude-ai[openai]". Create app/agent_analytics.py (never name it
amplitude_ai.py) that builds AmplitudeAI from
os.environ["AMPLITUDE_AI_API_KEY"], the wrapped AsyncOpenAI and OpenAI
clients from amplitude_ai, and the agent, all at module level; never call
ai.agent() inside a route. Use the same user ID format as our product
analytics, prefixed so it's at least 5 characters, and pass None when
signed out instead of a placeholder. In each async LLM route, open
agent.session(user_id, device_id, browser_session_id, session_id) with
"async with", the conversation ID as session_id, and await the async
client inside it. Don't call track_user_message() for a message you
pass to the wrapped client; the wrapper records it. Call ai.shutdown() after yield in a lifespan passed to
FastAPI(lifespan=...). For streaming, open the session inside a sync
generator passed to StreamingResponse and pass
stream_options={"include_usage": True}. Read device and session IDs from
the x-amplitude-device-id and x-amplitude-session-id headers. If the
frontend is cross-origin, allow POST and both headers in CORSMiddleware. Don't hardcode keys.
Report every file you changed.

What this unlocks

• Chart [Agent] Cost USD and [Agent] Latency Ms by model on AI Response events.
• Connect agent and product events in funnels when they use the same stable user ID.
• Review content-based quality signals on the Session Record after the session closes, when your privacy mode sends message text.

When browser session replay is already recording, forwarding both browser IDs adds a replay link to the agent session. Verify that recording separately from the server events.

Next steps