Build a Natural Language kubectl — Ask Questions to Your Cluster

$ k8s-ai "show me all pods in production namespace that are not running"
kubectl get pods -n production --field-selector=status.phase!=Running
 
$ k8s-ai "what's using the most memory in the default namespace"
kubectl top pods -n default --sort-by=memory
 
$ k8s-ai "delete all completed jobs older than 1 hour"
# kubectl delete jobs -n default $(kubectl get jobs -n default \
#   -o jsonpath='{.items[?(@.status.completionTime)].metadata.name}')
# ⚠️  Review before running — this deletes resources

pip install anthropic click rich python-dotenv
 
# .env
ANTHROPIC_API_KEY=sk-ant-...

# k8s_ai.py
import anthropic
import subprocess
import os
from dotenv import load_dotenv
 
load_dotenv()
client = anthropic.Anthropic()
 
SYSTEM_PROMPT = """You are a Kubernetes expert. Convert natural language requests to kubectl commands.
 
Rules:
1. Return ONLY the kubectl command, nothing else
2. If the request could delete/modify resources, prepend a comment: "# ⚠️ Review before running"
3. Use common flags: -n for namespace, -o for output format, --sort-by for sorting
4. For complex operations needing multiple commands, chain with && or use $(...)
5. Default namespace is "default" unless specified
6. For "recent" or "last hour" use --since=1h flag on logs, or --field-selector for events
7. Never use --force or --grace-period=0 unless explicitly asked
8. For listing resources by status, use --field-selector=status.phase=Running etc.
 
If the request is impossible with kubectl, respond: "# Not possible with kubectl alone: [brief explanation]"
"""
 
def natural_to_kubectl(query: str) -> str:
    """Convert natural language to kubectl command."""
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=300,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": query}]
    )
    return response.content[0].text.strip()

# cli.py
import click
from rich.console import Console
from rich.syntax import Syntax
from rich.prompt import Confirm
from k8s_ai import natural_to_kubectl
import subprocess
 
console = Console()
 
DESTRUCTIVE_KEYWORDS = ["delete", "remove", "kill", "drain", "cordon", "patch", "edit", "apply"]
 
def is_destructive(command: str) -> bool:
    cmd_lower = command.lower()
    return any(kw in cmd_lower for kw in DESTRUCTIVE_KEYWORDS)
 
@click.command()
@click.argument("query", nargs=-1)
@click.option("--execute", "-e", is_flag=True, help="Execute the command after generating")
@click.option("--dry-run", "-d", is_flag=True, help="Add --dry-run=client to the command")
def main(query, execute, dry_run):
    """Generate kubectl commands from natural language."""
    
    if not query:
        console.print("[red]Please provide a query[/red]")
        return
    
    query_str = " ".join(query)
    
    console.print(f"\n[dim]Generating kubectl command for:[/dim] {query_str}\n")
    
    command = natural_to_kubectl(query_str)
    
    # Add dry-run if requested
    if dry_run and "delete" in command:
        command = command.replace("kubectl delete", "kubectl delete --dry-run=client")
    
    # Display command with syntax highlighting
    syntax = Syntax(command, "bash", theme="monokai", line_numbers=False)
    console.print(syntax)
    
    if command.startswith("# Not possible"):
        return
    
    # Handle execution
    if execute:
        if is_destructive(command) or "⚠️" in command:
            console.print("\n[yellow]⚠️  This command modifies cluster resources[/yellow]")
            if not Confirm.ask("Execute anyway?"):
                console.print("[dim]Cancelled[/dim]")
                return
        
        console.print("\n[dim]Running...[/dim]\n")
        
        # Strip comments before executing
        clean_cmd = "\n".join(
            line for line in command.split("\n") 
            if not line.strip().startswith("#")
        )
        
        result = subprocess.run(clean_cmd, shell=True, text=True, capture_output=True)
        
        if result.stdout:
            console.print(result.stdout)
        if result.stderr:
            console.print(f"[red]{result.stderr}[/red]")
    else:
        console.print("\n[dim]Add -e to execute | -d for dry-run[/dim]")
 
if __name__ == "__main__":
    main()

# cluster_context.py
from kubernetes import client, config
import json
 
def get_cluster_summary() -> str:
    """Get current cluster state to give AI context."""
    try:
        config.load_kube_config()
        v1 = client.CoreV1Api()
        
        namespaces = [ns.metadata.name for ns in v1.list_namespace().items]
        
        context = {
            "current_context": config.list_kube_config_contexts()[1]["name"],
            "namespaces": namespaces[:10],  # First 10 namespaces
        }
        
        return json.dumps(context)
    except:
        return "{}"
 
# Update system prompt with context
def natural_to_kubectl_with_context(query: str) -> str:
    context = get_cluster_summary()
    
    enhanced_system = SYSTEM_PROMPT + f"\n\nCurrent cluster context:\n{context}"
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=300,
        system=enhanced_system,
        messages=[{"role": "user", "content": query}]
    )
    return response.content[0].text.strip()

# Setup
pip install -e .
 
# Or just alias it
echo 'alias k8s-ai="python /path/to/cli.py"' >> ~/.bashrc
source ~/.bashrc

# pyproject.toml
[project.scripts]
k8s-ai = "cli:main"

# Listing and filtering
k8s-ai "list all deployments with less than 2 ready replicas"
k8s-ai "show me pods older than 7 days in all namespaces"
k8s-ai "find all pods without resource limits set"
 
# Logs and debugging
k8s-ai "show logs for the api-server pod in production from last 30 minutes"
k8s-ai "get events sorted by time in production namespace"
 
# Resource usage
k8s-ai "which nodes are using more than 80% memory"
k8s-ai "show me top 5 CPU-consuming pods across all namespaces"
 
# With execution
k8s-ai "restart all pods in the staging namespace" -e
k8s-ai "scale the api deployment to 5 replicas" -e -d  # dry run first