--- title: "How to add agent analytics to a LangChain app" description: "Add Amplitude agent analytics to LangChain with a callback, one session per complete invocation, async calls, token costs, and a shutdown flush." canonical: "https://amplitude.com/en-us/build/langchain-agent-analytics" --- # How to add agent analytics to a LangChain app Add Amplitude agent analytics to LangChain with a callback, one session per complete invocation, async calls, token costs, and a shutdown flush. Source: https://amplitude.com/en-us/build/langchain-agent-analytics --- # How to add agent analytics to a LangChain app Install the Amplitude AI SDK with its LangChain extra, pass the Amplitude callback handler to a LangChain agent on every call, group each completed agent invocation into an agent session, and flush before the process exits. About 20 minutes, all copy-paste. - Python 3.10+ - LangChain 1.x - Amplitude AI SDK - 20 min Chris Tam · Last updated 21 October 2026 LangChain gives you the model, the tools, and the loop between them, and its callback system sees every model call and tool call that loop makes. [Agent analytics](https://amplitude.com/agent-analytics) turns those calls into answers about cost, quality, and where users give up. In this tutorial, I will pass Amplitude's callback handler to a LangChain agent, give each complete invocation its own agent session and cover async calls, and make sure buffered events ship before your process stops. A `create_agent` support assistant with one `lookup_order` tool. Each complete agent invocation runs inside a session keyed by its job ID, and the `AmplitudeCallbackHandler` in the call's config records the agent's model calls as `[Agent] AI Response` and its tool runs as `[Agent] Tool Call`, under your user ID and one `[Agent] Session ID` per invocation. You'll have sync and async versions of the same call. By the end you'll see those events in your user's event stream and the conversation in agent analytics. ## Prerequisites - Python 3.10 or later, an authenticated user ID, and a LangChain 1.x app that builds its agent with `create_agent`. The examples use OpenAI through `langchain-openai`. - An Amplitude project with agent analytics enabled, and its project API key. In Amplitude, go to Settings > Projects > your project > General. 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. 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. ## 1. Install the SDK with the LangChain extra The Python package is `amplitude-ai`, and it installs framework support as extras. The `langchain` extra adds the LangChain integration. This tutorial uses `amplitude-ai` 1.20.1 or later. Install LangChain and its OpenAI provider package next to it. For every option, see the [agent analytics SDK reference](https://amplitude.com/docs/sdks/agent-analytics/sdk). ```sh uv add "amplitude-ai[langchain]>=1.20.1" langchain langchain-openai ``` The docs name the variable `AMPLITUDE_AI_API_KEY`. It holds your standard project API key. Set it and your OpenAI key in the environment your app runs in. .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 for a LangChain agent. ## 2. Initialize once in a module of its own Create one module that constructs the Amplitude client and names your agent at import time. Everything else imports from it. app/agent\_analytics.py ```python # app/agent_analytics.py import os from amplitude_ai import AIConfig, AmplitudeAI 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", ), ) # Reuse the named agent and its configuration across requests. support_agent = ai.agent("support-assistant") ``` **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. 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): # This example serves authenticated users. Never invent a user ID. if user_id is None: raise ValueError("Pass an authenticated user ID") # Prefix integer primary keys: Amplitude's HTTP V2 API requires IDs of # at least 5 characters by default, 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 the explicit `True` keeps older installs safe. The SDK applies the privacy mode to framework integrations too, so the LangChain handler follows these settings. If your conversations are regulated, 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. ## 3. Build the LangChain agent once Build the agent at import time, the same way you would without analytics. Don't attach the Amplitude handler here: the agent is shared by every user, and the handler carries one user's identity. app/assistant.py ```python # app/assistant.py from langchain.agents import create_agent from langchain.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import InMemorySaver @tool def lookup_order(order_id: str) -> str: """Look up the status of an order by its ID.""" # Replace with your order service. return f"Order {order_id} shipped on Monday." assistant = create_agent( model=ChatOpenAI(model="gpt-4o-mini"), tools=[lookup_order], # Keeps each conversation's history under its thread_id. # Swap in a persistent checkpointer for production. checkpointer=InMemorySaver(), ) ``` 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 cannot price gets no cost. This example uses completed, non-streaming OpenAI calls through `invoke` or `ainvoke`. ## 4. Pass the handler in each call's config LangChain takes callback handlers at run time through the `callbacks` key of a call's `config`. Create an `AmplitudeCallbackHandler` per call, pass it there, and make the call inside an agent session keyed by this complete job. app/assistant.py ```python # app/assistant.py (add to the file) from amplitude_ai.integrations.langchain import AmplitudeCallbackHandler from .agent_analytics import ai, support_agent from .identity import amplitude_user_id def run_turn(user_id, conversation_id, message): uid = amplitude_user_id(user_id) with support_agent.session(user_id=uid, session_id=conversation_id): handler = AmplitudeCallbackHandler(ai, user_id=uid) result = assistant.invoke( {"messages": [{"role": "user", "content": message}]}, config={ "callbacks": [handler], "configurable": {"thread_id": conversation_id}, }, ) return result["messages"][-1].content ``` The job ID does two jobs. LangChain uses it as the `thread_id` for this invocation's history, and Amplitude uses it as the `[Agent] Session ID`. One invocation includes the model calls and tool calls needed to answer one request. Generate a fresh ID for each new job and never share it across users. **Why.** The handler goes in `config` on each call, not into the agent or model you built once at import. Built in, one user's ID would ride along on every other user's events. Created per call, each request gets a handler with its own user, and nothing leaks between concurrent requests. The `with` block closes the session after the complete invocation. Amplitude enriches a session once, after its first close, so finish the job before closing. If your application keeps one job open across several HTTP requests, use an explicit session lifecycle instead of opening and closing this scope for every request. Pass a stable user ID or device ID on every session; this example requires an authenticated user ID. **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 code 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). ## 5. Use the same pattern for async calls In async code, open the session with `async with` and call `ainvoke`. The handler goes in the same place. app/assistant.py ```python # app/assistant.py (add to the file) async def arun_turn(user_id, conversation_id, message): uid = amplitude_user_id(user_id) async with support_agent.session(user_id=uid, session_id=conversation_id): handler = AmplitudeCallbackHandler(ai, user_id=uid) result = await assistant.ainvoke( {"messages": [{"role": "user", "content": message}]}, config={ "callbacks": [handler], "configurable": {"thread_id": conversation_id}, }, ) return result["messages"][-1].content ``` ## 6. Flush before the process exits The SDK batches events in memory. A script, a notebook cell, a CLI, or a worker that exits right after the agent answers can take the last batch with it. Call `ai.shutdown()` once when the process is done; it flushes and then closes the analytics client. main.py ```python # main.py import uuid from app.agent_analytics import ai from app.assistant import run_turn def main(): conversation_id = str(uuid.uuid4()) try: print(run_turn(42, conversation_id, "Where is order A-1001?")) finally: # Ship everything still buffered, then close the client. ai.shutdown() if __name__ == "__main__": main() ``` In a long-running server, call `ai.shutdown()` from your shutdown hook instead: a `SIGTERM` handler, or the FastAPI lifespan. Don't call it per request, since it closes the client. On AWS Lambda and other serverless platforms the SDK detects, sessions flush on exit by themselves. **Why.** The `finally` runs even when the agent raises, so a failed run still ships the events that explain the failure. ## Verify it worked Run `AMPLITUDE_AI_DEBUG=true uv run python main.py`, 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-assistant model=gpt-4o-mini ...`. Nothing printing means the handler never reached the run, usually a `config` without `callbacks`. You can also run `amplitude-ai doctor`, or `amplitude-ai verify` for local checks. 2. **Ingestion Debugger.** In Amplitude, click Data, select Sources, then open the Ingestion Debugger tab. The successful requests chart should show your app's uploads. It shows counts, not individual events. 3. **User lookup.** Go to Users > User Profiles and search `user-42`. Within ten seconds to a minute the event stream shows `[Agent] AI Response` events for the model calls and an `[Agent] Tool Call` for `lookup_order`. Open an AI Response and confirm `[Agent] Model Name`, `[Agent] Input Tokens`, `[Agent] Output Tokens`, `[Agent] Cost USD`, `[Agent] Latency Ms`, `[Agent] Session ID` (your job ID), and `[Agent] Agent ID` (`support-assistant`) are set. 4. **Agent analytics.** Open agent analytics in the left nav and find the session by your job ID. Quality signals arrive on `[Agent] Session Record` typically 15 to 20 minutes after the first close, and cover the events received before it. ## 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. ### Nothing prints in the terminal The handler isn't in the call's config, or the call ran outside the session. Pass `config={"callbacks": [handler]}` on the `invoke` or `ainvoke` call itself, make the call inside the `with` or `async with` block, and check that a wrapper function isn't dropping the `config` argument on its way to the agent. ### One user's events show up under another user The handler was created once and shared. Create `AmplitudeCallbackHandler` inside the function that serves the request, as in step 4, never at module scope or on the agent. ### Events from one invocation have different `[Agent] Session ID` values The job ID changes inside the invocation. Generate it once before calling `run_turn` or `arun_turn`, and pass the same value as both `session_id` and `thread_id` for that complete run. ### The last events before exit never arrive The process ended with a batch still buffered. Call `ai.shutdown()` in a `finally` for scripts and workers, or from your server's shutdown hook. ### `[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. Check that the completed response includes provider token usage. ## 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 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", ), ) support_agent = ai.agent("support-assistant") # -- app/identity.py -- def amplitude_user_id(user_id): if user_id is None: raise ValueError("Pass an authenticated user ID") return f"user-{user_id}" # -- app/assistant.py -- from amplitude_ai.integrations.langchain import AmplitudeCallbackHandler from langchain.agents import create_agent from langchain.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import InMemorySaver from .agent_analytics import ai, support_agent from .identity import amplitude_user_id @tool def lookup_order(order_id: str) -> str: """Look up the status of an order by its ID.""" return f"Order {order_id} shipped on Monday." assistant = create_agent( model=ChatOpenAI(model="gpt-4o-mini"), tools=[lookup_order], checkpointer=InMemorySaver(), ) def run_turn(user_id, conversation_id, message): uid = amplitude_user_id(user_id) with support_agent.session(user_id=uid, session_id=conversation_id): handler = AmplitudeCallbackHandler(ai, user_id=uid) result = assistant.invoke( {"messages": [{"role": "user", "content": message}]}, config={ "callbacks": [handler], "configurable": {"thread_id": conversation_id}, }, ) return result["messages"][-1].content async def arun_turn(user_id, conversation_id, message): uid = amplitude_user_id(user_id) async with support_agent.session(user_id=uid, session_id=conversation_id): handler = AmplitudeCallbackHandler(ai, user_id=uid) result = await assistant.ainvoke( {"messages": [{"role": "user", "content": message}]}, config={ "callbacks": [handler], "configurable": {"thread_id": conversation_id}, }, ) return result["messages"][-1].content # -- main.py -- import uuid from app.agent_analytics import ai from app.assistant import run_turn def main(): conversation_id = str(uuid.uuid4()) try: print(run_turn(42, conversation_id, "Where is order A-1001?")) finally: ai.shutdown() if __name__ == "__main__": main() ``` ## 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 LangChain app by following https://amplitude.com/build/langchain-agent-analytics exactly. Install "amplitude-ai[langchain]>=1.20.1". Create app/agent_analytics.py (never name it amplitude_ai.py) that builds AmplitudeAI from os.environ["AMPLITUDE_AI_API_KEY"] and calls ai.agent() once at module level; never call ai.agent() per request. Keep the LangChain agent built once at import, with ChatOpenAI(model="gpt-4o-mini") as its model, and don't attach the Amplitude handler to it. For each complete invoke or ainvoke job, open agent.session(user_id, session_id) with a fresh job ID as session_id ("async with" in async code), create AmplitudeCallbackHandler(ai, user_id=...) inside the session, and pass it as config={"callbacks": [handler]} on that call, with the same job ID as the thread_id. Use the same user ID format as our product analytics, prefixed so it's at least 5 characters, and require an authenticated user for this example. Never invent a placeholder user ID. Call ai.shutdown() once on process exit, never per request. Don't hardcode keys. Report every file you changed. ``` ## What this unlocks - [Product analytics](https://amplitude.com/amplitude-analytics) funnels can run from `[Agent] Tool Call` for `lookup_order` to the `Support Ticket Created` event your app already sends, because both carry the same `user-42` user ID. - [Feature Experiment](https://amplitude.com/amplitude-experiment) can test `gpt-4o-mini` against another model with `[Agent] Cost USD` and `[Agent] Response Quality` as metrics. - [Guides and surveys](https://amplitude.com/guides-and-surveys) can reach users whose `[Agent] Session Record` shows `[Agent] Has User Friction` true. ## FAQ ### Does the callback replace LangChain? No. LangChain still calls the model and tools. The callback records their activity in Amplitude. ### When does this example close an agent session? After one complete invocation, including its model and tool calls. Use a fresh job ID for the next invocation. ## Next steps [Web frameworksHow to add agent analytics to a FastAPI backendServe this agent from an async route, flush from the lifespan, and link conversations to session replay.]()[Use caseHow to monitor agent tool callsFind the tools that fail, run slow, or never get called.]()[Web frameworksHow to add agent analytics to a Next.js appThe same setup for a TypeScript agent in Next.js route handlers.](https://amplitude.com/build/nextjs-agent-analytics) 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)[Instrument your agent with the AI SDK →](https://amplitude.com/docs/amplitude-ai/agent-analytics/instrument-ai-sdk)[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/langchain-agent-analytics.md)[llms.txt](/llms.txt) ## Ship the instrumentation, then read the data Use these events in funnels and retention charts alongside the product events your app already sends. [Create a free account](https://app.amplitude.com/signup)[Browse all Build tutorials](https://amplitude.com/build)