Skip to content

6. Text-to-SQL agents

Intermediate · 13 min read

The text-to-SQL pipeline puts the schema in the prompt and asks for one query. That breaks down when the database has hundreds of tables, when the right filter values aren't known in advance, or when answering needs several queries. A SQL agent gets tools to explore the database itself — list tables, inspect columns and sample rows, run queries — and works step by step until it can answer.

flowchart LR
    Q[Question] --> A{Agent LLM}
    A -->|list_tables| T[Tools]
    A -->|describe_table| T
    A -->|run_query| T
    T -->|results / errors| A
    A -->|final answer + SQL| U[User]

The tool loop itself is explained in Structured output & tool calling.

6.1 Sample database

import json
import re
import sqlite3

setup = sqlite3.connect("shop.db")
setup.executescript("""
DROP TABLE IF EXISTS customers; DROP TABLE IF EXISTS orders; DROP TABLE IF EXISTS tickets;
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, city TEXT, plan TEXT, email TEXT);
CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL, status TEXT, created_at TEXT);
CREATE TABLE tickets (id INTEGER PRIMARY KEY, customer_id INTEGER, category TEXT, created_at TEXT);
INSERT INTO customers VALUES
    (1,'Priya','Bengaluru','pro','priya@example.com'), (2,'Rahul','Pune','free','rahul@example.com'),
    (3,'Ananya','Pune','pro','ananya@example.com'),    (4,'Arjun','Delhi','free','arjun@example.com');
INSERT INTO orders VALUES
    (101,1,2499,'delivered','2026-09-02'), (102,2,799,'delivered','2026-09-05'),
    (103,1,1299,'shipped','2026-09-20'),   (104,3,5600,'delivered','2026-09-21'),
    (105,4,450,'cancelled','2026-09-22'),  (106,3,999,'shipped','2026-10-01');
INSERT INTO tickets VALUES
    (1,1,'billing','2026-09-03'), (2,2,'shipping','2026-09-06'), (3,1,'shipping','2026-09-21'), (4,3,'bug','2026-09-25');
""")
setup.commit()
setup.close()
ro = sqlite3.connect("file:shop.db?mode=ro", uri=True)

6.2 The tools

Each tool returns compact text — tool results go into the context window, so keep them small:

HIDDEN = {"customers.email"}
MAX_ROWS = 20

def list_tables() -> str:
    rows = ro.execute("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name").fetchall()
    return ", ".join(r[0] for r in rows)

def describe_table(table: str) -> str:
    if table not in list_tables().split(", "):
        return f"ERROR: no table named {table!r}. Available: {list_tables()}"
    cols = [(name, ctype) for _, name, ctype, *_ in ro.execute(f"PRAGMA table_info({table})")
            if f"{table}.{name}" not in HIDDEN]
    names = ", ".join(c for c, _ in cols)
    sample = ro.execute(f"SELECT {names} FROM {table} LIMIT 3").fetchall()
    return (f"{table}(" + ", ".join(f"{c} {t}" for c, t in cols) + ")\n"
            f"sample rows: {sample}")

def run_query(sql: str) -> str:
    sql = sql.strip().rstrip(";")
    if ";" in sql or not re.match(r"^\s*(select|with)\b", sql, re.I) or "email" in sql.lower():
        return "ERROR: only a single SELECT on permitted columns is allowed."
    try:
        cur = ro.execute(f"SELECT * FROM ({sql}) LIMIT {MAX_ROWS + 1}")
        cols, rows = [d[0] for d in cur.description], cur.fetchall()
    except sqlite3.Error as e:
        return f"ERROR: {e}"
    note = f" (first {MAX_ROWS} rows only)" if len(rows) > MAX_ROWS else ""
    return json.dumps({"columns": cols, "rows": rows[:MAX_ROWS]}) + note

TOOLS = {"list_tables": list_tables, "describe_table": describe_table, "run_query": run_query}

print(list_tables())
print(describe_table("orders"))
print(run_query("SELECT status, COUNT(*) FROM orders GROUP BY status"))
print(run_query("SELECT * FROM refunds"))
Output
customers, orders, tickets
orders(id INTEGER, customer_id INTEGER, amount REAL, status TEXT, created_at TEXT)
sample rows: [(101, 1, 2499.0, 'delivered', '2026-09-02'), (102, 2, 799.0, 'delivered', '2026-09-05'), (103, 1, 1299.0, 'shipped', '2026-09-20')]
{"columns": ["status", "COUNT(*)"], "rows": [["cancelled", 1], ["delivered", 3], ["shipped", 2]]}
ERROR: no such table: refunds

Sample rows are what let the agent discover that status values are lowercase, or that cities are spelled "Bengaluru" — the filter-value errors that sink one-shot text-to-SQL.

6.3 The agent loop

As in the LLM notes, a scripted fake model plays the LLM's part so every step is visible. With a real model, each decide call is one chat-completion request with these three tools defined.

SYSTEM = """You answer questions about a SQLite database by calling tools.
Explore with list_tables and describe_table before writing queries. Use run_query to get data.
Only SELECT queries. When you have the answer, reply with a short answer and the final SQL."""

class ScriptedAgentLLM:
    def __init__(self, steps): self.steps = iter(steps)
    def decide(self, messages): return next(self.steps)

def run_agent(question: str, llm, max_steps: int = 8) -> str:
    messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
    for step in range(1, max_steps + 1):
        action = llm.decide(messages)
        if "answer" in action:
            return action["answer"]
        result = TOOLS[action["tool"]](**action.get("args", {}))
        print(f"step {step}: {action['tool']}({action.get('args', {})})\n        → {result.replace(chr(10), ' | ')[:110]}")
        messages += [{"role": "assistant", "content": json.dumps(action)},
                     {"role": "tool", "content": result}]
    return "Stopped: too many steps without an answer."

