--- title: "How to add agent analytics to a FastAPI backend" description: "Add agent analytics to FastAPI with async OpenAI calls, stable conversation IDs, streaming replies, browser identity, and lifespan shutdown." canonical: "https://amplitude.com/en-us/build/fastapi-agent-analytics" --- # How to add agent analytics to a FastAPI backend Add agent analytics to FastAPI with async OpenAI calls, stable conversation IDs, streaming replies, browser identity, and lifespan shutdown. Source: https://amplitude.com/en-us/build/fastapi-agent-analytics --- # How to add agent analytics to a FastAPI backend Install the Amplitude AI SDK in a FastAPI app, wrap an async chat route in an agent session, flush buffered events from the lifespan, and stream replies without dropping the response event. About 20 minutes, all copy-paste. - FastAPI - Uvicorn - Amplitude AI SDK - 20 min Chris Tam · Last updated 19 October 2026 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](https://amplitude.com/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. A `POST /api/chat` route that records `[Agent] User Message` and `[Agent] AI Response` for every turn under one `[Agent] Session ID` per conversation, with `[Agent] Model Name`, `[Agent] Input Tokens`, `[Agent] Output Tokens`, `[Agent] Cost USD`, and `[Agent] Latency Ms` filled in by the OpenAI wrapper, plus a streaming `POST /api/chat/stream` route. Both are tied to the same user and browser session as your product analytics. By the end you'll see those events in your user's event stream and the conversation in agent analytics. ## 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](https://amplitude.com/build/react-product-analytics). 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](https://app.amplitude.com/signup) to get your project API key. These examples use a US-region Amplitude project. For an EU project, supply an [EU-configured analytics client](https://amplitude.com/docs/amplitude-ai/agent-analytics/privacy-modes#data-residency) 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](https://amplitude.com/docs/sdks/agent-analytics/sdk). ```sh 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 ```sh # .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 ```python # 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") ``` **Why.** The module isn't called `amplitude_ai.py` because that's the SDK's own package name, and a local file with that name shadows it, so `from amplitude_ai import AmplitudeAI` imports your file instead. `ai.agent(...)` lives here so requests reuse the same named agent and shared configuration. Add one helper for the user ID, and use the same format your product analytics already sends. app/identity.py ```python # 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 ```python # 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. **MCP.** Working in Cursor or Claude Code? Amplitude's MCP server lets your agent read your project's existing taxonomy before it adds anything, so the `Plan Upgraded` event your billing route already sends and the `[Agent]` events from this page end up in charts together instead of next to a duplicate `planUpgraded`. [Set up the MCP server](https://amplitude.com/docs/amplitude-ai/amplitude-mcp). ## 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 ```python # 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) ``` **Why.**` ai.shutdown()` flushes and then closes the analytics client, which the docs say to call once on process exit. FastAPI only runs the code after `yield` if you pass `lifespan=lifespan` to the constructor, so a lifespan function that isn't wired in silently does nothing. On AWS Lambda and other serverless platforms the SDK detects, sessions flush on exit by themselves. ## 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 ```python # 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 ```javascript // 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`. **Why.** The agent session and the browser session are different things. The agent session is the conversation, and your app owns its ID. The browser session is the visit that powers session replay, and the Browser SDK owns it. Forwarding the device ID also keeps a signed-out visitor from splitting into two users, since a backend that makes up its own device ID per request breaks the merge. ## 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. Full code reference ```python # -- .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", ) ``` Full code reference ```javascript // -- 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. Using Cursor or Claude Code? Paste this prompt ```text 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 [Web frameworksHow to add agent analytics to a Django appThe sync version, with a flush per request under Gunicorn.]()[Web frameworksHow to add agent analytics to a Node.js backendThe same setup for an Express server, with tool calls and server-sent events.]()[Use caseHow to stitch client and server events to one userMake the user ID your backend sends match the one the browser sets.](https://amplitude.com/build/stitch-client-server-events) Reference docs [Agent analytics quickstart →](https://amplitude.com/docs/amplitude-ai/agent-analytics/quickstart)[Amplitude AI SDK reference →](https://amplitude.com/docs/sdks/agent-analytics/sdk)[Agent analytics sessions and user identity →](https://amplitude.com/docs/amplitude-ai/agent-analytics/sessions)[Link agent sessions to session replay →](https://amplitude.com/docs/amplitude-ai/agent-analytics/link-session-replay)[Verify your agent analytics instrumentation →](https://amplitude.com/docs/amplitude-ai/agent-analytics/verify)[Amplitude MCP server →](https://amplitude.com/docs/amplitude-ai/amplitude-mcp) For agents [View raw .md](/build/fastapi-agent-analytics.md)[llms.txt](/llms.txt) ## Ship the instrumentation, then read the data Use the verification steps above to check your agent events, then configure the reports and replay links you need. [Create a free account](https://app.amplitude.com/signup)[Browse all Build tutorials](https://amplitude.com/build)