Build an AI Observability Alert Noise Reducer with Claude API

pip install anthropic requests

import anthropic
import requests
from collections import defaultdict
from datetime import datetime, timedelta
 
client = anthropic.Anthropic()
 
 
def get_alert_history(days: int = 90) -> list[dict]:
    """Pull historical alert firing/resolution data from Alertmanager."""
    resp = requests.get(
        "http://alertmanager:9093/api/v2/alerts/history",
        params={"since": (datetime.utcnow() - timedelta(days=days)).isoformat()}
    )
    return resp.json()
 
 
def compute_alert_stats(history: list[dict]) -> dict:
    """Aggregate per-alert-rule firing frequency, duration, and — critically —
    whether it ever led to a PagerDuty acknowledgment vs auto-resolved unattended."""
    stats = defaultdict(lambda: {"fire_count": 0, "durations": [], "was_acked": []})
 
    for event in history:
        rule = event["labels"]["alertname"]
        stats[rule]["fire_count"] += 1
        stats[rule]["durations"].append(event.get("duration_seconds", 0))
        stats[rule]["was_acked"].append(event.get("acknowledged", False))
 
    return dict(stats)

def get_incident_correlation(alert_rule: str, days: int = 90) -> dict:
    """Check PagerDuty history — did this alert ever correlate with an
    incident that got a real human response, or does it always
    auto-resolve with nobody looking at it?"""
    resp = requests.get(
        "https://api.pagerduty.com/incidents",
        headers={"Authorization": f"Token token={PAGERDUTY_TOKEN}"},
        params={"since": (datetime.utcnow() - timedelta(days=days)).isoformat()}
    )
    incidents = resp.json()["incidents"]
    matching = [i for i in incidents if alert_rule in i.get("title", "")]
 
    return {
        "total_incidents": len(matching),
        "resolved_by_human_action": len([i for i in matching if i.get("resolve_reason") == "manual"]),
        "auto_resolved": len([i for i in matching if i.get("resolve_reason") == "timeout"]),
    }

import json
 
ANALYZE_PROMPT = """Analyze this alert rule's firing history and recommend
specific tuning.
 
Alert rule: {rule_name}
Current threshold config: {current_config}
 
Firing statistics (last 90 days):
- Total fires: {fire_count}
- Average duration before auto-resolve: {avg_duration}s
- Percentage that were acknowledged by a human: {ack_rate}%
 
Incident correlation:
- Times this led to a PagerDuty incident: {total_incidents}
- Of those, resolved by actual human action (not auto-timeout): {resolved_by_human}
 
Determine:
1. Is this alert too sensitive (fires often, rarely acknowledged, rarely
   leads to real action) — recommend a specific threshold/duration adjustment
2. Is this alert well-tuned (fires rarely, but when it does, leads to
   real action) — leave alone, flag as healthy
3. Is this alert potentially UNDER-sensitive (rarely fires, but incident
   history suggests real problems happened without it firing) — recommend
   tightening
 
Respond with ONLY valid JSON:
{{"verdict": "too_noisy" | "well_tuned" | "under_sensitive",
  "recommended_change": "specific config change or null",
  "reasoning": "..."}}"""
 
 
def analyze_alert(rule_name: str, current_config: dict, stats: dict, correlation: dict) -> dict:
    ack_rate = (sum(stats["was_acked"]) / len(stats["was_acked"]) * 100) if stats["was_acked"] else 0
    avg_duration = sum(stats["durations"]) / len(stats["durations"]) if stats["durations"] else 0
 
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": ANALYZE_PROMPT.format(
                rule_name=rule_name, current_config=current_config,
                fire_count=stats["fire_count"], avg_duration=round(avg_duration, 1), ack_rate=round(ack_rate, 1),
                total_incidents=correlation["total_incidents"],
                resolved_by_human=correlation["resolved_by_human_action"],
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)