Agent Analytics SDK

This page is the developer reference for the Amplitude AI SDK. For the product-level setup overview, refer to Set up Agent Analytics. For the product concepts and how Amplitude uses the data, refer to the Agent Analytics overview and Analyze agent results.

The timeline below shows what your instrumentation produces. Click any event to inspect its shape and the call that emits it.

Prerequisites

• An Amplitude project with Agent Analytics enabled.
• The View Agent Analytics Objects permission. Admins grant access through role-based access control (RBAC).
• An agent codebase to instrument in Node.js or Python (or a runtime that can call the Amplitude HTTP API).
• The project's API key for the right data center. Agent Analytics runs in US and EU.

Install the SDK

Instrument this app with @amplitude/ai. Follow node_modules/@amplitude/ai/amplitude-ai.md

Paste this prompt into your AI coding agent (Cursor, Claude Code, Windsurf, GitHub Copilot, or Codex). The agent reads the linked instructions file, scans your codebase, finds every LLM call site and the session lifecycle, and instruments them.

Initialize the SDK

Initialize once at your application entry point and reuse the instance. The recommended pattern is a bootstrap module that exports ai plus wrapped provider clients.

// src/lib/amplitude.ts
import { AmplitudeAI, AIConfig, OpenAI } from "@amplitude/ai";

export const ai = new AmplitudeAI({
  apiKey: process.env.AMPLITUDE_AI_API_KEY!,
  config: new AIConfig({ contentMode: "full", redactPii: true }),
});

export const openai = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY!,
  amplitude: ai,
});

Import openai from this module instead of directly from 'openai'. Add more wrapped providers as needed.

To validate events during development without sending them, set dryRun: true (Node) or dry_run=True (Python) on AIConfig.

Configure the SDK

Pass an AIConfig to the AmplitudeAI constructor. All options are optional; the defaults work for most apps.

For redaction recipes (named replacements, custom scrubbers, international locales), refer to Choose a privacy mode.

Instrument an agent session

Wrap each agent invocation in a session. The session correlates every event (user message, model response, tool calls, spans) into a single record.

An agent session is one job the user hands the agent, from start to finish: the unit of work with a real outcome. Set the sessionId from an ID you already track, rather than inventing a new one:

• Chatbot or copilot: the conversation thread ID.
• Coding agent: the task or work-session ID.
• Support agent: the ticket ID.
• Voice agent: the call ID.
• Background or autonomous agent: the run or job ID.

import { ai, openai } from "@/lib/amplitude";

const agent = ai.agent("chat-handler", {
  description: "Customer support chatbot",
});

export async function POST(req: Request) {
  const { messages, userId } = await req.json();
  return agent.session({ userId }).run(async (s) => {
    s.trackUserMessage(messages[messages.length - 1].content);
    const response = await openai.chat.completions.create({
      model: "gpt-4o-mini",
      messages,
    });
    return Response.json(response);
  });
}

The Python SDK follows the same pattern with ai.agent(...).session(...). A session opened with run() closes when the callback returns. For sessions that span multiple requests, end them one of these ways:

• Close it explicitly (recommended): Call trackSessionEnd() (Node) or track_session_end() (Python) when the job finishes, such as a closed ticket or a completed run. Server-side evaluation runs on close. Closing is a completion marker, not a hard lock: events arriving after the close are still accepted and stored, but enrichment runs once per session, so late turns never appear in the session's signals, rollups, or Session Record.
• Let the idle timeout close it: The timeout defaults to 30 minutes of inactivity, measured from the last agent event received for the session. It's configurable per session with idleTimeoutMinutes (Node) or idle_timeout_minutes (Python). Raise it for jobs with long natural gaps, such as 240 for a support ticket worked over hours. Set it to -1 to raise the idle window to its 90-day maximum and waive the default 24-hour duration cap, so an explicit Session End is the only close in practice.

Where to set the idle override. Set the session parameter, and also include an idle_timeout_minutes key in the agent's context. Today the context route is what reliably reaches the server for sessions that never close explicitly; setting both guarantees the override takes effect and keeps working after the server-side fix ships.

const agent = ai.agent("support-bot", {
  context: { idle_timeout_minutes: 240 },   // reliable today
});
const session = agent.session({
  userId,
  sessionId: ticketId,
  idleTimeoutMinutes: 240,                  // the intended parameter
});

When the same user returns with a new goal, start a new session with a new sessionId rather than continuing the old one.

Two identity rules keep a single user from splitting into two:

• Don't pass a placeholder userId such as "anonymous", "", or a temporary ID. Omit the userId instead. Amplitude can't change a userId after it's set, so a placeholder creates a separate user that won't merge later.
• Reuse the same deviceId across a pre-account session. If your backend generates a new deviceId per request, the merge breaks. Read the deviceId from the Browser SDK and forward it.

Auto-instrument provider calls

The SDK offers two zero-code paths for capturing provider activity.

Provider wrappers

Wrap the provider client at construction time. The wrapper forwards calls to the underlying client and records request, response, tokens, latency, and cost.

import OpenAI from "openai";
const openaiWrapped = new OpenAI({ amplitude: ai });

If you can't change the construction site, use wrap(existingClient, ai) to instrument an existing client without modifying its creation.

Coverage varies by provider. All wrappers capture streaming, system prompts, and cost; the rest depends on what the provider's API exposes:

Bedrock model IDs such as us.anthropic.claude-3-5-sonnet are normalized for price lookup automatically.

patch()

Call patch({ amplitudeAI: ai }) once at startup for zero-code instrumentation. The SDK monkey-patches supported clients and auto-extracts [Agent] Tool Call events from message arrays for OpenAI Chat Completions, OpenAI Responses, and Anthropic Messages. Extracted tool calls land with latencyMs: 0 because execution timing isn't available through message inspection. Use tool() or trackToolCall() when you need real tool latency.

patch() requires an active session context. Patched calls that fire outside a session context are silently dropped: no event is emitted, no error is thrown, and the provider call itself is unaffected. This includes AMPLITUDE_AI_AUTO_PATCH=true with amplitude-ai-instrument, which patches at process start but establishes no ambient session. This is the most common cause of "I instrumented but see no events." To verify a patched setup, wrap one call in a session:

const ai = new AmplitudeAI({ apiKey: process.env.AMPLITUDE_AI_API_KEY! });
patch({ amplitudeAI: ai });

