Build a Slack Bot That Monitors Kubernetes Errors Using Claude API

mkdir k8s-slack-bot && cd k8s-slack-bot
python -m venv venv && source venv/bin/activate
 
pip install kubernetes anthropic slack-sdk python-dotenv

# .env
ANTHROPIC_API_KEY=sk-ant-...
SLACK_BOT_TOKEN=xoxb-...
SLACK_CHANNEL=#k8s-alerts
KUBERNETES_NAMESPACE=default  # or "all" for all namespaces

# watcher.py
import asyncio
from kubernetes import client, config, watch
from kubernetes.client.exceptions import ApiException
 
def get_k8s_client():
    try:
        config.load_incluster_config()  # inside cluster
    except:
        config.load_kube_config()       # local kubectl
    return client.CoreV1Api()
 
def watch_pod_events(namespace="default"):
    v1 = get_k8s_client()
    w = watch.Watch()
    
    error_reasons = {
        "CrashLoopBackOff",
        "OOMKilled", 
        "BackOff",
        "Failed",
        "FailedScheduling",
        "FailedMount",
        "Unhealthy",
        "ImagePullBackOff",
        "ErrImagePull",
    }
    
    print(f"Watching namespace: {namespace}")
    
    try:
        if namespace == "all":
            stream = w.stream(v1.list_event_for_all_namespaces, timeout_seconds=0)
        else:
            stream = w.stream(v1.list_namespaced_event, namespace=namespace, timeout_seconds=0)
        
        for event in stream:
            obj = event['object']
            reason = obj.reason or ""
            
            if reason in error_reasons:
                yield {
                    "reason": reason,
                    "message": obj.message,
                    "namespace": obj.metadata.namespace,
                    "name": obj.involved_object.name,
                    "kind": obj.involved_object.kind,
                    "count": obj.count,
                    "first_time": str(obj.first_timestamp),
                    "last_time": str(obj.last_timestamp),
                }
    except ApiException as e:
        print(f"Kubernetes API error: {e}")

# analyzer.py
import anthropic
import os
 
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
 
def analyze_k8s_error(error_event: dict) -> dict:
    prompt = f"""You are a Kubernetes expert helping a DevOps engineer debug a cluster issue.
 
A Kubernetes error event occurred:
- Kind: {error_event['kind']}
- Name: {error_event['name']}
- Namespace: {error_event['namespace']}
- Reason: {error_event['reason']}
- Message: {error_event['message']}
- Occurred: {error_event['count']} times
 
Provide:
1. A simple 2-sentence explanation of what happened (no jargon)
2. The 3 most likely root causes
3. Specific kubectl commands to investigate and fix
 
Keep it concise and actionable. Format as JSON with keys: explanation, causes (list), fix_commands (list of strings)."""
 
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=600,
        messages=[{"role": "user", "content": prompt}]
    )
    
    import json
    try:
        return json.loads(message.content[0].text)
    except:
        return {
            "explanation": message.content[0].text,
            "causes": [],
            "fix_commands": []
        }

# notifier.py
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import os
 
slack_client = WebClient(token=os.getenv("SLACK_BOT_TOKEN"))
channel = os.getenv("SLACK_CHANNEL", "#k8s-alerts")
 
def send_k8s_alert(error_event: dict, analysis: dict):
    causes_text = "\n".join([f"• {c}" for c in analysis.get("causes", [])])
    commands_text = "\n".join([f"`{cmd}`" for cmd in analysis.get("fix_commands", [])])
    
    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": f"🚨 {error_event['reason']} — {error_event['kind']}/{error_event['name']}"
            }
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*Namespace:*\n{error_event['namespace']}"},
                {"type": "mrkdwn", "text": f"*Count:*\n{error_event['count']} times"},
            ]
        },
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*What happened:*\n{analysis.get('explanation', 'No explanation available')}"
            }
        },
    ]
    
    if causes_text:
        blocks.append({
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"*Likely causes:*\n{causes_text}"}
        })
    
    if commands_text:
        blocks.append({
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"*Investigate with:*\n{commands_text}"}
        })
    
    try:
        slack_client.chat_postMessage(channel=channel, blocks=blocks)
        print(f"Alert sent for {error_event['name']}")
    except SlackApiError as e:
        print(f"Slack error: {e.response['error']}")

# main.py
import os
import time
from dotenv import load_dotenv
from watcher import watch_pod_events
from analyzer import analyze_k8s_error
from notifier import send_k8s_alert
 
load_dotenv()
 
# Deduplicate — don't spam the same error
seen_errors = {}
 
def should_alert(error_event: dict) -> bool:
    key = f"{error_event['namespace']}/{error_event['name']}/{error_event['reason']}"
    last_count = seen_errors.get(key, 0)
    
    # Alert on first occurrence, then only every 10 more
    if error_event['count'] == 1 or (error_event['count'] - last_count) >= 10:
        seen_errors[key] = error_event['count']
        return True
    return False
 
def main():
    namespace = os.getenv("KUBERNETES_NAMESPACE", "default")
    print(f"K8s Slack Bot started — watching {namespace}")
    
    for error_event in watch_pod_events(namespace):
        if not should_alert(error_event):
            continue
        
        print(f"Error detected: {error_event['reason']} on {error_event['name']}")
        
        # Analyze with Claude
        analysis = analyze_k8s_error(error_event)
        
        # Send to Slack
        send_k8s_alert(error_event, analysis)
        
        # Small delay to avoid rate limits
        time.sleep(1)
 
if __name__ == "__main__":
    main()

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: k8s-slack-bot
  namespace: monitoring
spec:
  replicas: 1
  selector:
    matchLabels:
      app: k8s-slack-bot
  template:
    metadata:
      labels:
        app: k8s-slack-bot
    spec:
      serviceAccountName: k8s-slack-bot-sa
      containers:
        - name: bot
          image: python:3.11-slim
          command: ["python", "/app/main.py"]
          env:
            - name: ANTHROPIC_API_KEY
              valueFrom:
                secretKeyRef:
                  name: bot-secrets
                  key: anthropic-key
            - name: SLACK_BOT_TOKEN
              valueFrom:
                secretKeyRef:
                  name: bot-secrets
                  key: slack-token
            - name: KUBERNETES_NAMESPACE
              value: "all"

# rbac.yaml — bot needs read access to events
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: k8s-slack-bot
rules:
  - apiGroups: [""]
    resources: ["events", "pods"]
    verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: k8s-slack-bot
subjects:
  - kind: ServiceAccount
    name: k8s-slack-bot-sa
    namespace: monitoring
roleRef:
  kind: ClusterRole
  name: k8s-slack-bot
  apiGroup: rbac.authorization.k8s.io