Raven — Notifications
Sends alerts and notifications via Discord and email. Triggered by Varys (service down/up) and Grafana (infrastructure alerts).
Updated · {total} total
Email alerts: {email_label}
Recent Notifications (last 20)
{"" if rows else 'No notifications yet.
'}
"""
def _render_md(history: list, timestamp: str) -> str:
now = timestamp[:16].replace("T", " ") + " UTC"
lines = [f"# Raven — Notification Log\n\nUpdated: {now} | {len(history)} total\n"]
for e in reversed(history[-10:]):
ts = e.get("timestamp", "")[:16].replace("T", " ") + " UTC"
lines.append(f"- [{e.get('severity','?').upper()}] {ts} · **{e.get('source','')}** — {e.get('message','')}")
return "\n".join(lines)
def _write_status(history: list, last_event: dict = None):
timestamp = datetime.now(timezone.utc).isoformat()
email_enabled = _load_state().get("email_enabled", True)
out_dir = SITES_DIR / "raven"
out_dir.mkdir(parents=True, exist_ok=True)
(out_dir / "index.html").write_text(_render_html(history, timestamp, email_enabled))
(out_dir / "last-output.md").write_text(_render_md(history, timestamp))
log_dir = AGENT_OS_DIR / "logs" / "raven-notify"
log_dir.mkdir(parents=True, exist_ok=True)
last_msg = (
f"last: [{last_event['severity']}] {last_event['source']} — {last_event['message'][:50]}"
if last_event else "running, no notifications yet"
)
(log_dir / "last-run.json").write_text(json.dumps({
"agent": "raven-notify",
"timestamp": timestamp,
"status": "success",
"result": f"{len(history)} notifications total; {last_msg}",
}, indent=2))
def _do_notify(message: str, severity: str, source: str) -> dict:
timestamp = datetime.now(timezone.utc).isoformat()
severity = severity.lower()
entry = {"timestamp": timestamp, "severity": severity, "source": source, "message": message}
history = _load_history()
history.append(entry)
_save_history(history)
_send_discord(message, severity, source, timestamp)
_send_email(message, severity, source)
_write_status(history, entry)
logger.info(f"notified [{severity}] {source}: {message[:80]}")
return {"ok": True, "timestamp": timestamp}
@asynccontextmanager
async def lifespan(app: FastAPI):
history = _load_history()
_write_status(history)
logger.info(f"raven-notify started on port {PORT} — {len(history)} notifications in history")
yield
app = FastAPI(title="raven-notify", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["POST"],
allow_headers=["*"],
)
@app.get("/health")
def health():
return {"status": "ok", "agent": "raven-notify"}
@app.post("/notify")
def handle_notify(req: NotifyRequest):
return _do_notify(req.message, req.severity, req.source)
@app.post("/email/toggle")
def toggle_email():
state = _load_state()
state["email_enabled"] = not state.get("email_enabled", True)
_save_state(state)
_write_status(_load_history())
logger.info(f"email alerts {'enabled' if state['email_enabled'] else 'disabled'} via toggle")
return {"ok": True, "email_enabled": state["email_enabled"]}
@app.post("/webhook/grafana")
def handle_grafana(payload: dict):
status = payload.get("status", "firing")
alerts = payload.get("alerts", [])
if alerts:
for alert in alerts:
alert_status = alert.get("status", status)
name = alert.get("labels", {}).get("alertname", "unknown")
if alert_status == "resolved":
# Grafana always sends the firing annotation text in resolved payloads —
# override with a clear recovery message so Discord/email isn't confusing.
msg = f"✅ RESOLVED: {name} — alert has cleared, service is back online."
sev = "info"
else:
msg = alert.get("annotations", {}).get("description", name)
sev = "critical"
_do_notify(msg, sev, f"grafana/{name}")
return {"ok": True, "processed": len(alerts)}
msg = payload.get("message", "") or payload.get("title", "Grafana Alert")
sev = "critical" if status == "firing" else "info"
return _do_notify(msg, sev, "grafana")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=PORT)
# ---------------------------------------------------------------------------
# Jon Snow approval gate — action notification endpoint
# ---------------------------------------------------------------------------
class ActionNotifyRequest(BaseModel):
description: str
approve_url: str
reject_url: str
action_type: str
source: str = "jon-snow"
@app.post("/notify-with-actions")
def handle_action_notify(req: ActionNotifyRequest):
"""Send a Discord embed with clickable approve/reject links for Jon Snow's approval gate."""
if not DISCORD_WEBHOOK_URL:
logger.warning("DISCORD_WEBHOOK_URL not set — skipping Discord")
return {"ok": False, "error": "no webhook configured"}
payload = {
"embeds": [{
"title": "⚡ Action Approval Required",
"description": f"**{req.source}** wants to: `{req.description}`",
"color": 16776960, # yellow
"fields": [
{"name": "Action Type", "value": f"`{req.action_type}`", "inline": True},
{"name": "Expires", "value": "15 minutes", "inline": True},
],
"footer": {"text": "jon-snow approval gate · nxm.co.za"},
}],
"content": f"✅ **Approve:** {req.approve_url}\n❌ **Reject:** {req.reject_url}",
}
try:
r = httpx.post(DISCORD_WEBHOOK_URL, json=payload, timeout=10)
r.raise_for_status()
logger.info(f"discord: action approval request sent for {req.action_type}")
return {"ok": True}
except Exception as e:
logger.error(f"discord action notify failed: {e}")
return {"ok": False, "error": str(e)}