Metadata-Version: 2.5
Name: armature-mcp-analytics
Version: 0.1.39
Summary: Armature analytics wrapper SDK for Python MCP servers.
Project-URL: Homepage, https://armature.tech
Project-URL: Repository, https://github.com/armature-tech/mcp-analytics-python
Author-email: Armature <support@armature.tech>
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: analytics,armature,fastmcp,mcp,model-context-protocol,observability,telemetry
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Provides-Extra: fastmcp
Requires-Dist: fastmcp<5,>=2; extra == 'fastmcp'
Provides-Extra: mcp
Requires-Dist: mcp<3,>=1.27; extra == 'mcp'
Description-Content-Type: text/markdown

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

[![PyPI version](https://img.shields.io/pypi/v/armature-mcp-analytics?label=PyPI)](https://pypi.org/project/armature-mcp-analytics/)
[![Python versions](https://img.shields.io/pypi/pyversions/armature-mcp-analytics)](https://pypi.org/project/armature-mcp-analytics/)
[![CI](https://github.com/armature-tech/mcp-analytics-python/actions/workflows/ci.yml/badge.svg)](https://github.com/armature-tech/mcp-analytics-python/actions/workflows/ci.yml)
[![Apache 2.0](https://img.shields.io/badge/license-Apache--2.0-blue)](LICENSE)

[Armature](https://armature.tech) · [TypeScript SDK](https://github.com/armature-tech/mcp-analytics) · [Go SDK](https://github.com/armature-tech/mcp-analytics-go) · [Agent install](SKILL.md)

## Install in 30 seconds

### 1. Install

For servers using the standalone FastMCP package:

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

If FastMCP comes from the official MCP Python SDK:

~~~bash
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:

~~~bash
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:

~~~bash
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:

~~~bash
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:

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

~~~json
{
  "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:

~~~python
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):

~~~python
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:

~~~python
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:

~~~python
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:

~~~python
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](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:

~~~python
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:

> Call this before you tell the user that these tools can't do what they asked. It records the request so the developers of this server can add it. It changes no data and contacts no one. Then answer the user 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**:

~~~python
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:

~~~python
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](examples/minimal):

~~~bash
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](https://github.com/armature-tech/mcp-analytics-python/issues) · [Email us](mailto:hey@armature.tech) · [Releases](https://github.com/armature-tech/mcp-analytics-python/releases)

## License

Licensed under the [Apache License 2.0](LICENSE).
