mcpeye 0.1.4

mcpeye Python SDK — open-source product analytics for MCP servers. See why your agent is failing.

mcpeye (Python SDK)

Open-source product analytics for MCP servers. See why your agent is failing.

mcpeye instruments your MCP server so you can answer the question the hero Intent Gap Report is built around: what did users ask the tools to do that the tools could not deliver?

It works by injecting an optional mcpeyeIntent parameter into every tool's input schema. The agent self-reports, in its own words, why it is calling a tool and any blocker the user hit — so intent is captured at near-zero cost, with no per-call LLM. The clustering LLM runs later, server-side, only to build reports.

This is the Python SDK. The TypeScript SDK is the reference implementation; behaviour and the wire contract are identical across SDKs.

Install

pip install mcpeye

Requires Python 3.9+. Depends on httpx and pydantic. The mcp package is not a hard dependency — it is your server's framework, imported lazily by track(). If you use the automatic track() path, install it alongside (or use the extra, which pins mcp>=1.0.0, itself Python 3.10+):

pip install "mcpeye[mcp]"

Quick start

Works out of the box with both official server styles — a low-level mcp.server.Server or a high-level FastMCP. Call track(...) once, after you have registered your tools (list_tools/call_tool handlers, or @mcp.tool() functions):

from mcp.server.lowlevel import Server
import mcpeye

server = Server("my-server", version="1.2.3")

@server.list_tools()
async def list_tools():
    return [ ... ]

@server.call_tool()
async def call_tool(name, arguments):
    ...

# Instrument it. mcpeyeIntent is injected into every tool schema, calls are
# captured, redacted, buffered, and shipped to the ingest API.
mcpeye.track(
    server,
    project_id="proj_123",
    ingest_url="http://localhost:3001",   # or set MCPEYE_INGEST_URL
    ingest_secret="...",                  # or set MCPEYE_INGEST_SECRET
)

track returns the same server instance, instrumented in place.

For FastMCP, pass the FastMCP object directly — mcpeye resolves its inner low-level server automatically (mcpeyeIntent is stripped before your function runs, so a plain def search(query) never sees the extra argument):

from mcp.server.fastmcp import FastMCP
import mcpeye

mcp = FastMCP("my-server")

@mcp.tool()
def search(query: str) -> str:
    ...

mcpeye.track(mcp, project_id="proj_123", ingest_url="http://localhost:3001")

What track does

1. Injects mcpeyeIntent into each tool's inputSchema (and teaches the server's tool cache about it, so the built-in input validation accepts the parameter). The agent fills it in; your tool handler never sees it — it is stripped from the arguments before your code runs.
2. Captures every tool call: tool name, redacted arguments, redacted result, error state + message, duration, and the reported intent. Each captured field is size-bounded (oversized or unserializable values become a small marker) so a multi-MB tool result can never blow the ingest body limit or grow the buffer.
3. Adds a reserved mcpeye_request_capability tool (active missing-capability capture). When the agent wants a capability none of your tools cover, it can call this tool to say so in the user's words. mcpeye answers it locally with a canned acknowledgement — it is never forwarded to your server — and records it as a normal tool call with tool_name = "mcpeye_request_capability", which the report folds into "Top missing capabilities" as a high-confidence, explicitly-requested entry. This catches the silent miss, where the right move is to call no tool at all. Disable with capture_missing_capabilities=False.
4. Ships batches as the shared IngestPayload JSON to <ingest_url>/ingest with the x-mcpeye-secret header, using httpx. Shipping happens on a background daemon thread — never on the tool-call thread — so a slow or unreachable ingest endpoint adds no latency to your tools. It flushes on a timer (default every 5s), eagerly when the batch fills (flush_at), and a final time at process exit. Shipping is best-effort: ingest failures are swallowed (routed to on_error) and retried; analytics can never take down your MCP server.

