---
title: Amplitude AI SDK reference
description: "Developer reference for the Amplitude AI SDK: install, instrument, and verify that your agent traffic lands in Amplitude."
product: general
lang: en
token_estimate: 23682
---
# Amplitude AI SDK reference

> For AI agents: a documentation index is available at [/docs/llms.txt](/docs/llms.txt). Append `.md` to any page URL for markdown, or send `Accept: text/markdown`.

This page is the developer reference for the Amplitude AI SDK. For the product-level setup overview, refer to [Set up Agent Analytics](https://amplitude.com/docs/amplitude-ai/agent-analytics/setup). For the product concepts and how Amplitude uses the data, refer to the [Agent Analytics overview](https://amplitude.com/docs/amplitude-ai/agent-analytics/overview) and [Analyze agent results](https://amplitude.com/docs/amplitude-ai/agent-analytics/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)](https://amplitude.com/docs/amplitude-ai/agent-analytics/overview#manage-access-with-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

#### Node

```bash
npm install @amplitude/ai @amplitude/analytics-node
```

#### Python

```bash
python3 -m pip install amplitude-ai
```

To have your AI coding agent install and instrument the SDK for you, refer to [Instrument your agent with the AI SDK](https://amplitude.com/docs/amplitude-ai/agent-analytics/instrument-ai-sdk).

The packages are [`@amplitude/ai` on npm](https://www.npmjs.com/package/@amplitude/ai) and [`amplitude-ai` on PyPI](https://pypi.org/project/amplitude-ai/). For what changed in each version, refer to the [Node](https://amplitude.com/docs/sdks/agent-analytics/sdk-release-notes) and [Python](https://amplitude.com/docs/sdks/agent-analytics/sdk-python-release-notes) release notes.

The Python package installs provider and framework support as extras, so you only install what you use:

| Extra | Installs support for |
| --- | --- |
| `openai`, `azure` | OpenAI and Azure OpenAI |
| `anthropic` | Anthropic |
| `gemini` | Google Gen AI (`google-genai`) |
| `bedrock` | Amazon Bedrock (`boto3`) |
| `mistral` | Mistral |
| `langchain`, `llamaindex`, `openai-agents`, `crewai` | The matching framework integration |
| `fastapi` | The FastAPI and Starlette middleware |
| `otel` | OpenTelemetry export |
| `tokens` | Local token counting (`tiktoken`) |
| `mcp` | The SDK's MCP server |
| `all` | Every extra in this table |

```bash
python3 -m pip install "amplitude-ai[openai,fastapi]"
```

In Node, install the provider package you use, such as `openai` or `@anthropic-ai/sdk`, next to `@amplitude/ai`. Provider packages are optional peer dependencies.

## 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.

#### Node

```typescript
// 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,
});
```

#### Python

```python
# src/lib/amplitude.py
import os

from amplitude_ai import AmplitudeAI, AIConfig

ai = AmplitudeAI(
    api_key=os.environ['AMPLITUDE_AI_API_KEY'],
    config=AIConfig(content_mode='full', redact_pii=True),
)
```

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`. To inspect the sanitized upload body, use [review mode](#review-outbound-traffic).

## Configure the SDK

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

#### Node

| Option | Description |
| --- | --- |
| `contentMode` | `'full'` (default), `'metadata_only'`, or `'customer_enriched'`. Controls what message content reaches Amplitude. Refer to [Choose a privacy mode](#choose-a-privacy-mode). |
| `redactPii` | Scrub emails, phone numbers, SSNs, credit card numbers, and IP addresses from tracked content before events leave the process. Defaults to `true`. Set to `false` to opt out. |
| `customRedactionPatterns` | Additional redaction patterns. Accepts regex strings (replaced with `[REDACTED]`) or `{ pattern, replacement }` objects for named labels. |
| `customRedactionFn` | A `(text) => string` callback for custom redaction logic (for example, an NER library). Runs after all regex-based redaction. |
| `debug` | Log every tracked event to stderr. |
| `dryRun` | Build and log events without sending them to Amplitude. Use during development. |
| `reviewMode` | Record each post-redaction event and print a sanitized upload preview without sending it. Defaults to `false`. Refer to [Review outbound traffic](#review-outbound-traffic). |
| `validate` | Enforce strict validation of required fields. |
| `onEventCallback` | A `(event, statusCode, message) => void` callback invoked exactly once per tracked event, from the delivery path. |
| `propagateContext` | Enable cross-service context propagation. Refer to [Propagate context across services](#propagate-context-across-services). |
| `strictCost` | Throw `CostCalculationError` when a response has tokens but the SDK can't calculate its cost. Use in development and CI to catch unpriced model names. Defaults to `false`. |
| `captureStackTrace` | Attach `error.stack` to error events as `[Agent] Stack Trace`. Stack traces can contain file paths and internal details, so this defaults to `false`. |

#### Python

| Option | Description |
| --- | --- |
| `content_mode` | `'full'` (default), `'metadata_only'`, or `'customer_enriched'`. Controls what message content reaches Amplitude. Refer to [Choose a privacy mode](#choose-a-privacy-mode). |
| `redact_pii` | Scrub emails, phone numbers, SSNs, credit card numbers, and IP addresses from tracked content before events leave the process. Defaults to `True` from version 1.17.0, and to `False` in earlier versions. Set to `False` to opt out. |
| `custom_redaction_patterns` | Additional redaction patterns. Accepts regex strings (replaced with `[REDACTED]`) or `{ pattern, replacement }` objects for named labels. |
| `custom_redaction_fn` | A `(text) -> str` callback for custom redaction logic (for example, an NER library). Runs after all regex-based redaction. |
| `debug` | Log every tracked event to stderr. |
| `dry_run` | Build and log events without sending them to Amplitude. Use during development. |
| `review_mode` | Record each post-redaction event and print a sanitized upload preview without sending it. Defaults to `False`. Refer to [Review outbound traffic](#review-outbound-traffic). |
| `validate` | Enforce strict validation of required fields. |
| `on_event_callback` | A `(event, status_code, message) -> None` callback invoked exactly once per tracked event, from the delivery path. |
| `propagate_context` | Enable cross-service context propagation. Refer to [Propagate context across services](#propagate-context-across-services). |
| `strict_cost` | Raise `CostCalculationError` when a response has tokens but the SDK can't calculate its cost. Use in development and CI to catch unpriced model names. Defaults to `False`. |
| `capture_stack_trace` | Attach the traceback to error events as `[Agent] Stack Trace`. Stack traces can contain file paths and internal details, so this defaults to `False`. |

For redaction recipes (named replacements, custom scrubbers, international locales), refer to [Choose a privacy mode](#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. Set the `sessionId` from an ID you already track, such as a conversation thread, ticket, call, or run ID, rather than inventing a new one. Refer to [Choose a session ID](https://amplitude.com/docs/amplitude-ai/agent-analytics/sessions#choose-a-session-id).

#### Node

```typescript
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);
  });
}
```

#### Python

```python
from amplitude_ai import AmplitudeAI
from openai import OpenAI
import time

