listo-mcp-observability 0.0.2

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 Insights in production
• MCP tool tracking — Decorator-based (@obs.track_tool()) or programmatic wrapping
• Payload sanitization — Redacts password, token, apiKey, secret, authorization
• Sampling — Configurable rates with guaranteed error/session capture
• Zero runtime dependencies — Core uses only Python stdlib

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
INSIGHTS_API_URL	Listo Insights API endpoint	—
INSIGHTS_API_KEY	API key for authentication	—
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, properties, status)

obs.record_business_event("purchase", {"amount": 99.99}, "ok")

obs.record_session(action, session_id)

obs.record_session("open", "session-123")

obs.record_ui_event(name, ...)

obs.record_ui_event(name="click", action="button_press", widget_id="w1")

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     — UI event ingestion

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

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"],
))

Development

# Install dependencies
uv sync --all-extras

# Run tests
uv run pytest -v

# Lint
uv run ruff check src/ tests/

# Build
uv build

License

MIT