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.

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. 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, changes nothing on the wire until you configure it. The rationale you passed to set_rationale is never returned to the client either way; this affects telemetry only.

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

Values are sent as provided — the SDK does not escape or redact them. Apply any 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) 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.