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 ingest key

Create a server in the Armature dashboard, copy its ingest key, and add it to your environment:

export ANALYTICS_INGEST_API_KEY="..."

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

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, reasoning, and frustration 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",
    "agent_thinking": "The payment lookup tool provides the requested status",
    "user_frustration": "low"
  }
}

All telemetry fields are optional. Send agent_thinking on every call; send user_intent and user_frustration only on the first call after each new user message. Their absence on later calls means the same turn continues. The earlier aliases remain accepted, while cached user_turn values are 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

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

Official MCP Python SDK

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

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

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, request, header, and authentication information in the dispatcher context when it is available.

Stateless HTTP and serverless

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.

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

Most servers only need ANALYTICS_INGEST_API_KEY. 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_id": "stable-user-or-tenant-seed",
            "enabled": True,
            "delivery": "await",
            "timeout_ms": 500,
            "emit": None,
            "on_error": None,
        }
    },
)

Telemetry capture and privacy

The SDK injects an optional telemetry parameter (user_intent, agent_thinking, user_frustration) 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.

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

Redaction and binary payloads

Before any preview is serialized, the SDK strips binary content automatically: image/audio content-block data, resource blobs, base64 data URIs, and long base64 strings are replaced with "[binary removed]" / "[base64 removed]" placeholders. A redact callable then runs over the sanitized inputs, outputs, error strings, and telemetry text, and must return the value to serialize. The pipeline is sanitize → redact → stringify → truncate. If the callable raises, the SDK fails closed: the affected payload is replaced with "[redaction failed]" and the event still ships.

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

Delivery

• "background" schedules delivery on the running event loop. Call await instrumentation.recorder.flush() during shutdown.
• "await" waits for the delivery attempt before returning. Use it for serverless functions and short-lived processes.

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.

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 and 3.x
• Official MCP Python SDK 1.27+
• Synchronous and asynchronous tool handlers

Environment variables

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="..." python server.py

Support

Open an issue · Email us · Releases

License

Licensed under the Apache License 2.0.