harnest.store

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`CustomStorage`	Class
`MemoryStore`	Class
`PostgresStore`	Class
`RedisStore`	Class
`StorageRegistry`	Class

CustomStorage

Lifecycle-owned application storage exposed by a stable authored name.

CustomStorage(*args, **kwargs)

Example

from harnest.store import CustomStorage

class MyCustomStorage(CustomStorage):
    ...

CustomStorage.start

async CustomStorage.start() -> None

Open the underlying storage resource.

Example

from harnest.store import CustomStorage

storage: CustomStorage
await storage.start()

CustomStorage.close

async CustomStorage.close() -> None

Release the underlying storage resource.

Example

from harnest.store import CustomStorage

storage: CustomStorage
await storage.close()

MemoryStore

Process-local session, checkpoint, and continuation store with atomic CAS.

MemoryStore() -> None

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`'harnest'`
`framework`	`str \| None`	`None`
`schema_id`	`str`	`'harnest-checkpoint/v1'`

Example

from harnest.store import MemoryStore

store = MemoryStore()

MemoryStore.start

async MemoryStore.start() -> None

Reject reopening because closed development state is intentionally lost.

Example

from harnest.store import MemoryStore

store: MemoryStore
await store.start()

MemoryStore.create

async MemoryStore.create(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord

Create one process-local tenant-scoped session.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state`	`Mapping[str, Any]`	Required

from harnest.store import MemoryStore

store: MemoryStore
record = await store.create(session_id=..., user_id=..., state=...)

MemoryStore.get

async MemoryStore.get(*, session_id: str, user_id: str) -> SessionRecord | None

Read one process-local tenant-scoped session.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
record = await store.get(session_id=..., user_id=...)

MemoryStore.list

async MemoryStore.list(*, user_id: str, after: str | None = None, limit: int | None = None) -> Sequence[SessionRecord]

List an ordered keyset page without materializing unrelated users.

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int \| None`	`None`

from harnest.store import MemoryStore

store: MemoryStore
result = await store.list(user_id=...)

MemoryStore.update

async MemoryStore.update(*, session_id: str, user_id: str, state_delta: Mapping[str, Any]) -> SessionRecord | None

Merge a state delta into a process-local session.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state_delta`	`Mapping[str, Any]`	Required

from harnest.store import MemoryStore

store: MemoryStore
record = await store.update(session_id=..., user_id=..., state_delta=...)

MemoryStore.delete

async MemoryStore.delete(*, session_id: str, user_id: str) -> bool

Delete one process-local session after its active lease exits.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.delete(session_id=..., user_id=...)

MemoryStore.acquire

MemoryStore.acquire(*, session_id: str, user_id: str) -> AsyncIterator[SessionLease]

Hold exclusive access to one process-local session.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
async with store.acquire(session_id=..., user_id=...):
    ...

MemoryStore.close

async MemoryStore.close() -> None

Close both session and checkpoint sides of the shared store.

Example

from harnest.store import MemoryStore

store: MemoryStore
await store.close()

MemoryStore.as_langgraph_checkpointer

MemoryStore.as_langgraph_checkpointer() -> Any

Return one stable native saver for advanced LangGraph construction.

Example

from harnest.store import MemoryStore

store: MemoryStore
result = store.as_langgraph_checkpointer()

MemoryStore.put_a2a_task

async MemoryStore.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord

Upsert one owner-scoped snapshot without allowing context reassignment.

Parameters and example

Parameter	Type	Default
`record`	`A2ATaskRecord`	Required

from harnest.store import MemoryStore

store: MemoryStore
record = await store.put_a2a_task(record=...)

MemoryStore.get_a2a_task

async MemoryStore.get_a2a_task(*, application_id: str, user_id: str, task_id: str) -> A2ATaskRecord | None

Read one snapshot only through its complete ownership key.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
record = await store.get_a2a_task(application_id=..., user_id=..., task_id=...)

MemoryStore.list_a2a_tasks

async MemoryStore.list_a2a_tasks(
    *,
    application_id: str,
    user_id: str,
    context_id: str | None = None,
    status: int | None = None,
    status_timestamp_after: str | None = None,
    cursor_task_id: str | None = None,
    limit: int = 51,
) -> A2ATaskPage

Return a stable inclusive-cursor page for the process-local backend.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`context_id`	`str \| None`	`None`
`status`	`int \| None`	`None`
`status_timestamp_after`	`str \| None`	`None`
`cursor_task_id`	`str \| None`	`None`
`limit`	`int`	`51`

from harnest.store import MemoryStore

store: MemoryStore
page = await store.list_a2a_tasks(application_id=..., user_id=...)

MemoryStore.delete_a2a_task

async MemoryStore.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool

Delete exactly one owner-scoped A2A snapshot.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.delete_a2a_task(application_id=..., user_id=..., task_id=...)

MemoryStore.begin_run

async MemoryStore.begin_run(
    *,
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
) -> RunRecord

Claim the session for one idempotent invocation run.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`run_id`	`str`	Required
`framework`	`Literal['adk', 'langgraph']`	Required

from harnest.store import MemoryStore

store: MemoryStore
record = await store.begin_run(
    application_id=...,
    user_id=...,
    session_id=...,
    run_id=...,
    framework=...,
)

MemoryStore.get_run

async MemoryStore.get_run(*, scope: RunScope) -> RunRecord | None

Read one process-local run through its complete ownership scope.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.store import MemoryStore

store: MemoryStore
record = await store.get_run(scope=...)

MemoryStore.get_checkpoint

async MemoryStore.get_checkpoint(
    *,
    scope: RunScope,
    checkpoint_id: str | None = None,
    namespace: str = '',
) -> CheckpointRecord | None

Load an exact checkpoint or the newest checkpoint in a namespace.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str \| None`	`None`
`namespace`	`str`	`''`

from harnest.store import MemoryStore

store: MemoryStore
record = await store.get_checkpoint(scope=...)

MemoryStore.list_checkpoints

MemoryStore.list_checkpoints(
    *,
    scope: RunScope,
    namespace: str | None = None,
    before: str | None = None,
    limit: int = 20,
) -> AsyncIterator[CheckpointRecord]

Yield a bounded newest-first checkpoint history.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`namespace`	`str \| None`	`None`
`before`	`str \| None`	`None`
`limit`	`int`	`20`

from harnest.store import MemoryStore

store: MemoryStore
async for item in store.list_checkpoints(scope=...):
    ...

MemoryStore.put

async MemoryStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord

Persist a checkpoint only when its expected revision still matches.

Parameters and example

Parameter	Type	Default
`checkpoint`	`CheckpointRecord`	Required
`scope`	`RunScope`	Required
`expected_revision`	`int \| None`	Required

from harnest.store import MemoryStore

store: MemoryStore
record = await store.put(checkpoint=..., scope=..., expected_revision=...)

MemoryStore.put_writes

async MemoryStore.put_writes(*, scope: RunScope, checkpoint_id: str, writes: Sequence[CheckpointWrite]) -> None

Add pending writes idempotently by task and channel.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required
`writes`	`Sequence[CheckpointWrite]`	Required

from harnest.store import MemoryStore

store: MemoryStore
await store.put_writes(scope=..., checkpoint_id=..., writes=...)

MemoryStore.get_writes

async MemoryStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]

