AutoGen Tutorial (Python): connecting to PostgreSQL for advanced developers

This tutorial shows how to wire an AutoGen agent to PostgreSQL in Python so the agent can read and write structured data through a real tool layer. You’d use this when your assistant needs durable state, transactional lookups, or access to business data that should not live in prompt memory.

What You'll Need

• •Python 3.10+
• •A running PostgreSQL instance
• •A PostgreSQL database and credentials
• •OPENAI_API_KEY
• •Python packages:
  • •pyautogen
  • •psycopg2-binary
  • •python-dotenv

Install them with:

pip install pyautogen psycopg2-binary python-dotenv

Step-by-Step

1. •Start by defining your environment variables and a small schema. For production work, keep credentials out of code and create a table that the agent can query safely.

import os
from dotenv import load_dotenv
import psycopg2

load_dotenv()

DB_HOST = os.getenv("POSTGRES_HOST", "localhost")
DB_NAME = os.getenv("POSTGRES_DB", "autogen_demo")
DB_USER = os.getenv("POSTGRES_USER", "postgres")
DB_PASSWORD = os.getenv("POSTGRES_PASSWORD", "postgres")
DB_PORT = int(os.getenv("POSTGRES_PORT", "5432"))

conn = psycopg2.connect(
    host=DB_HOST,
    dbname=DB_NAME,
    user=DB_USER,
    password=DB_PASSWORD,
    port=DB_PORT,
)
conn.autocommit = True

with conn.cursor() as cur:
    cur.execute("""
        CREATE TABLE IF NOT EXISTS customer_notes (
            id SERIAL PRIMARY KEY,
            customer_name TEXT NOT NULL,
            note TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT NOW()
        )
    """)

2. •Next, wrap PostgreSQL access in plain Python functions. AutoGen works best when tools are deterministic, narrow, and easy to test outside the agent loop.

def add_customer_note(customer_name: str, note: str) -> str:
    with conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO customer_notes (customer_name, note)
            VALUES (%s, %s)
            RETURNING id
            """,
            (customer_name, note),
        )
        row_id = cur.fetchone()[0]
    return f"Inserted note {row_id} for {customer_name}"

def get_customer_notes(customer_name: str) -> list[dict]:
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT id, customer_name, note, created_at
            FROM customer_notes
            WHERE customer_name = %s
            ORDER BY created_at DESC
            """,
            (customer_name,),
        )
        rows = cur.fetchall()

    return [
        {
            "id": row[0],
            "customer_name": row[1],
            "note": row[2],
            "created_at": row[3].isoformat(),
        }
        for row in rows
    ]

3. •Now expose those functions to AutoGen as tools. This is the part that makes the agent able to call PostgreSQL-backed operations instead of hallucinating answers.

import autogen

llm_config = {
    "model": "gpt-4o-mini",
    "api_key": os.environ["OPENAI_API_KEY"],
}

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
)

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config=False,
)

assistant.register_for_llm(name="add_customer_note", description="Insert a customer note into PostgreSQL")(add_customer_note)
assistant.register_for_llm(name="get_customer_notes", description="Fetch notes for a customer from PostgreSQL")(get_customer_notes)
user_proxy.register_for_execution(name="add_customer_note")(add_customer_note)
user_proxy.register_for_execution(name="get_customer_notes")(get_customer_notes)

4. •Give the assistant a task that forces tool usage. Keep the instruction specific so the model does not try to answer from memory when it should hit the database.

task = """
Use PostgreSQL tools only.
Add a note for customer 'Acme Bank' saying 'Requested duplicate statement history for Q4'.
Then fetch all notes for 'Acme Bank' and summarize them in one sentence.
"""

user_proxy.initiate_chat(assistant, message=task)

5. •If you want this to behave like production code, add transaction boundaries and basic error handling around writes. That gives you predictable behavior when the database is down or a query fails.

def safe_add_customer_note(customer_name: str, note: str) -> str:
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO customer_notes (customer_name, note)
                VALUES (%s, %s)
                RETURNING id
                """,
                (customer_name, note),
            )
            row_id = cur.fetchone()[0]
        return f"Inserted note {row_id} for {customer_name}"
    except Exception as exc:
        conn.rollback()
        return f"Database error: {exc}"

Testing It

Run the script once and confirm the table gets created without errors. Then check that the agent inserts a row and retrieves it through the tool calls instead of inventing output.

A good verification path is to query PostgreSQL directly after the chat finishes:

SELECT * FROM customer_notes ORDER BY created_at DESC;

You should see the inserted record for Acme Bank. If AutoGen says it used the tools but nothing appears in the table, your issue is usually credentials, network access, or a missing register_for_execution binding.

Next Steps

• •Add read-only tools for common reporting queries like counts, date filters, and status breakdowns.
• •Put your database calls behind a service layer so multiple agents can reuse the same Postgres functions.
• •Add SQLAlchemy or asyncpg if you need pooling, migrations, or higher throughput under load.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.