How to Connect Any AI Agent to Your Database Using Tool Calling

Most AI agent tutorials stop at toy examples — fake weather APIs and stubbed search results. But the moment you try to build something real, the first question is always the same: how do I connect this thing to my database?

This tutorial answers that question. You will build a Python agent that takes natural language questions, generates SQL, queries a PostgreSQL database, and returns human-readable answers. The entire thing runs on OpenAI's function calling API and about 120 lines of code.

If you are new to tool calling, read that first. If you have already built a basic agent, this is the natural next step.

Prerequisites

• Python 3.10+
• PostgreSQL running locally (or any accessible Postgres instance)
• An OpenAI API key
• Two Python packages:

pip install openai psycopg2-binary

Set your API key:

export OPENAI_API_KEY="your-key-here"

Step 1: Set Up a Sample Database

You need some data to query. Create a small e-commerce schema with products and orders.

Connect to Postgres and run:

CREATE DATABASE agent_demo;
\c agent_demo

CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    category TEXT NOT NULL,
    price DECIMAL(10,2) NOT NULL,
    stock INTEGER NOT NULL
);

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    product_id INTEGER REFERENCES products(id),
    quantity INTEGER NOT NULL,
    total DECIMAL(10,2) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Seed some data
INSERT INTO products (name, category, price, stock) VALUES
    ('Wireless Mouse', 'Electronics', 29.99, 150),
    ('Mechanical Keyboard', 'Electronics', 89.99, 75),
    ('USB-C Hub', 'Electronics', 45.99, 200),
    ('Standing Desk', 'Furniture', 499.99, 30),
    ('Monitor Arm', 'Furniture', 129.99, 60),
    ('Noise-Cancelling Headphones', 'Electronics', 249.99, 40),
    ('Webcam HD', 'Electronics', 79.99, 90),
    ('Ergonomic Chair', 'Furniture', 399.99, 25);

INSERT INTO orders (product_id, quantity, total, created_at) VALUES
    (1, 2, 59.98, '2026-01-15'),
    (2, 1, 89.99, '2026-01-16'),
    (3, 5, 229.95, '2026-01-18'),
    (4, 1, 499.99, '2026-01-20'),
    (1, 3, 89.97, '2026-02-01'),
    (6, 2, 499.98, '2026-02-03'),
    (5, 1, 129.99, '2026-02-05'),
    (2, 2, 179.98, '2026-02-10'),
    (7, 4, 319.96, '2026-02-12'),
    (8, 1, 399.99, '2026-02-14');

Create a Read-Only Database User

This is critical. Your agent should never connect with a user that can write, update, or delete data. Create a read-only role:

CREATE ROLE agent_reader WITH LOGIN PASSWORD 'readonly_pass';
GRANT CONNECT ON DATABASE agent_demo TO agent_reader;
GRANT USAGE ON SCHEMA public TO agent_reader;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_reader;

-- Auto-grant SELECT on future tables too
ALTER DEFAULT PRIVILEGES IN SCHEMA public
    GRANT SELECT ON TABLES TO agent_reader;

If you are on PostgreSQL 14+, you can use the built-in pg_read_all_data role instead — it grants SELECT on every table in the database without manual grants.

Step 2: Define the Database Tools

Your agent needs two tools to be useful:

• get_schema — discovers what tables and columns exist, so the LLM can write valid SQL
• run_query — executes a read-only SQL query and returns the results

import psycopg2
import json

DB_CONFIG = {
    "host": "localhost",
    "dbname": "agent_demo",
    "user": "agent_reader",
    "password": "readonly_pass",
}

def get_schema() -> str:
    """Return all table names and their columns."""
    conn = psycopg2.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT table_name, column_name, data_type
                FROM information_schema.columns
                WHERE table_schema = 'public'
                ORDER BY table_name, ordinal_position
            """)
            rows = cur.fetchall()
    finally:
        conn.close()

    schema = {}
    for table, column, dtype in rows:
        schema.setdefault(table, []).append(f"{column} ({dtype})")

    return json.dumps(schema, indent=2)


def run_query(sql: str) -> str:
    """Execute a SELECT query and return results as JSON."""
    # Safety check: block anything that is not a SELECT
    normalized = sql.strip().lower()
    if not normalized.startswith("select"):
        return json.dumps({"error": "Only SELECT queries are allowed."})

    conn = psycopg2.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            columns = [desc[0] for desc in cur.description]
            rows = cur.fetchall()
    finally:
        conn.close()

    results = [dict(zip(columns, row)) for row in rows]
    return json.dumps(results, indent=2, default=str)

Two safety layers are at work here. The database role only has SELECT privileges, so even if the LLM somehow generates a DROP TABLE, Postgres will reject it. The application-level check catches non-SELECT queries before they ever hit the database. Defense in depth.

Step 3: Register the Tools with OpenAI

Now tell the LLM what tools it has. Each tool gets a name, description, and a JSON Schema for its parameters. These descriptions matter — they are how the model decides when and how to call each tool.

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_schema",
            "description": (
                "Get the database schema including all table names and their "
                "columns with data types. Call this before writing any SQL query "
                "to understand what tables and columns are available."
            ),
            "parameters": {
                "type": "object",
                "properties": {},
                "required": [],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "run_query",
            "description": (
                "Execute a read-only SQL query against the PostgreSQL database "
                "and return the results. Only SELECT statements are allowed. "
                "Always call get_schema first to know the available tables."
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "The SELECT SQL query to execute.",
                    }
                },
                "required": ["sql"],
            },
        },
    },
]

Notice the descriptions nudge the model to call get_schema first. This is a simple but effective way to guide agent behavior without complex orchestration.

Step 4: Build the Agent Loop

This is the core pattern: send a message, check if the model wants to call a tool, execute it, feed the result back, and repeat until the model produces a final text answer.

from openai import OpenAI

client = OpenAI()

SYSTEM_PROMPT = """You are a helpful data analyst agent. You answer questions
about business data stored in a PostgreSQL database.