Read process-local pending writes for one owned checkpoint.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.get_writes(scope=..., checkpoint_id=...)

MemoryStore.get_writes_batch

async MemoryStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]

Load writes for several checkpoints without per-checkpoint calls.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_ids`	`Sequence[str]`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.get_writes_batch(scope=..., checkpoint_ids=...)

MemoryStore.transition

async MemoryStore.transition(
    *,
    scope: RunScope,
    expected_status: RunStatus,
    status: RunStatus,
    pending_action: PendingAction | None = None,
) -> RunRecord

Apply one valid compare-and-swap run-state transition.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`expected_status`	`RunStatus`	Required
`status`	`RunStatus`	Required
`pending_action`	`PendingAction \| None`	`None`

from harnest.store import MemoryStore

store: MemoryStore
record = await store.transition(scope=..., expected_status=..., status=...)

MemoryStore.suspend_continuation

async MemoryStore.suspend_continuation(*, record: Any, external_id: str) -> Any

Atomically register provider work and move its owned run to waiting.

Parameters and example

Parameter	Type	Default
`record`	`Any`	Required
`external_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.suspend_continuation(record=..., external_id=...)

MemoryStore.get_continuation

async MemoryStore.get_continuation(*, scope: RunScope, continuation_id: str) -> Any | None

