LangGraph agents with Postgres: tools, retrieval, and safe SQL

Most agent tutorials stop at mock dictionaries. Production copilots need live rows: accounts, tickets, policies — and they need them without turning the model into a SQL client.

This guide builds a tenant-scoped research agent with LangChain tools and a LangGraph loop that reads Postgres safely, then retrieves supporting articles. Examples are paired in Python and TypeScript. For ReAct basics and memory, start with Build an agent with LangChain.

What you will build

A support research agent that can:

1. Look up an account with parameterized SQL (get_account)
2. List open tickets for that account (list_open_tickets)
3. Retrieve knowledge articles via Postgres full-text search (search_knowledge)
4. Answer with citations to account IDs and article titles

Optional: swap or add a pgvector semantic search tool when keyword search is not enough.

Schema sketch

Keep the example schema small enough to reason about:

CREATE TABLE accounts (
  id text PRIMARY KEY,
  tenant_id text NOT NULL,
  name text NOT NULL,
  plan text NOT NULL,
  mrr numeric
);
 
CREATE TABLE tickets (
  id text PRIMARY KEY,
  tenant_id text NOT NULL,
  account_id text NOT NULL REFERENCES accounts(id),
  subject text NOT NULL,
  status text NOT NULL,
  priority text,
  updated_at timestamptz NOT NULL DEFAULT now()
);
 
CREATE TABLE knowledge_articles (
  id text PRIMARY KEY,
  tenant_id text NOT NULL,
  title text NOT NULL,
  body text NOT NULL
);

Prerequisites

• Python 3.11+ and Node 20+ (pick the stack you ship; examples cover both)
• Postgres locally or on Neon
• An OpenAI API key (or swap the chat model provider)

Install dependencies:

pip install langgraph langchain-openai langchain-core "psycopg[binary,pool]" python-dotenv

Set DATABASE_URL and OPENAI_API_KEY in the environment.

Why not give the model SQL

A generic execute_sql(query: str) tool is convenient in a demo and dangerous in a multi-tenant product. Prompt injection, confused models, and sloppy tool schemas all converge on the same failure: the model chooses the query string.

Rule of thumb: reads by default. Writes belong in a separate graph path with confirmation gates — see Prompt injection and LLM security for SaaS.

Allowlisted tools beat a SQL sandbox for product agents. You want get_account, not run_whatever.

Connection and tenancy

Create a shared pool. Every tool call uses parameters, never string interpolation of user or model text.

import os
import json
from psycopg.rows import dict_row
from psycopg_pool import ConnectionPool

DATABASE_URL = os.environ["DATABASE_URL"]

pool = ConnectionPool(
    conninfo=DATABASE_URL,
    min_size=1,
    max_size=10,
    kwargs={"row_factory": dict_row},
)


def query(sql: str, params: tuple) -> list[dict]:
    """Run a parameterized query. Never interpolate user or model text into SQL."""
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute(sql, params)
            return list(cur.fetchall())


def compact_json(rows: list[dict], limit: int = 20) -> str:
    return json.dumps(rows[:limit], default=str)

The critical design choice: tenant_id is injected when you build tools, after auth middleware resolves the signed-in user. The model never supplies it. If tenant_id is a tool argument, prompt injection becomes cross-tenant data access.

Define Postgres tools

Factory pattern: build_tools(tenant_id) returns tools closed over that tenant.

from langchain_core.tools import tool

def build_tools(tenant_id: str):
    """tenant_id comes from auth middleware — never from the model."""

    @tool
    def get_account(account_id: str) -> str:
        """Fetch one account by ID for the current tenant."""
        rows = query(
            """
            SELECT id, name, plan, mrr
            FROM accounts
            WHERE tenant_id = %s AND id = %s
            LIMIT 1
            """,
            (tenant_id, account_id),
        )
        return compact_json(rows) if rows else "Account not found."

    @tool
    def list_open_tickets(account_id: str) -> str:
        """List open support tickets for an account (max 20)."""
        rows = query(
            """
            SELECT id, subject, status, priority, updated_at
            FROM tickets
            WHERE tenant_id = %s
              AND account_id = %s
              AND status = 'open'
            ORDER BY updated_at DESC
            LIMIT 20
            """,
            (tenant_id, account_id),
        )
        return compact_json(rows) if rows else "No open tickets."

    @tool
    def search_knowledge(query_text: str) -> str:
        """Search knowledge articles with Postgres full-text search."""
        rows = query(
            """
            SELECT id, title,
                   ts_rank(search_vector, plainto_tsquery('english', %s)) AS rank
            FROM knowledge_articles
            WHERE tenant_id = %s
              AND search_vector @@ plainto_tsquery('english', %s)
            ORDER BY rank DESC
            LIMIT 8
            """,
            (query_text, tenant_id, query_text),
        )
        return compact_json(rows) if rows else "No matching articles."

    return [get_account, list_open_tickets, search_knowledge]

