harnest.channels

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`ChannelAdapter`	Class
`ChannelAdapterConflictError`	Class
`ChannelAdapterFactory`	Value or type alias
`ChannelAdapterNotFoundError`	Class
`ChannelBinding`	Class
`ChannelError`	Class
`ChannelEvent`	Class
`ChannelHTTPInvoker`	Class
`ChannelReply`	Class
`ChannelStore`	Class
`ChannelStoreConflictError`	Class
`ChannelWorker`	Class
`MemoryChannelStore`	Class
`create_channel_adapter`	Function
`discover_channel_bindings`	Function
`register_channel_adapter`	Function
`registered_channel_adapters`	Function

ChannelAdapter

Platform-specific transport that normalizes events and sends replies. Extensions implement this contract without exposing raw provider drivers. Normalization is synchronous and offline so fixtures can be tested without live credentials; sending is the only network-owned operation.

ChannelAdapter()

Fields (— means no declared default):

Field	Type	Default
`platform`	`str`	`—`

Example

from harnest.channels import ChannelAdapter

class MyChannelAdapter(ChannelAdapter):
    ...

ChannelAdapter.start

async ChannelAdapter.start() -> None

Initialize provider-owned clients before accepting invocations.

Example

from harnest.channels import ChannelAdapter

adapter: ChannelAdapter
await adapter.start()

ChannelAdapter.normalize_event

ChannelAdapter.normalize_event(raw: Mapping[str, Any]) -> ChannelEvent

Parse one provider payload into a ChannelEvent without any I/O.

Parameters and example

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

from harnest.channels import ChannelAdapter

adapter: ChannelAdapter
event = adapter.normalize_event(raw=...)

ChannelAdapter.send_reply

async ChannelAdapter.send_reply(reply: ChannelReply) -> Mapping[str, Any]

Send content through the transport and return an opaque receipt.

Parameters and example

Parameter	Type	Default
`reply`	`ChannelReply`	Required

from harnest.channels import ChannelAdapter

adapter: ChannelAdapter
result = await adapter.send_reply(reply=...)

ChannelAdapter.close

async ChannelAdapter.close() -> None

Release provider-owned clients after invocations have drained.

Example

from harnest.channels import ChannelAdapter

adapter: ChannelAdapter
await adapter.close()

ChannelAdapterConflictError

Two extensions registered a channel adapter under the same name.

Example

from harnest.channels import ChannelAdapterConflictError

try:
    ...
except ChannelAdapterConflictError as error:
    ...

ChannelAdapterFactory

ChannelAdapterFactory = Callable[[Mapping[str, Any]], harnest.channels.ChannelAdapter]

Example

from harnest.channels import ChannelAdapterFactory

ChannelAdapterNotFoundError

No extension registered a channel adapter under the requested name.

Example

from harnest.channels import ChannelAdapterNotFoundError

try:
    ...
except ChannelAdapterNotFoundError as error:
    ...

ChannelBinding

Local policy connecting one platform conversation scope to this agent. extension names the registered ChannelAdapter factory that owns the platform transport (for example a Fused-backed Slack extension, or a direct-API extension); the binding itself never depends on how that extension talks to the platform. Empty allow-lists mean nothing is authorized yet, not that everything is; an adapter must fail closed.

