feat: add AsyncMySQLDb (#5425)

## Summary

Describe key changes, mention related issues or motivation for the
changes.
Issue number: #5427
Docs Change PR: [PR](agno-agi/agno-docs#266)


## Type of change

- [ ] Bug fix
- [X] New feature
- [ ] Breaking change
- [ ] Improvement
- [ ] Model update
- [ ] Other:

---

## Checklist

- [X] Code complies with style guidelines
- [X] Ran format/validation scripts (`./scripts/format.sh` and
`./scripts/validate.sh`)
- [X] Self-review completed
- [X] Documentation updated (comments, docstrings)
- [X] Examples and guides: Relevant cookbook examples have been included
or updated (if applicable)
- [X] Tested in clean environment
- [X] Tests added/updated (if applicable)

---

## Additional Notes

Add any important context (deployment instructions, screenshots,
security considerations, etc.)

---------

Co-authored-by: Mustafa Esoofally <[email protected]>
Co-authored-by: Dirk Brand <[email protected]>
Co-authored-by: Yash Pratap Solanky <[email protected]>
Co-authored-by: ysolanky <[email protected]>

Author: 4 people

cookbook/agent_os/dbs/async_mysql_demo.py

@@ -0,0 +1,50 @@
+"""Example showing how to use AgentOS with a SQLite database"""
+
+from agno.agent import Agent
+from agno.db.mysql import AsyncMySQLDb
+from agno.models.openai import OpenAIChat
+from agno.os import AgentOS
+from agno.team.team import Team
+
+db_url = "mysql+asyncmy://ai:ai@localhost:3306/ai"
+
+db = AsyncMySQLDb(
+    id="mysql-demo",
+    db_url=db_url,
+    session_table="sessions",
+    eval_table="eval_runs",
+    memory_table="user_memories",
+    metrics_table="metrics",
+)
+
+
+# Setup a basic agent and a basic team
+basic_agent = Agent(
+    name="Basic Agent",
+    id="basic-agent",
+    model=OpenAIChat(id="gpt-4o"),
+    db=db,
+    enable_user_memories=True,
+    enable_session_summaries=True,
+    add_history_to_context=True,
+    num_history_runs=3,
+    add_datetime_to_context=True,
+    markdown=True,
+)
+team_agent = Team(
+    id="basic-team",
+    name="Team Agent",
+    model=OpenAIChat(id="gpt-4o"),
+    db=db,
+    members=[basic_agent],
+)
+
+agent_os = AgentOS(
+    description="Example OS setup",
+    agents=[basic_agent],
+    teams=[team_agent],
+)
+app = agent_os.get_app()
+
+if __name__ == "__main__":
+    agent_os.serve(app="async_mysql_demo:app", reload=True)

cookbook/db/mysql/README.md

@@ -1,28 +1,84 @@
 # MySQL Integration
 
-Examples demonstrating MySQL database integration with Agno agents and teams.
+Examples demonstrating MySQL database integration with Agno agents, teams, and workflows.
 
 ## Setup
 
+### Synchronous MySQL
+
 ```shell
-pip install mysql-connector-python
+pip install mysql-connector-python sqlalchemy pymysql
+```
+
+### Asynchronous MySQL
+
+```shell
+pip install sqlalchemy asyncmy
 ```
 
 ## Configuration
 
+### Synchronous MySQL
+
+```python
+from agno.agent import Agent
+from agno.db.mysql import MySQLDb
+
+db = MySQLDb(db_url="mysql+pymysql://username:password@localhost:3306/database")
+
+agent = Agent(
+    db=db,
+    add_history_to_context=True,
+)
+```
+
+### Asynchronous MySQL
+
 ```python
+import asyncio
 from agno.agent import Agent
-from agno.db.mysql import MySqlDb
+from agno.db.mysql import AsyncMySQLDb
 
-db = MySqlDb(db_url="mysql+pymysql://username:password@localhost:3306/database")
+db = AsyncMySQLDb(db_url="mysql+asyncmy://username:password@localhost:3306/database")
 
 agent = Agent(
     db=db,
     add_history_to_context=True,
 )
+
+asyncio.run(agent.aprint_response("Hello!"))
 ```
 
-## Examples
+## Synchronous Examples
 
 - [`mysql_for_agent.py`](mysql_for_agent.py) - Agent with MySQL storage
 - [`mysql_for_team.py`](mysql_for_team.py) - Team with MySQL storage
+
+## Asynchronous Examples
+
+- [`async_mysql/async_mysql_for_agent.py`](async_mysql/async_mysql_for_agent.py) - Agent with Async MySQL storage
+- [`async_mysql/async_mysql_for_team.py`](async_mysql/async_mysql_for_team.py) - Team with Async MySQL storage
+- [`async_mysql/async_mysql_for_workflow.py`](async_mysql/async_mysql_for_workflow.py) - Workflow with Async MySQL storage
+
+## Database URL Format
+
+### Synchronous Drivers
+
+- **PyMySQL**: `mysql+pymysql://user:password@host:port/database`
+- **MySQL Connector/Python**: `mysql+mysqlconnector://user:password@host:port/database`
+
+### Asynchronous Drivers
+
+- **asyncmy**: `mysql+asyncmy://user:password@host:port/database`
+
+## Async vs Sync
+
+Choose **AsyncMySQLDb** when:
+- Building high-concurrency applications
+- Working with async frameworks (FastAPI, Sanic, etc.)
+- Need non-blocking database operations
+
+Choose **MySQLDb** when:
+- Building traditional synchronous applications
+- Simpler deployment requirements
+- Working with sync-only libraries

cookbook/db/mysql/async_mysql/__init__.py

@@ -0,0 +1,2 @@
+# Async MySQL integration examples
+

cookbook/db/mysql/async_mysql/async_mysql_for_agent.py

@@ -0,0 +1,41 @@
+"""Use Async MySQL as the database for an agent.
+Run `pip install openai duckduckgo-search sqlalchemy asyncmy agno` to install dependencies.
+"""
+
+import asyncio
+import uuid
+
+from agno.agent import Agent
+from agno.db.base import SessionType
+from agno.db.mysql import AsyncMySQLDb
+from agno.tools.duckduckgo import DuckDuckGoTools
+
+db_url = "mysql+asyncmy://ai:ai@localhost:3306/ai"
+db = AsyncMySQLDb(db_url=db_url)
+
+agent = Agent(
+    db=db,
+    tools=[DuckDuckGoTools()],
+    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())

cookbook/db/mysql/async_mysql/async_mysql_for_team.py

@@ -0,0 +1,70 @@
+"""Use Async MySQL as the database for a team.
+Run `pip install openai duckduckgo-search newspaper4k lxml_html_clean agno sqlalchemy asyncmy` to install the dependencies
+"""
+
+import asyncio
+import uuid
+from typing import List
+
+from agno.agent import Agent
+from agno.db.base import SessionType
+from agno.db.mysql import AsyncMySQLDb
+from agno.team import Team
+from agno.tools.duckduckgo import DuckDuckGoTools
+from agno.tools.hackernews import HackerNewsTools
+from pydantic import BaseModel
+
+db_url = "mysql+asyncmy://ai:ai@localhost:3306/ai"
+db = AsyncMySQLDb(db_url=db_url)
+
+
+class Article(BaseModel):
+    title: str
+    summary: str
+    reference_links: List[str]
+
+
+hn_researcher = Agent(
+    name="HackerNews Researcher",
+    role="Gets top stories from hackernews.",
+    tools=[HackerNewsTools()],
+)
+
+web_searcher = Agent(
+    name="Web Searcher",
+    role="Searches the web for information on a topic",
+    tools=[DuckDuckGoTools()],
+    add_datetime_to_context=True,
+)
+
+
+hn_team = Team(
+    name="HackerNews Team",
+    members=[hn_researcher, web_searcher],
+    db=db,
+    instructions=[
+        "First, search hackernews for what the user is asking about.",
+        "Then, ask the web searcher to search for each story to get more information.",
+        "Finally, provide a thoughtful and engaging summary.",
+    ],
+    output_schema=Article,
+    markdown=True,
+    show_members_responses=True,
+)
+
+
+async def main():
+    """Run the agent queries in the same event loop"""
+    session_id = str(uuid.uuid4())
+    await hn_team.aprint_response(
+        "Write an article about the top 2 stories on hackernews", session_id=session_id
+    )
+    session_data = await db.get_session(
+        session_id=session_id, session_type=SessionType.TEAM
+    )
+    print("\n=== SESSION DATA ===")
+    print(session_data.to_dict())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

cookbook/db/mysql/async_mysql/async_mysql_for_workflow.py

@@ -0,0 +1,87 @@
+"""Use Async MySQL as the database for a workflow.
+Run `pip install openai duckduckgo-search sqlalchemy asyncmy agno` to install dependencies.
+"""
+
+import asyncio
+import uuid
+from typing import List
+
+from agno.agent import Agent
+from agno.db.base import SessionType
+from agno.db.mysql import AsyncMySQLDb
+from agno.tools.duckduckgo import DuckDuckGoTools
+from agno.workflow.types import WorkflowExecutionInput
+from agno.workflow.workflow import Workflow
+from pydantic import BaseModel
+
+db_url = "mysql+asyncmy://ai:ai@localhost:3306/ai"
+db = AsyncMySQLDb(db_url=db_url)
+
+
+class ResearchTopic(BaseModel):
+    topic: str
+    key_points: List[str]
+    summary: str
+
+
+# Create researcher agent
+researcher = Agent(
+    name="Researcher",
+    tools=[DuckDuckGoTools()],
+    instructions="Research the given topic thoroughly and provide key insights",
+    output_schema=ResearchTopic,
+)
+
+# Create writer agent
+writer = Agent(
+    name="Writer",
+    instructions="Write a well-structured blog post based on the research provided",
+)
+
+
+# Define the workflow
+async def blog_workflow(workflow: Workflow, execution_input: WorkflowExecutionInput):
+    """
+    A workflow that researches a topic and writes a blog post about it.
+    """
+    topic = execution_input.input
+
+    # Step 1: Research the topic
+    research_result = await researcher.arun(f"Research this topic: {topic}")
+
+    # Step 2: Write the blog post
+    if research_result and research_result.content:
+        blog_result = await writer.arun(
+            f"Write a blog post about {topic}. Use this research: {research_result.content.model_dump_json()}"
+        )
+        return blog_result.content
+
+    return "Failed to complete workflow"
+
+
+# Create and run the workflow
+workflow = Workflow(
+    name="Blog Generator",
+    steps=blog_workflow,
+    db=db,
+)
+
+
+async def main():
+    """Run the workflow with a sample topic"""
+    session_id = str(uuid.uuid4())
+
+    await workflow.aprint_response(
+        input="The future of artificial intelligence",
+        session_id=session_id,
+        markdown=True,
+    )
+    session_data = await db.get_session(
+        session_id=session_id, session_type=SessionType.WORKFLOW
+    )
+    print("\n=== SESSION DATA ===")
+    print(session_data.to_dict())
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

libs/agno/agno/db/mysql/__init__.py

@@ -1,3 +1,4 @@
+from agno.db.mysql.async_mysql import AsyncMySQLDb
 from agno.db.mysql.mysql import MySQLDb
 
-__all__ = ["MySQLDb"]
+__all__ = ["MySQLDb", "AsyncMySQLDb"]