SDK Integration

Complete guide to integrating the Spooled SDK into your agent.

Basic setup

import spooled

spooled.init(agent_id="my_agent")

# ... your agent code ...

spooled.shutdown(success=True)

Decorator API (recommended)

import spooled

@spooled.trace(agent_id="my_agent")
def run_agent(query: str):
    result = call_llm(query)
    data = search_tool(result)
    return summarize(data)

# Shutdown is handled automatically by the decorator

Mark individual functions as tools or observation points:

@spooled.tool()
def search(query: str):
    return api.search(query)

@spooled.observe()
def validate(result):
    assert result.is_valid

init() options

Parameter	Type	Default	Description
agent_id	Optional[str]	None	Falls back to SPOOLED_AGENT_ID, then "default"
session_id	Optional[str]	None	Multi-agent session correlation ID
parent_run_id	Optional[str]	None	Parent trace ID for child agents
correlation_context	Optional[Dict]	None	Key-value context for multi-agent sessions
sample_rate	Optional[float]	None	Override SPOOLED_SAMPLE_RATE (0.0–1.0)
tags	Optional[list]	None	Tags for fleet grouping
exporters	Optional[list]	None	Custom exporters (e.g., OTEL)

Multi-agent orchestration

Use spooled.spawn() to create child recorders that share a session:

import spooled

spooled.init(agent_id="orchestrator")

# Spawn a child agent — linked via session_id + parent_run_id
child = spooled.spawn(agent_id="researcher")
# child is a Recorder instance with its own trace
child.record_interaction(...)
child.stop()

spooled.shutdown(success=True)

Child traces are correlated via session_id and linked via parent_run_id, so behavioral changes in child agents are isolated and tracked independently.

Fleet tagging

spooled.init(
    agent_id="payment_agent",
    tags=["production", "finserv", "v2"]
)

Or via environment variable: SPOOLED_AGENT_TAGS=production,finserv,v2

Sampling

For high-volume agents, sample a fraction of runs:

spooled.init(agent_id="high_volume", sample_rate=0.1)  # Record 10% of runs

Or via SPOOLED_SAMPLE_RATE=0.1. Unsampled runs produce no trace.

OpenTelemetry export

from spooled.exporters import SpooledOTELExporter

spooled.init(
    agent_id="my_agent",
    exporters=[SpooledOTELExporter()]
)

Traces are exported via OTLP to your configured collector (Datadog, Grafana, Honeycomb, etc.).

Fingerprint mode

Mode	Behavior
sequence (default)	Order-sensitive. Hash includes the exact sequence of interactions.
structural	Order-insensitive. Hash includes the sorted unique set. Good for ReAct-style loops.

spooled.init(agent_id="react_agent", fingerprint_mode="structural")

Or via SPOOLED_FINGERPRINT_MODE=structural.

Privacy posture

On initialization, Spooled logs its privacy posture:

Spooled privacy posture
  agent_id: my_agent
  local_storage: true
  backend_url: (none)
  capture_mode: structure-only (content is never captured)
  data_sent_to_backend: none (local-only mode)

Call content (prompts, responses, tool argument values) is stripped at the SDK level. See Privacy Architecture for exactly what is and isn't transmitted.

Manual Recorder API

For full control, use the Recorder directly:

from spooled.recorder import Recorder
from spooled.models import InteractionType

recorder = Recorder(
    agent_id="my_agent",
    buffer_size=100,
    flush_interval=5.0,
)
recorder.start()

recorder.record_interaction(
    interaction_type=InteractionType.LLM_CALL,
    input_data={"model": "gpt-4", "prompt": "..."},
    output_data={"response": "..."},
    metadata={"latency_ms": 250, "tokens": 1500},
)

recorder.stop(status=TraceStatus.SUCCESS)

Note

Spooled fails open — if the SDK encounters an error, it silently degrades rather than blocking your agent.