Building a CrewAI tool that queries a SQL database

This crewai sql database tool tutorial walks through building a custom tool that lets a CrewAI agent run read-only SQL queries against a live database. We’ll use SQLAlchemy with SQLite so you can run everything locally, then wire the tool into an agent that answers natural-language questions with real data. By the end you’ll have a reusable pattern for giving any CrewAI crew structured database access.

Step 1: Set up the environment and seed a database

Install the framework and a database driver. CrewAI ships tooling for custom integrations; SQLAlchemy gives you a stable connection layer that works across SQLite, Postgres, and MySQL.

pip install crewai sqlalchemy

Create a small SQLite file with sample rows. Keeping the demo local avoids network and credential overhead while you validate the tool logic.

from sqlalchemy import create_engine, text

engine = create_engine("sqlite:///example.db")
with engine.connect() as conn:
    conn.execute(text("CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)"))
    conn.execute(text("INSERT INTO users (name, age) VALUES ('Alice', 30), ('Bob', 25), ('Carol', 41)"))
    conn.commit()

Run this once. You now have example.db with three users.

Step 2: Define the SQL database tool

CrewAI expects tools to subclass BaseTool and declare a Pydantic schema for arguments. The agent will generate the query string, so you must constrain what the tool accepts and how it executes.

The implementation below rejects anything that isn’t a SELECT statement. In production you’d also use a read-only database role, but the prefix check stops accidental writes from a poorly prompted agent.

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from sqlalchemy import text, SQLAlchemyError

class SQLQueryInput(BaseModel):
    query: str = Field(..., description="A read-only SQL SELECT statement to execute.")

class SQLDatabaseTool(BaseTool):
    name: str = "sql_database_query"
        "Executes a read-only SQL SELECT query against the connected database "
        "and returns rows as a list of dictionaries."
    )
    args_schema: type[BaseModel] = SQLQueryInput
    engine: object = None

    def __init__(self, engine, **kwargs):
        super().__init__(**kwargs)
        self.engine = engine

    def _run(self, query: str) -> str:
        if not query.strip().lower().startswith("select"):
            return "Error: only SELECT queries are permitted."
        try:
            with self.engine.connect() as conn:
                result = conn.execute(text(query))
                rows = [dict(zip(result.keys(), row)) for row in result.fetchall()]
                return str(rows)
        except SQLAlchemyError as e:
            return f"Database error: {e}"

A few notes from shipping this in anger: return a string, not a raw Python object. CrewAI serializes tool output into the agent’s context window; a stringified list of dicts is predictable and token-efficient. If you expect large result sets, truncate or aggregate inside the tool before returning.

Step 3: Instantiate the tool with a live engine

The tool is stateless aside from the engine reference. Build the engine separately so you can swap SQLite for Postgres by changing one connection string.

from sqlalchemy import create_engine

engine = create_engine("sqlite:///example.db")
sql_tool = SQLDatabaseTool(engine=engine)

That’s the entire tool. The rest is standard CrewAI agent assembly.

Step 4: Build an agent and crew that uses the tool

Define an agent with a clear role and grant it the tool. The task description should push the agent to use the tool rather than guess.

from crewai import Agent, Task, Crew

analyst = Agent(
    role="Data Analyst",
    goal="Answer questions accurately using the SQL database tool",
    backstory="You write careful, valid SQL and never modify the database.",
    tools=[sql_tool],
    verbose=True
)

task = Task(
    description="How many users are older than 26? Use the SQL tool to find out.",
    expected_output="A short sentence stating the count.",
    agent=analyst
)

crew = Crew(agents=[analyst], tasks=[task])

In this crewai sql database tool tutorial we keep the crew minimal: one agent, one task. For a real app you’d add a planner agent or a formatter, but the tool integration is identical.

Step 5: Configure a resilient LLM endpoint

CrewAI delegates reasoning to an LLM. By default it uses OpenAI, but the LLM class accepts any OpenAI-compatible base URL. If you want fallback when a provider is rate-limited, point at a gateway that handles that transparently.

from crewai import LLM

llm = LLM(
    model="gpt-4o-mini",
    base_url="https://api.n4n.ai/v1",  # OpenAI-compatible, 240+ models, auto fallback
    api_key="your-key"
)

analyst = Agent(
    role="Data Analyst",
    goal="Answer questions accurately using the SQL database tool",
    backstory="You write careful, valid SQL and never modify the database.",
    tools=[sql_tool],
    llm=llm,
    verbose=True
)

n4n.ai exposes a single OpenAI-compatible endpoint covering 240+ models and applies automatic fallback when a provider is degraded, which keeps long agent loops from stalling on a 429. The per-token metering also makes it easy to attribute cost per crew run.

If you don’t need that, set OPENAI_API_KEY and omit base_url. The tool code doesn’t care which backend produces the SQL.

Step 6: Run and verify success

Execute the crew and print the result.

result = crew.kickoff()
print("Final answer:", result)

Verification is straightforward: the agent should emit a query like SELECT COUNT(*) FROM users WHERE age > 26 and return 2 (Alice and Carol). If you see a database error string from the tool, check that the table exists and the engine path is correct. Enable verbose=True to watch the agent’s tool calls in the logs—you’ll see the exact SQL it generated.

A second verification: open the database and run the same query manually to confirm the tool’s output matches ground truth. That closes the loop on trust.

Security and operational notes

• Read-only access: The prefix check is a guardrail, not a security boundary. Create a database user with SELECT only and use that connection string in production.
• Injection: The agent controls the query string. If untrusted users can prompt the agent, treat the tool as executable SQL on your data. Consider a strict schema allowlist or a natural-language-to-SQL layer with validation.
• Result size: Never return thousands of rows to the LLM. Add LIMIT in the tool or summarize inside _run.
• Connection pooling: For high-throughput crews, pass pool_size to create_engine and reuse the engine across tool instances.

Extending the tool

The pattern above is a starting point. Useful extensions we’ve shipped:

• Schema introspection: Add a list_tables and describe_table method so the agent can discover columns before querying.
• Parameterized queries: Instead of raw string SQL, accept a table and filters dict and build the statement with bound parameters.
• Cache hints: If your gateway honors provider cache-control, mark repetitive introspection queries as cacheable to cut token spend.

This crewai sql database tool tutorial gave you a working, safe integration. The same BaseTool skeleton applies to any external system—swap the SQLAlchemy engine for an API client and you’re done.