Build an AI Kubernetes YAML Explainer and Validator with Claude API

pip install anthropic fastapi uvicorn python-multipart pyyaml python-dotenv

import yaml
import json
import os
from typing import Union
import anthropic
 
 
def load_kubernetes_yaml(content: str) -> list[dict]:
    """Parse YAML content — handles multi-document YAML (---)."""
    documents = []
    for doc in yaml.safe_load_all(content):
        if doc is not None:
            documents.append(doc)
    return documents
 
 
def extract_resource_context(doc: dict) -> dict:
    """Extract key fields for context-aware analysis."""
    kind = doc.get("kind", "Unknown")
    name = doc.get("metadata", {}).get("name", "unnamed")
    namespace = doc.get("metadata", {}).get("namespace", "default")
    spec = doc.get("spec", {})
 
    context = {
        "kind": kind,
        "name": name,
        "namespace": namespace,
    }
 
    # Extract container specs for Deployments, Pods, StatefulSets
    containers = []
    if kind in ["Deployment", "StatefulSet", "DaemonSet", "Job", "CronJob"]:
        template_spec = spec.get("template", {}).get("spec", {})
        containers = template_spec.get("containers", [])
        init_containers = template_spec.get("initContainers", [])
    elif kind == "Pod":
        containers = spec.get("containers", [])
        init_containers = spec.get("initContainers", [])
    else:
        init_containers = []
 
    context["containers"] = [
        {
            "name": c.get("name"),
            "image": c.get("image"),
            "resources": c.get("resources", {}),
            "securityContext": c.get("securityContext", {}),
            "readinessProbe": c.get("readinessProbe"),
            "livenessProbe": c.get("livenessProbe"),
            "env": [e.get("name") for e in c.get("env", [])]
        }
        for c in containers
    ]
 
    context["initContainers"] = [c.get("name") for c in init_containers]
 
    # Service-specific context
    if kind == "Service":
        context["type"] = spec.get("type", "ClusterIP")
        context["ports"] = spec.get("ports", [])
 
    # RBAC context
    if kind in ["Role", "ClusterRole"]:
        context["rules"] = doc.get("rules", [])
 
    return context

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
 
 
def analyze_kubernetes_manifest(yaml_content: str) -> dict:
    """
    Send Kubernetes YAML to Claude for comprehensive analysis.
    Returns explanation, issues, and improved YAML.
    """
    try:
        documents = load_kubernetes_yaml(yaml_content)
    except yaml.YAMLError as e:
        return {
            "error": f"Invalid YAML: {str(e)}",
            "explanation": None,
            "issues": [],
            "improved_yaml": None
        }
 
    if not documents:
        return {"error": "Empty YAML document", "explanation": None, "issues": [], "improved_yaml": None}
 
    # Extract context for each resource
    contexts = [extract_resource_context(doc) for doc in documents]
 
    yaml_block = "```yaml\n" + yaml_content + "\n```"
    context_json = json.dumps(contexts, indent=2)
 
    prompt = (
        "You are a Kubernetes expert reviewing a manifest. Analyze the following Kubernetes YAML "
        "and provide a comprehensive review.\n\n"
        "## YAML to Review\n\n" + yaml_block + "\n\n"
        "## Extracted Context (for your reference)\n" + context_json + "\n\n---\n\n"
        "Provide your analysis in the following exact JSON structure:\n\n"
        '{"explanation": "Plain English explanation...", '
        '"security_issues": [{"severity": "critical|high|medium|low", "field": "path.to.field", "issue": "...", "fix": "..."}], '
        '"best_practice_violations": [{"severity": "high|medium|low", "field": "...", "issue": "...", "fix": "..."}], '
        '"positive_observations": ["..."], '
        '"summary": {"overall_score": 0-100, "verdict": "Production-ready|Needs fixes|Not production-safe", "top_priority": "..."}, '
        '"improved_yaml": "Complete corrected YAML."}\n\n'
        "Security: check runAsRoot, privileged, hostNetwork, missing seccompProfile, wildcard RBAC, hardcoded secrets, NodePort.\n"
        "Best practices: missing resource limits, no health probes, :latest tags, missing PDB, single replica.\n"
        "Return ONLY valid JSON, no markdown code blocks around the JSON."
    )
 
    message = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=4000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    response_text = message.content[0].text.strip()
 
    # Parse JSON response
    try:
        # Sometimes Claude wraps in code block despite instructions
        if response_text.startswith("```"):
            response_text = response_text.split("```")[1]
            if response_text.startswith("json"):
                response_text = response_text[4:]
 
        result = json.loads(response_text)
        return result
    except json.JSONDecodeError:
        return {
            "explanation": response_text,
            "security_issues": [],
            "best_practice_violations": [],
            "positive_observations": [],
            "summary": {"overall_score": 0, "verdict": "Parse error", "top_priority": "N/A"},
            "improved_yaml": None
        }

