LangGraph Checkpoint for SurrealDB

SurrealDB-backed checkpointers for LangGraph.

Install

pip install langgraph-surrealdb

Configure SurrealDB

Set these environment variables (prefix: LANGGRAPH_SURREALDB_):

# required for all modes
export LANGGRAPH_SURREALDB_URL="ws://localhost:8000/rpc"
export LANGGRAPH_SURREALDB_NS="langgraph"
export LANGGRAPH_SURREALDB_DB="checkpoint"
export LANGGRAPH_SURREALDB_TABLE_PREFIX="prefix" # optionally add a prefix to checkpoint tables
# root mode
export LANGGRAPH_SURREALDB_AUTH_MODE="root"
export LANGGRAPH_SURREALDB_USER="root"
export LANGGRAPH_SURREALDB_PASSWORD="pass"

# record mode
# export LANGGRAPH_SURREALDB_AUTH_MODE="record"
# export LANGGRAPH_SURREALDB_USER="user"
# export LANGGRAPH_SURREALDB_PASSWORD="pass"
# export LANGGRAPH_SURREALDB_ACCESS="method"

# token mode
# export LANGGRAPH_SURREALDB_AUTH_MODE="token"
# export LANGGRAPH_SURREALDB_TOKEN="xyz"

Or create the saver directly from settings:

from langgraph_surrealdb import (
    AsyncSurrealSaver,
    RootAuth,
    SurrealSaver,
    SurrealSaverSettings,
    SurrealSaverDatabaseSettings,
)

settings = SurrealSaverSettings(
    writes_table="writes", # optional, defaults to 'writes'
    checkpoints_table="checkpoints", # optional, defaults to 'checkpoints'
    db=SurrealSaverDatabaseSettings(
        url="ws://localhost:8000/rpc",
        namespace="langgraph",
        database="checkpoint",
        # supports RootAuth, TokenAuth and RecordAuth
        auth=RootAuth(
            username="root",
            password="root"
        )
    )
)

with SurrealSaver.from_settings(settings) as checkpointer:
    ...

async with AsyncSurrealSaver.from_settings(settings) as checkpointer:
    ...

Initialize schema

[!IMPORTANT] When using SurrealDB checkpointers for the first time, call a setup method to create required tables and indexes before using saver operations.

# one-time setup
with SurrealSaver.from_env() as checkpointer:
    checkpointer.setup()

# async equivalent
async with AsyncSurrealSaver.from_env() as checkpointer:
    await checkpointer.setup()

Use with LangGraph (sync)

from langgraph.graph import StateGraph
from langgraph_surrealdb import SurrealSaver

# build your graph
builder = StateGraph(dict)
# ... add nodes and edges ...

with SurrealSaver.from_env() as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    result = graph.invoke(
        {"input": "hello"},
        config={"configurable": {"thread_id": "thread-1"}},
    )

Use with LangGraph (async)

from langgraph.graph import StateGraph
from langgraph_surrealdb import AsyncSurrealSaver

builder = StateGraph(dict)
# ... add nodes and edges ...

async with AsyncSurrealSaver.from_env() as checkpointer:
    await checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)
    result = await graph.ainvoke(
        {"input": "hello"},
        config={"configurable": {"thread_id": "thread-1"}},
    )

Then reuse the same thread_id to resume conversation state across calls.