Build an AI Kubernetes Cluster Migration Assistant with Claude API

pip install anthropic kubernetes pyyaml

import anthropic
from kubernetes import client, config
import yaml
 
client_ai = anthropic.Anthropic()
 
 
def inspect_cluster_capabilities(context_name: str) -> dict:
    """Gather what the target cluster actually provides — storage classes,
    ingress controller, available CRDs — everything a migrated manifest
    might implicitly depend on."""
    config.load_kube_config(context=context_name)
    v1 = client.CoreV1Api()
    storage_v1 = client.StorageV1Api()
    apiextensions = client.ApiextensionsV1Api()
 
    return {
        "storage_classes": [sc.metadata.name for sc in storage_v1.list_storage_class().items],
        "default_storage_class": next(
            (sc.metadata.name for sc in storage_v1.list_storage_class().items
             if sc.metadata.annotations and sc.metadata.annotations.get("storageclass.kubernetes.io/is-default-class") == "true"),
            None
        ),
        "ingress_class": get_ingress_controller(v1),
        "installed_crds": [crd.metadata.name for crd in apiextensions.list_custom_resource_definition().items],
        "node_labels": get_common_node_labels(v1),
    }

ANALYZE_PROMPT = """Analyze whether these Kubernetes manifests will work
unmodified on the target cluster, or need translation.
 
Manifests:
{manifests}
 
Source cluster capabilities:
{source_capabilities}
 
Target cluster capabilities:
{target_capabilities}
 
For each manifest, check specifically for:
1. StorageClass references that don't exist on target (common when
   moving between cloud providers — gp3 vs pd-ssd vs premium-lrs)
2. Ingress class/annotations that are provider-specific (alb vs nginx vs gce)
3. CRDs the manifest depends on (operators, custom resources) that
   aren't installed on the target cluster
4. NodeSelector/affinity rules referencing labels that won't exist
   on the target's node pools
 
Respond with ONLY valid JSON:
{{"issues": [
  {{"file": "...", "issue": "...", "fix": "...", "blocking": true/false}}
], "safe_to_migrate_unmodified": [...]}}"""
 
 
def analyze_compatibility(manifests: list[dict], source_caps: dict, target_caps: dict) -> dict:
    response = client_ai.messages.create(
        model="claude-sonnet-5",
        max_tokens=3000,
        messages=[{
            "role": "user",
            "content": ANALYZE_PROMPT.format(
                manifests=manifests, source_capabilities=source_caps, target_capabilities=target_caps
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    import json
    return json.loads(text)

SEQUENCE_PROMPT = """Generate a safe migration sequence for moving these
workloads from the source to target cluster.
 
Workloads and their dependencies (what depends on what):
{workload_graph}
 
Stateful resources requiring data migration (databases, PVCs with data):
{stateful_resources}
 
Generate an ordered migration plan considering:
1. Stateless services can usually migrate in any order, or blue-green cut over
2. Stateful services need explicit data migration steps BEFORE traffic cutover
3. Services with tight coupling (shared database, synchronous calls) should
   migrate together, not have one half on each cluster during the transition
4. What can run on BOTH clusters simultaneously during migration (dual-write,
   traffic splitting) vs what requires a hard cutover
 
Format as an ordered runbook."""
 
 
def generate_migration_sequence(workload_graph: dict, stateful_resources: list) -> str:
    response = client_ai.messages.create(
        model="claude-sonnet-5",
        max_tokens=2500,
        messages=[{
            "role": "user",
            "content": SEQUENCE_PROMPT.format(workload_graph=workload_graph, stateful_resources=stateful_resources)
        }]
    )
    return response.content[0].text