MongoDBStore#

class langgraph.store.mongodb.base.MongoDBStore(collection: Collection, ttl_config: TTLConfig | None = None, index_config: VectorIndexConfig | None = None, auto_index_timeout: int = 15, query_model: str | None = None, **kwargs: Any)[source]#

MongoDB’s persistent key-value stores for long-term memory.

Stores enable persistence and memory that can be shared across threads, scoped to user IDs, assistant IDs, or other arbitrary namespaces.

Supports semantic search capabilities through an optional index configuration. Only a single embedding is permitted per item, although embedded fields can be indexed via [“parent.child”].

Methods

`__init__`(collection[, ttl_config, ...])	Construct store and its indexes.
`abatch`(ops)	Execute multiple operations asynchronously in a single batch.
`adelete`(namespace, key)	Asynchronously delete an item.
`aget`(namespace, key, *[, refresh_ttl])	Asynchronously retrieve a single item.
`alist_namespaces`(*[, prefix, suffix, ...])	List and filter namespaces in the store asynchronously.
`aput`(namespace, key, value[, index, ttl])	Asynchronously store or update an item in the store.
`asearch`(namespace_prefix, /, *[, query, ...])	Asynchronously search for items within a namespace prefix.
`batch`(ops)	Execute multiple operations synchronously in a single batch.
`delete`(namespace, key)	Delete an item.
`ensure_index_filters`([filters])	Prepare filters for Atlas indexing.
`from_conn_string`([conn_string, db_name, ...])	Context manager to create a persistent MongoDB key-value store.
`get`(namespace, key, *[, refresh_ttl])	Retrieve a single item.
`list_namespaces`(*[, prefix, suffix, ...])	List and filter namespaces in the store.
`put`(namespace, key, value[, index, ttl])	Store or update an item in the store.
`search`(namespace_prefix, /, *[, query, ...])	Search for items within a namespace prefix.

Parameters:

collection (Collection)
ttl_config (TTLConfig | None)
index_config (VectorIndexConfig | None)
auto_index_timeout (int)
query_model (str | None)
kwargs (Any)

__init__(collection: Collection, ttl_config: TTLConfig | None = None, index_config: VectorIndexConfig | None = None, auto_index_timeout: int = 15, query_model: str | None = None, **kwargs: Any)[source]#

Construct store and its indexes.

Semantic search is supported by an Atlas vector search index TTL (time to live) is supported by a TTL index of field: updated_at.

Parameters:

collection (Collection) – Collection of Items backing the store.
ttl_config (TTLConfig | None) – Optionally define a TTL and whether to update on reads(get/search).
index_config (VectorIndexConfig | None) – Optionally define a VectorIndexConfig for semantic search.
auto_index_timeout (int) – Optional timeout for creation of indexes.
query_model (str | None) – For semantic search, optionally provide a different model for search than indexing.
kwargs (Any)

Returns:

Instance of MongoDBStore.

Execute multiple operations asynchronously in a single batch.

Parameters:: ops (Iterable[GetOp | SearchOp | PutOp | ListNamespacesOp]) – An iterable of operations to execute.
Returns:: A list of results, where each result corresponds to an operation in the input. The order of results matches the order of input operations.
Return type:: list[Item | list[Item] | list[SearchItem] | list[tuple[str, …]] | None]

async adelete(namespace: tuple[str, ...], key: str) → None#

Asynchronously delete an item.

Parameters:

namespace (tuple[str, ...]) – Hierarchical path for the item.
key (str) – Unique identifier within the namespace.

Return type:

None

async aget(namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None) → Item | None#

Asynchronously retrieve a single item.

Parameters:

namespace (tuple[str, ...]) – Hierarchical path for the item.
key (str) – Unique identifier within the namespace.
refresh_ttl (bool | None)

Returns:

The retrieved item or None if not found.

Return type:

Item | None

List and filter namespaces in the store asynchronously.

Used to explore the organization of data, find specific collections, or navigate the namespace hierarchy.