ChannelBinding(
    platform: str,
    extension: str,
    allowed_installations: tuple[str, ...] = (),
    allowed_conversations: tuple[str, ...] = (),
    config: Mapping[str, Any] = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`platform`	`str`	`—`
`extension`	`str`	`—`
`allowed_installations`	`tuple[str, ...]`	`()`
`allowed_conversations`	`tuple[str, ...]`	`()`
`config`	`Mapping[str, Any]`	`dict()`

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`extension`	`str`	Required
`allowed_installations`	`tuple[str, ...]`	`()`
`allowed_conversations`	`tuple[str, ...]`	`()`
`config`	`Mapping[str, Any]`	`...`

from harnest.channels import ChannelBinding

binding = ChannelBinding(platform=..., extension=...)

ChannelError

Base class for channel configuration and adapter failures.

Example

from harnest.channels import ChannelError

try:
    ...
except ChannelError as error:
    ...

ChannelEvent

One inbound platform message, normalized by a ChannelAdapter. delivery_id identifies this specific transport delivery and may repeat across retries; provider_event_id is the platform’s stable identity for deduplication. reply_to is an opaque, adapter-owned address a later reply must target and is never interpreted by the store.

ChannelEvent(
    platform: str,
    installation_id: str,
    provider_event_id: str,
    kind: str,
    sender_id: str,
    conversation_id: str,
    message_id: str,
    occurred_at: float,
    delivery_id: str = '',
    thread_id: str | None = None,
    content: str = '',
    reply_to: Mapping[str, Any] = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`platform`	`str`	`—`
`installation_id`	`str`	`—`
`provider_event_id`	`str`	`—`
`kind`	`str`	`—`
`sender_id`	`str`	`—`
`conversation_id`	`str`	`—`
`message_id`	`str`	`—`
`occurred_at`	`float`	`—`
`delivery_id`	`str`	`''`
`thread_id`	`str \| None`	`None`
`content`	`str`	`''`
`reply_to`	`Mapping[str, Any]`	`dict()`

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`provider_event_id`	`str`	Required
`kind`	`str`	Required
`sender_id`	`str`	Required
`conversation_id`	`str`	Required
`message_id`	`str`	Required
`occurred_at`	`float`	Required
`delivery_id`	`str`	`''`
`thread_id`	`str \| None`	`None`
`content`	`str`	`''`
`reply_to`	`Mapping[str, Any]`	`...`

from harnest.channels import ChannelEvent

event = ChannelEvent(
    platform=...,
    installation_id=...,
    provider_event_id=...,
    kind=...,
    sender_id=...,
    conversation_id=...,
    message_id=...,
    occurred_at=...,
)

ChannelHTTPInvoker

Use caller-owned HTTP transport and explicit per-actor authentication headers. The server’s authenticator must validate the service credential and map the supplied actor identity to its principal. Metadata never establishes identity.

ChannelHTTPInvoker(client: Any, headers_for_actor: Any) -> None

Parameters and example

Parameter	Type	Default
`client`	`Any`	Required
`headers_for_actor`	`Any`	Required

from harnest.channels import ChannelHTTPInvoker

invoker = ChannelHTTPInvoker(client=..., headers_for_actor=...)

ChannelHTTPInvoker.call

async ChannelHTTPInvoker.__call__(event: ChannelEvent, session_id: str, actor_id: str) -> str

Create/reuse the scoped session and refuse approval or incomplete responses.

Parameters and example

Parameter	Type	Default
`event`	`ChannelEvent`	Required
`session_id`	`str`	Required
`actor_id`	`str`	Required

from harnest.channels import ChannelHTTPInvoker

invoker: ChannelHTTPInvoker
result = await invoker(event=..., session_id=..., actor_id=...)

ChannelReply

One outbound answer queued for durable, at-least-once delivery. status is pending, sent, failed, or delivery_unknown (an ambiguous send that must be reconciled before replay, never blindly retried).

ChannelReply(
    reply_id: str,
    platform: str,
    installation_id: str,
    conversation_id: str,
    reply_to: Mapping[str, Any],
    content: str,
    status: str = 'pending',
    receipt: Mapping[str, Any] | None = None,
    created_at: float = 0.0,
    updated_at: float = 0.0,
) -> None

Fields (— means no declared default):

Field	Type	Default
`reply_id`	`str`	`—`
`platform`	`str`	`—`
`installation_id`	`str`	`—`
`conversation_id`	`str`	`—`
`reply_to`	`Mapping[str, Any]`	`—`
`content`	`str`	`—`
`status`	`str`	`'pending'`
`receipt`	`Mapping[str, Any] \| None`	`None`
`created_at`	`float`	`0.0`
`updated_at`	`float`	`0.0`

Parameters and example

Parameter	Type	Default
`reply_id`	`str`	Required
`platform`	`str`	Required
`installation_id`	`str`	Required
`conversation_id`	`str`	Required
`reply_to`	`Mapping[str, Any]`	Required
`content`	`str`	Required
`status`	`str`	`'pending'`
`receipt`	`Mapping[str, Any] \| None`	`None`
`created_at`	`float`	`0.0`
`updated_at`	`float`	`0.0`

from harnest.channels import ChannelReply

reply = ChannelReply(
    reply_id=...,
    platform=...,
    installation_id=...,
    conversation_id=...,
    reply_to=...,
    content=...,
)

ChannelStore

Atomic operations required for durable, deduplicated channel messaging. All identity, scope, and ordering predicates execute inside the datastore. A provider must durably commit before returning success.

ChannelStore(*args, **kwargs)

Example

from harnest.channels import ChannelStore

class MyChannelStore(ChannelStore):
    ...

ChannelStore.start

async ChannelStore.start() -> None

Open owned resources and initialize the provider’s durable schema.

Example

from harnest.channels import ChannelStore

store: ChannelStore
await store.start()

ChannelStore.close

async ChannelStore.close() -> None

Release owned resources after receivers and senders have stopped.

Example

from harnest.channels import ChannelStore

store: ChannelStore
await store.close()

ChannelStore.admit_event

async ChannelStore.admit_event(event: ChannelEvent) -> tuple[ChannelEvent, bool]

Atomically persist a new event or return an identical prior admission. Deduplication is scoped by platform, installation, and provider event ID. A conflicting definition under the same identity raises ChannelStoreConflictError. The second return value is True only for a newly admitted event.

Parameters and example

Parameter	Type	Default
`event`	`ChannelEvent`	Required

from harnest.channels import ChannelStore

store: ChannelStore
result = await store.admit_event(event=...)

ChannelStore.get_event

async ChannelStore.get_event(*, platform: str, installation_id: str, provider_event_id: str) -> ChannelEvent | None

Read one admitted event by its deduplication identity.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`provider_event_id`	`str`	Required

from harnest.channels import ChannelStore

store: ChannelStore
event = await store.get_event(platform=..., installation_id=..., provider_event_id=...)

ChannelStore.bind_session

async ChannelStore.bind_session(
    *,
    platform: str,
    installation_id: str,
    conversation_id: str,
    thread_id: str | None,
    session_id: str,
) -> None

Idempotently record which Harnest session owns a conversation scope.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`conversation_id`	`str`	Required
`thread_id`	`str \| None`	Required
`session_id`	`str`	Required

from harnest.channels import ChannelStore

store: ChannelStore
await store.bind_session(
    platform=...,
    installation_id=...,
    conversation_id=...,
    thread_id=...,
    session_id=...,
)

ChannelStore.get_session_binding

async ChannelStore.get_session_binding(
    *,
    platform: str,
    installation_id: str,
    conversation_id: str,
    thread_id: str | None,
) -> str | None

Read the session bound to a conversation scope, if any.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`conversation_id`	`str`	Required
`thread_id`	`str \| None`	Required

from harnest.channels import ChannelStore

store: ChannelStore
result = await store.get_session_binding(
    platform=...,
    installation_id=...,
    conversation_id=...,
    thread_id=...,
)

ChannelStore.enqueue_reply

async ChannelStore.enqueue_reply(reply: ChannelReply) -> ChannelReply

Durably queue one reply before the adapter sends it.

Parameters and example

Parameter	Type	Default
`reply`	`ChannelReply`	Required

from harnest.channels import ChannelStore

store: ChannelStore
reply = await store.enqueue_reply(reply=...)

ChannelStore.claim_replies

async ChannelStore.claim_replies(*, platform: str, installation_id: str, limit: int = 1) -> tuple[ChannelReply, ...]

Claim a bounded batch of pending replies for one sender to send.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`limit`	`int`	`1`

from harnest.channels import ChannelStore

store: ChannelStore
result = await store.claim_replies(platform=..., installation_id=...)

ChannelStore.finish_reply

async ChannelStore.finish_reply(*, reply_id: str, status: str, receipt: Mapping[str, Any] | None = None) -> bool

Record a claimed reply’s terminal outcome; unknown IDs return false.

Parameters and example

Parameter	Type	Default
`reply_id`	`str`	Required
`status`	`str`	Required
`receipt`	`Mapping[str, Any] \| None`	`None`

from harnest.channels import ChannelStore

store: ChannelStore
result = await store.finish_reply(reply_id=..., status=...)

ChannelStoreConflictError

An admitted event or reply has a different definition than the request.

Example

from harnest.channels import ChannelStoreConflictError

try:
    ...
except ChannelStoreConflictError as error:
    ...

ChannelWorker

Admit before ACK, fence work, and persist replies before any provider send. The caller owns the started task store and authenticated invoke/send callbacks. Jobs never automatically retry an ambiguous model execution or provider send. Pending jobs survive restart; interrupted claimed jobs require reconciliation.

ChannelWorker(
    *,
    application_id: str,
    binding_id: str,
    binding: ChannelBinding,
    store: TaskStore,
    invoke: Any,
    send: Any,
    allowed_senders: tuple[str, ...],
    allowed_kinds: tuple[str, ...] = ('app_mention',),
    timeout: float = 120,
    not_before: float = 0,
) -> None

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`binding_id`	`str`	Required
`binding`	`ChannelBinding`	Required
`store`	`TaskStore`	Required
`invoke`	`Any`	Required
`send`	`Any`	Required
`allowed_senders`	`tuple[str, ...]`	Required
`allowed_kinds`	`tuple[str, ...]`	`('app_mention',)`
`timeout`	`float`	`120`
`not_before`	`float`	`0`

from harnest.channels import ChannelWorker

worker = ChannelWorker(
    application_id=...,
    binding_id=...,
    binding=...,
    store=...,
    invoke=...,
    send=...,
    allowed_senders=...,
)

ChannelWorker.accepts

ChannelWorker.accepts(event: ChannelEvent) -> bool

Recheck policy at intake and execution, including work retained before changes.

Parameters and example

Parameter	Type	Default
`event`	`ChannelEvent`	Required

from harnest.channels import ChannelWorker

worker: ChannelWorker
result = worker.accepts(event=...)

ChannelWorker.admit

async ChannelWorker.admit(event: ChannelEvent) -> str | None

Commit input and dispatch intent in one atomic, idempotent task admission.

Parameters and example

Parameter	Type	Default
`event`	`ChannelEvent`	Required

from harnest.channels import ChannelWorker

worker: ChannelWorker
result = await worker.admit(event=...)

ChannelWorker.step

async ChannelWorker.step() -> int

Process bounded disjoint claims, prioritizing replies already durably queued.

Example

from harnest.channels import ChannelWorker

worker: ChannelWorker
result = await worker.step()

ChannelWorker.run

async ChannelWorker.run(stop: asyncio.Event) -> None

Drain bounded work until shutdown; the receiver must stop before this worker.

Parameters and example

Parameter	Type	Default
`stop`	`asyncio.Event`	Required

from harnest.channels import ChannelWorker

worker: ChannelWorker
await worker.run(stop=...)

MemoryChannelStore

Implement the atomic channel contract in one process; nothing survives restart.

MemoryChannelStore() -> None

Example

from harnest.channels import MemoryChannelStore

store = MemoryChannelStore()

MemoryChannelStore.start

async MemoryChannelStore.start() -> None

Satisfy lifecycle ownership without opening external resources.

Example

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
await store.start()

MemoryChannelStore.close

async MemoryChannelStore.close() -> None

Release no resources; in-memory data lives only with this instance.

Example

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
await store.close()

MemoryChannelStore.admit_event

async MemoryChannelStore.admit_event(event: ChannelEvent) -> tuple[ChannelEvent, bool]

Commit an isolated event snapshot keyed by its deduplication identity.

Parameters and example

Parameter	Type	Default
`event`	`ChannelEvent`	Required

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
result = await store.admit_event(event=...)

MemoryChannelStore.get_event

async MemoryChannelStore.get_event(*, platform: str, installation_id: str, provider_event_id: str) -> ChannelEvent | None

Read one detached event snapshot by its deduplication identity.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`provider_event_id`	`str`	Required

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
event = await store.get_event(platform=..., installation_id=..., provider_event_id=...)

MemoryChannelStore.bind_session

async MemoryChannelStore.bind_session(
    *,
    platform: str,
    installation_id: str,
    conversation_id: str,
    thread_id: str | None,
    session_id: str,
) -> None

Record the session owning a conversation scope without overwriting it.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`conversation_id`	`str`	Required
`thread_id`	`str \| None`	Required
`session_id`	`str`	Required

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
await store.bind_session(
    platform=...,
    installation_id=...,
    conversation_id=...,
    thread_id=...,
    session_id=...,
)

MemoryChannelStore.get_session_binding

async MemoryChannelStore.get_session_binding(
    *,
    platform: str,
    installation_id: str,
    conversation_id: str,
    thread_id: str | None,
) -> str | None

Read the session bound to a conversation scope, if any.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`conversation_id`	`str`	Required
`thread_id`	`str \| None`	Required

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
result = await store.get_session_binding(
    platform=...,
    installation_id=...,
    conversation_id=...,
    thread_id=...,
)

MemoryChannelStore.enqueue_reply

async MemoryChannelStore.enqueue_reply(reply: ChannelReply) -> ChannelReply

Queue one detached reply snapshot for later claim and send.

Parameters and example

Parameter	Type	Default
`reply`	`ChannelReply`	Required

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
reply = await store.enqueue_reply(reply=...)

MemoryChannelStore.claim_replies

async MemoryChannelStore.claim_replies(*, platform: str, installation_id: str, limit: int = 1) -> tuple[ChannelReply, ...]

Return a bounded scoped batch of pending replies, oldest first.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`limit`	`int`	`1`

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
result = await store.claim_replies(platform=..., installation_id=...)

MemoryChannelStore.finish_reply

async MemoryChannelStore.finish_reply(
    *,
    reply_id: str,
    status: str,
    receipt: Mapping[str, Any] | None = None,
) -> bool

Record a terminal outcome for one previously enqueued reply.

Parameters and example

Parameter	Type	Default
`reply_id`	`str`	Required
`status`	`str`	Required
`receipt`	`Mapping[str, Any] \| None`	`None`

from harnest.channels import MemoryChannelStore

store: MemoryChannelStore
result = await store.finish_reply(reply_id=..., status=...)

create_channel_adapter

Build the adapter a binding names, raising if its extension is absent.

create_channel_adapter(extension: str, config: Mapping[str, Any]) -> ChannelAdapter

Parameters and example

Parameter	Type	Default
`extension`	`str`	Required
`config`	`Mapping[str, Any]`	Required

from harnest.channels import create_channel_adapter

adapter = create_channel_adapter(extension=..., config=...)

discover_channel_bindings

Load every ChannelBinding exported by a project’s channels/ directory. Mirrors the discovery convention used for tools and MCP clients: one file per binding, named after the platform, exporting a callable binding().

discover_channel_bindings(directory: Path) -> tuple[ChannelBinding, ...]

Parameters and example

Parameter	Type	Default
`directory`	`Path`	Required

from harnest.channels import discover_channel_bindings

result = discover_channel_bindings(directory=...)

register_channel_adapter

register_channel_adapter(extension: str, factory: ChannelAdapterFactory) -> None

Parameters and example

Parameter	Type	Default
`extension`	`str`	Required
`factory`	`ChannelAdapterFactory`	Required

from harnest.channels import register_channel_adapter

register_channel_adapter(extension=..., factory=...)

registered_channel_adapters

List currently registered extension names for inspection and errors.

registered_channel_adapters() -> tuple[str, ...]

Example

from harnest.channels import registered_channel_adapters

result = registered_channel_adapters()

​Exports

​ChannelAdapter

​ChannelAdapter.start

​ChannelAdapter.normalize_event

​ChannelAdapter.send_reply

​ChannelAdapter.close

​ChannelAdapterConflictError

​ChannelAdapterFactory

​ChannelAdapterNotFoundError

​ChannelBinding

​ChannelError

​ChannelEvent

​ChannelHTTPInvoker

​ChannelHTTPInvoker.call

​ChannelReply

​ChannelStore

​ChannelStore.start

​ChannelStore.close

​ChannelStore.admit_event

​ChannelStore.get_event

​ChannelStore.bind_session

​ChannelStore.get_session_binding

​ChannelStore.enqueue_reply

​ChannelStore.claim_replies

​ChannelStore.finish_reply

​ChannelStoreConflictError

​ChannelWorker

​ChannelWorker.accepts

​ChannelWorker.admit

​ChannelWorker.step

​ChannelWorker.run

​MemoryChannelStore

​MemoryChannelStore.start

​MemoryChannelStore.close

​MemoryChannelStore.admit_event

​MemoryChannelStore.get_event

​MemoryChannelStore.bind_session

​MemoryChannelStore.get_session_binding

​MemoryChannelStore.enqueue_reply

​MemoryChannelStore.claim_replies

​MemoryChannelStore.finish_reply

​create_channel_adapter

​discover_channel_bindings

​register_channel_adapter

​registered_channel_adapters

Exports

ChannelAdapter

ChannelAdapter.start

ChannelAdapter.normalize_event

ChannelAdapter.send_reply

ChannelAdapter.close

ChannelAdapterConflictError

ChannelAdapterFactory

ChannelAdapterNotFoundError

ChannelBinding

ChannelError

ChannelEvent

ChannelHTTPInvoker

ChannelHTTPInvoker.call

ChannelReply

ChannelStore

ChannelStore.start

ChannelStore.close

ChannelStore.admit_event

ChannelStore.get_event

ChannelStore.bind_session

ChannelStore.get_session_binding

ChannelStore.enqueue_reply

ChannelStore.claim_replies

ChannelStore.finish_reply

ChannelStoreConflictError

ChannelWorker

ChannelWorker.accepts

ChannelWorker.admit

ChannelWorker.step

ChannelWorker.run

MemoryChannelStore

MemoryChannelStore.start

MemoryChannelStore.close

MemoryChannelStore.admit_event

MemoryChannelStore.get_event

MemoryChannelStore.bind_session

MemoryChannelStore.get_session_binding

MemoryChannelStore.enqueue_reply

MemoryChannelStore.claim_replies

MemoryChannelStore.finish_reply

create_channel_adapter

discover_channel_bindings

register_channel_adapter

registered_channel_adapters