osyra_ome.adapters.langgraph

Private pre-release / network boundary: The currently usable customer surface is offline verification of an existing signed artifact against an out-of-band trust anchor. Memory/network RPC APIs and host snippets on this page describe protocol and interface shape; the public Edge bridge, package publication, and release acceptance remain pending. Do not treat api.osyra.ai:443 examples as a released customer route.

LangGraph adapter — register OME as a LangGraph checkpoint saver.

Exports:

• :class:`OmeCheckpointSaver` — sync ``BaseCheckpointSaver`` for use with graph_builder.compile(checkpointer=...) on a sync :class:osyra_ome.OmeClient.
• :class:`OmeAsyncCheckpointSaver` — async variant for :class:osyra_ome.OmeAsyncClient.

Optional install::

pip install 'osyra-ome[langgraph]'

Classes

class OmeAsyncCheckpointSaver

Bases: BaseCheckpointSaver

Async LangGraph checkpoint saver backed by OME.

Wraps :class:`osyra_ome.OmeAsyncClient`. Mirrors the sync class method-for-method via aput / aget_tuple / alist / aput_writes.

Methods

__init__

def __init__(self, *, client: OmeAsyncClient, workspace_id: str | None = None) -> None

adelete_thread (async)

async def adelete_thread(self, thread_id: str) -> None

adopt_claim_ids

def adopt_claim_ids(self, thread_id: str, claim_ids: Sequence[str]) -> None

aget_tuple (async)

async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None

alist (async)

async def alist(self, config: RunnableConfig | None, *, filter: dict[str, Any] | None = None, before: RunnableConfig | None = None, limit: int | None = None) -> AsyncIterator[CheckpointTuple]

aput (async)

async def aput(self, config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: dict[str, Any]) -> RunnableConfig

aput_writes (async)

async def aput_writes(self, config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str, task_path: str = '') -> None

thread_claim_ids

def thread_claim_ids(self, thread_id: str) -> tuple[str, ...]

workspace_id @property

def workspace_id(self) -> str

class OmeCheckpointSaver

Bases: BaseCheckpointSaver

LangGraph checkpoint saver backed by the OME memory namespace.

Args: client: a :class:osyra_ome.OmeClient. workspace_id: OME workspace UUID-like id; if omitted, defers to the workspace_id bound on the wrapped client.

State model: The saver keeps a per-thread_id list of claim ids in memory. :meth:get_tuple returns the most recent checkpoint; :meth:list returns all checkpoints for the thread (newest first). Cross-process replay is supported via the OME service — but a fresh saver instance does NOT automatically know about claims written by a previous instance for the same thread. Customers running multi- process inference farms should wire their own thread_id → claim_ids mapping (typically Redis) and seed the saver with :meth:adopt_claim_ids.

Methods

__init__

def __init__(self, *, client: OmeClient, workspace_id: str | None = None) -> None

adopt_claim_ids

def adopt_claim_ids(self, thread_id: str, claim_ids: Sequence[str]) -> None

Seed the saver with externally-known CHECKPOINT claim ids.

Used to bridge multi-process inference where a prior worker wrote checkpoints and a new worker resuming the thread needs to know about them. Order matters: pass the claim ids in insertion order (oldest first) so :meth:`get_tuple` returns the most recent one as the last element.

ADAPT2-7: merges (append-and-dedupe in insertion order) rather than overwriting, so seeding externally-known ids does NOT drop checkpoints this saver instance already wrote for the thread.

delete_thread

def delete_thread(self, thread_id: str) -> None

get_tuple

def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None

list

def list(self, config: RunnableConfig | None, *, filter: dict[str, Any] | None = None, before: RunnableConfig | None = None, limit: int | None = None) -> Iterator[CheckpointTuple]

put

def put(self, config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: dict[str, Any]) -> RunnableConfig

put_writes

def put_writes(self, config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str, task_path: str = '') -> None

thread_claim_ids

def thread_claim_ids(self, thread_id: str) -> tuple[str, ...]

workspace_id @property

def workspace_id(self) -> str