Parameters:

prefix (tuple[str | Literal['*'], ...] | None) – Filter namespaces that start with this path.
suffix (tuple[str | Literal['*'], ...] | None) – Filter namespaces that end with this path.
max_depth (int | None) – Return namespaces up to this depth in the hierarchy. Namespaces deeper than this level will be truncated to this depth.
limit (int) – Maximum number of namespaces to return.
offset (int) – Number of namespaces to skip for pagination.

Returns:

A list of namespace tuples that match the criteria. Each tuple represents a: full namespace path up to max_depth.

Return type:

list[tuple[str, …]]

???+ example “Examples”

Setting max_depth=3 with existing namespaces: ```python # Given the following namespaces: # (“a”, “b”, “c”) # (“a”, “b”, “d”, “e”) # (“a”, “b”, “d”, “i”) # (“a”, “b”, “f”) # (“a”, “c”, “f”)

await store.alist_namespaces(prefix=(“a”, “b”), max_depth=3) # Returns: [(“a”, “b”, “c”), (“a”, “b”, “d”), (“a”, “b”, “f”)] ```

async aput(namespace: tuple[str, ...], key: str, value: dict[str, Any], index: Literal[False] | list[str] | None = None, *, ttl: float | None | NotProvided = NOT_GIVEN) → None#

Asynchronously store or update an item in the store.

Parameters:

namespace (tuple[str, ...]) – Hierarchical path for the item, represented as a tuple of strings. Example: (“documents”, “user123”)
key (str) – Unique identifier within the namespace. Together with namespace forms the complete path to the item.
value (dict[str, Any]) – Dictionary containing the item’s data. Must contain string keys and JSON-serializable values.
index (Literal[False] | list[str] | None) –
Controls how the item’s fields are indexed for search:
- None (default): Use fields you configured when creating the store (if any)
  If you do not initialize the store with indexing capabilities, the index parameter will be ignored
- False: Disable indexing for this item
- list[str]: List of field paths to index, supporting:
  - Nested fields: “metadata.title”
  - Array access: “chapters[*].content” (each indexed separately)
  - Specific indices: “authors[0].name”
ttl (float | None | NotProvided) – Time to live in minutes. Support for this argument depends on your store adapter. If specified, the item will expire after this many minutes from when it was last accessed. None means no expiration. Expired runs will be deleted opportunistically. By default, the expiration timer refreshes on both read operations (get/search) and write operations (put/update), whenever the item is included in the operation.

Return type:

None

Note

Indexing support depends on your store implementation. If you do not initialize the store with indexing capabilities, the index parameter will be ignored.

Similarly, TTL support depends on the specific store implementation. Some implementations may not support expiration of items.

???+ example “Examples”

Store item. Indexing depends on how you configure the store:

`python await store.aput(("docs",), "report", {"memory": "Will likes ai"}) `

Do not index item for semantic search. Still accessible through get() and search() operations but won’t have a vector representation.

`python await store.aput(("docs",), "report", {"memory": "Will likes ai"}, index=False) `

Index specific fields for search (if store configured to index items):

```python await store.aput(

(“docs”,), “report”, {

“memory”: “Will likes ai”, “context”: [{“content”: “…”}, {“content”: “…”}]

}, index=[“memory”, “context[*].content”]

async asearch(namespace_prefix: tuple[str, ...], /, *, query: str | None = None, filter: dict[str, Any] | None = None, limit: int = 10, offset: int = 0, refresh_ttl: bool | None = None) → list[SearchItem]#

Asynchronously search for items within a namespace prefix.

Parameters:

namespace_prefix (tuple[str, ...]) – Hierarchical path prefix to search within.
query (str | None) – Optional query for natural language search.
filter (dict[str, Any] | None) – Key-value pairs to filter results.
limit (int) – Maximum number of items to return.
offset (int) – Number of items to skip before returning results.
refresh_ttl (bool | None) – Whether to refresh TTLs for the returned items. If None, uses the store’s TTLConfig.refresh_default setting. If TTLConfig is not provided or no TTL is specified, this argument is ignored.

Returns:

List of items matching the search criteria.

Return type:

list[SearchItem]

???+ example “Examples”

Basic filtering:

```python # Search for documents with specific metadata results = await store.asearch(

