Add verified Next.js persistence, LangGraph and MCP client guides

_examples/langgraph-postgres/.dockerignore

@@ -0,0 +1,5 @@
+.venv/
+__pycache__/
+*.pyc
+.env*
+*.log

_examples/langgraph-postgres/.gitignore

@@ -0,0 +1,5 @@
+.venv/
+__pycache__/
+*.pyc
+.env*
+*.log

_examples/langgraph-postgres/.python-version

@@ -0,0 +1 @@
+3.13

_examples/langgraph-postgres/Procfile

@@ -0,0 +1 @@
+web: uv run --frozen python setup_db.py && exec uv run --frozen uvicorn app:app --host 0.0.0.0 --port 8000 --workers 1

_examples/langgraph-postgres/README.md

@@ -0,0 +1,13 @@
+# LangGraph with a Postgres checkpointer
+
+Runnable source for the [LangGraph persistence guide](https://lizard.build/docs/guides/langgraph-postgres/).
+
+Use Python 3.13 and `uv sync --frozen`. Set `DATABASE_URL` and a random `API_TOKEN` of at least 32 characters. Run the exact `Procfile` command: setup and the server both need `uv run --frozen`.
+
+The deterministic graph drafts uppercase text, interrupts for approval and resumes with the same thread ID. It needs no model API key. Its HTTP API authenticates all thread access with one shared token. Postgres advisory locks serialize requests for the same thread.
+
+With `APP_URL`, `API_TOKEN` and a UUID `THREAD_ID` exported, run `python3 check.py start`. Run `python3 restart_and_wait.py graph`, then the checks with `waiting` and `resume`. Run the restart helper once more and check `done`. The waiting and done checks only read checkpoints.
+
+The restart helper calls Lizard CLI and waits for three successful health responses from a new process. It compares the health endpoint's random `instanceId`, so a response from the old process cannot pass the check. It requires one replica and one HTTP worker, and fails after 120 seconds.
+
+This is a single-user persistence example, not a queue, user authorization system, backup test or exactly-once side-effect implementation. Long tasks, model calls and higher concurrency need separate testing.

_examples/langgraph-postgres/app.py

@@ -0,0 +1,97 @@
+import hmac
+import os
+from contextlib import contextmanager
+from uuid import UUID, uuid4
+from fastapi import FastAPI, Depends, HTTPException
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field, StrictBool
+from langgraph.checkpoint.postgres import PostgresSaver
+from langgraph.types import Command
+from graph import compile_graph
+
+TOKEN = os.environ["API_TOKEN"]
+DB = os.environ["DATABASE_URL"]
+INSTANCE_ID = str(uuid4())
+if len(TOKEN) < 32:
+    raise RuntimeError("API_TOKEN must contain at least 32 characters")
+app = FastAPI(title="LangGraph checkpoint example", docs_url=None, redoc_url=None, openapi_url=None)
+bearer = HTTPBearer(auto_error=False)
+
+
+def authorize(credentials: HTTPAuthorizationCredentials | None = Depends(bearer)):
+    if not credentials or not hmac.compare_digest(credentials.credentials, TOKEN):
+        raise HTTPException(401, "Unauthorized", headers={"WWW-Authenticate": "Bearer"})
+
+
+class StartInput(BaseModel):
+    message: str = Field(min_length=1, max_length=200)
+
+
+class ResumeInput(BaseModel):
+    approved: StrictBool
+
+
+@contextmanager
+def graph_for(thread: UUID):
+    # Connection-scoped lock serializes requests for the same thread across processes.
+    # Closing the connection releases the lock, including after an exception.
+    with PostgresSaver.from_conn_string(DB) as saver:
+        saver.conn.execute("SET lock_timeout = '5s'")
+        saver.conn.execute("SELECT pg_advisory_lock(hashtextextended(%s, 0))", (str(thread),))
+        yield compile_graph(saver), {"configurable": {"thread_id": str(thread)}}
+
+
+def snapshot(graph, config):
+    state = graph.get_state(config)
+    return {"thread_id": config["configurable"]["thread_id"], "values": state.values,
+            "next": list(state.next), "interrupts": [i.value for task in state.tasks for i in task.interrupts]}
+
+
+@app.middleware("http")
+async def no_cache(request, call_next):
+    response = await call_next(request)
+    response.headers["Cache-Control"] = "no-store"
+    response.headers["X-Robots-Tag"] = "noindex"
+    return response
+
+
+@app.exception_handler(Exception)
+async def internal_error(_request, _error):
+    return JSONResponse(status_code=503, content={"detail": "Database or graph unavailable"})
+
+
+@app.get("/health")
+def health():
+    with PostgresSaver.from_conn_string(DB) as saver:
+        saver.conn.execute("SELECT 1 FROM checkpoints LIMIT 1")
+    return {"ready": True, "instanceId": INSTANCE_ID}
+
+
+@app.post("/threads/{thread}", dependencies=[Depends(authorize)])
+def start(thread: UUID, data: StartInput):
+    with graph_for(thread) as (graph, config):
+        if graph.get_state(config).values:
+            raise HTTPException(409, "Thread already exists; read or resume it")
+        graph.invoke({"message": data.message}, config)
+        return snapshot(graph, config)
+
+
+@app.get("/threads/{thread}", dependencies=[Depends(authorize)])
+def read(thread: UUID):
+    with graph_for(thread) as (graph, config):
+        if not graph.get_state(config).values:
+            raise HTTPException(404, "Thread not found")
+        return snapshot(graph, config)
+
+
+@app.post("/threads/{thread}/resume", dependencies=[Depends(authorize)])
+def resume(thread: UUID, data: ResumeInput):
+    with graph_for(thread) as (graph, config):
+        state = graph.get_state(config)
+        if not state.values:
+            raise HTTPException(404, "Thread not found")
+        if not any(task.interrupts for task in state.tasks):
+            raise HTTPException(409, "Thread is not waiting for approval")
+        graph.invoke(Command(resume=data.approved), config)
+        return snapshot(graph, config)

_examples/langgraph-postgres/check.py

@@ -0,0 +1,44 @@
+import json
+import os
+import sys
+from urllib.request import Request, urlopen
+from urllib.error import HTTPError
+from uuid import uuid4
+
+url, token, thread = (os.environ[k] for k in ["APP_URL", "API_TOKEN", "THREAD_ID"])
+
+def call(path, method="GET", data=None, auth=True, expected=200):
+    headers = {"Content-Type": "application/json"}
+    if auth:
+        headers["Authorization"] = f"Bearer {token}"
+    req = Request(url + path, data=json.dumps(data).encode() if data is not None else None, headers=headers, method=method)
+    try:
+        response = urlopen(req, timeout=30)
+    except HTTPError as error:
+        response = error
+    assert response.status == expected, (path, response.status, expected)
+    return json.load(response)
+
+call("/health", auth=False)
+call(f"/threads/{thread}", auth=False, expected=401)
+call(f"/threads/{uuid4()}", expected=404)
+mode = sys.argv[1]
+if mode == "start":
+    state = call(f"/threads/{thread}", "POST", {"message": "saved before restart"})
+    assert state["next"] == ["review"] and state["interrupts"]
+    call(f"/threads/{thread}", "POST", {"message": "must not overwrite"}, expected=409)
+    call(f"/threads/{thread}/resume", "POST", {"approved": "yes"}, expected=422)
+elif mode == "waiting":
+    state = call(f"/threads/{thread}")
+    assert state["next"] == ["review"] and state["interrupts"]
+    assert state["values"]["draft"] == "SAVED BEFORE RESTART"
+elif mode == "resume":
+    state = call(f"/threads/{thread}/resume", "POST", {"approved": True})
+    assert not state["next"] and state["values"]["result"] == "SAVED BEFORE RESTART"
+    call(f"/threads/{thread}/resume", "POST", {"approved": True}, expected=409)
+elif mode == "done":
+    state = call(f"/threads/{thread}")
+    assert not state["next"] and state["values"]["result"] == "SAVED BEFORE RESTART"
+else:
+    raise ValueError("Use start, waiting, resume or done")
+print(f"{mode}: health, auth, thread state and persistence checks passed")

_examples/langgraph-postgres/graph.py

@@ -0,0 +1,32 @@
+from typing import TypedDict
+from langgraph.graph import StateGraph, START, END
+from langgraph.types import interrupt
+
+
+class State(TypedDict, total=False):
+    message: str
+    draft: str
+    approved: bool
+    result: str
+
+
+def draft(state: State):
+    # Deterministic so the persistence test needs no model API key.
+    return {"draft": state["message"].upper()}
+
+
+def review(state: State):
+    approved = interrupt({"draft": state["draft"], "question": "Approve this draft?"})
+    if not isinstance(approved, bool):
+        raise ValueError("Resume with a boolean")
+    return {"approved": approved}
+
+
+def finish(state: State):
+    return {"result": state["draft"] if state["approved"] else "Rejected"}
+
+
+def compile_graph(checkpointer):
+    return (StateGraph(State).add_node("draft", draft).add_node("review", review)
+            .add_node("finish", finish).add_edge(START, "draft").add_edge("draft", "review")
+            .add_edge("review", "finish").add_edge("finish", END).compile(checkpointer=checkpointer))

_examples/langgraph-postgres/pyproject.toml

@@ -0,0 +1,11 @@
+[project]
+name = "langgraph-postgres-example"
+version = "0.1.0"
+requires-python = ">=3.13,<3.14"
+dependencies = [
+  "langgraph==1.2.11",
+  "langgraph-checkpoint-postgres==3.1.2",
+  "psycopg[binary,pool]==3.3.5",
+  "fastapi==0.141.1",
+  "uvicorn==0.52.4",
+]

_examples/langgraph-postgres/restart_and_wait.py

@@ -0,0 +1,38 @@
+import json
+import os
+import subprocess
+import sys
+import time
+from urllib.error import URLError
+from urllib.request import Request, urlopen
+
+if len(sys.argv) != 2 or not os.environ.get("APP_URL"):
+    raise SystemExit("Set APP_URL and pass the service name")
+
+
+def health():
+    request = Request(os.environ["APP_URL"] + "/health", headers={"Cache-Control": "no-cache"})
+    with urlopen(request, timeout=5) as response:
+        body = json.load(response)
+    if body.get("ready") is not True or not isinstance(body.get("instanceId"), str) or not body["instanceId"]:
+        raise ValueError("Health must include ready and instanceId")
+    return body["instanceId"]
+
+
+before = health()
+subprocess.run(["lizard", "restart", "--service", sys.argv[1], "--json"], check=True, timeout=30)
+deadline = time.monotonic() + 120
+candidate, stable = None, 0
+while time.monotonic() < deadline:
+    try:
+        current = health()
+        stable = (stable + 1 if current == candidate else 1) if current != before else 0
+        candidate = current
+        if stable >= 3:
+            print(f"Replacement process ready: {before} -> {current}")
+            break
+    except (URLError, TimeoutError, ValueError, OSError):
+        stable = 0
+    time.sleep(2)
+else:
+    raise SystemExit("No stable replacement process within 120 seconds; inspect lizard events and logs")

_examples/langgraph-postgres/setup_db.py

@@ -0,0 +1,8 @@
+import os
+from langgraph.checkpoint.postgres import PostgresSaver
+
+if __name__ == "__main__":
+    with PostgresSaver.from_conn_string(os.environ["DATABASE_URL"]) as saver:
+        saver.conn.execute("SELECT pg_advisory_lock(19082027)")
+        saver.setup()
+    print("LangGraph checkpoint schema ready", flush=True)