const agent = ai.agent("verify", { userId: "verify-user-1" });
await agent.session().run(async () => {
  await openai.chat.completions.create({
    model: "gpt-4o",
    messages: [{ role: "user", content: "Hello" }],
  });
}); // one [Agent] AI Response

For request-scoped work, the Express middleware establishes the session context automatically (refer to Framework notes).

Track tools

The tool() higher-order function wraps a tool function so the SDK records each call:

import { tool } from "@amplitude/ai";

const searchProducts = tool(searchDB, { name: "search_products" });

// Inside session.run, call as usual:
const result = await searchProducts(query);
// [Agent] Tool Call event emitted with duration, success, input/output

For inline tool calls or unsupported flows, use s.trackToolCall(name, latencyMs, success, { input, output }) directly.

Track spans

Spans wrap internal sub-operations such as vector lookups, reranks, guardrails, or any timed work that sits inside a turn. They emit [Agent] Span events and share the trace's identity.

import { observe } from "@amplitude/ai";

// As a higher-order function:
const runSubAgent = observe(
  async (prompt: string) => {
    return await subAgent.execute(prompt);
  },
  { name: "sub-agent-execution" },
);

// Or explicitly when you need error capture:
const start = Date.now();
try {
  const result = await subAgent.execute(prompt);
  s.trackSpan({
    name: "sub-agent-execution",
    latencyMs: Date.now() - start,
    inputState: { prompt: prompt.slice(0, 1000) },
    outputState: { response: result.slice(0, 1000) },
  });
} catch (e) {
  s.trackSpan({
    name: "sub-agent-execution",
    latencyMs: Date.now() - start,
    isError: true,
    errorType: (e as Error).name,
    errorMessage: (e as Error).message,
  });
  throw e;
}

Manual instrumentation

For custom flows or unsupported providers, use the manual methods on the session object directly. Each maps to a single [Agent] event type.

For AI responses that don't go through a wrapper (proxies, custom gateways), pass usage from the completion response:

s.trackAiMessage(completedMessage.content, "gpt-4o", "openai", latencyMs, {
  inputTokens: usage.prompt_tokens,
  outputTokens: usage.completion_tokens,
  totalTokens: usage.total_tokens,
});

Pass the canonical provider model id (gpt-4o-mini, claude-sonnet-4-20250514), not an internal gateway label, so cost auto-calculates correctly.

Add segmentation with context

Pass a context dictionary to ai.agent(...) to attach arbitrary segmentation dimensions to every event. The SDK serializes it to [Agent] Context, so you can segment AI sessions without registering new global properties.

const agent = ai.agent("support-bot", {
  context: {
    agent_type: "executor",
    experiment_variant: "reasoning-enabled",
    surface: "chat",
  },
});

These keys cover the most common segmentation needs:

Merge context across child agents

Child agents inherit the parent's context. Keys on the child override matching parent keys; parent keys the child doesn't set are preserved.

const parent = ai.agent("orchestrator", {
  context: { experiment_variant: "treatment", surface: "chat" },
});
const child = parent.child("researcher", {
  context: { agent_type: "retriever" },
});
// child context = { experiment_variant: "treatment", surface: "chat", agent_type: "retriever" }

Query context in Amplitude

[Agent] Context is a JSON string. To query individual keys:

• Derived properties: For frequently-used keys, create a derived event property (Data > Properties > Derived > New) that extracts the value permanently.
• Filters: Use [Agent] Context contains "key":"value" for string matching in chart filters.

Use multiple tenants

On a multi-tenant platform, create a tenant-scoped handle with ai.tenant(orgId, opts?) (Node) or ai.tenant(org_id, ...) (Python). Every agent created from the handle pre-binds customerOrgId, which lands as [Agent] Customer Org ID on each event, so you can segment usage by end customer without threading the org ID through every call.

const tenant = ai.tenant("org-456", { env: "production" });
const agent = tenant.agent("support-bot", { userId: "user-123" });
// agent.track* calls carry [Agent] Customer Org ID = "org-456"

Classify model tiers

The SDK infers a model tier from the model name and attaches it as [Agent] Model Tier on every [Agent] AI Response. Tiers let you compare cost and performance across model classes without listing every model.

Call inferModelTier() / infer_model_tier() to resolve a tier directly:

import { inferModelTier } from "@amplitude/ai";

inferModelTier("gpt-4o-mini"); // 'fast'
inferModelTier("claude-3.5-sonnet"); // 'standard'
inferModelTier("o1-preview"); // 'reasoning'

For custom or fine-tuned models the name can't classify, pass modelTier / model_tier on the AI-message call to override the inferred value:

s.trackAiMessage(response.content, "ft:gpt-4o:my-org:custom", "openai", latencyMs, {
  modelTier: "standard",
});

Track attachments

Pass an attachments array to the user-message call to record files sent with a message (images, PDFs, URLs). Each entry carries type, name, and size_bytes.

s.trackUserMessage("Analyze this document", {
  attachments: [
    { type: "image", name: "chart.png", size_bytes: 102400 },
    { type: "pdf", name: "report.pdf", size_bytes: 2048576 },
  ],
});

The SDK derives these properties from the array, recording attachment metadata only, never file content: [Agent] Has Attachments, [Agent] Attachment Types, [Agent] Attachment Count, [Agent] Total Attachment Size Bytes, and [Agent] Attachments. Attachments also apply to AI responses, such as model-generated images. Pass the same attachments option to the AI-message call.

Capture implicit feedback

Behavioral signals indicate whether a response met the user's need without requiring an explicit rating. Set these options on the relevant track calls; the SDK maps them to queryable quality properties.

// AI response the user copied (positive)
s.trackAiMessage("To create a funnel, go to...", "gpt-4o", "openai", latencyMs, { wasCopied: true });

// User regenerates (negative — first response fell short)
s.trackUserMessage("How do I create a funnel?", { isRegeneration: true });

// User edits and resubmits their prompt
s.trackUserMessage("How do I create a conversion funnel for signups?", {
  isEdit: true,
  editedMessageId: originalMsgId,
});

// User left after the first AI response
agent.trackSessionEnd({ sessionId: "sess-1", abandonmentTurn: 1 });

Import existing conversations