ai = AmplitudeAI()
client = OpenAI()

agent = ai.agent("chat-handler")

@app.post("/chat")
async def chat(req):
    body = await req.json()
    with agent.session(user_id=body["userId"]) as s:
        s.track_user_message(body["messages"][-1]["content"])
        start = time.perf_counter()
        response = client.chat.completions.create(model="gpt-4o-mini", messages=body["messages"])
        s.track_ai_message(
            response.choices[0].message.content or "",
            "gpt-4o-mini",
            "openai",
            (time.perf_counter() - start) * 1000,
        )
        return {"response": response.choices[0].message.content}
```

A session opened with `run()` (Node) or the `session()` context manager (Python) closes when the scope exits. For sessions that span multiple requests, call `trackSessionEnd()` (Node) or `track_session_end()` (Python) when the job finishes, or let the idle timeout close it. Refer to [How sessions close](https://amplitude.com/docs/amplitude-ai/agent-analytics/sessions#how-sessions-close) and [Configure session timeouts](https://amplitude.com/docs/amplitude-ai/agent-analytics/configure-session-timeouts).

> **Tip:** Minimum viable instrumentation
>
> Agent Analytics needs four fields to correlate events: the API key, a user identifier (`userId` or `deviceId`), an `agentId`, and a `sessionId`. The recommended pattern automatically adds provider wrappers so the SDK captures model, token, cost, and latency data.

For the user ID rules that keep one person from splitting into two users, refer to [Set user IDs](https://amplitude.com/docs/amplitude-ai/agent-analytics/sessions#set-user-ids).

## 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 the wrapper classes from the AI SDK, not from the provider package.

#### Node

```typescript
import { OpenAI } from "@amplitude/ai";

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

| Provider | Wrapper |
| --- | --- |
| OpenAI (Chat Completions + Responses) | `new OpenAI({ apiKey, amplitude: ai })` |
| Anthropic | `new Anthropic({ apiKey, amplitude: ai })` |
| Azure OpenAI | `new AzureOpenAI({ apiKey, amplitude: ai })` |
| Gemini (`@google/generative-ai`) | `new Gemini({ apiKey, amplitude: ai })` |
| Google Gen AI (`@google/genai`) | `new GoogleGenAI({ apiKey, amplitude: ai })` |
| Bedrock (Converse APIs) | `new Bedrock({ amplitude: ai, client })` |
| Mistral | `new Mistral({ apiKey, amplitude: ai })` |

#### Python

```python
import os

from amplitude_ai import OpenAI

openai = OpenAI(amplitude=ai, api_key=os.environ["OPENAI_API_KEY"])
```

| Provider | Wrapper |
| --- | --- |
| OpenAI (Chat Completions + Responses) | `OpenAI(amplitude=ai, api_key=...)` |
| OpenAI, async | `AsyncOpenAI(amplitude=ai, api_key=...)` |
| Anthropic | `Anthropic(amplitude=ai, api_key=...)` |
| Azure OpenAI | `AzureOpenAI(amplitude=ai, azure_endpoint=..., api_key=..., api_version=...)` |
| Google Gen AI (`google-genai`) | `Gemini(amplitude=ai, api_key=...)` |
| Bedrock (Converse APIs) | `Bedrock(amplitude=ai, region_name=...)` |
| Mistral | `Mistral(amplitude=ai, api_key=...)` |

If you can't change the construction site, use `wrap(existingClient, ai)` (Node) or `wrap(existing_client, amplitude=ai)` (Python) 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:

| Feature | OpenAI | Anthropic | Gemini | Azure OpenAI | Bedrock | Mistral |
| --- | --- | --- | --- | --- | --- | --- |
| Streaming | Yes | Yes | Yes | Yes | Yes | Yes |
| Tool-call tracking | Yes | Yes | No | Yes | Yes | No |
| TTFB measurement | Yes | Yes | No | Yes | No | No |
| Cache token stats | Yes | Yes | Yes | Yes | Yes | No |
| Responses API | Yes | — | — | — | — | — |
| Reasoning content | Yes | Yes | No | Yes | No | No |
| Cost estimation | Yes | Yes | Yes | Yes | Yes | Yes |

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:

#### Node

```typescript
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
```

#### Python

```python
ai = AmplitudeAI(api_key=os.environ["AMPLITUDE_AI_API_KEY"])
patch(amplitude_ai=ai)

agent = ai.agent("verify", user_id="verify-user-1")
with agent.session() as s:
    client.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:

#### Node

```typescript
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
```

#### Python

```python
from amplitude_ai import tool

search_products = tool(search_db, name="search_products")

