Bot Platform Adapter

Build a new chat channel by subclassing BasePlatformAdapter — chunking, retries, typing, and edit-fallbacks come for free

Subclass BasePlatformAdapter to add a new chat channel: implement four methods, declare capabilities, and inherit robust chunking, retry, typing, and edit-fallback delivery. The runtime methods Bot / BotOS drive — start, stop, send_message, probe, health — are inherited too, so a four-method subclass runs under Bot(...).run() and BotOS(...) with no extra wiring.

An adapter also inherits canonicalize(platform, raw_user_id) — an optional identity-reconciliation hook that defaults to the identity function. See Identity canonicalization.

Quick Start

A minimal adapter implements four methods and calls deliver() to send.

1. Minimal adapter (4 methods)

   Subclass BasePlatformAdapter, declare capabilities, and implement connect, disconnect, send, and get_chat_info.

   from praisonaiagents import BasePlatformAdapter, SendResult
   from praisonaiagents.bots import PlatformCapabilities
   
   
   class AcmeBot(BasePlatformAdapter):
       capabilities = PlatformCapabilities(max_message_length=4096)
   
       async def connect(self, *, is_reconnect: bool = False) -> bool:
           # open your websocket / HTTP session here
           return True
   
       async def disconnect(self) -> None:
           # tear down cleanly
           ...
   
       async def send(self, chat_id, content, *, reply_to=None, metadata=None) -> SendResult:
           message_id = await acme_api.send(chat_id, content, reply_to=reply_to)
           return SendResult(ok=True, message_id=message_id, chat_id=chat_id)
   
       async def get_chat_info(self, chat_id):
           return {"id": chat_id}
   
   
   # Chunking, retry, typing, edit-fallback — all inherited.
   adapter = AcmeBot()
   await adapter.connect()
   result = await adapter.deliver(chat_id="C123", content="…very long text…")
   print(result.ok, result.message_ids)

2. Declare more capabilities to unlock defaults

   Turn on supports_edit and supports_typing, then override only what the platform genuinely does — here a lightweight edit_message.

   from praisonaiagents import BasePlatformAdapter, SendResult
   from praisonaiagents.bots import PlatformCapabilities
   
   
   class AcmeBot(BasePlatformAdapter):
       capabilities = PlatformCapabilities(
           supports_edit=True,
           supports_typing=True,
           max_message_length=4096,
       )
   
       async def connect(self, *, is_reconnect: bool = False) -> bool:
           return True
   
       async def disconnect(self) -> None:
           ...
   
       async def send(self, chat_id, content, *, reply_to=None, metadata=None) -> SendResult:
           message_id = await acme_api.send(chat_id, content, reply_to=reply_to)
           return SendResult(ok=True, message_id=message_id, chat_id=chat_id)
   
       async def get_chat_info(self, chat_id):
           return {"id": chat_id}
   
       async def send_typing(self, chat_id) -> None:
           await acme_api.typing(chat_id)
   
       async def edit_message(self, chat_id, message_id, content) -> SendResult:
           await acme_api.edit(chat_id, message_id, content)
           return SendResult(ok=True, message_id=message_id, chat_id=chat_id)

Runtime methods you get for free

A four-method adapter is a complete, runnable channel — register it and hand it an agent.

from praisonaiagents import Agent, BasePlatformAdapter, SendResult
from praisonaiagents.bots import PlatformCapabilities
from praisonai.bots import Bot
from praisonai.bots._registry import register_platform


class AcmeBot(BasePlatformAdapter):
    platform_name = "acme"
    capabilities = PlatformCapabilities(max_message_length=4096)

    def __init__(self, token=None, agent=None, config=None, **kwargs):
        self.token, self.agent, self.config = token, agent, config

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        return True  # open the session and register your inbound listener

    async def disconnect(self) -> None:
        ...

    async def send(self, chat_id, content, *, reply_to=None, metadata=None) -> SendResult:
        message_id = await acme_api.send(chat_id, content, reply_to=reply_to)
        return SendResult(ok=True, message_id=message_id, chat_id=chat_id)

    async def get_chat_info(self, chat_id):
        return {"id": chat_id}


register_platform("acme", AcmeBot)

agent = Agent(name="Acme Assistant", instructions="Help the user.")
Bot("acme", agent=agent, token="ACME_TOKEN").run()

If start() is cancelled or fails, the default still calls disconnect(), so no half-open client is left behind.

Probe and Health

Gateway preflight and health checks work on any subclass without overrides.

await adapter.probe()   # ProbeResult(ok=True, platform="acme", elapsed_ms=12.4, error=None)
await adapter.health()  # HealthResult(ok=True, is_running=True, uptime_seconds=42.0, ...)

A failing connect() surfaces in probe().error, and the platform field reads your adapter’s platform_name attribute when set. The default probe() opens and closes a real connection when the adapter is not running — override probe() with a cheaper credential check if your connect() is heavy.

How It Works