To backfill a full message history in one call, use trackConversation() (Node) or track_conversation() (Python). Pass an array of { role, content } messages; each becomes a [Agent] User Message or [Agent] AI Response, with turn IDs auto-incremented in order. system messages are skipped.

import { trackConversation } from "@amplitude/ai";
import * as amplitude from "@amplitude/analytics-node";

trackConversation({
  amplitude,
  userId: "user-123",
  sessionId: "sess-abc",
  agentId: "support-bot",
  messages: [
    { role: "user", content: "How do I reset my password?" },
    {
      role: "assistant",
      content: "Go to Settings > Security > Reset Password.",
      model: "gpt-4o",
      provider: "openai",
      latency_ms: 1200,
      input_tokens: 15,
      output_tokens: 42,
    },
    { role: "user", content: "Thanks, that worked!" },
  ],
});

Use this to import historical conversations or migrate data from external systems. The function accepts the same context fields as the individual tracking methods.

Send user feedback (scores)

Capture explicit user feedback, such as a thumbs up or down on a response or an optional rating, as a [Agent] Score event. Scores come only from your application; Amplitude's enrichment pipeline never generates them.

// Thumbs up/down on a specific AI response
ai.score({
  userId: "user-123",
  name: "user-feedback",
  value: 1.0,
  targetId: aiMessageId,
  targetType: "message",
  source: "user",
});

Use the name user-feedback for your binary thumbs control: that name has special semantics, overriding the detected negative feedback signal for the session. For the full pattern (in-session and post-request paths, CSAT scales), refer to the setup page's thumbs up / thumbs down section.

If you ingest events directly instead of using the SDK, send an [Agent] Score event with [Agent] Score Name set to your score name (for example, user-feedback).

Multi-agent architectures

Parent agents can delegate to child agents. Child agents inherit the parent's session, so all events stay correlated under one Session ID.

const orchestrator = ai.agent('shopping-agent', { description: 'Orchestrates shopping requests' });
const recipeAgent = orchestrator.child('recipe-agent', { description: 'Finds recipes' });

await orchestrator.session({ userId }).run(async (s) => {
  s.trackUserMessage(userInput);
  const result = await s.runAs(recipeAgent, async (cs) => {
    cs.trackUserMessage(delegatedQuery);
    return openai.chat.completions.create({ model: 'gpt-4o', messages: [...] });
  });
});

Wrap delegation calls with observe() or trackSpan if you want latency and error metrics on the dispatch itself, not only the child's LLM call.

Integration patterns

Single-request API endpoint

For a serverless function or one-shot endpoint, create the session inside the handler and flush before returning so the runtime doesn't freeze before events ship.

app.post("/chat", async (req, res) => {
  const agent = ai.agent("api-handler", { userId: req.userId });
  const result = await agent.session({ sessionId: req.sessionId }).run(async (s) => {
    s.trackUserMessage(req.body.message);
    const start = performance.now();
    const response = await openai.chat.completions.create({
      model: "gpt-4o",
      messages: req.body.messages,
    });
    s.trackAiMessage(
      response.choices[0].message.content ?? "",
      "gpt-4o",
      "openai",
      performance.now() - start,
      {
        inputTokens: response.usage?.prompt_tokens,
        outputTokens: response.usage?.completion_tokens,
      },
    );
    return response.choices[0].message.content;
  });
  await ai.flush();
  res.json({ response: result });
});

Long-lived session (chatbot)

For a multi-turn conversation, create the session once and reuse it across turns. Track a user/AI pair per turn; the session ends when run() returns or the idle timeout fires.

const agent = ai.agent("chatbot", { userId: "user-123", env: "production" });

await agent.session({ sessionId: conversationId }).run(async (s) => {
  s.trackUserMessage("What is Amplitude?");
  const r1 = await llm.chat("What is Amplitude?");
  s.trackAiMessage(r1.content, "gpt-4o", "openai", r1.latencyMs, {
    inputTokens: r1.usage.input,
    outputTokens: r1.usage.output,
  });

  s.trackUserMessage("How does it track events?");
  const r2 = await llm.chat("How does it track events?");
  s.trackAiMessage(r2.content, "gpt-4o", "openai", r2.latencyMs, {
    inputTokens: r2.usage.input,
    outputTokens: r2.usage.output,
  });
});

Multi-agent orchestration

When a parent agent delegates to specialized children, wrap each delegation in runAs() / arun_as(). Both manual tracking calls and provider wrappers inside the callback pick up the child's identity automatically. For the basic delegation shape, refer to Multi-agent architectures.

How runAs / arun_as works:

• Shares the parent session's sessionId, traceId, and turn counter.
• Sets agentId to the child and parentAgentId to the parent for the callback's duration.
• Suppresses auto user-message tracking, so internal role: "user" prompts in delegation calls don't create spurious user turns.
• Doesn't emit [Agent] Session End; the child runs inside the parent session, which emits one session end.
• Restores the parent context when the callback completes, even on error.
• Supports nesting: a child can runAs a grandchild.

Fan-out (parallel child calls, single user turn)

When one user turn triggers several parallel LLM calls, open a fresh trace with newTrace() / new_trace(), dispatch the children with Promise.all (Node) or asyncio.gather (Python), and emit a single AI response after they join. This keeps one trace, one user turn, and one AI response regardless of how many internal calls run.

await orchestrator.session({ sessionId }).run(async (s) => {
  s.newTrace();
  s.trackUserMessage("Generate plan from quiz results", { context: structuredState });

  const [a, b] = await Promise.all([
    s.runAs(scorer, () =>
      openai.chat.completions.create({ model: "gpt-4o", messages: scorerMessages }),
    ),
    s.runAs(matcher, () =>
      openai.chat.completions.create({ model: "gpt-4o", messages: matcherMessages }),
    ),
  ]);

  s.trackAiMessage(assemble(a, b), "gpt-4o", "openai", totalLatencyMs);
});

Stream responses

Streaming sessions must stay open until the stream is fully consumed. Closing the session before the stream finishes drops the AI response event.

// WRONG: session ends before stream is consumed
return agent.session({ userId }).run(async (s) => {
  const stream = await openai.chat.completions.create({
    model: "gpt-4o",
    messages,
    stream: true,
  });
  return new Response(stream.toReadableStream());
});

