amplitude-mcp-analytics

Amplitude MCP Analytics SDK — Model Context Protocol server usage tracking for Amplitude Analytics.

Status: Preview. Server and tool instrumentation, the default event set, identity resolution, and custom events are available now. Transport and correlation handling (stdio, Streamable HTTP — stateful and stateless — and SSE) is handled for you under the hood.

Install

uv add amplitude-mcp-analytics 'mcp>=1.16,<2'

amplitude-analytics comes along as a dependency — nothing else to install whichever way you construct the client. mcp is listed here because your MCP server needs it, not because this SDK depends on it: the coupling is duck-typed, so the SDK never pins a range against yours. The supported range is mcp>=1.16,<2. The mcp 2.x line restructures the server internals; a 2.x server is detected and rejected with a clear error rather than silently emitting nothing (2.x support is planned).

Agent-assisted setup: the instrument-mcp-server skill (part of the builder-skills engineering-skills plugin) exists for agent-assisted setup, but it was written for the Node SDK. A coding agent can still follow this README manually to instrument a Python server.

Quick start

import os

from mcp.server.fastmcp import FastMCP

from amplitude_mcp_analytics import AmplitudeMCPAnalytics

mcp = FastMCP("my-mcp-server")

analytics = AmplitudeMCPAnalytics(
    api_key=os.environ["AMPLITUDE_API_KEY"],
    server_name="my-mcp-server",
    server_version="1.0.0",
)

# Bind the server (enables analytics + emits connection events), then wrap your
# tool handlers. Order matters: instrument_server() must run before mcp.run().
analytics.instrument_server(mcp, auth_type="oauth")

@mcp.tool()  # @mcp.tool() stays outermost — see below
@analytics.instrument_tool(name="search", owner="docs-team")
async def search(query: str) -> str:
    return await do_search(query)  # your handler, unchanged

mcp.run()

To reuse an Amplitude client you already own, pass it instead of api_key:

from amplitude import Amplitude

AmplitudeMCPAnalytics(amplitude=Amplitude("KEY"), server_name="...", server_version="...")

Instrumenting your server

Two steps, both wrap things you already have — no handler signatures change.

instrument_server(server, ...) binds the SDK to your MCP server — a FastMCP or a low-level mcp.server.lowlevel.Server. The Python MCP SDK has no connect(transport); each run() call is one connection, so the SDK wraps run() instead. It auto-detects the transport from the message stream (stdio / streamable-http / sse; override with transport=), captures the client/server handshake, and emits the default connection events. Call it before the server runs. It's idempotent and returns the same server.

instrument_tool(...) wraps a tool handler. The returned function has the exact same signature and sync/async nature as the one you pass in (functools.wraps preserves the original signature), so FastMCP still builds the tool's input schema from your parameters. On each call it emits [MCP] Tool Call Response with timing, error, and size details.

analytics.instrument_server(mcp)

@mcp.tool()
@analytics.instrument_tool(
    name="search", owner="docs-team", extra={"feature flag": "new-ranker"}
)
async def search(query: str) -> str:
    return await do_search(query)

Decorator order is load-bearing. @mcp.tool() must be outermost, with @analytics.instrument_tool(...) directly under it. FastMCP builds the tool's input schema from the signature of the function it decorates — functools.wraps preserves that signature through the analytics wrapper, but only if the wrapper sits inside @mcp.tool(). Swapping the order would register the raw handler and instrument nothing.

Not using decorators? The same method wraps a handler directly: mcp.add_tool(analytics.instrument_tool(search, McpToolMeta(name="search"))).

instrument_tool requires instrument_server. If the server was never bound, the wrapper is a no-op passthrough: your handler runs untouched, nothing is emitted, and a one-time warning is logged. Instrumenting a tool can never change its behavior.

Default events

Once a server is bound and its tools wrapped, the SDK emits these automatically:

All event names and properties are prefixed [MCP] so they never collide with same-named events/properties from other Amplitude SDKs on the same project.

This table is a summary. The full reference — every property and when it's present, identity resolution, transport nuances, and the error taxonomy — lives in docs/events.md.

