API reference

The Python SDK exposes a single class – SpanlyClient – plus a MonitorOptions dataclass for hooks.

`SpanlyClient`

from spanly import SpanlyClient

client = SpanlyClient()  # reads SPANLY_API_KEY from env

`SpanlyClient(api_key=None, ingest_url=None)`

Argument	Type	Required	Description
`api_key`	`str \| None`	no	Your Spanly API key. Defaults to `os.environ["SPANLY_API_KEY"]`. Raises `ValueError` if neither is set.
`ingest_url`	`Callable[[SpanlyRegion], str] \| None`	no	Override the ingest endpoint. Used internally for E2E tests.

Region (us / eu) is parsed from the API key prefix. Invalid prefixes raise ValueError.

`client.monitor(server, options=None)`

Patch the server's run() method to capture every JSON-RPC frame on the read and write streams.

client.monitor(server)
# or with hooks
client.monitor(server, MonitorOptions(on_error=lambda e: print(e)))

Accepts either:

The high-level MCPServer / FastMCP instance (detected via _lowlevel_server).
The low-level mcp.server.Server instance.

Must be called before server.run(). The patch is in-place and process-local; you do not need to detach it on shutdown.

`MonitorOptions`

A simple dataclass:

from spanly import MonitorOptions, CollectWarning, SpanlyPacketContext

options = MonitorOptions(
    on_error=...,
    on_warning=...,
    on_collect=...,
)

`on_collect(direction, context, packet) -> dict | None`

def on_collect(
    direction: str,  # "from-client" | "to-client"
    context: SpanlyPacketContext,
    packet: dict,
) -> dict | None:
    ...

Called for every captured packet before it is queued for ingest. Return None to drop the packet. Return the (possibly mutated) dict to queue it.

Common writes on context:

context.environment_id = current_tenant_id()  # multi-tenant tagging
context.user_id = request.user.id              # join with auth
context.span_id = current_otel_span_id()       # APM correlation

`on_error(exc)`

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

Called when the SDK itself fails (network error, malformed packet). The SDK never re-raises into your server – it always routes the error through this hook.

`on_warning(warnings)`

def on_warning(warnings: list[CollectWarning]) -> None:
    ...

Called when the SDK detects something recoverable (unmatched response, malformed JSON, queue overflow). CollectWarning is a small dataclass with code: str and message: str.

Types

The package re-exports the types you'd want for typed hooks:

from spanly import (
    SpanlyClient,
    SpanlyRegion,         # Literal["us", "eu"]
    MonitorOptions,
    CollectWarning,
    SpanlyPacket,
    SpanlyPacketContext,
    McpServerInfo,
)

SpanlyPacket – the envelope sent to ingest. You won't usually touch this directly.
SpanlyPacketContext – mutable context: server/client identity, transport, custom fields like environment_id, user_id, span_id.
McpServerInfo – name + version from the MCP initialize handshake.

What the SDK does not do

It does not modify the responses your server returns. The captured payload is forwarded onward to the client unchanged before being copied for ingest.
It does not block your server on its network call to ingest. Capture is queued on a background task; if the queue is full, the oldest packet is dropped and on_warning fires.
It does not require any specific transport. stdio, HTTP, and any custom anyio stream work because monitor() wraps the streams passed to run().
It does not break on cancellation. The internal sender shields its flush from cancellation so packets aren't lost when the MCP session manager cancels its task group in stateless mode.

On this page