Return one record only when every run ownership field matches.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`continuation_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.get_continuation(scope=..., continuation_id=...)

MemoryStore.get_provider_continuation

async MemoryStore.get_provider_continuation(*, scope: RunScope, provider: str, continuation_id: str) -> Any | None

Return an exact continuation together with its provider external ID.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.get_provider_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
)

MemoryStore.get_continuation_by_external_id

async MemoryStore.get_continuation_by_external_id(*, application_id: str, provider: str, external_id: str) -> Any | None

Resolve one provider identity through the same indexed ownership key.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`external_id`	`str`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.get_continuation_by_external_id(
    application_id=...,
    provider=...,
    external_id=...,
)

MemoryStore.list_pending_continuations

async MemoryStore.list_pending_continuations(
    *,
    application_id: str,
    provider: str,
    after: str | None = None,
    limit: int = 100,
) -> Sequence[Any]

Return a bounded provider page for local reconciliation.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.store import MemoryStore

store: MemoryStore
result = await store.list_pending_continuations(application_id=..., provider=...)

MemoryStore.resolve_continuation

async MemoryStore.resolve_continuation(
    *,
    scope: RunScope,
    provider: str,
    external_id: str,
    schema_id: str,
    result: Any = None,
    failure: Any = None,
) -> Any

Commit a provider result or failure with one status compare-and-swap.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`external_id`	`str`	Required
`schema_id`	`str`	Required
`result`	`Any`	`None`
`failure`	`Any`	`None`

from harnest.store import MemoryStore

store: MemoryStore
result = await store.resolve_continuation(
    scope=...,
    provider=...,
    external_id=...,
    schema_id=...,
)

MemoryStore.cancel_continuation

async MemoryStore.cancel_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
    failure: Any,
) -> Any

Fail one pending continuation and cancel its exact waiting run.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required
`failure`	`Any`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.cancel_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
    failure=...,
)

MemoryStore.claim_continuation

async MemoryStore.claim_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> Any

Claim an outcome and return the waiting run to running atomically.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.claim_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

MemoryStore.arm_continuation

async MemoryStore.arm_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> Any

Arm a provider wait after its native suspension is checkpointed.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest.store import MemoryStore

store: MemoryStore
result = await store.arm_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

MemoryStore.delete_run

async MemoryStore.delete_run(*, scope: RunScope) -> None

Delete a run and every private checkpoint owned by it.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.store import MemoryStore

store: MemoryStore
await store.delete_run(scope=...)

PostgresStore

Durable combined store using asyncpg and one Harnest-owned schema. The driver is imported lazily by start, keeping module discovery independent from database connectivity.

PostgresStore(
    dsn: str,
    *,
    pool_options: Mapping[str, Any] | None = None,
    lease_pool_options: Mapping[str, Any] | None = None,
    setup_schema: bool = True,
) -> None

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`'harnest'`
`framework`	`str \| None`	`None`
`schema_id`	`str`	`'harnest-checkpoint/v1'`

Parameters and example

Parameter	Type	Default
`dsn`	`str`	Required
`pool_options`	`Mapping[str, Any] \| None`	`None`
`lease_pool_options`	`Mapping[str, Any] \| None`	`None`
`setup_schema`	`bool`	`True`

from harnest.store import PostgresStore

store = PostgresStore(dsn=...)

PostgresStore.as_langgraph_checkpointer

PostgresStore.as_langgraph_checkpointer() -> Any

Return one stable native saver for advanced LangGraph construction.

Example

from harnest.store import PostgresStore

store: PostgresStore
result = store.as_langgraph_checkpointer()

PostgresStore.start

async PostgresStore.start() -> None

Open the pool and create or validate the Harnest-owned schema.

Example

from harnest.store import PostgresStore

store: PostgresStore
await store.start()

PostgresStore.create

async PostgresStore.create(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord

Create one tenant-scoped session without overwriting conflicts.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state`	`Mapping[str, Any]`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.create(session_id=..., user_id=..., state=...)

PostgresStore.get

async PostgresStore.get(*, session_id: str, user_id: str) -> SessionRecord | None

Read one tenant-scoped session from PostgreSQL.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.get(session_id=..., user_id=...)

PostgresStore.list

async PostgresStore.list(*, user_id: str, after: str | None = None, limit: int | None = None) -> Sequence[SessionRecord]

Apply tenant filtering and optional keyset pagination in PostgreSQL.

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int \| None`	`None`

from harnest.store import PostgresStore

store: PostgresStore
result = await store.list(user_id=...)

PostgresStore.put_a2a_task

async PostgresStore.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord

Upsert an owner-scoped protobuf snapshot with immutable context.

Parameters and example

Parameter	Type	Default
`record`	`A2ATaskRecord`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.put_a2a_task(record=...)

PostgresStore.get_a2a_task

async PostgresStore.get_a2a_task(*, application_id: str, user_id: str, task_id: str) -> A2ATaskRecord | None

Read one A2A task using its complete application and owner scope.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.get_a2a_task(application_id=..., user_id=..., task_id=...)

PostgresStore.list_a2a_tasks

async PostgresStore.list_a2a_tasks(
    *,
    application_id: str,
    user_id: str,
    context_id: str | None = None,
    status: int | None = None,
    status_timestamp_after: str | None = None,
    cursor_task_id: str | None = None,
    limit: int = 51,
) -> A2ATaskPage

Apply A2A filters, ordering, and keyset pagination in PostgreSQL.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`context_id`	`str \| None`	`None`
`status`	`int \| None`	`None`
`status_timestamp_after`	`str \| None`	`None`
`cursor_task_id`	`str \| None`	`None`
`limit`	`int`	`51`

from harnest.store import PostgresStore

store: PostgresStore
page = await store.list_a2a_tasks(application_id=..., user_id=...)

PostgresStore.delete_a2a_task

async PostgresStore.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool

Delete one exact A2A owner record without revealing foreign tasks.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
result = await store.delete_a2a_task(application_id=..., user_id=..., task_id=...)

PostgresStore.update

async PostgresStore.update(*, session_id: str, user_id: str, state_delta: Mapping[str, Any]) -> SessionRecord | None

Atomically merge a state delta in PostgreSQL.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state_delta`	`Mapping[str, Any]`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.update(session_id=..., user_id=..., state_delta=...)

PostgresStore.delete

async PostgresStore.delete(*, session_id: str, user_id: str) -> bool

