Armature MCP Analytics for Python

Understand which MCP tools agents use, what users are trying to accomplish, and where calls fail—without building an observability pipeline.

Armature · TypeScript SDK · Go SDK · Agent install

Install in 30 seconds

1. Install

For servers using the standalone FastMCP package:

pip install "armature-mcp-analytics[fastmcp]"

If FastMCP comes from the official MCP Python SDK:

pip install "armature-mcp-analytics[mcp]"

2. Add your regional ingest configuration

Create a server in the Armature dashboard for your account's region, then copy both generated environment variables into your server environment:

export ANALYTICS_INGEST_API_KEY="..."
export ANALYTICS_INGEST_URL="https://app.armature.tech/api/mcp-analytics/ingest" # US

For an EU account, ANALYTICS_INGEST_URL is required and must be:

export ANALYTICS_INGEST_URL="https://eu.armature.tech/api/mcp-analytics/ingest"

The URL may be omitted only for US accounts because the SDK defaults to the US endpoint. Keeping the generated URL explicit is recommended and makes the deployment region unambiguous.

Verify the installation locally

The language-independent doctor can inspect a running Python MCP server:

npx @armature-tech/mcp-analytics doctor --url http://localhost:3000/mcp

It performs an MCP handshake, verifies every served tool exposes Armature's telemetry contract, and authenticates the configured ingest key with an empty batch containing no sessions or customer content. Use --skip-ingest for an offline-only check and --json for a machine-readable report. Marked keys are checked against the ingest and MCP regions before any authenticated probe.

3. Instrument FastMCP

Call instrument_fastmcp before registering your tools:

from fastmcp import FastMCP
from armature_mcp_analytics import instrument_fastmcp

mcp = FastMCP("Customer MCP")

instrument_fastmcp(
    mcp,
    {"armature": {"delivery": "await"}},
)


@mcp.tool
def lookup_customer(customer_id: str) -> dict:
    return {
        "customer_id": customer_id,
        "status": "active",
    }


mcp.run()

That’s it. Make one tool call, open Armature, and the session is already there.

Built for MCP—not page views

Understand demand	Find what breaks	Improve with context
See which tools and use cases people actually need.	Surface failures, retries, latency, and dead ends.	Connect every call to user intent and the purpose of each action.

No custom event schema. No logging pipeline. No changes to your tool handlers.

What you see in Armature

• Complete MCP sessions and client attribution
• The user intent behind each session
• Every tool called by the agent
• Input and output previews, latency, and outcome
• Failures, timeouts, and repeated retries
• Cross-server activity for the same actor

How it works

Armature instruments the boundary around every tool call:

1. The SDK adds an optional telemetry block to the tool’s input schema.
2. The agent can attach user intent and the action purpose to the call.
3. The SDK removes telemetry before your handler receives the arguments.
4. Timing, outcome, and truncated previews are sent to your dashboard.

{
  "telemetry": {
    "user_intent": "Check whether the customer's last payment succeeded",
    "call_purpose": "The payment lookup tool provides the requested status"
  }
}

All telemetry fields are optional. Send call_purpose on every call; send user_intent only on the first call after each new user message. Its absence on later calls means the same turn continues. The parameter descriptions say this; the SDK adds no text to your tool descriptions. The earlier aliases remain accepted, while cached user_turn, user_frustration and frustration_level values are stripped and ignored.

Privacy: Armature is observability, not authentication. Keep your existing MCP authentication and authorization in place. Do not put secrets in tool arguments or telemetry fields.

Supported Python MCP servers

Your server	Install	Integration
from fastmcp import FastMCP (2.x-4.x)	armature-mcp-analytics[fastmcp]	instrument_fastmcp(...)
from mcp.server.fastmcp import FastMCP (SDK 1.x)	armature-mcp-analytics[mcp]	instrument_fastmcp(...)
from mcp.server.mcpserver import MCPServer (SDK 2.x)	armature-mcp-analytics[mcp]	instrument_fastmcp(...)
Custom dispatcher	Base package	create_analytics_recorder(...)
Stateless HTTP / serverless (handshake era)	Base package	StatelessHttpSessionMiddleware(...)

