listo-mcp-observability

Lightweight telemetry SDK for Python MCP servers. Captures MCP tool invocations, HTTP requests, business events, and UI interactions with built-in payload sanitization.

Python port of @listo-labs-ltd/mcp-observability (TypeScript).

Features

• Easy setup — 3 lines with create_mcp_observability_easy()
• Local dashboard — Real-time metrics at /telemetry/dashboard
• Centralized analytics — Send to Listo in production
• MCP tool tracking — Decorator-based (@obs.track_tool()) or programmatic wrapping
• Payload sanitization — Redacts password, token, apiKey, secret, authorization, userid, email, phone, and more
• Sampling — Configurable rates with guaranteed error/session capture
• Zero runtime dependencies — Core uses only Python stdlib
• Event categories — Classify business and UI events as conversion, engagement, impression, navigation, or system
• Batch ingestion — POST /telemetry/events for efficient multi-event ingestion
• Platform detection — Identify which AI platform (Claude, ChatGPT, etc.) initiated each session and tool call
• Sink composition — combine_sinks() to route events to multiple destinations

Installation

# From PyPI
pip install listo-mcp-observability

# With uv
uv add listo-mcp-observability

# With optional MCP SDK support
pip install "listo-mcp-observability[mcp]"

# With optional Starlette dashboard support
pip install "listo-mcp-observability[starlette]"

Quick Start

from mcp.server.fastmcp import FastMCP
from mcp_observability import create_mcp_observability_easy

mcp = FastMCP("my-server")
obs = create_mcp_observability_easy(service_name="my-server")

@mcp.tool()
@obs.track_tool("search_hotels")
async def search_hotels(query: str) -> str:
    return f"Results for {query}"

Environment Variables

API

create_mcp_observability_easy()

Recommended entry point with sensible defaults:

obs = create_mcp_observability_easy(
    service_name="my-server",
    service_version="1.0.0",
    environment="dev",          # "dev" | "staging" | "production"
    sample_rate=1.0,            # 0.0 to 1.0 (errors always captured)
)

Dev mode (default): Console logging (errors only), in-memory dashboard, 100% sampling.

Production mode: No console, remote sink only, 10% sampling.

@obs.track_tool(name)

Decorator for tracking MCP tool calls:

@mcp.tool()
@obs.track_tool("my_tool")
async def my_tool(arg: str) -> str:
    ...

obs.wrap_mcp_handler(request_kind, handler, context)

Programmatic handler wrapping:

wrapped = obs.wrap_mcp_handler(
    "CallTool",
    original_handler,
    context=lambda *args, **kwargs: McpTrackingContext(
        tool_name="search",
        args=kwargs,
    ),
)

obs.record_business_event(name, *, ...)

obs.record_business_event(
    "purchase",
    properties={"amount": 99.99},
    status="ok",
    category="conversion",
    session_id="sess-123",
    user_id="user-456",
    tenant_id="tenant-789",
)

detect_platform(hints)

Detect which AI platform is connecting to your MCP server:

from mcp_observability import detect_platform, PlatformHints, McpClientInfo

platform = detect_platform(PlatformHints(
    headers={"User-Agent": "Claude-Desktop/2.0"},
))
# → "claude"

platform = detect_platform(PlatformHints(
    client_info=McpClientInfo(name="chatgpt"),
))
# → "chatgpt"

Three-tier detection (first match wins):

1. HTTP headers — user-agent and vendor-specific headers (x-openai-*, x-anthropic-*)
2. _meta key prefixes — openai/, anthropic/, claude/
3. clientInfo.name — exact or prefix match

Returns "claude", "chatgpt", or "other".

obs.record_session(action, session_id, *, platform=None)

obs.record_session("open", "session-123", platform="claude")

obs.record_ui_event(name, *, ...)

obs.record_ui_event(
    "click",
    action="button_press",
    category="engagement",
    widget_id="w1",
    user_id="user-456",
)

EventCategory

Type alias for valid event categories:

from mcp_observability import EventCategory
# "conversion" | "engagement" | "impression" | "navigation" | "system"

combine_sinks(*sinks)

Combine multiple sinks into one with error isolation:

from mcp_observability import combine_sinks, ConsoleSink, InMemorySink

combined = combine_sinks(ConsoleSink(), InMemorySink())

Dashboard Endpoints (Starlette)

from mcp_observability import create_telemetry_routes

routes = create_telemetry_routes()
# GET  /telemetry           — JSON metrics
# GET  /telemetry/dashboard — HTML dashboard
# POST /telemetry/event     — Single UI event ingestion (8KB max)
# POST /telemetry/events    — Batch UI event ingestion (max 100 events)

Remote Sink

from mcp_observability import RemoteSink, RemoteSinkOptions

sink = RemoteSink(RemoteSinkOptions(
    endpoint="https://api.listoai.co/v1/events/batch",
    api_key="your-key",
    batch_size=50,
    flush_interval=5.0,
    max_retries=3,
))

Payload Sanitization

The SDK automatically redacts sensitive fields (case-insensitive):

• password, token, apiKey, secret, authorization
• userId, userLocation, email, userEmail, phone, phoneNumber

Redaction is recursive (up to 6 levels deep) and limits arrays to 20 elements.

Custom redaction keys:

obs = McpObservability(ObservabilityOptions(
    service_name="my-server",
    redact_keys=["password", "token", "credit_card", "ssn"],
))

Event Types

The SDK captures five event types:

Security

• Sampling: Errors and sessions are always captured. All other events are subject to sample_rate.
• Redaction: Sensitive keys are recursively replaced with [redacted] up to depth 6.
• Data minimization: user_id and session_id are automatically hashed before emission. No user_location field exists in event types. User query text is truncated and hashed.

Documentation

Full documentation is available in the docs/ directory:

Development

# Install dependencies
uv sync --all-extras

# Run tests
uv run pytest -v

# Lint
uv run ruff check src/ tests/

# Type check
uv run mypy src/

# Build
uv build

License

MIT