Examples

Claude Desktop / Cursor / Windsurf

These hosts spawn your Python MCP server as a child process over stdio. Wire the SDK in before server.run():

import asyncio
import os
from mcp.server import Server
from mcp.server.stdio import stdio_server
from spanly import SpanlyClient


async def main() -> None:
    server = Server("my-server")
    # … register tools, prompts, resources …

    SpanlyClient(api_key=os.environ["SPANLY_API_KEY"]).monitor(server)

    async with stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())


if __name__ == "__main__":
    asyncio.run(main())

Pass the API key through the host config:

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["-m", "my_server"],
      "env": {
        "SPANLY_API_KEY": "spanly_us_xxxxxxxxxxxxxxxxxxxxxx"
      }
    }
  }
}

Multi-tenant tagging with `contextvars`

When the same MCP server handles requests from many tenants, attach the tenant id per request:

import contextvars
from spanly import SpanlyClient, MonitorOptions

current_tenant: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "current_tenant", default=None,
)


def on_collect(direction, context, packet):
    tenant = current_tenant.get()
    if tenant:
        context.environment_id = tenant
    return packet


SpanlyClient().monitor(server, MonitorOptions(on_collect=on_collect))

Set current_tenant in your transport / auth middleware. The dashboard's filter bar picks up environment_id automatically.

Filtering sensitive tools

Drop a packet entirely by returning None:

from spanly import MonitorOptions


def on_collect(direction, context, packet):
    if packet.get("method") == "tools/call":
        name = packet.get("params", {}).get("name")
        if name == "internal-debug-tool":
            return None  # never sent to Spanly
    return packet


SpanlyClient().monitor(server, MonitorOptions(on_collect=on_collect))

Or redact arguments in place:

def on_collect(direction, context, packet):
    params = packet.get("params", {})
    args = params.get("arguments", {})
    if "api_key" in args:
        args["api_key"] = "[REDACTED]"
    return packet

Correlate with OpenTelemetry

from opentelemetry import trace
from spanly import SpanlyClient, MonitorOptions

tracer = trace.get_tracer(__name__)


def on_collect(direction, context, packet):
    span = trace.get_current_span()
    span_context = span.get_span_context()
    if span_context.is_valid:
        context.trace_id = format(span_context.trace_id, "032x")
        context.span_id = format(span_context.span_id, "016x")
    return packet


SpanlyClient().monitor(server, MonitorOptions(on_collect=on_collect))

Click into a request in the Spanly dashboard and the trace_id field becomes the link back to your APM tool's trace view.

Error reporting via Sentry

import sentry_sdk
from spanly import SpanlyClient, MonitorOptions

sentry_sdk.init(dsn=os.environ["SENTRY_DSN"])


def on_error(exc):
    sentry_sdk.capture_exception(exc)


SpanlyClient().monitor(server, MonitorOptions(on_error=on_error))

The SDK never re-raises into your server, so capturing through on_error is the only way to surface its internal failures.

Claude Desktop / Cursor / Windsurf

Multi-tenant tagging with contextvars

Filtering sensitive tools

Correlate with OpenTelemetry

Error reporting via Sentry

On this page

Multi-tenant tagging with `contextvars`