Sessionless Streamable HTTP still performs the initialize handshake, so it emits [MCP] Session Initialized with [MCP] Session ID: no-session. It does not emit [MCP] Session Ended: a duration for a transport that lives for one request is not meaningful. Every event also carries the shared context properties (identity, client/server, transport, trace correlation).

Client name on sessionless servers

Through protocol 2025-11-25, clientInfo appears only on initialize. A host that creates a fresh server for every HTTP request therefore cannot read that name on a later tools/call. Supply it from request-local OAuth claims or headers with resolve_client_info:

from amplitude_mcp_analytics import McpClientInfo

analytics.instrument_server(
    mcp,
    resolve_client_info=lambda request: McpClientInfo(
        name=(request.auth_info or {}).get("client_name")
    ),
)

The callback wins over wire-derived sources and may return only the fields it knows; returning None, empty fields, or raising falls through safely. The SDK also emits [MCP] OAuth Client ID from the request's verified client_id. That value identifies a registration rather than a product, so it is never folded into [MCP] Client Name.

Identity

user_id must match whatever you already send to Amplitude for the same user. The SDK never guesses it from auth — you provide it, via whichever path fits:

from amplitude_mcp_analytics import McpTenant, SetIdentityInput

# 1. Bound with the server. Scoped to that binding — hosts that build one
#    server per request can pass per-request values safely; concurrent
#    bindings never overwrite each other.
analytics.instrument_server(
    mcp,
    user_id="user-123",
    tenant=McpTenant(group_type="org id", group_value="456"),
)

# 2. Per request, inside a handler (wins over everything else).
@mcp.tool()
@analytics.instrument_tool(name="search")
async def search(query: str) -> str:
    analytics.set_identity(user_id=my_auth.get_login_id())
    return await do_work(query)

# 3. Opt-in, derived from the request's verified auth info (you map the claims).
#    Pass the same resolver to the server and its tools so every event resolves
#    the caller consistently.
resolve_identity = lambda auth_info: SetIdentityInput(
    user_id=(auth_info or {}).get("subject"),
)
analytics.instrument_server(mcp, resolve_identity=resolve_identity)

@mcp.tool()
@analytics.instrument_tool(
    name="lookup",
    resolve_identity=resolve_identity,
)
async def lookup(doc_id: str) -> str: ...

set_identity also accepts a positional SetIdentityInput instead of the keyword arguments — handy for forwarding a value you already hold, which is exactly what a resolve_identity callback returns. Passing both spellings in one call raises ValueError.

The resolve_identity callback receives the request's auth info as a plain dict — the access-token information produced by the MCP server's token verifier (client_id, scopes, subject, and optional claims). The SDK passes this value to your resolver but never emits it. Your resolver owns the mapping and may ignore auth_info in favor of identity your application already made available through trusted request-local state. One transport nuance the SDK absorbs for you: under stateful Streamable HTTP the session runs in a different task than the HTTP request, so the SDK reads the token from the request's ASGI scope — the SDK's get_access_token() contextvar alone is not visible there.

Resolution order (first match wins): set_identity() → resolve_identity() → instrument_server options → correlation anchor → an anonymous floor. When no explicit identity is supplied but a correlation anchor exists (a stdio process, a transport session id, or a propagated W3C trace context), the SDK emits accurate aggregate-only data under a synthetic device_id derived from that anchor — never a polluting placeholder, never a fabricated user.

If there is no anchor either — the fully stateless case with no identity and no tenant — each request would mint a brand-new random device_id with no cross-call stitching, so those events are dropped by default rather than inflating your user counts (see docs/events.md). Opt in to emit them as anonymous, aggregate-only data with emit_anonymous_event=True:

from amplitude_mcp_analytics import MCPAnalyticsConfig

MCPAnalyticsConfig(emit_anonymous_event=True)

Rationale

Agent clients often supply a free-text rationale for a tool call ("why I'm calling this tool"). If your server receives one — as a tool argument, in _meta, a header, or however your convention works — pass it to the SDK and it is emitted as the reserved [MCP] Rationale property on the tool-call event and on every tool-scope custom event of the same invocation:

@mcp.tool()
@analytics.instrument_tool(name="search")
async def search(query: str, rationale: str | None = None) -> str:
    if isinstance(rationale, str):
        analytics.set_rationale(rationale)
    return await do_work(query)

