Build a Kubernetes Resource Optimizer with Claude API and Python

pip install anthropic kubernetes tabulate
# Metrics Server must be installed in your cluster
kubectl top pods -A  # Test that metrics work

import anthropic
import json
import subprocess
from kubernetes import client, config
from tabulate import tabulate
 
 
def get_cluster_resources() -> list[dict]:
    """Pull all Deployment resource specs from Kubernetes API."""
    config.load_kube_config()
    apps_v1 = client.AppsV1Api()
    core_v1 = client.CoreV1Api()
 
    workloads = []
 
    deployments = apps_v1.list_deployment_for_all_namespaces()
    for dep in deployments.items:
        namespace = dep.metadata.namespace
        name = dep.metadata.name
        replicas = dep.spec.replicas or 1
 
        for container in dep.spec.template.spec.containers:
            resources = container.resources or client.V1ResourceRequirements()
            req = resources.requests or {}
            lim = resources.limits or {}
 
            workloads.append({
                "name": f"{namespace}/{name}/{container.name}",
                "replicas": replicas,
                "cpu_request": req.get("cpu", "not set"),
                "cpu_limit": lim.get("cpu", "not set"),
                "memory_request": req.get("memory", "not set"),
                "memory_limit": lim.get("memory", "not set"),
            })
 
    return workloads
 
 
def get_actual_usage() -> dict[str, dict]:
    """Get actual CPU/memory usage from kubectl top."""
    usage = {}
    try:
        result = subprocess.run(
            ["kubectl", "top", "pods", "-A", "--no-headers"],
            capture_output=True, text=True, timeout=30
        )
        for line in result.stdout.strip().split("\n"):
            if not line:
                continue
            parts = line.split()
            if len(parts) >= 4:
                namespace, pod_name = parts[0], parts[1]
                cpu, memory = parts[2], parts[3]
                key = f"{namespace}/{pod_name}"
                usage[key] = {"cpu": cpu, "memory": memory}
    except Exception as e:
        print(f"Could not get metrics: {e}")
 
    return usage
 
 
def analyze_with_claude(workloads: list[dict], usage: dict) -> str:
    """Send resource data to Claude API for rightsizing analysis."""
    cl = anthropic.Anthropic()
 
    workload_json = json.dumps(workloads[:50], indent=2)  # Limit to 50 for token budget
    usage_json = json.dumps(dict(list(usage.items())[:50]), indent=2)
 
    prompt = f"""You are a Kubernetes FinOps expert. Analyze these workload resource configurations and actual usage data.
 
## Declared Resources (requests/limits)
{workload_json}
 
## Actual Usage (from kubectl top pods)
{usage_json}
 
Provide:
1. Top 10 most over-provisioned workloads (highest waste)
2. Specific recommended values for cpu_request, cpu_limit, memory_request, memory_limit for each
3. Estimated monthly savings if on AWS (assume $0.048/vCPU-hour, $0.006/GB-hour)
4. Workloads with NO resource limits set (risk flag)
5. One-sentence executive summary
 
Format as JSON:
{{
  "over_provisioned": [
    {{
      "workload": "namespace/name/container",
      "current_cpu_request": "...",
      "recommended_cpu_request": "...",
      "current_memory_request": "...",
      "recommended_memory_request": "...",
      "monthly_savings_usd": X,
      "reason": "..."
    }}
  ],
  "no_limits_set": ["list of workloads"],
  "total_monthly_savings_usd": X,
  "summary": "..."
}}"""
 
    response = cl.messages.create(
        model="claude-sonnet-5",
        max_tokens=3000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    return response.content[0].text
 
 
def parse_and_display(analysis_text: str):
    """Parse Claude's response and display as tables."""
    text = analysis_text.strip()
    if "```" in text:
        text = text.split("```")[1]
        if text.startswith("json"):
            text = text[4:]
 
    try:
        data = json.loads(text)
 
        print("\n" + "=" * 70)
        print("KUBERNETES RESOURCE OPTIMIZATION REPORT")
        print("=" * 70)
 
        print(f"\n{data.get('summary', '')}")
        print(f"\nTotal Potential Monthly Savings: ${data.get('total_monthly_savings_usd', 0):,.2f}")
 
        over = data.get("over_provisioned", [])
        if over:
            print("\n\nTOP OVER-PROVISIONED WORKLOADS")
            print("-" * 70)
            table_data = []
            for item in over:
                table_data.append([
                    item["workload"][:35],
                    item.get("current_cpu_request", "?"),
                    item.get("recommended_cpu_request", "?"),
                    item.get("current_memory_request", "?"),
                    item.get("recommended_memory_request", "?"),
                    f"${item.get('monthly_savings_usd', 0):.0f}/mo",
                ])
            print(tabulate(
                table_data,
                headers=["Workload", "CPU Req Now", "CPU Req Rec", "Mem Req Now", "Mem Req Rec", "Savings"],
                tablefmt="grid"
            ))
 
        no_limits = data.get("no_limits_set", [])
        if no_limits:
            print(f"\n\nWORKLOADS WITHOUT RESOURCE LIMITS ({len(no_limits)} total)")
            print("These can cause node OOM and cluster instability:")
            for w in no_limits[:10]:
                print(f"  - {w}")
 
    except json.JSONDecodeError:
        print("\nRaw Claude Analysis:")
        print(analysis_text)
 
 
def generate_patch_commands(analysis_text: str):
    """Generate kubectl patch commands from recommendations."""
    text = analysis_text.strip()
    if "```" in text:
        text = text.split("```")[1]
        if text.startswith("json"):
            text = text[4:]
 
    try:
        data = json.loads(text)
        print("\n\nKUBECTL PATCH COMMANDS (review before applying)")
        print("-" * 70)
 
        for item in data.get("over_provisioned", [])[:5]:
            parts = item["workload"].split("/")
            if len(parts) < 3:
                continue
            namespace, deployment, container = parts[0], parts[1], parts[2]
 
            cpu_req = item.get("recommended_cpu_request", "")
            mem_req = item.get("recommended_memory_request", "")
 
            if cpu_req and mem_req:
                print(f"""
# {deployment} in {namespace}
kubectl patch deployment {deployment} -n {namespace} --type='json' -p='[
  {{"op": "replace", "path": "/spec/template/spec/containers/0/resources/requests/cpu", "value": "{cpu_req}"}},
  {{"op": "replace", "path": "/spec/template/spec/containers/0/resources/requests/memory", "value": "{mem_req}"}}
]'""")
 
    except Exception:
        pass
 
 
if __name__ == "__main__":
    print("Collecting Kubernetes resource data...")
    workloads = get_cluster_resources()
    print(f"Found {len(workloads)} containers across all namespaces")
 
    print("Collecting actual usage metrics...")
    usage = get_actual_usage()
    print(f"Got metrics for {len(usage)} pods")
 
    print("Analyzing with Claude API...")
    analysis = analyze_with_claude(workloads, usage)
 
    parse_and_display(analysis)
    generate_patch_commands(analysis)

# Make sure kubectl context is set to your target cluster
kubectl config current-context
 
python k8s_optimizer.py