Core API

Platform-independent models, backend contracts, registries, conversion, authorization, mentions, and capability declarations.

Models

Base model classes for chatom.

This module provides the foundational model classes that all chatom data structures inherit from. It uses Pydantic for data validation and serialization.

pydantic model chatom.base.base.BaseModel

Bases: BaseModel

Base model class for all chatom data structures.

Provides common configuration and utilities for all models. Inherits from Pydantic’s BaseModel for validation and serialization.

Models can be marked as “incomplete” when they have partial information that needs to be resolved by a backend (e.g., a Channel with only a name but no id). Use is_incomplete to check, and mark_incomplete() / mark_complete() to change the state.

property is_incomplete: bool

Check if this object is marked as incomplete.

Incomplete objects have partial information and need to be resolved by a backend to populate missing fields.

Returns:

True if the object is incomplete.

Return type:

bool

mark_incomplete() → None

Mark this object as incomplete.

Call this when the object has partial information that needs to be resolved by a backend.

mark_complete() → None

Mark this object as complete.

Call this after a backend has resolved all missing fields.

to_dict() → dict[str, Any]

Convert model to dictionary representation.

Returns:

Dictionary representation of the model.

Return type:

Dict[str, Any]

copy_with(**kwargs: Any) → Self

Create a copy of the model with updated fields.

Parameters:

**kwargs – Fields to update in the copy.

Returns:

A new instance with the updated fields.

chatom.base.base.Field(default: Any = PydanticUndefined, *, default_factory: Callable[[], Any] | Callable[[dict[str, Any]], Any] | None = PydanticUndefined, alias: str | None = PydanticUndefined, alias_priority: int | None = PydanticUndefined, validation_alias: str | AliasPath | AliasChoices | None = PydanticUndefined, serialization_alias: str | None = PydanticUndefined, title: str | None = PydanticUndefined, field_title_generator: Callable[[str, FieldInfo], str] | None = PydanticUndefined, description: str | None = PydanticUndefined, examples: list[Any] | None = PydanticUndefined, exclude: bool | None = PydanticUndefined, exclude_if: Callable[[Any], bool] | None = PydanticUndefined, discriminator: str | types.Discriminator | None = PydanticUndefined, deprecated: Deprecated | str | bool | None = PydanticUndefined, json_schema_extra: JsonDict | Callable[[JsonDict], None] | None = PydanticUndefined, frozen: bool | None = PydanticUndefined, validate_default: bool | None = PydanticUndefined, repr: bool = PydanticUndefined, init: bool | None = PydanticUndefined, init_var: bool | None = PydanticUndefined, kw_only: bool | None = PydanticUndefined, pattern: str | re.Pattern[str] | None = PydanticUndefined, strict: bool | None = PydanticUndefined, coerce_numbers_to_str: bool | None = PydanticUndefined, gt: annotated_types.SupportsGt | None = PydanticUndefined, ge: annotated_types.SupportsGe | None = PydanticUndefined, lt: annotated_types.SupportsLt | None = PydanticUndefined, le: annotated_types.SupportsLe | None = PydanticUndefined, multiple_of: float | None = PydanticUndefined, allow_inf_nan: bool | None = PydanticUndefined, max_digits: int | None = PydanticUndefined, decimal_places: int | None = PydanticUndefined, min_length: int | None = PydanticUndefined, max_length: int | None = PydanticUndefined, union_mode: Literal['smart', 'left_to_right'] = PydanticUndefined, fail_fast: bool | None = PydanticUndefined, **extra: Unpack[_EmptyKwargs]) → Any

!!! abstract “Usage Documentation”

[Fields](../concepts/fields.md)

Create a field for objects that can be configured.

Used to provide extra information about a field, either for the model schema or complex validation. Some arguments apply only to number fields (int, float, Decimal) and some apply only to str.

Note

• Any _Unset objects will be replaced by the corresponding value defined in the _DefaultValues dictionary. If a key for the _Unset object is not found in the _DefaultValues dictionary, it will default to None

Parameters:

• default – Default value if the field is not set.

• default_factory – A callable to generate the default value. The callable can either take 0 arguments (in which case it is called as is) or a single argument containing the already validated data.

• alias – The name to use for the attribute when validating or serializing by alias. This is often used for things like converting between snake and camel case.

• alias_priority – Priority of the alias. This affects whether an alias generator is used.

• validation_alias – Like alias, but only affects validation, not serialization.

• serialization_alias – Like alias, but only affects serialization, not validation.

• title – Human-readable title.

• field_title_generator – A callable that takes a field name and returns title for it.

• description – Human-readable description.

• examples – Example values for this field.

• exclude – Whether to exclude the field from the model serialization.

• exclude_if – A callable that determines whether to exclude a field during serialization based on its value.

• discriminator – Field name or Discriminator for discriminating the type in a tagged union.

• deprecated – A deprecation message, an instance of warnings.deprecated or the typing_extensions.deprecated backport, or a boolean. If True, a default deprecation message will be emitted when accessing the field.

• json_schema_extra – A dict or callable to provide extra JSON schema properties.

• frozen – Whether the field is frozen. If true, attempts to change the value on an instance will raise an error.

• validate_default – If True, apply validation to the default value every time you create an instance. Otherwise, for performance reasons, the default value of the field is trusted and not validated.

• repr – A boolean indicating whether to include the field in the __repr__ output.

• init – Whether the field should be included in the constructor of the dataclass. (Only applies to dataclasses.)

• init_var – Whether the field should _only_ be included in the constructor of the dataclass. (Only applies to dataclasses.)

• kw_only – Whether the field should be a keyword-only argument in the constructor of the dataclass. (Only applies to dataclasses.)

• coerce_numbers_to_str – Whether to enable coercion of any Number type to str (not applicable in strict mode).

• strict – If True, strict validation is applied to the field. See [Strict Mode](../concepts/strict_mode.md) for details.

• gt – Greater than. If set, value must be greater than this. Only applicable to numbers.

• ge – Greater than or equal. If set, value must be greater than or equal to this. Only applicable to numbers.

• lt – Less than. If set, value must be less than this. Only applicable to numbers.

• le – Less than or equal. If set, value must be less than or equal to this. Only applicable to numbers.

• multiple_of – Value must be a multiple of this. Only applicable to numbers.

• min_length – Minimum length for iterables.

• max_length – Maximum length for iterables.

• pattern – Pattern for strings (a regular expression).

• allow_inf_nan – Allow inf, -inf, nan. Only applicable to float and [Decimal][decimal.Decimal] numbers.

• max_digits – Maximum number of allow digits for strings.

• decimal_places – Maximum number of decimal places allowed for numbers.