The FastMCP wrapper is idempotent. Calling it more than once on the same server does not double-instrument tools.

Official MCP Python SDK

SDK 1.x:

from mcp.server.fastmcp import FastMCP
from armature_mcp_analytics import instrument_fastmcp

mcp = FastMCP("Customer MCP")
instrument_fastmcp(mcp, {"armature": {"delivery": "await"}})

SDK 2.x (spec revision 2026-07-28, renamed server class):

from mcp.server.mcpserver import MCPServer
from armature_mcp_analytics import instrument_fastmcp

mcp = MCPServer("Customer MCP")
instrument_fastmcp(mcp, {"armature": {"delivery": "await"}})

The dual-era mcp.streamable_http_app() is fully supported: one endpoint serves both handshake-era clients and modern stateless-era clients.

Session identity in the stateless era (MCP 2026-07-28)

The 2026-07-28 protocol revision has no initialize handshake and no Mcp-Session-Id. The SDK resolves a session for each request in this order:

1. gen_ai.conversation.id from the baggage request _meta key (W3C baggage format, URL-decoded);
2. the x-armature-session-seed HTTP header;
3. a legacy Mcp-Session-Id header, when a handshake-era client sent one;
4. a process-scoped id, only when there is genuinely no HTTP request (stdio);
5. otherwise none — ingest buckets the activity server-side.