Delete one tenant-scoped session and report whether it existed.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
result = await store.delete(session_id=..., user_id=...)

PostgresStore.acquire

PostgresStore.acquire(*, session_id: str, user_id: str) -> AsyncIterator[SessionLease]

Hold a cross-replica advisory lock for one session execution.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
async with store.acquire(session_id=..., user_id=...):
    ...

PostgresStore.begin_run

async PostgresStore.begin_run(
    *,
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
) -> RunRecord

Create an idempotent run while allowing one active run per session.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`run_id`	`str`	Required
`framework`	`Literal['adk', 'langgraph']`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.begin_run(
    application_id=...,
    user_id=...,
    session_id=...,
    run_id=...,
    framework=...,
)

PostgresStore.get_run

async PostgresStore.get_run(*, scope: RunScope) -> RunRecord | None

Read one PostgreSQL checkpoint run through its ownership scope.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.get_run(scope=...)

PostgresStore.get_checkpoint

async PostgresStore.get_checkpoint(
    *,
    scope: RunScope,
    checkpoint_id: str | None = None,
    namespace: str = '',
) -> CheckpointRecord | None

Load an exact or newest checkpoint entirely in PostgreSQL.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str \| None`	`None`
`namespace`	`str`	`''`

from harnest.store import PostgresStore

store: PostgresStore
record = await store.get_checkpoint(scope=...)

PostgresStore.list_checkpoints

PostgresStore.list_checkpoints(
    *,
    scope: RunScope,
    namespace: str | None = None,
    before: str | None = None,
    limit: int = 20,
) -> AsyncIterator[CheckpointRecord]

Push checkpoint filtering, ordering, cursor, and limit into SQL.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`namespace`	`str \| None`	`None`
`before`	`str \| None`	`None`
`limit`	`int`	`20`

from harnest.store import PostgresStore

store: PostgresStore
async for item in store.list_checkpoints(scope=...):
    ...

PostgresStore.put

async PostgresStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord

Allocate a serialized revision and write only on matching CAS state.

Parameters and example

Parameter	Type	Default
`checkpoint`	`CheckpointRecord`	Required
`scope`	`RunScope`	Required
`expected_revision`	`int \| None`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.put(checkpoint=..., scope=..., expected_revision=...)

PostgresStore.put_writes

async PostgresStore.put_writes(*, scope: RunScope, checkpoint_id: str, writes: Sequence[CheckpointWrite]) -> None

Insert pending writes as one set and ignore existing identities.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required
`writes`	`Sequence[CheckpointWrite]`	Required

from harnest.store import PostgresStore

store: PostgresStore
await store.put_writes(scope=..., checkpoint_id=..., writes=...)

PostgresStore.get_writes

async PostgresStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]

Read pending writes for one owned PostgreSQL checkpoint.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
result = await store.get_writes(scope=..., checkpoint_id=...)

PostgresStore.get_writes_batch

async PostgresStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]

Fetch pending writes for several checkpoints in one query.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_ids`	`Sequence[str]`	Required

from harnest.store import PostgresStore

store: PostgresStore
result = await store.get_writes_batch(scope=..., checkpoint_ids=...)

PostgresStore.suspend_continuation

async PostgresStore.suspend_continuation(*, record: ContinuationRecord, external_id: str) -> ContinuationRecord

Atomically persist a provider wait and mark its owned run waiting.

Parameters and example

Parameter	Type	Default
`record`	`ContinuationRecord`	Required
`external_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.suspend_continuation(record=..., external_id=...)

PostgresStore.get_continuation

async PostgresStore.get_continuation(*, scope: RunScope, continuation_id: str) -> ContinuationRecord | None

Load one continuation with its complete ownership predicate in SQL.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`continuation_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.get_continuation(scope=..., continuation_id=...)

PostgresStore.get_provider_continuation

async PostgresStore.get_provider_continuation(*, scope: RunScope, provider: str, continuation_id: str) -> ProviderPendingContinuation | None

Load one exact provider wait and its private external identity.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
continuation = await store.get_provider_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
)

PostgresStore.get_continuation_by_external_id

async PostgresStore.get_continuation_by_external_id(
    *,
    application_id: str,
    provider: str,
    external_id: str,
) -> ProviderPendingContinuation | None

Use the unique provider key so replica callbacks never scan waits.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`external_id`	`str`	Required

from harnest.store import PostgresStore

store: PostgresStore
continuation = await store.get_continuation_by_external_id(
    application_id=...,
    provider=...,
    external_id=...,
)

PostgresStore.list_pending_continuations

async PostgresStore.list_pending_continuations(
    *,
    application_id: str,
    provider: str,
    after: str | None = None,
    limit: int = 100,
) -> Sequence[ProviderPendingContinuation]

Push provider reconciliation filtering and pagination into PostgreSQL.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.store import PostgresStore

store: PostgresStore
result = await store.list_pending_continuations(application_id=..., provider=...)

PostgresStore.resolve_continuation