// CORRECT: session stays open until stream completes
return agent.session({ userId }).run(async (s) => {
  const stream = await openai.chat.completions.create({
    model: "gpt-4o",
    messages,
    stream: true,
  });
  const readable = stream.toReadableStream();
  const [passthrough, forClient] = readable.tee();
  const reader = passthrough.getReader();
  (async () => {
    while (!(await reader.read()).done) {}
  })();
  return new Response(forClient);
});

With the Vercel AI SDK, flush in the onFinish callback:

const result = await streamText({
  model: openai("gpt-4o"),
  messages,
  onFinish: async () => {
    await ai.flush();
  },
});

Link to the standard-analytics session

When a session crosses the network boundary, pass Amplitude IDs through request headers so server-side events join the user's standard-analytics session ($session_id). Pass the value as the session's browserSessionId field:

const browserSessionId = req.headers.get("x-amplitude-session-id");
const deviceId = req.headers.get("x-amplitude-device-id");
const session = agent.session({ userId, browserSessionId, deviceId });

For cross-service propagation between back-end services, use injectContext() on the outbound side and extractContext(headers) on the inbound side.

Propagate context across services

When one back-end service calls another, propagate the active identity and session so the downstream events join the same trace instead of starting a new one. On the outbound side, injectContext() serializes the active context (session ID, trace ID, user ID) into request headers. On the inbound side, extractContext(headers) reads them back.

// --- Service A (outbound) ---
import { injectContext } from "@amplitude/ai";

await agent.session({ userId, sessionId }).run(async (s) => {
  s.trackUserMessage(message);
  const headers = injectContext({ "content-type": "application/json" });
  await fetch("https://service-b/internal/enrich", {
    method: "POST",
    headers,
    body: JSON.stringify({ message }),
  });
});

// --- Service B (inbound) ---
import { randomUUID } from "node:crypto";
import { extractContext, runWithContextAsync, SessionContext } from "@amplitude/ai";

export async function POST(req: Request) {
  const extracted = extractContext(Object.fromEntries(req.headers));
  const ctx = new SessionContext({
    sessionId: extracted.sessionId ?? randomUUID(),
    traceId: extracted.traceId ?? null,
    userId: extracted.userId ?? null,
  });
  return runWithContextAsync(ctx, async () => {
    await handleEnrichment(req);
  });
}

injectContext() returns a new headers object and never mutates the original. If no session is active, it returns the headers unchanged, so it's safe to call unconditionally.

Supported providers and frameworks

Providers with native wrappers: OpenAI (Chat Completions + Responses), Anthropic, Azure OpenAI, Gemini (@google/generative-ai), Google Gen AI (@google/genai), Mistral, Bedrock (Converse APIs).

Agent frameworks with first-party integrations: LangChain, LlamaIndex, OpenAI Agents SDK, Anthropic Tool Use, Claude Agent SDK (ClaudeAgentSDKTracker), Anthropic Managed Agents, CrewAI (Python only).

Framework integrations

The integrations below bridge an agent framework's own callback or tracing system into Agent Analytics. Each takes the ai instance plus identity fields, then hooks into the framework. CrewAI is Python-only; in Node, AmplitudeCrewAIHooks throws by design. Use the LangChain or OpenTelemetry path instead.

LangChain

Pass an AmplitudeCallbackHandler to LangChain's callbacks.

import { AmplitudeCallbackHandler } from "@amplitude/ai";

const handler = new AmplitudeCallbackHandler({ amplitudeAI: ai, userId: "user-123", sessionId: "sess-1" });
// Pass handler to any LangChain runnable via { callbacks: [handler] }

LlamaIndex

import { AmplitudeLlamaIndexHandler } from "@amplitude/ai";

const handler = new AmplitudeLlamaIndexHandler({ amplitudeAI: ai, userId: "user-123", sessionId: "sess-1" });

OpenAI Agents SDK

Register an AmplitudeTracingProcessor as a tracing processor.

import { AmplitudeTracingProcessor } from "@amplitude/ai";

const processor = new AmplitudeTracingProcessor({ amplitudeAI: ai, userId: "user-123", sessionId: "sess-1" });
// Register with the OpenAI Agents SDK trace provider.

Anthropic Tool Use

AmplitudeToolLoop runs Anthropic's multi-turn tool_use loop and tracks each AI response and tool call.

import { AmplitudeToolLoop } from "@amplitude/ai";

const loop = new AmplitudeToolLoop({ amplitudeAI: ai, userId: "user-123", sessionId: "sess-1" });
await loop.run({ client, model: "claude-sonnet-4-20250514", messages, tools, toolExecutor });

OpenTelemetry attribute mapping

If a framework already emits OpenTelemetry GenAI spans, the SDK maps them onto [Agent] properties. For how to enable this, including the span-first enableOtel() / enable_otel() path and the manual AmplitudeGenAIExporter / AmplitudeAgentExporter exporters, refer to Ingest OpenTelemetry spans. The mapping the exporter applies:

Some signals have no OTEL equivalent and require the native provider wrappers: reasoning content and tokens, TTFB, streaming detection, implicit feedback, file attachments, and event-graph linking through [Agent] Parent Message ID. You can run OTEL and a native wrapper together for the same call. The SDK de-duplicates, so no double events emit.

Provider-specific notes

Vercel AI SDK

Provider wrappers instrument the underlying SDK (openai), not the Vercel abstraction. If only @ai-sdk/openai is present, either add openai as a direct dependency or fall back to patch(). For streaming responses, use onFinish to call await ai.flush() (refer to Stream responses).

Claude Agent SDK

Use ClaudeAgentSDKTracker from @amplitude/ai/integrations/claude-agent-sdk. Two fields are required for the events to be useful: agentId on ai.agent() (identifies the AI feature in the LLM Usage Application Registry), and userId + sessionId on agent.session() (ties events into a single interaction).

import { AmplitudeAI } from "@amplitude/ai";
import { ClaudeAgentSDKTracker } from "@amplitude/ai/integrations/claude-agent-sdk";
import { query } from "@anthropic-ai/claude-agent-sdk";

const ai = new AmplitudeAI({ apiKey: process.env.AMPLITUDE_AI_API_KEY! });
const agent = ai.agent({ agentId: "code-reviewer" });
const tracker = new ClaudeAgentSDKTracker();