# Inside the session context, call as usual:
result = search_products(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.

> **Warning:**
>
> Tool content comes from the **`input`** and **`output`** options. Amplitude ignores other spellings without an error, so the span records that a tool ran but not what it did. That includes `toolInput` and `toolOutput`, which are the names of the resulting `[Agent] Tool Input` and `[Agent] Tool Output` properties. Tool calls that `patch()` extracts automatically carry no input or output either, so add an explicit `trackToolCall` when you need the content. Rendering agent visuals in the session view depends on it. Refer to [Render agent artifacts](https://amplitude.com/docs/amplitude-ai/agent-analytics/render-artifacts).

For scripts and notebooks, `ToolCallTracker.setAmplitude()` (Node) or `ToolCallTracker.set_amplitude()` (Python) sets a process-wide default for tool tracking outside a session. The setting is global, so don't use it in multi-tenant servers; open a session per request instead.

`tool()` and `@tool` capture input and output according to the active privacy config: the one passed to the decorator, then the session's `AmplitudeAI` client, then `ToolCallTracker`. When none is set, the SDK records metadata only, and the Python SDK logs a one-time warning that starts with `@tool: no privacy config found`. To capture content, pass `privacyConfig` (Node) or `privacy_config=` (Python), or run the tool inside a session.

## 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.

> **Note:** OTEL-enabled behavior
>
> When OTEL is enabled through `enable_otel()` / `enableOtel()`, `observe()` / `@observe` creates real OTEL spans instead of emitting events directly. The `SpanEventMapper` converts these spans into the appropriate `[Agent]` event type. Use the `type` parameter to control routing: `@observe(type="tool")` routes the span as `[Agent] Tool Call` rather than `[Agent] Span`.

#### Node

```typescript
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;
}
```

#### Python

```python
from amplitude_ai import observe

# As a decorator:
@observe(name="sub-agent-execution")
def run_sub_agent(prompt: str):
    return sub_agent.execute(prompt)

# Or explicitly when you need error capture:
start = time.perf_counter()
try:
    result = sub_agent.execute(prompt)
    s.track_span(
        name="sub-agent-execution",
        latency_ms=(time.perf_counter() - start) * 1000,
        input_state={"prompt": prompt[:1000]},
        output_state={"response": result[:1000]},
    )
except Exception as e:
    s.track_span(
        name="sub-agent-execution",
        latency_ms=(time.perf_counter() - start) * 1000,
        is_error=True,
        error_type=type(e).__name__,
        error_message=str(e),
    )
    raise
```

> **Warning:** Spans don't replace turn-level events
>
> Agent Analytics turn counts and interaction views are driven by `[Agent] User Message` and `[Agent] AI Response`, not spans. If you only emit spans around internal steps, dashboards show traces with no turn-level analytics. Always emit the User Message / AI Response pair for each user-visible cycle, and use spans on top.

## Track embeddings

For retrieval and RAG workloads, track each embedding call as an `[Agent] Embedding` event with its model, provider, latency, and optionally input tokens, dimensions, and cost. Call the provider's embeddings API with your regular client, then track the call. The LlamaIndex and OpenTelemetry integrations track embeddings for you.

#### Node

```typescript
import OpenAIClient from "openai";

const embedder = new OpenAIClient();
const start = Date.now();
const result = await embedder.embeddings.create({ model: "text-embedding-3-small", input: query });
s.trackEmbedding("text-embedding-3-small", "openai", Date.now() - start, {
  inputTokens: result.usage.prompt_tokens,
  dimensions: result.data[0].embedding.length,
});
```

#### Python

```python
import time

from openai import OpenAI as OpenAIClient

embedder = OpenAIClient()
start = time.monotonic()
result = embedder.embeddings.create(model="text-embedding-3-small", input=query)
s.track_embedding(
    model="text-embedding-3-small",
    provider="openai",
    latency_ms=(time.monotonic() - start) * 1000,
    input_tokens=result.usage.prompt_tokens,
    dimensions=len(result.data[0].embedding),
)
```

When you pass input tokens, the SDK calculates `[Agent] Cost USD` for you. To set the cost yourself, pass `totalCostUsd` (Node) or `total_cost_usd` (Python).

## 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.

| Method | Event | Use when |
| --- | --- | --- |
| `s.trackUserMessage(text)` | `[Agent] User Message` | User-authored input arrives |
| `s.trackAiMessage(text, model, provider, latencyMs, opts?)` | `[Agent] AI Response` | Provider wrapper can't auto-capture |
| `s.trackToolCall(name, latencyMs, success, opts?)` | `[Agent] Tool Call` | Calling a tool outside `tool()`. Content comes from `opts.input` / `opts.output` |
| `s.trackSpan({ name, latencyMs, ... })` | `[Agent] Span` | Wrapping an internal sub-step |
| `s.runAs(childAgent, fn)` | (delegation) | Routing to a child agent |

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

#### Node

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

#### Python

```python
s.track_ai_message(
    completed_message.content,
    "gpt-4o",
    "openai",
    latency_ms,
    input_tokens=usage.prompt_tokens,
    output_tokens=usage.completion_tokens,
    total_tokens=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.

#### Node

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

#### Python

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

These keys cover the most common segmentation needs:

| Key | Example values | Use case |
| --- | --- | --- |
| `agent_type` | `"planner"`, `"executor"`, `"retriever"`, `"router"` | Group analytics by agent role in multi-agent systems. |
| `experiment_variant` | `"control"`, `"treatment-v2"` | Compare quality, abandonment, or cost across A/B test arms. |
| `feature_flag` | `"new-rag-pipeline"` | Track which flags were active during the session. |
| `surface` | `"chat"`, `"search"`, `"copilot"` | Identify the UI surface that triggered the interaction. |
| `prompt_revision` | `"v7"`, `"2026-02-15"` | Track the prompt version; detect regressions alongside `agentVersion`. |
| `deployment_region` | `"us-east-1"`, `"eu-west-1"` | Segment by region for latency or compliance analysis. |
| `canary_group` | `"canary"`, `"stable"` | Separate canary from stable deployments during a rollout. |

### Tag sessions

Pass `tags` when you open a session to label every event in it, for example by feature or cohort. The SDK emits them as `[Agent] Tags` on each event in the session.

#### Node

```typescript
await agent.session({ userId, sessionId, tags: ["beta", "checkout"] }).run(async (s) => {
  // ...
});
```

#### Python

```python
with agent.session(user_id=user_id, session_id=session_id, tags=["beta", "checkout"]) as s:
    ...
```

Use `context` for key-value dimensions and `tags` for simple labels.

### 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.

#### Node

```typescript
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" }
```

#### Python

```python
parent = ai.agent("orchestrator", context={"experiment_variant": "treatment", "surface": "chat"})
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.

#### Node

```typescript
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"
```

#### Python

```python
tenant = ai.tenant("org-456", groups={"company": "org-456"}, env="production")
agent = tenant.agent("support-bot", user_id="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.

| Tier | Examples | When to use |
| --- | --- | --- |
| `fast` | `gpt-4o-mini`, `claude-3-haiku`, `gemini-flash`, `gpt-3.5-turbo` | High-volume, latency-sensitive work. |
| `standard` | `gpt-4o`, `claude-3.5-sonnet`, `gemini-pro`, `llama`, `command` | General purpose. |
| `reasoning` | `o1`, `o3-mini`, `deepseek-r1`, Claude with extended thinking | Complex reasoning tasks. |

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

#### Node

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

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

#### Python

```python
from amplitude_ai import infer_model_tier

infer_model_tier("gpt-4o-mini")  # 'fast'
infer_model_tier("claude-3.5-sonnet")  # 'standard'
infer_model_tier("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:

#### Node

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

#### Python

```python
s.track_ai_message(response.content, "ft:gpt-4o:my-org:custom", "openai", latency_ms, model_tier="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`.

#### Node

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

#### Python

```python
s.track_user_message(
    "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.

| Signal | Property | Interpretation |
| --- | --- | --- |
| Copy | `[Agent] Was Copied` | User copied the output, a positive signal. Set on the AI-message call. |
| Regeneration | `[Agent] Is Regeneration` | User asked for a redo, a negative signal. Set on the user-message call. |
| Edit | `[Agent] Is Edit` + `[Agent] Edited Message ID` | User refined a previous prompt, a friction signal. Set on the user-message call. |
| Abandonment | `[Agent] Abandonment Turn` | User left after N turns; a low value (such as `1`) signals first-response dissatisfaction. Set on session end. |

#### Node

```typescript
// 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 });
```

#### Python

```python
# AI response the user copied (positive)
s.track_ai_message("To create a funnel, go to...", "gpt-4o", "openai", latency_ms, was_copied=True)

# User regenerates (negative — first response fell short)
s.track_user_message("How do I create a funnel?", is_regeneration=True)

# User edits and resubmits their prompt
s.track_user_message(
    "How do I create a conversion funnel for signups?",
    is_edit=True,
    edited_message_id=original_msg_id,
)

# User left after the first AI response
agent.track_session_end(session_id="sess-1", abandonment_turn=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.

#### Node

```typescript
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!" },
  ],
});
```

#### Python

```python
from amplitude_ai import track_conversation

track_conversation(
    amplitude=amplitude_client,
    user_id="user-123",
    session_id="sess-abc",
    agent_id="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. Call `ai.score()` any time, including after the session closes, because it targets a message or session by ID rather than the active session.

#### Node

```typescript
// 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",
});
```

#### Python

```python
# Thumbs up/down on a specific AI response
ai.score(
    user_id="user-123",
    name="user-feedback",
    value=1.0,
    target_id=ai_message_id,
    target_type="message",
    source="user",
)
```

Use the name `user-feedback` for your binary thumbs control. A thumbs down sent this way counts as explicit dissatisfaction for the Negative Feedback signal. For the full pattern (in-session and post-request paths, CSAT scales), refer to the [Collect user feedback in Agent Analytics](https://amplitude.com/docs/amplitude-ai/agent-analytics/collect-user-feedback).

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.

#### Node

```typescript
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: [...] });
  });
});
```

#### Python

```python
orchestrator = ai.agent("shopping-agent", description="Orchestrates shopping requests")
recipe_agent = orchestrator.child("recipe-agent", description="Finds recipes")