async PostgresStore.resolve_continuation(
    *,
    scope: RunScope,
    provider: str,
    external_id: str,
    schema_id: str,
    result: Any = None,
    failure: ContinuationFailure | None = None,
) -> ContinuationRecord

Commit one validated provider outcome with an indexed SQL CAS.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`external_id`	`str`	Required
`schema_id`	`str`	Required
`result`	`Any`	`None`
`failure`	`ContinuationFailure \| None`	`None`

from harnest.store import PostgresStore

store: PostgresStore
record = await store.resolve_continuation(
    scope=...,
    provider=...,
    external_id=...,
    schema_id=...,
)

PostgresStore.claim_continuation

async PostgresStore.claim_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord

Claim an outcome and resume its run in one PostgreSQL statement.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.claim_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

PostgresStore.cancel_continuation

async PostgresStore.cancel_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
    failure: ContinuationFailure,
) -> ContinuationRecord

Fail a pending continuation and cancel its waiting run atomically.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required
`failure`	`ContinuationFailure`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.cancel_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
    failure=...,
)

PostgresStore.arm_continuation

async PostgresStore.arm_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord

Arm one wait with a SQL CAS after native checkpoint persistence.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest.store import PostgresStore

store: PostgresStore
record = await store.arm_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

PostgresStore.transition

async PostgresStore.transition(
    *,
    scope: RunScope,
    expected_status: RunStatus,
    status: RunStatus,
    pending_action: PendingAction | None = None,
) -> RunRecord

Apply a validated status transition with one SQL compare-and-swap.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`expected_status`	`RunStatus`	Required
`status`	`RunStatus`	Required
`pending_action`	`PendingAction \| None`	`None`

from harnest.store import PostgresStore

store: PostgresStore
record = await store.transition(scope=..., expected_status=..., status=...)

PostgresStore.delete_run

async PostgresStore.delete_run(*, scope: RunScope) -> None

Delete a run and cascade its private checkpoint data.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.store import PostgresStore

store: PostgresStore
await store.delete_run(scope=...)

PostgresStore.close

async PostgresStore.close() -> None

Close only pools created and owned by this store.

Example

from harnest.store import PostgresStore

store: PostgresStore
await store.close()

RedisStore

Combined Redis store with atomic CAS and renewable execution leases. Production durability depends on Redis persistence, replication, and failover configuration; Postgres remains the reference durable backend.