Return compact JSON strings, not driver objects. Cap rows (LIMIT 8–20) so a noisy table does not blow the context window. Tool descriptions matter — they are how the model chooses which query to run.

Wire LangGraph

Use the prebuilt ReAct agent. Pass a system prompt that restates the tenant boundary and citation rules.

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

def build_agent(tenant_id: str):
    model = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
    tools = build_tools(tenant_id)

    system = (
        "You are a support research agent. "
        f"You serve tenant {tenant_id} only. "
        "Use tools for live data. Cite account IDs and article titles. "
        "If a tool returns empty, say you do not have that data."
    )

    return create_react_agent(
        model,
        tools,
        prompt=system,
    )


def run_agent(tenant_id: str, message: str) -> str:
    agent = build_agent(tenant_id)
    result = agent.invoke(
        {"messages": [("user", message)]},
        config={"recursion_limit": 10},
    )
    return result["messages"][-1].content

Invoke only on the server (FastAPI route, Next.js Route Handler). The browser sends a message; middleware authenticates, resolves tenantId, builds tools, runs the graph.

Example user message the agent should handle:

What is Acme's plan, and are there open tickets? Also check our refund policy.

Expected tool sequence: get_account → list_open_tickets → search_knowledge → final answer with citations.

Retrieval patterns on Postgres

Not every question needs embeddings. Match the retrieval style to the question — same decision frame as When not to use RAG:

RAG is a tool inside the agent, not a replacement for SQL lookups. Looking up acme-001 with vectors is slower and less precise than WHERE id = $2.

Full-text search

Add a generated tsvector column and GIN index (migration):

ALTER TABLE knowledge_articles
  ADD COLUMN IF NOT EXISTS search_vector tsvector
  GENERATED ALWAYS AS (
    to_tsvector('english', coalesce(title, '') || ' ' || coalesce(body, ''))
  ) STORED;
 
CREATE INDEX IF NOT EXISTS knowledge_articles_search_idx
  ON knowledge_articles USING GIN (search_vector);

Query helper (same pattern as the tool body):

def search_knowledge_fts(tenant_id: str, query_text: str) -> str:
    rows = query(
        """
        SELECT id, title, left(body, 400) AS excerpt
        FROM knowledge_articles
        WHERE tenant_id = %s
          AND search_vector @@ plainto_tsquery('english', %s)
        ORDER BY ts_rank(search_vector, plainto_tsquery('english', %s)) DESC
        LIMIT 8
        """,
        (tenant_id, query_text, query_text),
    )
    return compact_json(rows)

For integrating retrieval without a separate vector platform, see RAG without the platform rewrite.

Optional: pgvector

When keyword search misses paraphrases, add an embedding vector(...) column and a semantic tool. Keep tenant filters and LIMIT the same as FTS.

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

@tool
def search_knowledge_semantic(query_text: str) -> str:
    """Semantic search over knowledge articles (pgvector)."""
    vector = embeddings.embed_query(query_text)
    rows = query(
        """
        SELECT id, title,
               1 - (embedding <=> %s::vector) AS score
        FROM knowledge_articles
        WHERE tenant_id = %s
        ORDER BY embedding <=> %s::vector
        LIMIT 8
        """,
        (vector, tenant_id, vector),
    )
    return compact_json(rows) if rows else "No matching articles."

Register both FTS and semantic tools if you need them — the model can choose. Start with FTS; add pgvector when evals show a real gap.

Production boundaries

Shared rate limits, provider routing, and cost controls belong in LLM middleware, regardless of LangGraph vs a thinner stack.

If external systems enter the tool surface later, MCP in production SaaS covers allowlists and least privilege — the same bar as native tools.

Common mistakes

Model-authored SQL. Free-form execute_sql turns prompt injection into database access. Use allowlisted tools.

Model-supplied tenantId. Always close over auth context. Treat tool arguments as untrusted.

Unbounded SELECT *. Hard LIMIT, selected columns, truncated excerpts.

Write tools in the same unconstrained loop. Separate graphs or human-in-the-loop nodes for mutations.

Client-side LangGraph with a database URL. Credentials and tool I/O stay on the server.

Skipping evals. Assert tool choice and refusals on a small golden set before widening traffic.

Next steps

1. Point tools at your staging schema with a read-only role
2. Add 20–30 golden questions (correct tool, citation present, refuse missing data)
3. Stream graph events to the UI once the tool path is stable
4. Ship behind a feature flag; watch tool error rate and tokens per session

Building the loop is straightforward. Making Postgres access permissioned, parameterized, and observable is the integration work. Describe your schema and workflow — we will map tools, tenancy, and retrieval for your stack.