The SDK never reads rationale out of tool inputs itself: it is content-bearing free text, so emitting it is an explicit opt-in, and where it lives is your convention. Callable at any depth inside an instrumented handler (like set_identity); truncated to 1000 characters; last write wins. Omitted entirely when never set.

Feedback

register_feedback_tool(server) adds a submit_feedback tool to a FastMCP server. The agent calls it when the user reacts to a result from this server ("thanks, that's it", "that's wrong") or asks to give feedback, and the SDK emits [MCP] Feedback Submitted. instrument_server does not add the tool. A server that never opts in is unchanged.

analytics.instrument_server(server)
analytics.register_feedback_tool(server)

server.run()

The tool description is what tells the agent when to call it. The result text tells it not to ask again in the same conversation unless the user brings feedback up. The call does not also emit [MCP] Tool Call Response.

helpful is a boolean. reason, when the user gave one, is one of wrong_result, incomplete, too_slow, wrong_tool, missing_capability, or other. An optional tools list is kept only for names registered on that server. solicited records whether the agent asked or the user volunteered.

Free-text comment is off by default, and the field is then absent from the schema. Opt in to emit [MCP] Feedback Comment, capped at 500 characters before and after sanitize_error_message and the same built-in PII redaction as [MCP] Error Message:

analytics.register_feedback_tool(server, capture_comment=True)

A name that is already registered is left alone. Pass name to register alongside an existing tool of the same name. On a low-level Server (no add_tool) this warns and does nothing.

The event follows the same anonymous-floor rule as every other event: a call with no identity, no tenant, and no correlation anchor is dropped unless emit_anonymous_event is on.

To name the tool in the server's own instructions (those are fixed when the server is constructed), append FEEDBACK_TOOL_INSTRUCTIONS:

from amplitude_mcp_analytics import FEEDBACK_TOOL_INSTRUCTIONS

mcp = FastMCP(
    "my-mcp-server",
    instructions=f"Your instructions.\n\n{FEEDBACK_TOOL_INSTRUCTIONS}",
)

feedback_tool_instructions(name) is the same paragraph for a renamed tool. create_feedback_tool_handler(analytics) is the callback on its own, for a framework that registers tools itself. Pass server so tools can be checked against that server's registry.

Tool parameter capture

For handlers that receive arguments, [MCP] Tool Call Response automatically includes content-free metadata describing which parameters were supplied and their shallow types:

• [MCP] Param Keys — sorted top-level keys, capped at 32
• [MCP] Param Count — the full top-level key count
• [MCP] Param Shape — deterministic types, array/object counts, and bucketed string lengths, capped at 1,024 characters
• [MCP] Param Fingerprint — a stable 12-character hash of the shape

Values are not included. Nested lists, dicts, and pydantic models are counted, not traversed, and exact string lengths are bucketed. Only identifier-like keys ([A-Za-z0-9_.:\\-/]{1,64}) appear on [MCP] Param Keys and [MCP] Param Shape; caller-controlled names from open schemas (for example an email used as a key) are omitted there but still counted in [MCP] Param Count. By default, rationale and context are also excluded because servers commonly use those names for content-bearing injected metadata. An injected FastMCP Context is not a parameter and is left out entirely. On a tools/call, capture keeps the argument names the client sent. FastMCP still calls the handler with every declared parameter filled in; those filled-in values are not counted. A function you call directly, rather than through that request, falls back to dropping a keyword that is the signature's default object. Enum members are captured as their value, so a str enum route is route=list rather than route=Action.LIST.

Multiplexed tools can include a safe route value in the shape (route_key is for small schema enums — string, finite number, or boolean — not user_id- style fields). Tools can also opt into bounded scalar facts derived from their inputs:

@mcp.tool()
@analytics.instrument_tool(
    name="manage_items",
    param_capture={
        "route_key": "action",
        "never": ["private_metadata"],
        "derive": lambda params: {
            "item_count": len(params["items"]) if isinstance(params.get("items"), list) else 0,
            "destructive": params.get("action") == "delete",
        },
    },
)
async def manage_items(action: str, items: list[str]) -> str:
    return await do_work(action, items)

