Claude's Corner: Sentrial - Monitoring the AI Agents Your Datadog Can't See

# Build Your Own AI Agent Monitoring Platform: A Step-by-Step Guide

A developer guide to building a Sentrial-style production monitoring platform for AI agents using Claude Code. This covers the core architecture: SDK integration, semantic failure detection, session replay, and real-time alerting.

---

## Step 1: Design the Data Model

Start with the schema that stores agent telemetry. Every session, step, and classification result needs a home.

```sql
CREATE TABLE agent_sessions (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id UUID NOT NULL REFERENCES customers(id),
  session_id TEXT NOT NULL,
  framework TEXT NOT NULL, -- langchain, openai, anthropic, custom
  started_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  ended_at TIMESTAMPTZ,
  total_tokens INTEGER,
  total_cost_usd NUMERIC(10, 6),
  step_count INTEGER DEFAULT 0,
  status TEXT DEFAULT 'active', -- active, completed, flagged, error
  created_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE agent_steps (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  session_id UUID NOT NULL REFERENCES agent_sessions(id) ON DELETE CASCADE,
  step_index INTEGER NOT NULL,
  step_type TEXT NOT NULL, -- llm_call, tool_call, memory_read, output
  input_text TEXT,
  output_text TEXT,
  tool_name TEXT,
  tool_args JSONB,
  tool_result TEXT,
  latency_ms INTEGER,
  tokens_used INTEGER,
  cost_usd NUMERIC(10, 6),
  recorded_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE failure_events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  session_id UUID NOT NULL REFERENCES agent_sessions(id),
  step_id UUID REFERENCES agent_steps(id),
  failure_type TEXT NOT NULL, -- hallucination, infinite_loop, bad_tool_call, forgetfulness, jailbreak, user_frustration
  confidence NUMERIC(4, 3) NOT NULL, -- 0.000 to 1.000
  classifier_id UUID NOT NULL REFERENCES classifiers(id),
  details JSONB,
  alerted BOOLEAN DEFAULT false,
  created_at TIMESTAMPTZ DEFAULT now()
);

CREATE TABLE classifiers (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  customer_id UUID REFERENCES customers(id), -- NULL = built-in classifier
  failure_type TEXT NOT NULL,
  model_path TEXT, -- path to fine-tuned model weights in object storage
  threshold NUMERIC(4, 3) DEFAULT 0.75,
  training_examples_count INTEGER DEFAULT 0,
  deployed_at TIMESTAMPTZ,
  created_at TIMESTAMPTZ DEFAULT now()
);
```

Key design decision: every step is its own row. This enables the replay-from-step feature. Never flatten steps into a session-level blob.

---

## Step 2: Build the SDK Integration Layer

The SDK is the customer-facing surface. It must be a callback handler that adds zero latency to the agent's hot path - classification runs asynchronously.

```python
# sentrial_sdk/handler.py
import asyncio
import httpx
from langchain.callbacks.base import BaseCallbackHandler
from typing import Any, Dict, List, Optional, Union
from uuid import uuid4

class SentrialCallbackHandler(BaseCallbackHandler):
    def __init__(self, api_key: str, customer_id: str, endpoint: str = "https://api.sentrial.com"):
        self.api_key = api_key
        self.customer_id = customer_id
        self.endpoint = endpoint
        self.session_id = str(uuid4())
        self._steps: List[Dict] = []
        self._client = httpx.AsyncClient(
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=5.0
        )

    def on_llm_start(self, serialized, prompts, **kwargs):
        self._current_step = {
            "step_type": "llm_call",
            "input_text": prompts[0] if prompts else None,
            "step_index": len(self._steps),
            "start_ms": self._now_ms()
        }

    def on_llm_end(self, response, **kwargs):
        step = self._current_step
        step["output_text"] = response.generations[0][0].text if response.generations else None
        step["latency_ms"] = self._now_ms() - step.pop("start_ms")
        step["tokens_used"] = response.llm_output.get("token_usage", {}).get("total_tokens")
        self._steps.append(step)
        asyncio.create_task(self._flush_step(step))

    def on_tool_start(self, serialized, input_str, **kwargs):
        self._current_tool_step = {
            "step_type": "tool_call",
            "tool_name": serialized.get("name"),
            "tool_args": {"input": input_str},
            "step_index": len(self._steps),
            "start_ms": self._now_ms()
        }

    def on_tool_end(self, output, **kwargs):
        step = self._current_tool_step
        step["tool_result"] = str(output)
        step["latency_ms"] = self._now_ms() - step.pop("start_ms")
        self._steps.append(step)
        asyncio.create_task(self._flush_step(step))

    async def _flush_step(self, step: Dict):
        try:
            await self._client.post(
                f"{self.endpoint}/v1/ingest/step",
                json={
                    "session_id": self.session_id,
                    "customer_id": self.customer_id,
                    **step
                }
            )
        except Exception:
            pass  # Never block the agent on telemetry failures

    def _now_ms(self) -> int:
        import time
        return int(time.time() * 1000)
```

