Build an AI Kubernetes Runbook Generator with Claude API

pip install anthropic kubernetes python-dotenv click

from kubernetes import client, config
from kubernetes.client.rest import ApiException
import json
import yaml
 
 
def load_config():
    try:
        config.load_incluster_config()
    except config.ConfigException:
        config.load_kube_config()
 
 
def get_resource_context(namespace: str, resource_type: str, name: str) -> dict:
    """
    Collect a Kubernetes resource and all related resources.
    Returns everything needed to write a meaningful runbook.
    """
    load_config()
    context = {
        "primary": {},
        "related": {},
        "errors": []
    }
 
    apps_v1 = client.AppsV1Api()
    core_v1 = client.CoreV1Api()
    autoscaling_v2 = client.AutoscalingV2Api()
    policy_v1 = client.PolicyV1Api()
    batch_v1 = client.BatchV1Api()
 
    try:
        if resource_type.lower() == "deployment":
            resource = apps_v1.read_namespaced_deployment(name, namespace)
            context["primary"] = _serialize_resource(resource)
 
            # Get pods
            label_selector = _make_selector(resource.spec.selector.match_labels)
            pods = core_v1.list_namespaced_pod(namespace, label_selector=label_selector)
            context["related"]["pods"] = [_serialize_resource(p) for p in pods.items[:3]]
 
            # Get HPA if it exists
            try:
                hpa = autoscaling_v2.read_namespaced_horizontal_pod_autoscaler(name, namespace)
                context["related"]["hpa"] = _serialize_resource(hpa)
            except ApiException:
                pass
 
            # Get PDB if it exists
            try:
                pdbs = policy_v1.list_namespaced_pod_disruption_budget(namespace)
                for pdb in pdbs.items:
                    if pdb.spec.selector and _matches_labels(
                        pdb.spec.selector.match_labels,
                        resource.spec.selector.match_labels
                    ):
                        context["related"]["pdb"] = _serialize_resource(pdb)
                        break
            except ApiException:
                pass
 
        elif resource_type.lower() == "statefulset":
            resource = apps_v1.read_namespaced_stateful_set(name, namespace)
            context["primary"] = _serialize_resource(resource)
 
            label_selector = _make_selector(resource.spec.selector.match_labels)
            pods = core_v1.list_namespaced_pod(namespace, label_selector=label_selector)
            context["related"]["pods"] = [_serialize_resource(p) for p in pods.items[:3]]
 
        elif resource_type.lower() == "cronjob":
            resource = batch_v1.read_namespaced_cron_job(name, namespace)
            context["primary"] = _serialize_resource(resource)
 
            # Get recent Jobs from this CronJob
            jobs = batch_v1.list_namespaced_job(namespace)
            related_jobs = [
                j for j in jobs.items
                if j.metadata.owner_references and
                any(ref.name == name for ref in j.metadata.owner_references)
            ]
            context["related"]["recent_jobs"] = [
                _serialize_resource(j) for j in sorted(
                    related_jobs,
                    key=lambda j: j.metadata.creation_timestamp or "",
                    reverse=True
                )[:5]
            ]
 
        # Get related Services (for all types)
        try:
            services = core_v1.list_namespaced_service(namespace)
            matching = []
            primary_labels = context["primary"].get("spec", {}).get("selector", {}) or \
                             context["primary"].get("spec", {}).get("template", {}).get("metadata", {}).get("labels", {})
 
            for svc in services.items:
                if svc.spec.selector and _matches_labels(svc.spec.selector, primary_labels):
                    matching.append(_serialize_resource(svc))
            if matching:
                context["related"]["services"] = matching
        except ApiException:
            pass
 
        # Get ConfigMaps and Secrets referenced in env
        try:
            containers = (context["primary"].get("spec", {})
                         .get("template", {})
                         .get("spec", {})
                         .get("containers", []))
            cm_names = set()
            for container in containers:
                for env_from in container.get("envFrom", []):
                    if "configMapRef" in env_from:
                        cm_names.add(env_from["configMapRef"]["name"])
 
            cms = {}
            for cm_name in cm_names:
                try:
                    cm = core_v1.read_namespaced_config_map(cm_name, namespace)
                    # Don't include data values (may be sensitive), just keys
                    cms[cm_name] = {"keys": list((cm.data or {}).keys())}
                except ApiException:
                    pass
            if cms:
                context["related"]["configmaps"] = cms
        except Exception:
            pass
 
    except ApiException as e:
        context["errors"].append(f"Could not fetch {resource_type}/{name}: {e.reason}")
 
    return context
 
 
def _serialize_resource(resource) -> dict:
    """Convert K8s resource object to dict, removing status noise."""
    data = resource.to_dict()
    # Clean up managed fields noise
    if "metadata" in data:
        data["metadata"].pop("managed_fields", None)
        data["metadata"].pop("annotations", None)
    return data
 
 
def _make_selector(labels: dict) -> str:
    return ",".join(f"{k}={v}" for k, v in (labels or {}).items())
 
 
