Skip to main content
Subclass BasePlatformAdapter to add a new chat channel: implement four methods, declare capabilities, and inherit robust chunking, retry, typing, and edit-fallback delivery.

​
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.

​
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.

​
Configuration Options

SendResult is the transport-neutral value returned by every send/edit path.
FieldTypeDefaultDescription
okboolTrueWhether the send succeeded.
message_idOptional[str]NonePlatform message id of the last message sent. For chunked delivery this is the final chunk.
chat_idOptional[str]NoneThe chat/channel the message was delivered to.
message_idsList[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.
errorOptional[str]NoneHuman-readable error string when ok=False.
retry_afterOptional[float]NoneSuggested seconds to wait before retrying, from the platform’s rate-limit response. Honoured by the default retry loop.
metadataDict[str, Any]{}Additional platform-specific result details. On a successful multi-chunk deliver() this includes {"chunks": N}.
SendResult.to_dict() returns a plain dict for logging and observability. BasePlatformAdapter class attributes declare adapter behaviour.
AttributeTypeDefaultDescription
capabilitiesPlatformCapabilitiesPlatformCapabilities()Platform features. Every default behaviour keys off this, degrading gracefully when a flag is absent.
max_retriesint3Retry attempts for the default resilient delivery loop.
retry_base_delayfloat0.5Base 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:
MethodSignaturePurpose
connectasync def connect(*, is_reconnect: bool = False) -> boolEstablish the platform connection.
disconnectasync def disconnect() -> NoneTear down the connection and release resources.
sendasync def send(chat_id, content, *, reply_to=None, metadata=None) -> SendResultSend one message (no chunking, no retry).
get_chat_infoasync 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).

​
Best Practices

Chunking, retry, and typing belong to deliver(). Keep send() a thin wrapper around one platform API call so the shared machinery stays in control.
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.
A truthful PlatformCapabilities gives the shared code the best information for graceful degradation. Overstating a capability breaks the fallback path.
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.
Slack mrkdwn, Telegram MarkdownV2, and Discord embeds each have quirks. format_message is the single method to apply platform-specific markup.

Bot Platform Capabilities

The PlatformCapabilities descriptor that gates adapter defaults.

Bot Platform Plugins

Runtime registration and discovery of platform adapters.

Run Status Controller

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