LangGraph Tutorial (Python): connecting to PostgreSQL for beginners

By Cyprian AaronsUpdated 2026-04-22
langgraphconnecting-to-postgresql-for-beginnerspython

This tutorial shows you how to build a small LangGraph app in Python that reads and writes state in PostgreSQL. You need this when your agent must survive process restarts, share state across workers, or keep conversation history outside memory.

What You'll Need

  • Python 3.10+
  • A running PostgreSQL instance
  • A PostgreSQL connection string, for example:
    • postgresql://postgres:postgres@localhost:5432/langgraph_demo
  • pip for installing packages
  • These Python packages:
    • langgraph
    • psycopg[binary]
    • python-dotenv if you want to load env vars from a .env file
  • Basic familiarity with:
    • LangGraph state graphs
    • Python functions and type hints

Step-by-Step

  1. Install the dependencies.
    We need LangGraph plus a PostgreSQL driver that supports the checkpoint store. Keep this minimal and pin versions in your own project later.
pip install langgraph psycopg[binary] python-dotenv
  1. Create a PostgreSQL database and set your connection string.
    The example below assumes you already have a database called langgraph_demo. If you prefer environment variables, put the URL in .env so your code stays clean.
export POSTGRES_URL="postgresql://postgres:postgres@localhost:5432/langgraph_demo"
  1. Build a simple graph with one node that updates state.
    This graph takes a counter in state, increments it, and returns the new value. The important part is that we compile it with a PostgreSQL checkpointer so state can be persisted between runs.
from typing import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver


class CounterState(TypedDict):
    count: int


def increment(state: CounterState) -> CounterState:
    return {"count": state["count"] + 1}


builder = StateGraph(CounterState)
builder.add_node("increment", increment)
builder.add_edge(START, "increment")
builder.add_edge("increment", END)

POSTGRES_URL = "postgresql://postgres:postgres@localhost:5432/langgraph_demo"
with PostgresSaver.from_conn_string(POSTGRES_URL) as checkpointer:
    checkpointer.setup()
    app = builder.compile(checkpointer=checkpointer)
  1. Run the graph with a thread ID so LangGraph can store checkpoints per conversation.
    The thread_id is the key that tells LangGraph which saved state belongs to which run. Run it twice with the same thread ID and you will see persisted state behavior.
from langchain_core.runnables import RunnableConfig

config = RunnableConfig(configurable={"thread_id": "demo-thread-1"})

result1 = app.invoke({"count": 0}, config=config)
print("First run:", result1)

result2 = app.invoke({"count": result1["count"]}, config=config)
print("Second run:", result2)
  1. Load the latest checkpoint back from PostgreSQL.
    This is the part beginners usually want to confirm first: the data is not only in memory. You can inspect the stored state by asking LangGraph for the latest snapshot using the same thread ID.
snapshot = app.get_state(config)
print("Latest saved state:", snapshot.values)
print("Next nodes:", snapshot.next)
  1. Wrap it into a reusable script.
    In production, keep your graph definition separate from runtime configuration so you can reuse the same graph across API workers, batch jobs, and tests.
from typing import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_core.runnables import RunnableConfig


class CounterState(TypedDict):
    count: int


def increment(state: CounterState) -> CounterState:
    return {"count": state["count"] + 1}


def build_app(postgres_url: str):
    builder = StateGraph(CounterState)
    builder.add_node("increment", increment)
    builder.add_edge(START, "increment")
    builder.add_edge("increment", END)

    with PostgresSaver.from_conn_string(postgres_url) as checkpointer:
        checkpointer.setup()
        return builder.compile(checkpointer=checkpointer)


app = build_app(POSTGRES_URL)
config = RunnableConfig(configurable={"thread_id": "demo-thread-2"})
print(app.invoke({"count": 10}, config=config))

Testing It

Run the script once and confirm you get an output like {'count': 1} or {'count': 11} depending on your input. Then run it again with the same thread_id and verify that app.get_state(config) returns the saved checkpoint from PostgreSQL.

If nothing persists, check these first:

  • Your PostgreSQL server is running
  • The database name in POSTGRES_URL exists
  • The user has permission to create tables
  • You called checkpointer.setup()

A good sanity test is to stop your Python process after one invocation, restart it, and query the same thread again. If the state comes back, your LangGraph-to-PostgreSQL connection is working correctly.

Next Steps

  • Add a second node and route between nodes using conditional edges.
  • Replace the counter with real agent state, such as messages or form fields.
  • Add FastAPI around this graph so each HTTP request gets its own thread_id.

Keep learning

By Cyprian Aarons, AI Consultant at Topiax.

Want the complete 8-step roadmap?

Grab the free AI Agent Starter Kit — architecture templates, compliance checklists, and a 7-email deep-dive course.

Get the Starter Kit

Related Guides