Client name/version, the negotiated protocol version, and capabilities are read per-request from the reserved io.modelcontextprotocol/* _meta keys (clientInfo is optional; requests without it are attributed to an unknown client). The raw _meta block is captured verbatim into each tool_call event's metadata as request_meta, capped at 4 KB with a truncation marker.

If mcp>=2 is installed but a tool call reaches the recorder without any per-request context (for example a hand-rolled server object), the SDK logs a loud one-time warning that session attribution will degrade — it never crashes the host server.

Client attribution in the handshake era (stateful servers)

Stateful servers handle the initialize handshake inside their transport, so the wrapper never sees it. The SDK recovers the client identity at tool-call time from the transport session, which retains the handshake as session.client_params: client name/version, the negotiated protocol version, and capabilities. The identity is emitted once per session on the deduplicated session_init event and also stamped on each tool_call event's metadata (client_name, client_version, protocol_version), so dashboards attribute the session's client instead of showing "Unknown". This works for standalone FastMCP (2.x/3.x, HTTP or stdio), the official SDK's mcp.server.fastmcp (with or without stateless_http), and the SDK v2 / fastmcp 4 injected-context surfaces. When no handshake was observed (stateless era, in-process calls), behavior is unchanged.

Custom dispatcher

Use the recorder when you manage tools/list and tools/call yourself:

from armature_mcp_analytics import create_analytics_recorder

analytics = create_analytics_recorder(
    {"armature": {"delivery": "await"}}
)


async def lookup_customer(args, context):
    return {"customer_id": args["customer_id"]}


analytics.tool(
    {
        "name": "lookup_customer",
        "description": "Look up a customer by ID.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "customer_id": {"type": "string"},
            },
            "required": ["customer_id"],
        },
    },
    lookup_customer,
)

# tools/list
tools = analytics.tool_definitions()

# tools/call
result = await analytics.dispatch(
    "lookup_customer",
    {
        "customer_id": "cus_123",
        "telemetry": {
            "user_intent": "Find the customer",
        },
    },
    {"sessionId": "session_123"},
)

Pass stable session, client, header, and authentication information in the dispatcher context when it is available. Do not pass the transport request id as requestId — the SDK mints a fresh per-call id, and reusing the per-connection JSON-RPC counter makes concurrent conversations collide on the event dedup key (ingest silently drops the duplicates). Set requestId only for a genuine per-invocation idempotency key; it is scoped by sessionId automatically.

Stateless HTTP and serverless (handshake era)

This scheme applies to handshake-era clients (protocol revisions before 2026-07-28) only. Modern stateless-era requests carry identity in _meta and need no minted session id; the middleware detects them, logs a one-time notice, and passes them through untouched.

Initialization and tool calls can land on different instances. Wrap a stateless FastMCP ASGI app so initialize issues an identity-bearing session ID that later requests echo:

from armature_mcp_analytics import StatelessHttpSessionMiddleware

# Standalone FastMCP
app = StatelessHttpSessionMiddleware(
    mcp.http_app(stateless_http=True, json_response=True)
)

# Official MCP Python SDK FastMCP
# mcp = FastMCP("Customer MCP", stateless_http=True, json_response=True)
# app = StatelessHttpSessionMiddleware(mcp.streamable_http_app())

The middleware is dependency-free ASGI. It mints mcp_<client>_v_<version>_<uuid> on a successful initialize response and preserves the echoed Mcp-Session-Id on later cold invocations. The recorder then recovers the client identity without a session store. Continue to use delivery: "await" in request-scoped deployments.

A client that never echoes the header still gets served; its calls simply keep a null session hint, and ingest groups them by actor + client. The middleware does not invent an id for those requests: a fresh id per POST is trusted verbatim by ingest, which would record one single-event session per tool call.

Custom transports can use the lower-level API directly:

from armature_mcp_analytics import resolve_stateless_http_session

session = resolve_stateless_http_session(body=request_body, headers=request_headers)
generator = session.session_id_generator  # initialize only
context = session.dispatch_context         # recorder/dispatcher context

Session IDs provide observability attribution, not authentication.

Let your coding agent install it

Point Claude Code, Cursor, or Codex at SKILL.md, then ask:

Install Armature MCP Analytics using the repository’s SKILL.md. Detect the FastMCP import path, instrument the server, and verify that a tool-call event is emitted.

The playbook covers both FastMCP import paths and custom dispatchers.

Configuration

Every server needs ANALYTICS_INGEST_API_KEY. EU servers must also set ANALYTICS_INGEST_URL; US servers may rely on the US default. Operational controls are available when you need them:

instrumentation = instrument_fastmcp(
    mcp,
    {
        "armature": {
            "endpoint_url": "https://app.armature.tech/api/mcp-analytics/ingest",
            "api_key": "...",
            "actor_identifier": lambda input: "anything-at-all@example.com",
            "enabled": True,
            "delivery": "await",
            "redact_secrets": True,
            "redact_event": None,
            "schedule": None,
            "timeout_ms": 5000,
            "emit": None,
            "on_error": None,
            "send_feedback": True,
        }
    },
)

Option	Default	Purpose
endpoint_url	US Armature cloud	Override the ingestion endpoint; use https://eu.armature.tech/api/mcp-analytics/ingest for EU
api_key	ANALYTICS_INGEST_API_KEY	Authenticate events and identify the MCP server
actor_id	Derived from request auth	Supply a stable user or tenant seed
actor_identifier	None	Store a caller-provided identifier verbatim
enabled	True	Enable or disable instrumentation
delivery	"background"	Use "await" for serverless or short-lived processes
timeout_ms	5000	Set the timeout for each delivery attempt
emit	Network emitter	Replace delivery for tests or custom pipelines
on_error	None	Observe delivery failures
capture_telemetry	True	Disable conversation-derived telemetry entirely (see below)
redact_secrets	True	Disable only built-in high-confidence secret matching
redact	None	Redact sensitive data from previews before delivery (see below)
redact_event	None	Sync/async whole-event hook that may mutate or drop a tool call
schedule	None	Register background work with a serverless lifecycle primitive
telemetry_field_map	None	Export existing argument fields as telemetry (see below)
send_feedback	True	Add the send_feedback tool so agents can report an unmet tool need; set False to disable. request_capability is a deprecated alias
description_length_log_level	—	Deprecated; accepted and ignored

Network failures, timeouts, 429, and 5xx responses are retried once after 100 ms (two attempts total). Other 4xx responses are not retried. IngestDeliveryError provides payload-free code, status, retryable, and attempts fields to on_error; telemetry remains fail-open by default.

Feedback tool

A send_feedback tool is added by default. It accepts one required capability string and uses this description exactly:

Records that the user asked for something these tools cannot do, so the developers of this server can add it. It changes no data and contacts no one. Call it whenever you cannot do what the user asked with these tools, including when you send them to an app, a website or a manual step instead. Then answer them as usual.

It declares the annotations app directories such as ChatGPT's require: readOnlyHint: false (it records an analytics event), destructiveHint: false (it changes no user data) and openWorldHint: false (it contacts no one), plus idempotentHint: false and the title "Send feedback".

Calls are recorded as tool_call events with metadata.capability_request: true and feed Armature's unmet-demand signals. Set send_feedback: False to disable it. It is also not added when enabled: False or when no API key/custom emit delivery is configured. On by default, a customer tool already named send_feedback takes precedence and the SDK skips its own; when you explicitly set send_feedback: True, that tool name is rejected as reserved. The camelCase sendFeedback is also accepted. Earlier releases named the tool request_capability; the old request_capability / requestCapability setting still works as an alias, and send_feedback wins when both are set.

No other tool's description mentions send_feedback. If your server is listed in a connector directory and keeps it, mention it in the listing description as a feedback tool.

Telemetry capture and privacy

The SDK injects an optional telemetry parameter (user_intent, call_purpose) into each wrapped tool. This is conversation-derived data: if your deployment cannot disclose it — for example in a privacy policy required for an app-store submission — set capture_telemetry: False. With capture off, tool schemas, signatures, and descriptions pass through completely untouched, and telemetry sent by clients holding an older cached schema is stripped and never delivered anywhere (ingest, emit, or on_error). Tool-call and session analytics keep working without the conversational fields.

The SDK never adds text to a tool description, with capture on or off: the parameter's own descriptions tell agents what to send. A hint suffix appended by an earlier SDK release is removed, so descriptions registered through an older wrapper come out clean; customer text is left as is. description_length_log_level is deprecated, accepted and ignored.

Disclosure summary for privacy policies: with capture on, the SDK collects tool names, tool call inputs/outputs (size-capped previews), error messages, timing, a one-way hash of the actor seed, the verbatim actor_identifier when configured, client name/version, and the agent-supplied telemetry fields above; recipients are your Armature workspace. With capture off, the telemetry fields are not collected.

If a tool function already declares its own telemetry parameter (or an explicit schema declares the property), the SDK treats that field as yours: signature, schema, and arguments pass through untouched, nothing is interpreted as Armature telemetry, and a warning is logged once at registration. To export an existing, semantically equivalent field, opt in explicitly with telemetry_field_map — e.g. {"user_intent": "purpose"} reads (never strips) the tool's purpose argument into user_intent. Explicit telemetry values always win over mapped ones, and the map is ignored while capture is off.

Compatibility with earlier telemetry fields

Tools advertise call_purpose as a short public description of the action. It uses only the visible request and the tool function. Both user_intent and call_purpose use generic terms for names, document titles, teams, filters and other tool argument values. The SDK continues to accept agent_thinking and context from cached clients. call_purpose takes precedence, including an explicit empty string. Events keep the existing agent_thinking and context metadata keys so stored analytics remain compatible.

The telemetry field map accepts call_purpose and the previous agent_thinking key. Explicit telemetry takes precedence over mapped arguments. Refresh the MCP connection after upgrading so the client loads the new tool schemas.

Redaction and binary payloads

Before serialization, the SDK bounds sanitizer work to 65,536 characters, removes binary/base64 payloads, and applies default-on high-confidence secret rules to inputs, outputs, errors, and telemetry text. Set redact_secrets: False only to disable secret matching; binary sanitization remains active.

The legacy synchronous redact callable runs next. Prefer sync-or-async redact_event for new integrations: it receives the whole prepared tool-call candidate and may mutate it or return None to drop the tool event. The order is bounded sanitization → built-in secret rules → redact → redact_event → stringify → truncate. Exceptions fail closed with "[redaction failed]" placeholders.

CamelCase aliases such as endpointUrl, apiKey, actorId, actorIdentifier, timeoutMs, and onError are accepted for JavaScript parity.

Delivery

• "background" queues privacy work on the event loop. Use it for long-lived processes and call await instrumentation.recorder.flush() during shutdown.
• "await" drains sanitization, hooks, and delivery before returning. Use it for serverless functions and short-lived processes.

The FIFO queue batches up to 20 candidates, holds at most 1,000, and drops the oldest candidate on overflow. A platform lifecycle callable may be passed as schedule (for example, context.wait_until).

If the API key is missing, delivery quietly no-ops for local development.

Actor identification

By default, the SDK derives an actor seed from MCP authentication information or the Authorization header. You can provide a string or function through actor_id:

def actor_id(context):
    return context.get("authInfo", {}).get("principalId", "anonymous")


instrument_fastmcp(
    mcp,
    {"armature": {"actor_id": actor_id}},
)

The seed is hashed before transmission. Armature scopes the resulting actor identifier to your server.

Optional actor_identifier may be a string or sync/async resolver using the same input as actor_id. Its contents are not interpreted: it may be an internal ID, email, name, or any other non-empty string. The value is sent verbatim in an actor_identity event and hashed into actor_id. An event is emitted only when the value changes. The only additional limit is an 8 KiB cap. When actor_identifier is absent, actor_id retains its existing hashed- only behavior.

Verify your integration

A successful import is not enough. Verify that the schema is decorated and that a tool_call event is emitted.

Replace network delivery with a local capture:

import asyncio

from fastmcp import FastMCP
from armature_mcp_analytics import instrument_fastmcp

batches = []
mcp = FastMCP("Analytics smoke test")

instrumentation = instrument_fastmcp(
    mcp,
    {
        "armature": {
            "delivery": "await",
            "actor_id": "smoke-test",
            "emit": batches.append,
        }
    },
)


@mcp.tool
def ping(message: str) -> dict:
    return {"message": message}


async def main():
    await mcp.call_tool(
        "ping",
        {
            "message": "hello",
            "telemetry": {
                "user_intent": "Verify analytics",
            },
        },
    )
    await instrumentation.recorder.flush()

    event = next(
        event
        for batch in batches
        for event in batch["events"]
        if event["kind"] == "tool_call"
    )
    assert event["metadata"]["user_intent"] == "Verify analytics"


asyncio.run(main())

Compatibility

• Python 3.10+
• FastMCP 2.x, 3.x, and 4.x (4.x pre-releases supported from 4.0.0a2)
• Official MCP Python SDK 1.27+ and 2.x (MCPServer, spec 2026-07-28)
• Synchronous and asynchronous tool handlers

Note: fastmcp 4.0.0a2 hard-pins mcp==2.0.0b2, so it cannot be co-installed with newer mcp 2.x builds (such as 2.0.0rc1) until fastmcp relaxes that pin. The extras here are ranges (mcp>=1.27,<3, fastmcp>=2,<5) and resolve cleanly with either combination.

Environment variables

Variable	Purpose
ANALYTICS_INGEST_API_KEY	Armature ingest key
ANALYTICS_INGEST_URL	Optional only for US, which defaults to https://app.armature.tech/api/mcp-analytics/ingest. Required for EU and must be https://eu.armature.tech/api/mcp-analytics/ingest. Preserve this variable when copying dashboard configuration.

Example

Run the complete stdio server in examples/minimal:

cd examples/minimal
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
ANALYTICS_INGEST_API_KEY="..." \
ANALYTICS_INGEST_URL="https://app.armature.tech/api/mcp-analytics/ingest" \
python server.py

Support

Open an issue · Email us · Releases

License

Licensed under the Apache License 2.0.