Build a Text-to-SQL Chatbot With Claude and PostgreSQL: End-to-End Python Tutorial (May 2026)

Image: Anthropic platform docs — Tool use with Claude overview (platform.claude.com), used for editorial coverage of the API surface discussed below.

TL;DR

This tutorial walks through a working command-line chatbot that turns plain-English questions into safe SELECT queries against a PostgreSQL copy of the Chinook sample database, executes them, and prints results as a Markdown table. The stack is Python 3.11+, the official anthropic SDK ¹ , psycopg 3 for PostgreSQL ² , and the public Chinook schema from the lerocha/chinook-database repository ³ .

The safety story leans on Anthropic’s documented tool-use loop. Claude proposes a query via a run_select tool call, the bot inspects the SQL, and only single-statement SELECT payloads run, inside a READ ONLY PostgreSQL transaction with a 5-second statement timeout ^{4 5} . Per Anthropic’s tool-use overview, Claude itself never touches the database; the bot keeps full control of execution ⁶ . Total walk-through time is roughly 60 minutes for a developer comfortable with pip, psql, and a terminal.

What you’ll need

• Python 3.11 or newer with pip on PATH.
• PostgreSQL 16 or 17 running locally (or a managed instance you can connect to from your laptop). The PostgreSQL 17 docs are the reference used below ^{4 5} .
• An Anthropic API key from console.anthropic.com.
• About 200 MB of free disk for the Chinook sample data plus Python virtualenv.

Step 1: Load the Chinook sample database

The Chinook dataset models a small digital media store: 11 tables covering customers, invoices, tracks, albums, artists, and genres. The canonical source is the lerocha/chinook-database GitHub repository, which ships ready-made PostgreSQL DDL plus data ³ .

Fetch the PostgreSQL bootstrap file and load it into a fresh database:

createdb chinook
curl -L -o Chinook_PostgreSql.sql \
  https://raw.githubusercontent.com/lerocha/chinook-database/master/ChinookDatabase/DataSources/Chinook_PostgreSql.sql
psql -d chinook -f Chinook_PostgreSql.sql

Confirm the load worked:

psql -d chinook -c "\dt"

You should see 11 tables: Album, Artist, Customer, Employee, Genre, Invoice, InvoiceLine, MediaType, Playlist, PlaylistTrack, Track.

Image: GitHub — lerocha/chinook-database README (github.com), used for editorial coverage of the sample dataset.

Chinook table and column identifiers are CamelCase and case- sensitive in PostgreSQL because the DDL quotes them. Every query in this article double-quotes identifiers for that reason: "Customer", not customer.

Step 2: Project scaffolding

Create the project and install dependencies:

mkdir claude-sql-bot && cd claude-sql-bot
python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install "anthropic>=0.40" "psycopg[binary]>=3.2" python-dotenv
pip freeze > requirements.txt

psycopg[binary] is the pure-wheel install of psycopg 3, so no local PostgreSQL build toolchain is needed ² . python-dotenv reads a local .env file into os.environ at startup ⁷ .

Create .env in the project root:

ANTHROPIC_API_KEY=sk-ant-your-key-here
DATABASE_URL=postgresql://localhost/chinook

Add .env to .gitignore straight away.

Step 3: Expose the schema to Claude

Claude needs to see the schema to write correct queries. Rather than dumping the entire pg_catalog, the bot queries information_schema once at startup and serialises a compact description: one line per column with table, name, and type.

Create schema.py:

import psycopg

SCHEMA_SQL = """
SELECT
    table_name,
    column_name,
    data_type
FROM information_schema.columns
WHERE table_schema = 'public'
ORDER BY table_name, ordinal_position;
"""


def load_schema(dsn: str) -> str:
    """Return a compact text description of every public-schema column."""
    lines: list[str] = []
    current_table: str | None = None
    with psycopg.connect(dsn) as conn:
        with conn.cursor() as cur:
            cur.execute(SCHEMA_SQL)
            for table, column, dtype in cur.fetchall():
                if table != current_table:
                    lines.append(f'\nTable "{table}":')
                    current_table = table
                lines.append(f'  "{column}" {dtype}')
    return "\n".join(lines).strip()

The resulting string is around 2 KB for Chinook, small enough to include verbatim in every Claude request, which gives the model ground truth on column names and types without retrieval.

Image: PyPI project page for psycopg (pypi.org), used for editorial coverage of the PostgreSQL driver discussed below.

Step 4: Define the run_select tool

Per Anthropic’s “Define tools” reference, every custom tool needs a name, a description, and a JSON Schema input_schema describing its parameters ⁸ . The chatbot exposes exactly one tool. Claude proposes a query; the bot decides whether to run it.

Create tools.py:

