Build an AI Deployment Health Checker with Claude API and Kubernetes

pip install anthropic kubernetes python-dotenv

from kubernetes import client, config
from kubernetes.client.rest import ApiException
from typing import Optional
import json
 
 
def load_kube_config(in_cluster: bool = False):
    """Load kubeconfig — in-cluster for prod, local for dev."""
    if in_cluster:
        config.load_incluster_config()
    else:
        config.load_kube_config()
 
 
def get_deployment_health(namespace: str, deployment_name: str) -> dict:
    """
    Collect everything needed to diagnose a deployment:
    - Deployment spec and status
    - Pod states and conditions
    - Recent pod events
    - Container resource usage
    """
    apps_v1 = client.AppsV1Api()
    core_v1 = client.CoreV1Api()
 
    result = {
        "deployment": {},
        "pods": [],
        "events": [],
        "errors": []
    }
 
    # Get deployment details
    try:
        deployment = apps_v1.read_namespaced_deployment(
            name=deployment_name,
            namespace=namespace
        )
        status = deployment.status
        spec = deployment.spec
 
        result["deployment"] = {
            "name": deployment_name,
            "namespace": namespace,
            "desired_replicas": spec.replicas,
            "ready_replicas": status.ready_replicas or 0,
            "available_replicas": status.available_replicas or 0,
            "updated_replicas": status.updated_replicas or 0,
            "conditions": [
                {
                    "type": c.type,
                    "status": c.status,
                    "reason": c.reason,
                    "message": c.message
                }
                for c in (status.conditions or [])
            ],
            "strategy": spec.strategy.type,
            "image": spec.template.spec.containers[0].image if spec.template.spec.containers else "unknown",
            "resources": _extract_resources(spec.template.spec.containers)
        }
    except ApiException as e:
        result["errors"].append(f"Could not fetch deployment: {e.reason}")
        return result
 
    # Get pods belonging to this deployment
    try:
        label_selector = _get_label_selector(deployment)
        pods = core_v1.list_namespaced_pod(
            namespace=namespace,
            label_selector=label_selector
        )
 
        for pod in pods.items:
            pod_info = {
                "name": pod.metadata.name,
                "phase": pod.status.phase,
                "conditions": [],
                "containers": [],
                "node": pod.spec.node_name
            }
 
            # Pod conditions
            for condition in (pod.status.conditions or []):
                pod_info["conditions"].append({
                    "type": condition.type,
                    "status": condition.status,
                    "reason": condition.reason,
                    "message": condition.message
                })
 
            # Container states
            for cs in (pod.status.container_statuses or []):
                container_info = {
                    "name": cs.name,
                    "ready": cs.ready,
                    "restart_count": cs.restart_count,
                    "image": cs.image,
                    "state": {}
                }
 
                if cs.state.running:
                    container_info["state"] = {"type": "running"}
                elif cs.state.waiting:
                    container_info["state"] = {
                        "type": "waiting",
                        "reason": cs.state.waiting.reason,
                        "message": cs.state.waiting.message
                    }
                elif cs.state.terminated:
                    container_info["state"] = {
                        "type": "terminated",
                        "reason": cs.state.terminated.reason,
                        "exit_code": cs.state.terminated.exit_code,
                        "message": cs.state.terminated.message
                    }
 
                # Last terminated state (useful for crash loops)
                if cs.last_state.terminated:
                    lt = cs.last_state.terminated
                    container_info["last_terminated"] = {
                        "reason": lt.reason,
                        "exit_code": lt.exit_code,
                        "message": lt.message
                    }
 
                pod_info["containers"].append(container_info)
 
            result["pods"].append(pod_info)
 
    except ApiException as e:
        result["errors"].append(f"Could not fetch pods: {e.reason}")
 
    # Get recent events
    try:
        events = core_v1.list_namespaced_event(
            namespace=namespace,
            field_selector=f"involvedObject.name={deployment_name}"
        )
 
        # Also get pod events
        for pod in result["pods"]:
            pod_events = core_v1.list_namespaced_event(
                namespace=namespace,
                field_selector=f"involvedObject.name={pod['name']}"
            )
            events.items.extend(pod_events.items)
 
        # Sort by time, take most recent 15
        sorted_events = sorted(
            events.items,
            key=lambda e: e.last_timestamp or e.event_time or "",
            reverse=True
        )[:15]
 
        result["events"] = [
            {
                "type": e.type,
                "reason": e.reason,
                "message": e.message,
                "count": e.count,
                "source": e.source.component if e.source else "unknown"
            }
            for e in sorted_events
        ]
 
    except ApiException as e:
        result["errors"].append(f"Could not fetch events: {e.reason}")
 
    return result
 
 
