> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# 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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart LR
A[Your Agent]:::agent -->|events| B[Custom Sink]:::tool
B -->|store| C[Any Destination]:::agent
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
```
PraisonAI uses a **protocol-driven** design. Implement 3 methods and your sink works with any agent.
## Quick Start
Implement 3 methods: `emit()`, `flush()`, `close()`
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
class MySink:
def emit(self, event):
print(f"Event: {event.event_type.value}")
def flush(self):
pass
def close(self):
pass
```
Use `trace_context()` to capture all agent events
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart LR
A[Agent]:::agent -->|emit| B[ContextTraceEmitter]:::tool
B -->|emit| C[Your Custom Sink]:::tool
C -->|store| D[Any Destination]:::agent
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
```
## The Protocol
Your sink must implement 3 methods. That's it!
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
class ContextTraceSinkProtocol:
def emit(self, event): ... # Receive event
def flush(self): ... # Flush buffer
def close(self): ... # Release resources
```
`ContextTraceSink` is a backward-compatible alias for `ContextTraceSinkProtocol`.
| Method | Purpose |
| ------------- | --------------------- |
| `emit(event)` | Receive a trace event |
| `flush()` | Flush buffered events |
| `close()` | Release resources |
## Example Sinks
Send events to a remote HTTP endpoint with batching.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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()
```
Store events in SQLite database for later analysis.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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()
```
Print events to console for debugging.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart TB
subgraph Session
SS[session_start]:::agent --> SE[session_end]:::agent
end
subgraph Agent
AS[agent_start]:::agent --> AE[agent_end]:::agent
AS --> AH[agent_handoff]:::agent
end
subgraph Operations
MA[message_added]:::agent
TS[tool_call_start]:::tool --> TE[tool_call_end]:::tool
LR[llm_request]:::tool --> LP[llm_response]:::tool
end
subgraph Memory
MS[memory_store]:::tool
MSE[memory_search]:::tool
end
subgraph Knowledge
KA[knowledge_add]:::tool
KS[knowledge_search]:::tool
end
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
```
| 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
event.event_type # ContextEventType enum
event.timestamp # Unix timestamp
event.session_id # Session identifier
event.agent_name # Agent name (if applicable)
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
```
## Architecture
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
flowchart TB
subgraph "Core SDK (praisonaiagents)"
P[ContextTraceSink Protocol]:::tool
E[ContextTraceEmitter]:::tool
N[ContextNoOpSink]:::tool
L[ContextListSink]:::tool
end
subgraph "Wrapper (praisonai)"
W[ContextTraceWriter]:::tool
end
subgraph "Your Code"
C[Custom Sink]:::agent
end
P -.->|implements| N
P -.->|implements| L
P -.->|implements| W
P -.->|implements| C
E -->|uses| P
classDef agent fill:#8B0000,color:#fff
classDef tool fill:#189AB4,color:#fff
```
## Best Practices
Batch events before sending to reduce network calls
Sink errors are caught silently - tracing never crashes agents
Automatic cleanup even if exceptions occur
Use locks if your sink is accessed from multiple threads
## Zero Overhead
When tracing is disabled, there's **zero performance impact**:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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 theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```bash List Traces theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
praisonai replay list
```
```bash Replay Session theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
praisonai replay context
```
```bash Show Events theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
praisonai replay show
```
```bash Cleanup Old Traces theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
praisonai replay cleanup --max-age 7
```
## Related
Debug agent execution step-by-step
20+ provider integrations
Register sinks via the `praisonai.observability_sinks` entry-point group