MySQL Plugin

Async MySQL/MariaDB operations with automatic connection pooling built on aiomysql.

Installation

pip install aiomysql

Quick Start

Direct Usage (Scripts)

from daita.plugins import mysql

# Direct usage in scripts
async with mysql(
    host="localhost",
    database="mydb",
    username="user",
    password="password"
) as db:
    results = await db.query("SELECT * FROM users")
    print(results)

Agent Integration (Recommended)

from daita import SubstrateAgent
from daita.plugins import mysql

# Create plugin
db = mysql(
    host="localhost",
    database="mydb",
    username="user",
    password="password"
)

# Agent uses database tools autonomously
agent = SubstrateAgent(
    name="DB Agent",
    prompt="You are a database analyst. Help users query and analyze data.",
    tools=[db]
)

await agent.start()
result = await agent.run("Show me all users")

Connection Parameters

mysql(
    host: str = "localhost",
    port: int = 3306,
    database: str = "",
    username: str = "",
    password: str = "",
    connection_string: Optional[str] = None,
    min_size: int = 1,
    max_size: int = 10,
    charset: str = "utf8mb4",
    autocommit: bool = True,
    **kwargs
)

Parameters

• host (str): Database host address (default: "localhost")
• port (int): Database port (default: 3306)
• database (str): Database name to connect to
• username (str): Username for authentication (note: use username not user)
• password (str): Password for authentication
• connection_string (str): Full MySQL connection string (overrides individual params)
• min_size (int): Minimum connections in pool (default: 1)
• max_size (int): Maximum connections in pool (default: 10)
• charset (str): Character set (default: "utf8mb4")
• autocommit (bool): Enable autocommit mode (default: True)
• **kwargs: Additional aiomysql configuration

Connection Methods

# Individual parameters
async with mysql(
    host="localhost",
    database="ecommerce",
    username="app_user",
    password="secure_password"
) as db:
    results = await db.query("SELECT * FROM products")

# Connection string
async with mysql(
    connection_string="mysql://user:pass@localhost:3306/ecommerce"
) as db:
    results = await db.query("SELECT * FROM products")

# Advanced configuration
async with mysql(
    host="localhost",
    database="mydb",
    username="user",
    password="password",
    charset="utf8mb4",      # Full Unicode support
    autocommit=True,
    min_size=5,
    max_size=20
) as db:
    results = await db.query("SELECT * FROM users")

Query Operations

SELECT queries - Use query() method, returns list of dictionaries:

async with mysql(host="localhost", database="app") as db:
    # Simple query
    users = await db.query("SELECT id, name, email FROM users")
    # Returns: [{"id": 1, "name": "John", "email": "john@example.com"}, ...]

    # Parameterized query (use %s, %s, %s, etc.)
    active_users = await db.query(
        "SELECT * FROM users WHERE status = %s AND created_at > %s",
        ["active", "2024-01-01"]
    )

Parameter syntax: MySQL uses %s for all types (not $1 like PostgreSQL)

Data Modification

INSERT, UPDATE, DELETE - Use execute() method, returns row count:

async with mysql(host="localhost", database="app") as db:
    # Insert
    await db.execute(
        "INSERT INTO users (name, email, status) VALUES (%s, %s, %s)",
        ["Jane Doe", "jane@example.com", "active"]
    )

    # Update
    await db.execute("UPDATE users SET last_login = NOW() WHERE id = %s", [user_id])

    # Delete
    await db.execute("DELETE FROM users WHERE status = %s", ["inactive"])

Bulk Operations & Schema

Bulk insert:

async with mysql(host="localhost", database="app") as db:
    users_data = [
        {"name": "Alice", "email": "alice@example.com"},
        {"name": "Bob", "email": "bob@example.com"}
    ]
    count = await db.insert_many("users", users_data)

List tables:

async with mysql(host="localhost", database="app") as db:
    tables = await db.tables()
    # Returns: ["users", "orders", "products"]

Describe table:

async with mysql(host="localhost", database="app") as db:
    schema = await db.describe("users")
    # Returns: [{"column_name": "id", "data_type": "int", ...}, ...]

Using with Agents

Tool-Based Integration (Recommended)

MySQL plugin exposes database operations as tools that agents can use autonomously:

from daita import SubstrateAgent
from daita.plugins import mysql
import os

# Create database plugin
db = mysql(
    host="localhost",
    database="ecommerce",
    username=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD")
)