deliver() formats, chunks, sends a typing heartbeat, then sends each chunk with retry — all keyed off capabilities.

What Do I Need to Override?

Start with the four abstract methods, then reach for defaults only when the platform can do better. An adapter that defers a send to a durable outbox returns SendResult(ok=True, queued=True, ...); one that re-sends after crash recovery returns SendResult(ok=True, duplicate=True, ...) — the status derives automatically so the caller can branch on the outcome.

Advertise honestly — the contract is verified at build time

A supports_* flag with no backing method is caught when the adapter is built, not mid-turn.

# BROKEN — the flag is a lie; caught at Bot(...) build time, not mid-turn.
class MyBot(BasePlatformAdapter):
    capabilities = PlatformCapabilities(supports_edit=True)

# Bot("mybot", agent=agent)
# -> CapabilityContractError: MyBot declares supports_edit=True but does not
#    override edit_message with a callable implementation

# FIXED — declaration is backed by a real implementation.
class MyBot(BasePlatformAdapter):
    capabilities = PlatformCapabilities(supports_edit=True)

    async def edit_message(self, chat_id, message_id, content):
        await acme_api.edit(chat_id, message_id, content)
        return SendResult(ok=True, message_id=message_id, chat_id=chat_id)

delete_message is now capability-gated for parity with edit_message. When supports_delete=False the default returns False (unsupported). When an adapter declares supports_delete=True, the default raises NotImplementedError — but the capability contract catches this earlier, at build time. Set the flag True only when you provide a real delete_message implementation. See Adapter Capability Contract.

Control-trust posture

Set allow_control=False on any BotMessage your adapter emits that is not an interactive human turn — webhook bodies, relayed or mirrored bot-authored content, and proactive/wake/scheduler injects. The default (True) is only correct for a real user typing into a live chat. When False, is_command returns False, so an untrusted /stop or /new is handled as plain text instead of a privileged command. See Control-Trust Provenance.

Identity canonicalization (optional override)

BasePlatformAdapter implements IdentityCanonicalizerProtocol out of the box — the default canonicalize() returns the raw id unchanged.

def canonicalize(self, platform: str, raw_user_id: str) -> str:
    return raw_user_id  # default: identity

Override it when a platform addresses the same person with two interchangeable ids — a WhatsApp LID vs phone JID, a handle-as-id rename, or a number↔UUID alias flip — so the conversation collapses to one session instead of silently forking.

class AcmeBot(BasePlatformAdapter):
    def canonicalize(self, platform: str, raw_user_id: str) -> str:
        # map any legacy handle to the stable numeric account id
        return self._alias_map.get(raw_user_id, raw_user_id)

The override must be deterministic, total (never raise), and fail-open — return raw_user_id unchanged when no mapping is known.

Because a vanilla subclass already satisfies the protocol, hand the adapter itself to identity_canonicalizer= wherever the protocol is accepted. See Identity Canonicalization for the full wiring story.

Opting Out of Default Supervision

The inherited start() is already a working push/callback loop, so most adapters never override it. Override start() (and stop() to unblock it) only when the platform has its own blocking loop — long-polling or a library-owned event loop.

When Bot(...) runs your adapter, it wraps the inbound run loop in ChannelSupervisor by default — auto-reconnect with capped backoff plus health-based restart. Set the class attribute supervised_inbound = False when your adapter already manages its own reconnect loop:

from praisonaiagents import BasePlatformAdapter, SendResult
from praisonaiagents.bots import PlatformCapabilities


class MyAdapter(BasePlatformAdapter):
    supervised_inbound = False   # my adapter manages its own reconnect loop
    capabilities = PlatformCapabilities(max_message_length=4096)

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        return True

    async def disconnect(self) -> None:
        ...

    async def send(self, chat_id, content, *, reply_to=None, metadata=None) -> SendResult:
        message_id = await my_api.send(chat_id, content, reply_to=reply_to)
        return SendResult(ok=True, message_id=message_id, chat_id=chat_id)

    async def get_chat_info(self, chat_id):
        return {"id": chat_id}

The supervised path drives the start()/stop() seam: start() runs the inbound source until stopped and raises on an unexpected drop; the default stop() releases the default start() and calls disconnect() — override it if your own start() blocks and needs an explicit unblock.

User Interaction Flow

A long user reply flows through deliver() as three chunks, with one retry honouring retry_after.

Four behaviours are worth calling out:

1. Chunking is text-only. Dict content passes straight through to send() without chunking or formatting — ideal for rich payloads like attachments or buttons.
2. Reply-to only on the first chunk. In a multi-chunk reply, chunk #1 carries reply_to; chunks 2..N are unthreaded follow-ups. SendResult.message_ids gives you every chunk’s id in order.
3. Typing is best-effort. Failures in send_typing() are swallowed, so a broken indicator never breaks delivery.
4. edit_message declares “not supported” instead of crashing. Callers on channels without edits use the edit_not_supported fallback to decide whether to re-send.