import sys
 
 
def format_analysis_report(analysis: dict, yaml_file: str) -> str:
    """Format the analysis as a readable terminal report."""
    lines = []
    lines.append(f"\n{'='*60}")
    lines.append(f"  KUBERNETES YAML ANALYSIS: {yaml_file}")
    lines.append(f"{'='*60}\n")
 
    if "error" in analysis:
        lines.append(f"ERROR: {analysis['error']}")
        return "\n".join(lines)
 
    # Overall score
    summary = analysis.get("summary", {})
    score = summary.get("overall_score", 0)
    verdict = summary.get("verdict", "Unknown")
    score_emoji = "✅" if score >= 80 else "⚠️" if score >= 60 else "❌"
 
    lines.append(f"{score_emoji}  Overall Score: {score}/100 — {verdict}")
    lines.append("")
 
    # Explanation
    if analysis.get("explanation"):
        lines.append("📖  WHAT THIS DOES")
        lines.append(f"   {analysis['explanation']}")
        lines.append("")
 
    # Security issues
    security = analysis.get("security_issues", [])
    if security:
        lines.append("🔒  SECURITY ISSUES")
        for issue in security:
            sev = issue.get("severity", "").upper()
            icon = "🔴" if sev == "CRITICAL" else "🟠" if sev == "HIGH" else "🟡"
            lines.append(f"   {icon} [{sev}] {issue.get('field', '')}")
            lines.append(f"      Issue: {issue.get('issue', '')}")
            lines.append(f"      Fix:   {issue.get('fix', '')}")
        lines.append("")
 
    # Best practice violations
    violations = analysis.get("best_practice_violations", [])
    if violations:
        lines.append("📋  BEST PRACTICE VIOLATIONS")
        for v in violations:
            sev = v.get("severity", "").upper()
            icon = "🟠" if sev == "HIGH" else "🟡" if sev == "MEDIUM" else "🔵"
            lines.append(f"   {icon} [{sev}] {v.get('field', '')}")
            lines.append(f"      Issue: {v.get('issue', '')}")
            lines.append(f"      Fix:   {v.get('fix', '')}")
        lines.append("")
 
    # Positive observations
    positive = analysis.get("positive_observations", [])
    if positive:
        lines.append("✅  WHAT'S DONE WELL")
        for obs in positive:
            lines.append(f"   • {obs}")
        lines.append("")
 
    # Top priority
    top_priority = summary.get("top_priority")
    if top_priority:
        lines.append(f"🎯  TOP PRIORITY: {top_priority}")
        lines.append("")
 
    return "\n".join(lines)
 
 
def cli():
    if len(sys.argv) < 2:
        print("Usage: python k8s_analyzer.py <path-to-yaml-file> [--save-fixed]")
        sys.exit(1)
 
    yaml_file = sys.argv[1]
    save_fixed = "--save-fixed" in sys.argv
 
    with open(yaml_file, "r") as f:
        yaml_content = f.read()
 
    print(f"Analyzing {yaml_file}...")
    analysis = analyze_kubernetes_manifest(yaml_content)
 
    print(format_analysis_report(analysis, yaml_file))
 
    if save_fixed and analysis.get("improved_yaml"):
        fixed_path = yaml_file.replace(".yaml", "-fixed.yaml").replace(".yml", "-fixed.yml")
        with open(fixed_path, "w") as f:
            f.write(analysis["improved_yaml"])
        print(f"\n💾  Improved YAML saved to: {fixed_path}")

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import JSONResponse
 
app = FastAPI(title="K8s YAML Analyzer", version="1.0.0")
 
 
@app.post("/analyze")
async def analyze_yaml(
    file: UploadFile = File(None),
    yaml_content: str = Form(None)
):
    """Accept either a file upload or raw YAML string."""
    if file:
        content = (await file.read()).decode("utf-8")
    elif yaml_content:
        content = yaml_content
    else:
        return JSONResponse({"error": "Provide either a file or yaml_content"}, status_code=400)
 
    result = analyze_kubernetes_manifest(content)
    return result
 
 
@app.get("/health")
def health():
    return {"status": "ok"}
 
 
if __name__ == "__main__":
    import uvicorn
    cli()  # Run CLI mode if yaml file passed, else run API

# .github/workflows/k8s-lint.yml
- name: Analyze Kubernetes manifests
  run: |
    pip install anthropic pyyaml
    for f in k8s/**/*.yaml; do
      python k8s_analyzer.py "$f"
    done
  env:
    ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}