Implementation:CrewAIInc CrewAI NL2SQL Tool
| Knowledge Sources | |
|---|---|
| Domains | Tools, Database, SQL |
| Last Updated | 2026-02-11 00:00 GMT |
Overview
NL2SQLTool enables CrewAI agents to execute SQL queries against a database, supporting natural language to SQL conversion workflows.
Description
The NL2SQLTool extends BaseTool and uses SQLAlchemy for database connectivity, guarded by a graceful SQLALCHEMY_AVAILABLE check. On initialization via model_post_init, it introspects the database schema by querying information_schema.tables for available tables and information_schema.columns for each table's column metadata, populating self.tables and self.columns attributes. Input is validated through the NL2SQLToolInput Pydantic schema which expects a sql_query string. When _run is called, it executes the SQL query via execute_sql, which creates a SQLAlchemy engine and session, runs the query with session.execute(text(sql_query)), commits the transaction, and returns results as a list of dictionaries for row-returning queries or a success message for non-returning queries. On error, it rolls back the transaction and provides the agent with schema context (tables and columns) along with the error message, enabling the agent to self-correct and retry with a corrected query.
Usage
Use this tool when a CrewAI agent needs to interact directly with a relational database by executing SQL queries, particularly in workflows where natural language is converted to SQL statements by the agent.
Code Reference
Source Location
- Repository: CrewAI
- File: lib/crewai-tools/src/crewai_tools/tools/nl2sql/nl2sql_tool.py
- Lines: 1-97
Signature
class NL2SQLToolInput(BaseModel):
sql_query: str = Field(title="SQL Query", description="The SQL query to execute.")
class NL2SQLTool(BaseTool):
name: str = "NL2SQLTool"
description: str = "Converts natural language to SQL queries and executes them."
db_uri: str = Field(title="Database URI", description="The URI of the database to connect to.")
tables: list = Field(default_factory=list)
columns: dict = Field(default_factory=dict)
args_schema: type[BaseModel] = NL2SQLToolInput
def model_post_init(self, __context: Any) -> None: ...
def _run(self, sql_query: str): ...
def execute_sql(self, sql_query: str) -> list | str: ...
Import
from crewai_tools import NL2SQLTool
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| db_uri | str | Yes | SQLAlchemy-compatible database connection URI |
| sql_query | str | Yes | The SQL query to execute against the database |
Outputs
| Name | Type | Description |
|---|---|---|
| _run() returns | list or str | List of row dictionaries for SELECT queries; success message string for non-returning queries; error context string with schema info on failure |
Usage Examples
Basic Usage
from crewai_tools import NL2SQLTool
tool = NL2SQLTool(db_uri="postgresql://user:password@localhost:5432/mydb")
result = tool._run(sql_query="SELECT * FROM users LIMIT 10")
Schema Introspection
from crewai_tools import NL2SQLTool
tool = NL2SQLTool(db_uri="postgresql://user:password@localhost:5432/mydb")
# After initialization, tables and columns are auto-populated
print(tool.tables) # Available tables
print(tool.columns) # Column metadata per table