> ## Documentation Index
> Fetch the complete documentation index at: https://docs.armature.tech/llms.txt
> Use this file to discover all available pages before exploring further.

# Python

> Instrument a Python MCP server with armature-mcp-analytics

The Python SDK instruments servers built with [FastMCP](https://gofastmcp.com) — both the standalone package and the `FastMCP` class from the official [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk). Python 3.10+.

## Install

<CodeGroup>
  ```bash FastMCP (standalone) theme={null}
  pip install "armature-mcp-analytics[fastmcp]"
  ```

  ```bash Official MCP SDK theme={null}
  pip install "armature-mcp-analytics[mcp]"
  ```
</CodeGroup>

The base package has no runtime dependencies; the extras pull in the framework you use (`fastmcp>=2,<4` or `mcp>=1.27,<2`).

## Set your API key

```bash theme={null}
ANALYTICS_INGEST_API_KEY=<your API key>
```

Create a key in the dashboard, then add it to your deployment secrets. If the key is missing, the SDK quietly no-ops — your server keeps working, and no data is sent.

## Instrument your server

Call `instrument_fastmcp` once, before registering tools:

```python server.py theme={null}
from fastmcp import FastMCP
from armature_mcp_analytics import instrument_fastmcp

mcp = FastMCP("Customer MCP")
instrument_fastmcp(mcp)

@mcp.tool
def lookup_customer(customer_id: str) -> dict:
    """Look up a customer by ID."""
    return {"customer_id": customer_id, "status": "active"}

mcp.run()
```

Works identically with the official SDK's FastMCP:

```python theme={null}
from mcp.server.fastmcp import FastMCP
from armature_mcp_analytics import instrument_fastmcp

mcp = FastMCP("Customer MCP")
instrument_fastmcp(mcp)
```

Instrumentation is idempotent — calling `instrument_fastmcp` twice does not double-count. Sync and async tool handlers are both supported.

## Custom dispatcher

If you handle `tools/list` and `tools/call` yourself, use the recorder directly:

```python theme={null}
from armature_mcp_analytics import create_analytics_recorder

analytics = create_analytics_recorder()

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

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 = analytics.tool_definitions()                       # tools/list
result = await analytics.dispatch(name, args, context)     # tools/call
```

<Note>
  Running on serverless or stateless HTTP? Wrap your ASGI app in `StatelessHttpSessionMiddleware` and set `"delivery": "await"`. See [Stateless & serverless servers](/sdks/stateless-servers).
</Note>

## Configuration

Pass options as a dict under the `armature` key. Both `snake_case` and `camelCase` keys are accepted:

```python theme={null}
instrument_fastmcp(mcp, {
    "armature": {
        "delivery": "background",
        "timeout_ms": 500,
        "on_error": lambda err, batch: print("analytics delivery failed", err),
    }
})
```

| Option         | Default                            | Description                                                                                                               |
| -------------- | ---------------------------------- | ------------------------------------------------------------------------------------------------------------------------- |
| `api_key`      | `ANALYTICS_INGEST_API_KEY` env var | API key used to authenticate ingestion.                                                                                   |
| `endpoint_url` | Armature ingest endpoint           | Override with `ANALYTICS_INGEST_URL` or this option.                                                                      |
| `enabled`      | `True`                             | Set `False` to disable capture entirely.                                                                                  |
| `delivery`     | `"background"`                     | `"background"` posts events off the request path; `"await"` sends before returning (use in serverless).                   |
| `timeout_ms`   | `500`                              | Delivery timeout in milliseconds.                                                                                         |
| `actor_id`     | Derived from request auth          | A string or callable (sync or async) used to attribute calls to a user. See [Identifying users](/sdks/identifying-users). |
| `emit`         | Network emitter                    | Replace delivery with your own callable — useful in tests.                                                                |
| `on_error`     | `None`                             | Called when delivery fails. Failures never reach your tools.                                                              |

## Flush on shutdown

With background delivery, drain pending events before the process exits:

```python theme={null}
instrumentation = instrument_fastmcp(mcp)
# ... on shutdown:
await instrumentation.recorder.flush()
```

## Verify locally

Capture batches in memory instead of sending them:

```python theme={null}
batches = []
instrument_fastmcp(mcp, {"armature": {"emit": batches.append, "delivery": "await"}})

# call a tool through the MCP client, then:
event = batches[0]["events"][0]
assert event["metadata"]["tool_name"] == "lookup_customer"
```

## Troubleshooting

<AccordionGroup>
  <Accordion title="Telemetry fields missing from tool schemas">
    FastMCP builds the advertised schema from your function's type hints, so the SDK adds a keyword-only `telemetry` parameter to instrumented functions. If you post-process signatures or schemas yourself, make sure that parameter survives.
  </Accordion>

  <Accordion title="HTTP sessions all look anonymous">
    The SDK reads the `Mcp-Session-Id` header, including on FastMCP versions that exclude it from `get_http_headers()` by default. If you run a custom ASGI stack, make sure the header reaches the app.
  </Accordion>

  <Accordion title="No data arrives">
    Check that `ANALYTICS_INGEST_API_KEY` is set in the server's environment (missing keys no-op silently by design) and pass an `on_error` hook to surface delivery failures.
  </Accordion>
</AccordionGroup>
