LangGraph MongoDB Checkpoint Saver

0.3.1 · active · verified Sat Apr 11

This library provides a MongoDB implementation of LangGraph's `CheckpointSaver` interface, enabling persistence for LangGraph agent states. It allows agents to maintain short-term memory, facilitate human-in-the-loop workflows, and provide fault tolerance by saving graph state checkpoints in a MongoDB database. The current version is 0.3.1, with releases tied to the broader LangChain/LangGraph ecosystem updates.

Warnings

breaking Breaking changes in the base `langgraph-checkpoint` library (e.g., between v0.x/v1.x and v2.x/v3.x) can lead to API mismatches and dependency conflicts. Ensure your `langgraph-checkpoint-mongodb` version is compatible with your `langgraph` and `langgraph-checkpoint` versions to avoid issues like the `langgraph-checkpoint@3.0` incompatibility reported previously.
Fix: Refer to the official documentation and release notes of `langgraph-checkpoint-mongodb` and `langgraph` for compatible version ranges. Upgrade both `langgraph` and `langgraph-checkpoint-mongodb` to their latest compatible versions.
gotcha The `MongoDBSaver` currently does not offer built-in mechanisms for automatic checkpoint retention or Time-To-Live (TTL) configuration. This means that checkpoints will accumulate indefinitely, potentially leading to significant storage growth in production environments with high conversation volume.
Fix: Implement a periodic cleanup job to manually prune old checkpoints based on `checkpoint_id` (which is time-sortable) or wrap the saver to add an `expiresAt`/`createdAt` top-level field for MongoDB's native TTL indexing. It is recommended to keep at least the last few checkpoints per thread.
gotcha For use with the official LangGraph Agent Server, a MongoDB replica set is a prerequisite; standalone `mongod` instances are not supported. Additionally, the MongoDB connection URI must include the database name in its path (e.g., `mongodb://localhost:27017/mydatabase`).
Fix: Ensure your MongoDB deployment is a replica set (or Atlas/`mongos` router). Update your MongoDB connection URI to include the database name, for example, `mongodb://<host>:<port>/<db_name>`.
deprecated The `AsyncMongoDBSaver` class has been removed. Users who previously relied on this for asynchronous operations will need to refactor their code to use the main `MongoDBSaver` which handles operations synchronously or manage async interaction at a higher level.
Fix: Migrate from `AsyncMongoDBSaver` to `MongoDBSaver`. Review LangGraph's asynchronous execution patterns to integrate the synchronous `MongoDBSaver` appropriately, potentially by running operations in a separate thread/executor if truly non-blocking database calls are required.

Install

pip install langgraph-checkpoint-mongodb Install stable version

Imports

MongoDBSaver

from langgraph.checkpoint.mongodb import MongoDBSaver

Quickstart

This quickstart demonstrates how to initialize `MongoDBSaver` and interact with it directly to save and load a dummy checkpoint. In a typical LangGraph application, the `checkpointer` instance is passed to the `graph.compile()` method to automatically manage state persistence.

import os
from langgraph.checkpoint.mongodb import MongoDBSaver
from pymongo import MongoClient

# NOTE: Ensure a MongoDB instance is running, e.g., locally at mongodb://localhost:27017
# Replace with your actual MongoDB URI
MONGODB_URI = os.environ.get('MONGODB_URI', 'mongodb://localhost:27017')
DB_NAME = "langgraph_checkpoints_db"
COLLECTION_NAME = "checkpoints_collection"

# Initialize the MongoDB client
client = MongoClient(MONGODB_URI)

# Initialize the checkpointer with a client, database name, and optional collection name
checkpointer = MongoDBSaver(
    client,
    db_name=DB_NAME,
    collection_name=COLLECTION_NAME
)

# Example usage with a dummy checkpoint and config (simplified for quickstart)
# In a real LangGraph application, this would be managed by the graph's execution.
config = {"configurable": {"thread_id": "test_thread_1", "checkpoint_ns": ""}}
dummy_checkpoint = {
    "v": 1,
    "ts": "2026-04-11T12:00:00.000000+00:00",
    "id": "12345678-abcd-1234-abcd-1234567890ab",
    "channel_values": {"my_state": "initial_value"},
    "channel_versions": {},
    "versions_seen": {},
    "pending_sends": []
}

print(f"Saving checkpoint for thread_id: {config['configurable']['thread_id']}")
checkpointer.put(config, dummy_checkpoint, {}, {})
print("Checkpoint saved.")

print(f"Loading checkpoint for thread_id: {config['configurable']['thread_id']}")
loaded_checkpoint_tuple = checkpointer.get(config)
if loaded_checkpoint_tuple:
    print(f"Loaded checkpoint state: {loaded_checkpoint_tuple.checkpoint.channel_values}")
else:
    print("No checkpoint found.")

# Clean up (optional, for demonstration)
checkpointer.delete_thread(config)
print(f"Deleted checkpoints for thread_id: {config['configurable']['thread_id']}")

client.close()

view raw JSON →