await agent.session({ userId: "u1", sessionId: "sess-abc" }).run(async (s) => {
  for await (const message of query({
    prompt: "Analyze this codebase",
    options: { hooks: tracker.hooks(s) },
  })) {
    tracker.process(s, message);
  }
});

tracker.hooks(session) returns PreToolUse / PostToolUse hooks with precise tool latency. tracker.process(session, message) processes the message stream for AI responses and user messages.

Anthropic Managed Agents

Provider wrappers don't work, because LLM calls happen in Anthropic's cloud, not your code. Use manual tracking and poll client.beta.sessions.events.list(). Map event types to SDK methods:

Deduplicate events across polls, because events.list() returns previously-seen events:

const seenIds = new Set<string>(savedState.seenIds);
for (const event of response.data) {
  if (seenIds.has(event.id)) continue;
  seenIds.add(event.id);
  // track event
}

Measure latency as wall-clock time between session.status_running and the event's processed_at, not poll round-trip. events.list() doesn't include usage or token counts, so cost tracking requires the Anthropic Admin API.

OpenAI Assistants API

Provider wrappers don't auto-instrument the Assistants API (async / polling-based). Use manual tracking: trackUserMessage() when creating a message, trackAiMessage() when polling completion events.

MCP servers

The MCP protocol doesn't pass the originating user prompt to tools, so MCP servers can't capture it. Add an optional rationale parameter to each tool so the LLM can self-explain its intent and you keep usable session content.

Framework notes

Next.js (App Router)

Initialize the SDK in a server-side module, never a client component. Add @amplitude/ai to serverExternalPackages in next.config.ts. Wrap session creation inside each route handler; in serverless deployments call await ai.flush() before the handler returns so the runtime doesn't freeze before events ship.

Express / Fastify / Hono

Use the bundled middleware to attach ai to every request:

import { createAmplitudeAIMiddleware } from "@amplitude/ai";

app.use(
  createAmplitudeAIMiddleware({
    amplitudeAI: ai,
    userIdResolver: (req) => req.headers["x-user-id"] ?? null,
  }),
);

Edge runtimes and Cloudflare Workers

@amplitude/ai cannot bundle in Cloudflare Workers. The SDK depends on node:async_hooks, node:module, and node:crypto. Workers Builds rejects the upload even with nodejs_compat_v2 enabled. @amplitude/analytics-node is also incompatible (depends on Node's http).

The only safe import is import type { ... } from '@amplitude/ai/types', which is erased at compile time. For runtime tracking, use a fetch-based transport that constructs [Agent] events directly:

import type { AmplitudeClientLike, AmplitudeEvent } from "@amplitude/ai/types";

class FetchAmplitudeClient implements AmplitudeClientLike {
  private _apiKey: string;
  private _buffer: AmplitudeEvent[] = [];
  constructor(apiKey: string) {
    this._apiKey = apiKey;
  }
  track(event: AmplitudeEvent): void {
    this._buffer.push(event);
  }
  async flush(): Promise<void> {
    if (!this._buffer.length) return;
    const events = this._buffer.splice(0);
    try {
      const resp = await fetch("https://api2.amplitude.com/2/httpapi", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ api_key: this._apiKey, events }),
      });
      if (!resp.ok) console.error(`[Amplitude] Flush failed: ${resp.status}`);
    } catch (err) {
      console.error(`[Amplitude] Flush error: ${(err as Error).message}`);
    }
  }
}

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
    if (env.AMPLITUDE_TRACKING_DISABLED) return handleRequest(request, env);
    const transport = new FetchAmplitudeClient(env.AMPLITUDE_API_KEY);
    transport.track({
      event_type: "[Agent] User Message",
      user_id: userId,
      event_properties: {
        "[Agent] Session ID": sessionId,
        "[Agent] Agent ID": "my-agent",
        $llm_message: { text: content },
      },
    });
    // After the LLM call completes:
    transport.track({
      event_type: "[Agent] AI Response",
      user_id: userId,
      event_properties: {
        "[Agent] Session ID": sessionId,
        "[Agent] Agent ID": "my-agent",
        "[Agent] Model Name": model,
        "[Agent] Provider": "anthropic",
        "[Agent] Latency Ms": latencyMs,
        $llm_message: { text: responseText },
      },
    });
    // Non-blocking flush so events ship before the isolate terminates
    ctx.waitUntil(transport.flush());
    return new Response("ok");
  },
};

Construct FetchAmplitudeClient per-request to avoid buffer leakage between requests. Use crypto.randomUUID() for event insert_id dedup, and gate tracking behind an AMPLITUDE_TRACKING_DISABLED env var to disable it.

Run in serverless environments

The SDK auto-detects serverless platforms (Vercel, AWS Lambda, Netlify, Google Cloud Functions, Azure Functions, Cloudflare Pages) from their environment variables. When it detects one, session.run() flushes pending events before the promise resolves, so you don't need an explicit ai.flush(). In a long-running server, it skips the per-session flush and lets the analytics client batch normally.

Control this per session with the autoFlush option (Node) or auto_flush (Python). Leave it unset to use auto-detection, set it to true to always flush on session exit, or false to never flush.

If you track events outside session.run(), flush before the handler returns or the runtime can freeze the process with events still buffered.

ai.flush() and ai.shutdown() serve different lifecycles:

• ai.flush() sends buffered events now and keeps the SDK running. Use it in serverless handlers and API endpoints to guarantee delivery before responding.
• ai.shutdown() flushes and then closes the underlying analytics client. Call it once on process exit, such as a SIGTERM handler. It only closes the client when you created it through apiKey; if you passed your own instance, you own its lifecycle.

process.on("SIGTERM", () => {
  ai.shutdown();
  process.exit(0);
});

Cloudflare Workers (edge isolates) aren't a supported serverless target; the full SDK can't bundle into a Worker. Refer to Edge runtimes and Cloudflare Workers for the fetch-based transport.

Ingest OpenTelemetry spans

The exporters below run in-process, inside a Node or Python app that already uses this SDK. If your spans already reach an OpenTelemetry Collector, you can skip the SDK entirely and point the Collector at Amplitude's OTLP endpoint instead. Refer to Send OpenTelemetry traces directly.

