Custom Tracing

Create custom trace sinks to send agent events anywhere

Send agent events to any destination – HTTP endpoints, databases, cloud services, or your own systems.

Diagram

Quick Start

Create Your Sink

Implement 3 methods: emit(), flush(), close()

python

class MySink:
    def emit(self, event):
        print(f"Event: {event.event_type.value}")

    def flush(self):
        pass

    def close(self):
        pass

Connect to Agent

Use trace_context() to capture all agent events

python

from praisonaiagents import Agent, ContextTraceEmitter, trace_context

sink = MySink()
emitter = ContextTraceEmitter(sink=sink, session_id="my-session", enabled=True)

with trace_context(emitter):
    agent = Agent(instructions="You are helpful")
    agent.chat("Hello!")  # Events go to MySink

How It Works

Diagram

The Protocol

Protocol

python

class ContextTraceSinkProtocol:
    def emit(self, event): ...   # Receive event
    def flush(self): ...         # Flush buffer
    def close(self): ...         # Release resources

Methods

Method	Purpose
`emit(event)`	Receive a trace event
`flush()`	Flush buffered events
`close()`	Release resources

Example Sinks

HTTP Sink

Send events to a remote HTTP endpoint with batching.

python

class HTTPSink:
    def __init__(self, url, batch_size=10):
        self.url = url
        self.buffer = []
        self.batch_size = batch_size

    def emit(self, event):
        self.buffer.append(event.to_dict())
        if len(self.buffer) >= self.batch_size:
            self.flush()

    def flush(self):
        if self.buffer:
            import requests
            requests.post(self.url, json=self.buffer)
            self.buffer.clear()

    def close(self):
        self.flush()

Database Sink

Store events in SQLite database for later analysis.

python

class SQLiteSink:
    def __init__(self, db_path="traces.db"):
        import sqlite3
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS events (
                id INTEGER PRIMARY KEY,
                session_id TEXT,
                event_type TEXT,
                agent_name TEXT,
                timestamp REAL,
                data TEXT
            )
        """)

    def emit(self, event):
        import json
        self.conn.execute(
            "INSERT INTO events VALUES (NULL, ?, ?, ?, ?, ?)",
            (event.session_id, event.event_type.value, 
             event.agent_name, event.timestamp, json.dumps(event.data))
        )

    def flush(self):
        self.conn.commit()

    def close(self):
        self.flush()
        self.conn.close()

Console Sink

Print events to console for debugging.

python

class ConsoleSink:
    def emit(self, event):
        print(f"[{event.event_type.value}] {event.agent_name or 'session'}")

    def flush(self):
        pass

    def close(self):
        pass

Event Types

Diagram

Event	Description
`session_start`	Session begins
`session_end`	Session ends
`agent_start`	Agent starts
`agent_end`	Agent finishes
`agent_handoff`	Agent hands off
`message_added`	Message added
`tool_call_start`	Tool starts
`tool_call_end`	Tool finishes
`llm_request`	LLM request sent
`llm_response`	LLM response received
`memory_store`	Memory stored
`memory_search`	Memory searched
`knowledge_add`	Knowledge added
`knowledge_search`	Knowledge searched

Event Data

python

event.event_type      # ContextEventType enum
event.timestamp       # Unix timestamp
event.session_id      # Session identifier
event.agent_name      # Agent name (None for an unnamed agent)
event.node_id         # "agent:<name>" or "tool:<name>/<tool>"; unnamed agents use "Agent"
event.sequence_num    # Event sequence number
event.data            # Event-specific data dict
event.to_dict()       # Convert to dictionary
event.to_json()       # Convert to JSON string

agent_end events carry event.data["status"]: "completed", "failed", or "cancelled".
event.agent_name is None for an unnamed agent; fall back to event.node_id ("agent:Agent") when grouping events by agent.
llm_request / llm_response arrive from async runs (astart/achat) as well as sync runs (start/chat).

Architecture

Diagram

Best Practices

Buffer Events

Batch events before sending to reduce network calls

Handle Errors

Sink errors are caught silently – tracing never crashes agents

Use trace_context

Automatic cleanup even if exceptions occur

Thread Safety

Use locks if your sink is accessed from multiple threads

Zero Overhead

When tracing is disabled, there’s zero performance impact:

python

# Disabled by default - zero overhead
emitter = ContextTraceEmitter(sink=sink, enabled=False)

# Or use NoOpSink
from praisonaiagents import ContextNoOpSink
emitter = ContextTraceEmitter(sink=ContextNoOpSink())

API Reference

ContextTraceEmitter

python

from praisonaiagents import ContextTraceEmitter

emitter = ContextTraceEmitter(
    sink=my_sink,           # Your sink implementation
    session_id="my-session", # Session identifier
    enabled=True,           # Enable/disable tracing
    redact=True,            # Redact sensitive data
)

trace_context

python

from praisonaiagents import trace_context

with trace_context(emitter) as ctx:
    # All agent events go to your sink
    agent.chat("Hello")
# Automatically resets when done

Built-in Sinks

Sink	Purpose
`ContextNoOpSink`	Does nothing (zero overhead)
`ContextListSink`	Collects events in memory (testing)
`ContextTraceWriter`	Writes to JSONL files (in praisonai)

CLI Commands

Use the CLI to manage traces:

List Traces

praisonai replay list

Replay Session

praisonai replay context <session_id>

Show Events

praisonai replay show <session_id>

Cleanup Old Traces

praisonai replay cleanup --max-age 7

Context Replay

Debug agent execution step-by-step

Observability

20+ provider integrations

Observability Hooks — Third-party sink plugins

Custom Tracing

Quick Start

How It Works

The Protocol

Protocol

Methods

Example Sinks

Event Types

Event Data

Architecture

Best Practices

Zero Overhead

API Reference

ContextTraceEmitter

trace_context

Built-in Sinks

CLI Commands

Related