Autonomous Database Migration Planning: What AI Agents Can (and Can't) Do Safely in 2026

import anthropic
import json
 
client = anthropic.Anthropic()
 
CLASSIFY_PROMPT = """Analyze this database migration for risk.
 
Migration SQL:
{migration_sql}
 
Table stats:
{table_stats}
 
Current database engine and version: {db_engine}
 
Classify the risk of running this migration on a live production table:
1. Does this DDL take a lock that blocks reads/writes, and for how long
   is that likely given the table size? (e.g. ADD COLUMN with a default
   on a large table locks differently across Postgres/MySQL versions)
2. Is this reversible with a clean rollback, or does it lose data if reverted?
3. Does application code need a deploy BEFORE or AFTER this migration to
   avoid a window where old code hits the new schema incorrectly?
 
Respond with ONLY valid JSON:
{{"risk_level": "safe_online" | "needs_maintenance_window" | "needs_manual_review",
  "lock_behavior": "...", "reversible": true/false,
  "deploy_ordering": "app_before_migration" | "app_after_migration" | "either_order_safe",
  "reasoning": "..."}}"""
 
 
def classify_migration_risk(migration_sql: str, table_stats: dict, db_engine: str) -> dict:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=800,
        messages=[{
            "role": "user",
            "content": CLASSIFY_PROMPT.format(
                migration_sql=migration_sql, table_stats=table_stats, db_engine=db_engine
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)

PLAN_PROMPT = """Generate a safe execution plan for this database migration.
 
Migration: {migration_sql}
Risk classification: {risk_classification}
 
Provide:
1. Pre-migration checks to run (verify replication lag is low, confirm
   no long-running transactions holding locks, etc.)
2. The exact execution command/sequence, including any needed session
   settings (lock_timeout, statement_timeout) to fail fast instead of
   blocking production traffic indefinitely
3. A rollback script — even for "irreversible" changes, provide the closest
   safe mitigation (e.g. can't un-drop a column, but CAN restore from
   pre-migration backup within X minutes)
4. Monitoring to watch during and after (replication lag, error rate,
   query latency on the affected table)
 
Format as a runbook a human operator follows step by step."""
 
 
def generate_execution_plan(migration_sql: str, risk_classification: dict) -> str:
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{
            "role": "user",
            "content": PLAN_PROMPT.format(migration_sql=migration_sql, risk_classification=risk_classification)
        }]
    )
    return response.content[0].text

-- Migration: ADD COLUMN status VARCHAR(20) NOT NULL DEFAULT 'pending';
-- Table: orders (140M rows)