Build an AI SRE Incident Commander with Claude API

import anthropic
import json
import subprocess
import os
import urllib.request
from datetime import datetime
from fastapi import FastAPI, Request
from pydantic import BaseModel
from typing import Optional
 
app = FastAPI()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
SLACK_WEBHOOK = os.getenv("SLACK_INCIDENT_WEBHOOK")
 
 
class Alert(BaseModel):
    title: str
    severity: str  # P1, P2, P3
    service: str
    namespace: Optional[str] = "default"
    details: Optional[str] = ""
    runbook_url: Optional[str] = None
    source: Optional[str] = "manual"
 
 
@app.post("/incident/create")
async def create_incident(alert: Alert):
    """Receive an alert and start the AI incident commander."""
    incident_id = f"INC-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
 
    # Post initial Slack message
    post_slack_message(
        f":rotating_light: *Incident {incident_id}* — {alert.severity}\n"
        f"*Service*: {alert.service}\n"
        f"*Alert*: {alert.title}\n"
        f"_AI Incident Commander gathering context..._"
    )
 
    # Run AI analysis
    brief = await run_incident_commander(alert, incident_id)
 
    # Post structured brief
    post_slack_message(brief)
 
    return {"incident_id": incident_id, "brief": brief}

INCIDENT_TOOLS = [
    {
        "name": "kubectl_get",
        "description": "Get Kubernetes resources. Returns output of kubectl get.",
        "input_schema": {
            "type": "object",
            "properties": {
                "resource": {"type": "string"},
                "namespace": {"type": "string", "default": "default"},
                "selector": {"type": "string", "description": "Label selector like app=my-api"}
            },
            "required": ["resource"]
        }
    },
    {
        "name": "kubectl_events",
        "description": "Get recent events in a namespace, sorted by time.",
        "input_schema": {
            "type": "object",
            "properties": {
                "namespace": {"type": "string"},
                "warning_only": {"type": "boolean", "default": True}
            }
        }
    },
    {
        "name": "kubectl_logs",
        "description": "Get logs from a pod.",
        "input_schema": {
            "type": "object",
            "properties": {
                "pod_name": {"type": "string"},
                "namespace": {"type": "string"},
                "lines": {"type": "integer", "default": 50},
                "previous": {"type": "boolean", "default": False},
                "container": {"type": "string"}
            },
            "required": ["pod_name"]
        }
    },
    {
        "name": "check_node_status",
        "description": "Check all nodes for resource pressure, conditions, and capacity.",
        "input_schema": {
            "type": "object",
            "properties": {}
        }
    },
    {
        "name": "get_deployment_history",
        "description": "Check recent deployment history for a service to identify if a recent deploy caused the incident.",
        "input_schema": {
            "type": "object",
            "properties": {
                "deployment_name": {"type": "string"},
                "namespace": {"type": "string"}
            },
            "required": ["deployment_name", "namespace"]
        }
    }
]
 
 
def execute_incident_tool(name: str, input_data: dict) -> str:
    """Execute incident tools — all read-only."""
    try:
        if name == "kubectl_get":
            cmd = ["kubectl", "get", input_data["resource"], "--no-headers"]
            if ns := input_data.get("namespace"):
                cmd.extend(["-n", ns])
            if sel := input_data.get("selector"):
                cmd.extend(["-l", sel])
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            return (result.stdout or result.stderr)[:3000]
 
        elif name == "kubectl_events":
            cmd = ["kubectl", "get", "events",
                   "-n", input_data.get("namespace", "default"),
                   "--sort-by='.lastTimestamp'"]
            if input_data.get("warning_only", True):
                cmd.extend(["--field-selector=type=Warning"])
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            return (result.stdout or result.stderr)[-2000:]
 
        elif name == "kubectl_logs":
            cmd = ["kubectl", "logs", input_data["pod_name"],
                   "-n", input_data.get("namespace", "default"),
                   f"--tail={input_data.get('lines', 50)}"]
            if input_data.get("previous"):
                cmd.append("--previous")
            if c := input_data.get("container"):
                cmd.extend(["-c", c])
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            return (result.stdout or result.stderr)[-3000:]
 
        elif name == "check_node_status":
            result = subprocess.run(
                ["kubectl", "get", "nodes", "-o", "wide"],
                capture_output=True, text=True, timeout=30
            )
            describe_result = subprocess.run(
                ["kubectl", "describe", "nodes"],
                capture_output=True, text=True, timeout=30
            )
            output = result.stdout + "\n\n" + describe_result.stdout
            return output[:4000]
 
        elif name == "get_deployment_history":
            result = subprocess.run(
                ["kubectl", "rollout", "history", "deployment",
                 input_data["deployment_name"],
                 "-n", input_data["namespace"]],
                capture_output=True, text=True, timeout=30
            )
            return result.stdout or result.stderr
 
    except Exception as e:
        return f"Tool error: {str(e)}"
 
    return "Unknown tool"

async def run_incident_commander(alert: Alert, incident_id: str) -> str:
    """
    Run the AI incident commander — tool use loop with structured output.
    """
    system_prompt = f"""You are an expert SRE acting as Incident Commander for a production incident.
 
Incident ID: {incident_id}
Severity: {alert.severity}
Service: {alert.service}
Namespace: {alert.namespace}
Alert: {alert.title}
Details: {alert.details}
 
Your job:
1. Gather context systematically — do not guess, verify with tools
2. Check pods, events, logs, and deployment history in that order
3. Form a hypothesis about the root cause
4. Generate an actionable runbook
 
Focus on: 
- Is there a recent deployment that could have caused this?
- Are pods healthy? Any CrashLoopBackOff, OOMKilled, Pending?
- What do the warning events show?
- What do the pod logs show around the time of the alert?
 
You have read-only access — diagnosis only, no remediation actions."""
 
    messages = [
        {
            "role": "user",
            "content": f"Alert received: {alert.title}. Service: {alert.service} in namespace {alert.namespace}. Start incident investigation."
        }
    ]
 
    for iteration in range(10):
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=4000,
            system=system_prompt,
            tools=INCIDENT_TOOLS,
            messages=messages
        )
 
        messages.append({"role": "assistant", "content": response.content})
 
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return format_incident_brief(incident_id, alert, block.text)
            break
 
        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_incident_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            messages.append({"role": "user", "content": tool_results})
 
    return "Incident analysis timed out. Manual investigation required."
 
 
def format_incident_brief(incident_id: str, alert: Alert, analysis: str) -> str:
    """Format the AI analysis as a Slack-friendly incident brief."""
    return f"""
:red_circle: *INCIDENT BRIEF — {incident_id}*
*Severity*: {alert.severity} | *Service*: {alert.service}
 
{analysis}
 
---
_Generated by AI Incident Commander at {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}_
_This is an AI-assisted first pass — verify all findings before taking action._
"""
 
 
def post_slack_message(message: str):
    """Post to Slack webhook."""
    if not SLACK_WEBHOOK:
        print(f"[SLACK] {message}")
        return
    payload = json.dumps({"text": message}).encode()
    req = urllib.request.Request(
        SLACK_WEBHOOK,
        data=payload,
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req)