# LangGraph Checkpoint Couchbase
A Couchbase implementation of the LangGraph `CheckpointSaver` interface that enables persisting agent state and conversation history in a Couchbase database.
[](https://opensource.org/licenses/MIT)
## Overview
This package provides a seamless way to persist LangGraph agent states in Couchbase, enabling:
- State persistence across application restarts
- Retrieval of historical conversation steps
- Continued conversations from previous checkpoints
- Both synchronous and asynchronous interfaces
## Installation
```bash
pip install langgraph-checkpointer-couchbase
```
## Requirements
- Python 3.8+
- Couchbase Server (7.0+ recommended)
- LangGraph 0.3.22+
- LangChain OpenAI 0.3.11+
## Prerequisites
- A running Couchbase cluster
- A bucket created for storing checkpoints
- Appropriate credentials with read/write access
## Quick Start
First, set up your agent tools and model:
```python
from typing import Literal
from langchain_openai import ChatOpenAI
@tool
def get_weather(city: Literal["nyc", "sf"]):
"""Use this to get weather information."""
if city == "nyc":
return "It might be cloudy in nyc"
elif city == "sf":
return "It's always sunny in sf"
else:
raise AssertionError("Unknown city")
tools = [get_weather]
model = ChatOpenAI(model_name="gpt-4o-mini", temperature=0)
```
### Synchronous Usage
```python
import os
from langgraph_checkpointer_couchbase import CouchbaseSaver
from langgraph.graph import create_react_agent
with CouchbaseSaver.from_conn_info(
cb_conn_str=os.getenv("CB_CLUSTER") or "couchbase://localhost",
cb_username=os.getenv("CB_USERNAME") or "Administrator",
cb_password=os.getenv("CB_PASSWORD") or "password",
bucket_name=os.getenv("CB_BUCKET") or "test",
scope_name=os.getenv("CB_SCOPE") or "langgraph",
) as checkpointer:
# Create the agent with checkpointing
graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)
# Configure with a unique thread ID
config = {"configurable": {"thread_id": "1"}}
# Run the agent
res = graph.invoke({"messages": [("human", "what's the weather in sf")]}, config)
# Retrieve checkpoints
latest_checkpoint = checkpointer.get(config)
latest_checkpoint_tuple = checkpointer.get_tuple(config)
checkpoint_tuples = list(checkpointer.list(config))
print(latest_checkpoint)
print(latest_checkpoint_tuple)
print(checkpoint_tuples)
```
### Asynchronous Usage
```python
import os
from acouchbase.cluster import Cluster as ACluster
from couchbase.auth import PasswordAuthenticator
from couchbase.options import ClusterOptions
from langgraph_checkpointer_couchbase import AsyncCouchbaseSaver
from langgraph.graph import create_react_agent
auth = PasswordAuthenticator(
os.getenv("CB_USERNAME") or "Administrator",
os.getenv("CB_PASSWORD") or "password",
)
options = ClusterOptions(auth)
cluster = await ACluster.connect(os.getenv("CB_CLUSTER") or "couchbase://localhost", options)
bucket_name = os.getenv("CB_BUCKET") or "test"
scope_name = os.getenv("CB_SCOPE") or "langgraph"
async with AsyncCouchbaseSaver.from_cluster(
cluster=cluster,
bucket_name=bucket_name,
scope_name=scope_name,
) as checkpointer:
# Create the agent with checkpointing
graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)
# Configure with a unique thread ID
config = {"configurable": {"thread_id": "2"}}
# Run the agent asynchronously
res = await graph.ainvoke(
{"messages": [("human", "what's the weather in nyc")]}, config
)
# Retrieve checkpoints asynchronously
latest_checkpoint = await checkpointer.aget(config)
latest_checkpoint_tuple = await checkpointer.aget_tuple(config)
checkpoint_tuples = [c async for c in checkpointer.alist(config)]
print(latest_checkpoint)
print(latest_checkpoint_tuple)
print(checkpoint_tuples)
# Close the cluster when done
await cluster.close()
```
## Configuration Options
| Parameter | Description | Default |
|-----------|-------------|---------|
| CB_CLUSTER | Couchbase connection string | couchbase://localhost |
| CB_USERNAME | Username for Couchbase | Administrator |
| CB_PASSWORD | Password for Couchbase | password |
| CB_BUCKET | Bucket to store checkpoints | test |
| CB_SCOPE | Scope within bucket | langgraph |
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
---
## 📢 Support Policy
We truly appreciate your interest in this project!
This project is **community-maintained**, which means it's **not officially supported** by our support team.
If you need help, have found a bug, or want to contribute improvements, the best place to do that is right here — by [opening a GitHub issue](https://github.com/Couchbase-Ecosystem/langgraph-checkpointer-couchbase/issues).
Our support portal is unable to assist with requests related to this project, so we kindly ask that all inquiries stay within GitHub.
Your collaboration helps us all move forward together — thank you!
Raw data
{
"_id": null,
"home_page": null,
"name": "langgraph-checkpointer-couchbase",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "checkpointer, couchbase, langchain, langgraph, persistence",
"author": null,
"author_email": "Lokesh Goel <lokesh.goel@couchbase.com>",
"download_url": "https://files.pythonhosted.org/packages/f0/23/0e6ab24cd2121de4fdc5ab183c28f5b93679a0a5d5a089426c378f198b4c/langgraph_checkpointer_couchbase-1.0.7.tar.gz",
"platform": null,
"description": "# LangGraph Checkpoint Couchbase\n\nA Couchbase implementation of the LangGraph `CheckpointSaver` interface that enables persisting agent state and conversation history in a Couchbase database.\n\n[](https://opensource.org/licenses/MIT)\n\n## Overview\n\nThis package provides a seamless way to persist LangGraph agent states in Couchbase, enabling:\n- State persistence across application restarts\n- Retrieval of historical conversation steps\n- Continued conversations from previous checkpoints\n- Both synchronous and asynchronous interfaces\n\n## Installation\n\n```bash\npip install langgraph-checkpointer-couchbase\n```\n\n## Requirements\n\n- Python 3.8+\n- Couchbase Server (7.0+ recommended)\n- LangGraph 0.3.22+\n- LangChain OpenAI 0.3.11+\n\n## Prerequisites\n\n- A running Couchbase cluster\n- A bucket created for storing checkpoints\n- Appropriate credentials with read/write access\n\n## Quick Start\n\nFirst, set up your agent tools and model:\n\n```python\nfrom typing import Literal\nfrom langchain_openai import ChatOpenAI\n\n@tool\ndef get_weather(city: Literal[\"nyc\", \"sf\"]):\n \"\"\"Use this to get weather information.\"\"\"\n if city == \"nyc\":\n return \"It might be cloudy in nyc\"\n elif city == \"sf\":\n return \"It's always sunny in sf\"\n else:\n raise AssertionError(\"Unknown city\")\n\n\ntools = [get_weather]\nmodel = ChatOpenAI(model_name=\"gpt-4o-mini\", temperature=0)\n```\n\n### Synchronous Usage\n\n```python\nimport os\nfrom langgraph_checkpointer_couchbase import CouchbaseSaver\nfrom langgraph.graph import create_react_agent\n\nwith CouchbaseSaver.from_conn_info(\n cb_conn_str=os.getenv(\"CB_CLUSTER\") or \"couchbase://localhost\",\n cb_username=os.getenv(\"CB_USERNAME\") or \"Administrator\",\n cb_password=os.getenv(\"CB_PASSWORD\") or \"password\",\n bucket_name=os.getenv(\"CB_BUCKET\") or \"test\",\n scope_name=os.getenv(\"CB_SCOPE\") or \"langgraph\",\n ) as checkpointer:\n # Create the agent with checkpointing\n graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)\n \n # Configure with a unique thread ID\n config = {\"configurable\": {\"thread_id\": \"1\"}}\n \n # Run the agent\n res = graph.invoke({\"messages\": [(\"human\", \"what's the weather in sf\")]}, config)\n \n # Retrieve checkpoints\n latest_checkpoint = checkpointer.get(config)\n latest_checkpoint_tuple = checkpointer.get_tuple(config)\n checkpoint_tuples = list(checkpointer.list(config))\n\n print(latest_checkpoint)\n print(latest_checkpoint_tuple)\n print(checkpoint_tuples)\n```\n\n### Asynchronous Usage\n\n```python\nimport os\nfrom acouchbase.cluster import Cluster as ACluster\nfrom couchbase.auth import PasswordAuthenticator\nfrom couchbase.options import ClusterOptions\nfrom langgraph_checkpointer_couchbase import AsyncCouchbaseSaver\nfrom langgraph.graph import create_react_agent\n\nauth = PasswordAuthenticator(\n os.getenv(\"CB_USERNAME\") or \"Administrator\",\n os.getenv(\"CB_PASSWORD\") or \"password\",\n)\noptions = ClusterOptions(auth)\ncluster = await ACluster.connect(os.getenv(\"CB_CLUSTER\") or \"couchbase://localhost\", options)\n\nbucket_name = os.getenv(\"CB_BUCKET\") or \"test\"\nscope_name = os.getenv(\"CB_SCOPE\") or \"langgraph\"\n\nasync with AsyncCouchbaseSaver.from_cluster(\n cluster=cluster,\n bucket_name=bucket_name,\n scope_name=scope_name,\n ) as checkpointer:\n # Create the agent with checkpointing\n graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)\n \n # Configure with a unique thread ID\n config = {\"configurable\": {\"thread_id\": \"2\"}}\n \n # Run the agent asynchronously\n res = await graph.ainvoke(\n {\"messages\": [(\"human\", \"what's the weather in nyc\")]}, config\n )\n\n # Retrieve checkpoints asynchronously\n latest_checkpoint = await checkpointer.aget(config)\n latest_checkpoint_tuple = await checkpointer.aget_tuple(config)\n checkpoint_tuples = [c async for c in checkpointer.alist(config)]\n\n print(latest_checkpoint)\n print(latest_checkpoint_tuple)\n print(checkpoint_tuples)\n\n# Close the cluster when done\nawait cluster.close()\n```\n\n## Configuration Options\n\n| Parameter | Description | Default |\n|-----------|-------------|---------|\n| CB_CLUSTER | Couchbase connection string | couchbase://localhost |\n| CB_USERNAME | Username for Couchbase | Administrator |\n| CB_PASSWORD | Password for Couchbase | password |\n| CB_BUCKET | Bucket to store checkpoints | test |\n| CB_SCOPE | Scope within bucket | langgraph |\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n## Contributing\n\nContributions are welcome! Please feel free to submit a Pull Request.\n---\n\n## \ud83d\udce2 Support Policy\n\nWe truly appreciate your interest in this project! \nThis project is **community-maintained**, which means it's **not officially supported** by our support team.\n\nIf you need help, have found a bug, or want to contribute improvements, the best place to do that is right here \u2014 by [opening a GitHub issue](https://github.com/Couchbase-Ecosystem/langgraph-checkpointer-couchbase/issues). \nOur support portal is unable to assist with requests related to this project, so we kindly ask that all inquiries stay within GitHub.\n\nYour collaboration helps us all move forward together \u2014 thank you!\n",
"bugtrack_url": null,
"license": null,
"summary": null,
"version": "1.0.7",
"project_urls": {
"Documentation": "https://github.com/couchbase-ecosystem/langgraph-checkpointer-couchbase#readme",
"Issues": "https://github.com/couchbase-ecosystem/langgraph-checkpointer-couchbase/issues",
"Source": "https://github.com/couchbase-ecosystem/langgraph-checkpointer-couchbase"
},
"split_keywords": [
"checkpointer",
" couchbase",
" langchain",
" langgraph",
" persistence"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "a557d356b48c0d460e05f8777fe38c37514768a154c867cb854939ddb58bff2b",
"md5": "4abe46169c58969580a470afd55c277d",
"sha256": "6b0b7aeb20e9ce435ec04c4838decdd54076ba2ccb6e8db43f8f98804fffbef8"
},
"downloads": -1,
"filename": "langgraph_checkpointer_couchbase-1.0.7-py3-none-any.whl",
"has_sig": false,
"md5_digest": "4abe46169c58969580a470afd55c277d",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 11839,
"upload_time": "2025-08-19T10:16:56",
"upload_time_iso_8601": "2025-08-19T10:16:56.705039Z",
"url": "https://files.pythonhosted.org/packages/a5/57/d356b48c0d460e05f8777fe38c37514768a154c867cb854939ddb58bff2b/langgraph_checkpointer_couchbase-1.0.7-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "f0230e6ab24cd2121de4fdc5ab183c28f5b93679a0a5d5a089426c378f198b4c",
"md5": "89e8f5f4526345ea10fc8ba85f9dd75a",
"sha256": "8258d51adf518c6a4d53a841bbf155b6f3f03752ac7e451c87a168642ec1123d"
},
"downloads": -1,
"filename": "langgraph_checkpointer_couchbase-1.0.7.tar.gz",
"has_sig": false,
"md5_digest": "89e8f5f4526345ea10fc8ba85f9dd75a",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 8631,
"upload_time": "2025-08-19T10:16:58",
"upload_time_iso_8601": "2025-08-19T10:16:58.515796Z",
"url": "https://files.pythonhosted.org/packages/f0/23/0e6ab24cd2121de4fdc5ab183c28f5b93679a0a5d5a089426c378f198b4c/langgraph_checkpointer_couchbase-1.0.7.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-19 10:16:58",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "couchbase-ecosystem",
"github_project": "langgraph-checkpointer-couchbase#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "langgraph-checkpointer-couchbase"
}
JSON
