Build an AI Kubernetes Admission Controller with Claude API and OPA

pip install fastapi uvicorn anthropic kubernetes

import base64
import json
import anthropic
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
 
app = FastAPI()
client = anthropic.Anthropic()
 
REVIEW_PROMPT = """You are a Kubernetes security reviewer. Review this manifest
being submitted to the cluster and decide if it should be ALLOWED or DENIED.
 
Deny if you see combinations of risk, not just single flags:
- image tag "latest" combined with no resource limits
- hostNetwork or hostPID set to true without an obvious system-level reason
- privileged: true or allowPrivilegeEscalation: true
- containers running as root (no runAsNonRoot) in a namespace that isn't kube-system
- secrets mounted as environment variables instead of volumes (softer flag)
 
Manifest:
{manifest}
 
Respond with ONLY valid JSON: {{"allowed": true/false, "reason": "one sentence, specific"}}"""
 
 
def review_with_claude(manifest: dict) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=300,
        messages=[{
            "role": "user",
            "content": REVIEW_PROMPT.format(manifest=json.dumps(manifest, indent=2))
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)
 
 
@app.post("/validate")
async def validate(request: Request):
    body = await request.json()
    admission_request = body["request"]
    uid = admission_request["uid"]
    manifest = admission_request["object"]
 
    # Skip system namespaces — Claude should not gate kube-system
    namespace = manifest.get("metadata", {}).get("namespace", "default")
    if namespace in ("kube-system", "kube-public"):
        return admission_response(uid, allowed=True)
 
    verdict = review_with_claude(manifest)
 
    return admission_response(uid, allowed=verdict["allowed"], reason=verdict["reason"])
 
 
def admission_response(uid: str, allowed: bool, reason: str = "") -> JSONResponse:
    return JSONResponse({
        "apiVersion": "admission.k8s.io/v1",
        "kind": "AdmissionReview",
        "response": {
            "uid": uid,
            "allowed": allowed,
            "status": {"message": reason} if reason else {},
        },
    })

apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
  name: claude-manifest-reviewer
webhooks:
- name: review.claude.devopsboys.com
  clientConfig:
    service:
      name: claude-reviewer
      namespace: security
      path: "/validate"
    caBundle: <base64-ca-cert>
  rules:
  - apiGroups: ["apps", ""]
    apiVersions: ["v1"]
    operations: ["CREATE", "UPDATE"]
    resources: ["deployments", "pods"]
  failurePolicy: Ignore    # Never block the cluster if the webhook is down
  timeoutSeconds: 5
  admissionReviewVersions: ["v1"]