If any chunk in a multi-chunk reply comes back queued or duplicate, deliver() carries that outcome onto the aggregate SendResult.status — a deferred or at-least-once delivery is no longer masked as "sent".

Branching on delivery outcome

A swallowed exception must never become a silent drop. Branch on result.status after deliver() — the four outcomes are a closed set, so the caller always knows what happened without depending on an exception a wrapper layer might eat.

from praisonaiagents.bots import SendStatus  # "sent" | "failed" | "queued" | "duplicate"

result = await adapter.deliver(chat_id, reply)

match result.status:
    case "sent":       ...   # landed inline
    case "queued":     ...   # persisted to outbox, will land later
    case "duplicate":  ...   # crash-recovered — may be a re-send
    case "failed":     ...   # honest failure (see result.error / result.error_kind)

An adapter that persists to a durable outbox returns SendResult(ok=True, queued=True, ...), and one that re-sends after crash recovery returns SendResult(ok=True, duplicate=True, ...) — the status property derives automatically from those flags.

praisonai-bot‘s outbound delivery treats a SendResult with status == "failed" as a failed send, alongside a bare False return — so the default send_message, which returns status="failed" once deliver() exhausts retries, feeds the outbound retry router with no extra code. None returns still count as delivered. See Send Error Taxonomy.

Configuration Options

SendResult is the transport-neutral value returned by every send/edit path.

SendResult.status is a closed union — the SendStatus literal alias "sent" | "failed" | "queued" | "duplicate" — so a caller can exhaustively branch on the outcome instead of treating it as an arbitrary string. Import both from the top-level package:

from praisonaiagents.bots import SendResult, SendStatus

SendResult.to_dict() returns a plain dict for logging and observability, now including the status, queued, and duplicate keys.

BasePlatformAdapter class attributes declare adapter behaviour.

The delay between attempts follows retry_after first, then exponential backoff:

delay_seconds =
    result.retry_after            # platform's own rate-limit hint, if any
    ELSE retry_base_delay * 2**attempt   # 0.5, 1.0, 2.0 with defaults

A send() implementation that raises is treated as a failure and retried — you do not have to catch transport errors yourself.

The four abstract methods every subclass must implement:

Common Patterns

Override chunk() when the platform needs code-fence-aware splitting.

class AcmeBot(BasePlatformAdapter):
    def chunk(self, text: str):
        blocks, buf, fence = [], [], False
        for line in text.splitlines(keepends=True):
            if line.lstrip().startswith("```"):
                fence = not fence
            buf.append(line)
            if not fence and sum(len(b) for b in buf) >= self.max_message_length:
                blocks.append("".join(buf))
                buf = []
        if buf:
            blocks.append("".join(buf))
        return blocks

Send a dict payload to pass rich content straight through — deliver() skips chunking and formatting.

await adapter.deliver(
    chat_id="C123",
    content={"text": "Choose an option", "buttons": ["Yes", "No"]},
)

Register the adapter at runtime via the platform registry — see Bot Platform Plugins for register_platform(name, cls).

The same adapter class is reachable three ways with no code change: register_platform(), a YAML adapter: "module:Class" import ref, or a .praisonai/channels/*.py drop-in file — one class, three registration surfaces. See Custom Gateway Channel (Zero-Code).

Formatting: the markdown_dialect seam

format_message is no longer identity — it auto-calls format_for_dialect(text, caps.markdown_dialect), so a custom adapter usually does not need to override it. Declare markdown_dialect on PlatformCapabilities and the base class renders each reply in the right flavour.

from praisonaiagents import BasePlatformAdapter, SendResult
from praisonaiagents.bots import PlatformCapabilities


class AcmeBot(BasePlatformAdapter):
    capabilities = PlatformCapabilities(markdown_dialect="telegram_markdown_v2")

    async def connect(self, *, is_reconnect: bool = False) -> bool:
        return True

    async def disconnect(self) -> None:
        ...

    async def send(self, chat_id, content, *, reply_to=None, metadata=None) -> SendResult:
        # deliver() already ran format_message() → MarkdownV2-safe text.
        message_id = await acme_api.send(chat_id, content, reply_to=reply_to)
        return SendResult(ok=True, message_id=message_id, chat_id=chat_id)

    async def get_chat_info(self, chat_id):
        return {"id": chat_id}

from praisonaiagents.bots import format_for_dialect


async def send(self, chat_id, content, *, reply_to=None, metadata=None):
    text, parse_mode = format_for_dialect(
        content, self._cap("markdown_dialect", "markdown")
    )
    message_id = await self._bot.send_message(
        chat_id, text, parse_mode=parse_mode, reply_to_message_id=reply_to
    )
    return SendResult(ok=True, message_id=message_id, chat_id=chat_id)

format_for_dialect(text, dialect) returns (rendered_text, parse_mode). See How markdown_dialect is consumed for the full dialect table.

Override format_message only when rendering can’t be expressed as a dialect string — embed builders, non-text payloads, or ML-driven templating.

Best Practices

Related