RUN_SELECT_TOOL = {
    "name": "run_select",
    "description": (
        "Execute a single read-only SELECT query against the "
        "Chinook PostgreSQL database and return the rows as JSON. "
        "Only one statement per call. No INSERT, UPDATE, DELETE, "
        "DDL, or multi-statement payloads. Identifiers in Chinook "
        "are CamelCase and must be double-quoted."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "sql": {
                "type": "string",
                "description": "A single SELECT statement.",
            },
            "row_limit": {
                "type": "integer",
                "description": "Maximum rows to return (1-200).",
                "minimum": 1,
                "maximum": 200,
            },
        },
        "required": ["sql"],
    },
}

The row_limit parameter caps result-set size so a stray SELECT * FROM "Track" does not return all 3,503 rows into the chat transcript.

Step 5: Enforce SELECT-only at the bot layer

Trusting the model’s promise to write read-only SQL is not enough. Per the Anthropic tool-use overview, your code is the one that actually executes tool calls; Claude only proposes them ⁶ . Two layers of defence run before any statement reaches PostgreSQL:

1. A static SQL check that rejects multi-statement payloads, DDL, DML, and anything that isn’t a single SELECT or WITH ... SELECT.
2. A PostgreSQL-side READ ONLY transaction with a 5-second statement_timeout ^{4 5} .

Add to tools.py:

import re

ALLOWED_PREFIXES = ("select", "with")
FORBIDDEN_TOKENS = (
    "insert", "update", "delete", "drop", "alter", "create",
    "truncate", "grant", "revoke", "copy", "vacuum", "reindex",
    "comment", "lock", "call", "do",
)


def is_safe_select(sql: str) -> tuple[bool, str]:
    """Return (ok, reason). Rejects anything not a single SELECT."""
    stripped = sql.strip().rstrip(";").strip()
    if not stripped:
        return False, "empty SQL"
    if ";" in stripped:
        return False, "multiple statements not allowed"
    lowered = stripped.lower()
    if not lowered.startswith(ALLOWED_PREFIXES):
        return False, "only SELECT / WITH ... SELECT allowed"
    # Reject forbidden tokens as whole words.
    pattern = r"\b(" + "|".join(FORBIDDEN_TOKENS) + r")\b"
    match = re.search(pattern, lowered)
    if match:
        return False, f"forbidden keyword: {match.group(1)}"
    return True, "ok"

The whole-word regex avoids false positives on column names like UpdatedAt; the \b boundary keeps the check on actual SQL keywords.

Step 6: Execute queries in a read-only transaction

Create db.py. The executor opens a READ ONLY transaction, sets a 5-second statement timeout, runs the query, and returns rows plus column names:

import psycopg


def run_query(dsn: str, sql: str, row_limit: int = 50) -> dict:
    """Execute SQL inside a READ ONLY transaction with a 5s timeout."""
    with psycopg.connect(dsn) as conn:
        conn.read_only = True
        with conn.cursor() as cur:
            cur.execute("SET LOCAL statement_timeout = '5s'")
            cur.execute(sql)
            rows = cur.fetchmany(row_limit)
            columns = [d.name for d in cur.description]
    return {
        "columns": columns,
        "rows": [list(r) for r in rows],
        "row_count": len(rows),
        "truncated": len(rows) == row_limit,
    }

conn.read_only = True maps to SET TRANSACTION READ ONLY on the PostgreSQL side, which blocks every write statement ⁴ . The SET LOCAL statement_timeout caps execution at 5 seconds per the PostgreSQL 17 client- configuration reference ⁵ . Both are belt-and- braces: even if the static check missed a forbidden statement, the database refuses to write or to run long.

Image: PostgreSQL project mark on postgresql.org, used for editorial coverage of the PostgreSQL 17 reference cited above.

Step 7: Render rows as a Markdown table

The chatbot prints results inline. A small helper turns the dict returned by run_query into a Markdown table:

def to_markdown_table(result: dict) -> str:
    columns = result["columns"]
    rows = result["rows"]
    if not rows:
        return "_(no rows)_"
    header = "| " + " | ".join(columns) + " |"
    sep = "| " + " | ".join("---" for _ in columns) + " |"
    body = "\n".join(
        "| " + " | ".join(str(c) if c is not None else "" for c in row) + " |"
        for row in rows
    )
    footer = ""
    if result["truncated"]:
        footer = f"\n\n_(truncated to {result['row_count']} rows)_"
    return f"{header}\n{sep}\n{body}{footer}"

Step 8: Wire the tool-use loop

The main script ties it together. Per Anthropic’s “How tool use works”, a client-tool turn runs as a loop: Claude returns stop_reason: "tool_use", your code executes the call, you append the assistant turn and a user turn carrying a tool_result content block, and Claude either calls another tool or returns a final text answer ⁹ .

Create app.py:

