# Async Postgres for Workflow (/database/providers/async-postgres/usage/async-postgres-for-workflow) `AsyncPostgresDb` stores a Workflow's sessions and run history asynchronously in [PostgreSQL](https://www.postgresql.org/). Use async Workflow methods such as `arun()` and `aprint_response()`. ## Usage [#usage] Start in a [virtual environment](/sdk/setup) and set the model key before running the example. ## Set OpenAI Key [#set-openai-key] Set your `OPENAI_API_KEY` as an environment variable. You can get one [from OpenAI](https://platform.openai.com/account/api-keys). Mac Windows PowerShell ```bash export OPENAI_API_KEY=sk-*** ``` ```powershell $env:OPENAI_API_KEY = "your-openai-api-key" ``` Install the `sqlalchemy`, `psycopg`, `openai`, and `ddgs` packages: ```shell uv pip install agno "sqlalchemy[asyncio]" "psycopg[binary]" openai ddgs ``` ### Run PostgreSQL [#run-postgresql] Install [Docker Desktop](https://docs.docker.com/get-started/get-docker/), then start PostgreSQL with pgvector on port `5532`: ```bash docker run -d \ -e POSTGRES_DB=ai \ -e POSTGRES_USER=ai \ -e POSTGRES_PASSWORD=ai \ -e PGDATA=/var/lib/postgresql \ -v pgvolume:/var/lib/postgresql \ -p 5532:5432 \ --name pgvector \ agnohq/pgvector:18 ``` ```python title="async_postgres_for_workflow.py" import asyncio from agno.agent import Agent from agno.db.postgres import AsyncPostgresDb 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 = "postgresql+psycopg_async://ai:ai@localhost:5532/ai" db = AsyncPostgresDb(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, ) async def 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], ) try: await content_creation_workflow.aprint_response( input="AI trends in 2024", markdown=True, ) finally: await db.close() if __name__ == "__main__": asyncio.run(main()) ``` ## Parameters [#parameters] | Parameter | Type | Default | Description | | ------------------------ | ----------------------- | ------- | ---------------------------------------------------------------------------------------------------------------- | | `id` | `Optional[str]` | - | The ID of the database instance. UUID by default. | | `db_url` | `Optional[str]` | - | The database URL to connect to. | | `db_engine` | `Optional[AsyncEngine]` | - | The SQLAlchemy async database engine to use. | | `db_schema` | `Optional[str]` | - | The database schema to use. | | `session_table` | `Optional[str]` | - | Name of the table to store Agent, Team and Workflow sessions. | | `runs_table` | `Optional[str]` | - | Name of the table to store the runs of each session. | | `memory_table` | `Optional[str]` | - | Name of the table to store memories. | | `metrics_table` | `Optional[str]` | - | Name of the table to store metrics. | | `eval_table` | `Optional[str]` | - | Name of the table to store evaluation runs data. | | `knowledge_table` | `Optional[str]` | - | Name of the table to store knowledge content. | | `traces_table` | `Optional[str]` | - | Name of the table to store traces. | | `spans_table` | `Optional[str]` | - | Name of the table to store spans. | | `versions_table` | `Optional[str]` | - | Name of the table to store schema versions. | | `components_table` | `Optional[str]` | - | Name of the table to store components. | | `learnings_table` | `Optional[str]` | - | Name of the table to store learnings. | | `schedules_table` | `Optional[str]` | - | Name of the table to store cron schedules. | | `schedule_runs_table` | `Optional[str]` | - | Name of the table to store schedule run history. | | `job_table` | `Optional[str]` | - | Name of the table to store durable background run jobs. | | `approvals_table` | `Optional[str]` | - | Name of the table to store human approval requests. | | `auth_tokens_table` | `Optional[str]` | - | Name of the table to store OAuth tokens for external services. | | `service_accounts_table` | `Optional[str]` | - | Name of the table to store service accounts. | | `create_schema` | `bool` | `True` | Whether to create the database schema if it doesn't exist. Set to `False` when the schema is managed externally. | ## Run the Example [#run-the-example] Save the code as `async_postgres_for_workflow.py`, complete the prerequisites above, then run: ```bash python async_postgres_for_workflow.py ```