Implementation:CrewAIInc CrewAI RAG MySQL Loader
| Knowledge Sources | |
|---|---|
| Domains | RAG, Data_Loading, Database |
| Last Updated | 2026-02-11 00:00 GMT |
Overview
Executes SQL queries against MySQL databases and formats the results into structured readable text for RAG ingestion.
Description
MySQLLoader extends BaseLoader to connect to MySQL databases and execute queries. It requires a db_uri parameter passed via the metadata kwargs, supporting mysql and mysql+pymysql URI schemes.
The loader parses the connection URI using urlparse to extract host, port (default 3306), user, password, and database name. It uses pymysql with DictCursor for dictionary-based row access and utf8mb4 charset.
The query execution formats results as structured text:
- A header line listing all column names.
- A total row count.
- Row-by-row data with each column value on an indented line, filtering out None values.
Content is truncated at 100KB to prevent excessive memory usage. The returned LoaderResult includes metadata with the source query, database name, row count, and column names. Empty results are handled with a descriptive message. Proper connection cleanup is ensured via try/finally.
Usage
Import MySQLLoader when you need to load data from a MySQL database into the RAG knowledge base. It is typically instantiated automatically by the DataType.MYSQL registry.
Code Reference
Source Location
- Repository: CrewAI
- File: lib/crewai-tools/src/crewai_tools/rag/loaders/mysql_loader.py
- Lines: 1-102
Signature
class MySQLLoader(BaseLoader):
def load(self, source: SourceContent, **kwargs) -> LoaderResult: ...
Import
from crewai_tools.rag.loaders.mysql_loader import MySQLLoader
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| source | SourceContent | Yes | Wraps a SQL query string (e.g., "SELECT * FROM table_name") |
| metadata (via kwargs) | dict | Yes | Must contain db_uri (str) with MySQL connection URI |
Outputs
| Name | Type | Description |
|---|---|---|
| return | LoaderResult | Formatted text with column headers and row data; metadata includes source query, database name, row_count, and columns |
Usage Examples
Basic Usage
from crewai_tools.rag.loaders.mysql_loader import MySQLLoader
from crewai_tools.rag.source_content import SourceContent
loader = MySQLLoader()
source = SourceContent("SELECT id, name, email FROM users LIMIT 100")
result = loader.load(
source,
metadata={"db_uri": "mysql://user:password@localhost:3306/mydb"},
)
print(result.content)
# Columns: id, name, email
# Total rows: 100
#
# Row 1:
# id: 1
# name: Alice
# email: alice@example.com
# ...
print(result.metadata)
# {'source': 'SELECT ...', 'database': 'mydb', 'row_count': 100, 'columns': ['id', 'name', 'email']}