AI agents for schema migration review and validation

Manual schema reviews don’t scale when your team ships database changes multiple times per day. An AI agent schema migration review process can parse diffs, enforce compatibility rules, and block destructive operations before they hit production. This guide lays out an ordered path to stand up that agent as a reliable CI gate rather than a toy demo.

Define the review contract

Before writing any agent code, write down the policies you expect the agent to enforce. The agent is a policy executor, not a free-form reviewer. Encode the rules as data so they can be version-controlled and unit-tested.

Split rules into two buckets: breaking changes that must be blocked, and expanding changes that require a companion step.

{
  "block": [
    "DROP COLUMN",
    "DROP TABLE",
    "ALTER COLUMN TYPE without USING clause",
    "ADD PRIMARY KEY to existing table"
  ],
  "require": [
    "backfill script for NOT NULL additions",
    "CREATE INDEX CONCURRENTLY on tables > 10M rows",
    "deprecated column marker in API schema"
  ]
}

Any migration that triggers a block rule fails the build. require rules generate warnings that a human must acknowledge with a PR comment. Keep the contract in schema-policy.json at the repo root so the agent and your tests read the same source of truth.

Capture the migration diff

The agent needs the exact DDL being applied and the current schema state. Don’t rely on the agent to introspect production; pass a baseline.

Generate the SQL from your migration tool, not from hand-written files:

# Alembic example
alembic upgrade head --sql > migration.sql
# Capture baseline from the target branch
git show origin/main:db/schema.sql > baseline.sql

Load both into the agent context with a strict delimiter so the model can’t confuse them:

def load_context(baseline_path: str, migration_path: str) -> str:
    with open(baseline_path) as f:
        baseline = f.read()
    with open(migration_path) as f:
        diff = f.read()
    return (
        "<<BASELINE_SCHEMA>>\n"
        f"{baseline}\n"
        "<<MIGRATION_DIFF>>\n"
        f"{diff}\n"
        "<<END>>"
    )

Use the baseline from the branch you’re merging into, not local dev. A stale baseline produces false alarms about columns that already exist.

Build the agent’s toolset

The agent should not guess table sizes or FK relationships. Give it read-only tools backed by your metadata store. Define them as JSON function specs:

[
  {
    "type": "function",
    "function": {
      "name": "get_table_rows",
      "description": "Estimate row count for a table",
      "parameters": {
        "type": "object",
        "properties": {"table": {"type": "string"}},
        "required": ["table"]
      }
    }
  },
  {
    "type": "function",
    "function": {
      "name": "get_foreign_keys",
      "description": "List FKs referencing the given column",
      "parameters": {
        "type": "object",
        "properties": {"table": {"type": "string"}, "column": {"type": "string"}},
        "required": ["table", "column"]
      }
    }
  }
]

Implement a dispatcher in Python that hits information_schema:

import psycopg2

def dispatch_tool(name: str, args: dict):
    conn = psycopg2.connect("dbname=meta user=readonly")
    cur = conn.cursor()
    if name == "get_table_rows":
        cur.execute("SELECT reltuples::bigint FROM pg_class WHERE relname=%s", (args["table"],))
        return {"rows": cur.fetchone()[0]}
    if name == "get_foreign_keys":
        cur.execute("""
            SELECT conname FROM pg_constraint
            WHERE contype='f' AND conrelid=%s::regclass
        """, (args["table"],))
        return {"fks": [r[0] for r in cur.fetchall()]}

Run the reviewer with a DB role that has SELECT only. A bug in this wrapper is the fastest way to let an agent truncate a table.

Prompt the agent with context

Use a strict system prompt that forces structured output. Set temperature to 0. Point the client at a single OpenAI-compatible endpoint that addresses 240+ models; n4n.ai does this and adds automatic fallback when a provider is degraded, keeping your CI pipeline unblocked.

from openai import OpenAI

client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="KEY")

resp = client.chat.completions.create(
    model="anthropic/claude-3.5-sonnet",
    temperature=0,
    messages=[
        {"role": "system", "content": (
            "You are a schema migration reviewer. Use the provided tools. "
            "Output ONLY JSON: {\"status\": \"pass\"|\"warn\"|\"block\", \"reasons\": [string]}. "
            "Never generate DDL."
        )},
        {"role": "user", "content": load_context("baseline.sql", "migration.sql")}
    ],
    tools=TOOL_SPEC,
    tool_choice="auto"
)

In an AI agent schema migration review flow, the model only inspects and judges. If it tries to emit CREATE TABLE, your parser should reject the run.

Validate the agent’s verdict

Parse the response and enforce it in code. Never trust the natural-language summary.

import json

def enforce(agent_output: str, policy: dict) -> str:
    try:
        verdict = json.loads(agent_output)
    except json.JSONDecodeError:
        raise RuntimeError("Agent did not return JSON")
    if verdict["status"] == "block":
        for reason in verdict["reasons"]:
            if not any(rule.lower() in reason.lower() for rule in policy["block"]):
                raise RuntimeError(f"Agent blocked for non-policy reason: {reason}")
        return "fail"
    if verdict["status"] == "warn":
        # require human ack handled outside
        return "warn"
    return "pass"

If the agent blocks on something outside the contract, fail closed but alert a human. The agent is a lint step, not a court of final appeal.

Integrate with CI as a merge gate

Add a job that runs after migrations are generated but before they are merged.

# .github/workflows/schema-review.yml
jobs:
  agent-review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install openai psycopg2
      - run: python review_agent.py --baseline baseline.sql --migration migration.sql
      - name: Fail on block
        if: ${{ failure() }}
        run: exit 1

Set the job to required in branch protection. Now an AI agent schema migration review runs on every PR that touches migrations/. Cache verifications per diff hash to skip re-running identical reviews.

Common pitfalls and tradeoffs

Context truncation

Large baselines blow the context window. Trim to only tables touched by the diff. A 5k-line baseline is unnecessary if the migration alters two tables. Compute the table set from the diff parser first.

False positives on renaming

Renames look like drop+add. Teach the agent your migration tool’s rename syntax (ALTER TABLE ... RENAME COLUMN), or it will block valid zero-downtime changes. Add a unit test with a rename diff to catch regressions.

Latency vs. thoroughness

A bigger model catches more subtle issues but adds minutes to CI. Use a smaller model for the first pass and escalate to a larger one only when the small model returns warn. This keeps median latency under 20s.

Don’t grant write access

The agent must be read-only. A bug in your tool wrapper could let it execute DDL. Run the reviewer with a DB role that has SELECT only and revoke all DDL privileges.

Human override is mandatory

For any block on a declared destructive operation, require a DBA to comment /approve-destructive in the PR. The agent cannot self-override. Store the override in the PR thread and have the CI job check for it before failing.

Metric collection

Log every verdict with the diff hash. Over a month you’ll see which rules fire most. If “ADD PRIMARY KEY” blocks ten PRs but none were actual mistakes, your contract is too strict.

Treat the AI agent schema migration review as a scalable first pass. It removes 80% of the toil: missing backfills, forgotten indexes, unguarded drops. The remaining 20% still needs a person who understands production traffic patterns and application semantics.