Usage (the "5 lines" pitch):

```python
from sentrial_sdk import SentrialCallbackHandler

handler = SentrialCallbackHandler(api_key="sk-...", customer_id="your-customer-id")
agent = initialize_agent(tools, llm, callbacks=[handler])
```

For OpenAI and Anthropic direct integrations, wrap at the client level rather than relying on LangChain's callback system.

---

## Step 3: Build the Telemetry Ingestion Pipeline

The ingestion endpoint receives step events in real time and queues them for classification. Speed here is critical - ingestion must not block.

```python
# api/ingest.py
from fastapi import FastAPI, BackgroundTasks, Depends
from pydantic import BaseModel
from typing import Optional
import asyncio

app = FastAPI()

class StepEvent(BaseModel):
    session_id: str
    customer_id: str
    step_index: int
    step_type: str
    input_text: Optional[str]
    output_text: Optional[str]
    tool_name: Optional[str]
    tool_args: Optional[dict]
    tool_result: Optional[str]
    latency_ms: Optional[int]
    tokens_used: Optional[int]

@app.post("/v1/ingest/step")
async def ingest_step(event: StepEvent, background_tasks: BackgroundTasks):
    # Write to DB immediately
    step_id = await db.insert_step(event)
    
    # Queue classification asynchronously - never in the hot path
    background_tasks.add_task(
        classify_step,
        step_id=step_id,
        customer_id=event.customer_id,
        step_data=event
    )
    
    # Check for infinite loop pattern inline (cheap heuristic)
    background_tasks.add_task(
        check_loop_pattern,
        session_id=event.session_id,
        customer_id=event.customer_id
    )
    
    return {"ok": True, "step_id": str(step_id)}
```

Use a message queue (Redis Streams or Kafka) between ingestion and classification if you expect high volume. The ingest endpoint should complete in under 5ms.

---

## Step 4: Build the Classifier Pipeline

This is the core of the product. Start with rule-based classifiers for infinite loops and bad tool calls (cheap and reliable), then layer in fine-tuned models for semantic failures.

```python
# classifiers/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass
class ClassificationResult:
    failure_type: str
    confidence: float
    details: dict

class BaseClassifier(ABC):
    failure_type: str
    
    @abstractmethod
    async def classify(self, session_steps: list, current_step: dict) -> Optional[ClassificationResult]:
        pass

# classifiers/loop_detector.py
class InfiniteLoopDetector(BaseClassifier):
    failure_type = "infinite_loop"
    
    async def classify(self, session_steps: list, current_step: dict) -> Optional[ClassificationResult]:
        # Look at last N tool calls for repetition
        tool_calls = [s for s in session_steps if s["step_type"] == "tool_call"]
        if len(tool_calls) < 3:
            return None
        
        last_three = tool_calls[-3:]
        if all(s["tool_name"] == last_three[0]["tool_name"] for s in last_three):
            if all(s["tool_args"] == last_three[0]["tool_args"] for s in last_three):
                return ClassificationResult(
                    failure_type="infinite_loop",
                    confidence=0.95,
                    details={"repeated_tool": last_three[0]["tool_name"], "count": 3}
                )
        return None

# classifiers/hallucination_detector.py
class HallucinationDetector(BaseClassifier):
    failure_type = "hallucination"
    
    def __init__(self, model_path: str, threshold: float = 0.75):
        self.model = load_fine_tuned_model(model_path)
        self.threshold = threshold
    
    async def classify(self, session_steps: list, current_step: dict) -> Optional[ClassificationResult]:
        if current_step["step_type"] != "llm_call":
            return None
        
        features = self._extract_features(session_steps, current_step)
        confidence = await self.model.predict(features)
        
        if confidence >= self.threshold:
            return ClassificationResult(
                failure_type="hallucination",
                confidence=float(confidence),
                details={"output_snippet": current_step["output_text"][:200]}
            )
        return None
    
    def _extract_features(self, steps: list, current: dict) -> dict:
        # Feature engineering: factual consistency signals, hedging language,
        # contradiction with prior steps, citation patterns, etc.
        return {
            "output_text": current["output_text"],
            "prior_context": [s["output_text"] for s in steps[-3:] if s.get("output_text")],
            "has_hedging": self._detect_hedging(current["output_text"])
        }
```

Fine-tuning: collect customer session logs, label failures (3-4 positive examples minimum), fine-tune a small classification model (distilbert-class works well), store weights in object storage keyed by customer + classifier type.

---

## Step 5: Implement Session Replay and Fork

The replay-and-fork system is what makes debugging practical. Store enough state at each step to reconstruct execution context and branch from any point.

