MySQL for Workflow

Store workflow sessions and run history in MySQL with MySQLDb.

MySQLDb stores a Workflow's sessions and run history in MySQL.

Usage

Start in a virtual environment.

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

uv pip install agno sqlalchemy "pymysql[rsa]" 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

Set your OpenAI API key:

Set OpenAI Key

Set your OPENAI_API_KEY as an environment variable. You can get one from OpenAI.

export OPENAI_API_KEY=sk-***
mysql_for_workflow.py
from agno.agent import Agent
from agno.db.mysql import MySQLDb
from agno.models.openai import OpenAIResponses
from agno.team import Team
from agno.tools.hackernews import HackerNewsTools
from agno.tools.websearch import WebSearchTools
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow

db_url = "mysql+pymysql://ai:ai@localhost:3306/ai"

db = MySQLDb(db_url=db_url)

hackernews_agent = Agent(
    name="HackerNews Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[HackerNewsTools()],
    role="Extract key insights and content from HackerNews posts",
)
web_agent = Agent(
    name="Web Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[WebSearchTools()],
    role="Search the web for the latest news and trends",
)

research_team = Team(
    name="Research Team",
    members=[hackernews_agent, web_agent],
    instructions="Research tech topics from HackerNews and the web",
)

content_planner = Agent(
    name="Content Planner",
    model=OpenAIResponses(id="gpt-5.2"),
    instructions=[
        "Plan a content schedule over 4 weeks for the provided topic and research content",
        "Ensure that I have posts for 3 posts per week",
    ],
)

research_step = Step(
    name="Research Step",
    team=research_team,
)

content_planning_step = Step(
    name="Content Planning Step",
    agent=content_planner,
)

if __name__ == "__main__":
    content_creation_workflow = Workflow(
        name="Content Creation Workflow",
        description="Automated content creation from blog posts to social media",
        db=db,
        steps=[research_step, content_planning_step],
    )
    content_creation_workflow.print_response(
        input="AI trends in 2024",
        markdown=True,
    )

Run the example

Save the code as mysql_for_workflow.py, start the database described above, and run:

python mysql_for_workflow.py

Parameters

ParameterTypeDefaultDescription
idOptional[str]NoneDatabase ID. Derived from the connection and schema when omitted.
db_engineOptional[Engine]NoneSQLAlchemy database engine. Provide this or db_url.
db_schemaOptional[str]NoneDatabase schema. Uses "ai" when omitted.
db_urlOptional[str]NoneDatabase URL. Provide this or db_engine.
session_tableOptional[str]NoneTable for Agent, Team, and Workflow sessions. Uses "agno_sessions" when omitted.
runs_tableOptional[str]NoneTable for session runs. Uses "agno_runs" with the default session table, or <session_table>_runs when session_table is customized.
memory_tableOptional[str]NoneTable for memories. Uses "agno_memories" when omitted.
metrics_tableOptional[str]NoneTable for metrics. Uses "agno_metrics" when omitted.
eval_tableOptional[str]NoneTable for evaluation run data. Uses "agno_eval_runs" when omitted.
knowledge_tableOptional[str]NoneTable for knowledge content. Uses "agno_knowledge" when omitted.
traces_tableOptional[str]NoneTable for traces. Uses "agno_traces" when omitted.
spans_tableOptional[str]NoneTable for spans. Uses "agno_spans" when omitted.
versions_tableOptional[str]NoneTable for schema versions. Uses "agno_schema_versions" when omitted.
create_schemaboolTrueCreate the database schema when it does not exist. Set to False when migrations manage it.