> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# 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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "Your Adapter"
C[π connect] --> S[βοΈ send]
S --> D[π΄ disconnect]
end
subgraph "Inherited for free"
F[π format] --> K[βοΈ chunk]
K --> T[β¨οΈ typing]
T --> R[π retry]
end
Cap[βοΈ capabilities] -.gates.-> F
Cap -.gates.-> T
Cap -.gates.-> K
classDef contract fill:#8B0000,stroke:#7C90A0,color:#fff
classDef inherited fill:#189AB4,stroke:#7C90A0,color:#fff
classDef config fill:#6366F1,stroke:#7C90A0,color:#fff
class C,S,D contract
class F,K,T,R inherited
class Cap config
```
An adapter also inherits `canonicalize(platform, raw_user_id)` β an optional identity-reconciliation hook that defaults to the identity function. See [Identity canonicalization](#identity-canonicalization-optional-override).
## Quick Start
A minimal adapter implements four methods and calls `deliver()` to send.
Subclass `BasePlatformAdapter`, declare `capabilities`, and implement `connect`, `disconnect`, `send`, and `get_chat_info`.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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)
```
Turn on `supports_edit` and `supports_typing`, then override only what the platform genuinely does β here a lightweight `edit_message`.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant Caller
participant Adapter as deliver()
participant Send as send()
Caller->>Adapter: deliver(chat_id, content)
Note over Adapter: dict content skips format + chunk
Adapter->>Adapter: send_typing() if supports_typing
Adapter->>Adapter: format_message(text) β format_for_dialect(markdown_dialect)
Adapter->>Adapter: chunk(text) β [c1, c2, c3]
loop each chunk (reply_to on chunk #1 only)
Adapter->>Send: send(chat_id, chunk)
Send-->>Adapter: SendResult
Note over Adapter: on failure, wait retry_after
(else base * 2**attempt)
end
Adapter-->>Caller: SendResult(ok, message_ids)
Note over Adapter: if any chunk was queued/duplicate,
the aggregate status carries it β not "sent"
```
***
## 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](#branching-on-delivery-outcome).
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Start[βοΈ New channel] --> Q1{Just send text?}
Q1 -->|Yes| Four[Override the 4
abstract methods only]
Q1 -->|Long messages| Chunk[Keep default chunk
or override for markdown]
Q1 -->|Supports edits| Edit[Set supports_edit=True
+ override edit_message]
Q1 -->|Rate-limited| Retry[Return SendResult with
retry_after from send]
classDef start fill:#8B0000,stroke:#7C90A0,color:#fff
classDef decision fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef action fill:#10B981,stroke:#7C90A0,color:#fff
class Start start
class Q1 decision
class Four,Chunk,Edit,Retry action
```
### 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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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](/docs/features/bot-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](/docs/features/bot-control-trust).
***
## Identity canonicalization (optional override)
`BasePlatformAdapter` implements `IdentityCanonicalizerProtocol` out of the box β the default `canonicalize()` returns the raw id unchanged.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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](/docs/features/gateway-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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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`.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant User
participant Wrapper as Bot Wrapper
participant Adapter as deliver()
User->>Wrapper: sends a long message
Wrapper->>Adapter: deliver(chat_id, reply_text)
Adapter->>Adapter: format β chunk β 3 chunks
Adapter->>User: chunk #1 (with reply_to)
Adapter->>User: chunk #2 β fails (retry_after=1.5)
Note over Adapter: sleep 1.5s
Adapter->>User: chunk #2 (retry) β ok
Adapter->>User: chunk #3 β ok
Adapter-->>Wrapper: SendResult(ok=True, message_ids=[m1, m2, m3])
```
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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)
```
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Send[βοΈ deliver returns SendResult] --> Q1{queued?}
Q1 -->|Yes| QD[status = queued]
Q1 -->|No| Q2{duplicate?}
Q2 -->|Yes| DP[status = duplicate]
Q2 -->|No| Q3{ok?}
Q3 -->|Yes| ST[status = sent]
Q3 -->|No| FL[status = failed]
classDef input fill:#8B0000,stroke:#7C90A0,color:#fff
classDef decision fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef result fill:#10B981,stroke:#7C90A0,color:#fff
class Send input
class Q1,Q2,Q3 decision
class QD,DP,ST,FL result
```
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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). |
Full auto-generated Python API surface for `BasePlatformAdapter` and `SendResult`.
***
## Common Patterns
Override `chunk()` when the platform needs code-fence-aware splitting.
````python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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](/features/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)](/features/gateway-custom-channel).
***
## 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.
Set `markdown_dialect` and inherit `format_message()` β replies render correctly with no extra code.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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}
```
Transports that need the transport `parse_mode` (Telegram requires it) call `format_for_dialect` in `send()` and forward the second return value.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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](/docs/docs/features/bot-platform-capabilities#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
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.
Setting `supports_edit=True` (or `supports_delete=True`) without overriding the corresponding method now raises `CapabilityContractError` when the adapter is built β not later, mid-turn. Declare only what your adapter actually implements; see [Adapter Capability Contract](/docs/features/bot-capability-contract).
***
## Related
The `PlatformCapabilities` descriptor that gates adapter defaults.
Declared capabilities are verified at build time so unbacked flags fail loudly.
Runtime registration and discovery of platform adapters.
Reach this adapter from YAML `adapter:` or a `.praisonai/channels/` drop-in.
Transport-agnostic run-progress state machine for your adapter.
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.