Python API

def init(
    api_key: str,                          # Project API key (required)
    user_id: str | None = None,            # Optional user identifier
    endpoint: str = "<default>",           # Event ingestion URL
    signing_key_path: str | None = None,   # Path to Ed25519 PEM for AARM receipts
) -> None

import mcp_tap

mcp_tap.init(
    api_key="tap_abc123",
    user_id="user@example.com",
)

def instrument_server(
    server: object,                  # MCP Server instance
    api_key: str,                    # Project API key (required)
    endpoint: str = "<default>",     # Event ingestion URL
) -> None

from mcp.server import Server
import mcp_tap

app = Server("my-server")
mcp_tap.instrument_server(app, api_key="tap_abc123")

import mcp_tap

with mcp_tap.session("user-query-123") as session_id:
    response = client.messages.create(...)

import mcp_tap

sid = mcp_tap.start_session("my-session")
# ... calls here are tagged with sid ...
mcp_tap.end_session()

from mcp_tap.emitter import Emitter

emitter = Emitter(api_key, endpoint)
emitter.emit(event_dict)

from mcp_tap.emitter import make_event, generate_event_id

event = make_event(
    project_id="tap_abc123",
    session_id="ses_abc123",
    user_id=None,
    tool_name="get_weather",
    inputs={"city": "SF"},
    output={"temp": 72},
    latency_ms=150,
    source="custom",
)

from mcp_tap.sessions import generate_session_id

from mcp_tap.receipts import ContextChain, ReceiptSigner, hash_output, enhance_event

chain = ContextChain()
context_hash = chain.chain(event_dict)

signer = ReceiptSigner(private_key_pem_bytes)
sig = signer.sign(event_dict)
# → {"algorithm": "Ed25519", "key_id": "...", "value": "base64..."}

output_hash = hash_output(some_output)

enhance_event(event_dict, context_chain=chain, signer=signer)