For stacks that already emit OpenTelemetry GenAI spans (OpenLIT, Traceloop, OpenAI's OTel instrumentation), map them into [Agent] events:

• AmplitudeGenAIExporter (inbound, production-ready): ingests GenAI semantic-convention spans and emits [Agent] events. Ignores non-GenAI spans, so it's safe in a mixed pipeline.
• AmplitudeAgentExporter (outbound, experimental): converts Amplitude events into flat OTel spans for forwarding to other backends. Doesn't preserve trace hierarchy.

Choose a privacy mode

Set contentMode on AIConfig. Every content-emitting channel in the SDK routes through a single privacy gate, so the mode applies uniformly across direct SDK calls, provider wrappers, observe() spans, OTel exception recording, and any framework integrations.

• full (default): captures prompt and response text. redactPii: true is on by default and scrubs emails, phone numbers, SSNs, credit card numbers, IP addresses, and base64-encoded image data before events leave the process. The SDK tunes phone and SSN detection for US formats; add customRedactionPatterns or customRedactionFn for international locales.
• metadata_only: no conversation text, exception message, stack trace, or system instruction reaches Amplitude. You still get token counts, latency, model, cost, and session grouping. Use for sensitive or regulated data.
• customer_enriched: no text by default. Send pre-scored summaries through trackSessionEnrichment(). Designed for teams with existing evaluation stacks.

metadata_only gates the following channels. This is the customer-visible expression of the single-gate contract:

• Message text ($llm_message.text) on User Message and AI Response
• Tool inputs and outputs ([Agent] Tool Input, [Agent] Tool Output)
• System prompts ([Agent] System Prompt)
• Exception messages and stack traces recorded by observe() and OTel span error paths
• Framework-emitted content attributes such as gen_ai.system_instructions and gen_ai.input.messages / gen_ai.output.messages
• Score comments ([Agent] Comment)

For managed-agent architectures, prefer full with redactPii: true. The managed API already stores message content server-side, so metadata_only adds no privacy benefit.

Provide your own session enrichments

In customer_enriched mode, the SDK sends no message text. You run your own evaluation pipeline and ship the results back as structured session-level enrichments. Use this when compliance requires zero-content transmission, or when your eval logic goes beyond Amplitude's built-in server-side enrichment.

Build a SessionEnrichments object and send it with trackSessionEnrichment() (Node) or track_session_enrichment() (Python). The enrichment lands as an [Agent] Session Enrichment event, serialized into the [Agent] Enrichments property; the same fields also attach to [Agent] Session End when you set enrichments before the session closes.

The substantive fields on a SessionEnrichments object:

import {
  AmplitudeAI,
  AIConfig,
  ContentMode,
  SessionEnrichments,
  RubricScore,
  TopicClassification,
} from "@amplitude/ai";

const ai = new AmplitudeAI({
  apiKey: process.env.AMPLITUDE_AI_API_KEY!,
  config: new AIConfig({ contentMode: ContentMode.CUSTOMER_ENRICHED }),
});
const agent = ai.agent("support-bot", { agentVersion: "2.1.0" });

// 1. Run the conversation — no content is sent, only metadata.
const { sessionId } = await agent.session({ userId: "user-42" }).run(async (s) => {
  s.trackUserMessage("Why was I charged twice?");
  s.trackAiMessage(aiResponse.content, "gpt-4o", "openai", latencyMs);
  return { sessionId: s.sessionId };
});

// 2. Score the raw messages with your own pipeline.
const evalResults = await myEvalPipeline(conversationHistory);

// 3. Ship the enrichments back to Amplitude.
const enrichments = new SessionEnrichments({
  qualityScore: evalResults.quality,
  sentimentScore: evalResults.sentiment,
  overallOutcome: evalResults.outcome,
  topicClassifications: {
    billing: new TopicClassification({ topic: "billing-dispute", confidence: 0.92 }),
  },
  rubricScores: [new RubricScore({ name: "accuracy", score: 4, maxScore: 5 })],
  customMetadata: { eval_model: "gpt-4o-judge-v2" },
});

agent.trackSessionEnrichment(enrichments, { sessionId });

This produces the same event properties as Amplitude's built-in enrichment (topics, rubrics, outcomes, message labels), sourced from your pipeline instead.

Message labels

Message labels are key-value pairs attached to individual messages for filtering and segmentation, such as routing tags (flow, surface), classifier output (intent, sentiment), or business context (tier, plan). They emit as [Agent] Message Labels on the message event. Attach them two ways:

• Inline, at tracking time, by passing labels to trackUserMessage() / track_user_message().
• Retrospectively, when classifier results arrive after the session, through SessionEnrichments.messageLabels keyed by the message ID returned from each tracking call.

Manage cost and tokens

s.trackAiMessage(...) auto-calculates [Agent] Cost USD from the model name and token counts through the bundled Pydantic genai-prices catalog.

Three things prevent automatic pricing:

1. Unrecognized model name. Vertex AI aliases like claude-sonnet-4-6 won't match the canonical claude-sonnet-4-20250514. Internal gateway labels won't resolve. Brand-new models may not yet be in genai-prices. Pass the canonical provider id, or set totalCostUsd explicitly to override.
2. Fine-tuned models. ft: model names never auto-price; pass totalCostUsd explicitly for fine-tuned and custom models.
3. Incorrect inputTokens with prompt caching. The SDK expects inputTokens to be cache-inclusive (cached tokens are a subset, never additive). Provider conventions differ:

Provider-name quirks to know:

• Gemini → google (Node). The Node SDK maps gemini to google internally for price lookup. Passing defaultProvider: 'gemini' still resolves cost, but if you inspect the raw catalog and expect to see gemini, look for google instead.
• Bedrock candidate generation. The lookup strips dot-separated prefixes progressively (region.vendor.model → vendor.model → model) and tries regional. and global. variants, which is why new AWS regions and Bedrock vendors work automatically without a catalog update.
• Local Python rate overrides. The Python SDK carries local rate tables in costs.py that fill gaps in the upstream catalog: gpt-5.6 (upstream misses the cache-write premium and long-context tier), Fireworks fast routers (billed at the fast tier even when the API response reports the base model ID), and a _MODEL_PRICE_ALIASES map that points, for example, gpt-5.4 at gpt-5.2 pricing when upstream lags an OpenAI release.

The built-in Anthropic, Bedrock, and Gemini wrappers handle this normalization for you. Manual trackAiMessage callers need to handle it themselves. Pass cacheReadTokens / cacheCreationTokens separately so the SDK applies the differential pricing.

When you need to compute cost yourself, call calculateCost({ modelName, inputTokens, outputTokens, cacheReadInputTokens, cacheCreationInputTokens }) and pass the result as totalCostUsd.

Rules for calculateCost():

• inputTokens is the total input, including cache reads and cache creations. For Anthropic, pass input_tokens + cache_read_input_tokens + cache_creation_input_tokens. For OpenAI, prompt_tokens already includes cached tokens, so pass it directly.
• outputTokens is the total output, including reasoning tokens where the provider emits them. OpenAI's completion_tokens already includes reasoning; don't add it separately.
• cacheReadInputTokens and cacheCreationInputTokens are subsets of inputTokens, used only to apply the differential pricing rate.
• The reasoningTokens parameter is deprecated and ignored. It's kept for backward compatibility; passing it separately over-costs the request.

Debugging missing costs

When the SDK can't calculate a cost, it logs one warning per unique (model, provider, reason) tuple, capped at 100 unique tuples per process. Grep your logs for Unable to calculate cost for model= to see which lookups failed and why.

Report run cost for non-conversational agents

Provider wrappers emit [Agent] Cost USD automatically on every [Agent] AI Response. Agents that finish a run without a final text response, such as batch jobs and artifact generators, may never emit an [Agent] AI Response, so their cost never lands. Call trackRunCost() at the end of the run to emit the cost the per-call events didn't cover:

s.trackRunCost(totalCostUsd, inputTokens, outputTokens, model, "openai", {
  latencyMs,
  content: "[Artifact: batch run]", // optional; omit for a cost-only event
});

s.track_run_cost(
    total_cost_usd,
    input_tokens,
    output_tokens,
    model,
    "openai",
    latency_ms=latency_ms,
    content="[Artifact: batch run]",  # optional; omit for a cost-only event
)

trackRunCost() emits delta cost only, so it reconciles with any per-call costs already tracked in the run. It defaults to empty content, so omit content for a cost-only balancing event, or pass a short string when you want a bubble in the session viewer. Attach cost to run completion this way rather than to [Agent] Session End, which is lifecycle-only and doesn't carry cost fields.

Keep pricing data current

Cost relies on the bundled genai-prices catalog, so a newly released model can report [Agent] Cost USD of 0 until the catalog updates. To fetch the latest prices at runtime, opt in at startup with enableLivePriceUpdates() (Node) or enable_live_price_updates() (Python).

• Default state: off. You must opt in.
• Default refresh interval: 1 hour (3_600_000 ms on Node, 3600 s on Python). Configurable via the first argument.
• Upstream endpoint: public raw.githubusercontent.com, the Pydantic genai-prices GitHub repo. Not an Amplitude-owned endpoint.
• Failure mode: silent. Network errors are swallowed; the bundled catalog stays in use.
• Idempotent: a second call is a no-op.
• How to know it's working: no debug flag or status output. The only signal is [Agent] Cost USD starting to populate for a model that was previously missing cost.

Enable it when: you've upgraded to a newly released model that isn't in the bundled catalog yet. Without the toggle, [Agent] Cost USD stays omitted for that model until the SDK version bumps.

Don't enable it when: your process runs air-gapped, or your environment blocks outbound HTTPS to raw.githubusercontent.com. There's no benefit and every startup pays a failed network round-trip.

Capture deploy metadata

The SDK reads deploy-identifying environment variables at startup and stamps them onto every [Agent] event. This lets you filter charts by SHA or release without threading the value through your code.

The repo URL is opt-in via env var only. Earlier SDK releases auto-captured the repo from git metadata; that was removed. If [Agent] Git Repo went empty after an SDK upgrade, set AMPLITUDE_GIT_REPO explicitly.

Track semantic cache hits

When you serve a full response from your own semantic or response cache, pass wasCached: true (Node) or was_cached=True (Python) on the AI-message call. It maps to [Agent] Was Cached, distinct from token-level prompt caching, so you can chart cache-hit rate and the cost it saves.

Shape message content

The first argument to trackUserMessage becomes $llm_message.text on [Agent] User Message. This is what session lists, segmentation, and enrichment treat as "what the user said". Two practical rules:

Do pass a short natural-language line as the message body. For example, the real prompt, or a canonical summary for headless jobs:

s.trackUserMessage(
  "Summarize the attached design doc and list open questions",
  {
    context: { structuredPayload: payloadRecord },
  },
);

Don't pass large JSON blobs as the message body. The product uses the JSON as the session title and breaks down charts by raw JSON:

// Session label becomes the JSON
s.trackUserMessage(JSON.stringify(payloadRecord));

Put structured segmentation dimensions in the context option (becomes [Agent] Context JSON, queryable in charts). For server-side enrichment to reason over structured data, also keep essential facts in content. Enrichments derive eval input primarily from turn text, not from [Agent] Context.

Instrument without the SDK

For unsupported runtimes (Java, Go, Ruby, edge environments), send events to the Amplitude HTTP API directly:

curl -X POST https://api2.amplitude.com/2/httpapi \
  -H 'Content-Type: application/json' \
  -d '{
    "api_key": "YOUR_API_KEY",
    "events": [{
      "event_type": "[Agent] User Message",
      "user_id": "user-123",
      "event_properties": {
        "[Agent] Session ID": "sess-abc",
        "[Agent] Agent ID": "support-chatbot",
        "$llm_message": { "text": "How do I cancel my subscription?" }
      }
    }]
  }'