Strict ingest URL. Unlike the TypeScript SDK (which defaults to http://localhost:3001), the Python SDK raises ValueError at setup time if no ingest URL is configured — there is no implicit localhost default, so a misconfigured server fails loudly at your call site instead of silently dropping telemetry into a dead port.

Attribute the end user (search by id / email)

The dashboard can search sessions by user id or email — but only if your server tells mcpeye who the end user is. MCP has no built-in end-user identity, so you pass identify: a callable resolved per tool call, on the request thread, so attribution is correct on a multi-user / stateless server where one flushed batch mixes users. Use contextvars for the per-request value:

import contextvars, mcpeye

_user = contextvars.ContextVar("mcpeye_user", default=None)
# In your request handler: _user.set({"id": uid, "email": email})

mcpeye.track(
    server,
    "your-project-id",
    ingest_url="http://localhost:3001",
    identify=lambda: (_user.get() or {}),   # returns {"userId": ..., "userEmail": ...}
)

Return {"userId": ..., "userEmail": ...}. Pass an opaque, stable userId; userEmail is optional and is PII you store only in your own deployment. (For a single-user process you can instead pass static user_id=/user_email=.) Without either, sessions read "user not identified" and search-by-user returns nothing.

Configuration

Argument	Env fallback	Default
ingest_url	MCPEYE_INGEST_URL	— (required, raises if unset)
ingest_secret	MCPEYE_INGEST_SECRET	none
redact	—	True
identify	—	none — callable resolved PER CALL for userId/userEmail (see below)
user_id	—	none (static fallback)
user_email	—	none (static fallback)
client	—	none
server_version	—	server.version when available
flush_at	—	20 buffered events
flush_interval_s	—	5.0 seconds (background timer)
denylist_fields	—	none (adds to the built-in denylist)
capture_missing_capabilities	—	True (inject mcpeye_request_capability)
on_error	—	debug log on the mcpeye logger

Manifest cost. With capture_missing_capabilities=True, your server's tools/list gains one extra tool — a few hundred tokens in any model context that lists tools, and one more entry in any tool picker / doc generator. That is the price of seeing silent misses; pass False to keep it out of the manifest.

Diagnostics

Every swallowed error (transport failure, capture failure) is routed to on_error, which defaults to a debug-level log on logging.getLogger("mcpeye"). Pass your own sink to see why telemetry might be silent — it is wrapped so it can never throw back into your server:

mcpeye.track(server, "proj_123", ingest_url="http://localhost:3001",
             on_error=lambda err: print("mcpeye:", err))

Redaction

When redact=True (the default), arguments, results, and the reported intent are scrubbed client-side before anything leaves your process. The v1 redaction is a conservative regex pass — it over-redacts rather than leak — covering:

• emails
• API keys: sk-…, sk-ant-…, ghp_…/gho_…/etc., AKIA…
• Bearer … tokens and JWTs
• credit-card-shaped digit runs
• loose international phone numbers
• a field-name denylist (password, secret, token, apiKey, authorization, …)

Self-hosting is the real privacy mitigation; redaction reduces the blast radius of obvious secrets in free-text. You can use the primitives directly:

from mcpeye import redact_string, redact_value

redact_string("ping me at [e-mail]")            # -> "ping me at [REDACTED_EMAIL]"
redact_value({"password": "hunter2"})           # -> {"password": "[REDACTED_FIELD]"}

Manual instrumentation (wrap_tool)

The mcp package's Server internals vary across versions. track attaches to server.request_handlers (resolving FastMCP's inner server automatically); if a future layout isn't recognized, track does not raise — it logs a clear WARNING and returns the server uninstrumented (a version mismatch must never break your server's boot). A genuine config error (no ingest URL) still raises. Either way, you can fall back to instrumenting individual tool handlers:

import mcpeye

@mcpeye.wrap_tool(project_id="proj_123", tool_name="search",
                  ingest_url="http://localhost:3001")
async def search(arguments):
    ...

wrap_tool pulls mcpeyeIntent out of the arguments, times the call, captures the outcome, and ships it — the same capture path as track, scoped to one tool. It accepts redact, ingest_url, ingest_secret, denylist_fields, flush_at, flush_interval_s, and on_error (not user_id/client/server_version, which are server-level identity). Note: it does not inject the parameter into a published schema, so add mcpeyeIntent to that tool's inputSchema yourself (see mcpeye.intent_param_json_schema) if you want agents to populate it.

Development

Unit tests stub the mcp package, so they run with only httpx, pydantic, and pytest installed (no mcp):

python3 -m venv .venv && .venv/bin/pip install httpx pydantic pytest
.venv/bin/pip install -e packages/sdk-python --no-deps
.venv/bin/python -m pytest packages/sdk-python/tests -q

License

MIT