# Pass plugin to agent - agent can now use database tools autonomously
agent = SubstrateAgent(
    name="Product Analyst",
    prompt="You are a product analyst. Help users analyze inventory and sales data.",
    llm_provider="openai",
    model="gpt-4",
    tools=[db]
)

await agent.start()

# Agent autonomously uses database tools to answer questions
result = await agent.run("Show me products that need restocking (inventory < 10)")

# The agent will autonomously:
# 1. Use list_tables to explore database
# 2. Use get_table_schema to understand structure
# 3. Use query_database to fetch the data
# 4. Analyze and present results in natural language

await agent.stop()

Direct Database Operations (Scripts)

For scripts that don't need agent capabilities:

from daita.plugins import mysql

async with mysql(
    host="localhost",
    database="analytics",
    username="analyst",
    password="password"
) as db:
    # Direct queries
    user = await db.query("SELECT * FROM users WHERE id = %s", [user_id])
    activity = await db.query(
        "SELECT DATE(created_at) as date, COUNT(*) as count "
        "FROM user_events WHERE user_id = %s GROUP BY DATE(created_at)",
        [user_id]
    )

    print(f"User: {user[0] if user else None}")
    print(f"Activity: {activity}")

Available Tools

The MySQL plugin exposes these tools to LLM agents:

Tool	Description	Parameters
query_database	Execute SELECT queries	sql (required), params (optional array)
execute_sql	Run INSERT/UPDATE/DELETE	sql (required), params (optional array)
list_tables	List all tables	None
get_table_schema	Get table column info	table_name (required)

Tool Categories: database Tool Source: plugin Parameter Syntax: Uses %s for all parameter placeholders

Tool Usage Example

from daita import SubstrateAgent
from daita.plugins import mysql

# Setup database with tool integration
db = mysql(host="localhost", database="inventory")

agent = SubstrateAgent(
    name="Inventory Manager",
    prompt="You are an inventory manager. Help users track and manage inventory.",
    llm_provider="openai",
    model="gpt-4",
    tools=[db]
)

await agent.start()

# Natural language query - agent uses tools autonomously
result = await agent.run("""
Check inventory status and identify:
1. Products below reorder point
2. Top 5 selling products this month
3. Products with zero stock
""")

# Agent orchestrates tool calls autonomously to answer the query
print(result)
await agent.stop()

Error Handling

try:
    async with mysql(host="localhost", database="mydb") as db:
        results = await db.query("SELECT * FROM users WHERE id = %s", [user_id])
except RuntimeError as e:
    if "aiomysql not installed" in str(e):
        print("Install aiomysql: pip install aiomysql")
    elif "connection" in str(e).lower():
        print(f"Connection failed: {e}")

MySQL-Specific Features

Character sets - Use utf8mb4 for full Unicode support (including emojis):

async with mysql(
    host="localhost",
    database="mydb",
    charset="utf8mb4"
) as db:
    await db.execute("INSERT INTO posts (content) VALUES (%s)", ["Hello 👋"])

Autocommit - Default is True (each statement commits immediately)

Best Practices

Connection Management:

• Always use context managers (async with) for automatic cleanup
• Configure pool size based on workload (min_size, max_size)
• Use utf8mb4 charset for full Unicode support

Security:

• Use parameterized queries to prevent SQL injection (%s placeholders)
• Store credentials in environment variables, never hardcode
• Use read-only accounts when possible
• Grant minimal privileges to application users

Performance:

• Use insert_many() for bulk inserts instead of loops
• Add indexes for frequently queried columns
• Limit result sets to avoid memory issues

PostgreSQL vs MySQL Differences

Feature	PostgreSQL	MySQL
Parameter syntax	$1, $2, $3	%s, %s, %s
Auto-increment	SERIAL	AUTO_INCREMENT
Boolean type	BOOLEAN	TINYINT(1)
String concat	|| operator	CONCAT() function

Troubleshooting

Issue	Solution
aiomysql not installed	pip install aiomysql
Connection refused	Check MySQL is running, verify host/port/credentials
Access denied	Verify username/password, check user privileges
Character encoding errors	Use charset="utf8mb4" in connection
Too many connections	Reduce max_size in pool, check for connection leaks

Next Steps

• PostgreSQL Plugin - PostgreSQL database operations
• MongoDB Plugin - Document database operations
• REST Plugin - API integrations
• Plugin Overview - All available plugins