mcp-observe 0.1.1

One-line SDK that captures every MCP tool call and sends it to a dashboard.

mcp-observe

One-line SDK that captures every LLM and MCP tool call and streams it to a real-time dashboard.

Zero runtime dependencies — uses only the Python standard library. Patches are silently skipped when a library isn't installed.

Install

pip install mcp-observe

Get Your API Key

Create a free account at app.themcp.company to get your API key (tap_...).

Quick Start

import mcp_observe
mcp_observe.init(api_key="tap_...")

# That's it. All supported SDK calls are now captured automatically.

Supported Providers

Provider	What's captured
Anthropic (anthropic)	messages.create (sync + async) — model, system prompt, messages, response, thinking blocks, token usage
OpenAI (openai)	chat.completions.create (sync + async) — model, messages, response, reasoning content, token usage
Google Gemini (google.genai)	generate_content (sync + async + streaming) — model, contents, response, token usage
MCP (mcp)	ClientSession.call_tool — tool name, arguments, result, errors
Composio (composio)	ComposioToolSet.execute_action — action name, parameters, result
HTTP (httpx, urllib3, aiohttp)	Requests to known AI API hosts (OpenAI, Anthropic, Gemini)

All patches are optional. If a library isn't installed, its patch is silently skipped — your app is never affected.

Server-Side Instrumentation

Instrument MCP servers directly to capture tool calls on the server side:

from mcp.server import Server
import mcp_observe

app = Server("my-server")
mcp_observe.instrument_server(app, api_key="tap_...")

Proxy CLI

Wrap any MCP server as a transparent proxy — no code changes needed:

mcp-observe --api-key tap_... -- python my_mcp_server.py

The proxy intercepts all JSON-RPC tools/call messages on stdio, captures them, and forwards everything unchanged.

Sessions

Sessions are detected automatically from conversation fingerprints. You can also set them explicitly:

import mcp_observe

# Context manager (scoped)
with mcp_observe.session("my-workflow") as session_id:
    # All events inside this block share the same session ID
    response = client.messages.create(...)

# Manual start/end
mcp_observe.start_session("my-workflow")
# ... your code ...
mcp_observe.end_session()

Traces

Group related operations across sessions with trace IDs:

import mcp_observe

trace_id = mcp_observe.start_trace()
# All events emitted until end_trace() share this trace_id
response = client.messages.create(...)
mcp_observe.end_trace()

Configuration

init() Parameters

Parameter	Type	Description
api_key	str	Your project API key (starts with tap_). Falls back to MCP_OBSERVE_API_KEY env var.
endpoint	str	Custom endpoint URL. Falls back to MCP_OBSERVE_ENDPOINT env var.
user_id	str | None	User ID to tag events with.
signing_key_path	str | None	Path to Ed25519 private key PEM for AARM cryptographic receipts. Requires pip install mcp-observe[aarm].

Environment Variables

Variable	Description
MCP_OBSERVE_API_KEY	API key fallback (used when api_key is not passed to init())
MCP_OBSERVE_ENDPOINT	Custom endpoint URL fallback

Debugging

If events aren't appearing in the dashboard:

1. Enable debug logging:

   import logging
   logging.basicConfig(level=logging.DEBUG)
   

   The SDK logs to the mcp_observe logger. You'll see messages for patch failures and send errors.

2. Verify your API key is correct and starts with tap_.

3. Make sure mcp_observe.init() is called before importing the SDK you want to instrument.

Dashboard

Sign up at app.themcp.company to get your API key and view events in real time.

Tests

python3 -m unittest discover -s tests

License

MIT