listo-mcp-observability 0.3.0

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

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

Variable	Description	Default
LISTO_API_URL	Listo API endpoint (primary)	—
LISTO_API_KEY	API key for authentication (primary)	—
INSIGHTS_API_URL	Deprecated, use LISTO_API_URL	—
INSIGHTS_API_KEY	Deprecated, use LISTO_API_KEY	—
HMAC_SECRET	Secret key for HMAC-SHA-256 hashing of session_id and user_id	Falls back to SHA-256
ENVIRONMENT or PYTHON_ENV	Controls dev/production defaults	dev

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:

Type	Description
http_request	HTTP request lifecycle
mcp_request	MCP tool call, resource read, prompt get
mcp_session	Session open/close (always captured)
business_event	Domain-specific events with optional category, session, tenant tracking
ui_event	Frontend interactions with optional category

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:

Page	Description
Overview	Architecture, quick start
Installation	pip install, prerequisites
Easy Setup	create_mcp_observability_easy() deep-dive
Configuration	Both init methods, full config tables
Wrap MCP Handlers	wrap_mcp_handler(), track_tool(), context fields
Sessions	record_session(), always-capture behavior
Business Events	record_business_event(), EventCategory
UI Events	record_ui_event() + POST endpoint
Sinks	InMemorySink, RemoteSink, ConsoleSink, combine_sinks
Endpoints	Dashboard, JSON API, event ingestion
Sampling & Privacy	Sampling, redaction, data minimization
Advanced	Custom sinks, metrics API
Platform Detection	Detect Claude, ChatGPT, etc.
Troubleshooting	Common issues, migration from v0.0.x

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