Gateway Runtime Contract

Let your custom channel adapter receive the gateway's five reliability seams — admission control, delivery routing, identity resolution, identity canonicalization, per-turn locking

Custom channel adapters implement SupportsGatewayRuntime to receive the gateway’s five reliability seams; miss the contract and the gateway fails loudly instead of silently dropping admission control, delivery routing, and cross-platform turn locking. The runtime methods Bot / BotOS drive (start, stop, send_message, probe, health) are already inherited from BasePlatformAdapter; this page covers the five gateway seams on top of them.

Quick Start

Adapters that build a BotSessionManager satisfy the contract for free; others implement one method.

1. You don't have to do anything

   Delegate to BotSessionManager — it already implements the contract, so your adapter inherits it through self._session.

   from praisonai_bot.bots._session import BotSessionManager
   
   
   class MyBot:
       def __init__(self, agent):
           # The gateway finds attach_gateway_runtime on self._session
           # and wires all five seams automatically.
           self._session = BotSessionManager(agent=agent)

2. Implement the contract yourself

   Define attach_gateway_runtime and copy each non-None seam exactly as BotSessionManager does — a None seam means “keep what you already have”.

   from praisonaiagents.bots import GatewayRuntimeSeams, SupportsGatewayRuntime
   
   
   class MyBot:
       def attach_gateway_runtime(self, runtime: GatewayRuntimeSeams) -> None:
           if runtime.identity_resolver is not None:
               self._identity_resolver = runtime.identity_resolver
           if runtime.identity_canonicalizer is not None:
               self._identity_canonicalizer = runtime.identity_canonicalizer
           if runtime.delivery_router is not None:
               self._delivery_router = runtime.delivery_router
           if runtime.admission_gate is not None:
               self._admission_gate = runtime.admission_gate
           if runtime.turn_lock_map is not None:
               self._locks = runtime.turn_lock_map
   
   
   # Runtime-checkable: isinstance confirms the contract is satisfied.
   assert isinstance(MyBot(), SupportsGatewayRuntime)

How It Works

The gateway builds each adapter once, then hands it a typed GatewayRuntimeSeams carrier through attach_gateway_runtime.

The gateway resolves the target in a fixed order:

The Five Seams

Each seam maps to a gateway reliability guarantee and a feature doc.

A seam left None means the gateway has nothing to inject for it; leave whatever value you already hold untouched.

The Contract

Three symbols define the contract, all imported from one place.

from praisonaiagents.bots import (
    GatewayRuntimeSeams,
    SupportsGatewayRuntime,
    GatewayAdapterContractError,
)

GatewayRuntimeSeams is the typed carrier the gateway fills and passes to your adapter.

from dataclasses import dataclass
from typing import Any, Optional


@dataclass
class GatewayRuntimeSeams:
    identity_resolver: Optional[Any] = None
    identity_canonicalizer: Optional[Any] = None
    delivery_router: Optional[Any] = None
    admission_gate: Optional[Any] = None
    turn_lock_map: Optional[Any] = None

SupportsGatewayRuntime is the runtime-checkable Protocol your adapter satisfies.

from typing import Protocol, runtime_checkable


@runtime_checkable
class SupportsGatewayRuntime(Protocol):
    def attach_gateway_runtime(self, runtime: "GatewayRuntimeSeams") -> None:
        ...

GatewayAdapterContractError is the loud failure raised when seams cannot be delivered.

class GatewayAdapterContractError(TypeError):
    """Raised when a channel adapter cannot receive the gateway runtime seams."""

When GatewayAdapterContractError Fires

The gateway raises this error only when it has seams to inject but your adapter neither implements attach_gateway_runtime nor exposes a _session / _session_mgr that does — the raise is intentional, replacing a silent loss of reliability guarantees.

The fix is one of the two Quick Start options: build a BotSessionManager on self._session, or implement attach_gateway_runtime directly.

Compatibility

The wrapper permits praisonaiagents >= 1.6.152; on a core that predates the contract the import in step 2 fails and the gateway transparently falls back to the pre-contract private-attribute splices (_identity_resolver, _delivery_router, _admission_gate, _locks) written onto adapter._session or adapter._session_mgr.

Best Practices

Related