Building a SQL Query Agent with CrewAI: Complete Guide

Introduction to Building a SQL Query Agent with CrewAI

Modern applications generate enormous volumes of data, and business stakeholders increasingly want to ask questions in plain English rather than write SQL themselves. A SQL Query Agent bridges that gap by translating natural language questions into valid SQL queries, executing them against a database, and returning human-readable insights. CrewAI, an open-source multi-agent framework, makes it straightforward to orchestrate this workflow with specialized agents that collaborate to deliver accurate results.

In this guide, you'll learn what a SQL Query Agent is, why it matters, how to build one end-to-end with CrewAI, and the best practices that separate a toy demo from a production-ready system.

What Is a SQL Query Agent?

A SQL Query Agent is an AI-powered assistant that understands a database schema, interprets natural language requests, generates SQL, executes the queries safely, and summarizes the results. Rather than relying on a single LLM call, a well-designed agent breaks the task into discrete steps: schema inspection, query generation, validation, execution, and reporting.

CrewAI structures this kind of workflow using the concept of crews, which are teams of agents that perform tasks using tools. Each agent has a role, a goal, and a backstory that shape its behavior. Tasks define what needs to be done and which agent should do it. Tools let agents interact with external systems — in our case, a relational database.

Key Components of a CrewAI SQL Agent

Agents: A schema reader, a SQL writer, a validator, and an analyst.
Tasks: Inspect schema, generate query, validate SQL, execute, and summarize.
Tools: Custom functions that connect to the database, run queries, and return results.
Process: Sequential or hierarchical orchestration that defines the order of execution.

Why It Matters

Hand-writing SQL is a bottleneck for non-technical users and even slows down engineers who must context-switch between application code and database queries. A SQL Query Agent democratizes data access, reduces the load on data teams, and accelerates decision-making. When built correctly, it also enforces guardrails — preventing destructive operations, validating syntax, and surfacing schema-aware corrections — that a naive LLM prompt cannot provide.

For organizations, this translates into faster analytics, fewer misinterpreted requests, and a consistent interface across multiple databases. For developers, CrewAI offers a clean abstraction that keeps orchestration logic readable and maintainable.

Prerequisites and Setup

Before writing code, ensure you have Python 3.10 or later and a working database. This tutorial uses SQLite for simplicity, but the same approach works with PostgreSQL, MySQL, or any SQLAlchemy-compatible database.

Installing Dependencies

Create a virtual environment and install the required packages:

python -m venv .venv
source .venv/bin/activate
pip install crewai crewai-tools sqlalchemy python-dotenv

You'll also need an OpenAI API key (or another supported LLM provider). Store it in a .env file:

OPENAI_API_KEY=sk-your-key-here

Preparing a Sample Database

Let's create a small e-commerce database with a few tables so the agent has something meaningful to query:

import sqlite3

conn = sqlite3.connect("shop.db")
cur = conn.cursor()

cur.executescript("""
CREATE TABLE IF NOT EXISTS customers (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE,
    country TEXT,
    signup_date TEXT
);

CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    category TEXT,
    price REAL
);

CREATE TABLE IF NOT EXISTS orders (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER,
    order_date TEXT,
    total REAL,
    FOREIGN KEY (customer_id) REFERENCES customers(id)
);

CREATE TABLE IF NOT EXISTS order_items (
    id INTEGER PRIMARY KEY,
    order_id INTEGER,
    product_id INTEGER,
    quantity INTEGER,
    FOREIGN KEY (order_id) REFERENCES orders(id),
    FOREIGN KEY (product_id) REFERENCES products(id)
);
""")

# Insert sample data
cur.executemany("INSERT INTO customers (name, email, country, signup_date) VALUES (?, ?, ?, ?)", [
    ("Alice", "alice@example.com", "USA", "2024-01-15"),
    ("Bob", "bob@example.com", "UK", "2024-02-20"),
    ("Carol", "carol@example.com", "USA", "2024-03-10"),
])