RedisStore(
    url: str,
    *,
    prefix: str = 'harnest',
    session_ttl_seconds: int | None = None,
    checkpoint_ttl_seconds: int = 604800,
    lease_seconds: int = 60,
    lease_wait_seconds: float = 30,
    client_options: Mapping[str, Any] | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`'harnest'`
`framework`	`str \| None`	`None`
`schema_id`	`str`	`'harnest-checkpoint/v1'`

Parameters and example

Parameter	Type	Default
`url`	`str`	Required
`prefix`	`str`	`'harnest'`
`session_ttl_seconds`	`int \| None`	`None`
`checkpoint_ttl_seconds`	`int`	`604800`
`lease_seconds`	`int`	`60`
`lease_wait_seconds`	`float`	`30`
`client_options`	`Mapping[str, Any] \| None`	`None`

from harnest.store import RedisStore

store = RedisStore(url=...)

RedisStore.as_langgraph_checkpointer

RedisStore.as_langgraph_checkpointer() -> Any

Return one stable native saver for advanced LangGraph construction.

Example

from harnest.store import RedisStore

store: RedisStore
result = store.as_langgraph_checkpointer()

RedisStore.start

async RedisStore.start() -> None

Open the client and establish the compatible schema marker.

Example

from harnest.store import RedisStore

store: RedisStore
await store.start()

RedisStore.create

async RedisStore.create(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord

Create one tenant-scoped session with an atomic Lua mutation.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state`	`Mapping[str, Any]`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.create(session_id=..., user_id=..., state=...)

RedisStore.get

async RedisStore.get(*, session_id: str, user_id: str) -> SessionRecord | None

Read one tenant-scoped session from Redis.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.get(session_id=..., user_id=...)

RedisStore.list

async RedisStore.list(*, user_id: str, after: str | None = None, limit: int | None = None) -> Sequence[SessionRecord]

Read one lexicographically ordered keyset page with one MGET.

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int \| None`	`None`

from harnest.store import RedisStore

store: RedisStore
result = await store.list(user_id=...)

RedisStore.put_a2a_task

async RedisStore.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord

Atomically update one A2A record and all of its query indexes.

Parameters and example

Parameter	Type	Default
`record`	`A2ATaskRecord`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.put_a2a_task(record=...)

RedisStore.get_a2a_task

async RedisStore.get_a2a_task(*, application_id: str, user_id: str, task_id: str) -> A2ATaskRecord | None

Read one task through its digested application and owner key.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.get_a2a_task(application_id=..., user_id=..., task_id=...)

RedisStore.list_a2a_tasks

async RedisStore.list_a2a_tasks(
    *,
    application_id: str,
    user_id: str,
    context_id: str | None = None,
    status: int | None = None,
    status_timestamp_after: str | None = None,
    cursor_task_id: str | None = None,
    limit: int = 51,
) -> A2ATaskPage

Use one prefiltered sorted index and one batched record read.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`context_id`	`str \| None`	`None`
`status`	`int \| None`	`None`
`status_timestamp_after`	`str \| None`	`None`
`cursor_task_id`	`str \| None`	`None`
`limit`	`int`	`51`

from harnest.store import RedisStore

store: RedisStore
page = await store.list_a2a_tasks(application_id=..., user_id=...)

RedisStore.delete_a2a_task

async RedisStore.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool

Delete a record and every selected-list index with optimistic CAS.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
result = await store.delete_a2a_task(application_id=..., user_id=..., task_id=...)

RedisStore.update

async RedisStore.update(*, session_id: str, user_id: str, state_delta: Mapping[str, Any]) -> SessionRecord | None

Atomically merge state while preserving session expiry policy.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state_delta`	`Mapping[str, Any]`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.update(session_id=..., user_id=..., state_delta=...)

RedisStore.delete

async RedisStore.delete(*, session_id: str, user_id: str) -> bool

Delete a session and its user index entry atomically.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
result = await store.delete(session_id=..., user_id=...)

RedisStore.acquire

RedisStore.acquire(*, session_id: str, user_id: str) -> AsyncIterator[SessionLease]

Hold and renew a token-bound distributed lease for one execution.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
async with store.acquire(session_id=..., user_id=...):
    ...

RedisStore.begin_run

async RedisStore.begin_run(
    *,
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
) -> RunRecord

Create an idempotent run while allowing one active run per session.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`run_id`	`str`	Required
`framework`	`Literal['adk', 'langgraph']`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.begin_run(
    application_id=...,
    user_id=...,
    session_id=...,
    run_id=...,
    framework=...,
)

RedisStore.get_run

async RedisStore.get_run(*, scope: RunScope) -> RunRecord | None

Read one Redis checkpoint run through its ownership scope.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.get_run(scope=...)

RedisStore.get_checkpoint

async RedisStore.get_checkpoint(*, scope: RunScope, checkpoint_id: str | None = None, namespace: str = '') -> CheckpointRecord | None

Resolve an exact or newest checkpoint through its Redis index.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str \| None`	`None`
`namespace`	`str`	`''`

from harnest.store import RedisStore

store: RedisStore
record = await store.get_checkpoint(scope=...)

RedisStore.list_checkpoints

RedisStore.list_checkpoints(
    *,
    scope: RunScope,
    namespace: str | None = None,
    before: str | None = None,
    limit: int = 20,
) -> AsyncIterator[CheckpointRecord]

Read bounded checkpoint history from sorted-set indexes.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`namespace`	`str \| None`	`None`
`before`	`str \| None`	`None`
`limit`	`int`	`20`

from harnest.store import RedisStore

store: RedisStore
async for item in store.list_checkpoints(scope=...):
    ...

RedisStore.put

async RedisStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord

Persist checkpoint data and revision indexes with one Lua CAS.

Parameters and example

Parameter	Type	Default
`checkpoint`	`CheckpointRecord`	Required
`scope`	`RunScope`	Required
`expected_revision`	`int \| None`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.put(checkpoint=..., scope=..., expected_revision=...)

RedisStore.put_writes

async RedisStore.put_writes(*, scope: RunScope, checkpoint_id: str, writes: Sequence[CheckpointWrite]) -> None

Add pending writes idempotently with one Lua operation.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required
`writes`	`Sequence[CheckpointWrite]`	Required

from harnest.store import RedisStore

store: RedisStore
await store.put_writes(scope=..., checkpoint_id=..., writes=...)

RedisStore.get_writes

async RedisStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]

Read pending writes for one owned Redis checkpoint.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
result = await store.get_writes(scope=..., checkpoint_id=...)

RedisStore.get_writes_batch

async RedisStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]

Fetch several pending-write hashes in one Redis round trip.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_ids`	`Sequence[str]`	Required

from harnest.store import RedisStore

store: RedisStore
result = await store.get_writes_batch(scope=..., checkpoint_ids=...)

RedisStore.suspend_continuation

async RedisStore.suspend_continuation(*, record: ContinuationRecord, external_id: str) -> ContinuationRecord

Atomically persist a provider wait and move its run to waiting.

Parameters and example

Parameter	Type	Default
`record`	`ContinuationRecord`	Required
`external_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.suspend_continuation(record=..., external_id=...)

RedisStore.get_continuation

async RedisStore.get_continuation(*, scope: RunScope, continuation_id: str) -> ContinuationRecord | None

