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

This tutorial shows you how to wire a CrewAI Python agent to PostgreSQL so it can read and write structured data instead of relying on prompts alone. You’d use this when your agent needs durable state, retrieval from business tables, or controlled writes into an operational database.

What You'll Need

• •Python 3.10+
• •PostgreSQL 14+ running locally or in your environment
• •A PostgreSQL user with access to the target database
• •OpenAI API key set as OPENAI_API_KEY
• •Python packages:
  • •crewai
  • •crewai-tools
  • •psycopg[binary]
  • •python-dotenv

Step-by-Step

1. •Install the dependencies and create a clean project.
   I’m using psycopg because it’s the current PostgreSQL driver I’d use in production Python code.

python -m venv .venv
source .venv/bin/activate

pip install crewai crewai-tools psycopg[binary] python-dotenv

2. •Create a .env file for your secrets and database connection string.
   Keep credentials out of code. For local development, a single DATABASE_URL is enough.

OPENAI_API_KEY=your_openai_key_here
DATABASE_URL=postgresql://postgres:postgres@localhost:5432/crewdb

3. •Create a small PostgreSQL helper that exposes safe read/write functions.
   CrewAI agents should not get raw SQL execution without guardrails unless you explicitly want that risk. This example gives the agent two narrow tools: one for fetching rows and one for inserting notes.

# db_tools.py
import os
import psycopg
from crewai_tools import tool
from dotenv import load_dotenv

load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")

@tool("fetch_customer_by_email")
def fetch_customer_by_email(email: str) -> str:
    """Fetch a customer record by email."""
    with psycopg.connect(DATABASE_URL) as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT id, name, email FROM customers WHERE email = %s LIMIT 1",
                (email,),
            )
            row = cur.fetchone()
    return str(row) if row else "No customer found"

@tool("add_customer_note")
def add_customer_note(customer_id: int, note: str) -> str:
    """Insert a note for a customer."""
    with psycopg.connect(DATABASE_URL) as conn:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO customer_notes (customer_id, note) VALUES (%s, %s) RETURNING id",
                (customer_id, note),
            )
            note_id = cur.fetchone()[0]
        conn.commit()
    return f"Inserted note_id={note_id}"

4. •Create the CrewAI agent and give it those tools.
   The agent can now answer questions from PostgreSQL and persist structured updates back into it.

# crew_postgres.py
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process, LLM
from db_tools import fetch_customer_by_email, add_customer_note

load_dotenv()

llm = LLM(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))

postgres_agent = Agent(
    role="Customer Data Analyst",
    goal="Look up customer records and store follow-up notes in PostgreSQL",
    backstory="You work with internal support data and must use database tools carefully.",
    llm=llm,
    tools=[fetch_customer_by_email, add_customer_note],
    verbose=True,
)

task = Task(
    description=(
        "Find the customer by email alice@example.com. "
        "If found, add a note that says 'Called customer about billing issue'. "
        "Return the final result clearly."
    ),
    expected_output="A short summary of what was found and what was written to the database.",
    agent=postgres_agent,
)

crew = Crew(agents=[postgres_agent], tasks=[task], process=Process.sequential)
result = crew.kickoff()
print(result)

5. •Create the tables before running the agent.
   The example expects two tables: one for customers and one for notes. If you already have schema management in place, map the tool SQL to your existing tables instead.

CREATE TABLE IF NOT EXISTS customers (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL
);

CREATE TABLE IF NOT EXISTS customer_notes (
    id SERIAL PRIMARY KEY,
    customer_id INTEGER NOT NULL REFERENCES customers(id),
    note TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

INSERT INTO customers (name, email)
VALUES ('Alice Johnson', 'alice@example.com')
ON CONFLICT (email) DO NOTHING;

6. •Run the script and inspect both the terminal output and the database rows.
   If everything is wired correctly, the agent should find Alice in PostgreSQL and write a new note without any manual SQL execution from your side.

python crew_postgres.py

Testing It

Start by checking that fetch_customer_by_email returns an actual row for alice@example.com. Then confirm that customer_notes has a new record tied to that customer ID after the crew finishes.

If you get connection errors, verify DATABASE_URL, network access, and that PostgreSQL is listening on the expected host and port. If the agent hallucinates SQL instead of using tools, tighten the task instructions and keep tool descriptions narrow.

For deeper validation, run the script twice and confirm inserts are duplicated only when intended. In production, you’d usually add idempotency keys or deduplication logic before allowing repeated writes.

Next Steps

• •Add a third tool for parameterized search across multiple business tables.
• •Wrap database access in a repository layer if you want stronger domain boundaries.
• •Add row-level permissions or service accounts so agents never connect with superuser credentials.

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.