Build an AI Infrastructure as Code Generator

pip install anthropic pyyaml

mkdir iac-gen && cd iac-gen
touch iac_gen.py

import anthropic
import json, yaml, sys

client = anthropic.Anthropic(
    api_key="sk-your-key",
    base_url="https://ezaiapi.com"
)

SYSTEM_PROMPT = """You are an infrastructure-as-code expert.
Given a plain English description, generate production-ready IaC.

RULES:
- Output ONLY the raw config (no markdown fences, no explanations)
- Use security best practices: non-root users, resource limits,
  encrypted connections, parameterized secrets via env vars
- Include health checks for all services
- Add meaningful comments explaining non-obvious choices
- For Terraform: include provider block, variables.tf content
  as comments, and use data sources where appropriate
- For Docker Compose: use version 3.8+, named networks/volumes
- For Kubernetes: include resource requests/limits, readiness
  and liveness probes, and anti-affinity rules for HA

FORMAT: {format}
CLOUD PROVIDER (if Terraform): {provider}
"""

def generate_iac(description, fmt="docker-compose", provider="aws"):
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        system=SYSTEM_PROMPT.format(
            format=fmt, provider=provider
        ),
        messages=[{
            "role": "user",
            "content": description
        }]
    )
    return response.content[0].text

def validate_output(content, fmt):
    """Verify the generated config is valid syntax."""
    if fmt in ("docker-compose", "kubernetes"):
        docs = list(yaml.safe_load_all(content))
        return len(docs) > 0
    elif fmt == "terraform":
        # HCL isn't JSON, but check for balanced braces
        return content.count("{") == content.count("}")
    return True

# CLI entry point
if __name__ == "__main__":
    fmt = sys.argv[1] if len(sys.argv) > 1 else "docker-compose"
    desc = input("Describe your infrastructure: ")
    result = generate_iac(desc, fmt)

    if validate_output(result, fmt):
        ext = {"terraform": ".tf", "docker-compose": ".yml", "kubernetes": ".yaml"}
        filename = f"output{ext[fmt]}"
        with open(filename, "w") as f:
            f.write(result)
        print(f"✅ Written to {filename}")
    else:
        print("⚠️ Validation failed — retrying...")
        result = generate_iac(desc, fmt)
        print(result)

python iac_gen.py docker-compose
# Input: "FastAPI app with PostgreSQL 16, Redis 7 for caching,
#         Nginx reverse proxy with SSL termination, and
#         pgAdmin for database management"

def generate_streaming(description, fmt, provider="aws"):
    """Stream IaC output for real-time feedback."""
    chunks = []

    with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        system=SYSTEM_PROMPT.format(
            format=fmt, provider=provider
        ),
        messages=[{
            "role": "user",
            "content": description
        }]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            chunks.append(text)

    full_output = "".join(chunks)
    return full_output

def detect_format(description):
    """Auto-detect output format from description."""
    desc_lower = description.lower()
    if any(w in desc_lower for w in
           ["terraform", "aws", "gcp", "azure", "ec2", "s3"]):
        return "terraform"
    if any(w in desc_lower for w in
           ["k8s", "kubernetes", "pod", "deployment", "cluster"]):
        return "kubernetes"
    return "docker-compose"

def refine_iac(messages, refinement, fmt, provider="aws"):
    """Refine existing config with follow-up instructions."""
    messages.append({
        "role": "user",
        "content": f"Update the config: {refinement}. "
                   "Output the COMPLETE updated config."
    })

    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        system=SYSTEM_PROMPT.format(
            format=fmt, provider=provider
        ),
        messages=messages
    )

    result = response.content[0].text
    messages.append({
        "role": "assistant",
        "content": result
    })
    return result, messages

# Usage: interactive refinement session
messages = []
desc = "3-node Redis cluster with Sentinel for HA"
result = generate_iac(desc, "docker-compose")
messages = [
    {"role": "user", "content": desc},
    {"role": "assistant", "content": result}
]

# Refine: add monitoring
result, messages = refine_iac(
    messages,
    "Add Prometheus with redis-exporter sidecar and Grafana "
    "dashboard preconfigured with Redis metrics",
    "docker-compose"
)

import subprocess, tempfile, os

def validate_terraform(tf_content):
    """Run terraform validate on generated config."""
    with tempfile.TemporaryDirectory() as tmpdir:
        tf_path = os.path.join(tmpdir, "main.tf")
        with open(tf_path, "w") as f:
            f.write(tf_content)

        # Initialize terraform (downloads providers)
        subprocess.run(
            ["terraform", "init", "-backend=false"],
            cwd=tmpdir, capture_output=True
        )

        # Validate syntax and references
        result = subprocess.run(
            ["terraform", "validate", "-json"],
            cwd=tmpdir, capture_output=True, text=True
        )

        validation = json.loads(result.stdout)
        if validation["valid"]:
            return True, "✅ Config is valid"
        else:
            errors = [d["detail"] for d in
                      validation.get("diagnostics", [])]
            return False, errors