Build an AI Code Reviewer for Kubernetes Manifests with Claude

# checkers.py — fast rules before hitting Claude
import yaml
from dataclasses import dataclass
from typing import List
 
 
@dataclass
class Issue:
    severity: str  # critical, warning, info
    resource: str
    message: str
    fix: str
 
 
def check_manifest(doc: dict) -> List[Issue]:
    issues = []
    kind = doc.get("kind", "")
    name = doc.get("metadata", {}).get("name", "unknown")
    resource_id = f"{kind}/{name}"
 
    if kind in ("Deployment", "DaemonSet", "StatefulSet"):
        containers = (
            doc.get("spec", {})
            .get("template", {})
            .get("spec", {})
            .get("containers", [])
        )
 
        for c in containers:
            cname = c.get("name", "unnamed")
 
            # Missing resource limits
            limits = c.get("resources", {}).get("limits", {})
            requests = c.get("resources", {}).get("requests", {})
            if not limits.get("cpu"):
                issues.append(Issue("critical", f"{resource_id}/{cname}",
                    "Missing CPU limit — pod can consume all node CPU",
                    "Add resources.limits.cpu (e.g., '500m')"))
            if not limits.get("memory"):
                issues.append(Issue("critical", f"{resource_id}/{cname}",
                    "Missing memory limit — pod will be OOMKilled unpredictably",
                    "Add resources.limits.memory (e.g., '512Mi')"))
 
            # Latest tag
            image = c.get("image", "")
            if image.endswith(":latest") or ":" not in image:
                issues.append(Issue("warning", f"{resource_id}/{cname}",
                    f"Image '{image}' uses :latest — non-deterministic deploys",
                    "Pin to a specific digest: image: nginx:1.27.0"))
 
            # Hardcoded secrets in env
            for env in c.get("env", []):
                val = str(env.get("value", "")).lower()
                key = env.get("name", "").lower()
                if any(s in key for s in ("password", "secret", "token", "key", "api_key")):
                    if env.get("value"):  # hardcoded (not valueFrom)
                        issues.append(Issue("critical", f"{resource_id}/{cname}",
                            f"Hardcoded secret in env var '{env['name']}'",
                            "Use secretKeyRef: valueFrom.secretKeyRef.name/key"))
 
            # Missing probes
            if not c.get("readinessProbe"):
                issues.append(Issue("warning", f"{resource_id}/{cname}",
                    "Missing readiness probe — traffic routed before app is ready",
                    "Add readinessProbe with httpGet or exec"))
            if not c.get("livenessProbe"):
                issues.append(Issue("info", f"{resource_id}/{cname}",
                    "Missing liveness probe — hung pods won't be restarted",
                    "Add livenessProbe"))
 
        # Security context
        pod_spec = doc.get("spec", {}).get("template", {}).get("spec", {})
        for c in containers:
            sc = c.get("securityContext", {})
            if sc.get("privileged"):
                issues.append(Issue("critical", f"{resource_id}/{c['name']}",
                    "Container runs privileged — full host access",
                    "Remove securityContext.privileged or set to false"))
            if sc.get("allowPrivilegeEscalation") is not False:
                issues.append(Issue("warning", f"{resource_id}/{c['name']}",
                    "allowPrivilegeEscalation not disabled",
                    "Add: securityContext.allowPrivilegeEscalation: false"))
 
    return issues

# reviewer.py
import anthropic
import yaml
from checkers import check_manifest, Issue
from typing import List
 
client = anthropic.Anthropic()
 