def _matches_labels(selector: dict, resource_labels: dict) -> bool:
    if not selector or not resource_labels:
        return False
    return all(resource_labels.get(k) == v for k, v in selector.items())

import anthropic
import os
from dotenv import load_dotenv
 
load_dotenv()
 
 
def build_runbook_prompt(resource_type, name, namespace, primary_resource, related_resources):
    json_fence = "```json"
    bash_fence = "```bash"
    close_fence = "```"
    return (
        f"Generate a complete operational runbook for the following Kubernetes resource.\n\n"
        f"## Resource Information\n\n"
        f"**Type:** {resource_type}\n**Name:** {name}\n**Namespace:** {namespace}\n\n"
        f"## Resource Configuration\n{json_fence}\n{primary_resource}\n{close_fence}\n\n"
        f"## Related Resources\n{json_fence}\n{related_resources}\n{close_fence}\n\n"
        "---\n\n"
        "Generate a runbook with these sections:\n\n"
        "## 1. Overview — what this resource does, business criticality, key dependencies\n"
        "## 2. Normal Operation — exact kubectl commands to verify health, expected pod count\n"
        "## 3. Common Issues and Fixes — CrashLoopBackOff, PVC issues, HPA not scaling\n"
        "## 4. Scaling Operations — manual scale commands, HPA behavior\n"
        "## 5. Deployment / Update Procedure — safe rollout and rollback commands\n"
        "## 6. Emergency Procedures — restart pods, take offline, restore from backup\n"
        f"## 7. Useful Commands Reference\n{bash_fence}\n# Commands with exact names\n{close_fence}\n\n"
        "Make the runbook specific — use actual name, namespace, labels. Include exact kubectl commands."
    )
 
 
def generate_runbook(
    namespace: str,
    resource_type: str,
    name: str,
    output_format: str = "markdown"
) -> str:
    """Generate a complete operational runbook using Claude API."""
 
    print(f"Collecting K8s context for {resource_type}/{name}...")
    context = get_resource_context(namespace, resource_type, name)
 
    if context["errors"] and not context["primary"]:
        return f"Error: {chr(10).join(context['errors'])}"
 
    prompt = build_runbook_prompt(
        resource_type=resource_type,
        name=name,
        namespace=namespace,
        primary_resource=json.dumps(context["primary"], indent=2, default=str)[:4000],
        related_resources=json.dumps(context["related"], indent=2, default=str)[:3000]
    )
 
    print("Generating runbook with Claude API...")
    anthropic_client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
 
    message = anthropic_client.messages.create(
        model="claude-sonnet-5",
        max_tokens=3000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    return message.content[0].text
 
 
def save_runbook(content: str, name: str, resource_type: str, output_dir: str = "runbooks"):
    """Save runbook to a Markdown file."""
    import os
    os.makedirs(output_dir, exist_ok=True)
    filename = f"{output_dir}/{resource_type}-{name}-runbook.md"
    with open(filename, "w") as f:
        f.write(content)
    print(f"Runbook saved: {filename}")
    return filename

import click
 
 
@click.command()
@click.argument("resource_type")
@click.argument("name")
@click.option("--namespace", "-n", default="default", help="Kubernetes namespace")
@click.option("--output", "-o", default="runbooks", help="Output directory")
@click.option("--print", "print_output", is_flag=True, help="Print to stdout instead of saving")
def main(resource_type, name, namespace, output, print_output):
    """
    Generate an operational runbook for a Kubernetes resource.
 
    Examples:
      python runbook_gen.py deployment my-api -n production
      python runbook_gen.py statefulset postgres -n databases
      python runbook_gen.py cronjob nightly-backup -n production
    """
    runbook = generate_runbook(namespace, resource_type, name)
 
    if print_output:
        print(runbook)
    else:
        save_runbook(runbook, name, resource_type, output)
 
 
if __name__ == "__main__":
    main()

# Generate runbook for a production deployment
python runbook_gen.py deployment payment-api -n production
 
# Generate runbook for a StatefulSet
python runbook_gen.py statefulset postgres -n databases
 
# Generate runbook for a CronJob and print to stdout
python runbook_gen.py cronjob nightly-cleanup -n production --print
 
# Generate and save to custom directory
python runbook_gen.py deployment api-gateway -n staging -o docs/runbooks

kubectl get deployment payment-api -n production
# Expected: READY 3/3
 
kubectl get pods -n production -l app=payment-api
# All pods should show STATUS=Running, RESTARTS=0

# Check logs from previous container
kubectl logs -n production -l app=payment-api --previous
 
# Verify secret exists
kubectl get secret payment-secrets -n production

# .github/workflows/update-runbooks.yml
- name: Update runbooks after deploy
  run: |
    pip install anthropic kubernetes click
    python runbook_gen.py deployment ${{ env.SERVICE_NAME }} \
      -n ${{ env.NAMESPACE }} \
      -o docs/runbooks
    git add docs/runbooks/
    git commit -m "Auto-update runbook for ${{ env.SERVICE_NAME }}"
    git push
  env:
    ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}