harnest.session

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`ADKSessionStorage`	Class
`InMemorySessionStore`	Class
`SessionLease`	Class
`SessionStore`	Class

ADKSessionStorage

Deployment-owned ADK session backend configuration.

ADKSessionStorage(uri: str, database_kwargs: Mapping[str, Any] = ...) -> None

Fields (— means no declared default):

Field	Type	Default
`uri`	`str`	`—`
`database_kwargs`	`Mapping[str, Any]`	`dict()`

Parameters and example

Parameter	Type	Default
`uri`	`str`	Required
`database_kwargs`	`Mapping[str, Any]`	`...`

from harnest.session import ADKSessionStorage

storage = ADKSessionStorage(uri=...)

ADKSessionStorage.create_service

ADKSessionStorage.create_service(base_directory: str) -> Any

Create the native ADK session service for a compiled application.

Parameters and example

Parameter	Type	Default
`base_directory`	`str`	Required

from harnest.session import ADKSessionStorage

storage: ADKSessionStorage
result = storage.create_service(base_directory=...)

InMemorySessionStore

Development store; inject a durable SessionStore in production.

InMemorySessionStore() -> None

Example

from harnest.session import InMemorySessionStore

store = InMemorySessionStore()

InMemorySessionStore.start

async InMemorySessionStore.start() -> None

Satisfy the application-owned store lifecycle without external I/O.

Example

from harnest.session import InMemorySessionStore

store: InMemorySessionStore
await store.start()

InMemorySessionStore.create

async InMemorySessionStore.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.session import InMemorySessionStore

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

InMemorySessionStore.get

async InMemorySessionStore.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.session import InMemorySessionStore

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

InMemorySessionStore.list

async InMemorySessionStore.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.session import InMemorySessionStore

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

InMemorySessionStore.update

async InMemorySessionStore.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.session import InMemorySessionStore

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

InMemorySessionStore.delete

async InMemorySessionStore.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.session import InMemorySessionStore

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

InMemorySessionStore.acquire

InMemorySessionStore.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.session import InMemorySessionStore

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

InMemorySessionStore.close

async InMemorySessionStore.close() -> None

Discard all process-local session state.

Example

from harnest.session import InMemorySessionStore

store: InMemorySessionStore
await store.close()

SessionLease

Exclusive session access held across one graph execution.

SessionLease(*args, **kwargs)

Example

from harnest.session import SessionLease

class MySessionLease(SessionLease):
    ...

SessionLease.record

Return the current session snapshot held by this lease.

Example

from harnest.session import SessionLease

lease: SessionLease
value = lease.record

SessionLease.patch_state

async SessionLease.patch_state(delta: Mapping[str, Any]) -> SessionRecord

Merge a state delta and return the updated session snapshot.

Parameters and example

Parameter	Type	Default
`delta`	`Mapping[str, Any]`	Required

from harnest.session import SessionLease

lease: SessionLease
record = await lease.patch_state(delta=...)

SessionLease.replace_state

async SessionLease.replace_state(state: Mapping[str, Any]) -> SessionRecord

Replace framework-owned state and return the updated session.

Parameters and example

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

from harnest.session import SessionLease

lease: SessionLease
record = await lease.replace_state(state=...)

SessionLease.replace_application_data

async SessionLease.replace_application_data(data: Mapping[str, Any]) -> SessionRecord

Replace application-owned data and return the updated session.

Parameters and example

Parameter	Type	Default
`data`	`Mapping[str, Any]`	Required

from harnest.session import SessionLease

lease: SessionLease
record = await lease.replace_application_data(data=...)

SessionStore

Persistent session state with tenant scoping and execution leases. Production implementations should use set-based listing, durable writes, and a distributed lease. Durable mutations are also responsible for the repository’s privacy-safe OTEL audit contract after commit.

SessionStore(*args, **kwargs)

Example

from harnest.session import SessionStore

class MySessionStore(SessionStore):
    ...

SessionStore.start

async SessionStore.start() -> None

Initialize resources required by the session store.

Example

from harnest.session import SessionStore

store: SessionStore
await store.start()

SessionStore.create

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

Create one tenant-scoped session.

Parameters and example

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

from harnest.session import SessionStore

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

SessionStore.get

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

Read one tenant-scoped session when it exists.

Parameters and example

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

from harnest.session import SessionStore

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

SessionStore.list

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

Return an ordered, optionally paginated session list for one user.

Parameters and example

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

from harnest.session import SessionStore

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

SessionStore.update

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

Merge a state delta into an existing tenant-scoped session.

Parameters and example

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

from harnest.session import SessionStore

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

SessionStore.delete

async SessionStore.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.session import SessionStore

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

SessionStore.acquire

SessionStore.acquire(*, session_id: str, user_id: str) -> AsyncContextManager[SessionLease]

Acquire exclusive execution access to one tenant-scoped session.

Parameters and example

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

from harnest.session import SessionStore

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

SessionStore.close

async SessionStore.close() -> None

Release resources owned by the session store.

Example

from harnest.session import SessionStore

store: SessionStore
await store.close()

Overview

Get started

Harnest Studio

Guides

API reference

harnest.session

Exports

ADKSessionStorage

ADKSessionStorage.create_service

InMemorySessionStore

InMemorySessionStore.start

InMemorySessionStore.create

InMemorySessionStore.get

InMemorySessionStore.list

InMemorySessionStore.update

InMemorySessionStore.delete

InMemorySessionStore.acquire

InMemorySessionStore.close

SessionLease

SessionLease.record

SessionLease.patch_state

SessionLease.replace_state

SessionLease.replace_application_data

SessionStore

SessionStore.start

SessionStore.create

SessionStore.get

SessionStore.list

SessionStore.update

SessionStore.delete

SessionStore.acquire

SessionStore.close

​Exports

​ADKSessionStorage

​ADKSessionStorage.create_service

​InMemorySessionStore

​InMemorySessionStore.start

​InMemorySessionStore.create

​InMemorySessionStore.get

​InMemorySessionStore.list

​InMemorySessionStore.update

​InMemorySessionStore.delete

​InMemorySessionStore.acquire

​InMemorySessionStore.close

​SessionLease

​SessionLease.record

​SessionLease.patch_state

​SessionLease.replace_state

​SessionLease.replace_application_data

​SessionStore

​SessionStore.start

​SessionStore.create

​SessionStore.get

​SessionStore.list

​SessionStore.update

​SessionStore.delete

​SessionStore.acquire

​SessionStore.close

Exports

ADKSessionStorage

ADKSessionStorage.create_service

InMemorySessionStore

InMemorySessionStore.start

InMemorySessionStore.create

InMemorySessionStore.get

InMemorySessionStore.list

InMemorySessionStore.update

InMemorySessionStore.delete

InMemorySessionStore.acquire

InMemorySessionStore.close

SessionLease

SessionLease.record

SessionLease.patch_state

SessionLease.replace_state

SessionLease.replace_application_data

SessionStore

SessionStore.start

SessionStore.create

SessionStore.get

SessionStore.list

SessionStore.update

SessionStore.delete

SessionStore.acquire

SessionStore.close