REVIEW_PROMPT = """You are a senior DevOps engineer reviewing Kubernetes manifests.
After the static checks listed, look for:
1. Architectural issues (wrong service type for use case, anti-patterns)
2. Operational risks (missing PDB for critical deployments, no HPA)
3. Security posture (RBAC overpermissioning, network exposure)
4. Performance (wrong resource ratios, missing affinity rules)
 
For each issue found, give: severity (critical/warning/info), what's wrong, and the exact YAML fix.
Be specific and actionable. Focus on issues the static checker didn't catch."""
 
 
def ai_review(manifest_yaml: str, static_issues: List[Issue]) -> str:
    static_summary = "\n".join(
        f"- [{i.severity.upper()}] {i.resource}: {i.message}"
        for i in static_issues
    )
 
    yaml_block = "```yaml\n" + manifest_yaml + "\n```"
    static_note = static_summary if static_summary else "No issues found by static checker."
    user_content = (
        f"Review this Kubernetes manifest:\n\n{yaml_block}\n\n"
        f"Static checker already found:\n{static_note}\n\n"
        "Find additional issues the static checker missed."
    )
 
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        system=REVIEW_PROMPT,
        messages=[{"role": "user", "content": user_content}],
    )
    return response.content[0].text

# main.py
import sys
import yaml
import click
from pathlib import Path
from checkers import check_manifest
from reviewer import ai_review
from rich.console import Console
from rich.table import Table
from rich import print as rprint
 
console = Console()
 
SEVERITY_COLORS = {
    "critical": "red",
    "warning": "yellow",
    "info": "blue",
}
 
 
@click.command()
@click.argument("manifest", type=click.Path(exists=True))
@click.option("--ai/--no-ai", default=True, help="Enable AI review (uses Claude API)")
@click.option("--fail-on", default="critical", type=click.Choice(["critical", "warning", "info"]))
def review(manifest: str, ai: bool, fail_on: str):
    """Review a Kubernetes manifest for issues."""
    content = Path(manifest).read_text()
 
    # Parse all YAML documents (multi-doc files)
    docs = list(yaml.safe_load_all(content))
    all_issues = []
 
    for doc in docs:
        if doc:
            issues = check_manifest(doc)
            all_issues.extend(issues)
 
    # Display static issues
    if all_issues:
        table = Table(title="Static Check Results", show_lines=True)
        table.add_column("Severity", style="bold", width=10)
        table.add_column("Resource", width=30)
        table.add_column("Issue", width=50)
        table.add_column("Fix", width=40)
 
        for issue in all_issues:
            color = SEVERITY_COLORS.get(issue.severity, "white")
            table.add_row(
                f"[{color}]{issue.severity.upper()}[/{color}]",
                issue.resource,
                issue.message,
                issue.fix,
            )
        console.print(table)
    else:
        console.print("[green]✓ No static check issues found[/green]")
 
    # AI review
    if ai:
        console.print("\n[cyan]Running AI review...[/cyan]")
        ai_feedback = ai_review(content, all_issues)
        console.print("\n[bold]AI Review:[/bold]")
        console.print(ai_feedback)
 
    # Exit code
    severity_order = ["info", "warning", "critical"]
    fail_level = severity_order.index(fail_on)
    highest = max(
        (severity_order.index(i.severity) for i in all_issues),
        default=-1
    )
    if highest >= fail_level:
        console.print(f"\n[red]❌ Review failed ({fail_on}+ issues found)[/red]")
        sys.exit(1)
    else:
        console.print(f"\n[green]✅ Review passed[/green]")
 
 
if __name__ == "__main__":
    review()

# Install
pip install anthropic pyyaml click rich
 
# Review a manifest
python main.py deployment.yaml
 
# Fail CI on critical issues only (default)
python main.py deployment.yaml --fail-on critical
 
# Skip AI review (just static checks)
python main.py deployment.yaml --no-ai
 
# Review whole directory
for f in k8s/*.yaml; do python main.py "$f"; done

# .github/workflows/k8s-review.yaml
name: Kubernetes Manifest Review
on:
  pull_request:
    paths:
    - 'k8s/**/*.yaml'
    - 'helm/**'
 
jobs:
  review:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v4
    - uses: actions/setup-python@v4
      with:
        python-version: '3.11'
    - run: pip install anthropic pyyaml click rich
    - name: Review changed manifests
      env:
        ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
      run: |
        # Get changed YAML files
        CHANGED=$(git diff --name-only origin/main...HEAD | grep '\.yaml$' || true)
        for f in $CHANGED; do
          echo "Reviewing $f..."
          python k8s-review/main.py "$f" --fail-on critical
        done