Building a SQL Query Agent with Claude Code: Complete Guide
Modern applications increasingly rely on natural language interfaces to databases. Instead of forcing users to learn SQL syntax, developers can build intelligent agents that translate plain English questions into executable queries, run them safely, and explain the results. Claude Code—Anthropic's agentic coding tool—provides an excellent foundation for this kind of project because it can read your schema, generate queries, execute them, and iterate when something goes wrong. In this guide, we'll build a complete SQL Query Agent from scratch, covering architecture, implementation, safety considerations, and deployment.
What Is a SQL Query Agent?
A SQL Query Agent is an AI-powered system that accepts natural language questions, converts them into SQL queries, executes those queries against a target database, and returns human-readable answers. Unlike a simple text-to-SQL converter, an agent operates in a loop: it inspects the schema, drafts a query, runs it, observes errors or unexpected results, and refines its approach until it produces a correct answer.
Claude Code is particularly well-suited for this because it is designed to operate autonomously within a codebase. It can read files, run shell commands, inspect database metadata, and write code—all within a controlled environment. By combining Claude Code's agentic capabilities with a well-structured prompt and a few helper scripts, you get a robust agent that can handle complex analytical questions.
Why It Matters
- Accessibility: Non-technical stakeholders—product managers, analysts, executives—can query data without learning SQL.
- Speed: Developers spend less time writing one-off reports and more time building features.
- Accuracy through iteration: Because the agent executes and validates queries, it catches its own mistakes before returning answers.
- Auditability: Every query the agent runs can be logged, reviewed, and replayed, giving you a transparent trail of how answers were derived.
- Extensibility: The same agent can be extended to generate charts, build dashboards, or trigger downstream workflows.
Project Architecture
Our SQL Query Agent will consist of four main components: a database connection layer, a schema introspection module, an agent runner that orchestrates Claude Code, and a safety layer that validates queries before execution. We'll use Python for the glue code, SQLite for the demo database (so you can run everything locally), and Claude Code as the reasoning engine.
The flow is straightforward: a user submits a question, the agent retrieves the current schema, Claude Code generates a SQL query, the safety layer validates it, the query executes against the database, and the agent formats the result. If an error occurs, the agent feeds the error back to Claude Code for a retry.
Setting Up the Environment
Before writing any code, make sure you have Python 3.10 or later installed, the Claude Code CLI authenticated, and a working directory for the project. Create a new folder and set up a virtual environment.
mkdir sql-query-agent
cd sql-query-agent
python -m venv .venv
source .venv/bin/activate
pip install anthropic python-dotenv
Next, ensure Claude Code is installed and authenticated. You can verify this by running a simple command:
claude --version
claude /doctor
If both commands succeed, you're ready to proceed. Create a project structure that looks like this:
sql-query-agent/
├── agent.py
├── db.py
├── schema.py
├── safety.py
├── seed.py
├── sample.db
├── .env
└── CLAUDE.md
Creating a Sample Database
To make the tutorial concrete, let's create a small e-commerce database with customers, orders, and products. The seed.py script will generate this database with realistic sample data.
# seed.py
import sqlite3
import random
from datetime import datetime, timedelta
def create_database():
conn = sqlite3.connect("sample.db")
cur = conn.cursor()
cur.executescript("""
DROP TABLE IF EXISTS order_items;
DROP TABLE IF EXISTS orders;
DROP TABLE IF EXISTS products;
DROP TABLE IF EXISTS customers;
CREATE TABLE customers (
customer_id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
signup_date TEXT NOT NULL,
country TEXT NOT NULL
);
CREATE TABLE products (
product_id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT NOT NULL,
price REAL NOT NULL,
stock INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE orders (
order_id INTEGER PRIMARY KEY,
customer_id INTEGER NOT NULL,
order_date TEXT NOT NULL,
status TEXT NOT NULL CHECK(status IN ('pending','shipped','delivered','cancelled')),
FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
);
CREATE TABLE order_items (
order_item_id INTEGER PRIMARY KEY,
order_id INTEGER NOT NULL,
product_id INTEGER NOT NULL,
quantity INTEGER NOT NULL,
unit_price REAL NOT NULL,
FOREIGN KEY (order_id) REFERENCES orders(order_id),
FOREIGN KEY (product_id) REFERENCES products(product_id)
);
""")
countries = ["USA", "Canada", "UK", "Germany", "France", "Japan", "Australia"]
categories = ["Electronics", "Books", "Clothing", "Home", "Sports"]
statuses = ["pending", "shipped", "delivered", "delivered", "delivered", "cancelled"]
for i in range(1, 101):
cur.execute(
"INSERT INTO customers (name, email, signup_date, country) VALUES (?, ?, ?, ?)",
(f"Customer {i}", f"customer{i}@example.com",
(datetime(2023,1,1) + timedelta(days=random.randint(0, 500))).isoformat(),
random.choice(countries))
)
for i in range(1, 51):
cur.execute(
"INSERT INTO products (name, category, price, stock) VALUES (?, ?, ?, ?)",
(f"Product {i}", random.choice(categories),
round(random.uniform(5, 500), 2), random.randint(0, 200))
)
for i in range(1, 201):
cust_id = random.randint(1, 100)
order_date = (datetime(2023,1,1) + timedelta(days=random.randint(0, 600))).isoformat()
cur.execute(
"INSERT INTO orders (customer_id, order_date, status) VALUES (?, ?, ?)",
(cust_id, order_date, random.choice(statuses))
)
order_id = cur.lastrowid
num_items = random.randint(1, 4)
for _ in range(num_items):
prod_id = random.randint(1, 50)
qty = random.randint(1, 5)
cur.execute("SELECT price FROM products WHERE product_id = ?", (prod_id,))
price = cur.fetchone()[0]
cur.execute(
"INSERT INTO order_items (order_id, product_id, quantity, unit_price) VALUES (?, ?, ?, ?)",
(order_id, prod_id, qty, price)
)
conn.commit()
conn.close()
print("Database created with sample data.")
if __name__ == "__main__":
create_database()
Run the script to generate the database:
python seed.py
Building the Database Connection Layer
The db.py module wraps SQLite operations so the agent can execute queries in a controlled manner. It enforces read-only access and returns structured results.
# db.py
import sqlite3
from typing import List, Tuple, Optional
DB_PATH = "sample.db"
def execute_query(sql: str, params: Optional[Tuple] = None) -> Tuple[List[str], List[Tuple]]:
"""Execute a SQL query and return (column_names, rows)."""
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cur = conn.cursor()
try:
cur.execute(sql, params or ())
rows = cur.fetchall()
columns = [desc[0] for desc in cur.description] if cur.description else []
data = [tuple(row) for row in rows]
return columns, data
finally:
conn.close()
def execute_scalar(sql: str, params: Optional[Tuple] = None):
"""Execute a query and return a single scalar value."""
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
try:
cur.execute(sql, params or ())
result = cur.fetchone()
return result[0] if result else None
finally:
conn.close()
Schema Introspection
The agent needs to understand the database structure before it can write queries. The schema.py module extracts table definitions, column types, and foreign key relationships, then formats them as a readable string that Claude Code can consume.
# schema.py
import sqlite3
from db import DB_PATH
def get_schema_text() -> str:
"""Return a formatted string describing all tables, columns, and relationships."""
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
cur.execute("""
SELECT name FROM sqlite_master
WHERE type='table' AND name NOT LIKE 'sqlite_%'
ORDER BY name
""")
tables = [row[0] for row in cur.fetchall()]
lines = []
for table in tables:
cur.execute(f"PRAGMA table_info({table})")
columns = cur.fetchall()
cur.execute(f"PRAGMA foreign_key_list({table})")
fks = cur.fetchall()
col_defs = []
for col in columns:
col_name = col[1]
col_type = col[2]
not_null = " NOT NULL" if col[3] else ""
pk = " PRIMARY KEY" if col[5] else ""
col_defs.append(f" {col_name} {col_type}{not_null}{pk}")
lines.append(f"TABLE {table} (")
lines.append(",\n".join(col_defs))
for fk in fks:
lines.append(f" -- FOREIGN KEY ({fk[3]}) REFERENCES {fk[2]}({fk[4]})")
lines.append(");")
lines.append("")
conn.close()
return "\n".join(lines)
if __name__ == "__main__":
print(get_schema_text())
Running python schema.py will print a human-readable schema that includes every table, its columns, types, constraints, and foreign keys. This output becomes a critical part of the agent's context.
Implementing the Safety Layer
Before any query reaches the database, it must pass through validation. The safety layer blocks destructive operations, enforces query limits, and strips potentially dangerous patterns. This is essential even in a read-only demo because it establishes good habits for production systems.
# safety.py
import re
BLOCKED_KEYWORDS = [
"DROP", "DELETE", "INSERT", "UPDATE", "ALTER",
"CREATE", "TRUNCATE", "ATTACH", "DETACH", "PRAGMA",
"REPLACE", "VACUUM", "REINDEX"
]
MAX_RESULT_ROWS = 1000
def validate_query(sql: str) -> tuple[bool, str]:
"""Validate a SQL query for safety. Returns (is_safe, message)."""
if not sql or not sql.strip():
return False, "Empty query."
normalized = re.sub(r'\s+', ' ', sql.strip().upper())
for keyword in BLOCKED_KEYWORDS:
pattern = rf'\b{keyword}\b'
if re.search(pattern, normalized):
return False, f"Blocked keyword detected: {keyword}"
if not normalized.startswith("SELECT") and not normalized.startswith("WITH"):
return False, "Only SELECT or WITH queries are allowed."
semicolons = sql.count(';')
if semicolons > 1 or (semicolons == 1 and not sql.strip().endswith(';')):
return False, "Multiple statements are not allowed."
return True, "Query is safe."
def add_limit(sql: str, limit: int = MAX_RESULT_ROWS) -> str:
"""Append a LIMIT clause if the query doesn't already have one."""
normalized = sql.strip().upper()
if "LIMIT" in normalized:
return sql
return f"{sql.rstrip(';')} LIMIT {limit};"
Writing the Agent Orchestrator
The agent.py file is the heart of the system. It uses the Anthropic Python SDK to communicate with Claude, provides the schema as context, sends the user's question, receives a SQL query, validates it, executes it, and returns the result. If execution fails, it feeds the error back to Claude for a retry—up to a configurable maximum number of attempts.
# agent.py
import os
import json
import sys
from anthropic import Anthropic
from schema import get_schema_text
from safety import validate_query, add_limit
from db import execute_query
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
MODEL = "claude-sonnet-4-20250514"
MAX_RETRIES = 3
SYSTEM_PROMPT = """You are a SQL query agent. Your job is to translate natural language
questions into SQL queries for a SQLite database.
Rules:
1. Only produce SELECT or WITH queries. Never produce DML or DDL.
2. Use the exact table and column names from the provided schema.
3. Return ONLY the SQL query inside a sql code block. No explanations.
4. If the question is ambiguous, make reasonable assumptions and proceed.
5. Always alias computed columns clearly.
6. Prefer CTEs for complex multi-step queries for readability.
"""
def build_user_message(question: str, schema: str) -> str:
return f"""Database schema:
{schema}
Question: {question}
Write a SQL query that answers this question. Return only the query in a sql block.
"""
def extract_sql(response_text: str) -> str:
"""Extract SQL from a markdown code block."""
if "sql" in response_text:
start = response_text.index("sql") + 6
end = response_text.index("", start)
return response_text[start:end].strip()
if "" in response_text:
start = response_text.index("") + 3
end = response_text.index("", start)
return response_text[start:end].strip()
return response_text.strip()
def run_agent(question: str) -> dict:
schema = get_schema_text()
messages = [{"role": "user", "content": build_user_message(question, schema)}]
for attempt in range(1, MAX_RETRIES + 1):
response = client.messages.create(
model=MODEL,
max_tokens=2048,
system=SYSTEM_PROMPT,
messages=messages
)
assistant_text = response.content[0].text
sql = extract_sql(assistant_text)
is_safe, safety_msg = validate_query(sql)
if not is_safe:
messages.append({"role": "assistant", "content": assistant_text})
messages.append({
"role": "user",
"content": f"The query was rejected: {safety_msg}. Please rewrite it."
})
continue
sql = add_limit(sql)
try:
columns, rows = execute_query(sql)
return {
"success": True,
"sql": sql,
"columns": columns,
"rows": rows,
"row_count": len(rows),
"attempts": attempt
}
except Exception as e:
messages.append({"role": "assistant", "content": assistant_text})
messages.append({
"role": "user",
"content": f"The query failed with error: {e}. Please fix and retry."
})
return {"success": False, "error": "Max retries exceeded.", "attempts": MAX_RETRIES}
def format_result(result: dict) -> str:
if not result["success"]:
return f"Agent failed after {result['attempts']} attempts: {result.get('error', 'Unknown error')}"
lines = []
lines.append(f"Query (attempt {result['attempts']}):")
lines.append(result["sql"])
lines.append("")
lines.append(f"Results: {result['row_count']} rows")
lines.append("")
if result["row_count"] == 0:
lines.append("(no rows returned)")
return "\n".join(lines)
col_widths = [
max(len(str(col)), max((len(str(row[i])) for row in result["rows"]), default=0))
for i, col in enumerate(result["columns"])
]
header = " | ".join(str(col).ljust(col_widths[i]) for i, col in enumerate(result["columns"]))
separator = "-+-".join("-" * w for w in col_widths)
lines.append(header)
lines.append(separator)
for row in result["rows"][:50]:
lines.append(" | ".join(str(val).ljust(col_widths[i]) for i, val in enumerate(row)))
if result["row_count"] > 50:
lines.append(f"... and {result['row_count'] - 50} more rows")
return "\n".join(lines)
if __name__ == "__main__":
question = " ".join(sys.argv[1:]) or "What are the top 5 customers by total order value?"
result = run_agent(question)
print(format_result(result))
Set your API key in a .env file or export it as an environment variable:
export ANTHROPIC_API_KEY="sk-ant-..."
Now run the agent with a natural language question:
python agent.py "What are the top 5 customers by total order value?"
You should see output that includes the generated SQL query and a formatted table of results. Try a few more questions to test the agent's range:
python agent.py "How many orders were cancelled last month?"
python agent.py "Which product category generates the most revenue?"
python agent.py "List customers who have never placed a delivered order"
Integrating with Claude Code
While the Python script above uses the Anthropic API directly, you can also leverage Claude Code's CLI for a more interactive, file-aware experience. Create a CLAUDE.md file in your project root that instructs Claude Code how to behave as a SQL agent.
# CLAUDE.md
## SQL Query Agent
This project is a SQL Query Agent for a SQLite database at `sample.db`.
### Your Role
You are a SQL query agent. When asked a question about the data, you should:
1. Read the schema by running `python schema.py`
2. Write a SQL query to answer the question
3. Validate it is a SELECT-only query
4. Execute it using `python -c "from db import execute_query; ..."` or by writing a small script
5. Present the results in a readable format
### Rules
- NEVER run DROP, DELETE, INSERT, UPDATE, or any DML/DDL
- Always verify column names against the actual schema
- If a query fails, read the error, fix the query, and retry
- Present results as a formatted table
- Explain your reasoning briefly before showing the query
Now you can launch Claude Code in the project directory and ask questions directly:
claude
> What are the top 5 products by sales volume in the Electronics category?
Claude Code will read the CLAUDE.md instructions, inspect the schema, write and execute a query, and present the results—all within the terminal session. This approach is powerful because Claude Code can also create visualization scripts, write the query to a file for reuse, or build a small web interface on top of the agent.
Adding a Conversational Interface
For a more user-friendly experience, wrap the agent in a simple REPL loop that maintains conversation history. This allows follow-up questions like "break that down by country" or "show me only the top 3."
# chat.py
import sys
from agent import run_agent, format_result
def main():
print("SQL Query Agent - Interactive Mode")
print("Type 'quit' to exit, 'schema' to view the database schema.\n")
while True:
try:
question = input("question> ").strip()
except (EOFError, KeyboardInterrupt):
print("\nGoodbye!")
break
if not question:
continue
if question.lower() in ("quit", "exit", "q"):
print("Goodbye!")
break
if question.lower() == "schema":
from schema import get_schema_text
print(get_schema_text())
continue
print("\nThinking...\n")
result = run_agent(question)
print(format_result(result))
print()
if __name__ == "__main__":
main()
Run it with:
python chat.py
Best Practices
Always Validate Before Executing
Never trust AI-generated SQL blindly. The safety layer should be non-negotiable, even if you're only using a read-only database. In production, consider running queries against a read replica or a snapshot to eliminate any risk of side effects.
Provide Rich Schema Context
The more context you give the agent, the better its queries will be. Beyond table and column names, include sample values, column descriptions, and common business rules. You can extend schema.py to read from a metadata table or a YAML file that documents each column's meaning.
# Example: extended schema metadata
COLUMN_DESCRIPTIONS = {
"orders.status": "One of: pending, shipped, delivered, cancelled. 'delivered' means the customer received the order.",
"order_items.unit_price": "The price at time of purchase, which may differ from products.price due to discounts.",
"customers.signup_date": "ISO 8601 date string. Used for cohort analysis."
}
Log Every Query
Maintain an audit log of every query the agent executes, including the original question, the generated SQL, the result count, and any errors. This is invaluable for debugging, compliance, and improving the agent over time.
# Add to agent.py
import logging
from datetime import datetime
logging.basicConfig(
filename="agent.log",
level=logging.INFO,
format="%(asctime)s | %(message)s"
)
# Inside run_agent, after successful execution:
logging.info(f"Q: {question} | SQL: {sql} | Rows: {len(rows)} | Attempts: {attempt}")
Use the Right Model
For simple queries, Claude Sonnet is fast and cost-effective. For complex analytical questions involving multiple joins, window functions, or nuanced business logic, Claude Opus may produce better results. You can dynamically choose the model based on query complexity—for example, routing questions with more than 20 words or containing words like "correlation," "trend," or "cohort" to the more capable model.
Handle Ambiguity Gracefully
Some questions are genuinely ambiguous. Rather than guessing silently, you can instruct the agent to ask clarifying questions when the intent is unclear. Modify the system prompt to include a rule like: "If the question is ambiguous in a way that would materially change the query, ask for clarification instead of guessing."
Rate-Limit and Cache
In production, implement rate limiting to prevent abuse and caching to avoid re-running identical queries. A simple cache keyed on the normalized question string can dramatically reduce API costs and response times for repeated questions.
# Simple in-memory cache
from functools import lru_cache
@lru_cache(maxsize=128)
def run_agent_cached(question: str) -> str:
result = run_agent(question)
return format_result(result)
Test with Edge Cases
Build a test suite of questions that exercise edge cases: empty result sets, questions referencing non-existent columns, questions requiring complex joins, questions with date ranges, and questions that might tempt the agent to write destructive queries. This ensures your safety layer and prompt engineering hold up under pressure.
# test_agent.py
import pytest
from safety import validate_query
def test_select_allowed():
ok, _ = validate_query("SELECT * FROM customers")
assert ok
def test_drop_blocked():
ok, msg = validate_query("DROP TABLE customers")
assert not ok
assert "DROP" in msg
def test_multiple_statements_blocked():
ok, _ = validate_query("SELECT 1; SELECT 2")
assert not ok
def test_update_blocked():
ok, _ = validate_query("UPDATE customers SET name = 'x'")
assert not ok
Extending the Agent
Once the core agent is working, there are many directions to take it. You can add chart generation by having the agent write Python scripts that use matplotlib or plotly. You can connect it to PostgreSQL or MySQL by swapping the connection layer in db.py. You can expose it as a REST API using FastAPI, or integrate it into a Slack bot so team members can ask data questions in a channel.
Here's a minimal FastAPI wrapper:
# api.py
from fastapi import FastAPI
from pydantic import BaseModel
from agent import run_agent, format_result
app = FastAPI(title="SQL Query Agent API")
class Question(BaseModel):
question: str
@app.post("/ask")
def ask(q: Question):
result = run_agent(q.question)
return {
"success": result["success"],
"sql": result.get("sql"),
"columns": result.get("columns", []),
"rows": result.get("rows", []),
"row_count": result.get("row_count", 0),
"attempts": result.get("attempts", 0),
"error": result.get("error")
}
Install FastAPI and uvicorn, then start the server:
pip install fastapi uvicorn
uvicorn api:app --reload
You can now POST questions to http://localhost:8000/ask and receive structured JSON responses.
Conclusion
Building a SQL Query Agent with Claude Code is a practical way to democratize data access within your organization. By combining schema introspection, a robust safety layer, an iterative retry loop, and Claude's strong reasoning capabilities, you get a system that can handle real-world analytical questions with surprising accuracy. The key to a production-ready agent is not just the model—it's the guardrails you put around it. Validate every query, log every execution, provide rich context, and test relentlessly. Start with the SQLite demo in this guide, then gradually extend it to your real database, add visualization, and expose it through whatever interface your team prefers. With the foundation in place, the agent becomes a versatile tool that grows alongside your data needs.