API Reference

Complete SDK reference for the spooled Python package (v0.3.1).

spooled.init()

Initialize Spooled and start recording.

def init(
    agent_id: Optional[str] = None,
    *,
    auto_instrument: bool = False,
    session_id: Optional[str] = None,
    parent_run_id: Optional[str] = None,
    correlation_context: Optional[Dict[str, str]] = None,
    sample_rate: Optional[float] = None,
    tags: Optional[list] = None,
    exporters: Optional[list] = None,
    **kwargs: Any,
) -> Recorder

Parameter	Type	Default	Description
agent_id	Optional[str]	None	Agent identifier. Falls back to SPOOLED_AGENT_ID env var, then "default"
auto_instrument	bool	False	Enable monkey-patching of supported libraries (deprecated — use explicit wrappers)
session_id	Optional[str]	None	Multi-agent session correlation ID
parent_run_id	Optional[str]	None	Parent trace ID for child/spawned agents
correlation_context	Optional[Dict[str, str]]	None	Key-value context for multi-agent correlation
sample_rate	Optional[float]	None	Override SPOOLED_SAMPLE_RATE (0.0–1.0)
tags	Optional[list]	None	Tags for fleet grouping and filtering
exporters	Optional[list]	None	Custom exporters (e.g., SpooledOTELExporter)

spooled.shutdown()

def shutdown(success: bool = True) -> None

Flush and close the recorder. Call this when your agent finishes. Pass success=False if the agent failed.

spooled.spawn()

def spawn(agent_id: str, **kwargs: Any) -> Recorder

Create a child recorder from the global recorder. The child shares the same session_id and links via parent_run_id. Used for multi-agent orchestration.

Decorators

@spooled.trace()

@spooled.trace(agent_id="my_agent")
def run_agent(query: str):
    ...

Wraps a function as a traced agent run. Automatically calls init() before and shutdown() after. Recommended over manual init/shutdown.

@spooled.tool()

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

Marks a function as a tool call. Records input shapes, output shapes, latency, and errors.

@spooled.observe()

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

Marks a function as an observation point. Records execution without modifying behavior.

Recorder class

class Recorder:
    def __init__(
        self,
        agent_id: str,
        backend_url: Optional[str] = None,
        buffer_size: int = 100,
        flush_interval: float = 5.0,
        session_id: Optional[str] = None,
        parent_run_id: Optional[str] = None,
        correlation_context: Optional[Dict[str, str]] = None,
        sample_rate: Optional[float] = None,
        tags: Optional[List[str]] = None,
        fingerprint_mode: Optional[str] = None,
        exporters: Optional[List[Any]] = None,
    ) -> None

record_interaction()

def record_interaction(
    self,
    interaction_type: InteractionType,
    input_data: Dict[str, Any],
    output_data: Optional[Dict[str, Any]] = None,
    error: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    parallel_group: Optional[str] = None,
) -> Interaction

Parameter	Type	Description
interaction_type	InteractionType	LLM_CALL, TOOL_CALL, HTTP_REQUEST, or OTHER
input_data	Dict[str, Any]	Input to the interaction (prompt, function args, URL)
output_data	Optional[Dict]	Output from the interaction (response, result)
error	Optional[str]	Error message if the interaction failed
metadata	Optional[Dict]	Additional metadata (latency_ms, tokens, status_code)
parallel_group	Optional[str]	Group ID for concurrent interactions

InteractionType enum

class InteractionType(str, Enum):
    LLM_CALL = "LLM_CALL"
    TOOL_CALL = "TOOL_CALL"
    HTTP_REQUEST = "HTTP_REQUEST"
    OTHER = "OTHER"

Auto-instrumented libraries

Library	Sync	Async
OpenAI	✓	✓
Anthropic	✓	✓
requests	✓	—
httpx	✓	✓
aiohttp	—	✓
AWS Bedrock	✓	—

Framework integrations

Framework	Integration class
LangChain	SpooledLangChainCallbackHandler
LlamaIndex	SpooledLlamaIndexCallbackHandler
AutoGen	install_autogen_hook(recorder)

Install with pip install spooled-ai[integrations]. See Frameworks for usage details.