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, AsyncUrsulaSaver, UrsulaStore and AsyncUrsulaStore.

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.

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() before acting on one.
Reads do not wait, and neither does setup(). A reset raises the UrsulaResetError 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.

UrsulaSaver

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: ...

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` refusal, which committed nothing, runs again after `synced()`.
`from_conn_string()`	Raises `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` 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` 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

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: ...

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

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: ...

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` 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` refusal, which committed nothing, runs again after `synced()`.

AsyncUrsulaStore

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: ...

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 in Connect from Python.