platforms.base module

Abstract base types for platform adapters.

Every chat platform (Matrix, Discord, Slack, …) implements PlatformAdapter. Incoming events are normalised into IncomingMessage before being handed to the shared MessageProcessor.

class platforms.base.Attachment(data, mimetype, filename, source_url='')

Bases: object

A downloaded media attachment, already in raw bytes.

Parameters:

• data (bytes)

• mimetype (str)

• filename (str)

• source_url (str)

data: bytes

mimetype: str

filename: str

source_url: str = ''

Original URL the media was fetched from (MXC, CDN, …).

class platforms.base.IncomingMessage(platform, channel_id, user_id, user_name, text, is_addressed, attachments=<factory>, channel_name='', timestamp=<factory>, message_id='', reply_to_id='', extra=<factory>, reactions='')

Bases: object

Platform-agnostic representation of an incoming chat message.

Platform adapters construct one of these for every event and pass it to the MessageProcessor.

Parameters:

• platform (str)

• channel_id (str)

• user_id (str)

• user_name (str)

• text (str)

• is_addressed (bool)

• attachments (list[Attachment])

• channel_name (str)

• timestamp (datetime)

• message_id (str)

• reply_to_id (str)

• extra (dict[str, Any])

• reactions (str)

platform: str

Short identifier for the originating platform.

For example: "matrix", "discord".

channel_id: str

Platform-specific channel / room identifier.

user_id: str

Platform-specific sender identifier.

user_name: str

Human-readable display name of the sender.

text: str

Plain-text body of the message (may be empty for media-only messages).

is_addressed: bool

Whether the bot was explicitly addressed (mention, DM, reply).

attachments: list[Attachment]

Media attachments that have already been downloaded.

channel_name: str = ''

Human-readable channel / room name (for prompt context).

timestamp: datetime

UTC timestamp of the message.

message_id: str = ''

Platform-specific message identifier.

reply_to_id: str = ''

ID of the message being replied to, if any.

extra: dict[str, Any]

Arbitrary platform-specific metadata.

reactions: str = ''

Serialized reaction summary, e.g. "👍×3, 🔥×1".

class platforms.base.HistoricalMessage(user_id, user_name, text, timestamp, message_id='', is_bot=False, reply_to_id='', reactions='')

Bases: object

Lightweight representation of a message fetched from platform history.

Used by PlatformAdapter.fetch_history() to return recent messages for backfilling conversation context after downtime.

Parameters:

• user_id (str)

• user_name (str)

• text (str)

• timestamp (datetime)

• message_id (str)

• is_bot (bool)

• reply_to_id (str)

• reactions (str)

user_id: str

user_name: str

text: str

timestamp: datetime

message_id: str = ''

is_bot: bool = False

True when this message was sent by the bot itself.

reply_to_id: str = ''

ID of the message this one replies to, if any.

reactions: str = ''

Serialized reaction summary, e.g. "👍×3, 🔥×1".

class platforms.base.PlatformAdapter(message_handler)

Bases: ABC

Interface that every platform must implement.

Subclasses wire up their SDK’s event loop, convert native events to IncomingMessage, and forward them to the message_handler callback supplied at construction time.

Parameters:

message_handler (MessageHandler)

__init__(message_handler)

Initialize the instance.

Parameters:

message_handler (Callable[[IncomingMessage, PlatformAdapter], Awaitable[None]]) – The message handler value.

Return type:

None

abstract property name: str

Short lowercase identifier, e.g. "matrix" or "discord".

abstract property is_running: bool

Return True while the platform event loop is active.

property bot_identity: dict[str, str]

Return the bot’s own identity on this platform.

Returns a dict with at minimum platform and user_id. Adapters should override to provide display_name and mention where available. The default returns an empty user_id (safe to call before login completes).

abstractmethod async start()

Connect to the service, authenticate, and begin listening.

Return type:

None

abstractmethod async stop()

Gracefully disconnect and release resources.

Return type:

None

abstractmethod async send(channel_id, text)

Send a plain-text message to channel_id.

Returns the platform message ID of the sent message, or "" if the send failed or the platform does not expose one.

Return type:

str

Parameters:

• channel_id (str)

• text (str)

abstractmethod async send_file(channel_id, data, filename, mimetype='application/octet-stream')

Send a file/media attachment to channel_id.

Parameters:

• channel_id (str) – Platform-specific channel / room identifier.

• data (bytes) – Raw file bytes.

• filename (str) – Suggested filename for the attachment.

• mimetype (str) – MIME type of the file (used for determining how to present the attachment on platforms that distinguish images, audio, video, and generic files).

Returns:

A platform-specific content URL for the uploaded file (mxc:// on Matrix, CDN URL on Discord), or None if the upload failed.

Return type:

str | None

async send_with_buttons(channel_id, text, view=None)

Send a message with interactive buttons attached.

Parameters:

• channel_id (str) – Platform-specific channel / room identifier.

• text (str) – Message text content.

• view (Any) – Platform-specific UI view object (e.g. discord.ui.View). Platforms that don’t support interactive components should

• ID (Returns the platform message)

• failure. (or "" on)

Return type:

str ignore this and send plain text.

:param The default implementation falls back to send().:

async edit_message(channel_id, message_id, new_text)

Edit an existing message sent by the bot.

Parameters:

• channel_id (str) – Platform-specific channel / room identifier.

• message_id (str) – Platform-specific ID of the message to edit.

• new_text (str) – Replacement text content.

• success (Returns True on)

• not (False if the platform does)

• default (support editing or the operation failed. The)

• False. (implementation is a no-op that returns)

Return type:

bool

async start_typing(channel_id)

Begin showing a typing indicator in channel_id.

Implementations should spawn a background task that periodically refreshes the indicator until stop_typing() is called. The default implementation is a no-op.

Return type:

None

Parameters:

channel_id (str)

async stop_typing(channel_id)

Stop showing the typing indicator in channel_id.

Must be safe to call even if start_typing() was never called for the given channel. The default is a no-op.

Return type:

None

Parameters:

channel_id (str)

async fetch_history(channel_id, limit=100)

Fetch recent messages from the platform for channel_id.

Returns up to limit messages in chronological order (oldest first). The default implementation returns an empty list; platform adapters should override when the underlying SDK supports history retrieval.

Return type:

list[HistoricalMessage]

Parameters:

• channel_id (str)

• limit (int)

async get_channel_webhooks(channel_id)

Return webhooks configured for channel_id.

The default implementation returns an empty list. Platform adapters that support webhooks (e.g. Discord) should override.

Return type:

list[dict[str, Any]]

Parameters:

channel_id (str)

async list_servers_and_channels()

Return all servers/guilds and their channels.

Each platform adapter should override this to return a list of dicts describing servers/guilds (or rooms) the bot is active in, along with their channels. The format is platform-specific but should include at minimum server_name, server_id, and channels (a list of channel dicts).

The default implementation returns an empty list.

Return type:

list[dict[str, Any]]

async get_guild_members(guild_id)

Return all members of guild_id with role information.

The default implementation returns an empty list. Platform adapters with guild/server membership APIs (e.g. Discord) should override and implement caching.

Return type:

list[dict[str, str]]

Parameters:

guild_id (str)