• union_mode – The strategy to apply when validating a union. Can be smart (the default), or left_to_right. See [Union Mode](../concepts/unions.md#union-modes) for details.

• fail_fast – If True, validation will stop on the first error. If False, all validation errors will be collected. This option can be applied only to iterable types (list, tuple, set, and frozenset).

• extra –

  (Deprecated) Extra fields that will be included in the JSON schema.

  !!! warning Deprecated

  The extra kwargs is deprecated. Use json_schema_extra instead.

Returns:

A new [FieldInfo][pydantic.fields.FieldInfo]. The return annotation is Any so Field can be used on

type-annotated fields without causing a type error.

pydantic model chatom.base.base.Identifiable

Bases: BaseModel

Base class for models with an identifier.

Provides common id and name fields that most chat entities have.

Objects can be created with partial information (e.g., just a name) and later resolved by a backend to populate missing fields like id. Use the is_complete property to check if an object has all required fields populated.

field id: str = ''

Platform-specific unique identifier.

field name: str = ''

Human-readable name or display name.

property is_complete: bool

Check if this object has all required fields populated.

An object is considered complete if it has an id. Subclasses may override this to add additional requirements.

Returns:

True if the object is complete.

Return type:

bool

property is_resolvable: bool

Check if this object has enough info to be resolved by a backend.

An object is resolvable if it has at least an id OR a name that can be used to look it up.

Returns:

True if the object can potentially be resolved.

Return type:

bool

User model for chatom.

This module provides the base User class representing a chat platform user.

pydantic model chatom.base.user.User

Bases: Identifiable

Represents a user on a chat platform.

id

Platform-specific unique identifier.

name

Display name of the user.

display_name

Display name (alias, can differ from name).

handle

Username or handle (e.g., @username).

email

Email address of the user, if available.

avatar_url

URL to the user’s avatar image.

is_bot

Whether the user is a bot.

app_id

App ID associated with the bot (for bot users).

field display_name: str = ''

Display name (can differ from name).

field handle: str = ''

Username or handle (e.g., @username).

field email: str = ''

Email address of the user, if available.

field avatar: Avatar | None = None

User’s Avatar

field is_bot: bool = False

Whether the user is a bot.

field app_id: str | None = None

App ID associated with the bot (for bot users).

property best_display_name: str

Get the best available display name for the user.

Returns:

The display_name, name, handle, or id (in order of preference).

Return type:

str

property mention_name: str

Get the best name to use when mentioning the user.

Returns:

The handle or name, whichever is available.

Return type:

str

property avatar_url: str

Get the URL to the user’s avatar image.

Returns:

The avatar URL, or empty string if no avatar is set.

Return type:

str

property is_resolvable: bool

Check if this user can be resolved by a backend.

A user is resolvable if it has an id, name, handle, or email.

Returns:

True if the user can potentially be resolved.

Return type:

bool

Channel model for chatom.

This module provides the base Channel class representing a chat channel or room.

pydantic model chatom.base.channel.Channel

Bases: Identifiable

Represents a channel or room on a chat platform.

id

Platform-specific unique identifier.

name

Display name of the channel.

topic

Channel topic or description.

channel_type

Type of the channel (public, private, etc.).

is_archived

Whether the channel is archived.

member_count

Number of members in the channel.

parent

Parent channel (for threads/subchannels).

parent_id

ID of the parent channel (derived from parent).

field topic: str = ''

Channel topic or description.

field channel_type: ChannelType = ChannelType.UNKNOWN

Type of the channel.

field is_archived: bool = False

Whether the channel is archived.

field member_count: int | None = None

Number of members in the channel.

field parent: Channel | None = None

Parent channel (for threads/subchannels).

field users: list[User] [Optional]

Users in a DM channel (for DIRECT/GROUP types).

property is_complete: bool

Check if this channel has all required fields populated.

A channel is complete if it has an ID. DM channels with users but no ID are considered incomplete (need resolution to get/create ID).

Returns:

True if the channel is complete.

Return type:

bool

property parent_id: str

Get the parent channel’s ID.

Returns:

The parent channel ID or empty string if no parent.

Return type:

str

property is_thread: bool

Check if this channel is a thread.

Returns:

True if this is a thread channel.

Return type:

bool

property is_direct_message: bool

Check if this channel is a direct message.

Returns:

True if this is a DM or group DM.

Return type:

bool

property is_dm: bool

Alias for is_direct_message.

Returns:

True if this is a DM or group DM.

Return type:

bool

property is_public: bool

Check if this channel is public.

Returns:

True if this is a public channel.

Return type:

bool

property is_private: bool

Check if this channel is private.

Returns:

True if this is a private channel.

Return type:

bool

property is_resolvable: bool

Check if this channel can be resolved by a backend.

A channel is resolvable if it has an id, name, or users (for DMs).

Returns:

True if the channel can potentially be resolved.

Return type:

bool

classmethod dm_to(user: User) → Channel

Create an incomplete DM channel to a user.

The returned channel is marked incomplete and will be resolved by the backend (looking up or creating the DM channel).

Parameters:

user – The user to DM.

Returns:

An incomplete Channel that can be resolved to a DM.

Example

>>> dm_channel = Channel.dm_to(user)
>>> await backend.send_message(dm_channel, "Hello!")

classmethod group_dm_to(users: list[User]) → Channel

Create an incomplete group DM channel to multiple users.

The returned channel is marked incomplete and will be resolved by the backend (looking up or creating the group DM).

Parameters:

users – The users to include in the group DM.

Returns:

An incomplete Channel that can be resolved to a group DM.

Example

>>> group = Channel.group_dm_to([user1, user2])
>>> await backend.send_message(group, "Hello everyone!")

class chatom.base.channel.ChannelType(value)

Bases: str, Enum

Types of chat channels.

PUBLIC = 'public'

A public channel visible to all members.

PRIVATE = 'private'

A private channel with restricted access.

DIRECT = 'direct'

A direct message between two users.

GROUP = 'group'

A group direct message between multiple users.

THREAD = 'thread'

A thread within another channel.

FORUM = 'forum'

A forum channel for organized discussions.

ANNOUNCEMENT = 'announcement'

An announcement or broadcast channel.

UNKNOWN = 'unknown'

Unknown channel type.

Organization model for chatom.

This module provides the base Organization class representing a chat platform organization (e.g., Discord guild, Slack workspace, Symphony pod).

pydantic model chatom.base.organization.Organization

Bases: Identifiable

Represents an organization on a chat platform.

An organization is the top-level container for users and channels. This maps to different concepts on different platforms: - Discord: Guild (server) - Slack: Workspace (team) - Symphony: Pod

id

Platform-specific unique identifier.

name

Display name of the organization.

description

Description or purpose of the organization.

icon_url

URL to the organization’s icon/logo image.

member_count

Approximate number of members, if available.

owner

The organization owner, if applicable.

field description: str = ''

Description or purpose of the organization.

field icon_url: str = ''

URL to the organization’s icon/logo image.

field member_count: int | None = None

Approximate number of members, if available.

field owner: User | None = None

The organization owner, if applicable.

property owner_id: str

Get the owner’s ID.

Returns:

The owner ID or empty string if no owner.

Return type:

str

property display_name: str

Get the best available display name for the organization.

Returns:

The name or id (in order of preference).

Return type:

str

Thread model for chatom.

This module provides the Thread class representing a message thread.

pydantic model chatom.base.thread.Thread

Bases: Identifiable

Represents a thread within a channel.

id

Platform-specific unique identifier.

name

Display name of the thread.

parent_channel

The channel this thread belongs to.

parent_message

The message that started the thread.

message_count

Number of messages in the thread.

is_locked

Whether the thread is locked from new messages.

created_at

When the thread was created.

last_message_at

When the last message was posted.

field parent_channel: Channel | None = None

The channel this thread belongs to.

field parent_message: Message | None = None

The message that started the thread.

field message_count: int = 0

Number of messages in the thread.

field is_locked: bool = False

Whether the thread is locked from new messages.

field created_at: datetime | None = None

When the thread was created.

field last_message_at: datetime | None = None

When the last message was posted.

property parent_message_id: str

Get the parent message’s ID.

Returns:

The parent message ID or empty string if no parent message.

Return type:

str

property is_resolvable: bool

Check if this thread can be resolved by a backend.

A thread is resolvable if it has an id or a parent_message.

Returns:

True if the thread can potentially be resolved.

Return type:

bool

Message model for chatom.

This module provides the Message class representing a chat message.

pydantic model chatom.base.message.Message

Bases: Identifiable

Represents a message on a chat platform.

id

Platform-specific unique identifier.

content

Text content of the message.

author

User who sent the message.

channel

Channel where the message was sent.

thread

Thread the message belongs to, if any.

message_type

Type of message.

created_at

When the message was created.

edited_at

When the message was last edited.

is_edited

Whether the message has been edited.

is_pinned

Whether the message is pinned.

is_bot

Whether the message was sent by a bot.

is_system

Whether this is a system message.

mentions

Users mentioned in the message.

attachments

File attachments on the message.

embeds

Rich embeds in the message.

reactions

Reactions on the message.

reference

Reference to another message (for replies).

reply_to

The message this is replying to.

formatted_content

Rich/formatted version of content (HTML, MessageML, etc.).

raw

The raw message data from the backend.

backend

The backend this message originated from.

metadata

Additional platform-specific data.

field content: str = ''

Text content of the message.

field author: User | None = None

User who sent the message.

field channel: Channel | None = None

Channel where the message was sent.

field thread: Thread | None = None

Thread the message belongs to, if any.

field organization: Organization | None = None

Organization the message belongs to, if applicable.

field message_type: MessageType = MessageType.DEFAULT

Type of message.

field created_at: datetime | None = None

When the message was created.

field edited_at: datetime | None = None

When the message was last edited.

field is_edited: bool = False

Whether the message has been edited.

field is_pinned: bool = False

Whether the message is pinned.

field is_bot: bool = False

Whether the message was sent by a bot.

field is_system: bool = False

Whether this is a system message.

field mentions: list[User] [Optional]

Users mentioned in the message.

field attachments: list[Attachment] [Optional]

File attachments on the message.

field embeds: list[Embed] [Optional]

Rich embeds in the message.

field components: Any | None = None

Interactive UI components attached to the message. This is a chatom.format.ComponentContainer but is typed as Any here to avoid a circular import; use the typed accessor FormattedMessage.components when possible.

field reactions: list[Reaction] [Optional]

Reactions on the message.

field reference: MessageReference | None = None

Reference to another message (for replies).

field reply_to: Message | None = None

The message this is replying to.

field forwarded_from: Message | None = None

The original message if this is a forwarded message.

field formatted_content: str = ''

Rich/formatted version of content (HTML, MessageML, etc.).

field raw: Any | None = None

The raw message data from the backend.

field backend: str = ''

The backend this message originated from.

field metadata: dict[str, Any] [Optional]

Additional platform-specific data.

property text: str

Alias for content for backwards compatibility.

Returns:

The message content.

Return type:

str

property user: User | None

Alias for author for backwards compatibility.

Returns:

The message author.

Return type:

Optional[User]

property tags: list[User]

Alias for mentions for backwards compatibility.

Returns:

The mentioned users.

Return type:

List[User]

property mention_ids: list[str]

Get the IDs of mentioned users.

Returns:

List of user IDs mentioned in this message.

Return type:

List[str]

property author_id: str

Get the author’s ID.

Returns:

The author ID or empty string if no author.

Return type:

str

property channel_id: str

Get the channel’s ID.

Returns:

The channel ID or empty string if no channel.

Return type:

str

property thread_id: str

Get the thread’s ID.

Returns:

The thread ID or empty string if no thread.

Return type:

str

property reply_to_id: str

Get the ID of the message this is replying to.

Returns:

The reply-to message ID or empty string if not a reply.

Return type:

str

property is_reply: bool

Check if this message is a reply.

Returns:

True if this is a reply to another message.

Return type:

bool

property has_attachments: bool

Check if message has any attachments.

Returns:

True if message has attachments.

Return type:

bool

property has_embeds: bool

Check if message has any embeds.

Returns:

True if message has embeds.

Return type:

bool

property is_dm: bool

Check if this message is from a direct message.

Returns:

True if from a DM channel.

Return type:

bool

property is_direct_message: bool

Alias for is_dm.

Returns:

True if from a DM channel.

Return type:

bool

property channel_name: str

Get the channel name.

Returns:

The channel name or empty string if not available.

Return type:

str

property author_name: str

Get the author name.

Returns:

The author name or empty string if not available.

Return type:

str

get_mentioned_user_ids() → list[str]

Parse and extract user IDs mentioned in the message content.

This parses the message content using the backend’s mention format and returns a list of user IDs that were mentioned. This is useful for detecting when specific users are mentioned in a message.

Note: This parses the content text. For mentions that were already parsed by the backend, use the mentions property instead.

Returns:

List of user IDs mentioned in the content.

Return type:

List[str]

Example

>>> msg = Message(content="Hey <@U123> and <@U456>!", backend="slack")
>>> msg.get_mentioned_user_ids()
['U123', 'U456']

get_mentioned_channel_ids() → list[str]

Parse and extract channel IDs mentioned in the message content.

This parses the message content using the backend’s channel mention format and returns a list of channel IDs that were referenced.

Returns:

List of channel IDs mentioned in the content.

Return type:

List[str]

Example

>>> msg = Message(content="Join <#C123> and <#C456>!", backend="slack")
>>> msg.get_mentioned_channel_ids()
['C123', 'C456']

mentions_user(user: User) → bool

Check if a specific user is mentioned in this message.

Checks both the parsed mentions list and parses the content to find mentions of the given user.

Parameters:

user – The User to check for.

Returns:

True if the user is mentioned.

Return type:

bool

Example

>>> if message.mentions_user(bot_user):
...     await handle_bot_mention(message)

to_formatted() → FormattedMessage

Convert this message to a FormattedMessage.

Creates a FormattedMessage from the message content, preserving formatting based on the backend’s format.

Returns:

The formatted message representation.

Return type:

FormattedMessage

Example

>>> msg = Message(content="Hello **world**", backend="discord")
>>> formatted = msg.to_formatted()
>>> formatted.render(Format.SLACK_MARKDOWN)
'Hello *world*'

classmethod from_formatted(formatted: FormattedMessage, backend: str = '', **kwargs: Any) → Message

Create a Message from a FormattedMessage.

Renders the FormattedMessage in the appropriate format for the target backend and creates a Message instance.

Parameters:

• formatted – The FormattedMessage to convert.

• backend – The target backend (e.g., ‘slack’, ‘discord’).

• **kwargs – Additional message attributes.

Returns:

A new message instance.

Return type:

Message

Example

>>> from chatom.format import MessageBuilder
>>> fm = MessageBuilder().bold("Hello").text(" world").build()
>>> msg = Message.from_formatted(fm, backend="slack")
>>> msg.content
'*Hello* world'

render_for(backend: str) → str

Render this message’s content for a specific backend.

Converts the message to a FormattedMessage and renders it for the target backend’s format.

Parameters:

backend – The target backend (e.g., ‘slack’, ‘discord’).

Returns:

The rendered message content.

Return type:

str

Example

>>> msg = DiscordMessage(content="Hello **world**")
>>> msg.render_for("slack")
'Hello *world*'

is_message_to_user(user: User) → bool

Check if this message mentions or is directed at a specific user.

This method checks if the given user is mentioned in the message. It checks both the mentions list (User objects) and the mention_ids list (user IDs as strings).

Parameters:

user – The user to check for.

Returns:

True if the user is mentioned in this message.

Return type:

bool

Example

>>> bot_user = User(id="U123", name="MyBot")
>>> if message.is_message_to_user(bot_user):
...     # This message mentions the bot
...     await handle_command(message)

is_in_thread() → bool

Check if this message is part of a thread.

Returns:

True if the message is in a thread.

Return type:

bool

property is_forwarded: bool

Check if this message is a forwarded message.

Returns:

True if this message was forwarded from another message.

Return type:

bool

property forwarded_from_id: str

Get the ID of the original message if this is a forward.

Returns:

The original message ID or empty string if not a forward.

Return type:

str

as_reply(content: str, **kwargs: Any) → Message

Create a new message as a reply to this message.

Constructs a new message with the same channel, with this message set as the reply_to reference.

Parameters:

• content – The reply content.

• **kwargs – Additional message attributes to set.

Returns:

A new Message instance configured as a reply.

Example

>>> reply = message.as_reply("Thanks for letting me know!")
>>> reply.reply_to is message
True

as_thread_reply(content: str, **kwargs: Any) → Message

Create a new message as a reply in this message’s thread.

If this message is already in a thread, the new message is placed in that thread. Otherwise, a new thread is started on this message.

Parameters:

• content – The reply content.

• **kwargs – Additional message attributes to set.

Returns:

A new Message instance configured as a thread reply.

Example

>>> reply = message.as_thread_reply("Following up on this...")
>>> reply.thread is not None
True

async reply(content: str, backend: Any, *, in_thread: bool = True, **kwargs: Any) → Message

Reply to this message, threading by default when supported.

Convenience wrapper around backend.send_message that removes the per-backend branching needed to “reply in the same thread as the user’s message”.

• When in_thread is True (default), passes thread=self so the backend posts into this message’s thread (starting one if needed, where the platform supports it).

• When in_thread is False, passes reply_to=self so the backend references this message without forcing a thread.

Backends that lack the corresponding concept (e.g. Symphony has no threads) silently treat both as a plain top-level send.

Parameters:

• content – The reply content.

• backend – The backend to send through.

• in_thread – If True (default), reply in the thread. If False, use a plain reply reference.

• **kwargs – Additional options forwarded to send_message.

Returns:

The sent reply Message.

Example

>>> reply = await message.reply("thanks!", backend=slack)
>>> reply = await message.reply("ack", backend=slack, in_thread=False)

as_dm_to_author(content: str, **kwargs: Any) → Message

Create a new message as a DM to this message’s author.

Constructs a new message in an incomplete DM channel that will be resolved by the backend when sent.

Parameters:

• content – The DM content.

• **kwargs – Additional message attributes to set.

Returns:

A new Message instance configured for a DM to the author.

Example

>>> dm = message.as_dm_to_author("I'll follow up privately.")
>>> dm.channel.users[0] is message.author
True

as_forward(target_channel: Channel, **kwargs: Any) → Message

Create a new message as a forward of this message to another channel.

Constructs a new message with forwarded content and attribution to the original author and channel.

Parameters:

• target_channel – The channel to forward to.

• **kwargs – Additional message attributes to set.

Returns:

A new Message instance configured as a forward.

Example

>>> forward = message.as_forward(log_channel)
>>> forward.forwarded_from is message
True

as_quote_reply(content: str, **kwargs: Any) → Message

Create a new message that quotes this message.

Constructs a new message with the original content quoted, followed by the reply content.

Parameters:

• content – The reply content (after the quote).

• **kwargs – Additional message attributes to set.

Returns:

A new Message instance with quoted content.

Example

>>> quote = message.as_quote_reply("I agree with this point!")
>>> "> " in quote.content
True

reply_context() → dict[str, Any]

Get context information for creating a reply.

Returns useful objects for manually constructing a reply.

Returns:

Dict with channel, message, thread, author objects.

Example

>>> ctx = message.reply_context()
>>> new_msg = Message(
...     channel=ctx["channel"],
...     content="My reply",
...     reply_to=ctx["message"],
... )

pydantic model chatom.base.message.MessageReference

Bases: BaseModel

Reference to another message (for replies, forwards, etc.).

message_id

ID of the referenced message.

channel_id

ID of the channel containing the message.

guild_id

ID of the guild/server, if applicable.

field message_id: str = ''

ID of the referenced message.

field channel_id: str = ''

ID of the channel containing the message.

field guild_id: str = ''

ID of the guild/server, if applicable.

class chatom.base.message.MessageType(value)

Bases: str, Enum

Types of messages.

DEFAULT = 'default'

A normal user message.

REPLY = 'reply'

A reply to another message.

SYSTEM = 'system'

A system-generated message.

JOIN = 'join'

User joined notification.

LEAVE = 'leave'

User left notification.

PIN = 'pin'

Message pin notification.

THREAD_CREATED = 'thread_created'

Thread creation notification.

FORWARD = 'forward'

A forwarded message.

CALL = 'call'

Voice/video call notification.

UNKNOWN = 'unknown'

Unknown message type.

Attachment model for chatom.

This module provides classes for file attachments and media.

pydantic model chatom.base.attachment.Attachment

Bases: BaseModel

Base class for file attachments.

id

Platform-specific unique identifier.

filename

Name of the file.

url

URL to download the attachment.

data

Raw file bytes for direct upload.

content_type

MIME type of the attachment.

size

File size in bytes.

attachment_type

Type of attachment.

field id: str = ''

Platform-specific unique identifier.

field filename: str = ''

Name of the file.

field url: str = ''

URL to download the attachment.

field data: bytes | None = None

Raw file bytes for direct upload.

field content_type: str = ''

MIME type of the attachment.

field size: int | None = None

File size in bytes.

field attachment_type: AttachmentType = AttachmentType.UNKNOWN

Type of attachment.

field metadata: dict[str, Any] [Optional]

Platform-specific data needed to resolve or download the attachment (e.g. Symphony stream/message IDs).

property has_data: bool

Whether this attachment carries in-memory binary data.

classmethod from_content_type(content_type: str) → AttachmentType

Determine attachment type from MIME type.

Parameters:

content_type – MIME type string.

Returns:

The determined attachment type.

Return type:

AttachmentType

class chatom.base.attachment.AttachmentType(value)

Bases: str, Enum

Types of attachments.

FILE = 'file'

Generic file attachment.

IMAGE = 'image'

Image attachment.

VIDEO = 'video'

Video attachment.

AUDIO = 'audio'

Audio attachment.

DOCUMENT = 'document'

Document attachment (PDF, DOC, etc.).

ARCHIVE = 'archive'

Archive file (ZIP, TAR, etc.).

CODE = 'code'

Code snippet or file.

UNKNOWN = 'unknown'

Unknown attachment type.

pydantic model chatom.base.attachment.File

Bases: Attachment

Generic file attachment.

preview

Preview text or snippet of the file content.

field attachment_type: AttachmentType = AttachmentType.FILE

Type of attachment.

field preview: str = ''

Preview text or snippet of the file content.

pydantic model chatom.base.attachment.Image

Bases: Attachment

Image attachment with additional image-specific metadata.

width

Image width in pixels.

height

Image height in pixels.

alt_text

Alternative text description.

thumbnail_url

URL to a thumbnail version.

field attachment_type: AttachmentType = AttachmentType.IMAGE

Type of attachment.

field width: int | None = None

Image width in pixels.

field height: int | None = None

Image height in pixels.

field alt_text: str = ''

Alternative text description.

field thumbnail_url: str = ''

URL to a thumbnail version.

Embed model for chatom.

This module provides the Embed class for rich message embeds.

pydantic model chatom.base.embed.Embed

Bases: BaseModel

Rich embed for messages.

Supports images, fields, authors, footers, and more. Compatible with Discord embeds and similar rich content systems.

title

Title of the embed.

description

Description text.

url

URL the title links to.

color

Color of the embed sidebar (as hex integer).

timestamp

Timestamp to display.

author

Author information.

footer

Footer information.

thumbnail

Thumbnail image.

image

Main image.

video

Video content.

fields

List of embed fields.

field title: str = ''

Title of the embed.

field description: str = ''

Description text.

field url: str = ''

URL the title links to.

field color: int | None = None

Color of the embed sidebar (as hex integer).

field timestamp: datetime | None = None

Timestamp to display.

field author: EmbedAuthor | None = None

Author information.

field footer: EmbedFooter | None = None

Footer information.

field thumbnail: EmbedMedia | None = None

Thumbnail image.

field image: EmbedMedia | None = None

Main image.

field video: EmbedMedia | None = None

Video content.

field fields: list[EmbedField] [Optional]

List of embed fields.

add_field(name: str, value: str, inline: bool = False) → Embed

Add a field to the embed.

Parameters:

• name – Field name.

• value – Field value.

• inline – Whether to display inline.

Returns:

Self for method chaining.

pydantic model chatom.base.embed.EmbedAuthor

Bases: BaseModel

Author information for an embed.

name

Name of the author.

url

URL for the author link.

icon_url

URL for the author’s icon.

field name: str = ''

Name of the author.

field url: str = ''

URL for the author link.

field icon_url: str = ''

URL for the author’s icon.

pydantic model chatom.base.embed.EmbedField

Bases: BaseModel

A field within an embed.

name

Field name/title.

value

Field value/content.

inline

Whether to display inline with other fields.

field name: str = ''

Field name/title.

field value: str = ''

Field value/content.

field inline: bool = False

Whether to display inline with other fields.

pydantic model chatom.base.embed.EmbedFooter

Bases: BaseModel

Footer information for an embed.

text

Footer text.

icon_url

URL for the footer icon.

field text: str = ''

Footer text.

field icon_url: str = ''

URL for the footer icon.

pydantic model chatom.base.embed.EmbedMedia

Bases: BaseModel

Media (image/video/thumbnail) within an embed.

url

URL of the media.

proxy_url

Proxy URL of the media.

width

Width in pixels.

height

Height in pixels.

field url: str = ''

URL of the media.

field proxy_url: str = ''

Proxy URL of the media.

field width: int | None = None

Width in pixels.

field height: int | None = None

Height in pixels.

Reaction model for chatom.

This module provides the Reaction class representing an emoji reaction.

pydantic model chatom.base.reaction.Emoji

Bases: BaseModel

Represents an emoji that can be used in reactions.

name

The name of the emoji (e.g., ‘thumbsup’, ‘smile’).

id

Platform-specific ID for custom emojis.

unicode

Unicode representation for standard emojis.

is_custom

Whether this is a custom emoji.

url

URL of the emoji image for custom emojis.

field name: str = ''

The name of the emoji.

field id: str = ''

Platform-specific ID for custom emojis.

field unicode: str = ''

Unicode representation for standard emojis.

field is_custom: bool = False

Whether this is a custom emoji.

field url: str = ''

URL of the emoji image for custom emojis.

pydantic model chatom.base.reaction.Reaction

Bases: BaseModel

Represents a reaction to a message.

emoji

The emoji used in the reaction.

count

Number of times this reaction was added.

users

List of users who added this reaction.

me

Whether the current user added this reaction.

field emoji: Emoji [Required]

The emoji used in the reaction.

field count: int = 1

Number of times this reaction was added.

field users: list[User] [Optional]

List of users who added this reaction.

field me: bool = False

Whether the current user added this reaction.

pydantic model chatom.base.reaction.ReactionEvent

Bases: BaseModel

Represents a reaction event (add or remove) on a message.

This model is used for handling incoming reaction events from backends. It provides the information needed to track who reacted to which message and with what emoji.

message

The message the reaction is on.

user

The user who added/removed the reaction.

emoji

The emoji that was added/removed.

event_type

Whether the reaction was added or removed.

timestamp

When the reaction event occurred.

Example

>>> # Handle a reaction event in a bot
>>> async def on_reaction(event: ReactionEvent):
...     if event.event_type == ReactionEventType.ADDED:
...         print(f"User {event.user.id} added {event.emoji}")

field message: Message | None = None

The message the reaction is on.

field user: User | None = None

The user who added/removed the reaction.

field emoji: Emoji [Required]

The emoji that was added/removed.

field event_type: ReactionEventType [Required]

Whether the reaction was added or removed.

field timestamp: datetime | None = None

When the reaction event occurred.

property message_id: str

Get the message ID.

Returns:

The message ID, or empty string if not set.

Return type:

str

property channel_id: str

Get the channel ID from the message.

Returns:

The channel ID, or empty string if not set.

Return type:

str

property user_id: str

Get the user ID.

Returns:

The user ID, or empty string if not set.

Return type:

str

class chatom.base.reaction.ReactionEventType(value)

Bases: str, Enum

Types of reaction events.

ADDED = 'added'

A reaction was added to a message.

REMOVED = 'removed'

A reaction was removed from a message.

Presence model for chatom.

This module provides the Presence class representing a user’s online status.

pydantic model chatom.base.presence.Activity

Bases: BaseModel

Represents a user’s current activity.

name

Name of the activity.

activity_type

Type of activity.

details

Additional details about the activity.

url

URL associated with the activity (e.g., stream URL).

started_at

When the activity started.

field name: str = ''

Name of the activity.

field activity_type: ActivityType = ActivityType.CUSTOM

Type of activity.

field details: str = ''

Additional details about the activity.

field url: str = ''

URL associated with the activity.

field started_at: datetime | None = None

When the activity started.

class chatom.base.presence.ActivityType(value)

Bases: str, Enum

Types of user activities.

PLAYING = 'playing'

Playing a game.

STREAMING = 'streaming'

Streaming content.

LISTENING = 'listening'

Listening to music.

WATCHING = 'watching'

Watching content.

CUSTOM = 'custom'

Custom status.

COMPETING = 'competing'

Competing in something.

pydantic model chatom.base.presence.Presence

Bases: BaseModel

Represents a user’s presence/online status.

user

The user this presence belongs to.

status

Current presence status.

status_text

Custom status text/message.

activity

Current activity, if any.

last_seen

When the user was last active.

is_mobile

Whether the user is on a mobile device.

field user: User | None = None

The user this presence belongs to.

field status: PresenceStatus = PresenceStatus.UNKNOWN

Current presence status.

field status_text: str = ''

Custom status text/message.

field activity: Activity | None = None

Current activity, if any.

field last_seen: datetime | None = None

When the user was last active.

field is_mobile: bool = False

Whether the user is on a mobile device.

property is_online: bool

Check if user is currently online.

Returns:

True if user is online (any non-offline status).

Return type:

bool

property is_available: bool

Check if user is available for messaging.

Returns:

True if user is online and not DND.

Return type:

bool

class chatom.base.presence.PresenceStatus(value)

Bases: str, Enum

User presence status.

ONLINE = 'online'

User is online and active.

IDLE = 'idle'

User is online but idle/away.

DND = 'dnd'

User is in do not disturb mode.

OFFLINE = 'offline'

User is offline.

INVISIBLE = 'invisible'

User is online but appearing offline.

UNKNOWN = 'unknown'

Unknown status.

Interaction model for chatom.

Represents a user interaction with a message component (button click, select menu choice, modal submit, etc.). Platform-agnostic.

pydantic model chatom.base.interaction.Interaction

Bases: Identifiable

A user interaction with a message component.

Emitted by backends when a user clicks a button, picks from a select menu, or submits a modal. Handlers can be registered against action_id via chatom.handlers.InteractionRegistry or consumed as a stream via backend.stream_interactions().

id

Platform-specific interaction ID (e.g. Slack action_ts or Discord interaction snowflake).

type

The kind of interaction.

action_id

The action_id declared on the source component. This is the primary dispatch key.

values

Selected/submitted values. For buttons this is typically a single-element list with the button’s value; for selects it’s the picked option values; for modals it’s all submitted input values keyed by their action_id.

user

The user who triggered the interaction.

channel

The channel the source message lives in.

message_id

The ID of the message that contained the component.

response_token

Opaque, short-lived token some platforms require to reply to the interaction (e.g. Discord interaction token, Slack response_url).

created_at

When the interaction happened.

raw

The raw event payload from the backend.

backend

Name of the backend that produced this interaction.

metadata

Additional platform-specific data.

field type: InteractionType = InteractionType.OTHER

The kind of component interaction.

field action_id: str = ''

Action identifier from the source component.

field values: list[str] [Optional]

Selected/submitted values.

field user: User | None = None

The user who triggered the interaction.

field channel: Channel | None = None

The channel containing the source message.

field message_id: str = ''

ID of the message that contained the component.

field response_token: str = ''

Short-lived token for replying to this interaction.

field created_at: datetime | None = None

When the interaction happened.

field raw: Any | None = None

Raw event payload from the backend.

field backend: str = ''

Backend that produced this interaction.

field metadata: dict[str, Any] [Optional]

Additional platform-specific data.

property value: str

the first value, or empty string.

Type:

Convenience

property channel_id: str

the channel ID, or empty string.

Type:

Convenience

property user_id: str

the user ID, or empty string.

Type:

Convenience

class chatom.base.interaction.InteractionType(value)

Bases: str, Enum

The kind of component interaction.

BUTTON = 'button'

A button click.

SELECT = 'select'

A select menu selection.

MODAL_SUBMIT = 'modal_submit'

A modal form submission.

OTHER = 'other'

Any other interaction (future types).

Capabilities and conversion

Backend capabilities for chatom.

This module defines the capabilities that different chat backends support.

pydantic model chatom.base.capabilities.BackendCapabilities

Bases: BaseModel

Describes the capabilities of a chat backend.

capabilities

Set of supported capabilities.

max_message_length

Maximum message length in characters.

max_attachment_size

Maximum attachment size in bytes.

max_attachments

Maximum number of attachments per message.

max_embeds

Maximum number of embeds per message.

max_reactions

Maximum reactions per message.

field capabilities: frozenset[Capability] [Optional]

Set of supported capabilities.

field max_message_length: int = 4000

Maximum message length in characters.

field max_attachment_size: int = 26214400

Maximum attachment size in bytes.

field max_attachments: int = 10

Maximum number of attachments per message.

field max_embeds: int = 10

Maximum number of embeds per message.

field max_reactions: int = 20

Maximum reactions per message.

supports(capability: Capability) → bool

Check if a capability is supported.

Parameters:

capability – The capability to check.

Returns:

True if the capability is supported.

Return type:

bool

supports_all(*capabilities: Capability) → bool

Check if all capabilities are supported.

Parameters:

*capabilities – Capabilities to check.

Returns:

True if all capabilities are supported.

Return type:

bool

supports_any(*capabilities: Capability) → bool

Check if any capability is supported.

Parameters:

*capabilities – Capabilities to check.

Returns:

True if any capability is supported.

Return type:

bool

class chatom.base.capabilities.Capability(value)

Bases: str, Enum

Capabilities that a chat backend may support.

PLAINTEXT = 'plaintext'

Supports plain text messages.

MARKDOWN = 'markdown'

Supports standard Markdown formatting.

RICH_TEXT = 'rich_text'

Supports rich text formatting.

HTML = 'html'

Supports HTML formatting.

CODE_BLOCKS = 'code_blocks'

Supports code blocks with syntax highlighting.

IMAGES = 'images'

Supports image attachments.

FILES = 'files'

Supports file attachments.

EMBEDS = 'embeds'

Supports rich embeds.

VIDEOS = 'videos'

Supports video attachments.

AUDIO = 'audio'

Supports audio attachments.

EMOJI_REACTIONS = 'emoji_reactions'

Supports emoji reactions on messages.

CUSTOM_EMOJI = 'custom_emoji'

Supports custom emoji.

THREADS = 'threads'

Supports message threads.

REPLIES = 'replies'

Supports direct replies to messages.

USER_MENTIONS = 'user_mentions'

Supports mentioning users.

CHANNEL_MENTIONS = 'channel_mentions'

Supports mentioning channels.

EVERYONE_MENTION = 'everyone_mention'

Supports mentioning everyone/all.

ROLE_MENTIONS = 'role_mentions'

Supports mentioning roles/groups.

EDITING = 'editing'

Supports editing sent messages.

DELETING = 'deleting'

Supports deleting messages.

PINNING = 'pinning'

Supports pinning messages.

TABLES = 'tables'

Supports table rendering.

PRESENCE = 'presence'

Supports user presence/status.

TYPING_INDICATORS = 'typing_indicators'

Supports typing indicators.

BUTTONS = 'buttons'

Supports interactive buttons.

FORMS = 'forms'

Supports forms/dialogs.

SELECT_MENUS = 'select_menus'

Supports select menus/dropdowns.

ORGANIZATIONS = 'organizations'

Supports organization/guild/workspace operations.

MESSAGE_SEARCH = 'message_search'

Supports searching messages by content.

FORWARDING = 'forwarding'

Supports forwarding messages to other channels.

Backend type conversion and validation for chatom.

This module provides functionality for validating and converting between base chatom types and backend-specific types. It enables:

1. Validation: Check if a base type instance can be promoted to a backend type

2. Promotion: Convert a base type (e.g., User) to a backend type (e.g., DiscordUser)

3. Demotion: Convert a backend type back to a base type

Example usage:

>>> from chatom import User
>>> from chatom.base.conversion import can_promote, promote, demote
>>>
>>> user = User(id="123", name="Test")
>>> if can_promote(user, "discord"):
...     discord_user = promote(user, "discord")
>>>
>>> base_user = demote(discord_user)

exception chatom.base.conversion.BackendNotFoundError

Bases: ConversionError

Raised when a backend or backend type is not found.

exception chatom.base.conversion.ConversionError

Bases: Exception

Raised when a type conversion fails.

class chatom.base.conversion.ValidationResult(valid: bool = True, missing_required: list[str] | None = None, invalid_fields: dict[str, str] | None = None, warnings: list[str] | None = None)

Bases: object

Result of validating a base type for a backend.

valid

Whether the instance is valid for the backend.

missing_required

List of required fields that are missing values.

invalid_fields

Dict of field names to validation error messages.

warnings

List of warning messages (non-fatal issues).

chatom.base.conversion.can_promote(instance: BaseModel, backend: str) → bool

Check if a base type instance can be promoted to a backend type.

This is a convenience wrapper around validate_for_backend that returns a simple boolean.

Parameters:

• instance – The base type instance to check.

• backend – The backend identifier.

Returns:

True if the instance can be promoted, False otherwise.

Raises:

BackendNotFoundError – If the backend type is not registered.

chatom.base.conversion.demote(instance: BaseModel) → BaseModel

Demote a backend-specific type instance to its base type.

Creates a new instance of the base type using only the base type’s fields, stripping away backend-specific fields.

Parameters:

instance – The backend-specific type instance to demote.

Returns:

A new instance of the base type.

Raises:

ConversionError – If the instance is not a registered backend type.

chatom.base.conversion.get_backend_type(base_type: type[T], backend: str) → type[BaseModel] | None

Get the backend-specific type for a base type.

Parameters:

• base_type – The base type class.

• backend – The backend identifier.

Returns:

The backend-specific type class, or None if not registered.

chatom.base.conversion.get_base_type(backend_type: type[BaseModel]) → type[BaseModel] | None

Get the base type for a backend-specific type.

Parameters:

backend_type – The backend-specific type class.

Returns:

The base type class, or None if not registered.

chatom.base.conversion.list_backends_for_type(base_type: type[BaseModel]) → list[str]

List all backends that have a registered type for the given base type.

Parameters:

base_type – The base type class.

Returns:

List of backend identifiers.

chatom.base.conversion.promote(instance: T, backend: str, **extra_fields: Any) → BaseModel

Promote a base type instance to a backend-specific type.

Creates a new instance of the backend-specific type using the data from the base instance, plus any additional backend-specific fields provided.

Parameters:

• instance – The base type instance to promote.

• backend – The backend identifier.

• **extra_fields – Additional fields for the backend type.

Returns:

A new instance of the backend-specific type.

Raises:

• BackendNotFoundError – If the backend type is not registered.

• ConversionError – If the promotion fails validation.

chatom.base.conversion.register_backend_type(backend: str, base_type: type[BaseModel], backend_type: type[BaseModel]) → None

Register a backend-specific type for a base type.

This function registers the relationship between a base chatom type and its backend-specific variant. Called during module initialization.

Parameters:

• backend – The backend identifier (e.g., “discord”, “slack”).

• base_type – The base chatom type (e.g., User).

• backend_type – The backend-specific type (e.g., DiscordUser).

chatom.base.conversion.validate_for_backend(instance: BaseModel, backend: str) → ValidationResult

Validate if a base type instance can be promoted to a backend type.

This performs validation to check if the instance has all required fields and if the values are compatible with the backend type.

Parameters:

• instance – The base type instance to validate.

• backend – The backend identifier.

Returns:

ValidationResult with details about validity and any issues.

Raises:

BackendNotFoundError – If the backend type is not registered.

Mention utilities for chatom.

This module provides a single-dispatch based system for generating platform-specific mention strings from User objects.

class chatom.base.mention.ChannelMentionMatch(channel_id: str, start: int, end: int, raw: str)

Bases: NamedTuple

Result of parsing a channel mention from content.

channel_id

The extracted channel ID.

Type:

str

start

Start position in the original string.

Type:

int

end

End position in the original string.

Type:

int

raw

The raw mention string as it appeared.

Type:

str

channel_id: str

Alias for field number 0

start: int

Alias for field number 1

end: int

Alias for field number 2

raw: str

Alias for field number 3

class chatom.base.mention.MentionMatch(user_id: str, start: int, end: int, raw: str)

Bases: NamedTuple

Represents a mention found in message content.

user_id

The extracted user ID.

Type:

str

start

Start position in the original string.

Type:

int

end

End position in the original string.

Type:

int

raw

The raw mention string as it appeared.

Type:

str

user_id: str

Alias for field number 0

start: int

Alias for field number 1

end: int

Alias for field number 2

raw: str

Alias for field number 3

chatom.base.mention.extract_channel_ids(content: str, backend: str) → list[str]

Extract just the channel IDs from mentions in content.

This is a convenience wrapper around parse_channel_mentions that returns only the channel IDs as strings.

Parameters:

• content – The message content to parse.

• backend – The backend platform identifier.

Returns:

List of channel IDs mentioned in the content.

Return type:

List[str]

Example

>>> ids = extract_channel_ids("Join <#C123> and <#C456>!", "slack")
>>> ids
['C123', 'C456']

chatom.base.mention.extract_mention_ids(content: str, backend: str) → list[str]

Extract just the user IDs from mentions in content.

This is a convenience wrapper around parse_mentions that returns only the user IDs as strings.

Parameters:

• content – The message content to parse.

• backend – The backend platform identifier.

Returns:

List of user IDs mentioned in the content.

Return type:

List[str]

Example

>>> ids = extract_mention_ids("Hey <@U123> and <@U456>!", "slack")
>>> ids
['U123', 'U456']

chatom.base.mention.mention_channel(channel: Channel) → str

chatom.base.mention.mention_channel(channel: DiscordChannel) → str

chatom.base.mention.mention_channel(channel: SlackChannel) → str

chatom.base.mention.mention_channel(channel: TelegramChannel) → str

Generate a mention string for a channel.

This is a single-dispatch function that can be overridden for platform-specific channel types.

Parameters:

channel – The channel to mention.

Returns:

The formatted channel mention string.

Return type:

str

Example

>>> from chatom import Channel, mention_channel
>>> channel = Channel(name="general", id="456")
>>> mention_channel(channel)
'#general'

chatom.base.mention.mention_channel_for_backend(channel: Channel, backend: BACKEND) → str

Generate a mention string for a channel based on the backend platform.

Parameters:

• channel – The channel to mention.

• backend – The backend platform identifier.

Returns:

The formatted channel mention string for that backend.

Return type:

str

Example

>>> from chatom import Channel, mention_channel_for_backend
>>> channel = Channel(id="C123", name="general")
>>> mention_channel_for_backend(channel, "slack")
'<#C123>'

chatom.base.mention.mention_user(user: User) → str

chatom.base.mention.mention_user(user: DiscordUser) → str

chatom.base.mention.mention_user(user: SlackUser) → str

chatom.base.mention.mention_user(user: SymphonyUser) → str

chatom.base.mention.mention_user(user: TelegramUser) → str

Generate a mention string for a user.

This is a single-dispatch function that can be overridden for platform-specific user types.

Parameters:

user – The user to mention.

Returns:

The formatted mention string.

Return type:

str

Example

>>> from chatom import User, mention_user
>>> user = User(name="John", id="123")
>>> mention_user(user)
'John'

chatom.base.mention.mention_user_for_backend(user: User, backend: BACKEND) → str

Generate a mention string for a user based on the backend platform.

This is a convenience function that dispatches to the appropriate backend-specific mention format based on the backend parameter.

Parameters:

• user – The user to mention.

• backend – The backend platform identifier.

Returns:

The formatted mention string for that backend.

Return type:

str

Example

>>> from chatom import User, mention_user_for_backend
>>> user = User(id="123", name="John", email="john@example.com")
>>> mention_user_for_backend(user, "slack")
'<@123>'
>>> mention_user_for_backend(user, "discord")
'<@123>'
>>> mention_user_for_backend(user, "symphony")
'<mention uid="123"/>'

chatom.base.mention.parse_channel_mentions(content: str, backend: str) → list[ChannelMentionMatch]

Parse channel mentions from message content.

Extracts channel mentions from a message based on the backend’s mention format. Returns a list of ChannelMentionMatch objects containing the channel IDs and positions of mentions in the content.

Parameters:

• content – The message content to parse.

• backend – The backend platform identifier.

Returns:

List of channel mention matches found.

Return type:

List[ChannelMentionMatch]

Example

>>> # Parse Slack channel mentions
>>> mentions = parse_channel_mentions("Join <#C123>!", "slack")
>>> mentions[0].channel_id
'C123'

>>> # Parse Discord channel mentions
>>> mentions = parse_channel_mentions("Check <#987654321>", "discord")
>>> mentions[0].channel_id
'987654321'

chatom.base.mention.parse_mentions(content: str, backend: str) → list[MentionMatch]

Parse user mentions from message content.

Extracts user mentions from a message based on the backend’s mention format. Returns a list of MentionMatch objects containing the user IDs and positions of mentions in the content.

Parameters:

• content – The message content to parse.

• backend – The backend platform identifier.

Returns:

List of mention matches found.

Return type:

List[MentionMatch]

Example

>>> # Parse Slack mentions
>>> mentions = parse_mentions("Hey <@U123>, check this!", "slack")
>>> mentions[0].user_id
'U123'

>>> # Parse Discord mentions
>>> mentions = parse_mentions("<@123456789> Hello!", "discord")
>>> mentions[0].user_id
'123456789'

>>> # Parse Symphony mentions
>>> mentions = parse_mentions('<mention uid="123"/>!', "symphony")
>>> mentions[0].user_id
'123'

Authorization utilities for chatom.

This module provides a simple authorization framework for bot development. It allows defining policies for what users can do in different channels.

class chatom.base.authorization.AuthorizationPolicy

Bases: ABC

Abstract base class for authorization policies.

Implement this class to define custom authorization logic for your bot. The policy can check user permissions, roles, channel membership, or any other criteria.

Example

>>> class MyPolicy(AuthorizationPolicy):
...     def __init__(self, admin_users: List[str]):
...         self.admin_users = set(admin_users)
...
...     async def is_authorized(
...         self, user: User, permission: str, channel: Optional[Channel] = None
...     ) -> AuthorizationResult:
...         if user.id in self.admin_users:
...             return AuthorizationResult(authorized=True)
...         return AuthorizationResult(
...             authorized=False,
...             reason="Not an admin user"
...         )

abstractmethod async is_authorized(user: User, permission: str, channel: Channel | None = None, **context: Any) → AuthorizationResult

Check if a user is authorized for a permission.

Parameters:

• user – The user to check.

• permission – The permission to check for (can be a Permission enum value or a custom string).

• channel – Optional channel context for channel-specific permissions.

• **context – Additional context that may be needed for the check.

Returns:

AuthorizationResult indicating whether authorized and why.

async check_permissions(user: User, permissions: list[str], channel: Channel | None = None, require_all: bool = True, **context: Any) → AuthorizationResult

Check multiple permissions at once.

Parameters:

• user – The user to check.

• permissions – List of permissions to check.

• channel – Optional channel context.

• require_all – If True, user must have all permissions. If False, user needs any one permission.

• **context – Additional context.

Returns:

AuthorizationResult with missing permissions listed.

pydantic model chatom.base.authorization.AuthorizationResult

Bases: BaseModel

Result of an authorization check.

authorized

Whether the action is authorized.

reason

Human-readable reason for the result.

required_permissions

Permissions that were required.

missing_permissions

Permissions that the user lacks.

field authorized: bool [Required]

Whether the action is authorized.

field reason: str = ''

Human-readable reason for the result.

field required_permissions: list[str] [Optional]

Permissions that were required.

field missing_permissions: list[str] [Optional]

Permissions that the user lacks.

class chatom.base.authorization.Permission(value)

Bases: str, Enum

Common permissions that can be checked.

SEND_MESSAGES = 'send_messages'

Can send messages in a channel.

READ_MESSAGES = 'read_messages'

Can read messages in a channel.

DELETE_MESSAGES = 'delete_messages'

Can delete messages (own or others).

EDIT_MESSAGES = 'edit_messages'

Can edit messages.

EXECUTE_COMMANDS = 'execute_commands'

Can execute bot commands.

ADMIN_COMMANDS = 'admin_commands'

Can execute administrative commands.

MANAGE_CHANNEL = 'manage_channel'

Can modify channel settings.

INVITE_USERS = 'invite_users'

Can invite users to channel.

USE_BOT = 'use_bot'

Can interact with the bot at all.

CONFIGURE_BOT = 'configure_bot'

Can configure bot settings.

class chatom.base.authorization.SimpleAuthorizationPolicy(admin_users: list[str] | None = None, admin_channels: list[str] | None = None, default_authorized: bool = False)

Bases: AuthorizationPolicy

A simple authorization policy based on user/channel allow lists.

This policy allows you to define: - Global admins who can do anything - Per-permission allow lists of user IDs - Per-channel permission overrides - Default behavior (allow or deny)

Example

>>> policy = SimpleAuthorizationPolicy(
...     admin_users=["U123", "U456"],  # These users can do anything
...     default_authorized=False,  # Deny by default
... )
>>> # Allow specific users to use commands
>>> policy.allow_permission("execute_commands", ["U789", "U999"])
>>>
>>> # Check authorization
>>> result = await policy.is_authorized(user, "execute_commands")

add_admin(user_id: str) → None

Add a user as a global admin.

Parameters:

user_id – The user ID to add as admin.

remove_admin(user_id: str) → None

Remove a user from global admins.

Parameters:

user_id – The user ID to remove.

allow_permission(permission: str, user_ids: list[str], channel_id: str | None = None) → None

Allow specific users to have a permission.

Parameters:

• permission – The permission to allow.

• user_ids – List of user IDs to allow.

• channel_id – If provided, only allow in this channel.

block_permission_in_channel(permission: str, channel_ids: list[str]) → None

Block a permission in specific channels.

Parameters:

• permission – The permission to block.

• channel_ids – List of channel IDs where to block.

async is_authorized(user: User, permission: str, channel: Channel | None = None, **context: Any) → AuthorizationResult

Check if a user is authorized for a permission.

Parameters:

• user – The user to check.

• permission – The permission to check for.

• channel – Optional channel context.

• **context – Additional context (ignored by this policy).

Returns:

AuthorizationResult indicating whether authorized.

async chatom.base.authorization.is_user_authorized(user: User, permission: str, policy: AuthorizationPolicy, channel: Channel | None = None, **context: Any) → bool

Convenience function to check user authorization.

This is a simple wrapper that returns just a boolean.

Parameters:

• user – The user to check.

• permission – The permission to check for.

• policy – The authorization policy to use.

• channel – Optional channel context.

• **context – Additional context for the policy.

Returns:

True if authorized, False otherwise.

Return type:

bool

Example

>>> policy = SimpleAuthorizationPolicy(admin_users=["U123"])
>>> if await is_user_authorized(user, Permission.ADMIN_COMMANDS, policy):
...     await handle_admin_command(message)

Connections and backends

Connection and registry classes for chatom.

This module provides base classes for backend connections and registries for looking up users and channels.

pydantic model chatom.base.connection.ChannelRegistry

Bases: Registry[Channel]

Registry for looking up channels by ID or name.

This can be subclassed by backends to provide additional lookup capabilities like fetching from an API.

Example

>>> registry = ChannelRegistry()
>>> registry.add(Channel(id="C123", name="general"))
>>> registry.get_by_id("C123")
Channel(id='C123', name='general', ...)

lookup(*, id: str | None = None, name: str | None = None) → Channel | None

Look up a channel by any available identifier.

Parameters:

• id – Channel ID to look up.

• name – Channel name to look up.

Returns:

The channel if found, None otherwise.

channel_to_id(channel: Channel) → str

Get the ID for a channel.

Parameters:

channel – The channel object.

Returns:

The channel’s ID.

Raises:

LookupError – If channel has no ID.

channel_to_name(channel: Channel) → str

Get the name for a channel.

Parameters:

channel – The channel object.

Returns:

The channel’s name.

id_to_channel(id: str) → Channel

Look up a channel by ID.

Parameters:

id – The channel ID.

Returns:

The channel.

Raises:

LookupError – If channel not found.

name_to_channel(name: str) → Channel

Look up a channel by name.

Parameters:

name – The channel name.

Returns:

The channel.

Raises:

LookupError – If channel not found.

pydantic model chatom.base.connection.Connection

Bases: BaseModel

Base class for a connection to a backend service.

This provides a unified interface for connecting to chat platforms, managing users and channels, and sending messages.

Subclasses should implement the abstract methods for platform-specific behavior.

backend

The backend type identifier (e.g., ‘slack’, ‘discord’).

connected

Whether currently connected.

users

Registry of users.

channels

Registry of channels.

field backend: str = ''

The backend type identifier.

field connected: bool = False

Whether currently connected.

field users: UserRegistry [Optional]

Registry of users.

field channels: ChannelRegistry [Optional]

Registry of channels.

abstractmethod async connect() → None

Establish connection to the backend.

Raises:

NotImplementedError – If not implemented by subclass.

abstractmethod async disconnect() → None

Disconnect from the backend.

Raises:

NotImplementedError – If not implemented by subclass.

abstractmethod async fetch_user(id: str) → User | None

Fetch a user from the backend by ID.

Parameters:

id – The user ID.

Returns:

The user if found, None otherwise.

abstractmethod async fetch_channel(id: str) → Channel | None

Fetch a channel from the backend by ID.

Parameters:

id – The channel ID.

Returns:

The channel if found, None otherwise.

async get_user(*, id: str | None = None, name: str | None = None, email: str | None = None, handle: str | None = None) → User | None

Get a user by any identifier, fetching from backend if needed.

First checks the local registry, then fetches from backend if not found and an ID was provided.

Parameters:

• id – User ID.

• name – User name.

• email – User email.

• handle – User handle.

Returns:

The user if found, None otherwise.

async get_channel(*, id: str | None = None, name: str | None = None) → Channel | None

Get a channel by any identifier, fetching from backend if needed.

First checks the local registry, then fetches from backend if not found and an ID was provided.

Parameters:

• id – Channel ID.

• name – Channel name.

Returns:

The channel if found, None otherwise.

exception chatom.base.connection.LookupError

Bases: Exception

Raised when a lookup fails.

pydantic model chatom.base.connection.UserRegistry

Bases: Registry[User]

Registry for looking up users by ID, name, or email.

This can be subclassed by backends to provide additional lookup capabilities like fetching from an API.

Example

>>> registry = UserRegistry()
>>> registry.add(User(id="123", name="John", email="john@example.com"))
>>> registry.get_by_id("123")
User(id='123', name='John', ...)

add(entity: User) → None

Add a user to the registry.

Parameters:

entity – The user to add.

get_by_email(email: str) → User | None

Get a user by email address.

Parameters:

email – The email address.

Returns:

The user if found, None otherwise.

get_by_handle(handle: str) → User | None

Get a user by handle/username.

Parameters:

handle – The handle/username.

Returns:

The user if found, None otherwise.

lookup(*, id: str | None = None, name: str | None = None, email: str | None = None, handle: str | None = None) → User | None

Look up a user by any available identifier.

Parameters:

• id – User ID to look up.

• name – User name to look up.

• email – User email to look up.

• handle – User handle to look up.

Returns:

The user if found, None otherwise.

user_to_id(user: User) → str

Get the ID for a user.

Parameters:

user – The user object.

Returns:

The user’s ID.

Raises:

LookupError – If user has no ID.

user_to_name(user: User) → str

Get the display name for a user.

Parameters:

user – The user object.

Returns:

The user’s display name.

user_to_email(user: User) → str | None

Get the email for a user.

Parameters:

user – The user object.

Returns:

The user’s email or None.

id_to_user(id: str) → User

Look up a user by ID.

Parameters:

id – The user ID.

Returns:

The user.

Raises:

LookupError – If user not found.

name_to_user(name: str) → User

Look up a user by name.

Parameters:

name – The user name.

Returns:

The user.

Raises:

LookupError – If user not found.

email_to_user(email: str) → User

Look up a user by email.

Parameters:

email – The email address.

Returns:

The user.

Raises:

LookupError – If user not found.

clear() → None

Clear all cached users.

Backend configuration for chatom.

This module provides base classes for configuring chat backends.

pydantic model chatom.backend.backend_config.BackendConfig

Bases: BaseModel

Base configuration for a chat backend.

Subclass this for backend-specific configuration options.

This base class provides common patterns for configuration: - get_secret(field_name): Get a SecretStr field’s value as a plain string - has_field(field_name): Check if a field has a non-empty value

api_token

Authentication token for the backend API.

api_url

Base URL for the backend API.

timeout

Request timeout in seconds.

retry_count

Number of retries for failed requests.

extra

Additional backend-specific configuration.

field api_token: str = ''

Authentication token for the backend API.

field api_url: str = ''

Base URL for the backend API.

field timeout: float = 30.0

Request timeout in seconds.

field retry_count: int = 3

Number of retries for failed requests.

field extra: dict[str, Any] [Optional]

Additional backend-specific configuration.

get_secret(field_name: str) → str

Get a SecretStr field’s value as a plain string.

This is a helper method to avoid repeating the pattern of self.field.get_secret_value() in every config class.

Parameters:

field_name – The name of the SecretStr field.

Returns:

The secret value as a plain string, or empty string if not set.

Raises:

• AttributeError – If the field doesn’t exist.

• TypeError – If the field is not a SecretStr.

Example

>>> config.get_secret("password")
"my-secret-password"

has_field(field_name: str) → bool

Check if a field has a non-empty value.

This is a helper method to check if optional configuration fields are set. Works with strings, SecretStr, and other types.

Parameters:

field_name – The name of the field to check.

Returns:

True if the field has a non-empty value.

Example

>>> config.has_field("api_token")
True

property has_token: bool

Check if an API token is configured.

Returns:

True if api_token is set.

property has_url: bool

Check if an API URL is configured.

Returns:

True if api_url is set.

Backend registry for chatom.

This module provides the central registry for backend implementations.

class chatom.backend.backend_registry.BackendRegistry

Bases: object

Central registry for all backend implementations.

This registry allows discovering and instantiating backends by name. Backends can be registered via entry points or programmatically.

Example

>>> # Get a registered backend class
>>> SlackBackend = BackendRegistry.get("slack")
>>> backend = SlackBackend()
>>>
>>> # List all available backends
>>> for name in BackendRegistry.list():
...     print(name)

classmethod register(backend_class: type[BackendBase], name: str | None = None) → type[BackendBase]

Register a backend class.

Parameters:

• backend_class – The backend class to register.

• name – Optional name override. If not provided, uses backend_class.name.

Returns:

The registered backend class (for use as decorator).

Example

>>> @BackendRegistry.register
... class MyBackend(BackendBase):
...     name = "my_backend"
...     ...

classmethod get(name: str) → type[BackendBase]

Get a backend class by name.

Parameters:

name – The backend name.

Returns:

The backend class.

Raises:

KeyError – If the backend is not registered.

classmethod get_instance(name: str, **kwargs: Any) → BackendBase

Get or create a backend instance.

If an instance already exists for this name, returns it. Otherwise creates a new instance.

Parameters:

• name – The backend name.

• **kwargs – Arguments passed to the backend constructor.

Returns:

The backend instance.

classmethod get_format(name: str) → Format

Get the preferred format for a backend.

Parameters:

name – The backend name.

Returns:

The Format enum value.

classmethod list() → list[str]

List all registered backend names.

Returns:

List of backend names.

classmethod items() → Iterator[tuple[str, type[BackendBase]]]

Iterate over all registered backends.

Yields:

Tuples of (name, backend_class).

classmethod clear_instances() → None

Clear all cached backend instances.

classmethod register_all_types() → None

Register all backend types for type conversion.

This iterates over all registered backends and registers their user_class, channel_class, and presence_class with the conversion module. Each backend class should define these ClassVar attributes.

This is called lazily when conversion functions are first used.

chatom.backend.backend_registry.get_backend(name: str) → type[BackendBase]

Get a backend class by name.

Parameters:

name – The backend name.

Returns:

The backend class.

chatom.backend.backend_registry.get_backend_format(name: str) → Format

Get the preferred format for a backend.

This function replaces get_format_for_backend and uses the backend registry.

Parameters:

name – The backend name.

Returns:

The Format enum value.

chatom.backend.backend_registry.list_backends() → list[str]

List all available backend names.

Returns:

List of registered backend names.

chatom.backend.backend_registry.register_backend(backend_class: type[BackendBase] | None = None, *, name: str | None = None) → type[BackendBase] | Callable[[type[BackendBase]], type[BackendBase]]

Register a backend class with the registry.

Can be used as a decorator with or without arguments.

Parameters:

• backend_class – The backend class to register.

• name – Optional name override.

Returns:

The registered class or a decorator.

Example

>>> @register_backend
... class MyBackend(BackendBase):
...     name = "my_backend"
...
>>> @register_backend(name="custom_name")
... class AnotherBackend(BackendBase):
...     ...

Backend base class for chatom.

This module provides the base class that all backends must implement.

chatom.backend.backend.Backend

alias of BackendBase

pydantic model chatom.backend.backend.BackendBase

Bases: BaseModel

Base class for all chat backend implementations.

This provides a unified interface for interacting with chat platforms. All backends must implement the abstract methods.

Backends should be registered with the BackendRegistry via entry points or by calling register_backend().

name

The backend identifier (e.g., ‘slack’, ‘discord’).

display_name

Human-readable name for the backend.

format

The preferred output format for this backend.

capabilities

The capabilities supported by this backend.

connected

Whether currently connected.

users

Registry of cached users.

channels

Registry of cached channels.

Example

>>> class MyBackend(BackendBase):
...     name = "my_backend"
...     display_name = "My Backend"
...     format = Format.MARKDOWN
...
...     async def connect(self):
...         # Implementation
...         pass

name: ClassVar[str] = ''

display_name: ClassVar[str] = ''

format: ClassVar[Format] = 'markdown'

mention_pattern: ClassVar[Pattern[str] | None] = None

field capabilities: BackendCapabilities | None = None

The capabilities supported by this backend.

field connected: bool = False

Whether currently connected.

field users: UserRegistry [Optional]

Registry of cached users.

field channels: ChannelRegistry [Optional]

Registry of cached channels.

field config: Any | None = None

Backend-specific configuration. Subclasses override with their config type.

property sync: SyncHelper

Get the sync helper for calling async methods synchronously.

Returns:

SyncHelper instance that wraps async methods.

Example

>>> backend = MyBackend()
>>> backend.sync.connect()  # Calls connect() synchronously
>>> user = backend.sync.lookup_user(id="123")

get_format() → Format

Get the preferred format for this backend.

Returns:

The Format enum value for this backend.

normalize_channel_id(channel_id: str) → str

Return a canonical form of a channel id for equality comparison.

Some platforms expose the same channel under multiple equivalent id encodings. Backends where that happens should override this to return a single canonical form so that equal channels compare equal. The default returns the id unchanged.

Parameters:

channel_id – The channel id to canonicalize.

Returns:

The canonical channel id.

abstractmethod async connect() → None

Establish connection to the backend.

This should authenticate and establish a connection to the chat platform. After successful connection, connected should be set to True.

Raises:

ConnectionError – If connection fails.

abstractmethod async disconnect() → None

Disconnect from the backend.

This should cleanly close the connection and release resources. After disconnection, connected should be set to False.

async lookup_user(*, id: str | None = None, name: str | None = None, email: str | None = None, handle: str | None = None) → User | None

Look up a user by any identifier.

First checks the local cache, then fetches from the backend if not found. This method will attempt to use the backend’s fetch_user with whatever identifiers are provided.

Parameters:

• id – User ID.

• name – User name.

• email – User email address.

• handle – User handle/username.

Returns:

The user if found, None otherwise.

abstractmethod async fetch_user(identifier: str | User | None = None, *, id: str | None = None, name: str | None = None, email: str | None = None, handle: str | None = None) → User | None

Fetch a user from the backend.

This method accepts flexible input types for convenience: - Pass a User object to validate/refresh it - Pass an ID string as the first positional argument - Use keyword arguments for lookup by name, email, or handle

The backend will attempt to resolve the user using the most efficient method available for the platform.

Parameters:

• identifier – A User object or user ID string.

• id – User ID (alternative to positional identifier).

• name – User name or display name to search for.

• email – Email address to search for.

• handle – Username/handle to search for.

Returns:

The user if found, None otherwise.

Example

>>> # All of these work:
>>> user = await backend.fetch_user("U123456")
>>> user = await backend.fetch_user(id="U123456")
>>> user = await backend.fetch_user(name="John Doe")
>>> user = await backend.fetch_user(email="john@example.com")
>>> user = await backend.fetch_user(existing_user)  # refresh

async lookup_channel(*, id: str | None = None, name: str | None = None) → Channel | None

Look up a channel by any identifier.

First checks the local cache, then fetches from the backend if not found. This method will attempt to use the backend’s fetch_channel with whatever identifiers are provided.

Parameters:

• id – Channel ID.

• name – Channel name.

Returns:

The channel if found, None otherwise.

async lookup_room(*, id: str | None = None, name: str | None = None) → Channel | None

Look up a room by any identifier.

This is an alias for lookup_channel. Use whichever terminology fits your platform (room for Symphony/Matrix, channel for Slack/Discord).

Parameters:

• id – Room ID.

• name – Room name.

Returns:

The room/channel if found, None otherwise.

abstractmethod async fetch_channel(identifier: str | Channel | None = None, *, id: str | None = None, name: str | None = None) → Channel | None

Fetch a channel from the backend.

This method accepts flexible input types for convenience: - Pass a Channel object to validate/refresh it - Pass an ID string as the first positional argument - Use keyword arguments for lookup by name

The backend will attempt to resolve the channel using the most efficient method available for the platform.

Parameters:

• identifier – A Channel object or channel ID string.

• id – Channel ID (alternative to positional identifier).

• name – Channel name to search for.

Returns:

The channel if found, None otherwise.

Example

>>> # All of these work:
>>> channel = await backend.fetch_channel("C123456")
>>> channel = await backend.fetch_channel(id="C123456")
>>> channel = await backend.fetch_channel(name="general")
>>> channel = await backend.fetch_channel(existing_channel)  # refresh

async fetch_room(identifier: str | Channel | None = None, *, id: str | None = None, name: str | None = None) → Channel | None

Fetch a room from the backend.

This is an alias for fetch_channel. Use whichever terminology fits your platform (room for Symphony/Matrix, channel for Slack/Discord).

Parameters:

• identifier – A Channel object or room ID string.

• id – Room ID (alternative to positional identifier).

• name – Room name to search for.

Returns:

The room/channel if found, None otherwise.

async resolve_user(user: User) → User

Resolve an incomplete user to a complete one.

If the user is already complete (has all required fields), returns it as-is. Otherwise, fetches the full user data from the backend.

Parameters:

user – A User object that may be incomplete.

Returns:

A complete User object with all fields populated.

Raises:

ValueError – If the user cannot be resolved (no id or name).

Example

>>> # User from a message might only have id
>>> incomplete_user = message.author
>>> full_user = await backend.resolve_user(incomplete_user)
>>> print(full_user.email)  # Now populated

async resolve_channel(channel: Channel) → Channel

Resolve an incomplete channel to a complete one.

If the channel is already complete (has all required fields), returns it as-is. Otherwise, fetches the full channel data from the backend.

Parameters:

channel – A Channel object that may be incomplete.

Returns:

A complete Channel object with all fields populated.

Raises:

ValueError – If the channel cannot be resolved (no id or name).

Example

>>> # Channel from config might only have name
>>> incomplete_channel = Channel(name="general")
>>> full_channel = await backend.resolve_channel(incomplete_channel)
>>> print(full_channel.id)  # Now populated

async resolve_room(room: Channel) → Channel

Resolve an incomplete room to a complete one.

This is an alias for resolve_channel. Use whichever terminology fits your platform (room for Symphony/Matrix, channel for Slack/Discord).

Parameters:

room – A Channel object that may be incomplete.

Returns:

A complete Channel object with all fields populated.

async fetch_organization(identifier: str | Organization | None = None, *, id: str | None = None, name: str | None = None) → Organization | None

Fetch an organization from the backend.

An organization is the top-level container (guild, workspace, pod, etc.).

Parameters:

• identifier – An Organization object or organization ID string.

• id – Organization ID (alternative to positional identifier).

• name – Organization name to search for.

Returns:

The organization if found, None otherwise.

Raises:

NotImplementedError – If the backend doesn’t support organizations.

async list_organizations() → list[Organization]

List all organizations the bot has access to.

Returns:

List of organizations.

Raises:

NotImplementedError – If the backend doesn’t support organizations.

async fetch_channel_members(identifier: str | Channel | None = None, *, id: str | None = None, name: str | None = None) → list[User]

Fetch members of a channel.

Retrieves the list of users who are members of the specified channel. This is useful for authorization checks, mention validation, or building user interfaces.

This method accepts flexible input types for convenience: - Pass a Channel object to use its ID - Pass an ID string as the first positional argument - Use keyword arguments for lookup by id or name

Parameters:

• identifier – A Channel object or channel ID string.

• id – Channel ID (alternative to positional identifier).

• name – Channel name to search for.

Returns:

List of users who are members of the channel.

Raises:

NotImplementedError – If the backend doesn’t support member listing.

Example

>>> # All of these work:
>>> members = await backend.fetch_channel_members("C123")
>>> members = await backend.fetch_channel_members(id="C123")
>>> members = await backend.fetch_channel_members(name="general")
>>> members = await backend.fetch_channel_members(channel)
>>> for user in members:
...     print(user.name)

async fetch_room_members(identifier: str | Channel | None = None, *, id: str | None = None, name: str | None = None) → list[User]

Fetch members of a room.

This is an alias for fetch_channel_members. Use whichever terminology fits your platform.

Parameters:

• identifier – A Channel object or room ID string.

• id – Room ID (alternative to positional identifier).

• name – Room name to search for.

Returns:

List of users who are members of the room.

async resolve_message(message: Message) → Message

Resolve incomplete nested objects in a Message.

Resolves the message’s author and channel if they are incomplete. This is useful when a message is created with partial information that needs to be filled in before sending.

Parameters:

message – The message to resolve.

Returns:

The message with resolved author and channel.

Example

>>> msg = Message(
...     content="Hello",
...     channel=Channel(name="general"),
...     author=User(email="john@example.com"),
... )
>>> resolved = await backend.resolve_message(msg)
>>> print(resolved.channel.id)  # Now populated

abstractmethod async fetch_messages(channel: str | Channel, limit: int = 100, before: str | Message | datetime | None = None, after: str | Message | datetime | None = None) → list[Message]

Fetch messages from a channel, newest-first.

Returns up to limit messages, ordered newest-to-oldest. before and after bound the range; each accepts a message id, a Message, or a timezone-aware datetime:

• after: only messages at or after this point (lower bound).

• before: only messages at or before this point (upper bound).

When a range is given, implementations page the underlying API to cover the whole range without dropping messages (subject to limit); when no range is given, they return the most recent limit messages.

Parameters:

• channel – The channel to fetch messages from (ID string or Channel object).

• limit – Maximum number of messages to return.

• before – Upper bound — message id, Message, or datetime.

• after – Lower bound — message id, Message, or datetime.

Returns:

List of messages, ordered newest-to-oldest.

Example

>>> # Most recent 50:
>>> msgs = await backend.fetch_messages("C123", limit=50)
>>> # Everything in the last 30 minutes:
>>> from datetime import datetime, timedelta, timezone
>>> since = datetime.now(timezone.utc) - timedelta(minutes=30)
>>> msgs = await backend.fetch_messages("C123", after=since)

async search_messages(query: str, channel: str | Channel | None = None, limit: int = 50, **kwargs: Any) → list[Message]

Search for messages matching a query.

Searches message content across channels. Requires the MESSAGE_SEARCH capability to be supported by the backend.

Parameters:

• query – The search query string.

• channel – Optional channel to limit search to (ID string or Channel object).

• limit – Maximum number of results to return.

• **kwargs – Additional platform-specific search options (e.g., from_user, has_file, date range).

Returns:

List of messages matching the query.

Raises:

NotImplementedError – If the backend doesn’t support MESSAGE_SEARCH capability.

Example

>>> # Search all channels
>>> results = await backend.search_messages("important meeting")
>>> # Search specific channel
>>> results = await backend.search_messages("bug fix", channel="C123")
>>> # With filters
>>> results = await backend.search_messages("report", from_user="U123")

async fetch_new_messages(channel: str | Channel, after: str | None = None) → list[Message]

Fetch new messages from a channel.

This is a convenience method that fetches messages after a specific point, typically used for getting updates.

Parameters:

• channel – The channel to fetch messages from (ID string or Channel object).

• after – Fetch messages after this message ID.

Returns:

List of new messages.

abstractmethod async send_message(channel: str | Channel, content: str, **kwargs: Any) → Message

Send a message to a channel.

Standardized optional kwargs (recognized by every backend):

• thread: str | Thread | Message | None — send into an existing thread. When a Message is passed, the message’s thread is used if set, otherwise the message itself becomes the thread root. Backends translate this to their native concept (Slack thread_ts, Discord thread channel, Telegram message_thread_id). Symphony has no thread concept and silently ignores this.

• reply_to: str | Message | None — reply referencing a specific message (Discord reference=, Telegram reply_to_message_id, Slack thread_ts). Symphony has no native reply and silently ignores this.

Parameters:

• channel – The channel to send to (ID string or Channel object).

• content – The message content.

• **kwargs – Additional platform-specific options (e.g., embeds, attachments, thread, reply_to).

Returns:

The sent message.

Example

>>> # All of these work:
>>> msg = await backend.send_message("C123", "Hello!")
>>> msg = await backend.send_message(Channel(id="C123"), "Hello!")
>>> msg = await backend.send_message(Channel(name="general"), "Hello!")  # Resolves
>>> # Thread and reply:
>>> msg = await backend.send_message("C123", "In thread", thread=parent_msg)
>>> msg = await backend.send_message("C123", "Replying", reply_to=parent_msg)

async edit_message(message: str | Message, content: str, channel: str | Channel | None = None, **kwargs: Any) → Message

Edit an existing message.

Parameters:

• message – The message to edit (ID string or Message object).

• content – The new message content.

• channel – The channel containing the message (required if message is a string).

• **kwargs – Additional platform-specific options.

Returns:

The edited message.

Raises:

NotImplementedError – If the backend doesn’t support editing.

Example

>>> # Edit using Message object
>>> edited = await backend.edit_message(msg, "Updated content")
>>> # Edit using IDs
>>> edited = await backend.edit_message("M123", "Updated", channel="C123")

async upload_file(channel: str | Channel, data: bytes, filename: str = 'file', content_type: str = '', title: str = '', content: str = '', **kwargs: Any) → Message

Upload a file with binary data to a channel.

Backends should override this to use their native file upload API (e.g. Slack files.uploadV2, Discord File, Telegram send_document / send_photo, Symphony attachment API).

The default implementation falls back to send_message with a text-only placeholder.

Parameters:

• channel – The channel to upload to (ID string or Channel object).

• data – Raw file bytes.

• filename – Name of the file.

• content_type – MIME type of the file.

• title – Optional title for the upload.

• content – Optional accompanying text message.

• **kwargs – Additional platform-specific options.

Returns:

The sent message.

async download_attachment(attachment: Attachment, *, message: Message | None = None) → bytes

Download the binary content of an attachment.

Returns the raw bytes of a file/image/document that was received in chat. The default implementation returns attachment.data if it is already populated, otherwise performs an HTTP GET against attachment.url.

Backends override this to add platform authentication (Slack url_private bearer token, Telegram getFile, Symphony attachment API) where a plain public download is not possible.

Parameters:

• attachment – The attachment to download. Must carry either in-memory data, a downloadable url, or a platform file id (for backends that resolve by ID).

• message – The message the attachment belongs to. Some backends (e.g. Symphony) require the message and channel context to resolve the download.

Returns:

The raw file bytes.

Raises:

NotImplementedError – If the attachment cannot be resolved to a downloadable source.

Example

>>> for att in message.attachments:
...     data = await backend.download_attachment(att, message=message)
...     Path(att.filename).write_bytes(data)

async delete_message(message: str | Message, channel: str | Channel | None = None) → None

Delete a message.

Parameters:

• message – The message to delete (ID string or Message object).

• channel – The channel containing the message (required if message is a string).

Raises:

NotImplementedError – If the backend doesn’t support deletion.

Example

>>> # Delete using Message object
>>> await backend.delete_message(msg)
>>> # Delete using IDs
>>> await backend.delete_message("M123", channel="C123")

async reply_in_thread(message: Message, content: str, **kwargs: Any) → Message

Reply to a message in its thread.

This is a convenience method that sends a reply in the thread of the given message. If the message is not in a thread, it creates a new thread from that message.

This method simplifies thread-based conversations for bot developers by abstracting the platform-specific details of thread handling.

Parameters:

• message – The message to reply to. Thread context is extracted from message.thread_id or message.id.

• content – The reply content.

• **kwargs – Additional platform-specific options (e.g., embeds, attachments).

Returns:

The sent reply message.

Raises:

NotImplementedError – If the backend doesn’t support threading.

Example

>>> # Reply in a thread
>>> async def on_message(message: Message):
...     reply = await backend.reply_in_thread(
...         message,
...         "Thanks for your message!"
...     )

async forward_message(message: Message, to_channel: str | Channel, *, include_attribution: bool = True, prefix: str | None = None, **kwargs: Any) → Message

Forward a message to another channel.

This method forwards a message from one channel to another, optionally including attribution about the original source. The forwarded message can include the original content, attachments, and optionally embeds.

Different backends handle forwarding differently: - Some have native forwarding (Discord has reference links) - Others simulate it by re-sending with attribution

Parameters:

• message – The message to forward.

• to_channel – The destination channel (ID string or Channel object).

• include_attribution – If True (default), include information about the original message source (author, channel, time).

• prefix – Optional text to prepend to the forwarded message.

• **kwargs – Additional platform-specific options (e.g., embeds, thread_id for threading).

Returns:

The forwarded message in the destination channel.

Raises:

• NotImplementedError – If the backend doesn’t support forwarding.

• ValueError – If the message or channel cannot be resolved.

Example

>>> # Forward a message to another channel
>>> forwarded = await backend.forward_message(
...     incoming_message,
...     to_channel="C456789",
... )
>>>
>>> # Forward with custom prefix
>>> forwarded = await backend.forward_message(
...     incoming_message,
...     to_channel=Channel(name="alerts"),
...     prefix="⚠️ Escalated: ",
... )
>>>
>>> # Forward without attribution
>>> forwarded = await backend.forward_message(
...     incoming_message,
...     to_channel="C456789",
...     include_attribution=False,
... )

async listen(channel: str | Channel | None = None, skip_own: bool = True) → AsyncIterator[Message]

Listen for incoming messages in real-time.

This async generator yields Message objects as they arrive from the chat platform. It abstracts away the platform-specific details of real-time messaging (WebSockets, Socket Mode, datafeed, etc.).

This is an alias for stream_messages() with skip_history=True.

Parameters:

• channel – Optional channel to filter messages to a specific channel (ID string or Channel object). If None, yields messages from all channels the bot has access to.

• skip_own – If True (default), skip messages sent by the bot itself.

Yields:

Message – Each message as it arrives.

Raises:

• NotImplementedError – If the backend doesn’t support streaming.

• ConnectionError – If the real-time connection fails.

Example

>>> async for message in backend.listen():
...     print(f"Received: {message.content}")
...     if "!ping" in message.content:
...         await backend.reply_in_thread(message, "Pong!")

async stream_messages(channel: str | Channel | None = None, skip_own: bool = True, skip_history: bool = True) → AsyncIterator[Message]

Stream incoming messages in real-time.

This async generator yields Message objects as they arrive from the chat platform. It abstracts away the platform-specific details of real-time messaging (WebSockets, Socket Mode, datafeed, etc.).

The stream continues until the generator is closed or an error occurs.

Parameters:

• channel – Optional channel to filter messages to a specific channel (ID string or Channel object). If None, yields messages from all channels the bot has access to.

• skip_own – If True (default), skip messages sent by the bot itself.

• skip_history – If True (default), skip messages that existed before the stream started. Only yields new messages.

Yields:

Message – Each message as it arrives.

Raises:

• NotImplementedError – If the backend doesn’t support streaming.

• ConnectionError – If the real-time connection fails.

Example

>>> async for message in backend.stream_messages():
...     print(f"Received: {message.content}")
...     if message.mentions_bot:
...         await backend.reply_in_thread(message, "Hello!")
>>>
>>> # Filter to a specific channel
>>> async for message in backend.stream_messages(channel="C123"):
...     await process_message(message)
>>>
>>> # Filter using Channel object
>>> async for message in backend.stream_messages(channel=Channel(name="general")):
...     await process_message(message)

async stream_interactions(channel: str | Channel | None = None) → AsyncIterator[Interaction]

Stream incoming component interactions in real-time.

Yields Interaction objects each time a user clicks a button, picks from a select menu, or submits a modal that was sent via this backend.

Backends that don’t natively support interactive components, or where interaction streaming hasn’t been implemented yet, should leave this as NotImplementedError.

Parameters:

channel – Optional channel filter.

Yields:

Interaction – Each component interaction as it arrives.

Raises:

NotImplementedError – If the backend doesn’t support interaction streaming.

Example

>>> async for event in backend.stream_interactions():
...     if event.action_id == "confirm":
...         await handle_confirm(event)

async read_messages(channel: str | Channel, limit: int = 100, before: str | None = None, after: str | None = None) → AsyncIterator[Message]

Read message history from a channel as an async iterator.

This is a convenience async generator that yields messages from a channel’s history one at a time. It wraps fetch_messages for easier iteration.

Parameters:

• channel – The channel to read messages from (ID string or Channel object).

• limit – Maximum number of messages to read.

• before – Read messages before this message ID (for pagination).

• after – Read messages after this message ID (for pagination).

Yields:

Message – Each message from the history.

Example

>>> async for message in backend.read_messages("C123", limit=50):
...     print(f"{message.author.name}: {message.content}")

async read_thread(channel: str | Channel, thread_id: str, limit: int = 100) → AsyncIterator[Message]

Read messages from a thread.

This async generator yields messages from a thread/conversation one at a time. Threads are platform-specific: - Slack: message thread (thread_ts) - Discord: thread channel - Symphony: reply chain

Parameters:

• channel – The parent channel containing the thread (ID string or Channel object).

• thread_id – The thread identifier (message ID that started the thread).

• limit – Maximum number of messages to read.

Yields:

Message – Each message from the thread.

Raises:

NotImplementedError – If the backend doesn’t support threads.

Example

>>> async for message in backend.read_thread("C123", "1234567890.123456"):
...     print(f"{message.author.name}: {message.content}")

async create_thread(channel: str | Channel, message_id: str, name: str, **kwargs: Any) → Channel

Create a thread from a message.

Creates a new thread attached to the specified message. This is primarily used for Discord-style threads where a thread is a separate channel derived from a message.

For Slack-style threads that are implicit (just replies to a message), use reply_in_thread() instead.

Parameters:

• channel – The channel containing the message (ID string or Channel object).

• message_id – The ID of the message to create a thread from.

• name – The name/title for the thread.

• **kwargs – Additional platform-specific options: - auto_archive_duration: Discord thread archive duration in minutes. - type: Discord thread type (public, private, announcement). - reason: Audit log reason.

Returns:

The created thread as a Channel object.

Raises:

NotImplementedError – If the backend doesn’t support thread creation.

Example

>>> thread = await backend.create_thread(
...     channel="C123",
...     message_id="M456",
...     name="Discussion Thread",
... )
>>> await backend.send_message(thread.id, "First message in thread!")

async reply_to_message(channel: str | Channel, message_id: str, content: str, **kwargs: Any) → Message

Reply to a specific message.

Sends a reply that references the original message. How this is displayed depends on the platform: - Slack: In thread (if thread support) or as inline reply - Discord: As a reply with reference link - Symphony: As a reply to the message

For thread-based replies, consider using reply_in_thread() instead which handles the thread context automatically.

Parameters:

• channel – The channel containing the message (ID string or Channel object).

• message_id – The ID of the message to reply to.

• content – The reply content.

• **kwargs – Additional platform-specific options.

Returns:

The sent reply message.

Raises:

NotImplementedError – If the backend doesn’t support replies.

Example

>>> reply = await backend.reply_to_message(
...     channel="C123",
...     message_id="M456",
...     content="Thanks for your message!",
... )

async get_bot_info() → User | None

Get information about the connected bot user.

Returns the User object representing the bot/service account that is currently connected. This is useful for checking if messages mention the bot.

Returns:

The bot’s User object, or None if not available.

Raises:

NotImplementedError – If the backend doesn’t support this.

Example

>>> bot = await backend.get_bot_info()
>>> print(f"Connected as: {bot.name} ({bot.id})")

async set_presence(status: str, status_text: str | None = None, **kwargs: Any) → None

Set the current user’s presence status.

Parameters:

• status – The presence status (e.g., ‘online’, ‘away’, ‘dnd’).

• status_text – Optional status message/text.

• **kwargs – Additional platform-specific options.

Raises:

NotImplementedError – If the backend doesn’t support presence.

async get_presence(user: str | User) → Presence | None

Get a user’s presence status.

Parameters:

user – The user to get presence for (ID string or User object).

Returns:

The user’s presence, or None if not available.

Raises:

NotImplementedError – If the backend doesn’t support presence.

Example

>>> # All of these work:
>>> presence = await backend.get_presence("U123")
>>> presence = await backend.get_presence(User(id="U123"))
>>> presence = await backend.get_presence(User(email="john@example.com"))  # Resolves

start_presence_heartbeat(interval_seconds: int = 60, status: str = 'online', status_text: str | None = None) → None

Start an automatic presence heartbeat.

This method periodically sets the user’s presence to keep the bot appearing online. This is useful for platforms that require regular presence updates or for bots that need to maintain an active status.

The heartbeat runs in the background and can be stopped with stop_presence_heartbeat().

Parameters:

• interval_seconds – How often to send presence updates (default 60).

• status – The presence status to set (default ‘online’).

• status_text – Optional status message/text to display.

Example

>>> # Start keeping the bot online
>>> backend.start_presence_heartbeat(60, "online", "Ready to help!")
>>>
>>> # Later, stop the heartbeat
>>> backend.stop_presence_heartbeat()

stop_presence_heartbeat() → None

Stop the automatic presence heartbeat.

Cancels any running presence heartbeat task started by start_presence_heartbeat().

Example

>>> backend.stop_presence_heartbeat()

property is_presence_heartbeat_active: bool

Check if the presence heartbeat is currently running.

Returns:

True if the heartbeat is active.

Return type:

bool

async add_reaction(message: str | Message, emoji: str, channel: str | Channel | None = None) → None

Add a reaction to a message.

Parameters:

• message – The message to react to (ID string or Message object).

• emoji – The emoji to add (name or unicode).

• channel – The channel containing the message (required if message is a string).

Raises:

NotImplementedError – If the backend doesn’t support reactions.

Example

>>> # React using Message object
>>> await backend.add_reaction(msg, "👍")
>>> # React using IDs
>>> await backend.add_reaction("M123", "👍", channel="C123")

async remove_reaction(message: str | Message, emoji: str, channel: str | Channel | None = None) → None

Remove a reaction from a message.

Parameters:

• message – The message to remove reaction from (ID string or Message object).

• emoji – The emoji to remove (name or unicode).

• channel – The channel containing the message (required if message is a string).

Raises:

NotImplementedError – If the backend doesn’t support reactions.

Example

>>> # Remove reaction using Message object
>>> await backend.remove_reaction(msg, "👍")
>>> # Remove reaction using IDs
>>> await backend.remove_reaction("M123", "👍", channel="C123")

async create_dm(users: list[str | User]) → str | None

Create a direct message (DM) or instant message (IM) channel.

Creates a private conversation with one or more users. For single users, this creates a 1:1 DM. For multiple users, this may create a group DM/MIM depending on the platform.

Parameters:

users – List of users to include in the DM (ID strings or User objects).

Returns:

The channel/stream ID of the created DM, or None if failed.

Raises:

NotImplementedError – If the backend doesn’t support DM creation.

Example

>>> # Create DM with user IDs
>>> dm_id = await backend.create_dm(["U123", "U456"])
>>> # Create DM with User objects
>>> dm_id = await backend.create_dm([User(id="U123")])
>>> # Create DM with incomplete User (will resolve)
>>> dm_id = await backend.create_dm([User(email="john@example.com")])

async create_im(users: list[str | User]) → str | None

Create an instant message (IM) channel.

This is an alias for create_dm. Use whichever terminology fits your platform (IM for Symphony, DM for Discord/Slack).

Parameters:

users – List of users to include in the IM (ID strings or User objects).

Returns:

The channel/stream ID of the created IM, or None if failed.

async send_dm(user: str | User, content: str, **kwargs: Any) → Message

Send a direct message to a user.

This is a convenience method that creates a DM channel with the user if needed, then sends the message. It simplifies the common pattern of:

dm_id = await backend.create_dm([user]) await backend.send_message(dm_id, content)

Parameters:

• user – The user to send to (ID string or User object).

• content – The message content.

• **kwargs – Additional platform-specific options passed to send_message.

Returns:

The sent message.

Raises:

• ValueError – If the DM channel could not be created.

• NotImplementedError – If the backend doesn’t support DM creation.

Example

>>> # Send DM using user ID
>>> msg = await backend.send_dm("U123", "Hello!")
>>> # Send DM using User object
>>> msg = await backend.send_dm(user, "Hello!")
>>> # With additional options
>>> msg = await backend.send_dm(user, "Check this!", thread_id="T123")

async create_channel(name: str, description: str = '', public: bool = True, **kwargs: Any) → str | None

Create a new channel.

Creates a channel/room for group communication.

Parameters:

• name – The channel name.

• description – Optional channel description/purpose.

• public – Whether the channel is public (default True).

• **kwargs – Additional platform-specific options: - read_only: Whether the channel is read-only (Symphony). - topic: Channel topic (Slack). - category_id: Category to create under (Discord).

Returns:

The channel ID of the created channel, or None if failed.

Raises:

NotImplementedError – If the backend doesn’t support channel creation.

async create_room(name: str, description: str = '', public: bool = True, **kwargs: Any) → str | None

Create a new room.

This is an alias for create_channel. Use whichever terminology fits your platform (room for Symphony/Matrix, channel for Slack/Discord).

Parameters:

• name – The room name.

• description – Optional room description.

• public – Whether the room is public (default True).

• **kwargs – Additional platform-specific options.

Returns:

The room/stream ID of the created room, or None if failed.

async join_channel(channel: str | Channel, **kwargs: Any) → None

Join a channel.

Makes the bot/user a member of the specified channel.

Parameters:

• channel – The channel to join (ID string or Channel object).

• **kwargs – Additional platform-specific options: - key: Channel password/key (IRC). - invite_code: Invite code (Discord).

Raises:

NotImplementedError – If the backend doesn’t support joining channels.

Example

>>> # Join using channel ID
>>> await backend.join_channel("C123")
>>> # Join using Channel object
>>> await backend.join_channel(Channel(name="general"))

async join_room(room: str | Channel, **kwargs: Any) → None

Join a room.

This is an alias for join_channel. Use whichever terminology fits your platform.

Parameters:

• room – The room to join (ID string or Channel object).

• **kwargs – Additional platform-specific options.

async leave_channel(channel: str | Channel, **kwargs: Any) → None

Leave a channel.

Removes the bot/user from the specified channel.

Parameters:

• channel – The channel to leave (ID string or Channel object).

• **kwargs – Additional platform-specific options: - message: Part message (IRC).

Raises:

NotImplementedError – If the backend doesn’t support leaving channels.

Example

>>> # Leave using channel ID
>>> await backend.leave_channel("C123")
>>> # Leave using Channel object
>>> await backend.leave_channel(Channel(id="C123", name="general"))

async leave_room(room: str | Channel, **kwargs: Any) → None

Leave a room.

This is an alias for leave_channel. Use whichever terminology fits your platform.

Parameters:

• room – The room to leave (ID string or Channel object).

• **kwargs – Additional platform-specific options.

async send_action(target: str | Channel | User, action: str) → None

Send an action/emote message.

Sends an action message (like IRC’s /me command). On IRC this is a CTCP ACTION. On other platforms, this may be formatted as italicized text or similar.

Parameters:

• target – The channel or user to send to (ID string, Channel, or User).

• action – The action text (e.g., “waves hello”).

Raises:

NotImplementedError – If the backend doesn’t support actions.

Example

>>> # Send action to channel
>>> await backend.send_action(Channel(name="#general"), "waves hello")
>>> # Send action to user
>>> await backend.send_action(User(id="U123"), "waves hello")

async send_notice(target: str | Channel | User, text: str) → None

Send a notice message.

Sends a notice (typically displayed differently from regular messages). On IRC this is a NOTICE. Other platforms may not distinguish notices.

Parameters:

• target – The channel or user to send to (ID string, Channel, or User).

• text – The notice text.

Raises:

NotImplementedError – If the backend doesn’t support notices.

Example

>>> # Send notice to channel
>>> await backend.send_notice(Channel(name="#general"), "Server maintenance")
>>> # Send notice to user
>>> await backend.send_notice(User(id="U123"), "You have been warned")

mention_user(user: User) → str

Format a user mention for this backend.

Parameters:

user – The user to mention.

Returns:

The formatted mention string.

mention_channel(channel: Channel) → str

Format a channel mention for this backend.

Parameters:

channel – The channel to mention.

Returns:

The formatted mention string.

channel_link(channel: str | Channel) → str

Generate a clickable channel link/mention for this backend.

This is a convenience method that accepts either a channel ID string or a Channel object and returns the appropriate platform-specific channel reference.

Parameters:

channel – A channel ID string or Channel object.

Returns:

The formatted channel link/mention string.

Example

>>> # From channel ID
>>> link = backend.channel_link("C123")
>>> # From Channel object
>>> link = backend.channel_link(channel)
>>> # Use in message
>>> await backend.send_message(ch, f"Join us in {backend.channel_link('general')}")

mention_here() → str

Format an @here mention for this backend.

Notifies all users who are currently active/online in the channel. Subclasses should override this with platform-specific format.

Returns:

The formatted @here mention string.

mention_everyone() → str

Format an @everyone mention for this backend.

Notifies all members of the channel/server. Subclasses should override this with platform-specific format.

Returns:

The formatted @everyone mention string.

mention_channel_all() → str

Format an @channel mention for this backend.

Notifies all members of the current channel (Slack-specific concept). For platforms without this distinction, defaults to @everyone. Subclasses should override this with platform-specific format.

Returns:

The formatted @channel mention string.

class chatom.backend.backend.SyncHelper(backend: BackendBase)

Bases: object

Helper class to run async methods synchronously.

This provides a convenient way to call async methods from sync code by managing an event loop in a background thread. Uses __getattr__ to dynamically wrap any async method on the backend.

Example

>>> backend = MyBackend()
>>> # Call async method synchronously
>>> user = backend.sync.lookup_user(id="123")
>>> # Any async method can be called:
>>> backend.sync.connect()
>>> backend.sync.send_message(channel_id="C123", content="Hello")

close() → None

Clean up resources.