llm = ScriptedAgentLLM([
    {"tool": "list_tables"},
    {"tool": "describe_table", "args": {"table": "tickets"}},
    {"tool": "run_query", "args": {"sql": "SELECT c.name, COUNT(*) AS n FROM tickets t "
                                          "JOIN customer c ON c.id = t.customer_id GROUP BY c.name"}},     # typo: customer
    {"tool": "run_query", "args": {"sql": "SELECT c.name, COUNT(*) AS n FROM tickets t "
                                          "JOIN customers c ON c.id = t.customer_id GROUP BY c.name ORDER BY n DESC"}},
    {"answer": "Priya raised the most support tickets (2). Everyone else raised 1.\n"
               "SQL: SELECT c.name, COUNT(*) AS n FROM tickets t JOIN customers c ON c.id = t.customer_id "
               "GROUP BY c.name ORDER BY n DESC"},
])
print(run_agent("Which customer has raised the most support tickets?", llm))
Output
step 1: list_tables({})
        → customers, orders, tickets
step 2: describe_table({'table': 'tickets'})
        → tickets(id INTEGER, customer_id INTEGER, category TEXT, created_at TEXT) | sample rows: [(1, 1, 'billing', '20
step 3: run_query({'sql': 'SELECT c.name, COUNT(*) AS n FROM tickets t JOIN customer c ON c.id = t.customer_id GROUP BY c.name'})
        → ERROR: no such table: customer
step 4: run_query({'sql': 'SELECT c.name, COUNT(*) AS n FROM tickets t JOIN customers c ON c.id = t.customer_id GROUP BY c.name ORDER BY n DESC'})
        → {"columns": ["name", "n"], "rows": [["Priya", 2], ["Rahul", 1], ["Ananya", 1]]}
Priya raised the most support tickets (2). Everyone else raised 1.
SQL: SELECT c.name, COUNT(*) AS n FROM tickets t JOIN customers c ON c.id = t.customer_id GROUP BY c.name ORDER BY n DESC

Step 3 shows the key benefit of the agent design: the database error went back to the model as a tool result, and it corrected the table name on its own.

6.4 Making answers trustworthy

  • Show the SQL (and the rows) behind every answer, so users and analysts can verify it.
  • Check the answer against the data: the final numbers should appear in the last query result. A simple check — every number in the answer must exist in the tool results — catches invented figures.
  • Ask, don't guess: if the question is ambiguous ("top customers"), the agent should ask which measure, or state its assumption in the answer.
def numbers_grounded(answer: str, tool_results: list[str]) -> bool:
    nums = set(re.findall(r"\d+(?:\.\d+)?", answer.split("SQL:")[0]))
    return all(any(n in r for r in tool_results) for n in nums)

results = ['{"columns": ["name", "n"], "rows": [["Priya", 2], ["Ananya", 1], ["Rahul", 1]]}']
print(numbers_grounded("Priya raised the most tickets (2).", results))
print(numbers_grounded("Priya raised the most tickets (5).", results))
Output
True
False

6.5 Production guardrails

Risk Guardrail
Writes or destructive SQL read-only database role; validation; no DDL/DML tools at all
Seeing other customers' or other tenants' data row-level security or per-user views; the connection is opened as the user, not as an admin
Sensitive columns (emails, phone numbers, salaries) expose views without them; hide them from describe_table
Expensive queries statement timeout; row limits; run against a read replica or warehouse, not the production primary
Endless loops max_steps; a token/cost budget per question
Wrong business logic a semantic layer: named, pre-defined metrics ("revenue", "active customer") the agent must use
Prompt injection via data (e.g. a ticket text saying "ignore your rules") treat query results as data; the database permissions above limit the damage

The semantic layer

Rather than letting the agent invent "revenue" each time, give it a tool like get_metric(name="revenue", group_by="city", period="last_month") backed by definitions you control (dbt metrics, Cube, or your own views). Free-form SQL becomes the fallback, not the default. Answers become consistent with the company dashboards — which is what business users expect.

6.6 Frameworks

LangChain (SQLDatabaseToolkit, create_sql_agent), LlamaIndex (NLSQLTableQueryEngine) and Vanna provide ready-made versions of these tools and loops. They're quick to start with; the guardrails above are still your job.

Interview questions

When would you use a SQL agent instead of a single text-to-SQL call?

When the schema is too big for one prompt, when the right tables or filter values must be discovered (sample rows, distinct values), or when answering needs several dependent queries. The agent explores with tools and corrects errors from database feedback, at the cost of more LLM calls and latency.

How do you stop a SQL agent from leaking data?

Enforce access in the database, not the prompt: a read-only role, views or row-level security scoped to the current user/tenant, sensitive columns removed, query timeouts and row limits. Hide sensitive columns from schema tools, log every query, and cap steps and cost per question.

What is a semantic layer and why does it help text-to-SQL?

A set of governed definitions of business metrics and dimensions (revenue, active user, region). Having the agent call these instead of writing raw SQL makes answers consistent with official reports and avoids errors in business logic, joins and filters.

Practice

  • Add a distinct_values(table, column) tool and use it in a scripted run where the user asks about "Bangalore".
  • Record every run_query call (question, SQL, row count, error) in a log table, then query it with the techniques from Querying LLM logs.

Next: Pandas — work with query results as DataFrames.