Read one private record only when all ownership fields match.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`continuation_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.get_continuation(scope=..., continuation_id=...)

RedisStore.get_provider_continuation

async RedisStore.get_provider_continuation(*, scope: RunScope, provider: str, continuation_id: str) -> ProviderPendingContinuation | None

Load an exact provider envelope only while its owned run exists.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
continuation = await store.get_provider_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
)

RedisStore.get_continuation_by_external_id

async RedisStore.get_continuation_by_external_id(*, application_id: str, provider: str, external_id: str) -> ProviderPendingContinuation | None

Resolve a callback through its provider-owned Redis index.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`external_id`	`str`	Required

from harnest.store import RedisStore

store: RedisStore
continuation = await store.get_continuation_by_external_id(
    application_id=...,
    provider=...,
    external_id=...,
)

RedisStore.list_pending_continuations

async RedisStore.list_pending_continuations(
    *,
    application_id: str,
    provider: str,
    after: str | None = None,
    limit: int = 100,
) -> Sequence[ProviderPendingContinuation]

Load one indexed provider page with a single batched value read.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.store import RedisStore

store: RedisStore
result = await store.list_pending_continuations(application_id=..., provider=...)

RedisStore.resolve_continuation

async RedisStore.resolve_continuation(
    *,
    scope: RunScope,
    provider: str,
    external_id: str,
    schema_id: str,
    result: Any = None,
    failure: ContinuationFailure | None = None,
) -> ContinuationRecord

Resolve an indexed provider callback with an atomic status CAS.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`external_id`	`str`	Required
`schema_id`	`str`	Required
`result`	`Any`	`None`
`failure`	`ContinuationFailure \| None`	`None`

from harnest.store import RedisStore

store: RedisStore
record = await store.resolve_continuation(
    scope=...,
    provider=...,
    external_id=...,
    schema_id=...,
)

RedisStore.claim_continuation

async RedisStore.claim_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord

Claim a provider outcome and resume its run with one Lua CAS.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.claim_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

RedisStore.cancel_continuation

async RedisStore.cancel_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
    failure: ContinuationFailure,
) -> ContinuationRecord

Cancel one exact waiting run and fail its continuation in one script.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required
`failure`	`ContinuationFailure`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.cancel_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
    failure=...,
)

RedisStore.arm_continuation

async RedisStore.arm_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord

Arm one persisted wait after the framework commits its checkpoint.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest.store import RedisStore

store: RedisStore
record = await store.arm_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

RedisStore.transition

async RedisStore.transition(
    *,
    scope: RunScope,
    expected_status: RunStatus,
    status: RunStatus,
    pending_action: PendingAction | None = None,
) -> RunRecord

Apply one status compare-and-swap and release terminal ownership.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`expected_status`	`RunStatus`	Required
`status`	`RunStatus`	Required
`pending_action`	`PendingAction \| None`	`None`

from harnest.store import RedisStore

store: RedisStore
record = await store.transition(scope=..., expected_status=..., status=...)

RedisStore.delete_run

async RedisStore.delete_run(*, scope: RunScope) -> None

Make run data unreachable immediately and let private blobs expire.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.store import RedisStore

store: RedisStore
await store.delete_run(scope=...)

RedisStore.close

async RedisStore.close() -> None

Close only clients created and owned by this store.

Example

from harnest.store import RedisStore

store: RedisStore
await store.close()

StorageRegistry

One normalized storage boundary consumed by compiler and runtime adapters.

StorageRegistry(
    sessions: SessionStore | ADKStore | None = None,
    checkpoints: CheckpointAuthority | None = None,
    assets: Mapping[str, AssetStore] = ...,
    custom: Mapping[str, CustomStorage] = ...,
    tasks: TaskStore | None = None,
    cron: CronStore | None = None,
    memory: MemoryStore | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`sessions`	`SessionStore \| ADKStore \| None`	`None`
`checkpoints`	`CheckpointAuthority \| None`	`None`
`assets`	`Mapping[str, AssetStore]`	`dict()`
`custom`	`Mapping[str, CustomStorage]`	`dict()`
`tasks`	`TaskStore \| None`	`None`
`cron`	`CronStore \| None`	`None`
`memory`	`MemoryStore \| None`	`None`

Parameters and example

Parameter	Type	Default
`sessions`	`SessionStore \| ADKStore \| None`	`None`
`checkpoints`	`CheckpointAuthority \| None`	`None`
`assets`	`Mapping[str, AssetStore]`	`...`
`custom`	`Mapping[str, CustomStorage]`	`...`
`tasks`	`TaskStore \| None`	`None`
`cron`	`CronStore \| None`	`None`
`memory`	`MemoryStore \| None`	`None`

from harnest.store import StorageRegistry

registry = StorageRegistry()

StorageRegistry.default_assets

Return the explicitly named default without choosing by insertion order.

Example

from harnest.store import StorageRegistry

registry: StorageRegistry
value = registry.default_assets

StorageRegistry.owned_resources

StorageRegistry.owned_resources() -> tuple[Any, ...]

Return each lifecycle-owned object once in deterministic role order.

Example

from harnest.store import StorageRegistry

registry: StorageRegistry
result = registry.owned_resources()

​Exports

​CustomStorage

​CustomStorage.start

​CustomStorage.close

​MemoryStore

​MemoryStore.start

​MemoryStore.create

​MemoryStore.get

​MemoryStore.list

​MemoryStore.update

​MemoryStore.delete

​MemoryStore.acquire

​MemoryStore.close

​MemoryStore.as_langgraph_checkpointer

​MemoryStore.put_a2a_task

​MemoryStore.get_a2a_task

​MemoryStore.list_a2a_tasks

​MemoryStore.delete_a2a_task

​MemoryStore.begin_run

​MemoryStore.get_run

​MemoryStore.get_checkpoint

​MemoryStore.list_checkpoints

​MemoryStore.put

​MemoryStore.put_writes

​MemoryStore.get_writes

​MemoryStore.get_writes_batch

​MemoryStore.transition

​MemoryStore.suspend_continuation

​MemoryStore.get_continuation

​MemoryStore.get_provider_continuation

​MemoryStore.get_continuation_by_external_id

​MemoryStore.list_pending_continuations

​MemoryStore.resolve_continuation

​MemoryStore.cancel_continuation

​MemoryStore.claim_continuation

​MemoryStore.arm_continuation

​MemoryStore.delete_run

​PostgresStore

​PostgresStore.as_langgraph_checkpointer

​PostgresStore.start

​PostgresStore.create

​PostgresStore.get

​PostgresStore.list

​PostgresStore.put_a2a_task

​PostgresStore.get_a2a_task

​PostgresStore.list_a2a_tasks

​PostgresStore.delete_a2a_task

​PostgresStore.update

​PostgresStore.delete

​PostgresStore.acquire

​PostgresStore.begin_run

​PostgresStore.get_run

​PostgresStore.get_checkpoint

​PostgresStore.list_checkpoints

​PostgresStore.put

​PostgresStore.put_writes

​PostgresStore.get_writes

​PostgresStore.get_writes_batch

​PostgresStore.suspend_continuation

​PostgresStore.get_continuation

​PostgresStore.get_provider_continuation

​PostgresStore.get_continuation_by_external_id

​PostgresStore.list_pending_continuations

​PostgresStore.resolve_continuation

​PostgresStore.claim_continuation

​PostgresStore.cancel_continuation

​PostgresStore.arm_continuation

​PostgresStore.transition

​PostgresStore.delete_run

​PostgresStore.close

​RedisStore

​RedisStore.as_langgraph_checkpointer

​RedisStore.start

​RedisStore.create

​RedisStore.get

​RedisStore.list

​RedisStore.put_a2a_task

​RedisStore.get_a2a_task

​RedisStore.list_a2a_tasks

​RedisStore.delete_a2a_task

Exports

CustomStorage

CustomStorage.start

CustomStorage.close

MemoryStore

MemoryStore.start

MemoryStore.create

MemoryStore.get

MemoryStore.list

MemoryStore.update

MemoryStore.delete

MemoryStore.acquire

MemoryStore.close

MemoryStore.as_langgraph_checkpointer

MemoryStore.put_a2a_task

MemoryStore.get_a2a_task

MemoryStore.list_a2a_tasks

MemoryStore.delete_a2a_task

MemoryStore.begin_run

MemoryStore.get_run

MemoryStore.get_checkpoint

MemoryStore.list_checkpoints

MemoryStore.put

MemoryStore.put_writes

MemoryStore.get_writes

MemoryStore.get_writes_batch

MemoryStore.transition

MemoryStore.suspend_continuation

MemoryStore.get_continuation

MemoryStore.get_provider_continuation

MemoryStore.get_continuation_by_external_id

MemoryStore.list_pending_continuations

MemoryStore.resolve_continuation

MemoryStore.cancel_continuation

MemoryStore.claim_continuation

MemoryStore.arm_continuation

MemoryStore.delete_run

PostgresStore

PostgresStore.as_langgraph_checkpointer

PostgresStore.start

PostgresStore.create

PostgresStore.get

PostgresStore.list

PostgresStore.put_a2a_task

PostgresStore.get_a2a_task

PostgresStore.list_a2a_tasks

PostgresStore.delete_a2a_task

PostgresStore.update

PostgresStore.delete

PostgresStore.acquire

PostgresStore.begin_run

PostgresStore.get_run

PostgresStore.get_checkpoint

PostgresStore.list_checkpoints

PostgresStore.put

PostgresStore.put_writes

PostgresStore.get_writes

PostgresStore.get_writes_batch

PostgresStore.suspend_continuation

PostgresStore.get_continuation

PostgresStore.get_provider_continuation

PostgresStore.get_continuation_by_external_id

PostgresStore.list_pending_continuations

PostgresStore.resolve_continuation

PostgresStore.claim_continuation

PostgresStore.cancel_continuation

PostgresStore.arm_continuation

PostgresStore.transition

PostgresStore.delete_run

PostgresStore.close

RedisStore

RedisStore.as_langgraph_checkpointer

RedisStore.start

RedisStore.create

RedisStore.get

RedisStore.list

RedisStore.put_a2a_task

RedisStore.get_a2a_task

RedisStore.list_a2a_tasks

RedisStore.delete_a2a_task