Build an AI-Powered Kubernetes YAML Generator Using LLMs

pip install anthropic
export ANTHROPIC_API_KEY=sk-ant-your-key

# k8s_yaml_generator.py
import anthropic
import os
import sys
import yaml
import json
from typing import Optional
 
SYSTEM_PROMPT = """You are a Kubernetes expert. When given a description of what to deploy, 
you generate production-ready Kubernetes YAML manifests.
 
Always follow these rules:
1. Always specify resource requests AND limits (never omit)
2. Always add readiness and liveness probes for long-running services
3. Always use specific image tags (never 'latest' unless user specifies)
4. Always create non-root user in pods via securityContext
5. Always add namespace if mentioned
6. Add PodDisruptionBudget for deployments with multiple replicas
7. Add HorizontalPodAutoscaler when scalability is mentioned
8. Label all resources with app, version, and managed-by labels
9. Output valid YAML only — no prose between YAML blocks
10. After the YAML, add a brief "# Notes:" comment block explaining key decisions
 
Output format:
- Multiple YAML documents separated by ---
- Each resource on its own document
- Comments explaining non-obvious choices"""
 
 
def generate_k8s_yaml(description: str, context: Optional[str] = None) -> str:
    client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
    
    user_message = f"Generate Kubernetes YAML for: {description}"
    if context:
        user_message += f"\n\nAdditional context:\n{context}"
    
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}]
    )
    
    return message.content[0].text
 
 
def validate_yaml(yaml_content: str) -> tuple[bool, str]:
    """Validate the generated YAML is parseable"""
    try:
        # Split by --- to get individual documents
        docs = [doc for doc in yaml_content.split('---') 
                if doc.strip() and not doc.strip().startswith('#')]
        
        valid_docs = 0
        for doc in docs:
            parsed = yaml.safe_load(doc)
            if parsed and isinstance(parsed, dict):
                valid_docs += 1
        
        return True, f"Valid: {valid_docs} Kubernetes resources"
    except yaml.YAMLError as e:
        return False, f"YAML Error: {e}"
 
 
def extract_yaml_blocks(response: str) -> str:
    """Extract YAML from response (handles markdown code blocks)"""
    if '```yaml' in response:
        lines = response.split('\n')
        in_block = False
        yaml_lines = []
        for line in lines:
            if line.startswith('```yaml'):
                in_block = True
            elif line.startswith('```') and in_block:
                in_block = False
                yaml_lines.append('')
            elif in_block:
                yaml_lines.append(line)
        return '\n'.join(yaml_lines)
    return response
 
 
def main():
    if len(sys.argv) < 2:
        # Interactive mode
        print("Kubernetes YAML Generator")
        print("=" * 40)
        print("Describe what you want to deploy:")
        description = input("> ")
        print("\nAny additional constraints? (press Enter to skip):")
        context = input("> ") or None
    else:
        description = ' '.join(sys.argv[1:])
        context = None
    
    print("\n⚙️  Generating Kubernetes YAML...\n")
    response = generate_k8s_yaml(description, context)
    yaml_content = extract_yaml_blocks(response)
    
    # Validate
    is_valid, validation_msg = validate_yaml(yaml_content)
    
    print(response)
    print(f"\n{'✅' if is_valid else '❌'} Validation: {validation_msg}")
    
    # Save to file
    filename = "generated-k8s.yaml"
    with open(filename, 'w') as f:
        f.write(yaml_content)
    print(f"\n💾 Saved to: {filename}")
    print(f"Apply with: kubectl apply -f {filename}")
 
 
if __name__ == "__main__":
    main()

apiVersion: apps/v1
kind: Deployment
metadata:
  name: fastapi-app
  labels:
    app: fastapi-app
    version: "1.0"
    managed-by: k8s-yaml-generator
spec:
  replicas: 3
  selector:
    matchLabels:
      app: fastapi-app
  template:
    metadata:
      labels:
        app: fastapi-app
    spec:
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
      containers:
        - name: fastapi
          image: fastapi-app:1.0.0
          ports:
            - containerPort: 8000
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: postgres-secret
                  key: connection-string
          resources:
            requests:
              cpu: "100m"
              memory: "128Mi"
            limits:
              cpu: "500m"
              memory: "512Mi"
          readinessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 10
            periodSeconds: 5
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 30
            periodSeconds: 10
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: fastapi-app-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: fastapi-app
  minReplicas: 3
  maxReplicas: 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70
---
apiVersion: v1
kind: Service
metadata:
  name: fastapi-app
spec:
  selector:
    app: fastapi-app
  ports:
    - port: 80
      targetPort: 8000
  type: ClusterIP
---
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: fastapi-app-pdb
spec:
  minAvailable: 2
  selector:
    matchLabels:
      app: fastapi-app

# slack_k8s_bot.py
from slack_bolt import App
from k8s_yaml_generator import generate_k8s_yaml, extract_yaml_blocks, validate_yaml
import os
 
app = App(token=os.environ["SLACK_BOT_TOKEN"])
 
@app.message("generate k8s")
def handle_generate(message, say):
    description = message['text'].replace('generate k8s', '').strip()
    
    say(f"⚙️ Generating Kubernetes YAML for: `{description}`")
    
    response = generate_k8s_yaml(description)
    yaml_content = extract_yaml_blocks(response)
    is_valid, validation_msg = validate_yaml(yaml_content)
    
    # Post YAML in code block
    say(f"```yaml\n{yaml_content[:2900]}\n```\n{validation_msg}")
 
if __name__ == "__main__":
    app.start(port=3000)

# FastAPI endpoint for your developer portal
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from k8s_yaml_generator import generate_k8s_yaml, extract_yaml_blocks, validate_yaml
import os
 
app = FastAPI(title="K8s YAML Generator API")
 
class GenerateRequest(BaseModel):
    description: str
    namespace: str = "default"
    environment: str = "production"
 
@app.post("/generate")
async def generate(request: GenerateRequest):
    context = f"Namespace: {request.namespace}, Environment: {request.environment}"
    response = generate_k8s_yaml(request.description, context)
    yaml_content = extract_yaml_blocks(response)
    is_valid, validation_msg = validate_yaml(yaml_content)
    
    if not is_valid:
        raise HTTPException(status_code=422, detail=f"Generated invalid YAML: {validation_msg}")
    
    return {
        "yaml": yaml_content,
        "validation": validation_msg,
        "explanation": response
    }

context = """
Company standards:
- All pods must have team: backend label
- Resource limits required (fail if missing)
- AWS EKS environment with gp3 storage class
- Use aws-secrets-manager for secrets via External Secrets Operator
"""