Async MongoDB for Workflow

Agno supports using MongoDB asynchronously as a storage backend for Workflows, with the AsyncMongoDb class.

Usage

Install the pymongo (4.9 or later), openai, and ddgs packages:

uv pip install "pymongo>=4.9" openai ddgs

motor clients are also supported, but motor is deprecated. Use PyMongo’s async client instead.

Run MongoDB

Install docker desktop and run MongoDB on port 27017 using:

docker run -d \
  --name local-mongo \
  -p 27017:27017 \
  -e MONGO_INITDB_ROOT_USERNAME=mongoadmin \
  -e MONGO_INITDB_ROOT_PASSWORD=secret \
  mongo

async_mongodb_for_workflow.py

import asyncio

from agno.agent import Agent
from agno.db.mongo import AsyncMongoDb
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 = "mongodb://mongoadmin:secret@localhost:27017"

db = AsyncMongoDb(db_url=db_url)

# Define agents
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",
)

# Define research team for complex analysis
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",
    ],
)

# Define steps
research_step = Step(
    name="Research Step",
    team=research_team,
)

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

# Create and use workflow
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],
    )
    asyncio.run(
        content_creation_workflow.aprint_response(
            input="AI trends in 2024",
            markdown=True,
        )
    )

Params

Parameter	Type	Default	Description
id	Optional[str]	-	The ID of the database instance. UUID by default.
db_client	Optional[Union[AsyncIOMotorClient, AsyncMongoClient]]	-	The MongoDB async client to use. Supports Motor and PyMongo async clients.
db_name	Optional[str]	-	The name of the database to use.
db_url	Optional[str]	-	The database URL to connect to.
session_collection	Optional[str]	-	Name of the collection to store sessions.
memory_collection	Optional[str]	-	Name of the collection to store memories.
metrics_collection	Optional[str]	-	Name of the collection to store metrics.
eval_collection	Optional[str]	-	Name of the collection to store evaluation runs.
knowledge_collection	Optional[str]	-	Name of the collection to store knowledge documents.
culture_collection	Optional[str]	-	Name of the collection to store cultural knowledge.
traces_collection	Optional[str]	-	Name of the collection to store traces.
spans_collection	Optional[str]	-	Name of the collection to store spans.
learnings_collection	Optional[str]	-	Name of the collection to store learnings.
schedules_collection	Optional[str]	-	Name of the collection to store cron schedules.
schedule_runs_collection	Optional[str]	-	Name of the collection to store schedule run history.