def _get_label_selector(deployment) -> str:
    """Extract label selector from deployment spec."""
    selector = deployment.spec.selector.match_labels or {}
    return ",".join(f"{k}={v}" for k, v in selector.items())
 
 
def _extract_resources(containers) -> list:
    """Extract resource requests and limits from containers."""
    resources = []
    for container in (containers or []):
        if container.resources:
            r = container.resources
            resources.append({
                "container": container.name,
                "requests": {
                    "cpu": r.requests.get("cpu") if r.requests else None,
                    "memory": r.requests.get("memory") if r.requests else None
                },
                "limits": {
                    "cpu": r.limits.get("cpu") if r.limits else None,
                    "memory": r.limits.get("memory") if r.limits else None
                }
            })
    return resources

import anthropic
import json
import sys
from k8s_client import load_kube_config, get_deployment_health
from dotenv import load_dotenv
import os
 
load_dotenv()
 
 
def diagnose_deployment(namespace: str, deployment_name: str) -> str:
    """
    Main function: collect K8s data, send to Claude, return diagnosis.
    """
    print(f"Collecting health data for {namespace}/{deployment_name}...")
 
    # Load kubeconfig
    load_kube_config(in_cluster=False)
 
    # Collect all deployment context
    health_data = get_deployment_health(namespace, deployment_name)
 
    if health_data["errors"] and not health_data["deployment"]:
        return f"Error: Could not fetch deployment data.\n" + "\n".join(health_data["errors"])
 
    # Build the prompt
    prompt = _build_diagnosis_prompt(deployment_name, namespace, health_data)
 
    # Send to Claude
    print("Sending to Claude API for diagnosis...")
    client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
 
    message = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=1500,
        messages=[
            {
                "role": "user",
                "content": prompt
            }
        ]
    )
 
    return message.content[0].text
 
 
def _build_diagnosis_prompt(deployment_name: str, namespace: str, health_data: dict) -> str:
    """Build a structured prompt with all K8s context."""
 
    deployment = health_data["deployment"]
    pods = health_data["pods"]
    events = health_data["events"]
 
    # Determine health status
    desired = deployment.get("desired_replicas", 0)
    ready = deployment.get("ready_replicas", 0)
    is_healthy = desired > 0 and desired == ready
 
    prompt = f"""You are a Kubernetes expert helping diagnose a deployment issue.
 
## Deployment: {deployment_name} (namespace: {namespace})
 
### Status
- Desired replicas: {desired}
- Ready replicas: {ready}
- Available replicas: {deployment.get('available_replicas', 0)}
- Health: {'HEALTHY' if is_healthy else 'UNHEALTHY'}
- Strategy: {deployment.get('strategy', 'Unknown')}
- Image: {deployment.get('image', 'Unknown')}
 
### Deployment Conditions
{json.dumps(deployment.get('conditions', []), indent=2)}
 
### Resource Configuration
{json.dumps(deployment.get('resources', []), indent=2)}
 
### Pod States ({len(pods)} pods found)
{json.dumps(pods, indent=2)}
 
### Recent Events (most recent first)
{json.dumps(events, indent=2)}
 
---
 
Based on this data, provide:
 
1. **Health Assessment**: Is this deployment healthy or not? One sentence.
 
2. **Root Cause**: What is the specific problem (if any)? Be precise — reference exact pod names, container states, exit codes, or event messages.
 
3. **Explanation**: Why is this happening? Keep it concise and practical.
 
4. **Fix**: Exact kubectl commands or configuration changes to resolve the issue. If multiple fixes are possible, list them in order of likelihood.
 
5. **Prevention**: One practical tip to prevent this in the future.
 
Keep the response focused and actionable. Skip any issue that is clearly not present in the data."""
 
    return prompt
 
 
def format_output(diagnosis: str, deployment_name: str) -> str:
    """Format the output nicely for terminal."""
    separator = "=" * 60
    return f"""
{separator}
AI DEPLOYMENT HEALTH DIAGNOSIS
Deployment: {deployment_name}
{separator}
 
{diagnosis}
 
{separator}
"""
 
 
if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("Usage: python checker.py <namespace> <deployment-name>")
        print("Example: python checker.py default my-api")
        sys.exit(1)
 
    namespace = sys.argv[1]
    deployment_name = sys.argv[2]
 
    diagnosis = diagnose_deployment(namespace, deployment_name)
    print(format_output(diagnosis, deployment_name))

ANTHROPIC_API_KEY=sk-ant-your-key-here

# Check a specific deployment
python checker.py production my-api
 
# Check a deployment in a specific namespace
python checker.py staging payment-service