DynamoDB for Workflow

Store workflow runs in DynamoDB with DynamoDb.

DynamoDb stores a Workflow's sessions and run history in DynamoDB.

Usage

Start in a virtual environment and set the model key before running the example.

Set OpenAI Key

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

export OPENAI_API_KEY=sk-***

Pass a configured DynamoDB client as db_client. Otherwise, set the AWS_REGION, AWS_ACCESS_KEY_ID, and AWS_SECRET_ACCESS_KEY environment variables or pass region_name, aws_access_key_id, and aws_secret_access_key directly.

On first use, DynamoDb creates missing tables with provisioned capacity. The client or credentials must allow table creation and read/write operations.

Install dependencies:

uv pip install agno boto3 openai ddgs
dynamo_for_workflow.py
from agno.agent import Agent
from agno.db.dynamo import DynamoDb
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 = DynamoDb()

# 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 there are 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],
    )
    content_creation_workflow.print_response(
        input="Recent AI trends",
        markdown=True,
    )

Parameters

ParameterTypeDefaultDescription
idOptional[str]-Database ID. Derived deterministically from explicit client or constructor region/key values when omitted.
db_clientOptional[Any]-The DynamoDB client to use.
region_nameOptional[str]-AWS region name.
aws_access_key_idOptional[str]-AWS access key ID.
aws_secret_access_keyOptional[str]-AWS secret access key.
session_tableOptional[str]-The name of the session table.
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]-The name of the memory table.
metrics_tableOptional[str]-The name of the metrics table.
eval_tableOptional[str]-The name of the eval table.
knowledge_tableOptional[str]-The name of the knowledge table.
traces_tableOptional[str]-The name of the traces table.
spans_tableOptional[str]-The name of the spans table.

Run the Example

Save the code as dynamo_for_workflow.py, complete the prerequisites above, then run:

python dynamo_for_workflow.py