> Fetch the index of every Tonbo docs page as Markdown at https://tonbo.dev/llms.txt

# LangGraph

Last updated 3 November 2026

`sqlite_ursula.langgraph` holds LangGraph’s SQLite checkpointers and stores on a database, each a subclass of LangGraph’s own whose writes return once Tonbo holds them: [`UrsulaSaver`](https://tonbo.dev/docs/reference/python/langgraph.md#ursula-saver), [`AsyncUrsulaSaver`](https://tonbo.dev/docs/reference/python/langgraph.md#async-ursula-saver), [`UrsulaStore`](https://tonbo.dev/docs/reference/python/langgraph.md#ursula-store) and [`AsyncUrsulaStore`](https://tonbo.dev/docs/reference/python/langgraph.md#async-ursula-store).

Install them with the `langgraph` extra, which adds `langgraph-checkpoint-sqlite<3.2,>=3.0.3`, the releases the package is tested with. The package itself never imports the module.

```shell
uv pip install 'sqlite-ursula[langgraph]==0.11.3.dev0'
```

## How they work

- Each takes a connection `db` that `open()` made, and runs on a connection of its own on the same database, which it opens and closes. LangGraph writes checkpoints from threads of its own, so the application’s connection would otherwise share its transactions. Keep `db` open while the graph runs.
- Every write returns only once `db.synced()` returned after it. LangGraph waits for its background checkpoint writes before `invoke` returns, so when `invoke` returns, every checkpoint of the run is in Tonbo, in every durability mode. `durability="sync"` also gates what a node does after the step before it, at one round trip per step. A streamed chunk is not gated: call [db.synced()](https://tonbo.dev/docs/reference/python/connection.md#synced) before acting on one.
- Reads do not wait, and neither does `setup()`. A reset raises the [UrsulaResetError](https://tonbo.dev/docs/reference/python/errors.md#ursula-reset-error) itself, from a write or a read, and the other errors are the package’s too.
- Given anything but a connection `open()` made, each raises [`invalid_argument`](https://tonbo.dev/docs/reference/python/errors.md#setup-invalid-argument).

## UrsulaSaver

```python
class UrsulaSaver(SqliteSaver):
    conn: _Connection
    def __init__(
        self,
        db: UrsulaConnection,
        *,
        serde: SerializerProtocol | None = None,
    ) -> None: ...
    def close(self) -> None: ...
    def delete_thread(self, thread_id: str) -> None: ...
    @classmethod
    def from_conn_string(cls, conn_string: str) -> Any: ...
    def put(
        self,
        config: Any,
        checkpoint: Any,
        metadata: Any,
        new_versions: Any,
    ) -> Any: ...
    def put_writes(
        self,
        config: Any,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
        task_path: str = '',
    ) -> None: ...
    def setup(self) -> None: ...
```

```python
from sqlite_ursula.langgraph import UrsulaSaver

with UrsulaSaver(db) as saver:
    graph = builder.compile(checkpointer=saver)
    reply = graph.invoke(inputs, {"configurable": {"thread_id": "1"}})
send(reply)  # every checkpoint of the run is in Tonbo
```

LangGraph’s `SqliteSaver` on a database. `close()`, or `with`, closes its connection.

| Parameter | Type | Default | What it does |
| --- | --- | --- | --- |
| `db` | `UrsulaConnection` | Required | A connection `open()` made. The saver opens a connection of its own on the same database, and `db` stays yours: keep it open while the graph runs. Any other object raises `invalid_argument`. |
| `serde` | `SerializerProtocol \| None` | `None`: LangGraph's default serializer | LangGraph’s serializer for checkpoints, as SqliteSaver takes it. |

| Member | What it does |
| --- | --- |
| `close()` | Closes the saver’s own connection, after the call that holds it. `db` stays open. `with` calls it. |
| `delete_thread()` | Deletes a thread’s checkpoints and writes. Returns once `db.synced()` returned after it; a [`backlog_full`](https://tonbo.dev/docs/reference/python/errors.md#write-refused-backlog-full) refusal, which committed nothing, runs again after `synced()`. |
| `from_conn_string()` | Raises [`invalid_argument`](https://tonbo.dev/docs/reference/python/errors.md#setup-invalid-argument): the saver replicates a database, not a file. Use `UrsulaSaver(sqlite_ursula.open(url))`. |
| `put()` | Writes a checkpoint, as SqliteSaver’s. Returns once `db.synced()` returned after it; a [`backlog_full`](https://tonbo.dev/docs/reference/python/errors.md#write-refused-backlog-full) refusal, which committed nothing, runs again after `synced()`. |
| `put_writes()` | Writes a task’s pending writes. Returns once `db.synced()` returned after it; a [`backlog_full`](https://tonbo.dev/docs/reference/python/errors.md#write-refused-backlog-full) refusal, which committed nothing, runs again after `synced()`. |
| `setup()` | Creates LangGraph’s tables, once. It waits for nothing: the next write’s wait covers it. A statement refused with `backlog_full` waits and runs again by itself. |

## AsyncUrsulaSaver

```python
class AsyncUrsulaSaver(AsyncSqliteSaver):
    async def aclose(self) -> None: ...
    async def adelete_thread(self, thread_id: str) -> None: ...
    async def aput(
        self,
        config: Any,
        checkpoint: Any,
        metadata: Any,
        new_versions: Any,
    ) -> Any: ...
    async def aput_writes(
        self,
        config: Any,
        writes: Sequence[tuple[str, Any]],
        task_id: str,
        task_path: str = '',
    ) -> None: ...
    @classmethod
    def from_conn_string(cls, conn_string: str) -> Any: ...
    @classmethod
    async def open(
        cls,
        db: UrsulaConnection,
        *,
        serde: SerializerProtocol | None = None,
    ) -> Self: ...
    async def setup(self) -> None: ...
```

```python
from sqlite_ursula.langgraph import AsyncUrsulaSaver

db = await sqlite_ursula.aopen(url, token=key)
async with await AsyncUrsulaSaver.open(db) as saver:
    graph = builder.compile(checkpointer=saver)
    reply = await graph.ainvoke(inputs, {"configurable": {"thread_id": "1"}})
await send(reply)  # every checkpoint of the run is in Tonbo
```

LangGraph’s `AsyncSqliteSaver` on an aiosqlite connection of its own. Make it with `await AsyncUrsulaSaver.open(db)`, not its constructor, and close it with `aclose()` or `async with`. Its synchronous calls, which LangGraph runs from other threads, are these.

### open()

| Parameter | Type | Default | What it does |
| --- | --- | --- | --- |
| `db` | `UrsulaConnection` | Required | A connection `open()` or `aopen()` made. The saver opens an aiosqlite connection of its own on the same database, and `db` stays yours. |
| `serde` | `SerializerProtocol \| None` | `None`: LangGraph's default serializer | LangGraph’s serializer for checkpoints, as AsyncSqliteSaver takes it. |

| Member | What it does |
| --- | --- |
| `aclose()` | Closes the saver’s connection. `db` stays open. `async with` calls it. |
| `adelete_thread()` | Deletes a thread’s checkpoints and writes. Returns once `db.asynced()` returned after it; a `backlog_full` refusal runs again after `asynced()`. |
| `aput()` | Writes a checkpoint, as AsyncSqliteSaver’s. Returns once `db.asynced()` returned after it; a `backlog_full` refusal runs again after `asynced()`. |
| `aput_writes()` | Writes a task’s pending writes. Returns once `db.asynced()` returned after it; a `backlog_full` refusal runs again after `asynced()`. |
| `from_conn_string()` | Raises `invalid_argument`: use `await AsyncUrsulaSaver.open(sqlite_ursula.open(url))`. |
| `open()` | Makes the saver of a connection: the way to construct it. |
| `setup()` | Creates LangGraph’s tables, once, waiting for nothing. A statement refused with `backlog_full` waits on aiosqlite’s thread and runs again, which cancelling the caller does not end. |

## UrsulaStore

```python
class UrsulaStore(SqliteStore):
    conn: _Connection
    def __init__(
        self,
        db: UrsulaConnection,
        *,
        deserializer: Callable[[Any], dict[str, Any]] | None = None,
        index: Any = None,
        ttl: Any = None,
    ) -> None: ...
    def batch(self, ops: Iterable[Op]) -> list[Result]: ...
    def close(self) -> None: ...
    @classmethod
    def from_conn_string(cls, conn_string: str, **kwargs: Any) -> Any: ...
    def setup(self) -> None: ...
    def sweep_ttl(self) -> int: ...
```

```python
from sqlite_ursula.langgraph import UrsulaStore

with UrsulaStore(db) as store:
    store.setup()
    store.put(("users", "42"), "prefs", {"theme": "dark"})  # returns once Tonbo holds it
    item = store.get(("users", "42"), "prefs")
```

LangGraph’s `SqliteStore` on a database, on a connection without implicit transactions, as `SqliteStore` runs its own. `close()`, or `with`, stops its TTL sweeper and closes its connection.

| Parameter | Type | Default | What it does |
| --- | --- | --- | --- |
| `db` | `UrsulaConnection` | Required | A connection `open()` made. The store opens a connection of its own on the same database, without implicit transactions, and `db` stays yours. |
| `deserializer` | `Callable[[Any], dict[str, Any]] \| None` | `None`: JSON | Turns a stored value back into a dict, as SqliteStore takes it. |
| `index` | `Any` | `None`: no vector search | SqliteStore's index configuration, for semantic search over the values. |
| `ttl` | `Any` | `None`: items never expire | SqliteStore’s TTL configuration. With `refresh_on_read`, a read that refreshes TTLs writes, so it waits for Tonbo as a write does. |

| Member | What it does |
| --- | --- |
| `batch()` | Runs a batch of operations, as SqliteStore’s; `put`, `get`, `search` and `delete` go through it. A batch that writes (a put, a delete, or a get or search that refreshes TTLs) Returns once `db.synced()` returned after it; a [`backlog_full`](https://tonbo.dev/docs/reference/python/errors.md#write-refused-backlog-full) refusal, which committed nothing, runs again after `synced()`. A batch that only reads waits for nothing. |
| `close()` | Stops the TTL sweeper, waiting at most a second for it, and closes the store’s connection. `db` stays open. `with` calls it. |
| `from_conn_string()` | Raises `invalid_argument`: use `UrsulaStore(sqlite_ursula.open(url))`. |
| `setup()` | Runs the store’s migrations, once, waiting for nothing. A statement refused with `backlog_full` waits and runs again by itself, since the migrations are not safe to run twice. |
| `sweep_ttl()` | Deletes the expired items and returns how many. Returns once `db.synced()` returned after it; a [`backlog_full`](https://tonbo.dev/docs/reference/python/errors.md#write-refused-backlog-full) refusal, which committed nothing, runs again after `synced()`. |

## AsyncUrsulaStore

```python
class AsyncUrsulaStore(AsyncSqliteStore):
    async def aclose(self) -> None: ...
    async def adelete(self, *args: Any, **kwargs: Any) -> None: ...
    async def aget(self, *args: Any, **kwargs: Any) -> Any: ...
    async def alist_namespaces(self, *args: Any, **kwargs: Any) -> Any: ...
    async def aput(self, *args: Any, **kwargs: Any) -> None: ...
    async def asearch(self, *args: Any, **kwargs: Any) -> Any: ...
    @classmethod
    def from_conn_string(cls, conn_string: str, **kwargs: Any) -> Any: ...
    @classmethod
    async def open(
        cls,
        db: UrsulaConnection,
        *,
        deserializer: Callable[[Any], dict[str, Any]] | None = None,
        index: Any = None,
        ttl: Any = None,
    ) -> Self: ...
    async def setup(self) -> None: ...
    async def sweep_ttl(self) -> int: ...
```

```python
from sqlite_ursula.langgraph import AsyncUrsulaStore

async with await AsyncUrsulaStore.open(db) as store:
    await store.setup()
    await store.aput(("users", "42"), "prefs", {"theme": "dark"})
```

LangGraph’s `AsyncSqliteStore` on an aiosqlite connection of its own. Make it with `await AsyncUrsulaStore.open(db)`, and close it with `aclose()` or `async with`, which also stop its TTL sweeper. The waits are the callers’, so a read never waits behind another call’s write. `abatch()` and `batch()`, which take operations, are LangGraph’s own and do not wait: call `await db.asynced()` after them.

### open()

| Parameter | Type | Default | What it does |
| --- | --- | --- | --- |
| `db` | `UrsulaConnection` | Required | A connection `open()` or `aopen()` made. The store opens an aiosqlite connection of its own on the same database, and `db` stays yours. |
| `deserializer` | `Callable[[Any], dict[str, Any]] \| None` | `None`: JSON | Turns a stored value back into a dict, as AsyncSqliteStore takes it. |
| `index` | `Any` | `None`: no vector search | AsyncSqliteStore's index configuration, for semantic search over the values. |
| `ttl` | `Any` | `None`: items never expire | AsyncSqliteStore’s TTL configuration. With `refresh_on_read`, a read that refreshes TTLs waits for Tonbo as a write does. |

| Member | What it does |
| --- | --- |
| `aclose()` | Stops the TTL sweeper, cancelling it after a second, and closes the store’s connection. `db` stays open. `async with` calls it. |
| `adelete()` | Deletes an item. Returns once `db.asynced()` returned after it; a `backlog_full` refusal runs again after `asynced()`. |
| `aget()` | Reads an item. It waits as a write does only when it refreshes TTLs (the store’s `ttl` sets `refresh_on_read`, and `refresh_ttl` is not `False`). |
| `alist_namespaces()` | Lists namespaces. It never waits. |
| `aput()` | Puts an item. Returns once `db.asynced()` returned after it; a `backlog_full` refusal runs again after `asynced()`. |
| `asearch()` | Searches items, waiting only when it refreshes TTLs, as `aget`. |
| `from_conn_string()` | Raises `invalid_argument`: use `await AsyncUrsulaStore.open(sqlite_ursula.open(url))`. |
| `open()` | Makes the store of a connection: the way to construct it. |
| `setup()` | Runs the store’s migrations, once, waiting for nothing. A refused statement waits on aiosqlite’s thread and runs again. |
| `sweep_ttl()` | Deletes the expired items and returns how many. Returns once `db.asynced()` returned after it; a `backlog_full` refusal runs again after `asynced()`. |

For a first graph on a database, see [LangGraph](https://tonbo.dev/docs/databases/python.md#langgraph) in Connect from Python.