Async MySQL for Workflow

Store workflow sessions and run history asynchronously in MySQL with AsyncMySQLDb.

AsyncMySQLDb stores a Workflow's sessions and run history asynchronously in MySQL. Use async Workflow methods such as arun() and aprint_response().

Usage

Install the sqlalchemy, asyncmy, openai, and ddgs packages:

uv pip install sqlalchemy asyncmy openai ddgs

Run MySQL

Install Docker Desktop, then start MySQL on port 3306:

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
async_mysql_for_workflow.py
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.websearch import WebSearchTools
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


researcher = Agent(
    name="Researcher",
    tools=[WebSearchTools()],
    instructions="Research the topic and return key points and a summary",
    output_schema=ResearchTopic,
)

writer = Agent(
    name="Writer",
    instructions="Write a blog post based on the research provided",
)


async def blog_workflow(workflow: Workflow, execution_input: WorkflowExecutionInput):
    topic = execution_input.input

    research_result = await researcher.arun(f"Research this topic: {topic}")

    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"


workflow = Workflow(
    name="Blog Generator",
    steps=blog_workflow,
    db=db,
)


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


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

Parameters

ParameterTypeDefaultDescription
idOptional[str]-The ID of the database instance. UUID by default.
db_urlOptional[str]-The database URL to connect to.
db_engineOptional[AsyncEngine]-The SQLAlchemy async database engine to use.
db_schemaOptional[str]-The database schema to use.
session_tableOptional[str]-Name of the table to store Agent, Team and Workflow sessions.
runs_tableOptional[str]NoneStorage name for individual runs. Defaults to agno_runs, or <session_table>_runs when a custom session name is supplied.
memory_tableOptional[str]-Name of the table to store memories.
metrics_tableOptional[str]-Name of the table to store metrics.
eval_tableOptional[str]-Name of the table to store evaluation runs data.
knowledge_tableOptional[str]-Name of the table to store knowledge content.
traces_tableOptional[str]-Name of the table to store traces.
spans_tableOptional[str]-Name of the table to store spans.
versions_tableOptional[str]-Name of the table to store schema versions.
create_schemaboolTrueWhether to create the database schema if it doesn't exist. Set to False when the schema is managed externally.