Derived facts are emitted as [MCP] Param: <key>. The SDK accepts at most eight per tool call and drops non-scalars, non-finite numbers, long strings, unsafe property names, and strings containing email or free-text indicators such as @, quotes, or newlines. never applies to both input keys and those derived fact names. A raising derive callback is ignored and never affects the tool response. A mistyped opt-in field (route_key, derive, never) logs a warning and is ignored; default shape capture still runs. Only a non-object param_capture disables capture for that tool.

Disable automatic shape capture, or replace the global exclusion list, through the SDK config. never_keys replaces the default ("rationale", "context") — include those names if you still want them omitted. Opted-in derive facts still run when shape is disabled:

from amplitude_mcp_analytics import DEFAULT_PARAM_NEVER_KEYS, MCPAnalyticsConfig

MCPAnalyticsConfig(
    param_capture={
        "shape": False,
        "never_keys": [*DEFAULT_PARAM_NEVER_KEYS, "reason", "customer_context"],
    },
)

The SDK may inspect and serialize inputs to derive this metadata and [MCP] Request Size, but it never emits input content unless the server author explicitly opts into a derived fact or calls a content-bearing API such as set_rationale.

Error HTTP status

When a tool call fails on a raised error that carries an HTTP status (err.status, err.status_code, or err.response.status_code — the common Python conventions, covering aiohttp, httpx, and requests), the tool-call event includes [MCP] Error HTTP Status. This is the status of the failure the tool hit (an upstream API response, an HTTP-shaped error), NOT the MCP transport status — per the MCP spec, tool failures are returned in-band, so the transport typically answers 200 even when this property is a 4xx/5xx.

For error shapes the SDK can't sniff, set it explicitly when building the error: analytics.tool_error(ctx, code=..., message=..., http_status=502).

Related but distinct: [MCP] Response HTTP Status is the transport-level status of the HTTP response itself. The instrumented-tool wrapper never emits it (dispatched tool calls answer 200; the wrapper emits before the response is written). The default [MCP] Tool Call Rejected event carries it on the HTTP transports (protocol-level rejections answer 200 with the error in the JSON-RPC body). For events you emit yourself, set ctx.request.response_http_status before calling track_tool_event.

Choosing what's captured

All default events are on by default. Toggle them with autocapture — a boolean for everything, or a mapping to control families independently:

import os

from amplitude_mcp_analytics import AmplitudeMCPAnalytics, MCPAnalyticsConfig

AmplitudeMCPAnalytics(
    api_key=os.environ["AMPLITUDE_API_KEY"],
    server_name="my-mcp-server",
    server_version="1.0.0",
    config=MCPAnalyticsConfig(
        # keep tool-call events, drop connection events
        autocapture={"server_events": False},
    ),
)

autocapture=False disables all default events; {"server_events": ..., "tool_calls": ...} toggles each family. tool_calls covers both [MCP] Tool Call Response and [MCP] Tool Call Rejected. server_events can be split further with session_lifecycle ([MCP] Session Initialized/Ended) and tools_listed ([MCP] Tools Listed) — e.g. servers built per HTTP request typically want {"session_lifecycle": False, "tools_listed": True}, since their transports close at the end of every request rather than at session end. Custom events (below) are unaffected.

Redacting free-text properties

Two properties carry text the SDK didn't compose, and each has a hook that rewrites or drops it before emission.

[MCP] Error Message carries free text the SDK didn't compose — a failing tool's own message, or the MCP SDK's input-validation text, which quotes the rejected argument value. Either may contain end-user data. sanitize_error_message rewrites or drops it before emission, on every event that carries it:

import os
import re

from amplitude_mcp_analytics import AmplitudeMCPAnalytics, MCPAnalyticsConfig

AmplitudeMCPAnalytics(
    api_key=os.environ["AMPLITUDE_API_KEY"],
    server_name="my-mcp-server",
    server_version="1.0.0",
    config=MCPAnalyticsConfig(
        sanitize_error_message=lambda message: re.sub(
            r"[\w.+-]+@[\w-]+\.[\w.]+", "<email>", message
        ),
    ),
)

