When deploying agents to LangSmith, the server provides a built-in Postgres-backed checkpointer that handles state persistence across graph runs. You can replace this with your own BaseCheckpointSaver implementation to use a different storage backend.
You provide a path to an async context manager that yields a BaseCheckpointSaver instance, and the server manages its lifecycle automatically.
Custom checkpointers are in alpha. This feature may experience breaking changes in minor version updates.
Define the checkpointer
Starting from an existing LangSmith application, create a file that defines an async context manager yielding your custom checkpointer. If you are beginning a new project, you can create an app from a template using the CLI.
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()
checkpointer key to your langgraph.json configuration file. The path points to the async context manager you defined earlier.
{
"dependencies": ["."],
"graphs": {
"agent": "./src/agent/graph.py:graph"
},
"env": ".env",
"checkpointer": {
"path": "./src/agent/checkpointer.py:generate_checkpointer"
}
}
Start server
Test the server out locally:
langgraph dev --no-browser
Capabilities
The server checks your checkpointer for base (required) and extended (optional) capabilities at startup. If an extended capability is missing, the server either uses a fallback or disables the corresponding feature.
Base capabilities (required)
| Method | Description |
|---|
aput | Store a checkpoint |
aput_writes | Store pending writes |
aget_tuple | Retrieve a checkpoint |
alist | List checkpoints |
adelete_thread | Delete a thread |
Extended capabilities (optional)
| Method | Description | Fallback if missing |
|---|
adelete_for_runs | Delete checkpoints for specific runs | Rollback multitask strategy unavailable |
acopy_thread | Copy a thread | Slow fallback (re-inserts checkpoints one by one) |
aprune | Prune thread history | Thread history pruning unavailable |
Deploying
You can deploy this app as-is to LangSmith or to your self-hosted platform.
Next steps
