Documentation

Step 1: Install the SDK

bashpip install aeneas-agentwatch

Step 2: Initialize in your code

pythonimport agentwatch
agentwatch.init()
# That's it. Every LLM call is now monitored.

Step 3: Make an LLM call

pythonfrom openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
# Trace captured automatically. No callbacks, no wrappers.

Step 4: View your traces

Open http://localhost:3001 to see your dashboard, or use the API:

bashcurl http://localhost:3001/api/traces

How It Works

Thresholds

Alert Cooldown

Duplicate alerts are suppressed for 60 seconds to prevent spam. Each unique threshold violation has its own cooldown timer.

Catching Blocked Requests + Recovery

pythonfrom agentwatch import CircuitBreakerException

try:
    result = client.chat.completions.create(...)
except CircuitBreakerException as e:
    print(f"Blocked: {e.reason}")
    print(f"State: {e.state}")  # Full monitor snapshot

    # Recovery via e.monitor:
    e.monitor.pause(60)           # Half-Open after 60s (probe call)
    e.monitor.increase_budget(5)  # Or: raise budget, resume immediately
    e.monitor.reset_windows()     # Emergency: clear all windows

    result = client.chat.completions.create(...)  # Next call goes through

State Machine

Pre-Block Hook (on_block)

Save state before the exception propagates. Supports sync and async callbacks.

pythonasync def save_state(event):
    await db.save({"agent": event.scope_id, "spent": event.current})
    event.monitor.pause(30)  # Pause instead of hard kill

agentwatch.init(
    budget_per_hour=5.0,
    block_on_threshold=True,
    on_block=save_state,  # Fires BEFORE exception
)

LangGraph Checkpoint Resume

pythontry:
    result = graph.invoke({"messages": [...]}, config=config)
except CircuitBreakerException as e:
    # LangGraph auto-checkpoints before exception propagates
    e.monitor.increase_budget(10.0)
    result = graph.invoke(None, config=config)  # Resumes from checkpoint

EU AI Act Compliance

Every state transition (triggered, paused, budget_increased, reset, recovered, retrip) emits a compliance-relevant span with eu_ai_act_art12_relevant flag. No state change goes unlogged.

LangChain

pythonimport agentwatch                          # <-- line 1
agentwatch.init(api_key="your-key")         # <-- line 2. Done.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

chain = ChatPromptTemplate.from_template("Explain {topic}") | ChatOpenAI(model="gpt-4o") | StrOutputParser()

with agentwatch.trace("research-chain"):
    result = chain.invoke({"topic": "EU AI Act"})
# Every LLM call traced. No LangSmith required.

CrewAI

pythonimport agentwatch                          # <-- line 1
agentwatch.init(api_key="your-key")         # <-- line 2. Done.

from crewai import Agent, Task, Crew

researcher = Agent(role="Researcher", goal="Find data", verbose=True)
writer = Agent(role="Writer", goal="Write content", verbose=True)
crew = Crew(agents=[researcher, writer], tasks=[...])

with agentwatch.trace("compliance-crew"):
    result = crew.kickoff()
# Both agents traced — tokens, cost, latency per step.

AutoGen

pythonimport agentwatch                          # <-- line 1
agentwatch.init(api_key="your-key")         # <-- line 2. Done.

from autogen import ConversableAgent

assistant = ConversableAgent(name="Assistant", llm_config={"model": "gpt-4o"})
reviewer = ConversableAgent(name="Reviewer", llm_config={"model": "gpt-4o"})

with agentwatch.trace("code-review"):
    assistant.initiate_chat(reviewer, message="Review this code", max_turns=3)
# Full conversation flow traced: Assistant -> Reviewer -> Assistant.

Any Framework / Direct HTTP

pythonimport agentwatch                          # <-- line 1
agentwatch.init(api_key="your-key")         # <-- line 2. Done.

# Works with LlamaIndex, Haystack, Semantic Kernel, raw httpx/requests —
# anything that calls an AI API over HTTP is captured automatically.
# Supported: OpenAI, Anthropic, Gemini, Mistral, Groq, Cohere,
# Together AI, Fireworks, Ollama, Azure OpenAI, and more.

Minimal (Development)

bashdocker compose -f docker-compose.local.yml up -d
# Starts: ClickHouse + Backend
# Dashboard: http://localhost:3001
# No Kafka. No auth. LOCAL_MODE=true.

Full Stack (Production)

bashdocker compose up -d
# Starts: ClickHouse + Kafka + Backend + Proxy
# Configure via .env file
# Services: clickhouse (8123), kafka (9092), backend (3001), proxy (8080)

Verify It Works

bash# Check server health:
curl http://localhost:3001/health

# Send a test trace:
python -c "import agentwatch; agentwatch.init(); agentwatch.verify()"

Core

Authentication

Data & Privacy

Email (Optional)

SDK Alerts (In-Process)

Configured via init() parameters. Fires immediately when thresholds are exceeded. 60-second cooldown between duplicate alerts.

pythondef my_alert_handler(alert):
    print(f"Alert: {alert['reason']}")
    print(f"Scope: {alert['scope']} / {alert['scope_id']}")
    print(f"Threshold: {alert['threshold']} | Current: {alert['current']}")
    # Send to Slack, PagerDuty, email, etc.

agentwatch.init(
    budget_per_hour=10.0,
    on_alert=my_alert_handler
)

Backend Alerts (Server-Side)

Create rules via the dashboard or API. Backend evaluates rules against incoming trace data and sends email notifications via Resend.

bash# Create an alert rule via API:
curl -X POST http://localhost:3001/api/alerts \
  -H "Content-Type: application/json" \
  -d '{
    "name": "High Cost Alert",
    "condition": "cost_per_hour",
    "threshold": 50.0,
    "action_type": "email"
  }'

From Langfuse

Langfuse requires decorators or callbacks on every function. AeneasSoft requires zero code changes beyond init().

python# Before (Langfuse):
from langfuse.decorators import observe
from langfuse.openai import openai

@observe()
def my_function():
    client = openai.OpenAI()
    return client.chat.completions.create(...)

# After (AeneasSoft):
import agentwatch
agentwatch.init()

def my_function():
    client = OpenAI()
    return client.chat.completions.create(...)
# That's it. Remove all decorators. Remove langfuse imports.

From LangSmith

LangSmith is tightly coupled to LangChain. AeneasSoft works with any framework.

python# Before (LangSmith):
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls_..."
# Only works with LangChain

# After (AeneasSoft):
import agentwatch
agentwatch.init()
# Works with LangChain, CrewAI, raw OpenAI, httpx, any framework

From Helicone

Helicone requires routing all traffic through a proxy (single point of failure). AeneasSoft runs in-process.

python# Before (Helicone):
client = OpenAI(
    base_url="https://oai.helicone.ai/v1",  # Proxy SPOF
    default_headers={"Helicone-Auth": "Bearer sk-..."}
)

# After (AeneasSoft):
import agentwatch
agentwatch.init()
client = OpenAI()  # Direct connection. No proxy. No SPOF.