async with orchestrator.session(user_id=user_id) as s:
    s.track_user_message(user_input)
    async with s.arun_as(recipe_agent) as cs:
        cs.track_user_message(delegated_query)
        result = client.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.

#### Node

```typescript
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 });
});
```

#### Python

```python
@app.post("/chat")
async def chat(req):
    agent = ai.agent("api-handler", user_id=req.user_id)
    with agent.session(session_id=req.session_id) as s:
        s.track_user_message(req.message)
        start = time.perf_counter()
        response = client.chat.completions.create(model="gpt-4o", messages=req.messages)
        s.track_ai_message(
            response.choices[0].message.content or "",
            "gpt-4o",
            "openai",
            (time.perf_counter() - start) * 1000,
            input_tokens=response.usage.prompt_tokens,
            output_tokens=response.usage.completion_tokens,
        )
    ai.flush()
    return {"response": response.choices[0].message.content}
```

### 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.

#### Node

```typescript
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,
  });
});
```

#### Python

```python
agent = ai.agent("chatbot", user_id="user-123", env="production")

with agent.session(session_id=conversation_id) as s:
    s.track_user_message("What is Amplitude?")
    r1 = llm.chat("What is Amplitude?")
    s.track_ai_message(
        r1.content, "gpt-4o", "openai", r1.latency_ms,
        input_tokens=r1.usage.input, output_tokens=r1.usage.output,
    )

    s.track_user_message("How does it track events?")
    r2 = llm.chat("How does it track events?")
    s.track_ai_message(
        r2.content, "gpt-4o", "openai", r2.latency_ms,
        input_tokens=r2.usage.input, output_tokens=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](#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.

#### Node

```typescript
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);
});
```

#### Python

```python
async with orchestrator.session(session_id=sid) as s:
    s.new_trace()
    s.track_user_message("Generate plan from quiz results", context=structured_state)

    async def run_child(child, messages):
        async with s.arun_as(child):
            return client.chat.completions.create(model="gpt-4o", messages=messages)

    a, b = await asyncio.gather(
        run_child(scorer, scorer_messages),
        run_child(matcher, matcher_messages),
    )
    s.track_ai_message(assemble(a, b), "gpt-4o", "openai", total_latency_ms)
```

## 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.

```typescript
// 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);
});
```

In Python, consume the stream inside the session block. For a streaming HTTP response, open the session inside the generator, so it stays open until your app sends the last chunk:

```python
from fastapi.responses import StreamingResponse

def stream_answer(user_id: str, session_id: str, messages: list):
    with agent.session(user_id=user_id, session_id=session_id):
        stream = openai.chat.completions.create(model="gpt-4o", messages=messages, stream=True)
        for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                yield chunk.choices[0].delta.content

@app.post("/chat")
def chat(body: ChatRequest):
    return StreamingResponse(stream_answer(body.user_id, body.session_id, body.messages))
