Examples

Claude Desktop / Cursor / Windsurf

These hosts spawn your MCP server as a child process over stdio. The Python SDK is ASGI middleware, so it does not apply to a stdio server; wrap it with the Spanly CLI instead:

npx -y @spanly/spanly run -- python -m my_mcp

If your server also exposes an HTTP transport for other clients, mount SpanlyMiddleware on that ASGI app as usual; the two integration methods are independent.

Multi-tenant attribution

When the same MCP server handles requests from many tenants, attach the tenant id to captured traffic with identity.resolve. The resolved user's account_id is exactly the tenant-scoped field the dashboard's filter bar picks up automatically:

import os

from spanly import IdentityOptions, ResolveContext, SpanlyMiddleware


def resolve(ctx: ResolveContext):
    tenant = extract_tenant_from_auth(ctx.headers.get("authorization"))
    if tenant is None:
        return None
    return {"id": tenant.user_id, "accountId": tenant.org_id}


app.add_middleware(
    SpanlyMiddleware,
    api_key=os.environ["SPANLY_API_KEY"],
    identity=IdentityOptions(resolve=resolve),
)

resolve runs as a background task started at the transaction, and never blocks serving; a packet emitted before resolution settles simply goes out without a user, and the batcher backfills the rest of that session downstream.

What the middleware does not filter

There is no hook to drop or rewrite an individual captured packet: the middleware is a byte-copy tee, not an interception point for your JSON-RPC payloads. If you need to keep certain tools or paths out of Spanly entirely, scope capture at the transport level instead:

app.add_middleware(
    SpanlyMiddleware,
    api_key=os.environ["SPANLY_API_KEY"],
    paths=["/mcp/public"],  # /mcp/internal is never inspected
)

This is coarser than per-packet filtering: it excludes a whole path prefix, not a specific tool call. Credential-bearing headers are always redacted automatically; there is nothing to configure there.

Error reporting via Sentry

import os

import sentry_sdk
from spanly import SpanlyMiddleware

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


def on_error(exc: Exception) -> None:
    sentry_sdk.capture_exception(exc)


app.add_middleware(
    SpanlyMiddleware,
    api_key=os.environ["SPANLY_API_KEY"],
    on_error=on_error,
)

The middleware never raises into your app, so capturing through on_error is the only way to surface its internal failures (it falls back to logging through the spanly logger when on_error is omitted).

Tests: capture locally with `ingest_url`

There is no client-side queue to flush on a fixed schedule; each packet is delivered on a background task as soon as its request or response leg finishes. For a local test, point ingest_url at a server that records the payloads and drive the app through an ASGI test client:

import asyncio
import json
import os
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

import httpx
import pytest
from spanly import SpanlyMiddleware


class _CaptureHandler(BaseHTTPRequestHandler):
    received: list[dict] = []

    def do_POST(self):
        length = int(self.headers["Content-Length"])
        self._CaptureHandler.received.append(json.loads(self.rfile.read(length)))
        body = json.dumps({"success": True}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass  # keep test output quiet


@pytest.mark.asyncio
async def test_captures_a_tool_call(my_asgi_app):
    server = HTTPServer(("127.0.0.1", 0), _CaptureHandler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    port = server.server_address[1]

    app = SpanlyMiddleware(
        my_asgi_app,
        api_key="spanly_us_test",
        ingest_url=lambda region: f"http://127.0.0.1:{port}",
    )

    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test"
    ) as client:
        await client.post(
            "/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/call"}
        )

    await asyncio.sleep(0.1)  # let the background delivery task run
    assert _CaptureHandler.received

    server.shutdown()

Use on_error in tests to fail loudly when delivery breaks instead of silently losing packets.

Correlate with your APM (Datadog, Sentry, …)

If your infrastructure propagates a W3C traceparent value, on an HTTP header or embedded in params._meta.traceparent on the JSON-RPC message, it survives untouched in the captured packet: neither is stripped or rewritten before the packet leaves your process. Pick your APM provider in Settings, Integrations and the request detail view links straight to the corresponding trace in your APM. See Trace context propagation.