cur.executemany("INSERT INTO products (name, category, price) VALUES (?, ?, ?)", [
    ("Laptop", "Electronics", 1200.00),
    ("Mouse", "Electronics", 25.00),
    ("Notebook", "Stationery", 5.00),
])

cur.executemany("INSERT INTO orders (customer_id, order_date, total) VALUES (?, ?, ?)", [
    (1, "2024-04-01", 1225.00),
    (2, "2024-04-05", 30.00),
    (3, "2024-04-10", 1200.00),
])

cur.executemany("INSERT INTO order_items (order_id, product_id, quantity) VALUES (?, ?, ?)", [
    (1, 1, 1),
    (1, 2, 1),
    (2, 2, 1),
    (2, 3, 1),
    (3, 1, 1),
])

conn.commit()
conn.close()
print("Database created.")

Building Custom Tools

Tools are the bridge between agents and the outside world. We'll create three tools: one to retrieve the schema, one to execute read-only queries, and one to validate SQL syntax. CrewAI tools are typically defined using the @tool decorator from crewai.tools.

import sqlite3
from crewai.tools import tool

DB_PATH = "shop.db"

@tool("Get Database Schema")
def get_schema() -> str:
    """Returns the schema of all tables in the database, including column names and types."""
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    cur.execute("SELECT name, sql FROM sqlite_master WHERE type='table';")
    tables = cur.fetchall()
    conn.close()
    schema_lines = [f"Table: {name}\nDDL: {ddl}\n" for name, ddl in tables]
    return "\n".join(schema_lines)

@tool("Execute SQL Query")
def execute_query(query: str) -> str:
    """Executes a read-only SQL query against the database and returns the results as text.
    Only SELECT statements are allowed."""
    normalized = query.strip().lower()
    if not normalized.startswith("select"):
        return "Error: Only SELECT queries are permitted."

    conn = sqlite3.connect(DB_PATH)
    try:
        cur = conn.cursor()
        cur.execute(query)
        rows = cur.fetchall()
        columns = [desc[0] for desc in cur.description] if cur.description else []
        if not rows:
            return "Query executed successfully. No rows returned."
        result = ", ".join(columns) + "\n"
        result += "\n".join(", ".join(str(v) for v in row) for row in rows)
        return result
    except Exception as e:
        return f"Query failed: {e}"
    finally:
        conn.close()

Notice the safety check in execute_query: we reject any statement that doesn't begin with SELECT. This is a minimal guardrail; in production you'd also want query timeouts, row limits, and connection pooling.

Defining Agents

With tools in place, we define the agents. Each agent has a role, goal, backstory, and the tools it's allowed to use. We'll create two agents: a SQL Developer that writes queries, and a Data Analyst that interprets results.

from crewai import Agent, LLM

llm = LLM(model="gpt-4o-mini", temperature=0)

sql_developer = Agent(
    role="Senior SQL Developer",
    goal="Translate natural language questions into accurate, efficient SQL queries using the provided schema.",
    backstory=(
        "You are a meticulous SQL developer with 15 years of experience across "
        "PostgreSQL, MySQL, and SQLite. You always inspect the schema before writing "
        "queries, prefer explicit JOINs, and never use SELECT * in production code."
    ),
    tools=[get_schema, execute_query],
    llm=llm,
    verbose=True,
)

data_analyst = Agent(
    role="Business Data Analyst",
    goal="Interpret query results and present clear, actionable insights to non-technical stakeholders.",
    backstory=(
        "You translate raw data into business narratives. You highlight trends, "
        "call out anomalies, and always frame numbers in terms a product manager "
        "would understand."
    ),
    llm=llm,
    verbose=True,
)

Setting temperature=0 reduces randomness, which is desirable for SQL generation where precision matters more than creativity.

Defining Tasks

Tasks describe the work to be done and assign it to an agent. The first task generates and executes a SQL query; the second task takes the output and produces a human-readable summary.

from crewai import Task