(“docs”,), filter={“type”: “article”, “status”: “published”}

Natural language search (requires vector store implementation):

```python # Initialize store with embedding configuration store = YourStore( # e.g., InMemoryStore, AsyncPostgresStore

index={
“dims”: 1536, # embedding dimensions “embed”: your_embedding_function, # function to create embeddings “fields”: [“text”] # fields to embed

}

)

# Search for semantically similar documents

results = await store.asearch(
(“docs”,), query=”machine learning applications in healthcare”, filter={“type”: “research_paper”}, limit=5

!!! note

Natural language search support depends on your store implementation and requires proper embedding configuration.

Execute multiple operations synchronously in a single batch.

Get, Search, and List operations are performed on state before batch. Put and Delete change state. They are deduplicated and applied in order, but only after the read operations have completed.

Parameters:: ops (Iterable[GetOp | SearchOp | PutOp | ListNamespacesOp]) – An iterable of operations to execute.
Returns:: A list of results, where each result corresponds to an operation in the input. The order of results matches the order of input operations. The length of output may not match the input as PutOp returns None.
Return type:: list[Item | list[Item] | list[SearchItem] | list[tuple[str, …]] | None]

delete(namespace: tuple[str, ...], key: str) → None[source]#

Delete an item.

Parameters:

namespace (tuple[str, ...]) – Hierarchical path for the item.
key (str) – Unique identifier within the namespace.

Return type:

None

classmethod ensure_index_filters(filters: list[str] | None = None) → list[str][source]#

Prepare filters for Atlas indexing.

We must ensure that namespace_prefix is included in the filter. We also must ensure that the implicit value field is added.

Parameters:: filters (list[str] | None)
Return type:: list[str]

classmethod from_conn_string(conn_string: str | None = None, db_name: str = 'checkpointing_db', collection_name: str = 'persistent-store', ttl_config: TTLConfig | None = None, index_config: VectorIndexConfig | None = None, **kwargs: Any) → Iterator[MongoDBStore][source]#

Context manager to create a persistent MongoDB key-value store.

A unique compound index will be added to the collections backing the store on (namespace_str, key). On first initialization of an existing collection, legacy documents are backfilled with a namespace_str field and the old (namespace, key) index is replaced. On every initialization, documents missing namespace_str are backfilled; all other steps are no-ops if the collection is already up to date.

If the ttl argument is provided, TTL functionality will be employed. This is done automatically via MongoDB’s TTL Indexes, based on the updated_at field of the collection. The index will be created if it does not already exist.

Parameters:

conn_string (str | None) – MongoDB connection string. See [class:~pymongo.MongoClient].
db_name (str) – Database name. It will be created if it doesn’t exist.
collection_name (str) – Collection name backing the store. Created if it doesn’t exist.
ttl_config (TTLConfig | None) – Defines a TTL (in seconds) and whether to update on reads(get/search).
index_config (VectorIndexConfig | None) – Defines a VectorIndexConfig for semantic search queries.
kwargs (Any)

Return type:

Iterator[MongoDBStore]

Yields: A new MongoDBStore.

get(namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None) → Item | None[source]#

Retrieve a single item.

Parameters:

namespace (tuple[str, ...]) – Hierarchical path for the item.
key (str) – Unique identifier within the namespace.
refresh_ttl (bool | None) – Whether to refresh TTLs for the returned item. If None (default), uses the store’s default refresh_ttl setting. If no TTL is specified, this argument is ignored.

Returns:

The retrieved item or None if not found.

Return type:

Item | None

List and filter namespaces in the store.

Parameters:

prefix (tuple[str | Literal['*'], ...] | None) – Filter namespaces that start with this path.
suffix (tuple[str | Literal['*'], ...] | None) – Filter namespaces that end with this path.
max_depth (int | None) – Return namespaces up to this depth in the hierarchy.
limit (int) – Maximum number of namespaces to return (default 100).
offset (int) – Number of namespaces to skip for pagination. [Not implemented.]

Return type:

list[tuple[str, …]]

Returns: A list of namespace tuples that match the criteria.

put(namespace: tuple[str, ...], key: str, value: dict[str, Any], index: Literal[False] | list[str] | None = None, *, ttl: float | None | NotProvided = NOT_GIVEN) → None#

Store or update an item in the store.

Parameters:

namespace (tuple[str, ...]) – Hierarchical path for the item, represented as a tuple of strings. Example: (“documents”, “user123”)
key (str) – Unique identifier within the namespace. Together with namespace forms the complete path to the item.
value (dict[str, Any]) – Dictionary containing the item’s data. Must contain string keys and JSON-serializable values.
index (Literal[False] | list[str] | None) –
Controls how the item’s fields are indexed for search:
- None (default): Use fields you configured when creating the store (if any)
  If you do not initialize the store with indexing capabilities, the index parameter will be ignored
- False: Disable indexing for this item
- list[str]: List of field paths to index, supporting:
  - Nested fields: “metadata.title”
  - Array access: “chapters[*].content” (each indexed separately)
  - Specific indices: “authors[0].name”
ttl (float | None | NotProvided) – Time to live in minutes. Support for this argument depends on your store adapter. If specified, the item will expire after this many minutes from when it was last accessed. None means no expiration. Expired runs will be deleted opportunistically. By default, the expiration timer refreshes on both read operations (get/search) and write operations (put/update), whenever the item is included in the operation.

Return type:

None

Note

Indexing support depends on your store implementation. If you do not initialize the store with indexing capabilities, the index parameter will be ignored.

Similarly, TTL support depends on the specific store implementation. Some implementations may not support expiration of items.

???+ example “Examples”

Store item. Indexing depends on how you configure the store:

`python store.put(("docs",), "report", {"memory": "Will likes ai"}) `

Do not index item for semantic search. Still accessible through get() and search() operations but won’t have a vector representation.

`python store.put(("docs",), "report", {"memory": "Will likes ai"}, index=False) `

Index specific fields for search:

`python store.put(("docs",), "report", {"memory": "Will likes ai"}, index=["memory"]) `

search(namespace_prefix: tuple[str, ...], /, *, query: str | None = None, filter: dict[str, Any] | None = None, limit: int = 10, offset: int = 0, refresh_ttl: bool | None = None, **kwargs: Any) → list[SearchItem][source]#

Search for items within a namespace prefix.

Values are stored in the collection as a document of name ‘value’. One uses dot notation to access embedded fields. For example, value.text, value.address.city and for arrays value.titles.3.

Note that when search will not update the last_updated field and thus not affect TTL, unlike get.

Parameters:

namespace_prefix (tuple[str, ...]) – Hierarchical path prefix to search within.
query (str | None) – Optional query for natural language search.
filter (dict[str, Any] | None) – Key-value pairs to filter results.
limit (int) – Maximum number of items to return.
offset (int) – Number of items to skip before returning results.
refresh_ttl (bool | None) – TTL is not supported for search. Use get if needed.
kwargs (Any)

Returns:

List of items matching the search criteria.

Return type:

list[SearchItem]

???+ example “Examples”

Basic filtering: ```python # Search for documents with specific metadata results = store.search(

(“docs”,), filter={“value.type”: “article”, “value.status”: “published”}

Natural language search (requires vector store implementation): ```python # Initialize store with embedding configuration store = YourStore( # e.g., InMemoryStore, AsyncPostgresStore

index={
“dims”: 1536, # embedding dimensions “embed”: your_embedding_function, # function to create embeddings “fields”: [“text”] # fields to embed. Defaults to [“$”]

}

)