Return None to omit the property entirely. A sanitizer that raises fails closed — the property is omitted, never the raw message. [MCP] Error Code and [MCP] Error Type are unaffected, so failures stay segmentable. The text sent to the client never changes. Built-in PII redaction still runs on whatever string the sanitizer returns. See Redacting [MCP] Error Message.

[MCP] Rationale is the other free-text property, and sanitize_rationale is its counterpart — same signature, same fail-closed contract. A rationale is model-written prose about why a tool was called, so it can quote the end user's prompt verbatim even when every argument your tool receives is clean:

MCPAnalyticsConfig(
    sanitize_rationale=lambda rationale: rationale[:120],  # or None to drop it
)

It applies wherever the property is lowered — the default [MCP] Tool Call Response event and every tool-scope custom event of the same invocation — and, like sanitize_error_message, the hook itself changes nothing until you configure it. Built-in PII redaction still runs on whatever string it returns. The rationale you passed to set_rationale is never returned to the client either way; this affects telemetry only.

Privacy and redaction

By default the SDK redacts personally identifiable information from the free-form content of every event it emits:

• host enrichment on extra (server scope and tool metadata)
• properties passed to track_server_event / track_tool_event
• [MCP] Rationale (after sanitize_rationale, when that hook is set)
• [MCP] Error Message (after sanitize_error_message, when that hook is set)
• opt-in derived [MCP] Param: values

The built-in patterns cover emails, phone numbers (including international + numbers), credit cards, SSNs, and IPv4/IPv6 addresses. A value that is entirely a base64-encoded image is replaced with [base64 image redacted].

Identity and dimension fields are never redacted — user id, device id, group, session id, server and tool names, error type, and the rest of the reserved [MCP] properties other than rationale and error message. Redacting those would corrupt attribution. Raw tool arguments are not sent at all; see Tool parameter capture.

import os
import re

from amplitude_mcp_analytics import AmplitudeMCPAnalytics, MCPAnalyticsConfig

analytics = AmplitudeMCPAnalytics(
    api_key=os.environ["AMPLITUDE_API_KEY"],
    server_name="my-mcp-server",
    server_version="1.0.0",
    config=MCPAnalyticsConfig(
        # redact_pii=True,  # built-in PII patterns (default)
        custom_redaction_patterns=[
            r"secret-\d+",  # bare string → "[REDACTED]"
            {"pattern": r"\bACME-\d+\b", "replacement": "[ticket]"},
        ],
        custom_redaction_fn=lambda text: re.sub(r"internal-\w+", "[hidden]", text),
    ),
)

redact_pii=False turns off the built-in patterns (for example when you already redact upstream). Custom patterns, custom_redaction_fn, and base64-image replacement still run. A custom_redaction_fn that raises or returns a non-string replaces that value with [content redacted]. The failure is logged once, and the log names the failure kind rather than the input or the custom pattern. An invalid custom_redaction_patterns entry, or a custom_redaction_fn that is not a function, raises ConfigurationError when the config is constructed.

AmplitudeMCPAnalytics.track_server_event and track_tool_event always use that config. The module-level functions of the same names take the same three fields on TrackEventOptions for one call. An invalid entry there raises ConfigurationError from the call instead of dropping the event.

Context (ctx)

Every tracked event carries a per-invocation context object. You can construct one and pass it explicitly to the tracking APIs, or expose it via run_with_context so deeper call stacks can read it through get_current_context().

from amplitude_mcp_analytics import (
    McpServerInfo,
    McpToolMeta,
    create_server_context,
    create_tool_context,
    run_with_context,
)

server_ctx = create_server_context(
    server=McpServerInfo(name="my-mcp-server", version="1.0.0"),
    transport="stdio",
)

tool_ctx = create_tool_context(server_ctx, McpToolMeta(name="search_docs"))

def emit() -> None:
    ...  # get_current_context() is available here if needed

run_with_context(tool_ctx, emit)

Types and helpers live in the amplitude_mcp_analytics.context subpackage and are also re-exported from the main package (amplitude_mcp_analytics).

