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.

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)
   

---

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.

---

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

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}

supervised_inbound	What Bot(...) does
True (default, or unset)	Wraps adapter.start() in ChannelSupervisor. Your start() should raise on an unexpected drop so the supervisor can reconnect with backoff.
False	Calls adapter.start() raw — no ChannelSupervisor wrap. Your adapter owns reconnect.

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

Built-in Telegram sets supervised_inbound = False because it already runs its own reconnect loop internally, and the relay transport opts out because the connector owns out-of-process reconnect.

---

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.

---

Configuration Options

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

Field	Type	Default	Description
ok	bool	True	Whether the send succeeded.
message_id	Optional[str]	None	Platform message id of the last message sent. For chunked delivery this is the final chunk.
chat_id	Optional[str]	None	The chat/channel the message was delivered to.
message_ids	List[str]	[]	Ids of every chunk that landed, in send order. On a partial failure the ids of chunks that landed before the error are preserved.
error	Optional[str]	None	Human-readable error string when ok=False.
retryable	bool	True	Whether the failure is worth retrying. Permanent failures set this False so the retry loop short-circuits.
retry_after	Optional[float]	None	Suggested seconds to wait before retrying, from the platform’s rate-limit response. Honoured by the default retry loop.
queued	bool	False	True when the send was persisted to a durable outbox for later delivery rather than delivered inline. Drives status == "queued".
duplicate	bool	False	True when a crash-recovered re-send may be a duplicate — an honest at-least-once outcome. Drives status == "duplicate".
status	SendStatus (property)	derived	Closed, branchable delivery outcome. Precedence: queued beats duplicate beats ("sent" if ok else "failed").
metadata	Dict[str, Any]	{}	Additional platform-specific result details. On a successful multi-chunk deliver() this includes {"chunks": N}.

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.

Backward compatible. send_message may still return a legacy BotMessage — it is accepted and treated as status == "sent". New or refactored adapters SHOULD return a SendResult so the caller can branch on the typed status.

BasePlatformAdapter class attributes declare adapter behaviour.

Attribute	Type	Default	Description
capabilities	PlatformCapabilities	PlatformCapabilities()	Platform features. Every default behaviour keys off this, degrading gracefully when a flag is absent.
max_retries	int	3	Retry attempts for the default resilient delivery loop.
retry_base_delay	float	0.5	Base backoff (seconds) for exponential retry when retry_after is not supplied.

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:

Method	Signature	Purpose
connect	async def connect(*, is_reconnect: bool = False) -> bool	Establish the platform connection.
disconnect	async def disconnect() -> None	Tear down the connection and release resources.
send	async def send(chat_id, content, *, reply_to=None, metadata=None) -> SendResult	Send one message (no chunking, no retry).
get_chat_info	async def get_chat_info(chat_id) -> Dict[str, Any]	Return chat/channel metadata (at least an id key).

BasePlatformAdapter SDK Reference

Full auto-generated Python API surface for BasePlatformAdapter and SendResult.

---

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.

Declarative (base class handles it)

Set markdown_dialect and inherit format_message() — replies render correctly with no extra code.

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}

Direct (forward parse_mode)

Transports that need the transport parse_mode (Telegram requires it) call format_for_dialect in send() and forward the second return value.

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.

Plain-text fallback preserves identifiers. The default "markdown" dialect returns a safe plain-text reduction. As of PraisonAI #3506, strip_markdown unwraps only paired emphasis/code spans, so identifiers (svc_1), globs (*.py), and arithmetic (a*b) are preserved in the fallback rendering rather than mangled.

---

Best Practices

Keep send() a single API call

Chunking, retry, and typing belong to deliver(). Keep send() a thin wrapper around one platform API call so the shared machinery stays in control.

Return retry_after when the platform tells you to

Populate SendResult(ok=False, retry_after=X) from send() when the platform reports a rate limit. The default retry loop honours it and beats fixed backoff.

Declare only what your platform actually does

A truthful PlatformCapabilities gives the shared code the best information for graceful degradation. Overstating a capability breaks the fallback path.

Don't override edit_message unless supports_edit=True

Callers rely on the built-in edit_not_supported fallback to decide whether to re-send. Setting supports_edit=True without an override raises NotImplementedError.

---

Related

Bot Platform Capabilities

The PlatformCapabilities descriptor that gates adapter defaults.

Bot Platform Plugins

Runtime registration and discovery of platform adapters.

Custom Gateway Channel (Zero-Code)

Reach this adapter from YAML adapter: or a .praisonai/channels/ drop-in.

Run Status Controller

Transport-agnostic run-progress state machine for your adapter.