```

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

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

If you stream from a provider the SDK doesn't wrap, collect the stream with `StreamingAccumulator`, imported from `@amplitude/ai` in Node and `amplitude_ai.utils` in Python, and track one AI response when it ends. Add each text chunk with `addContent()` (Node) or `add_content()` (Python), which also measures time to first byte, record token usage from the final chunk, then pass the collected content and tokens to `trackAiMessage()` / `track_ai_message()`.

## 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:

#### Node

```typescript
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 });
```

#### Python

```python
browser_session_id = request.headers.get("x-amplitude-session-id")
device_id = request.headers.get("x-amplitude-device-id")
session = agent.session(
    user_id=user_id,
    browser_session_id=browser_session_id,
    device_id=device_id,
)
```

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.

#### Node

```typescript
// --- 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);
  });
}
```

#### Python

```python
# --- Service A (outbound) ---
from amplitude_ai import inject_context

with agent.session(user_id=user_id, session_id=session_id).run() as s:
    s.track_user_message(message)
    headers = inject_context({"content-type": "application/json"})
    requests.post("https://service-b/internal/enrich", headers=headers, json={"message": message})

# --- Service B (inbound): reopen the same session with the extracted IDs ---
from amplitude_ai import extract_context

extracted = extract_context(dict(request.headers))
with agent.session(
    session_id=extracted.get("session_id"),
    user_id=extracted.get("user_id"),
).run() as s:
    handle_enrichment(request)
```

`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, or `ai.amplitude` for CrewAI, 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.

#### Node

```typescript
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] }
```

#### Python

```python
from amplitude_ai.integrations.langchain import AmplitudeCallbackHandler

handler = AmplitudeCallbackHandler(amplitude_ai=ai, user_id="user-123", session_id="sess-1")
# chain.invoke(input, config={"callbacks": [handler]})
```

### LlamaIndex

#### Node

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

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

#### Python

```python
from amplitude_ai.integrations.llamaindex import AmplitudeLlamaIndexHandler

handler = AmplitudeLlamaIndexHandler(amplitude_ai=ai, user_id="user-123", session_id="sess-1")
```

### OpenAI Agents SDK

Register an `AmplitudeTracingProcessor` as a tracing processor.

#### Node

```typescript
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.
```

#### Python

```python
from amplitude_ai.integrations.openai_agents import AmplitudeTracingProcessor

processor = AmplitudeTracingProcessor(amplitude_ai=ai, user_id="user-123", session_id="sess-1")
# add_trace_processor(processor)
```

### Anthropic Tool Use

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

#### Node

```typescript
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 });
```

#### Python

```python
from amplitude_ai.integrations.anthropic_tools import AmplitudeToolLoop

loop = AmplitudeToolLoop(amplitude_ai=ai, user_id="user-123", session_id="sess-1")
loop.run(client=client, model="claude-sonnet-4-20250514", messages=messages, tools=tools, tool_executor=execute_tool)
```

### CrewAI (Python)

`AmplitudeCrewAIHooks` registers CrewAI hooks for LLM calls and tool calls. Pass it the underlying Amplitude client, `ai.amplitude`. Activate it before the crew runs and deactivate it afterward.

```python
from amplitude_ai.integrations.crewai import AmplitudeCrewAIHooks

hooks = AmplitudeCrewAIHooks(amplitude=ai.amplitude, user_id="user-123")
hooks.activate()
result = crew.kickoff()
hooks.deactivate()
```

### 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](#ingest-opentelemetry-spans). The mapping the exporter applies:

| OTEL span attribute | `[Agent]` property | Notes |
| --- | --- | --- |
| `gen_ai.response.model` / `gen_ai.request.model` | `[Agent] Model Name` | Response model preferred. |
| `gen_ai.system` / `gen_ai.provider.name` | `[Agent] Provider` | Required; spans without it are ignored. |
| `gen_ai.usage.input_tokens` | `[Agent] Input Tokens` |  |
| `gen_ai.usage.output_tokens` | `[Agent] Output Tokens` |  |
| `gen_ai.usage.total_tokens` | `[Agent] Total Tokens` | Derived from input + output if absent. |
| `gen_ai.request.temperature` | `[Agent] Temperature` |  |
| `gen_ai.request.top_p` | `[Agent] Top P` |  |
| `gen_ai.request.max_tokens` | `[Agent] Max Output Tokens` |  |
| `gen_ai.response.finish_reasons` | `[Agent] Finish Reason` | First reason if an array. |
| `gen_ai.tool.name` | `[Agent] Tool Name` | Routes the span as `[Agent] Tool Call`. |
| `gen_ai.input.messages` | `$llm_message` | User-role messages only, and only if the privacy mode allows. |
| Span duration | `[Agent] Latency Ms` |  |
| Span status `ERROR` | `[Agent] Is Error`, `[Agent] Error Message` |  |

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).

```typescript
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);
  }
});
```

In Python, import the same tracker from `amplitude_ai.integrations.claude_agent_sdk`:

```python
from amplitude_ai.integrations.claude_agent_sdk import ClaudeAgentSDKTracker
from claude_agent_sdk import ClaudeAgentOptions, query

agent = ai.agent("code-reviewer")
tracker = ClaudeAgentSDKTracker()

async with agent.session(user_id="u1", session_id="sess-abc") as s:
    options = ClaudeAgentOptions(hooks=tracker.hooks(s))
    async for message in query(prompt="Analyze this codebase", options=options):
        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:

| Anthropic event | SDK call |
| --- | --- |
| `user.message` | `trackUserMessage(text)` (track when sending, not when polling) |
| `agent.message` | `trackAiMessage(text, model, 'anthropic', latencyMs)` |
| `agent.tool_use` / `agent.mcp_tool_use` / `agent.custom_tool_use` | `trackToolCall(name, latencyMs, success)` |
| `agent.tool_result` / `agent.mcp_tool_result` | skip (latency captured at tool\_use time) |
| `session.error` | `trackAiMessage(errorMsg, model, 'anthropic', latencyMs, { isError: true })` |

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

#### Node

```typescript
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
}
```

#### Python

```python
seen_ids = set(saved_state.seen_ids)
for event in response.data:
    if event.id in seen_ids:
        continue
    seen_ids.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.

