Database Agents: Natural Language to SQL in Production

“Show me sales by region for the last 6 months” is a simple question with a not-simple SQL query — JOINs, grouping, and date arithmetic that a business user shouldn’t have to write. Text-to-SQL agents promise to bridge that gap, but in production the hard part isn’t writing SQL; it’s writing SQL you’d actually run on your production database.

In this post, you will learn how to build a safe production text-to-SQL agent: ground the model in your real schema, generate candidate queries, validate them before execution, and enforce read-only, cost-limited access to PostgreSQL.

Prerequisites

• PostgreSQL 13+ with an app user that has SELECT-only grants
• A Gemini API key, and Python 3.10+

1. Ground the model in your actual schema

The single biggest accuracy lever: show the model your real schema, not a description of it. Serialize information_schema into the prompt so table and column names match reality:

SELECT table_name,
       string_agg(column_name || ' ' || data_type, ', ') AS columns
FROM information_schema.columns
WHERE table_schema = 'public'
GROUP BY table_name;

import psycopg
from google import genai

client = genai.Client()
conn = psycopg.connect("dbname=analytics user=app_readonly")

schema = conn.execute(
    """SELECT table_name,
              string_agg(column_name || ' ' || data_type, ', ') AS columns
       FROM information_schema.columns
       WHERE table_schema = 'public'
       GROUP BY table_name"""
).fetchall()

SYSTEM = f"""You translate natural language into PostgreSQL queries.
Use ONLY these tables and columns:
{chr(10).join(f'- {t}: {cols}' for t, cols in schema)}

Rules:
- Output a single SQL statement, no commentary, no markdown fence.
- Prefer explicit column lists; never SELECT *.
- Use PostgreSQL-specific functions (date_trunc, extract) for time grouping.
- If the question is ambiguous, ask in a comment (-- ...) instead of guessing.
"""

The connection is a read-only user, so even a wrong query can’t mutate anything.

2. Generate and validate before you execute

Never run the model’s SQL directly. Two gates: a syntax/read-only check and a cost guard. Postgres gives you EXPLAIN to preview cost without executing — that’s your safety net:

def validate_query(conn, sql: str) -> tuple[bool, str]:
    if not sql.strip().lower().startswith(("select", "with")):
        return False, "only SELECT/WITH queries are allowed"
    try:
        plan = conn.execute(f"EXPLAIN {sql}").fetchall()
    except psycopg.errors.Error as e:
        return False, f"invalid SQL: {e}"
    cost_line = " ".join(row[0] for row in plan)
    if "seq scan" in cost_line.lower() and "where" in cost_line.lower():
        return False, "full-table scan detected"
    return True, cost_line

Add an execution cap at the Postgres level so a runaway query can’t stall production:

-- cancel any single query over 30 seconds
SET statement_timeout = 30000;

3. Build the agent loop

With generation and validation in place, the agent loop is simple and safe: generate → validate → (if invalid) feed the error back and regenerate → execute only validated SQL → summarize the result:

def ask_database(question: str) -> str:
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user", "content": question}]

    for attempt in range(3):
        reply = client.models.generate_content(model="gemini-2.5-pro", contents=messages)
        sql = strip_fences(reply.text)
        ok, msg = validate_query(conn, sql)
        if ok:
            rows = conn.execute(sql).fetchmany(50)
            return summarize(question, sql, rows)   # LLM turns rows into an answer
        messages.append({"role": "assistant", "content": sql})
        messages.append({"role": "user",
                         "content": f"That SQL failed: {msg}. Fix it."})
    return "I couldn't form a safe query. Please rephrase."

The feedback loop is where text-to-SQL stops being magic and starts being reliable: most failures are small syntax issues the model can fix when shown the actual error.

4. Guardrails for production

Beyond validation, the production checklist is short and non-negotiable:

• Permissions: the DB user has GRANT SELECT on specific tables, and no write privileges at all.
• Timeouts: statement_timeout and a row limit (FETCH FIRST 50 ROWS ONLY) on every query.
• Audit log: log question, generated SQL, validation outcome, and result summary to a separate table — you’ll need it when someone asks “how did the agent answer that?”
• Schema hygiene: drop unused tables from the prompt; a cluttered schema produces worse SQL.

Putting It All Together

The complete agent — schema serialization, the validate-then-execute gate, the retry loop, and the audit logger — is in this gist. Point it at a read-only user and a real schema, ask five questions from your own domain, and watch it refine SQL against the error messages.

Conclusion & Next Steps

You’ve built a text-to-SQL agent that only ever runs validated, read-only, cost-limited queries. Next steps: add an embedding index over “question → SQL” pairs so the agent retrieves a similar past query as a template, evaluate on a fixed set of questions with hand-written reference SQL, and wire the audit log into your existing observability stack.

References / Sources

• PostgreSQL documentation (EXPLAIN, information_schema, permissions). https://www.postgresql.org/docs
• Gemini API for generation. https://ai.google.dev/gemini-api/docs
• LlamaIndex text-to-SQL patterns. https://docs.llamaindex.ai