```python
# replay/engine.py
class SessionReplayEngine:
    async def get_session_state_at_step(self, session_id: str, step_index: int) -> dict:
        steps = await db.get_steps_up_to(session_id, step_index)
        
        # Reconstruct memory, context, and tool state at step N
        return {
            "conversation_history": self._build_conversation_history(steps),
            "tool_outputs": {s["tool_name"]: s["tool_result"] for s in steps if s["step_type"] == "tool_call"},
            "agent_state": self._extract_agent_state(steps),
            "step_index": step_index,
            "session_id": session_id
        }
    
    async def fork_from_step(self, session_id: str, step_index: int, modified_input: str = None) -> str:
        state = await self.get_session_state_at_step(session_id, step_index)
        
        # Create new session with pre-loaded state
        new_session_id = str(uuid4())
        await db.create_forked_session(
            new_session_id=new_session_id,
            parent_session_id=session_id,
            fork_point=step_index,
            initial_state=state
        )
        
        return new_session_id
    
    def _build_conversation_history(self, steps: list) -> list:
        history = []
        for step in steps:
            if step["step_type"] == "llm_call":
                history.append({"role": "user", "content": step["input_text"]})
                if step.get("output_text"):
                    history.append({"role": "assistant", "content": step["output_text"]})
        return history
```

Store the full input and output at every step. Compression is worth considering for high-volume production deployments - use columnar storage (Parquet in S3) for long-term session archives.

---

## Step 6: Build the Alert Routing System

Real-time Slack alerts with source-code-level detail. The key is including enough context that engineers can act immediately.

```python
# alerting/slack.py
import httpx
from jinja2 import Template

SLACK_ALERT_TEMPLATE = Template("""
:red_circle: *Agent Failure Detected*

*Type:* {{ failure_type | title }}
*Confidence:* {{ confidence | round(2) }}
*Session:* `{{ session_id }}`
*Step:* {{ step_index }}

*Details:*
```
{{ details_text }}
```

*Quick Actions:*
• <{{ dashboard_url }}/sessions/{{ session_id }}|View full session>
• <{{ dashboard_url }}/replay/{{ session_id }}/{{ step_index }}|Fork from this step>
""")

class SlackAlerter:
    def __init__(self, webhook_url: str, dashboard_url: str):
        self.webhook_url = webhook_url
        self.dashboard_url = dashboard_url
    
    async def send_failure_alert(self, failure_event: dict, session: dict, step: dict):
        message = SLACK_ALERT_TEMPLATE.render(
            failure_type=failure_event["failure_type"],
            confidence=failure_event["confidence"],
            session_id=session["session_id"],
            step_index=step["step_index"],
            details_text=self._format_details(failure_event, step),
            dashboard_url=self.dashboard_url
        )
        
        async with httpx.AsyncClient() as client:
            await client.post(self.webhook_url, json={"text": message})
    
    def _format_details(self, failure: dict, step: dict) -> str:
        lines = []
        if step.get("output_text"):
            lines.append(f"Output: {step['output_text'][:300]}")
        if step.get("tool_name"):
            lines.append(f"Tool: {step['tool_name']}({step.get('tool_args', {})})")
        lines.extend(f"{k}: {v}" for k, v in failure.get("details", {}).items())
        return "\n".join(lines)
```

Add PagerDuty integration for high-confidence failures in critical production agents. Slack is sufficient for investigation; PD is for when an agent failure has direct user impact.

---

## Step 7: Deploy and Scale

Production architecture recommendations for a Sentrial-style platform:

```yaml
# docker-compose.yml (development)
services:
  api:
    build: ./api
    ports: ["8000:8000"]
    environment:
      DATABASE_URL: postgresql://postgres:password@db:5432/sentrial
      REDIS_URL: redis://redis:6379
    depends_on: [db, redis]
  
  classifier_worker:
    build: ./classifiers
    environment:
      DATABASE_URL: postgresql://postgres:password@db:5432/sentrial
      REDIS_URL: redis://redis:6379
      MODEL_STORAGE: s3://sentrial-models
    command: python -m workers.classifier
    replicas: 4  # Scale based on classification queue depth
  
  db:
    image: postgres:16
    volumes: ["pgdata:/var/lib/postgresql/data"]
  
  redis:
    image: redis:7-alpine

volumes:
  pgdata:
```

Production notes:
- **Classifier serving:** Use a model registry (MLflow, Weights and Biases) to version per-customer models. Serve via a dedicated model API with caching - a customer's fine-tuned hallucination detector should load once per worker, not per request.
- **Session storage:** Partition the `agent_steps` table by `customer_id` and `recorded_at`. At scale, steps volume dwarfs session volume by 100x and naive queries become painful fast.
- **Full-capture ingestion:** If you want true full fidelity at high volume, OpenTelemetry Collector as a sidecar absorbs bursts before they hit your database.
- **Session replay storage:** Archive completed sessions to Parquet in S3 after 30 days. Keep the last 30 days hot in Postgres for the replay interface.
- **Pricing instrumentation:** Count sessions, not steps or tokens - it simplifies billing and aligns your incentives with your customers' incentives.

The stack that gets you to first 10 customers: FastAPI, Postgres 16, Redis for the classification queue, S3 for model storage, Cloudflare Workers for the SDK CDN. Total cloud spend under $500/month at early traction volumes.