### Instrument your own MCP server

This section covers tracking tool calls inside an MCP server you build. For the MCP server that ships with the AI SDK and helps coding agents instrument your app, refer to `amplitude-ai mcp` in [Auto-instrument and CLI tools](#auto-instrument-and-cli-tools).

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, FastAPI, and Starlette

Use the bundled middleware to open a session for every request. Provider wrappers and tool calls inside the request handler inherit that session. The Node middleware supports Express, Fastify, and Hono. The Python middleware is ASGI middleware for FastAPI and Starlette; it ends the session and flushes events after each response.

#### Node

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

app.use(
  createAmplitudeAIMiddleware({
    amplitudeAI: ai,
    userIdResolver: (req) => req.user?.id ?? null, // authenticated user, not a request header
  }),
);
```

#### Python

```python
from amplitude_ai.middleware import AmplitudeAIMiddleware

app.add_middleware(
    AmplitudeAIMiddleware,
    amplitude_ai=ai,
    user_id_resolver=lambda request: request.state.user.id,  # authenticated user, not a request header
)
```

For Flask, Django, and other WSGI frameworks, open a session in each request handler with `ai.agent(...).session(...)` instead.

### 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:

```typescript
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.

#### Node

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

#### Python

```python
import signal
import sys

def handle_sigterm(signum, frame):
    ai.shutdown()
    sys.exit(0)

signal.signal(signal.SIGTERM, handle_sigterm)
```

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](#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](https://amplitude.com/docs/amplitude-ai/agent-analytics/instrument-opentelemetry).

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.

In Python, `enable_otel()` maps only spans that carry `gen_ai.*` or `amplitude.*` attributes. To also send other spans on the provider, such as HTTP or database spans, as `[Agent] Span` events, pass `otel_span_filter="all"`. Span names from those instrumentations can contain user data, so review them before you turn this on.

```python
ai = AmplitudeAI(api_key=os.environ["AMPLITUDE_AI_API_KEY"]).enable_otel(otel_span_filter="all")
```

## Choose a privacy mode

Set `contentMode` (Node) or `content_mode` (Python) on `AIConfig` to `full` (default), `metadata_only`, or `customer_enriched`. For what each mode sends and how to choose, refer to [Agent Analytics privacy modes](https://amplitude.com/docs/amplitude-ai/agent-analytics/privacy-modes).

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. `metadata_only` gates these channels:

- 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`)

In `full` mode, PII redaction 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.

Redaction is on by default in both SDKs:

- **Node**: `redactPii` defaults to `true`.
- **Python**: `redact_pii` defaults to `True` from version 1.17.0. Earlier versions default to `False` on `AIConfig`, so upgrading to 1.17.0 starts scrubbing content that was sent raw before. To keep raw content, pass `AIConfig(redact_pii=False)`.

For managed-agent architectures, prefer `full` with redaction on. 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](#choose-a-privacy-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:

| Field | Purpose |
| --- | --- |
| `qualityScore`, `sentimentScore` | Numeric quality and sentiment of the session. |
| `overallOutcome` | Terminal result, such as `resolved` or `escalated`. |
| `topicClassifications` | Map of taxonomy name to a `TopicClassification` (topic, confidence, subcategories). |
| `rubricScores` | Array of `RubricScore` (name, score, rationale, evidence). |
| `agentChain`, `rootAgentName` | Agent topology for multi-agent runs. |
| `requestComplexity` | Difficulty bucket, such as `low`, `medium`, or `high`. |
| `errorCategories` | Categorized failure signals from your pipeline. |
| `messageLabels` | Per-message labels keyed by the message ID returned from each tracking call. |
| `customMetadata` | Arbitrary key/value data for your own analytics. |

#### Node

```typescript
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 });
```

#### Python

```python
from amplitude_ai import (
    AmplitudeAI,
    AIConfig,
    ContentMode,
    SessionEnrichments,
    RubricScore,
    TopicClassification,
)

ai = AmplitudeAI(
    api_key=os.environ["AMPLITUDE_AI_API_KEY"],
    config=AIConfig(content_mode=ContentMode.CUSTOMER_ENRICHED),
)
agent = ai.agent("support-bot", agent_version="2.1.0")

# 1. Run the conversation — no content is sent, only metadata.
with agent.session(user_id="user-42") as s:
    s.track_user_message("Why was I charged twice?")
    s.track_ai_message(ai_response.content, "gpt-4o", "openai", latency_ms)
    session_id = s.session_id

# 2. Score the raw messages with your own pipeline.
eval_results = my_eval_pipeline(conversation_history)

# 3. Ship the enrichments back to Amplitude.
enrichments = SessionEnrichments(
    quality_score=eval_results.quality,
    sentiment_score=eval_results.sentiment,
    overall_outcome=eval_results.outcome,
    topic_classifications={
        "billing": TopicClassification(topic="billing-dispute", confidence=0.92),
    },
    rubric_scores=[RubricScore(name="accuracy", score=4, max_score=5)],
    custom_metadata={"eval_model": "gpt-4o-judge-v2"},
)

agent.track_session_enrichment(enrichments, session_id=session_id)
```

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.

| Provider | Raw API behavior | What to pass as inputTokens |
| --- | --- | --- |
| OpenAI | `prompt_tokens` already includes `cached_tokens` | Use directly |
| Anthropic / Bedrock (Converse) | `input_tokens` excludes cache tokens | `input_tokens + cache_read_input_tokens + cache_creation_input_tokens` |
| Gemini | `promptTokenCount` includes cached; `cachedContentTokenCount` reports separately | Use `promptTokenCount` directly |

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:

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

```python
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.

| Env var | Fallback | Emits as | Behavior |
| --- | --- | --- | --- |
| `AMPLITUDE_GIT_SHA` | `GIT_SHA` | `[Agent] Git SHA` | Read verbatim. |
| `AMPLITUDE_GIT_REF` | `GIT_REF` | `[Agent] Git Ref` | Read verbatim. Typically a branch name or tag. |
| `AMPLITUDE_GIT_REPO` | `GIT_REPO` | `[Agent] Git Repo` | Opt-in only. Sanitized: userinfo (`user:pass@`) is stripped, and the SDK logs one warning if a credential was stripped. |

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:

