aidputils.agents.toolkit.memory_helper

class aidputils.agents.toolkit.memory_helper.AbstractCheckpointSaver

Bases: ABC

Local abstract interface for Oracle Memory checkpointers.

This is not LangGraph’s BaseCheckpointSaver; it is a project-level interface that combines the public surface area of: - ProxyCheckpointClient (sync) - AsyncProxyCheckpointClient (async)

It exists so code in this repo can depend on a single type that supports either sync or async execution.

health() → Dict[str, Any]

get_tuple(config: Dict[str, Any], kwargs: Dict[str, Any] | None = None) → Dict[str, Any] | None

list(config: Dict[str, Any] | None, filter: Dict[str, Any] | None = None, before: Dict[str, Any] | None = None, limit: int | None = None, kwargs: Dict[str, Any] | None = None) → List[Dict[str, Any]]

put(config: Dict[str, Any], checkpoint: Dict[str, Any], metadata: Dict[str, Any], new_versions: Dict[str, Any], kwargs: Dict[str, Any] | None = None) → Dict[str, Any]

put_writes(*args, **options)

delete_thread(thread_id: str) → Dict[str, Any]

delete(config_or_thread: Any) → Dict[str, Any]

async ahealth() → Dict[str, Any]

async aget_tuple(config: Dict[str, Any], kwargs: Dict[str, Any] | None = None) → Dict[str, Any] | None

async alist(config: Dict[str, Any] | None, filter: Dict[str, Any] | None = None, before: Dict[str, Any] | None = None, limit: int | None = None, kwargs: Dict[str, Any] | None = None) → List[Dict[str, Any]]

async aput(config: Dict[str, Any], checkpoint: Dict[str, Any], metadata: Dict[str, Any], new_versions: Dict[str, Any], kwargs: Dict[str, Any] | None = None) → Dict[str, Any]

async aput_writes(*args, **options)

async adelete_thread(thread_id: str) → Dict[str, Any]

async adelete(config_or_thread: Any) → Dict[str, Any]

async aclose() → None

class aidputils.agents.toolkit.memory_helper.InMemoryCheckpointSaver(*args: Any, **kwargs: Any)

Bases: InMemorySaver, AbstractCheckpointSaver

In-memory checkpointer that satisfies our project-level AbstractCheckpointSaver.

This is mainly useful for local/dev/test runs where you want LangGraph-compatible checkpointing without a backing memory server.

health() → Dict[str, Any]

async ahealth() → Dict[str, Any]

delete(config_or_thread: Any) → Dict[str, Any]

Delete a thread by thread_id or config.

Mirrors the convenience helpers on ProxyCheckpointClient / AsyncProxyCheckpointClient.

async adelete(config_or_thread: Any) → Dict[str, Any]

async aclose() → None

class aidputils.agents.toolkit.memory_helper.CheckpointerType(*values)

Bases: StrEnum

SYNC = 'SYNC'

ASYNC = 'ASYNC'

IN_MEMORY = 'IN_MEMORY'

aidputils.agents.toolkit.memory_helper.create_checkpoint_saver(agent_name: str, platform_name: str = 'langgraph', checkpointer_type: str = 'ASYNC', base_url: str = 'http://localhost:21100/') → AbstractCheckpointSaver | None