Use $llm_message.text for message content (the ingestion pipeline reads this property for interaction text). For the full property reference and event JSON examples, refer to the Agent Analytics taxonomy.

When you send events directly, you're responsible for what the SDK otherwise handles:

Verify your data

Run the doctor to validate env vars, installed dependencies, and the event-pipeline connection:

npx amplitude-ai doctor

Default behavior reads AMPLITUDE_AI_API_KEY. When your app names the key differently, point the doctor at the right variable with -key-env:

npx amplitude-ai doctor -key-env MY_KEY_NAME

If you pass -key-env without a value, the doctor errors with a usage message rather than silently falling back to the default. This rule only applies when the flag is present.

Then confirm events land in Amplitude:

1. Open the project's Live Events stream.
2. Send a test session from the instrumented code.
3. Within seconds, an [Agent] AI Response event should appear with these properties populated:
   • [Agent] Session ID, [Agent] Agent ID
   • [Agent] Model Name, [Agent] Provider
   • [Agent] Latency Ms
   • [Agent] Input Tokens, [Agent] Output Tokens
   • [Agent] Cost USD

Local verification with summary()

Before deploying, use MockAmplitudeAI.summary() to get a fill-rate report of all captured events. It checks eight verification gates and flags gaps before data reaches Amplitude.

