How to Implement Custom Trace Sinks for Observability in aisuite

Implementing custom trace sinks in aisuite requires creating a class with an emit(event: TraceEvent) method and registering it via ai.tracing.configure() to capture every agent run, model call, and tool invocation.

The aisuite framework provides a lightweight tracing system that streams observability data through pluggable sinks. To implement custom trace sinks for observability in aisuite, you implement the TraceSink protocol defined in aisuite/tracing/sinks.py and register your implementation with the global configuration. This architecture decouples your observability backend from the agent execution logic, allowing you to route trace events to files, HTTP endpoints, message queues, or custom analytics platforms.

Understanding the TraceSink Protocol

The foundation of aisuite's observability system rests on a simple protocol defined in aisuite/tracing/sinks.py. Any object implementing the TraceSink interface must provide a single method:

def emit(self, event: TraceEvent) -> None

This minimal contract allows the AgentRunner in aisuite/agents/runner.py to forward execution events without knowing the underlying storage mechanism. When an agent executes, the runner calls emit_event(sinks, event), which iterates through all configured sinks and invokes their emit method with the current TraceEvent.

Built-in Trace Sinks

aisuite ships with three production-ready implementations that demonstrate the protocol's flexibility.

LocalTraceSink

The LocalTraceSink persists events to a JSON Lines file on disk, defaulting to .aisuite/events.jsonl. It uses an internal JsonlTraceStore to handle file I/O, making it ideal for local development and debugging.

HttpTraceSink

For centralized observability, HttpTraceSink POSTs each TraceEvent to a remote HTTP endpoint. This enables integration with third-party APM tools and custom dashboards without modifying agent code.

InMemoryTraceSink

The InMemoryTraceSink accumulates events in a Python list, providing programmatic access to trace data during testing or short-lived scripts. This sink is particularly useful for unit tests verifying agent behavior.

Creating a Custom Trace Sink

To implement custom trace sinks for observability in aisuite, subclass the protocol and define your persistence logic.

File-Based Custom Sink

The following example writes events to a custom JSON file with specific formatting:

import json
from pathlib import Path
from aisuite.tracing.sinks import TraceSink, TraceEvent, configure

class JsonFileSink:
    def __init__(self, file: str | Path):
        self.path = Path(file)
        self.path.parent.mkdir(parents=True, exist_ok=True)

    def emit(self, event: TraceEvent) -> None:
        with self.path.open("a", encoding="utf-8") as f:
            json.dump(event.to_dict(), f)
            f.write("\n")

configure(JsonFileSink("my_custom_trace.jsonl"))

Message Queue Integration

For distributed systems, implement a sink that publishes to RabbitMQ:

import json
import pika
from aisuite.tracing.sinks import TraceSink, TraceEvent, configure

class RabbitMQSink:
    def __init__(self, url: str, exchange: str = "aisuite.traces"):
        self.conn = pika.BlockingConnection(pika.URLParameters(url))
        self.channel = self.conn.channel()
        self.exchange = exchange
        self.channel.exchange_declare(exchange=self.exchange, durable=True)

    def emit(self, event: TraceEvent) -> None:
        body = json.dumps(event.to_dict()).encode()
        self.channel.basic_publish(
            exchange=self.exchange,
            routing_key="trace",
            body=body,
            properties=pika.BasicProperties(content_type="application/json")
        )

configure(RabbitMQSink("amqp://guest:guest@localhost:5672/"))

Registering and Configuring Sinks

The global registry in aisuite/tracing/sinks.py maintains the list of active sinks via the _configured_sinks list. Use the configure() function to register your implementations before initializing any agents:

from aisuite.tracing.sinks import InMemoryTraceSink, LocalTraceSink, configure

mem = InMemoryTraceSink()
disk = LocalTraceSink("runs/events.jsonl")
configure(mem, disk)  # Both receive every event

Calling configure() replaces the current sink list entirely. The get_configured_sinks() function retrieves the current configuration, which the AgentRunner uses as a fallback when no explicit sinks are provided.

Per-Run Sink Configuration

For temporary observability without global side effects, pass sinks directly to AgentRunner:

from aisuite.tracing.sinks import InMemoryTraceSink
from aisuite.agents.runner import AgentRunner

temp_sink = InMemoryTraceSink()
runner = AgentRunner(..., trace_sinks=[temp_sink])
runner.run()
print(temp_sink.events)  # Inspect captured events

This approach bypasses the global registry, ensuring that only the specified sink receives events for that particular execution.

Summary

• Implement the protocol: Create a class with an emit(self, event: TraceEvent) method to satisfy the TraceSink interface defined in aisuite/tracing/sinks.py.
• Register globally: Use ai.tracing.configure() to activate your sink for all subsequent agent runs.
• Use built-in options: Leverage LocalTraceSink, HttpTraceSink, or InMemoryTraceSink for common observability patterns.
• Isolate per-run: Pass trace_sinks directly to AgentRunner for temporary or test-specific logging.
• Access trace data: The TraceEvent object contains all execution metadata; call to_dict() for serialization.

Frequently Asked Questions

What methods must a custom trace sink implement?

A custom trace sink must implement only the emit(self, event: TraceEvent) -> None method as defined by the TraceSink protocol in aisuite/tracing/sinks.py. This minimal interface allows aisuite to forward execution events without imposing storage or transport requirements.

How do I route trace events to multiple destinations simultaneously?

Pass multiple sink instances to configure() in aisuite/tracing/sinks.py. The framework iterates through all configured sinks via emit_event(), calling each sink's emit method with every TraceEvent. You can combine file-based, HTTP, and custom sinks in a single configuration.

Can I use different trace sinks for different agent runs?

Yes. Instead of using the global configuration, instantiate AgentRunner from aisuite/agents/runner.py with the trace_sinks parameter set to a list of specific sink instances. This overrides the global registry for that runner instance only, allowing isolated observability per execution.

Where does aisuite store trace events by default?

By default, aisuite uses LocalTraceSink to write events to .aisuite/events.jsonl in the current working directory. This behavior activates when no custom sinks are configured via configure() or when AgentRunner falls back to get_configured_sinks().

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}