Async MySQL for Agent

Agno supports using MySQL asynchronously, with the AsyncMySQLDb class.

Usage

Run MySQL

Install docker desktop and run MySQL on port 3306 using:

docker run -d \
  --name mysql \
  -e MYSQL_ROOT_PASSWORD=ai \
  -e MYSQL_DATABASE=ai \
  -e MYSQL_USER=ai \
  -e MYSQL_PASSWORD=ai \
  -p 3306:3306 \
  -d mysql:8

async_mysql_for_agent.py

import asyncio
import uuid

from agno.agent import Agent
from agno.db.base import SessionType
from agno.db.mysql import AsyncMySQLDb
from agno.tools.hackernews import HackerNewsTools

db_url = "mysql+asyncmy://ai:ai@localhost:3306/ai"
db = AsyncMySQLDb(db_url=db_url)

agent = Agent(
    db=db,
    tools=[HackerNewsTools()],
    add_history_to_context=True,
    add_datetime_to_context=True,
)

async def main():
    """Run the agent queries in the same event loop"""
    session_id = str(uuid.uuid4())
    await agent.aprint_response(
        "How many people live in Canada?", session_id=session_id
    )
    await agent.aprint_response(
        "What is their national anthem called?", session_id=session_id
    )
    session_data = await db.get_session(
        session_id=session_id, session_type=SessionType.AGENT
    )
    print("\n=== SESSION DATA ===")
    print(session_data.to_dict())

if __name__ == "__main__":
    asyncio.run(main())

Params

Parameter	Type	Default	Description
id	Optional[str]	-	The ID of the database instance. UUID by default.
db_url	Optional[str]	-	The database URL to connect to.
db_engine	Optional[AsyncEngine]	-	The SQLAlchemy async database engine to use.
db_schema	Optional[str]	-	The database schema to use.
session_table	Optional[str]	-	Name of the table to store Agent, Team and Workflow sessions.
memory_table	Optional[str]	-	Name of the table to store memories.
metrics_table	Optional[str]	-	Name of the table to store metrics.
eval_table	Optional[str]	-	Name of the table to store evaluation runs data.
knowledge_table	Optional[str]	-	Name of the table to store knowledge content.