You usually don't build ctx by hand — instrument_server / instrument_tool construct and inject it for you (inside an instrumented handler, get_current_context() returns the current invocation's context). Reach for these factories when emitting events outside an instrumented handler.

Custom event properties

Every event carries a set of reserved properties the SDK derives from the context — identity, session/trace correlation, client/server identity, and (for tool events) the tool metadata. You can attach your own properties on top of these from two places:

• extra — an enrichment bag carried on the context. Put domain values at the server scope (extra= in instrument_server) or on a tool (extra= in instrument_tool) and they ride along on every event derived from that scope — including the default events.
• properties — the per-call argument to track_server_event / track_tool_event, for values specific to that one event.

Precedence

When the same key appears in more than one place, the merge order is fixed — later sources overwrite earlier ones:

reserved (SDK-derived)  <  extra (context bag)  <  properties (per call)

• A properties value wins over anything with the same key — including a reserved property (the explicit, per-call value is the most intentional one).
• An extra value overrides a reserved property but loses to properties.
• On the default events, the SDK's outcome values ([MCP] Is Error, [MCP] Response Duration, …) ride as per-call properties, so a colliding extra key can't overwrite them.

Reserved names all carry the [MCP] prefix — avoid it in your own keys and collisions never arise.

Dropping the extra bag

extra properties are included by default. To omit them for a single event, pass TrackEventOptions(drop_extra_props=True):

from amplitude_mcp_analytics import TrackEventOptions

analytics.track_tool_event(
    ctx, "my event", {"foo": "bar"}, TrackEventOptions(drop_extra_props=True)
)

Free-form strings in extra and properties are run through PII redaction before they are sent. Reserved dimension fields are not. The SDK does not otherwise escape values — apply output encoding where the data is rendered.

Architecture decisions

Separate repo from the agent-analytics SDKs

MCP server analytics is a distinct product from agent analytics. Different audience (MCP server operators vs. agent developers), different domain model (server / session / tool invocation vs. agent / turn / message), and a different release cadence. Keeping the repos separate lets each evolve on its own timeline without coupling unrelated breaking changes.

The sibling the -node suffix left room for

The Node SDK shipped first as Amplitude-MCP-Analytics-Node, its suffix deliberately leaving room for a Python SDK without forcing a future rename. This repo is that sibling: same product, same wire contract, one repo per runtime.

Mimic the Node SDK for DX, not for the domain model

Repo layout, constructor shape, mock test client, and docs mirror Amplitude-MCP-Analytics-Node (which in turn mirrors @amplitude/ai), so contributors moving between the repos see familiar patterns — with Python-native build tooling (uv, pytest, pyright, ruff) in place of the Node stack. The domain model — events, properties, identity, context — is MCP-native and intentionally does not reuse agent vocabulary.

One runtime dependency, and mcp is not it

The Node SDK keeps the MCP SDK and the Amplitude client as peer dependencies. Python has no peer-dependency concept, so the two peers split:

• amplitude-analytics is a hard dependency. A required peer maps to a plain runtime dep — that is what "you must have this" means in Python. Behind an extra it would instead be a construction-time failure for anyone who missed the extra, which is a worse trade than one small, already-transitive package.
• mcp is not a dependency at all. The coupling is structural — duck-typed and confined to one adapter module (core/mcp.py) — and every consumer is an MCP server that already depends on mcp. Declaring a range here could only fight theirs, so the supported range is documented and asserted at runtime instead of pinned. It is a dev-group dependency for the test suite.

The low-level delivery utilities (delivery hooks, serverless flush accounting, privacy/PII redaction) are ported rather than depended on — no shared package, no version coupling at runtime.

Ported from the Node SDK

This SDK is a port of @amplitude/mcp-analytics (Amplitude-MCP-Analytics-Node). The wire contract — event names, [MCP] property names, precedence rules, identity math — is shared, so dashboards can slice both SDKs without special-casing; the integration seams are Python-native. The intentional divergences, and the policy for pulling upstream changes over, are recorded in PORTING.md.

Development

uv sync
uv run pytest
uv run pyright src
uv run ruff check src tests

Releases are automated: conventional-commit PR titles drive the version bump, release-please keeps a Release PR open against main, and merging it publishes to PyPI. See CONTRIBUTING.md.