#### Node

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

#### Python

```python
s.track_user_message(
    "Summarize the attached design doc and list open questions",
    context={"structured_payload": payload_record},
)
```

**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:

#### Node

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

#### Python

```python
import json

# Session label becomes the JSON
s.track_user_message(json.dumps(payload_record))
```

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, such as Java, Go, Ruby, browsers, and edge environments, send `[Agent]` events to the Amplitude HTTP API directly. Refer to [Send agent events without the AI SDK](https://amplitude.com/docs/amplitude-ai/agent-analytics/instrument-http-api) for the steps and the [Event contract](https://amplitude.com/docs/amplitude-ai/agent-analytics/taxonomy#event-contract) for required properties.

## Verify your data

To check your instrumentation, including the `amplitude-ai doctor` command, the Live Events gates, and the `summary()` fill-rate report, refer to [Verify your Agent Analytics instrumentation](https://amplitude.com/docs/amplitude-ai/agent-analytics/verify).

## Test against a mock client

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

#### Node

```typescript
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);
}
```

#### Python

```python
from amplitude_ai import AIConfig
from amplitude_ai.testing import MockAmplitudeAI

mock = MockAmplitudeAI(AIConfig(content_mode="full"))
agent = mock.agent("test-agent", user_id="u1")

with agent.session(session_id="s1") as s:
    s.track_user_message("hello")
    s.track_ai_message("response", "gpt-4o-mini", "openai", 150)

mock.assert_event_tracked("[Agent] User Message", user_id="u1")
mock.assert_session_closed("s1")

# Data quality gate: every AI Response must carry the eight verification fields
for e in mock.events_of_type("[Agent] AI Response"):
    p = e.event_properties or {}
    assert e.user_id or e.device_id
    assert p["[Agent] Session ID"]
    assert p["[Agent] Model Name"]
    assert p["[Agent] Provider"]
    assert p["[Agent] Latency Ms"] > 0
    assert p["[Agent] Input Tokens"] > 0
    assert p["[Agent] Output Tokens"] > 0
    assert p["[Agent] Cost USD"] > 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](#initialize-the-sdk).

#### Node

```bash
# 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
```

#### Python

```bash
AMPLITUDE_AI_API_KEY=xxx AMPLITUDE_AI_AUTO_PATCH=true amplitude-ai-instrument python app.py
```

Both runtimes read the same environment variables:

| Variable | Description |
| --- | --- |
| `AMPLITUDE_AI_API_KEY` | Required to enable auto-patch. |
| `AMPLITUDE_AI_AUTO_PATCH` | Must be `"true"` to turn auto-patching on. |
| `AMPLITUDE_AI_CONTENT_MODE` | `full` (default), `metadata_only`, or `customer_enriched`. |
| `AMPLITUDE_AI_DEBUG` | `"true"` to log each event to stderr. |
| `AMPLITUDE_AI_REVIEW` | `"true"` to capture sanitized upload previews without sending them during auto-instrumentation. Setting `AMPLITUDE_AI_DEBUG` doesn't enable this. |

Other CLI commands:

| Command | Runtime | What it does |
| --- | --- | --- |
| `amplitude-ai` | Node, Python | Prints the prompt to paste into your AI coding agent. |
| `amplitude-ai doctor` | Node, Python | Checks dependencies, environment, and the event pipeline. In Python, also `amplitude-ai-doctor`. |
| `amplitude-ai status` | Node, Python | Prints the SDK version, detected provider packages, and environment configuration. |
| `amplitude-ai mcp` | Node, Python | Starts a local stdio MCP server that gives your AI coding agent tools to scan, instrument, and verify your app. It works on your source code and doesn't read your Amplitude data. In Python, install `amplitude-ai[mcp]` first. |
| `amplitude-ai verify` | Python | Runs local checks: SDK installed, API key set, provider detected, and a verify test present. Add `--live` for steps to confirm sessions in Amplitude. |
| `amplitude-ai-completions bash` or `zsh` | Node | Prints a shell completion script for the `amplitude-ai` commands. |

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](#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](https://amplitude.com/docs/data/create-tracking-plan), 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**.

#### Node

The bundled CLI reads the event catalog and prints a bash script of curl commands. It makes no network requests itself. Write the script to a file, review it, then run it. The script reads `AMPLITUDE_API_KEY` and `AMPLITUDE_SECRET_KEY` from the environment when it runs, so export them in the shell that runs it.

```bash
# Write the script (add --eu for EU data residency)
npx -y -p @amplitude/ai amplitude-ai-register-catalog > register.sh

# Review register.sh, then run it with AMPLITUDE_API_KEY and AMPLITUDE_SECRET_KEY exported
bash register.sh
```

#### Python

The Python CLI calls the Taxonomy API directly, with retry logic and a progress summary. It reads `AMPLITUDE_API_KEY` and `AMPLITUDE_SECRET_KEY` from the environment. To keep the secret key out of the environment, pass `--secret-key -` and pipe the secret on stdin.

```bash
export AMPLITUDE_API_KEY=...
export AMPLITUDE_SECRET_KEY=...

# Preview without calling the API
amplitude-ai-register-catalog --dry-run

amplitude-ai-register-catalog

# EU data residency
amplitude-ai-register-catalog --eu
```

> **Note:**
>
> Neither CLI accepts a secret key as a command-line value, because other local users can read process arguments and shells keep them in history. The Node CLI ignores `--api-key` and `--secret-key` and prints a warning. The Python CLI rejects a literal `--secret-key` value.

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

Debug and dry-run are two `AIConfig` flags for inspecting events locally. Each has an environment-variable equivalent for auto-instrumentation. For review mode, refer to [Review outbound traffic](#review-outbound-traffic).

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

```text
[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.

## Review outbound traffic

Review mode records the event handed to the analytics client after privacy redaction and prints a sanitized upload preview. It sends nothing. Use it to inspect the request instrumentation would make during a security review.

Review mode labels previews `pre-transport`. They show the event after redaction, before gzip, retries, or fields the analytics SDK adds later, such as `insert_id` and library metadata. The envelope API key is `[REDACTED]`. Previews keep prompt and message content intact, so don't commit the artifact.

#### Node

```typescript
const ai = new AmplitudeAI({
  apiKey: process.env.AMPLITUDE_AI_API_KEY!,
  config: new AIConfig({ reviewMode: true, validate: true }),
});

// Exercise one representative session, then:
await ai.flush();
console.log(ai.getReviewTraffic());
```

`patch({ amplitudeAI: ai })`, `wrap()`, and `agent().session()` all use this client, so `reviewMode: true` covers those paths. For zero-code instrumentation, set `AMPLITUDE_AI_REVIEW=true`.

#### Python

```python
ai = AmplitudeAI(
    api_key=os.environ["AMPLITUDE_AI_API_KEY"],
    config=AIConfig(review_mode=True, validate=True),
)
# Exercise one representative session, then:
ai.flush()
print(ai.get_review_traffic())
```

`wrap()`, agents, and sessions use this `AmplitudeAI` instance, so `review_mode=True` on its config covers them. `patch(amplitude_ai=ai)` does too. `patch(amplitude, review_mode=True)` covers a raw Amplitude client that doesn't go through `AmplitudeAI`. `AMPLITUDE_AI_REVIEW=true` enables that raw-client path. Read those previews with `get_patch_review_traffic()`. `unpatch()` restores `track` and keeps the captured previews until `clear_patch_review_traffic()`.

Pass `review_mode=True` on `AIConfig` when you construct `AmplitudeAI`. That class doesn't read `AMPLITUDE_AI_REVIEW`.

The event callback fires once per captured event with status `-1` and message `review-mode`. `flush()` still flushes the underlying Amplitude client, so it doesn't strand unrelated product events on a shared client. Confirm in the network panel that no Amplitude upload left the process. `AMPLITUDE_AI_DEBUG` doesn't enable review mode.

## Troubleshooting

For symptoms such as missing Session Records, events that don't group into sessions, empty message content, and missing costs, refer to [Troubleshoot Agent Analytics setup](https://amplitude.com/docs/amplitude-ai/agent-analytics/troubleshooting).

## API reference

### Core classes

| API | Purpose |
| --- | --- |
| `new AmplitudeAI({ apiKey, config? })` | Initialize the SDK |
| `new AIConfig({ contentMode?, redactPii?, customRedactionPatterns?, customRedactionFn?, dryRun?, reviewMode?, debug? })` | Privacy, debug, and review config |
| `ai.agent(agentId, opts?)` | Create a bound agent |
| `agent.child(agentId, opts?)` | Create a child agent for delegation |
| `agent.session(opts?)` | Create a session (auto-flushes in serverless) |
| `session.run(fn)` | Execute work with session context |
| `s.runAs(childAgent, fn)` | Delegate to a child agent |
| `ai.enableOtel()` / `ai.enable_otel()` | Enable OTEL span-first instrumentation |
| `ai.otelEnabled` / `ai.otel_enabled` | Whether OTEL mode is active (read-only) |
| `ai.flush()` | Flush buffered events (serverless / streaming) |
| `ai.getReviewTraffic()` / `ai.get_review_traffic()` | Closed upload previews plus the still-open batch, when review mode is on |
| `get_patch_review_traffic()` | Previews captured by Python `patch(review_mode=True)` or `AMPLITUDE_AI_REVIEW` |
| `ai.shutdown()` | Flush, then close the analytics client (process exit) |
| `ai.tenant(orgId, opts?)` | Tenant-scoped handle that pre-binds `customerOrgId` |
| `ai.score({ userId, name, value, targetId?, targetType?, source? })` | Record explicit user feedback as `[Agent] Score` |

### Session tracking methods

| Method | Event |
| --- | --- |
| `s.trackUserMessage(content, opts?)` | `[Agent] User Message` |
| `s.trackAiMessage(content, model, provider, latencyMs, opts?)` | `[Agent] AI Response` |
| `s.trackToolCall(name, latencyMs, success, opts?)` | `[Agent] Tool Call` |
| `s.trackSpan({ name, latencyMs, ... })` | `[Agent] Span` |
| `s.trackEmbedding(model, provider, latencyMs, opts?)` | `[Agent] Embedding` |
| `s.trackSessionEnrichment({...})` | Session-level enrichment (customer\_enriched mode) |

### Higher-order functions

| HOF | Event | Use |
| --- | --- | --- |
| `tool(fn, { name })` | `[Agent] Tool Call` | Wrap tool functions |
| `observe(fn, { name, type? })` | `[Agent] Span` | Wrap any function for observability (with OTEL: creates real spans; `type` controls event routing) |

### Other APIs

| API | Use |
| --- | --- |
| `patch({ amplitudeAI: ai })` / `unpatch()` | Zero-code instrumentation, auto-extracts Tool Calls from message arrays |
| `wrap(client, ai)` | Wrap an existing provider client without modifying its construction |
| `injectContext()` / `extractContext(headers)` | Cross-service propagation |
| `usingAttributes(attrs, fn)` / `using_attributes(**attrs)` | Attach identity and session context to OTEL spans |
| `updateCurrentSpan(attrs)` / `update_current_span(**attrs)` | Update attributes on the active OTEL span |
| `createAmplitudeAIMiddleware(opts)` / `AmplitudeAIMiddleware` | Express, Fastify, and Hono middleware (Node); FastAPI and Starlette middleware (Python) |
| `StreamingAccumulator` | Collect a stream you track manually into one AI response |
| `calculateCost({ modelName, ... })` | Compute cost directly when you need to override `totalCostUsd` |
| `trackRunCost(...)` / `track_run_cost(...)` | Emit run-end delta `[Agent] Cost USD` for non-conversational agents |
| `trackConversation({ ... })` | Backfill a full message history as events |
| `inferModelTier(model)` | Resolve a model's tier (`fast` / `standard` / `reasoning`) |
| `enableLivePriceUpdates()` | Refresh `genai-prices` cost data at runtime |
| `MockAmplitudeAI` (`@amplitude/ai/testing`) | Deterministic test double; call `.summary()` for a fill-rate report |
| `ClaudeAgentSDKTracker` (`@amplitude/ai/integrations/claude-agent-sdk`) | Claude Agent SDK integration |