98 lines
3 KiB
Python
98 lines
3 KiB
Python
import os
|
|
import json
|
|
import requests
|
|
from flask import Flask, request, jsonify
|
|
|
|
app = Flask(__name__)
|
|
|
|
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")
|
|
NTFY_URL = os.environ.get("NTFY_URL", "https://ntfy.aery.tech/alerts")
|
|
NTFY_TOKEN = os.environ.get("NTFY_TOKEN", "")
|
|
|
|
|
|
def ask_claude(alert_name, alert_state, labels, annotations, values):
|
|
prompt = f"""You are a homelab infrastructure assistant. Analyze this Grafana alert and provide a concise, plain-English summary with suggested actions.
|
|
|
|
Alert Name: {alert_name}
|
|
State: {alert_state}
|
|
Labels: {json.dumps(labels, indent=2)}
|
|
Annotations: {json.dumps(annotations, indent=2)}
|
|
Values: {json.dumps(values, indent=2)}
|
|
|
|
Respond in 2-3 sentences max. Be direct and actionable. Focus on what happened and what to check."""
|
|
|
|
response = requests.post(
|
|
"https://api.anthropic.com/v1/messages",
|
|
headers={
|
|
"x-api-key": ANTHROPIC_API_KEY,
|
|
"anthropic-version": "2023-06-01",
|
|
"content-type": "application/json",
|
|
},
|
|
json={
|
|
"model": "claude-haiku-4-5-20251001",
|
|
"max_tokens": 256,
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
},
|
|
timeout=15,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()["content"][0]["text"]
|
|
|
|
|
|
def send_ntfy(title, message, priority="default", tags=None):
|
|
headers = {
|
|
"Title": title,
|
|
"Priority": priority,
|
|
"Content-Type": "text/plain",
|
|
}
|
|
if tags:
|
|
headers["Tags"] = ",".join(tags)
|
|
if NTFY_TOKEN:
|
|
headers["Authorization"] = f"Basic {NTFY_TOKEN}"
|
|
|
|
requests.post(NTFY_URL, data=message.encode("utf-8"), headers=headers, timeout=10)
|
|
|
|
|
|
@app.route("/webhook", methods=["POST"])
|
|
def webhook():
|
|
data = request.get_json(silent=True)
|
|
if not data:
|
|
return jsonify({"error": "Invalid JSON"}), 400
|
|
|
|
alerts = data.get("alerts", [])
|
|
if not alerts:
|
|
return jsonify({"status": "no alerts"}), 200
|
|
|
|
for alert in alerts:
|
|
alert_name = alert.get("labels", {}).get("alertname", "Unknown Alert")
|
|
alert_state = alert.get("status", "unknown")
|
|
labels = alert.get("labels", {})
|
|
annotations = alert.get("annotations", {})
|
|
values = alert.get("values", {})
|
|
|
|
# Determine priority and tags based on state
|
|
if alert_state == "firing":
|
|
priority = "high"
|
|
tags = ["warning", "grafana"]
|
|
else:
|
|
priority = "default"
|
|
tags = ["white_check_mark", "grafana"]
|
|
|
|
try:
|
|
summary = ask_claude(alert_name, alert_state, labels, annotations, values)
|
|
except Exception as e:
|
|
summary = f"Alert {alert_state}: {alert_name}. (Claude enrichment failed: {e})"
|
|
|
|
title = f"{alert_name} [{alert_state.upper()}]"
|
|
send_ntfy(title, summary, priority=priority, tags=tags)
|
|
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
|
|
@app.route("/health", methods=["GET"])
|
|
def health():
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=5000)
|