How to use a custom checkpointer

langgraph new --template=new-langgraph-project-python my_new_project

# ./src/agent/checkpointer.py
import contextlib

class MyCheckpointer(BaseCheckpointSaver):
    def __init__(self):
        super().__init__()
        # Initialize your custom checkpointer here
    ...

    @contextlib.asynccontextmanager
    async def aget(self, config: RunnableConfig):
        # Your custom logic to create a connection pool and initialize your checkpointer here.
        yield


@contextlib.asynccontextmanager
async def generate_checkpointer():
    """Yield a BaseCheckpointSaver, open for the duration of the server."""
    async with AsyncSqliteSaver.from_conn_string("./checkpoints.db") as saver:
        await saver.setup()
        yield saver

pip install langgraph-checkpoint-conformance

import asyncio

from langgraph.checkpoint.conformance import checkpointer_test, validate


@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
    async with MyCheckpointer(...) as saver:
        yield saver


async def main():
    report = await validate(my_checkpointer)
    report.print_report()
    assert report.passed_all_base()


asyncio.run(main())

import pytest

from langgraph.checkpoint.conformance import checkpointer_test, validate


@checkpointer_test(name="MyCheckpointer")
async def my_checkpointer():
    async with MyCheckpointer(...) as saver:
        yield saver


@pytest.mark.asyncio
async def test_conformance():
    report = await validate(my_checkpointer)
    report.print_report()
    assert report.passed_all_base()

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./src/agent/graph.py:graph"
  },
  "env": ".env",
  "checkpointer": {
    "path": "./src/agent/checkpointer.py:generate_checkpointer"
  }
}

langgraph dev --no-browser