Async MySQL for Agent

uv pip install sqlalchemy asyncmy openai ddgs

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 \
  mysql:8

import asyncio
import uuid

from agno.agent import Agent
from agno.db.base import SessionType
from agno.db.mysql import AsyncMySQLDb
from agno.tools.websearch import WebSearchTools

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

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


async def main():
    try:
        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())
    finally:
        await db.close()


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