tralo-monitor 0.1.0

AI agent monitoring SDK — zero-config observability

tralo-monitor

AI agent monitoring SDK for Tralo. The SDK captures events and raw provider responses; the Tralo server validates, deduplicates, extracts tokens, and interprets provider formats.

Installation

pip install tralo-monitor
pip install tralo-monitor[openai]   # with OpenAI autopatch
pip install tralo-monitor[all]      # all providers

Quickstart (autopatch — recommended)

from tralo_monitor import Tralo
from tralo_monitor.wrappers import patch_openai
import openai

monitor = Tralo(api_key="trl_live_YOUR_KEY", agent_id="my-agent")
client = openai.OpenAI()
patch_openai(client, monitor)

# Your existing provider call is unchanged.
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Help me"}]
)

Autopatch sends the raw provider response to /v1/ingest. Token extraction comes from the server-side format registry, so format updates do not require an SDK release.

Context manager pattern

from tralo_monitor import Tralo

monitor = Tralo(api_key="trl_live_YOUR_KEY", agent_id="support-agent")

with monitor.run(run_id="ticket-123") as run:
    result = call_llm("Classify this ticket")
    run.step(
        step_name="classify",
        model="gpt-4o",
        tokens_in=150,
        tokens_out=80,
        latency_ms=850,
        input={"prompt": "Classify this ticket"},
        output={"label": "billing"},
    )

The context manager sends run_start automatically. It sends run_end with success on clean exit and failure with error_msg if an exception is raised. The original exception is re-raised.

Decorator pattern

from tralo_monitor import Tralo

monitor = Tralo(api_key="trl_live_YOUR_KEY", agent_id="support-agent")

@monitor.track(step_name="classify_ticket")
def classify_ticket(text: str) -> str:
    return call_llm(text)

label = classify_ticket("Refund request")

Async functions work the same way:

@monitor.track(step_name="async_classify")
async def classify_ticket(text: str) -> str:
    return await call_llm_async(text)

The decorator captures latency and exceptions. It does not capture tokens by itself; use autopatch for provider responses.

Manual pattern

from tralo_monitor import Tralo

monitor = Tralo(api_key="trl_live_YOUR_KEY", agent_id="support-agent")

run = monitor.start_run()
try:
    run.step("classify", model="gpt-4o", tokens_in=150, tokens_out=80)
    run.step("respond", model="gpt-4o", tokens_in=300, tokens_out=200)
    run.end(status="success")
except Exception as error:
    run.end(status="failure", error_msg=str(error))
    raise

Provider wrappers

from tralo_monitor import Tralo
from tralo_monitor.wrappers import patch_anthropic, patch_auto, patch_groq, patch_openai

monitor = Tralo(api_key="trl_live_YOUR_KEY", agent_id="my-agent")
patch_openai(openai_client, monitor)
patch_anthropic(anthropic_client, monitor)
patch_groq(groq_client, monitor)
patch_auto(custom_client, monitor)

Patches are idempotent and reversible with unpatch_openai(), unpatch_anthropic(), and unpatch_groq(). Monitoring failures never propagate to user code.

Reliability

Every event gets an X-Idempotency-Key header so retries cannot create duplicate database rows. If the API is unavailable, events are written to ~/.tralo/failed_events.ndjson and retried on the next SDK initialization. API keys are redacted in repr, str, and debug output.

Configuration

monitor = Tralo(
    api_key="trl_live_YOUR_KEY",
    agent_id="my-agent",
    base_url="https://api.tralo.dev",
    debug=False,
)

Use base_url="http://localhost:8000" for a local Tralo API. Call monitor.flush() before process exit in short-lived serverless functions.