Bot Platform Adapter

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.

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)

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.

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:

Chunking is text-only. Dict content passes straight through to send() without chunking or formatting — ideal for rich payloads like attachments or buttons.
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.
Typing is best-effort. Failures in send_typing() are swallowed, so a broken indicator never breaks delivery.
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.

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`.
`retry_after`	`Optional[float]`	`None`	Suggested seconds to wait before retrying, from the platform’s rate-limit response. Honoured by the default retry loop.
`metadata`	`Dict[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.

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"]},
)

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.

Advanced: override format_message for markup

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.

​Quick Start

​How It Works

​What Do I Need to Override?

​Opting Out of Default Supervision

​User Interaction Flow

​Configuration Options

BasePlatformAdapter SDK Reference

​Common Patterns

​Best Practices

​Related

Bot Platform Capabilities

Bot Platform Plugins

Run Status Controller

Quick Start

How It Works

What Do I Need to Override?

Opting Out of Default Supervision

User Interaction Flow

Configuration Options

Common Patterns

Best Practices

Related