Metadata-Version: 2.4
Name: listo-mcp-observability
Version: 0.0.2
Summary: Lightweight telemetry SDK for MCP servers. Captures MCP tool invocations, HTTP requests, business events, and UI interactions with built-in payload sanitization.
Project-URL: Homepage, https://github.com/Listo-Labs-Ltd/mcp-observability-python
Project-URL: Repository, https://github.com/Listo-Labs-Ltd/mcp-observability-python
Project-URL: Issues, https://github.com/Listo-Labs-Ltd/mcp-observability-python/issues
Author: Listo Labs Ltd
License-Expression: MIT
License-File: LICENSE
Keywords: analytics,mcp,model-context-protocol,monitoring,observability,telemetry
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Monitoring
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: mypy>=1.13; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.8; extra == 'dev'
Provides-Extra: mcp
Requires-Dist: mcp>=1.0; extra == 'mcp'
Provides-Extra: starlette
Requires-Dist: starlette>=0.27; extra == 'starlette'
Description-Content-Type: text/markdown

# 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](https://github.com/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

```bash
# 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

```python
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:

```python
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:

```python
@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:

```python
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)`

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

### `obs.record_session(action, session_id)`

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

### `obs.record_ui_event(name, ...)`

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

### Dashboard Endpoints (Starlette)

```python
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

```python
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:

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

## Development

```bash
# Install dependencies
uv sync --all-extras

# Run tests
uv run pytest -v

# Lint
uv run ruff check src/ tests/

# Build
uv build
```

## License

MIT
