Build an AI Disaster Recovery Runbook Validator with Claude API

pip install anthropic boto3 pyyaml

import re
import yaml
import anthropic
 
client = anthropic.Anthropic()
 
 
def extract_commands(runbook_text: str) -> list[dict]:
    """Pull every shell/AWS CLI command block out of a markdown runbook."""
    commands = []
    code_blocks = re.findall(r"```(?:bash|sh)?\n(.*?)```", runbook_text, re.DOTALL)
 
    for i, block in enumerate(code_blocks):
        lines = [l.strip() for l in block.split("\n") if l.strip() and not l.strip().startswith("#")]
        for line in lines:
            resource_refs = re.findall(r"(?:--resources?|--db-instance-identifier|--role-name|--cluster-name)[= ]([\w-]+)", line)
            commands.append({"command": line, "block_index": i, "referenced_resources": resource_refs})
 
    return commands

import boto3
 
def verify_resource_exists(resource_id: str, resource_type: str) -> dict:
    """Check whether a resource referenced in the runbook still exists."""
    try:
        if resource_type == "ec2_instance":
            ec2 = boto3.client("ec2")
            ec2.describe_instances(InstanceIds=[resource_id])
            return {"exists": True}
        elif resource_type == "rds_instance":
            rds = boto3.client("rds")
            rds.describe_db_instances(DBInstanceIdentifier=resource_id)
            return {"exists": True}
        elif resource_type == "iam_role":
            iam = boto3.client("iam")
            iam.get_role(RoleName=resource_id)
            return {"exists": True}
    except Exception as e:
        return {"exists": False, "error": str(e)}
    return {"exists": None, "note": "unknown resource type, could not verify"}

ANALYZE_PROMPT = """Review this disaster recovery runbook step and its verification result.
 
Runbook command:
{command}
 
Referenced resources and their current status:
{verification_results}
 
Runbook context (surrounding steps, for understanding intent):
{context}
 
Determine:
1. Would this command succeed if run right now, given the verification results?
2. If it would fail, what is the most likely fix (updated resource ID, updated
   permission, or the step needs to be rewritten entirely)?
3. Is there a newer/better AWS CLI pattern for this operation that the runbook
   should be updated to use?
 
Respond with ONLY valid JSON:
{{"would_succeed": true/false, "issue": "description or null", "suggested_fix": "description or null"}}"""
 
 
def analyze_step(command: str, verification_results: list[dict], context: str) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=400,
        messages=[{
            "role": "user",
            "content": ANALYZE_PROMPT.format(
                command=command,
                verification_results=verification_results,
                context=context,
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    import json
    return json.loads(text)

def validate_runbook(runbook_path: str) -> dict:
    with open(runbook_path) as f:
        runbook_text = f.read()
 
    commands = extract_commands(runbook_text)
    results = {"total_steps": len(commands), "issues": [], "verified_ok": 0}
 
    for cmd in commands:
        verifications = []
        resource_type = infer_resource_type(cmd["command"])
        for resource_id in cmd["referenced_resources"]:
            verifications.append({
                "resource_id": resource_id,
                **verify_resource_exists(resource_id, resource_type)
            })
 
        if not cmd["referenced_resources"]:
            continue    # No verifiable resource reference, skip
 
        if any(v["exists"] is False for v in verifications):
            analysis = analyze_step(cmd["command"], verifications, runbook_text[:500])
            results["issues"].append({
                "command": cmd["command"],
                "problem": analysis["issue"],
                "suggested_fix": analysis["suggested_fix"],
            })
        else:
            results["verified_ok"] += 1
 
    return results
 
 
def infer_resource_type(command: str) -> str:
    if "ec2" in command or "instance" in command:
        return "ec2_instance"
    if "rds" in command or "db-instance" in command:
        return "rds_instance"
    if "iam" in command or "role" in command:
        return "iam_role"
    return "unknown"

python validate_runbook.py runbooks/payments-db-failover.md
 
# Validation Report: payments-db-failover.md
# Total steps: 18
# Verified OK: 15
# Issues found: 3
#
# Step: aws rds promote-read-replica --db-instance-identifier payments-replica-old
# Problem: DB instance "payments-replica-old" no longer exists
# Suggested fix: The current replica identifier is "payments-replica-us-east-2"
# based on the Terraform state — update the runbook reference

# .github/workflows/dr-runbook-validation.yml
name: DR Runbook Validation
on:
  schedule:
    - cron: "0 6 * * 1"    # Weekly, Monday morning
  workflow_dispatch:
 
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: python validate_runbook.py runbooks/*.md --fail-on-issues