Rules:
- Always call get_schema before writing any SQL query.
- Write efficient, read-only SQL. Use LIMIT to avoid returning too many rows.
- Format your final answer in a clear, human-readable way.
- If you cannot answer a question from the available data, say so."""

# Map tool names to Python functions
TOOL_MAP = {
    "get_schema": lambda _: get_schema(),
    "run_query": lambda args: run_query(args["sql"]),
}


def run_agent(user_message: str) -> str:
    """Run the agent loop until we get a final text response."""
    messages = [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": user_message},
    ]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
        )

        message = response.choices[0].message
        messages.append(message)

        # If the model did not call any tools, we have our answer
        if not message.tool_calls:
            return message.content

        # Execute each tool call and feed results back
        for tool_call in message.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)

            print(f"  → Calling {fn_name}({fn_args})")
            result = TOOL_MAP[fn_name](fn_args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result,
            })

If you have read the build an agent from scratch tutorial, this pattern should look familiar. The only difference is the tools — instead of weather and search stubs, you have real database access.

Step 5: Run It

Add a simple entry point and test it:

if __name__ == "__main__":
    questions = [
        "What are the top 3 products by total revenue?",
        "How many orders were placed in February 2026?",
        "Which product category has the highest average price?",
    ]

    for question in questions:
        print(f"\n{'='*60}")
        print(f"Q: {question}")
        print(f"{'='*60}")
        answer = run_agent(question)
        print(f"\nA: {answer}")

Run it:

python agent.py

You will see the agent make two tool calls for each question — first get_schema to learn the table structure, then run_query with the generated SQL. The output looks something like:

============================================================
Q: What are the top 3 products by total revenue?
============================================================
  → Calling get_schema({})
  → Calling run_query({'sql': "SELECT p.name, SUM(o.total) as revenue FROM orders o JOIN products p ON o.product_id = p.id GROUP BY p.name ORDER BY revenue DESC LIMIT 3"})

A: Here are the top 3 products by total revenue:

1. **Noise-Cancelling Headphones** — $499.98
2. **Standing Desk** — $499.99
3. **Ergonomic Chair** — $399.99

The LLM wrote the JOIN, the aggregation, and the ORDER BY — all from a natural language question.

Lock It Down: Security Best Practices

The tutorial above works, but shipping it to production needs more guardrails. Here is what to add:

• Read-only database role — already covered. This is your most important defense. Even if every other check fails, Postgres itself will block writes.
• Statement validation — the startswith("select") check is a floor, not a ceiling. For production, use a SQL parser like sqlglot to inspect the AST and reject subqueries with side effects.
• Row limits — inject a LIMIT 100 clause if the LLM forgets to add one. Returning 10 million rows into the agent loop will burn tokens and crash your process.
• Timeouts — set a statement timeout on the database role: ALTER ROLE agent_reader SET statement_timeout = '5s';. Expensive queries get killed automatically.
• No direct production access — for real deployments, query a read replica or a data warehouse, never the primary database. One bad query should not affect your users.
• Audit logging — log every SQL query the agent generates. You need a trail when something goes wrong.

For a deeper look at the architectural patterns, the Zuplo blog has a solid writeup on securing database access for LLM-powered agents.

Where to Go From Here

You built a working database agent in about 120 lines of Python. Here is where to take it next:

• Try a different LLM provider. The same pattern works with Anthropic's tool use API or any provider that supports function calling. Only the message format changes — the loop stays the same.
• Use MCP for a protocol-based approach. Instead of hardcoding tools in your script, expose them as an MCP server. This lets any MCP-compatible client (Claude Desktop, Cursor, etc.) use your database tools without custom integration. Google's open-source MCP Toolbox for Databases handles connection pooling, auth, and observability out of the box.
• Add memory. Right now the agent forgets everything between runs. Adding conversation history or a vector store for past queries means it can learn from previous interactions and avoid redundant schema lookups.
• Support multiple databases. Add a list_databases tool and let the agent pick which data source to query. The pattern scales naturally.

Database access is the gateway use case for AI agents. Once an agent can query your data, it can answer questions no dashboard anticipated, generate reports on demand, and spot patterns humans miss. The hard part is not the code — it is building the right guardrails around it.

Copy the snippets above into a single agent.py file, point it at your own database, and start asking questions.