import json
import os
import sys

from dotenv import load_dotenv
from anthropic import Anthropic

from db import run_query
from schema import load_schema
from tools import RUN_SELECT_TOOL, is_safe_select

load_dotenv()

MODEL = "claude-opus-4-7"
MAX_TURNS = 6


def system_prompt(schema_text: str) -> str:
    return (
        "You are a careful PostgreSQL analyst answering questions "
        "about the Chinook sample database. Use the run_select "
        "tool to fetch data; never invent rows. Identifiers are "
        "CamelCase and must be double-quoted (\"Customer\", "
        "\"Invoice\", \"Track\"). Prefer joins over multiple "
        "round-trips. Cap your queries with LIMIT when the user "
        "asks for a top-N answer. After you have the data, write "
        "a one-sentence answer followed by the Markdown table "
        "exactly as returned by the tool.\n\n"
        f"Schema:\n{schema_text}"
    )


def execute_tool(name: str, arguments: dict, dsn: str) -> str:
    if name != "run_select":
        return json.dumps({"error": f"unknown tool: {name}"})
    sql = arguments.get("sql", "")
    row_limit = int(arguments.get("row_limit", 50))
    ok, reason = is_safe_select(sql)
    if not ok:
        return json.dumps({"error": f"rejected: {reason}"})
    try:
        result = run_query(dsn, sql, row_limit=row_limit)
    except Exception as exc:
        return json.dumps({"error": f"db error: {exc}"})
    return json.dumps(result, default=str)


def ask(client: Anthropic, dsn: str, schema_text: str, question: str) -> str:
    messages = [{"role": "user", "content": question}]
    for _ in range(MAX_TURNS):
        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=system_prompt(schema_text),
            tools=[RUN_SELECT_TOOL],
            messages=messages,
        )
        if response.stop_reason == "end_turn":
            return "".join(
                block.text for block in response.content
                if block.type == "text"
            )
        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    output = execute_tool(block.name, block.input, dsn)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": output,
                    })
            messages.append({"role": "user", "content": tool_results})
            continue
        return f"(unexpected stop_reason: {response.stop_reason})"
    return "(turn limit reached)"


def main() -> None:
    dsn = os.environ["DATABASE_URL"]
    client = Anthropic()
    schema_text = load_schema(dsn)
    print("Chinook chatbot ready. Ctrl-D to exit.\n")
    for line in sys.stdin:
        question = line.strip()
        if not question:
            continue
        answer = ask(client, dsn, schema_text, question)
        print(answer + "\n")


if __name__ == "__main__":
    main()

The loop has a MAX_TURNS ceiling of six so a confused model can not spend the whole API budget retrying. Per Anthropic’s tool-use documentation, the conversation ends when stop_reason flips back to "end_turn" ⁹ .

Image: Anthropic platform docs — How tool use works (platform.claude.com), used for editorial coverage of the loop described above.

Step 9: Try it

Run the bot and ask a few questions:

python app.py

top 5 customers by total revenue

A typical run goes: Claude proposes SELECT "Customer"."FirstName", "Customer"."LastName", SUM("Invoice"."Total") AS revenue FROM "Customer" JOIN "Invoice" ON "Customer"."CustomerId" = "Invoice"."CustomerId" GROUP BY "Customer"."CustomerId" ORDER BY revenue DESC LIMIT 5;, the bot validates and executes, and Claude prints a one-line summary plus the Markdown table.

Two more questions worth trying:

• which 3 genres have the highest average track price?
• list the 5 longest tracks on the album 'Master Of Puppets'

If you ask something the schema cannot answer (“what is the weather today”), Claude’s system prompt instructs it to say so rather than hallucinate a query.

Common failure modes

A few patterns recur the first time you run a build like this:

• Quoted identifier mistakes. Claude occasionally writes customer instead of "Customer"; PostgreSQL returns relation "customer" does not exist. The system prompt names the convention; if errors persist, the tool_result content can feed the error back so Claude self-corrects on the next turn.
• Statement timeout fires. A 5-second cap is generous for Chinook (a few thousand rows). On a larger schema, raise the SET LOCAL statement_timeout to a number that matches your data, but never remove the cap.
• Tool-use loop runs forever. The MAX_TURNS ceiling exists for this reason. Six turns is enough headroom for a two-step query plus error recovery.

Where to take this next

This build covers the smallest version of the pattern. Natural extensions:

• EXPLAIN before execute. Run EXPLAIN on every proposed query and reject any plan whose estimated cost exceeds a threshold.
• Cite the rows. Have Claude include the tool_use_id of the underlying query in its final answer, so a reader can verify the exact SQL that produced each table.
• Wrap it in a web UI. Swap the sys.stdin loop for FastAPI plus htmx. The tool-use loop stays identical.