import { AIConfig } from "@amplitude/ai";
import { MockAmplitudeAI } from "@amplitude/ai/testing";

const mock = new MockAmplitudeAI(new AIConfig({ contentMode: "full" }));
const agent = mock.agent("test-agent", { userId: "u1" });

await agent.session({ sessionId: "s1" }).run(async (s) => {
  s.trackUserMessage("hello");
  s.trackAiMessage("response", "gpt-4o-mini", "openai", 150);
});

console.log(mock.summary());

The summary output looks like:

Agent Analytics fill-rate report
================================
Events captured: 2
  [Agent] User Message:  1
  [Agent] AI Response:   1

Verification gates (8/8 passing):
  ✓ user_id or device_id present
  ✓ [Agent] Session ID present
  ✓ [Agent] Agent ID present
  ✓ [Agent] Model Name present
  ✓ [Agent] Provider present
  ✓ [Agent] Latency Ms > 0
  ✓ [Agent] Input Tokens > 0
  ✓ [Agent] Output Tokens > 0
  ✓ [Agent] Cost USD > 0

Fixing common issues:

Test against a mock client

For CI, use MockAmplitudeAI from @amplitude/ai/testing to assert your events emit correctly:

import { AIConfig } from "@amplitude/ai";
import { MockAmplitudeAI } from "@amplitude/ai/testing";

const mock = new MockAmplitudeAI(new AIConfig({ contentMode: "full" }));
const agent = mock.agent("test-agent", { userId: "u1" });

await agent.session({ sessionId: "s1" }).run(async (s) => {
  s.trackUserMessage("hello");
  s.trackAiMessage("response", "gpt-4o-mini", "openai", 150);
});

mock.assertEventTracked("[Agent] User Message", { userId: "u1" });
mock.assertSessionClosed("s1");

// Data quality gate: every AI Response must carry the eight verification fields
for (const e of mock.eventsOfType("[Agent] AI Response")) {
  const p = e.event_properties ?? {};
  expect(e.user_id || e.device_id).toBeTruthy();
  expect(p["[Agent] Session ID"]).toBeTruthy();
  expect(p["[Agent] Model Name"]).toBeTruthy();
  expect(p["[Agent] Provider"]).toBeTruthy();
  expect(p["[Agent] Latency Ms"]).toBeGreaterThan(0);
  expect(p["[Agent] Input Tokens"]).toBeGreaterThan(0);
  expect(p["[Agent] Output Tokens"]).toBeGreaterThan(0);
  expect(p["[Agent] Cost USD"]).toBeGreaterThan(0);
}

Keep this test in CI to catch silent instrumentation regressions such as bad model names or missing token counts produce broken dashboards without throwing at runtime.

Reliability and error handling

Instrumentation can't take your application down:

• Tracking calls never throw. Every track* method catches and logs its own errors internally. A serialization bug or a bad field can't interrupt your agent's request path.
• The SDK buffers and retries events. The underlying @amplitude/analytics-node client batches events and retries failed sends from its transport layer.
• Failures degrade gracefully. If Amplitude is unreachable, the SDK drops events silently after exhausting retries. Your application keeps operating.

For development, set validate: true (Node) or validate=True (Python) on AIConfig to surface missing required fields, such as userId or sessionId, early. Validation errors throw ValidationError so you can catch them in tests before they reach production. Combine with dryRun / dry_run for the strictest CI checking.

Auto-instrument and CLI tools

To instrument without editing any call sites, auto-patch supported providers at process start. This is the fastest way to confirm the SDK is wired up; for the full event model (user messages, sessions, scores), use agents and sessions as shown in Initialize the SDK.

# Wrapper command
AMPLITUDE_AI_API_KEY=xxx AMPLITUDE_AI_AUTO_PATCH=true amplitude-ai-instrument node app.js

# Or Node's ESM preload flag directly
AMPLITUDE_AI_API_KEY=xxx AMPLITUDE_AI_AUTO_PATCH=true node --import @amplitude/ai/register app.js

Both runtimes read the same environment variables:

To inspect the environment without running your app, use amplitude-ai status. It prints the installed SDK version, the provider packages it detects, and the current environment-variable configuration. To validate dependencies and the event-pipeline connection, refer to Verify your data for the doctor command.

Register the event schema in your data catalog

The SDK ships a CLI that registers all [Agent] event types and their properties in Amplitude's Data Catalog, so events arrive documented with descriptions, types, and required flags instead of being inferred from ingestion.

Prerequisites: a plan with Taxonomy API access, and the project's API key and secret key from Settings > Projects.

# Print commands with your keys
npx amplitude-ai-register-catalog --api-key YOUR_KEY --secret-key YOUR_SECRET

# Execute immediately
npx amplitude-ai-register-catalog --api-key YOUR_KEY --secret-key YOUR_SECRET | bash

# EU data residency
npx amplitude-ai-register-catalog --api-key YOUR_KEY --secret-key YOUR_SECRET --eu | bash

The commands are idempotent: they create missing events and properties and update existing ones, so it's safe to re-run after an SDK upgrade adds new fields.

Debug and dry-run

Two AIConfig flags help you inspect events locally, each with an environment-variable equivalent for use with auto-instrumentation.

debug: true (Node) / debug=True (Python) logs a one-line summary of every event to stderr and still sends events to Amplitude:

[amplitude-ai] [Agent] AI Response | user=user-123 session=sess-abc agent=my-agent model=gpt-4o latency=1203ms tokens=150→847 cost=$0.0042

dryRun: true (Node) / dry_run=True (Python) logs the full event JSON to stderr and never transmits anything. Use it to validate event shape in local development and CI without a live API key. With auto-instrumentation, set AMPLITUDE_AI_DEBUG=true on the command instead.

Troubleshooting

API reference

Core classes

Session tracking methods

Higher-order functions

Other APIs