def build_tasks(question: str):
    query_task = Task(
        description=(
            f"Use the get_schema tool to inspect the database schema, then write a SQL "
            f"query that answers the following question: '{question}'. Execute the query "
            f"using the execute_query tool. If the query fails, inspect the error, correct "
            f"the SQL, and retry. Return the final query and its raw results."
        ),
        expected_output="The SQL query used and the raw rows returned by the database.",
        agent=sql_developer,
    )

    analysis_task = Task(
        description=(
            "Take the raw query results from the previous task and write a concise, "
            "non-technical summary that directly answers the original question. Include "
            "key numbers and a short explanation."
        ),
        expected_output="A plain-English summary of the findings with supporting figures.",
        agent=data_analyst,
        context=[query_task],
    )

    return [query_task, analysis_task]

The context parameter on the analysis task ensures the analyst receives the output of the query task, creating a clean handoff between agents.

Assembling and Running the Crew

Now we bring everything together into a Crew and execute it with a sample question:

from crewai import Crew, Process

question = "Which country has generated the highest total order revenue?"

crew = Crew(
    agents=[sql_developer, data_analyst],
    tasks=build_tasks(question),
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff()
print("\n=== FINAL ANSWER ===\n")
print(result.raw)

When you run this script, the SQL Developer first calls get_schema, inspects the tables, writes a JOIN between orders and customers, executes it via execute_query, and passes the rows to the Data Analyst. The analyst then produces a summary like: "The USA generated the highest total order revenue of $2,425.00 across two customers, compared to $30.00 from the UK."

Adding a Validation Layer

For more robust systems, introduce a dedicated validation agent that reviews generated SQL before execution. This catches common mistakes like referencing non-existent columns or forgetting JOIN conditions.

sql_validator = Agent(
    role="SQL Validator",
    goal="Review SQL queries for correctness, safety, and performance before execution.",
    backstory=(
        "You are a database reliability engineer who has seen every kind of query "
        "mistake. You check for missing JOIN conditions, ambiguous column references, "
        "and potential performance pitfalls."
    ),
    llm=llm,
    verbose=True,
)

validation_task = Task(
    description=(
        "Review the SQL query produced for the question. Confirm it references only "
        "valid tables and columns, uses proper JOINs, and is read-only. If issues are "
        "found, return a corrected query. Otherwise, approve it unchanged."
    ),
    expected_output="An approved or corrected SQL query ready for execution.",
    agent=sql_validator,
)

Insert this task between query generation and execution to create a three-stage pipeline: generate, validate, execute. This pattern dramatically reduces the rate of runtime errors on complex schemas.

Best Practices

Restrict permissions: Connect with a read-only database user. Never let the agent run DDL or DML statements in production.
Limit result sets: Add LIMIT clauses or wrap queries with row-count caps to avoid returning massive payloads to the LLM.
Cache schema lookups: Schema rarely changes between requests. Cache the output of get_schema to reduce token usage and latency.
Use structured outputs: For programmatic consumers, define a Pydantic model and use CrewAI's structured output support to return JSON instead of free text.
Log every query: Persist each generated SQL statement, its parameters, and execution time for auditing and debugging.
Test with edge cases: Include ambiguous questions, empty result sets, and schema mismatches in your test suite to evaluate agent robustness.
Choose the right model: Use a capable model like GPT-4o for complex schemas, but a smaller model like GPT-4o-mini is often sufficient for simple databases and keeps costs low.
Provide few-shot examples: Embedding example question-query pairs in the agent's backstory or task description improves accuracy on domain-specific schemas.

Conclusion

Building a SQL Query Agent with CrewAI turns a fragile single-prompt approach into a structured, multi-step workflow with clear separation of concerns. By combining schema-aware tools, specialized agents, and a validation pipeline, you create a system that is safer, more accurate, and easier to maintain. Start with the minimal two-agent crew on a sample database, then layer in validation, logging, and structured outputs as your needs grow. With the patterns in this guide, you're well equipped to ship a natural-language data interface that your whole team can rely on.