# How to Implement Custom Trace Sinks for Observability in aisuite

> Learn to implement custom trace sinks in aisuite. Capture agent runs, model calls, and tool invocations by registering your own emit method with ai.tracing.configure and enhance your observability.

- Repository: [Andrew Ng/aisuite](https://github.com/andrewyng/aisuite)
- Tags: how-to-guide
- Published: 2026-06-15

---

**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`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/aisuite/tracing/sinks.py). Any object implementing the `TraceSink` interface must provide a single method:

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

```

This minimal contract allows the `AgentRunner` in [`aisuite/agents/runner.py`](https://github.com/andrewyng/aisuite/blob/main/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:

```python
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:

```python
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`](https://github.com/andrewyng/aisuite/blob/main/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:

```python
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`:

```python
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`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/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`](https://github.com/andrewyng/aisuite/blob/main/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()`.