Backend APIs¶

Concrete configuration, models, mention helpers, backends, and testing backends for Discord, Slack, Symphony, and Telegram.

Discord¶

Discord backend configuration.

This module provides configuration classes for the Discord backend.

pydantic model chatom.discord.config.DiscordConfig[source]¶

Bases: BackendConfig

Configuration for Discord backend.

This configuration is used to connect to Discord using the discord.py library. You need a bot token from the Discord Developer Portal.

bot_token¶: Discord bot token (can be a file path).

application_id¶: Discord application ID.

guild_id¶: Default guild/server ID (optional).

intents¶: Discord gateway intents to request.

command_prefix¶: Prefix for bot commands (if using commands extension).

Example

>>> config = DiscordConfig(
...     bot_token="your-bot-token",
...     application_id="123456789",
...     intents=["guilds", "messages", "message_content"],
... )
>>> backend = DiscordBackend(config=config)

field token: str | SecretStr = SecretStr('')¶: Discord bot token from Developer Portal (can be a file path).

field application_id: str = ''¶: Discord application ID.

field guild_id: str = ''¶: Default guild/server ID (optional).

field intents: list [Optional]¶: Discord gateway intents to request.

field command_prefix: str = '!'¶: Prefix for bot commands.

field shard_id: int | None = None¶: Shard ID for sharded bots.

field shard_count: int | None = None¶: Total number of shards.

property bot_token_str: str¶

Get the bot token as a plain string.

Returns:: The bot token string.

property has_token: bool¶

Check if a bot token is configured.

Returns:: True if a bot token is set.

Discord-specific User model.

This module provides the Discord-specific User class.

pydantic model chatom.discord.user.DiscordUser[source]¶

Bases: User

Discord-specific user with additional Discord fields.

discriminator¶: The user’s 4-digit discriminator (legacy).

global_name¶: The user’s global display name.

is_system¶: Whether this is a Discord system user.

accent_color¶: The user’s banner color.

banner_url¶: URL to the user’s banner image.

field discriminator: str = '0'¶: The user’s 4-digit discriminator (legacy).

field global_name: str | None = None¶: The user’s global display name.

field is_system: bool = False¶: Whether this is a Discord system user.

field accent_color: int | None = None¶: The user’s banner color.

field banner_url: str = ''¶: URL to the user’s banner image.

property full_username: str¶

Get the full username with discriminator (legacy format).

Returns:: Username#discriminator or just username for new usernames.
Return type:: str

classmethod from_discord_user(user: Any) → DiscordUser[source]¶

Create a DiscordUser from a discord.py User or Member object.

This factory method converts a discord.py User or Member object into a chatom DiscordUser.

Parameters:: user – A discord.py User or Member object.
Returns:: A DiscordUser instance.

Example

>>> # From a discord.py event
>>> discord_user = DiscordUser.from_discord_user(event.author)

Discord-specific Channel model.

This module provides the Discord-specific Channel class.

pydantic model chatom.discord.channel.DiscordChannel[source]¶

Bases: Channel

Discord-specific channel with additional Discord fields.

guild¶: The guild/server this channel belongs to.

position¶: Sorting position of the channel.

nsfw¶: Whether the channel is NSFW.

slowmode_delay¶: Slowmode delay in seconds.

discord_type¶: Discord-specific channel type.

bitrate¶: Voice channel bitrate.

user_limit¶: Voice channel user limit.

rate_limit_per_user¶: Slowmode rate limit.

field guild: Organization | None = None¶: The guild/server this channel belongs to.

field position: int = 0¶: Sorting position of the channel.

field nsfw: bool = False¶: Whether the channel is NSFW.

field slowmode_delay: int = 0¶: Slowmode delay in seconds.

field discord_type: DiscordChannelType = DiscordChannelType.GUILD_TEXT¶: Discord-specific channel type.

field bitrate: int | None = None¶: Voice channel bitrate.

field user_limit: int | None = None¶: Voice channel user limit.

field rate_limit_per_user: int = 0¶: Slowmode rate limit.

property guild_id: str¶

Get the guild/server ID.

Returns:: The guild ID, or empty string if not set.
Return type:: str

property is_voice: bool¶

Check if this is a voice channel.

Returns:: True if this is a voice channel.
Return type:: bool

property is_text: bool¶

Check if this is a text channel.

Returns:: True if this is a text channel.
Return type:: bool

class chatom.discord.channel.DiscordChannelType(value)[source]¶

Bases: str, Enum

Discord-specific channel types.

Discord-specific Guild (Organization) model.

This module provides the Discord-specific Guild class which extends Organization.

pydantic model chatom.discord.guild.DiscordGuild[source]¶

Bases: Organization

Discord-specific guild (server) with additional Discord fields.

A Discord guild is the organization unit in Discord (commonly called a “server”).

premium_tier¶: The guild’s boost level (0-3).

nsfw_level¶: The guild’s NSFW content level.

preferred_locale¶: The guild’s preferred locale.

approximate_member_count¶: Approximate number of members (may differ from member_count).

approximate_presence_count¶: Approximate number of online members.

vanity_url_code¶: The guild’s vanity invite code if set.

features¶: List of guild features (e.g., “COMMUNITY”, “PARTNERED”).

field premium_tier: int = 0¶: The guild’s boost level (0-3).

field nsfw_level: int = 0¶: The guild’s NSFW content level.

field preferred_locale: str = 'en-US'¶: The guild’s preferred locale.

field approximate_member_count: int | None = None¶: Approximate number of members.

field approximate_presence_count: int | None = None¶: Approximate number of online members.

field vanity_url_code: str | None = None¶: The guild’s vanity invite code if set.

field features: list [Optional]¶: List of guild features.

classmethod from_discord_guild(guild: Any) → DiscordGuild[source]¶

Create a DiscordGuild from a discord.py Guild object.

Parameters:: guild – A discord.py Guild object.
Returns:: DiscordGuild instance.

Discord-specific Message model.

This module provides the Discord-specific Message class.

pydantic model chatom.discord.message.DiscordMessage[source]¶

Bases: Message

Discord-specific message with additional Discord fields.

Based on the Discord API message structure.

Note: Use the inherited channel field for the DiscordChannel object, channel_id is synced from channel.id. Similarly for author/author_id. The mentions field here is a List[str] of user IDs for Discord-specific use, while the base class mention_ids is synced from it.

discord_type¶: The Discord message type.

guild¶: The guild/server this message was sent in.

member¶: Guild member data for the author.

mention_everyone¶: Whether @everyone was mentioned.

mention_roles¶: List of mentioned role IDs.

mention_channels¶: List of mentioned channel IDs.

nonce¶: Used for message send verification.

pinned¶: Whether the message is pinned.

webhook_id¶: Webhook ID if sent by a webhook.

flags¶: Message flags.

interaction¶: Interaction data if from an interaction.

components¶: Message components (buttons, etc.).

sticker_items¶: Stickers in the message.

position¶: Position in thread.

field discord_type: DiscordMessageType = DiscordMessageType.DEFAULT¶: The Discord message type.

field guild: Organization | None = None¶: The guild/server this message was sent in.

field member: dict[str, Any] | None = None¶: Guild member data for the author.

field mention_everyone: bool = False¶: Whether @everyone was mentioned.

field mention_roles: list[str] [Optional]¶: List of mentioned role IDs.

field mention_channels: list[str] [Optional]¶: List of mentioned channel IDs.

field nonce: str | None = None¶: Used for message send verification.

field pinned: bool = False¶: Whether the message is pinned.

field webhook_id: str | None = None¶: Webhook ID if sent by a webhook.

field flags: int = 0¶: Message flags.

field interaction: dict[str, Any] | None = None¶: Interaction data if from an interaction.

field components: list[dict[str, Any]] [Optional]¶: Message components (buttons, select menus, etc.).

field sticker_items: list[dict[str, Any]] [Optional]¶: Stickers in the message.

field position: int | None = None¶: Position in thread.

property guild_id: str¶

Get the guild/server ID.

Returns:: The guild ID, or empty string if not set.
Return type:: str

property is_reply: bool¶: Check if this message is a reply.

property is_ephemeral: bool¶: Check if this is an ephemeral message.

property is_crossposted: bool¶: Check if this message was crossposted.

property has_thread: bool¶: Check if this message has a thread.

property is_voice_message: bool¶: Check if this is a voice message.

property suppresses_embeds: bool¶: Check if embeds are suppressed.

property suppresses_notifications: bool¶: Check if notifications are suppressed.

has_flag(flag: DiscordMessageFlags) → bool[source]¶

Check if a specific flag is set.

Parameters:: flag – The flag to check.
Returns:: True if the flag is set.

to_formatted() → FormattedMessage[source]¶

Convert this Discord message to a FormattedMessage.

Parses Discord markdown formatting and converts to a FormattedMessage that can be rendered for other backends.

Returns:: The formatted message representation.
Return type:: FormattedMessage

classmethod from_formatted(formatted: FormattedMessage, backend: str = '', **kwargs: Any) → DiscordMessage[source]¶

Create a DiscordMessage from a FormattedMessage.

Renders the FormattedMessage in Discord markdown format.

Parameters:

formatted – The FormattedMessage to convert.
backend – Target backend (ignored, always uses discord format).
**kwargs – Additional message attributes.

Returns:

A new DiscordMessage instance.

Return type:

DiscordMessage

classmethod from_api_response(data: dict[str, Any]) → DiscordMessage[source]¶

Create a DiscordMessage from a Discord API response.

Parameters:: data – The API response data.
Returns:: A DiscordMessage instance.

class chatom.discord.message.DiscordMessageFlags(value)[source]¶

Bases: IntEnum

Discord message flags.

Based on Discord API message flags.

class chatom.discord.message.DiscordMessageType(value)[source]¶

Bases: IntEnum

Discord message types.

Based on Discord API message types.

Discord-specific Presence model.

This module provides the Discord-specific Presence class.

class chatom.discord.presence.DiscordActivityType(value)[source]¶

Bases: str, Enum

Discord-specific activity types.

pydantic model chatom.discord.presence.DiscordPresence[source]¶

Bases: Presence

Discord-specific presence.

activities¶: List of activities the user is engaged in.

client_status¶: Status per client (desktop, mobile, web).

field activities: list [Optional]¶: List of activities the user is engaged in.

field desktop_status: PresenceStatus = PresenceStatus.OFFLINE¶: Status on desktop client.

field mobile_status: PresenceStatus = PresenceStatus.OFFLINE¶: Status on mobile client.

field web_status: PresenceStatus = PresenceStatus.OFFLINE¶: Status on web client.

Discord-specific mention utilities.

This module registers Discord-specific mention formatting.

chatom.discord.mention.mention_channel(channel: Channel) → str[source]¶

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

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

chatom.discord.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.discord.mention.mention_user(user: User) → str[source]¶

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

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

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

chatom.discord.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'

Discord backend implementation for chatom.

This module provides the Discord backend using the discord.py library.

pydantic model chatom.discord.backend.DiscordBackend[source]¶

Bases: BackendBase

Discord backend implementation using discord.py.

This provides the backend interface for Discord using the discord.py library. It supports all standard backend operations including messaging, presence, and reactions.

name¶: The backend identifier (‘discord’).

display_name¶: Human-readable name.

format¶: Discord uses its own markdown flavor.

capabilities¶: Discord-specific capabilities.

config¶: Discord-specific configuration.

Example

>>> from chatom.discord import DiscordBackend, DiscordConfig
>>> config = DiscordConfig(bot_token="your-token")
>>> backend = DiscordBackend(config=config)
>>> await backend.connect()
>>> user = await backend.fetch_user("123456789")

name: ClassVar[str] = 'discord'¶

display_name: ClassVar[str] = 'Discord'¶

format: ClassVar[Format] = 'discord-markdown'¶

mention_pattern: ClassVar[Pattern | None] = re.compile('<@!?(\\d+)>')¶

user_class¶: alias of DiscordUser

channel_class¶: alias of DiscordChannel

presence_class¶: alias of DiscordPresence

guild_class¶: alias of DiscordGuild

field capabilities: BackendCapabilities | None = BackendCapabilities(capabilities=frozenset({<Capability.DELETING: 'deleting'>, <Capability.TYPING_INDICATORS: 'typing_indicators'>, <Capability.PINNING: 'pinning'>, <Capability.USER_MENTIONS: 'user_mentions'>, <Capability.THREADS: 'threads'>, <Capability.ORGANIZATIONS: 'organizations'>, <Capability.VIDEOS: 'videos'>, <Capability.BUTTONS: 'buttons'>, <Capability.EMBEDS: 'embeds'>, <Capability.MARKDOWN: 'markdown'>, <Capability.PLAINTEXT: 'plaintext'>, <Capability.ROLE_MENTIONS: 'role_mentions'>, <Capability.CUSTOM_EMOJI: 'custom_emoji'>, <Capability.CHANNEL_MENTIONS: 'channel_mentions'>, <Capability.CODE_BLOCKS: 'code_blocks'>, <Capability.FILES: 'files'>, <Capability.EVERYONE_MENTION: 'everyone_mention'>, <Capability.IMAGES: 'images'>, <Capability.PRESENCE: 'presence'>, <Capability.FORWARDING: 'forwarding'>, <Capability.REPLIES: 'replies'>, <Capability.AUDIO: 'audio'>, <Capability.SELECT_MENUS: 'select_menus'>, <Capability.EDITING: 'editing'>, <Capability.MESSAGE_SEARCH: 'message_search'>, <Capability.EMOJI_REACTIONS: 'emoji_reactions'>}), max_message_length=2000, max_attachment_size=26214400, max_attachments=10, max_embeds=10, max_reactions=20)¶

field config: DiscordConfig [Optional]¶

property bot_user_id: str | None¶: Get the bot’s user ID (cached from connect/get_bot_info).

property bot_user_name: str | None¶: Get the bot’s username (cached from connect/get_bot_info).

class Config[source]¶

Bases: object

Pydantic config.

async connect() → None[source]¶

Connect to Discord using the configured bot token.

This creates a Discord client and logs in using the bot token. Note that for full event handling, you may need to run the client with client.start() instead.

Raises:: RuntimeError – If discord.py is not installed or token is missing.

async disconnect() → None[source]¶: Disconnect from Discord.

Fetch a user from Discord.

Accepts flexible inputs: - User ID as positional arg or id= - User object (returns as-is or refreshes) - name= to search by display name (limited) - handle= to search by username (limited)

Note: Discord API has limited user search capabilities. ID-based lookup is most reliable.

Parameters:

identifier – A User object or user ID string.
id – User ID.
name – Display name to search for.
email – Email (not supported by Discord).
handle – Username to search for.

Returns:

The user if found, None otherwise.

Fetch a channel from Discord.

Accepts flexible inputs: - Channel ID as positional arg or id= - Channel object (returns as-is or refreshes) - name= to search by channel name (requires guild context)

Note: Discord requires channel ID for direct lookup. Name-based search checks cache only.

Parameters:

identifier – A Channel object or channel ID string.
id – Channel ID.
name – Channel name to search for (cache only).

Returns:

The channel if found, None otherwise.

Fetch messages from a Discord channel, newest-first.

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.

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

Search for messages matching a query.

Note: Discord’s API doesn’t have a direct message search endpoint for bots. This implementation fetches recent messages and filters locally. For production use with large channels, consider using Discord’s user account search or a database-backed solution.

Parameters:

query – The search query string (case-insensitive substring match).
channel – Channel to search in (required for Discord).
limit – Maximum number of results.
**kwargs – Additional options: - from_user: Filter by author ID

Returns:

List of messages containing the query.

async send_message(channel: str | Channel, content: str, **kwargs: Any) → Message[source]¶

Send a message to a Discord channel.

Parameters:

channel – The channel to send to (ID string or Channel object).
content – The message content. kwargs: Discord thread, reply, embed, file, and text-to-speech options.

Returns:

The sent message.

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

Upload a file to a Discord channel.

Uses discord.File to send binary data directly.

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

Edit a Discord message.

Parameters:

message – The message to edit (ID string or Message object).
content – The new content.
channel – The channel containing the message (required if message is a string).
**kwargs – Additional options.

Returns:

The edited message.

async delete_message(message: str | Message, channel: str | Channel | None = None) → None[source]¶

Delete a Discord message.

Parameters:

message – The message to delete (ID string or Message object).
channel – The channel containing the message (required if message is a string).

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

Forward a message to another Discord channel.

Discord doesn’t have native forwarding, so this creates a new message with the original content and optional attribution using embeds.

Parameters:

message – The message to forward (DiscordMessage object or message ID).
to_channel – The destination channel (ID string or Channel object).
include_attribution – If True, include info about original source.
prefix – Optional text to prepend to the forwarded message.
**kwargs – Additional options (embed, reference, etc.).

Returns:

The forwarded message in the destination channel.

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

Set the bot’s presence on Discord.

Parameters:

status – Presence status (‘online’, ‘idle’, ‘dnd’, ‘invisible’).
status_text – Activity text (game name).
**kwargs – Additional options: - activity_type: Type of activity (‘playing’, ‘streaming’, ‘listening’, ‘watching’, ‘competing’). - url: Streaming URL (for streaming activity).

async get_presence(user: str | User) → Presence | None[source]¶

Get a user’s presence on Discord.

Note: This requires the GUILD_PRESENCES intent and caching. The user must share a guild with the bot.

Parameters:: user_id – The user ID.
Returns:: The user’s presence or None if not available.

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

Add a reaction to a message.

Parameters:

message – The message to react to (ID string or Message object).
emoji – The emoji (unicode or custom format <:name:id>).
channel – The channel containing the message (required if message is a string).

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

Remove a reaction from a message.

Parameters:

message – The message to remove reaction from (ID string or Message object).
emoji – The emoji to remove.
channel – The channel containing the message (required if message is a string).

mention_user(user: User) → str[source]¶

Format a user mention for Discord.

Parameters:: user – The user to mention.
Returns:: Discord user mention format (<@user_id>).

mention_channel(channel: Channel) → str[source]¶

Format a channel mention for Discord.

Parameters:: channel – The channel to mention.
Returns:: Discord channel mention format (<#channel_id>).

mention_here() → str[source]¶

Format an @here mention for Discord.

Returns:: Discord @here mention format.

mention_everyone() → str[source]¶

Format an @everyone mention for Discord.

Returns:: Discord @everyone mention format.

async get_bot_info() → User | None[source]¶

Get information about the connected bot user.

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

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

Create a DM channel with a user.

Note: Discord only supports 1:1 DMs from bots, so only the first user in the list is used.

Parameters:: users – List of users to create a DM with. Only first user is used.
Returns:: The DM channel ID, or None if creation failed.

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

Create a new guild text channel.

Note: This requires the bot to have MANAGE_CHANNELS permission in the target guild.

Parameters:

name – The channel name.
description – Optional channel topic/description.
public – If False, creates a private channel (not implemented yet).
guild_id – The guild ID to create the channel in. If not provided, uses config.guild_id.
**kwargs – Additional options: - category_id: Category ID to put the channel under.

Returns:

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

async fetch_channel_by_name(name: str, guild_id: str | None = None) → Channel | None[source]¶

Fetch a channel by name from a guild.

This method searches through the guild’s channels to find one matching the given name.

Parameters:

name – The channel name to search for (case-insensitive).
guild_id – The guild ID to search in. If not provided, uses config.guild_id.

Returns:

The channel if found, None otherwise.

async fetch_user_by_name(name: str, guild_id: str | None = None) → User | None[source]¶

Fetch a user by username from a guild.

This method searches through the guild’s members to find one matching the given username or display name.

Parameters:

name – The username or display name to search for (case-insensitive).
guild_id – The guild ID to search in. If not provided, uses config.guild_id.

Returns:

The user if found, None otherwise.

Fetch a guild (organization) from Discord.

Parameters:

identifier – A DiscordGuild object or guild ID string.
id – Guild ID (alternative to positional identifier).
name – Guild name to search for (case-insensitive).

Returns:

The guild if found, None otherwise.

async list_organizations() → list[Organization][source]¶

List all guilds the bot has access to.

Returns:: List of guilds.

Fetch a guild from Discord.

This is an alias for fetch_organization using Discord terminology.

async list_guilds() → list[Organization][source]¶

List all guilds the bot has access to.

This is an alias for list_organizations using Discord terminology.

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

Stream incoming messages in real-time using Discord gateway.

This creates a connection to the Discord gateway and yields messages as they arrive. Requires the bot to be in a guild and have MESSAGE_CONTENT intent enabled.

Parameters:

channel – Optional channel to filter messages (ID string or Channel object).
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:

DiscordMessage – Each message as it arrives.

Mock Discord backend for testing.

This module provides a mock implementation of the Discord backend for use in testing without requiring an actual Discord connection.

pydantic model chatom.discord.testing.MockDiscordBackend[source]¶

Bases: DiscordBackend

Mock Discord backend for testing.

This class provides a mock implementation of the Discord backend that doesn’t require an actual Discord connection. It stores all data in memory and provides methods to set up mock data for tests.

Example

>>> backend = MockDiscordBackend()
>>> backend.add_mock_user("123", "TestUser", "testuser")
>>> backend.add_mock_channel("456", "general")
>>> await backend.connect()
>>> user = await backend.fetch_user("123")
>>> assert user.name == "TestUser"

add_mock_user(id: str, name: str, handle: str, *, avatar_url: str = '', discriminator: str = '0', global_name: str | None = None, is_bot: bool = False, is_system: bool = False) → DiscordUser[source]¶

Add a mock user for testing.

Parameters:

id – The user ID.
name – The user’s display name.
handle – The username.
avatar_url – URL to user’s avatar.
discriminator – Legacy discriminator (e.g., “1234”).
global_name – Global display name.
is_bot – Whether the user is a bot.
is_system – Whether it’s a Discord system user.

Returns:

The created mock user.

add_mock_channel(id: str, name: str, channel_type: str = 'text', *, topic: str = '', guild_id: str = '', position: int = 0, nsfw: bool = False, discord_type: DiscordChannelType | None = None) → DiscordChannel[source]¶

Add a mock channel for testing.

Parameters:

id – The channel ID.
name – The channel name.
channel_type – The channel type (“text”, “voice”, “dm”, etc.).
topic – The channel topic.
guild_id – The guild/server ID.
position – Channel position.
nsfw – Whether channel is NSFW.
discord_type – The Discord channel type enum (overrides channel_type).

Returns:

The created mock channel.

add_mock_message(channel_id: str, user_id: str, content: str, *, message_id: str | None = None, timestamp: datetime | None = None, guild_id: str = '', edited: bool = False) → str[source]¶

Add a mock message for testing.

Parameters:

channel_id – The channel containing the message.
user_id – The author’s user ID.
content – The message content.
message_id – Optional message ID (auto-generated if not provided).
timestamp – Message timestamp (defaults to now).
guild_id – The guild ID.
edited – Whether the message was edited.

Returns:

The message ID.

set_mock_presence(user_id: str, status: PresenceStatus = PresenceStatus.ONLINE, *, activities: list[dict[str, Any]] | None = None, desktop_status: PresenceStatus = PresenceStatus.OFFLINE, mobile_status: PresenceStatus = PresenceStatus.OFFLINE, web_status: PresenceStatus = PresenceStatus.OFFLINE) → DiscordPresence[source]¶

Set mock presence for a user.

Parameters:

user_id – The user ID.
status – The presence status.
activities – List of activity data.
desktop_status – Desktop client status.
mobile_status – Mobile client status.
web_status – Web client status.

Returns:

The created mock presence.

property sent_messages: list[DiscordMessage]¶

Get all messages sent through this backend.

Returns:: List of sent messages.

property edited_messages: list[DiscordMessage]¶

Get all messages edited through this backend.

Returns:: List of edited messages.

property deleted_messages: list[dict[str, str]]¶

Get all message IDs deleted through this backend.

Returns:: List of deleted message IDs.

get_sent_messages() → list[DiscordMessage][source]¶

Get all messages sent through this backend.

Returns:: List of sent messages (copy).

get_edited_messages() → list[DiscordMessage][source]¶

Get all messages edited through this backend.

Returns:: List of edited messages.

get_deleted_messages() → list[dict[str, str]][source]¶

Get all messages deleted through this backend.

Returns:: List of deleted message info (channel_id, message_id).

get_reactions() → list[dict[str, str]][source]¶

Get all reactions added/removed through this backend.

Returns:: List of reaction info (channel_id, message_id, emoji, action).

get_presence_updates() → list[dict[str, Any]][source]¶

Get all presence updates made through this backend.

Returns:: List of presence update info.

clear() → None[source]¶: Clear all mock data and tracking stores.

async connect() → None[source]¶: Mock connect - always succeeds.

async disconnect() → None[source]¶: Mock disconnect.

Fetch a mock user by ID or other attributes.

Parameters:

identifier – A User object or user ID string.
id – User ID.
name – Display name to search for.
email – Unused (Discord has no email lookup).
handle – Username to search for.

Returns:

The mock user if found, None otherwise.

Fetch a mock channel by ID or name.

Parameters:

identifier – A Channel object or channel ID string.
id – Channel ID.
name – Channel name to search for.

Returns:

The mock channel if found, None otherwise.

Fetch mock messages from a channel.

Parameters:

channel – The channel to fetch from (ID string or Channel object).
limit – Maximum number of messages.
before – Fetch messages before this ID or message.
after – Fetch messages after this ID or message.

Returns:

List of mock messages.

async send_message(channel: str | Channel, content: str, **kwargs: Any) → Message[source]¶

Send a mock message.

Parameters:

channel – The channel to send to (ID string or Channel object).
content – The message content.
**kwargs – Additional options.

Returns:

The mock sent message.

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

Edit a mock message.

Parameters:

message – The message to edit (ID string or Message object).
content – The new content.
channel – The channel containing the message (required if message is a string).
**kwargs – Additional options.

Returns:

The mock edited message.

async delete_message(message: str | Message, channel: str | Channel | None = None) → None[source]¶

Delete a mock message.

Parameters:

message – The message to delete (ID string or Message object).
channel – The channel containing the message (required if message is a string).

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

Forward a mock message to another channel.

Parameters:

message – The message to forward (DiscordMessage object).
to_channel – The destination channel (ID string or Channel object).
include_attribution – If True, include info about original source.
prefix – Optional text to prepend to the forwarded message.
**kwargs – Additional options.

Returns:

The forwarded message in the destination channel.

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

Set mock presence.

Parameters:

status – The status string.
status_text – Activity text.
**kwargs – Additional options.

async get_presence(user: str | User) → Presence | None[source]¶

Get mock presence for a user.

Parameters:: user – The user ID string or User object.
Returns:: The mock presence if set, None otherwise.

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

Add a mock reaction.

Parameters:

message – The message to react to (ID string or Message object).
emoji – The emoji to add.
channel – The channel containing the message (required if message is a string).

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

Remove a mock reaction.

Parameters:

message – The message to remove reaction from (ID string or Message object).
emoji – The emoji to remove.
channel – The channel containing the message (required if message is a string).

property created_dms: list[list[str]]¶

Get all DMs created through this backend.

Returns:: List of user ID lists for each created DM.

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

Create a mock DM channel with the specified users.

Parameters:: users – List of users to include in the DM (ID strings or User objects).
Returns:: The DM channel ID.

Slack¶

Slack backend configuration.

This module provides configuration classes for the Slack backend.

pydantic model chatom.slack.config.SlackConfig[source]¶

Bases: BackendConfig

Configuration for Slack backend.

bot_token¶: Slack Bot User OAuth Token (xoxb-…) or path to file.

app_token¶: Slack App-Level Token (xapp-…) or path to file.

signing_secret¶: Slack signing secret for request verification.

team_id¶: The Slack workspace/team ID.

default_channel¶: Default channel ID for sending messages.

socket_mode¶: Whether to use Socket Mode for events.

ssl¶: Optional SSL context for connections.

Example

>>> config = SlackConfig(
...     bot_token="xoxb-your-token",
...     app_token="xapp-your-app-token",  # For Socket Mode
...     signing_secret="your-signing-secret",
... )
>>> backend = SlackBackend(config=config)

field bot_token: str | SecretStr = SecretStr('')¶: Slack Bot User OAuth Token (xoxb-…) or path to file.

field app_token: str | SecretStr = SecretStr('')¶: Slack App-Level Token (xapp-…) or path to file.

field signing_secret: str | SecretStr = SecretStr('')¶: Slack signing secret for request verification.

field team_id: str = ''¶: The Slack workspace/team ID.

field default_channel: str = ''¶: Default channel ID for sending messages.

field socket_mode: bool = False¶: Whether to use Socket Mode for events.

field ssl: SSLContext | None = None¶: Optional SSL context for connections.

property bot_token_str: str¶

Get the bot token as a plain string.

Returns:: The bot token string.

property app_token_str: str¶

Get the app token as a plain string.

Returns:: The app token string.

property signing_secret_str: str¶

Get the signing secret as a plain string.

Returns:: The signing secret string.

property has_socket_mode: bool¶

Check if Socket Mode is configured.

Returns True if both socket_mode is enabled and app_token is set.

Returns:: True if Socket Mode is configured, False otherwise.

Slack-specific User model.

This module provides the Slack-specific User class.

pydantic model chatom.slack.user.SlackUser[source]¶

Bases: User

Slack-specific user with additional Slack fields.

real_name¶: The user’s real name from their profile.

display_name¶: The user’s display name.

team_id¶: The ID of the user’s workspace/team.

is_admin¶: Whether the user is a workspace admin.

is_owner¶: Whether the user is a workspace owner.

is_restricted¶: Whether the user is a multi-channel guest.

is_ultra_restricted¶: Whether the user is a single-channel guest.

tz¶: The user’s timezone identifier.

tz_offset¶: The user’s timezone offset in seconds.

title¶: The user’s job title.

phone¶: The user’s phone number.

status_text¶: The user’s current status text.

status_emoji¶: The user’s current status emoji.

field real_name: str = ''¶: The user’s real name from their profile.

field display_name: str = ''¶: The user’s display name.

field team_id: str = ''¶: The ID of the user’s workspace/team.

field is_admin: bool = False¶: Whether the user is a workspace admin.

field is_owner: bool = False¶: Whether the user is a workspace owner.

field is_restricted: bool = False¶: Whether the user is a multi-channel guest.

field is_ultra_restricted: bool = False¶: Whether the user is a single-channel guest.

field tz: str = ''¶: The user’s timezone identifier.

field tz_offset: int = 0¶: The user’s timezone offset in seconds.

field title: str = ''¶: The user’s job title.

field phone: str = ''¶: The user’s phone number.

field status_text: str = ''¶: The user’s current status text.

field status_emoji: str = ''¶: The user’s current status emoji.

property mention_name: str¶

Get the best name to use when mentioning.

Returns:: The display name, real name, or handle.
Return type:: str

Slack-specific Channel model.

This module provides the Slack-specific Channel class.

pydantic model chatom.slack.channel.SlackChannel[source]¶

Bases: Channel

Slack-specific channel with additional Slack fields.

is_channel¶: Whether this is a public channel.

is_group¶: Whether this is a private channel.

is_im¶: Whether this is a direct message.

is_mpim¶: Whether this is a multi-party direct message.

is_private¶: Whether this is a private channel.

is_shared¶: Whether this channel is shared with other workspaces.

is_ext_shared¶: Whether this channel is shared externally.

is_org_shared¶: Whether this channel is shared org-wide.

creator¶: User ID of the channel creator.

purpose¶: Channel purpose text.

num_members¶: Number of members in the channel.

unread_count¶: Number of unread messages.

last_read¶: Timestamp of last read message.

latest¶: Timestamp of latest message.

field is_channel: bool = False¶: Whether this is a public channel.

field is_group: bool = False¶: Whether this is a private channel.

field is_im: bool = False¶: Whether this is a direct message.

field is_mpim: bool = False¶: Whether this is a multi-party direct message.

field is_shared: bool = False¶: Whether this channel is shared with other workspaces.

field is_ext_shared: bool = False¶: Whether this channel is shared externally.

field is_org_shared: bool = False¶: Whether this channel is shared org-wide.

field creator: User | None = None¶: The user who created the channel.

field purpose: str = ''¶: Channel purpose text.

field num_members: int | None = None¶: Number of members in the channel.

field unread_count: int = 0¶: Number of unread messages.

field last_read: str = ''¶: Timestamp of last read message.

field latest: str = ''¶: Timestamp of latest message.

property creator_id: str¶

Get the creator’s user ID.

Returns:: The creator’s ID, or empty string if not set.
Return type:: str

property slack_channel_type: ChannelType¶

Determine the channel type from Slack flags.

Returns:: The generic channel type.
Return type:: ChannelType

Slack-specific Message model.

This module provides the Slack-specific Message class.

pydantic model chatom.slack.message.SlackMessage[source]¶

Bases: Message

Slack-specific message with additional Slack fields.

Based on the Slack API message structure.

Note: Use the inherited channel field for the SlackChannel object, and channel_id for the channel ID string. The sender_id field is deprecated - use author or author_id instead.

team¶: The team/workspace ID.

subtype¶: The message subtype.

blocks¶: Slack Block Kit blocks.

text¶: The message text (may differ from content due to mentions).

latest_reply¶: Timestamp of latest reply.

reply_users¶: List of user IDs who replied.

is_locked¶: Whether the thread is locked.

subscribed¶: Whether user is subscribed to thread.

last_read¶: Timestamp of last read message in thread.

files¶: List of attached files.

upload¶: Whether this is a file upload message.

display_as_bot¶: Whether to display as bot.

edited¶: Edit information if message was edited.

field subtype: SlackMessageSubtype | None = None¶: The message subtype.

field blocks: list[dict[str, Any]] [Optional]¶: Slack Block Kit blocks.

field reply_count: int = 0¶: Number of replies in thread.

field latest_reply: str | None = None¶: Timestamp of latest reply.

field reply_users: list[SlackUser] [Optional]¶: List of user IDs who replied.

field is_locked: bool = False¶: Whether the thread is locked.

field subscribed: bool = False¶: Whether user is subscribed to thread.

field last_read: str | None = None¶: Timestamp of last read message in thread.

field files: list[dict[str, Any]] [Optional]¶: List of attached files.

field upload: bool = False¶: Whether this is a file upload message.

field display_as_bot: bool = False¶: Whether to display as bot.

field edited: dict[str, Any] | None = None¶: Edit information if message was edited.

property ts: str | None¶: Get the message timestamp (ts).

property thread_ts: str | None¶

Get the thread timestamp (thread_ts).

Returns the thread’s id if this message is in a thread, None otherwise.

property team: Organization | None¶: Get the team/workspace ID.

property sender_id: str | None¶: Get sender ID (deprecated - use author_id instead).

property bot_id: str | None¶

Get bot ID from author if author is a bot.

For bots, the bot_id is the same as the author’s id.

property app_id: str | None¶: Get app ID from author if available.

property is_thread_reply: bool¶: Check if this message is a reply in a thread.

property is_thread_parent: bool¶: Check if this message started a thread.

property is_bot_message: bool¶: Check if this message is from a bot.

property has_blocks: bool¶: Check if this message has Block Kit blocks.

property has_files: bool¶: Check if this message has attached files.

mentions_user(user: User) → bool[source]¶

Check if this message mentions a specific user.

Slack mentions are in the format <@USER_ID> in the message text.

Parameters:: user – The User to check for.
Returns:: True if the user is mentioned, False otherwise.

property permalink: str | None¶: Get the message permalink if channel_id and ts are available.

to_formatted() → FormattedMessage[source]¶

Convert this Slack message to a FormattedMessage.

Parses Slack mrkdwn formatting and converts to a FormattedMessage that can be rendered for other backends.

Returns:: The formatted message representation.
Return type:: FormattedMessage

classmethod from_formatted(formatted: FormattedMessage, backend: str = '', **kwargs: Any) → SlackMessage[source]¶

Create a SlackMessage from a FormattedMessage.

Renders the FormattedMessage in Slack mrkdwn format.

Parameters:

formatted – The FormattedMessage to convert.
backend – Target backend (ignored, always uses slack format).
**kwargs – Additional message attributes.

Returns:

A new SlackMessage instance.

Return type:

SlackMessage

classmethod from_api_response(data: dict[str, Any]) → SlackMessage[source]¶

Create a SlackMessage from a Slack API response.

Parameters:: data – The API response data.
Returns:: A SlackMessage instance.

class chatom.slack.message.SlackMessageSubtype(value)[source]¶

Bases: str, Enum

Slack message subtypes.

Based on Slack API message subtypes.

Slack-specific Presence model.

This module provides the Slack-specific Presence class.

pydantic model chatom.slack.presence.SlackPresence[source]¶

Bases: Presence

Slack-specific presence.

slack_presence¶: Slack-specific presence (active/away).

auto_away¶: Whether user was automatically marked away.

manual_away¶: Whether user manually set away.

connection_count¶: Number of active connections.

last_activity¶: Timestamp of last activity.

field slack_presence: SlackPresenceStatus = SlackPresenceStatus.AWAY¶: Slack-specific presence.

field auto_away: bool = False¶: Whether user was automatically marked away.

field manual_away: bool = False¶: Whether user manually set away.

field connection_count: int = 0¶: Number of active connections.

field last_activity: int | None = None¶: Timestamp of last activity.

property generic_status: PresenceStatus¶

Convert Slack presence to generic status.

Returns:: The generic presence status.
Return type:: PresenceStatus

class chatom.slack.presence.SlackPresenceStatus(value)[source]¶

Bases: str, Enum

Slack-specific presence statuses.

classmethod from_base(status: PresenceStatus) → SlackPresenceStatus[source]¶

Convert a base PresenceStatus to SlackPresenceStatus.

Parameters:: status – The base PresenceStatus to convert.
Returns:: The corresponding SlackPresenceStatus.

Example

>>> slack_status = SlackPresenceStatus.from_base(PresenceStatus.ONLINE)
>>> print(slack_status)  # SlackPresenceStatus.ACTIVE

Slack-specific mention utilities.

This module registers Slack-specific mention formatting.

chatom.slack.mention.mention_channel(channel: Channel) → str[source]¶

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

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

chatom.slack.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.slack.mention.mention_user(user: User) → str[source]¶

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

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

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

chatom.slack.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'

Slack backend implementation for chatom.

pydantic model chatom.slack.backend.SlackBackend[source]¶

Bases: BackendBase

Slack backend implementation.

This provides the backend interface for Slack using the slack_sdk library for API calls.

name¶: The backend identifier (‘slack’).

display_name¶: Human-readable name.

format¶: Slack uses its own mrkdwn format.

capabilities¶: Slack-specific capabilities.

config¶: Slack-specific configuration.

Example

>>> config = SlackConfig(bot_token="xoxb-your-token")
>>> backend = SlackBackend(config=config)
>>> await backend.connect()
>>> messages = await backend.fetch_messages("C123456")

name: ClassVar[str] = 'slack'¶

display_name: ClassVar[str] = 'Slack'¶

format: ClassVar[Format] = 'slack-markdown'¶

mention_pattern: ClassVar[Pattern | None] = re.compile('<@(U[A-Z0-9]+)>')¶

user_class¶: alias of SlackUser

channel_class¶: alias of SlackChannel

presence_class¶: alias of SlackPresence

field capabilities: BackendCapabilities | None = BackendCapabilities(capabilities=frozenset({<Capability.DELETING: 'deleting'>, <Capability.TYPING_INDICATORS: 'typing_indicators'>, <Capability.PINNING: 'pinning'>, <Capability.RICH_TEXT: 'rich_text'>, <Capability.USER_MENTIONS: 'user_mentions'>, <Capability.FORMS: 'forms'>, <Capability.THREADS: 'threads'>, <Capability.BUTTONS: 'buttons'>, <Capability.EMBEDS: 'embeds'>, <Capability.MARKDOWN: 'markdown'>, <Capability.PLAINTEXT: 'plaintext'>, <Capability.CUSTOM_EMOJI: 'custom_emoji'>, <Capability.CHANNEL_MENTIONS: 'channel_mentions'>, <Capability.CODE_BLOCKS: 'code_blocks'>, <Capability.FILES: 'files'>, <Capability.EVERYONE_MENTION: 'everyone_mention'>, <Capability.IMAGES: 'images'>, <Capability.PRESENCE: 'presence'>, <Capability.FORWARDING: 'forwarding'>, <Capability.TABLES: 'tables'>, <Capability.REPLIES: 'replies'>, <Capability.SELECT_MENUS: 'select_menus'>, <Capability.EDITING: 'editing'>, <Capability.MESSAGE_SEARCH: 'message_search'>, <Capability.EMOJI_REACTIONS: 'emoji_reactions'>}), max_message_length=40000, max_attachment_size=1073741824, max_attachments=10, max_embeds=20, max_reactions=23)¶

field config: SlackConfig [Optional]¶

property bot_user_id: str | None¶: Get the bot’s user ID (cached from connect/get_bot_info).

property bot_user_name: str | None¶: Get the bot’s username (cached from connect/get_bot_info).

class Config[source]¶: Bases: object

async connect() → None[source]¶

Connect to Slack using the configured credentials.

Initializes the Slack WebClient with the bot token from config.

Raises:

ImportError – If slack_sdk is not installed.
SlackApiError – If authentication fails.

async disconnect() → None[source]¶: Disconnect from Slack.

Fetch a user from Slack.

Accepts flexible inputs: - User ID as positional arg or id= - SlackUser object (returns as-is or refreshes) - name= to search by display name or real name - email= to search by email address - handle= to search by username

Parameters:

identifier – A SlackUser object or user ID string.
id – User ID.
name – Display name or real name to search for.
email – Email address to search for.
handle – Username to search for.

Returns:

The user if found, None otherwise.

Fetch a channel from Slack.

Accepts flexible inputs: - Channel ID as positional arg or id= - SlackChannel object (returns as-is or refreshes) - name= to search by channel name

Parameters:

identifier – A SlackChannel object or channel ID string.
id – Channel ID.
name – Channel name to search for.

Returns:

The channel if found, None otherwise.

Fetch messages from a Slack channel, newest-first.

Uses the conversations.history API with oldest/latest bounds, paging via cursor to cover the full range up to limit.

Parameters:

channel – The channel to fetch messages from (ID string or Channel object).
limit – Maximum number of messages to return.
before – Upper bound — Slack ts, Message, or datetime (latest).
after – Lower bound — Slack ts, Message, or datetime (oldest).

Returns:

List of messages, ordered newest-to-oldest.

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

Search for messages matching a query.

Uses the search.messages API. Requires search:read scope.

Parameters:

query – The search query string. Supports Slack search modifiers like “from:@user”, “in:#channel”, “has:link”, etc.
channel – Optional channel to limit search to (ID string or Channel object).
limit – Maximum number of results (1-100).
**kwargs – Additional options: - sort: “score” or “timestamp” (default: “score”) - sort_dir: “asc” or “desc” (default: “desc”)

Returns:

List of messages matching the query.

Example

>>> results = await backend.search_messages("important")
>>> results = await backend.search_messages("from:@john has:file")

async send_message(channel: str | Channel, content: str, **kwargs: Any) → SlackMessage[source]¶

Send a message to a Slack channel.

Uses the chat.postMessage API.

Parameters:

channel – The channel to send to (ID string or Channel object).
content – The message content. kwargs: Slack thread, reply, block, attachment, and unfurl options.

Returns:

The sent message.

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

Upload a file to a Slack channel.

Uses the files_upload_v2 API (Slack SDK >=3.19).

async download_attachment(attachment: Any, *, message: Message | None = None) → bytes[source]¶

Download a Slack attachment’s bytes.

Slack url_private links require the bot token as a bearer token; this override supplies it. Falls back to files.info to resolve the private URL when the attachment only carries a file ID.

async edit_message(message: str | Message, content: str, channel: str | Channel | None = None, **kwargs: Any) → SlackMessage[source]¶

Edit a Slack message.

Uses the chat.update API.

Parameters:

message – The message to edit (ts string or SlackMessage object).
content – The new content.
channel – The channel containing the message (required if message is a string).
**kwargs – Additional options.

Returns:

The edited message.

async delete_message(message: str | Message, channel: str | Channel | None = None) → None[source]¶

Delete a Slack message.

Uses the chat.delete API.

Parameters:

message – The message to delete (ts string or SlackMessage object).
channel – The channel containing the message (required if message is a string).

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

Forward a message to another Slack channel.

Slack doesn’t have native forwarding, so this creates a new message with the original content and optional attribution.

Parameters:

message – The message to forward (SlackMessage object or message ts).
to_channel – The destination channel (ID string or Channel object).
include_attribution – If True, include info about original source.
prefix – Optional text to prepend to the forwarded message.
**kwargs – Additional options (thread_id, blocks, etc.).

Returns:

The forwarded message in the destination channel.

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

Set user presence on Slack.

Uses the users.setPresence API for presence and users.profile.set for status text.

Parameters:

status – Presence status (‘auto’ or ‘away’).
status_text – Status text to display.
**kwargs – Additional options (status_emoji, status_expiration, etc.).

async get_presence(user: str | User) → SlackPresence | None[source]¶

Get a user’s presence on Slack.

Uses the users.getPresence API.

Parameters:: user – The user ID string or User object.
Returns:: The user’s presence.

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

Add a reaction to a message.

Uses the reactions.add API.

Parameters:

message – The message to react to (ts string or SlackMessage object).
emoji – The emoji name (without colons).
channel – The channel containing the message (required if message is a string).

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

Remove a reaction from a message.

Uses the reactions.remove API.

Parameters:

message – The message to remove reaction from (ts string or SlackMessage object).
emoji – The emoji name to remove.
channel – The channel containing the message (required if message is a string).

mention_user(user: User) → str[source]¶

Format a user mention for Slack.

Parameters:: user – The user to mention.
Returns:: Slack user mention format (<@user_id>).

mention_channel(channel: Channel) → str[source]¶

Format a channel mention for Slack.

Parameters:: channel – The channel to mention.
Returns:: Slack channel mention format (<#channel_id>).

mention_here() → str[source]¶

Format an @here mention for Slack.

Returns:: Slack @here mention format (<!here>).

mention_everyone() → str[source]¶

Format an @everyone mention for Slack.

Returns:: Slack @everyone mention format (<!everyone>).

mention_channel_all() → str[source]¶

Format an @channel mention for Slack.

Notifies all members of the current channel.

Returns:: Slack @channel mention format (<!channel>).

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

Create a DM/IM channel with the specified users.

Uses conversations.open API to create or retrieve a DM channel. For a single user, creates a 1:1 DM. For multiple users, creates a group DM (multi-party DM).

Parameters:: users – List of users to include in the DM (ID strings or User objects). Can be a single user or a list.
Returns:: The DM channel ID, or None if creation failed.

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

Create a new Slack channel.

Uses conversations.create API to create a public or private channel.

Parameters:

name – The channel name (will be lowercased and spaces replaced).
description – Optional channel description/purpose.
public – Whether the channel is public (default True). Private channels are created with is_private=True.
**kwargs – Additional options: - team_id: Workspace ID for Enterprise Grid.

Returns:

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

async get_bot_info() → User | None[source]¶

Get information about the connected bot user.

Returns:: The bot’s User object.

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

Stream incoming messages in real-time using Socket Mode.

This requires an app token (xapp-…) to be configured.

Parameters:

channel – Optional channel to filter messages (ID string or Channel object).
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.

Mock Slack backend for testing.

This module provides a mock implementation of the Slack backend for use in tests without requiring actual Slack API credentials.

pydantic model chatom.slack.testing.MockSlackBackend[source]¶

Bases: SlackBackend

Mock Slack backend for testing.

This backend simulates Slack API responses without making actual network calls. Useful for unit tests and development.

mock_users¶: Dictionary of mock users by ID.

mock_channels¶: Dictionary of mock channels by ID.

mock_messages¶: Dictionary of messages by channel_id.

mock_presence¶: Dictionary of presence by user_id.

sent_messages¶: List of all sent messages (for assertions).

deleted_messages¶: List of deleted message IDs.

reactions¶: Dictionary of reactions by (channel_id, message_id).

Example

>>> backend = MockSlackBackend()
>>> backend.add_mock_user(SlackUser(id="U123", name="alice"))
>>> backend.add_mock_channel(SlackChannel(id="C123", name="general"))
>>> await backend.connect()
>>> user = await backend.fetch_user("U123")
>>> assert user.name == "alice"

name: ClassVar[str] = 'mock_slack'¶

display_name: ClassVar[str] = 'Mock Slack'¶

add_mock_user(id: str, name: str, handle: str | None = None, *, display_name: str | None = None, avatar_url: str = '', is_bot: bool = False) → SlackUser[source]¶

Add a mock user to the backend.

Parameters:

id – The user ID.
name – The user’s display name.
handle – The username/handle.
display_name – Optional display name.
avatar_url – URL to user’s avatar.
is_bot – Whether the user is a bot.

Returns:

The created mock user.

add_mock_channel(id: str, name: str, *, topic: str = '', is_private: bool = False, is_archived: bool = False) → SlackChannel[source]¶

Add a mock channel to the backend.

Parameters:

id – The channel ID.
name – The channel name.
topic – The channel topic.
is_private – Whether the channel is private.
is_archived – Whether the channel is archived.

Returns:

The created mock channel.

add_mock_message(channel_id: str, user_id: str, content: str, *, message_id: str | None = None, timestamp: datetime | None = None) → str[source]¶

Add a mock message to a channel.

Parameters:

channel_id – The channel ID.
user_id – The sender’s user ID.
content – The message content.
message_id – Optional message ID (auto-generated if not provided).
timestamp – Optional timestamp.

Returns:

The message ID.

set_mock_presence(user_id: str, status: PresenceStatus = PresenceStatus.ONLINE, *, status_text: str = '') → SlackPresence[source]¶

Set mock presence for a user.

Parameters:

user_id – The user ID.
status – The presence status.
status_text – Optional status text.

Returns:

The created presence.

property sent_messages: list[SlackMessage]¶

Get all messages sent through this backend.

Returns:: List of sent messages.

property added_reactions: list[tuple]¶

Get all reactions added through this backend.

Returns:: List of (channel_id, message_id, emoji) tuples.

property removed_reactions: list[tuple]¶

Get all reactions removed through this backend.

Returns:: List of (channel_id, message_id, emoji) tuples.

property presence_changes: list[dict[str, Any]]¶

Get all presence changes made through this backend.

Returns:: List of presence changes.

property created_dms: list[list[str]]¶

Get all DMs created through this backend.

Returns:: List of user ID lists for each created DM.

property mock_users: dict[str, SlackUser]¶

Get all mock users.

Returns:: Dictionary of mock users by ID.

reset() → None[source]¶: Reset all mock data and tracking stores.

get_sent_messages() → list[SlackMessage][source]¶

Get all messages sent through this backend.

Returns:: List of sent messages (copy).

get_deleted_messages() → list[tuple][source]¶

Get all deleted message references.

Returns:: List of (channel_id, message_id) tuples.

get_reactions(channel_id: str, message_id: str) → list[str][source]¶

Get reactions for a message.

Parameters:

channel_id – The channel ID.
message_id – The message ID.

Returns:

List of emoji names.

clear() → None[source]¶: Clear all mock data.

async connect() → None[source]¶: Connect to the mock backend.

async disconnect() → None[source]¶: Disconnect from the mock backend.

Fetch a mock user by ID or other attributes.

Parameters:

identifier – A User object or user ID string.
id – User ID.
name – Display name to search for.
email – Email address to search for.
handle – Username/handle to search for.

Returns:

The user if found, None otherwise.

Fetch a mock channel by ID or name.

Parameters:

identifier – A Channel object or channel ID string.
id – Channel ID.
name – Channel name to search for.

Returns:

The channel if found, None otherwise.

Fetch mock messages from a channel.

Parameters:

channel – The channel to fetch from (ID string or Channel object).
limit – Maximum number of messages.
before – Fetch messages before this timestamp.
after – Fetch messages after this timestamp.

Returns:

List of messages.

async send_message(channel: str | Channel, content: str, **kwargs: Any) → SlackMessage[source]¶

Send a mock message.

Parameters:

channel – The channel to send to (ID string or Channel object).
content – The message content.
**kwargs – Additional options.

Returns:

The sent message.

async edit_message(message: str | Message, content: str, channel: str | Channel | None = None, **kwargs: Any) → SlackMessage[source]¶

Edit a mock message.

Parameters:

message – The message to edit (ts string or Message object).
content – The new content.
channel – The channel containing the message (required if message is a string).
**kwargs – Additional options.

Returns:

The edited message.

async delete_message(message: str | Message, channel: str | Channel | None = None) → None[source]¶

Delete a mock message.

Parameters:

message – The message to delete (ts string or Message object).
channel – The channel containing the message (required if message is a string).

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

Forward a mock message to another channel.

Parameters:

message – The message to forward (SlackMessage object).
to_channel – The destination channel (ID string or Channel object).
include_attribution – If True, include info about original source.
prefix – Optional text to prepend to the forwarded message.
**kwargs – Additional options.

Returns:

The forwarded message in the destination channel.

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

Set mock presence.

Parameters:

status – Presence status (‘auto’ or ‘away’).
status_text – Status text.
**kwargs – Additional options.

async get_presence(user: str | User) → SlackPresence | None[source]¶

Get mock presence for a user.

Parameters:: user – The user ID string or User object.
Returns:: The user’s presence.

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

Add a mock reaction.

Parameters:

message – The message to react to (ts string or Message object).
emoji – The emoji name.
channel – The channel containing the message (required if message is a string).

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

Remove a mock reaction.

Parameters:

message – The message to remove reaction from (ts string or Message object).
emoji – The emoji name.
channel – The channel containing the message (required if message is a string).

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

Create a mock DM channel with the specified users.

Parameters:: users – List of users to include in the DM (ID strings or User objects).
Returns:: The DM channel ID.

Symphony¶

Symphony backend configuration.

This module provides the SymphonyConfig class for configuring the Symphony backend with the Symphony BDK.

pydantic model chatom.symphony.config.SymphonyConfig[source]¶

Bases: BackendConfig

Configuration for the Symphony backend.

This class holds all Symphony-specific configuration needed for the Symphony BDK, including pod URL, bot credentials, and optional application settings.

host¶: The Symphony pod hostname (e.g., “mycompany.symphony.com”).

port¶: The Symphony pod port (default: 443).

scheme¶: The URL scheme (default: “https”).

context¶: Optional context path (default: “”).

bot_username¶: The bot’s username.

bot_private_key_path¶: Path to the bot’s RSA private key file.

bot_private_key_content¶: Direct content of the RSA private key.

bot_certificate_path¶: Path to the bot’s certificate (for cert auth).

bot_certificate_content¶: Direct content of the certificate (PEM format). If provided without bot_certificate_path, a temp file is created automatically since Symphony BDK requires a file path.

bot_certificate_password¶: Password for the certificate.

app_id¶: Optional extension app ID.

app_private_key_path¶: Path to the app’s RSA private key.

agent_host¶: Optional separate agent host (if different from pod).

agent_port¶: Optional agent port.

session_auth_host¶: Optional session auth host (if different from pod).

session_auth_port¶: Optional session auth port.

key_manager_host¶: Optional key manager host.

key_manager_port¶: Optional key manager port.

trust_store_path¶: Path to custom trust store.

proxy_host¶: Optional proxy host.

proxy_port¶: Optional proxy port.

proxy_username¶: Optional proxy username.

proxy_password¶: Optional proxy password.

Example

>>> # Using RSA key file
>>> config = SymphonyConfig(
...     host="mycompany.symphony.com",
...     bot_username="my-bot",
...     bot_private_key_path="/path/to/private-key.pem",
... )
>>> backend = SymphonyBackend(config=config)

>>> # Using certificate content (temp file created automatically)
>>> config = SymphonyConfig(
...     host="mycompany.symphony.com",
...     bot_username="my-bot",
...     bot_certificate_content=SecretStr(cert_pem_string),
... )

field host: str = ''¶

field port: int = 443¶

field scheme: str = 'https'¶

field context: str = ''¶

field bot_username: str = ''¶

field bot_private_key_path: str | None = None¶

field bot_private_key_content: str | SecretStr | None = None¶

field bot_certificate_path: str | None = None¶

field bot_certificate_content: str | SecretStr | None = None¶: Direct content of the certificate PEM. If provided without bot_certificate_path, a temp file will be created automatically.

field bot_certificate_password: str | SecretStr | None = None¶

field app_id: str | None = None¶

field app_private_key_path: str | None = None¶

field pod_host: str | None = None¶

field pod_port: int | None = None¶

field agent_host: str | None = None¶

field agent_port: int | None = None¶

field session_auth_host: str | None = None¶

field session_auth_port: int | None = None¶

field key_manager_host: str | None = None¶

field key_manager_port: int | None = None¶

field trust_store_path: str | None = None¶

field proxy_host: str | None = None¶

field proxy_port: int | None = None¶

field proxy_username: str | None = None¶

field proxy_password: SecretStr | None = None¶

field timeout: int = 30¶: Request timeout in seconds

field error_room: str | None = None¶: A room to direct error messages to, if a message fails to be sent.

field inform_client: bool = False¶: Whether to inform the intended recipient of a failed message.

field max_attempts: int = 10¶: Max attempts for datafeed and message post requests before raising exception.

field initial_interval_ms: int = 500¶: Initial interval to wait between attempts, in milliseconds.

field multiplier: float = 2.0¶: Multiplier between attempt delays for exponential backoff.

field max_interval_ms: int = 300000¶: Maximum delay between retry attempts, in milliseconds.

field datafeed_version: str = 'v2'¶: Version of datafeed to use (‘v1’ or ‘v2’).

field ssl_trust_store_path: str | None = None¶: Path to a custom CA certificate bundle file.

field ssl_verify: bool = True¶: Whether to verify SSL certificates.

field message_create_url: str | None = None¶: Custom URL for message creation. Template: https://{host}/agent/v4/stream/{sid}/message/create

field datafeed_create_url: str | None = None¶: Custom URL for datafeed creation. Template: https://{host}/agent/v5/datafeeds

field datafeed_delete_url: str | None = None¶: Custom URL for datafeed deletion. Template: https://{host}/agent/v5/datafeeds/{datafeed_id}

field datafeed_read_url: str | None = None¶: Custom URL for datafeed reading. Template: https://{host}/agent/v5/datafeeds/{datafeed_id}/read

field room_search_url: str | None = None¶: Custom URL for room search. Template: https://{host}/pod/v3/room/search

field room_info_url: str | None = None¶: Custom URL for room info. Template: https://{host}/pod/v3/room/{room_id}/info

field im_create_url: str | None = None¶: Custom URL for IM creation. Template: https://{host}/pod/v1/im/create

field room_members_url: str | None = None¶: Custom URL for room members. Template: https://{host}/pod/v2/room/{room_id}/membership/list

field presence_url: str | None = None¶: Custom URL for user presence. Template: https://{host}/pod/v2/user/presence

field user_detail_url: str | None = None¶: Custom URL for user detail. Template: https://{host}/pod/v2/admin/user/{uid}

field user_search_url: str | None = None¶: Custom URL for user search. Template: https://{host}/pod/v3/users

field user_lookup_url: str | None = None¶: Custom URL for user lookup by email/username. Template: https://{host}/pod/v3/users

property bot_private_key_str: str | None¶: Get the bot private key content as string.

property bot_certificate_content_str: str | None¶: Get the certificate content as string.

property bot_certificate_password_str: str | None¶: Get the certificate password as string.

property proxy_password_str: str | None¶: Get the proxy password as string.

property has_rsa_auth: bool¶: Check if RSA authentication is configured.

property has_cert_auth: bool¶: Check if certificate authentication is configured.

property is_using_temp_cert: bool¶: Check if a temporary certificate file was created.

cleanup_temp_cert() → None[source]¶

Manually cleanup temporary certificate file if created.

This is called automatically on process exit, but can be called manually for explicit cleanup.

property pod_url: str¶: Build the pod URL.

to_bdk_config() → dict[str, Any][source]¶

Convert to Symphony BDK configuration format.

Returns:: Dictionary suitable for passing to SymphonyBdk.

get_bdk_config()[source]¶

Build a BdkConfig from chatom config fields.

Returns:: A BdkConfig instance for use with symphony-bdk-python.

class chatom.symphony.config.SymphonyRoomMapper(stream_service=None, backend: SymphonyBackend | None = None)[source]¶

Bases: object

Thread-safe mapper for Symphony room names and IDs.

This class maintains a cache of room name to ID mappings and vice versa, using the stream service to resolve unknown rooms.

set_stream_service(stream_service)[source]¶: Set the stream service for room resolution.

set_backend(backend: SymphonyBackend)[source]¶: Set the chatom backend for room resolution.

get_room_id(room_name: str) → str | None[source]¶

Get the room ID for a given room name.

Parameters:: room_name – The display name of the room.
Returns:: The room’s stream ID, or None if not found.

async get_room_id_async(room_name: str) → str | None[source]¶

Get the room ID for a given room name, using async calls if needed.

Parameters:: room_name – The display name of the room.
Returns:: The room’s stream ID, or None if not found.

get_room_name(room_id: str) → str | None[source]¶

Get the room name for a given room ID.

Parameters:: room_id – The room’s stream ID.
Returns:: The room’s display name, or None if not found.

async get_room_name_async(room_id: str) → str | None[source]¶

Get the room name for a given room ID, using async calls if needed.

Parameters:: room_id – The room’s stream ID.
Returns:: The room’s display name, or None if not found.

set_im_id(user_identifier: str, stream_id: str)[source]¶

Parameters:

user_identifier – The user’s display name or user ID.
stream_id – The IM stream ID.

register_room(room_name: str, room_id: str)[source]¶

Manually register a room name to ID mapping.

Parameters:

room_name – The display name of the room.
room_id – The room’s stream ID.

Symphony-specific User model.

This module provides the Symphony-specific User class.

pydantic model chatom.symphony.user.SymphonyUser[source]¶

Bases: User

Symphony-specific user with additional Symphony fields.

first_name¶: The user’s first name.

last_name¶: The user’s last name.

display_name¶: The user’s display name.

company¶: The user’s company name.

department¶: The user’s department.

title¶: The user’s job title.

location¶: The user’s location.

work_phone¶: The user’s work phone number.

mobile_phone¶: The user’s mobile phone number.

account_type¶: The type of Symphony account.

roles¶: List of user roles.

field first_name: str = ''¶: The user’s first name.

field last_name: str = ''¶: The user’s last name.

field display_name: str = ''¶: The user’s display name.

field company: str = ''¶: The user’s company name.

field department: str = ''¶: The user’s department.

field title: str = ''¶: The user’s job title.

field location: str = ''¶: The user’s location.

field work_phone: str = ''¶: The user’s work phone number.

field mobile_phone: str = ''¶: The user’s mobile phone number.

field account_type: str = ''¶: The type of Symphony account.

field roles: list[str] [Optional]¶: List of user roles.

property full_name: str¶

Get the user’s full name.

Returns:: First and last name combined.
Return type:: str

property mention_name: str¶

Get the best name to use when mentioning.

Returns:: The display name or full name.
Return type:: str

Symphony-specific Channel model.

This module provides the Symphony-specific Channel class. In Symphony, channels are called “streams” or “rooms”.

pydantic model chatom.symphony.channel.SymphonyChannel[source]¶

Bases: Channel

Symphony-specific channel (stream) with additional Symphony fields.

stream_type¶: The type of Symphony stream.

external¶: Whether the stream includes external users.

cross_pod¶: Whether the stream is cross-pod.

active¶: Whether the stream is active.

read_only¶: Whether the stream is read-only.

public¶: Whether the stream is public.

creation_date¶: When the stream was created.

last_message_date¶: When the last message was sent.

field stream_type: SymphonyStreamType = SymphonyStreamType.ROOM¶: The type of Symphony stream.

field external: bool = False¶: Whether the stream includes external users.

field cross_pod: bool = False¶: Whether the stream is cross-pod.

field active: bool = True¶: Whether the stream is active.

field read_only: bool = False¶: Whether the stream is read-only.

field public: bool = False¶: Whether the stream is public.

field creation_date: datetime | None = None¶: When the stream was created.

field last_message_date: datetime | None = None¶: When the last message was sent.

property generic_channel_type: ChannelType¶

Convert Symphony stream type to generic channel type.

Returns:: The generic channel type.
Return type:: ChannelType

property stream_id: str¶

Alias for id - Symphony uses stream_id terminology.

Returns:: The stream/channel ID.
Return type:: str

chatom.symphony.channel.SymphonyRoom¶: alias of SymphonyChannel

class chatom.symphony.channel.SymphonyStreamType(value)[source]¶

Bases: str, Enum

Symphony stream types.

IM = 'IM'¶

1 chat).

Type:: Instant message (1

MIM = 'MIM'¶: Multi-party instant message.

ROOM = 'ROOM'¶: Chat room.

POST = 'POST'¶: Wall post.

Symphony-specific Message model.

This module provides the Symphony-specific Message class.

pydantic model chatom.symphony.message.SymphonyMessage[source]¶

Bases: Message

Symphony-specific message with additional Symphony fields.

Based on the Symphony REST API message structure.

message_ml¶: The MessageML content.

presentation_ml¶: The PresentationML rendered content.

entity_data¶: Entity data for structured objects.

data¶: The JSON data associated with the message.

shared_message¶: Shared/forwarded message info.

ingestion_date¶: When the message was ingested.

diagnostic¶: Diagnostic information.

sid¶: Session ID.

original_format¶: The original message format.

is_chime¶: Whether this is a chime message.

is_copy_disabled¶: Whether copying is disabled.

attachments_metadata¶: Metadata about attachments.

hashtags¶: List of hashtags in the message.

cashtags¶: List of cashtags in the message.

mentions¶: List of user mentions in the message.

field message_ml: str | None = None¶: The MessageML content.

field presentation_ml: str | None = None¶: The PresentationML rendered content.

field entity_data: dict[str, Any] [Optional]¶: Entity data for structured objects.

field data: str | None = None¶: The JSON data associated with the message.

field shared_message: dict[str, Any] | None = None¶: Shared/forwarded message info.

field ingestion_date: datetime | None = None¶: When the message was ingested.

field diagnostic: str | None = None¶: Diagnostic information.

field sid: str | None = None¶: Session ID.

field original_format: SymphonyMessageFormat = SymphonyMessageFormat.MESSAGEML¶: The original message format.

field is_chime: bool = False¶: Whether this is a chime message.

field is_copy_disabled: bool = False¶: Whether copying is disabled.

field attachments_metadata: list[dict[str, Any]] [Optional]¶: Metadata about attachments.

field hashtags: list[str] [Optional]¶: List of hashtags in the message.

field cashtags: list[str] [Optional]¶: List of cashtags in the message.

property is_shared_message: bool¶: Check if this is a shared/forwarded message.

property message_id: str¶

Get the message ID (same as id).

Returns the message’s id for backward compatibility.

property stream_id: str¶

Get the stream ID (channel.id).

Returns the channel’s id, or empty string if no channel.

property has_entity_data: bool¶: Check if this message has entity data.

property has_hashtags: bool¶: Check if this message contains hashtags.

property has_cashtags: bool¶: Check if this message contains cashtags.

property has_mentions: bool¶: Check if this message contains user mentions.

mentions_user(user: User) → bool[source]¶

Check if this message mentions a specific user.

Parameters:: user – The User to check for.
Returns:: True if the user is mentioned in this message.

static extract_mentions_from_data(data: str | None) → list[int][source]¶

Extract user IDs from Symphony data field (JSON entity data).

Symphony encodes mentions in the data field as JSON with entity references like {“mention0”: {“type”: “com.symphony.user.mention”, “id”: [{“value”: “123”}]}}.

Parameters:: data – The JSON data string from the message.
Returns:: List of user IDs (as integers) mentioned in the message.

property rendered_content: str¶: Get the rendered content, preferring PresentationML.

classmethod from_api_response(data: dict[str, Any]) → SymphonyMessage[source]¶

Create a SymphonyMessage from an API response.

Parameters:: data – The API response data.
Returns:: A SymphonyMessage instance.

to_formatted() → FormattedMessage[source]¶

Convert this Symphony message to a FormattedMessage.

Parses Symphony MessageML/PresentationML formatting and converts to a FormattedMessage that can be rendered for other backends.

Returns:: The formatted message representation.
Return type:: FormattedMessage

classmethod from_formatted(formatted: FormattedMessage, backend: str = '', **kwargs: Any) → SymphonyMessage[source]¶

Create a SymphonyMessage from a FormattedMessage.

Renders the FormattedMessage in Symphony MessageML format.

Parameters:

formatted – The FormattedMessage to convert.
backend – Target backend (ignored, always uses symphony format).
**kwargs – Additional message attributes.

Returns:

A new SymphonyMessage instance.

Return type:

SymphonyMessage

class chatom.symphony.message.SymphonyMessageFormat(value)[source]¶

Bases: str, Enum

Symphony message format types.

Symphony-specific Presence model.

This module provides the Symphony-specific Presence class.

pydantic model chatom.symphony.presence.SymphonyPresence[source]¶

Bases: Presence

Symphony-specific presence.

symphony_status¶: Symphony-specific status enum.

category¶: Presence category.

timestamp¶: When the presence was last updated.

field symphony_status: SymphonyPresenceStatus = SymphonyPresenceStatus.OFFLINE¶: Symphony-specific status.

field category: str = ''¶: Presence category.

field timestamp: int | None = None¶: When the presence was last updated (epoch ms).

property generic_status: PresenceStatus¶

Convert Symphony status to generic status.

Returns:: The generic presence status.
Return type:: PresenceStatus

class chatom.symphony.presence.SymphonyPresenceStatus(value)[source]¶

Bases: str, Enum

Symphony-specific presence statuses.

Symphony-specific mention utilities.

This module registers Symphony-specific mention formatting using MessageML.

chatom.symphony.mention.mention_user(user: User) → str[source]¶

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

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

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

chatom.symphony.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.symphony.mention.mention_user_by_email(email: str) → str[source]¶

Generate a Symphony mention by email address.

Parameters:: email – The email address to mention.
Returns:: The Symphony MessageML mention tag.
Return type:: str

chatom.symphony.mention.mention_user_by_uid(uid: str) → str[source]¶

Generate a Symphony mention by user ID.

Parameters:: uid – The user ID to mention.
Returns:: The Symphony MessageML mention tag.
Return type:: str

Symphony backend implementation for chatom.

This module provides the Symphony backend using the Symphony BDK (Bot Development Kit).

pydantic model chatom.symphony.backend.SymphonyBackend[source]¶

Bases: BackendBase

Symphony backend implementation using Symphony BDK.

This provides the backend interface for Symphony using the official Symphony Bot Development Kit (BDK).

name¶: The backend identifier (‘symphony’).

display_name¶: Human-readable name.

format¶: Symphony uses MessageML format.

capabilities¶: Symphony-specific capabilities.

config¶: Symphony-specific configuration.

Example

>>> from chatom.symphony import SymphonyBackend, SymphonyConfig
>>> config = SymphonyConfig(
...     host="mycompany.symphony.com",
...     bot_username="my-bot",
...     bot_private_key_path="/path/to/private-key.pem",
... )
>>> backend = SymphonyBackend(config=config)
>>> await backend.connect()

name: ClassVar[str] = 'symphony'¶

display_name: ClassVar[str] = 'Symphony'¶

format: ClassVar[Format] = 'symphony-messageml'¶

mention_pattern: ClassVar[Pattern | None] = re.compile('<mention\\s+uid="(\\d+)"\\s*/>')¶

user_class¶: alias of SymphonyUser

channel_class¶: alias of SymphonyChannel

presence_class¶: alias of SymphonyPresence

field capabilities: BackendCapabilities | None = BackendCapabilities(capabilities=frozenset({<Capability.DELETING: 'deleting'>, <Capability.CODE_BLOCKS: 'code_blocks'>, <Capability.FILES: 'files'>, <Capability.IMAGES: 'images'>, <Capability.RICH_TEXT: 'rich_text'>, <Capability.USER_MENTIONS: 'user_mentions'>, <Capability.EDITING: 'editing'>, <Capability.FORWARDING: 'forwarding'>, <Capability.TABLES: 'tables'>, <Capability.HTML: 'html'>, <Capability.PLAINTEXT: 'plaintext'>, <Capability.FORMS: 'forms'>, <Capability.MESSAGE_SEARCH: 'message_search'>, <Capability.EMOJI_REACTIONS: 'emoji_reactions'>}), max_message_length=40000, max_attachment_size=26214400, max_attachments=10, max_embeds=10, max_reactions=20)¶

field config: SymphonyConfig [Optional]¶

property bot_user_id: str | None¶: Get the bot’s user ID as a string (cached from connect).

property bot_user_name: str | None¶: Get the bot’s username (from config or cached from connect).

class Config[source]¶

Bases: object

Pydantic config.

normalize_channel_id(channel_id: str) → str[source]¶

Canonicalize a Symphony stream id for equality comparison.

Symphony returns stream ids in both standard and URL-safe base64, with or without = padding, depending on which API produced them (for example the datafeed vs. stream lookups). Decode whichever form was given and re-encode to standard padded base64 so the same stream compares equal regardless of source. Non-base64 values are returned unchanged.

Parameters:: channel_id – The stream id to canonicalize.
Returns:: The canonical stream id.

async connect() → None[source]¶

Connect to Symphony using the BDK.

This initializes the Symphony BDK and authenticates the bot.

Raises:: RuntimeError – If symphony-bdk is not installed or connection fails.

async disconnect() → None[source]¶: Disconnect from Symphony.

Fetch a user from Symphony.

Accepts flexible inputs: - User ID as positional arg or id= - User object (returns as-is or refreshes) - name= to search by display name - email= to search by email address - handle= to search by username

Parameters:

identifier – A User object or user ID string.
id – User ID.
name – Display name to search for.
email – Email address to search for.
handle – Username to search for.

Returns:

The user if found, None otherwise.

Fetch a channel (stream/room) from Symphony.

Accepts flexible inputs: - Stream ID as positional arg or id= - Channel object (returns as-is or refreshes) - name= to search by room name

Parameters:

identifier – A Channel object or stream ID string.
id – Stream ID.
name – Room name to search for.

Returns:

The channel if found, None otherwise.

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

Fetch members of a Symphony room.

Parameters:

identifier – A Channel object or stream ID string.
id – Stream/room ID.
name – Room name (will be resolved via search).

Returns:

List of users who are members of the room.

Fetch messages from a Symphony stream, newest-first.

Returns up to limit messages ordered newest-to-oldest. after and before bound the range and accept a millisecond-epoch id string, a Message (its created_at is used), or a datetime. The stream is paged in full over the bounded range (Symphony’s API returns oldest-first from a since timestamp; we widen/skip to cover everything) before trimming to the most recent limit.

Parameters:

channel – The stream to fetch from (ID string or Channel object).
limit – Maximum number of messages to return.
before – Upper bound — epoch-ms id, Message, or datetime.
after – Lower bound — epoch-ms id, Message, or datetime.

Returns:

List of messages, ordered newest-to-oldest.

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

Search for messages matching a query.

Uses Symphony’s message search API.

Parameters:

query – The search query string.
channel – Optional stream to limit search to (ID string or Channel object).
limit – Maximum number of results.
**kwargs – Additional options: - from_user: Filter by sender user ID - hashtags: List of hashtags to filter by - cashtags: List of cashtags to filter by

Returns:

List of messages matching the query.

async send_message(channel: str | Channel, content: str, **kwargs: Any) → Message[source]¶

Send a message to a Symphony stream.

Parameters:

channel – The stream to send to (ID string or Channel object).
content – The message content (MessageML).
kwargs – Symphony template data and attachments; thread and reply options are ignored.

Returns:

The sent message.

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

Upload a file to a Symphony stream.

Sends the file as an attachment via the Symphony BDK message API. The binary data is written to a temporary file which is passed to the BDK’s attachment parameter.

async download_attachment(attachment: Any, *, message: Message | None = None) → bytes[source]¶

Download a Symphony attachment’s bytes.

Symphony attachments have no public URL; they are fetched with the BDK get_attachment(stream_id, message_id, attachment_id) call, which returns the content base64-encoded. The required stream and message IDs come from the attachment metadata (populated when the message was received) or from the supplied message.

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

Edit a Symphony message.

Note: Symphony has limited support for message editing via the update_message API.

Parameters:

message – The message to edit (ID string or Message object).
content – The new content (MessageML).
channel – The stream containing the message (required if message is a string).
**kwargs – Additional options.

Returns:

The edited message.

async delete_message(message: str | Message, channel: str | Channel | None = None) → None[source]¶

Delete (suppress) a Symphony message.

Parameters:

message – The message to delete (ID string or Message object).
channel – The stream containing the message (not used for Symphony).

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

Forward a message to another Symphony room/stream.

Symphony doesn’t have native forwarding, so this creates a new message with the original content and optional attribution in MessageML format.

Parameters:

message – The message to forward (SymphonyMessage object or message ID).
to_channel – The destination stream (ID string or Channel object).
include_attribution – If True, include info about original source.
prefix – Optional text to prepend to the forwarded message.
**kwargs – Additional options (data for entity data, etc.).

Returns:

The forwarded message in the destination stream.

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

Set user presence on Symphony.

Parameters:

status – Presence status (available, busy, away, etc.).
status_text – Not supported by Symphony.
**kwargs – Additional options: - soft: If True, respect current activity state.

async get_presence(user: str | User) → Presence | None[source]¶

Get a user’s presence on Symphony.

Parameters:: user – The user ID string or User object.
Returns:: The user’s presence.

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

Add a reaction to a message.

Note: Symphony doesn’t support reactions in the same way. This raises NotImplementedError.

Parameters:

message – The message to react to (ID string or Message object).
emoji – The emoji.
channel – The stream containing the message (not used).

Raises:

NotImplementedError – Symphony doesn’t support emoji reactions.

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

Remove a reaction from a message.

Note: Symphony doesn’t support reactions.

Parameters:

message – The message to remove reaction from (ID string or Message object).
emoji – The emoji to remove.
channel – The stream containing the message (not used).

Raises:

NotImplementedError – Symphony doesn’t support emoji reactions.

mention_user(user: User) → str[source]¶

Format a user mention for Symphony.

Parameters:: user – The user to mention.
Returns:: Symphony user mention format (<mention uid=”…”/>).

mention_channel(channel: Channel) → str[source]¶

Format a channel mention for Symphony.

Symphony doesn’t have channel mentions in the same way. Returns the channel name.

Parameters:: channel – The channel to mention.
Returns:: The channel name.

mention_here() → str[source]¶

Format an @here mention for Symphony.

Symphony doesn’t have an @here equivalent. Returns an empty string (no-op).

Returns:: Empty string (Symphony doesn’t support @here).

mention_everyone() → str[source]¶

Format an @everyone mention for Symphony.

Uses Symphony’s mention all users tag.

Returns:: Symphony MessageML mention all tag.

mention_channel_all() → str[source]¶

Format an @channel mention for Symphony.

Symphony uses the same tag for all broadcast mentions.

Returns:: Symphony MessageML mention all tag.

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

Create a direct message (IM) or multi-party IM.

Parameters:: users – List of users to include in the DM (ID strings or User objects). The calling bot is implicitly included in the conversation.
Returns:: The stream ID of the created DM.

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

Create an instant message (IM) or multi-party IM.

This is an alias for create_dm, using Symphony’s terminology.

Parameters:: users – List of users to include in the IM (ID strings or User objects).
Returns:: The stream ID of the created IM.

async create_channel(name: str, description: str = '', public: bool = False, **kwargs: Any) → str | None[source]¶

Create a room (chat room).

Parameters:

name – The room name.
description – The room description.
public – Whether the room is public.
**kwargs – Additional options: - read_only: Whether the room is read-only.

Returns:

The stream ID of the created room.

async create_room(name: str, description: str = '', public: bool = False, **kwargs: Any) → str | None[source]¶

Create a room (chat room).

This is an alias for create_channel, using Symphony’s terminology.

Parameters:

name – The room name.
description – The room description.
public – Whether the room is public.
**kwargs – Additional options: - read_only: Whether the room is read-only.

Returns:

The stream ID of the created room.

async get_bot_info() → User | None[source]¶

Get information about the connected bot user.

Returns:: The bot’s User object.

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

Stream incoming messages in real-time using Symphony datafeed.

This async generator yields Message objects as they arrive.

Parameters:

channel – Optional stream to filter messages (ID string or Channel object).
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.

Mock Symphony backend for testing.

This module provides a mock implementation of the Symphony backend that doesn’t require actual Symphony servers.

pydantic model chatom.symphony.testing.MockSymphonyBackend[source]¶

Bases: BackendBase

Mock Symphony backend for testing.

This provides a testing-friendly implementation that simulates Symphony operations without requiring actual servers.

mock_users¶: Dictionary of mock users by ID.

mock_streams¶: Dictionary of mock streams (channels) by ID.

mock_messages¶: Dictionary of messages by stream ID.

mock_presence¶: Dictionary of presence by user ID.

sent_messages¶: List of sent messages for verification.

edited_messages¶: List of edited message IDs.

deleted_messages¶: List of deleted message IDs.

presence_changes¶: List of presence changes for verification.

Example

>>> from chatom.symphony import MockSymphonyBackend, SymphonyConfig
>>> backend = MockSymphonyBackend()
>>> await backend.connect()
>>> # Add mock data
>>> backend.add_mock_user(123456789, "Test User", "testuser")
>>> backend.add_mock_stream("stream123", "Test Room")
>>> # Verify operations
>>> await backend.send_message("stream123", "<messageML>Hello</messageML>")
>>> assert len(backend.sent_messages) == 1

name: ClassVar[str] = 'symphony'¶

display_name: ClassVar[str] = 'Symphony (Mock)'¶

format: ClassVar[Format] = 'symphony-messageml'¶

field capabilities: BackendCapabilities | None = BackendCapabilities(capabilities=frozenset({<Capability.DELETING: 'deleting'>, <Capability.CODE_BLOCKS: 'code_blocks'>, <Capability.FILES: 'files'>, <Capability.IMAGES: 'images'>, <Capability.RICH_TEXT: 'rich_text'>, <Capability.USER_MENTIONS: 'user_mentions'>, <Capability.EDITING: 'editing'>, <Capability.FORWARDING: 'forwarding'>, <Capability.TABLES: 'tables'>, <Capability.HTML: 'html'>, <Capability.PLAINTEXT: 'plaintext'>, <Capability.FORMS: 'forms'>, <Capability.MESSAGE_SEARCH: 'message_search'>, <Capability.EMOJI_REACTIONS: 'emoji_reactions'>}), max_message_length=40000, max_attachment_size=26214400, max_attachments=10, max_embeds=10, max_reactions=20)¶

field config: SymphonyConfig [Optional]¶

field mock_users: dict[str, dict[str, Any]] [Optional]¶

field mock_streams: dict[str, dict[str, Any]] [Optional]¶

field mock_messages: dict[str, list[dict[str, Any]]] [Optional]¶

field mock_presence: dict[str, SymphonyPresenceStatus] [Optional]¶

field sent_messages: list[dict[str, Any]] [Optional]¶

field edited_messages: list[dict[str, Any]] [Optional]¶

field deleted_messages: list[str] [Optional]¶

field presence_changes: list[dict[str, Any]] [Optional]¶

field created_ims: list[list[int]] [Optional]¶

field created_rooms: list[dict[str, Any]] [Optional]¶

class Config[source]¶

Bases: object

Pydantic config.

add_mock_user(user_id: int, display_name: str, username: str, email: str | None = None) → None[source]¶

Add a mock user.

Parameters:

user_id – The user’s Symphony ID.
display_name – The user’s display name.
username – The user’s username.
email – The user’s email address.

add_mock_stream(stream_id: str, name: str, stream_type: str = 'ROOM') → None[source]¶

Add a mock stream (channel).

Parameters:

stream_id – The stream ID.
name – The stream name.
stream_type – The stream type (ROOM, IM, MIM).

add_mock_message(stream_id: str, user_id: int, content: str, message_id: str | None = None, timestamp: datetime | None = None) → str[source]¶

Add a mock message to a stream.

Parameters:

stream_id – The stream ID.
user_id – The sender’s user ID.
content – The message content (MessageML).
message_id – Optional message ID. Generated if not provided.
timestamp – Optional timestamp. Uses current time if not provided.

Returns:

The message ID.

set_mock_presence(user_id: str, status: SymphonyPresenceStatus) → None[source]¶

Set mock presence for a user.

Parameters:

user_id – The user ID.
status – The presence status.

async connect() → None[source]¶

Connect to mock Symphony.

Always succeeds immediately.

async disconnect() → None[source]¶: Disconnect from mock Symphony.

Fetch a mock user by ID or other attributes.

Parameters:

identifier – A User object or user ID string.
id – User ID.
name – Display name to search for.
email – Email address to search for.
handle – Username to search for.

Returns:

The user if found, None otherwise.

Fetch a mock channel (stream) by ID or name.

Parameters:

identifier – A Channel object or stream ID string.
id – Stream ID.
name – Stream name to search for.

Returns:

The channel if found, None otherwise.

Fetch messages from a mock stream.

Parameters:

channel – The channel to fetch from (ID string or Channel object).
limit – Maximum number of messages.
before – Fetch messages before this timestamp (ms).
after – Fetch messages after this timestamp (ms).

Returns:

List of messages.

async send_message(channel: str | Channel, content: str, **kwargs: Any) → Message[source]¶

Send a mock message.

Parameters:

channel – The channel to send to (stream ID string or Channel object).
content – The message content (MessageML).
**kwargs – Additional options (data, attachments).

Returns:

The sent message.

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

Edit a mock message.

Parameters:

message – The message to edit (ID string or Message object).
content – The new content.
channel – The channel containing the message (required if message is a string).
**kwargs – Additional options.

Returns:

The edited message.

async delete_message(message: str | Message, channel: str | Channel | None = None) → None[source]¶

Delete (suppress) a mock message.

Parameters:

message – The message to delete (ID string or Message object).
channel – The channel containing the message (required if message is a string).

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

Forward a mock message to another stream.

Parameters:

message – The message to forward (SymphonyMessage object).
to_channel – The destination stream (ID string or Channel object).
include_attribution – If True, include info about original source.
prefix – Optional text to prepend to the forwarded message.
**kwargs – Additional options.

Returns:

The forwarded message in the destination stream.

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

Set mock presence.

Parameters:

status – Presence status.
status_text – Not used in Symphony.
**kwargs – Additional options.

async get_presence(user: str | User) → Presence | None[source]¶

Get mock presence for a user.

Parameters:: user – The user to get presence for (ID string or User object).
Returns:: The user’s presence.

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

Add a reaction.

Symphony doesn’t support reactions.

Raises:: NotImplementedError – Symphony doesn’t support reactions.

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

Remove a reaction.

Symphony doesn’t support reactions.

Raises:: NotImplementedError – Symphony doesn’t support reactions.

mention_user(user: User) → str[source]¶

Format a user mention.

Parameters:: user – The user to mention.
Returns:: Symphony mention format.

mention_channel(channel: Channel) → str[source]¶

Format a channel mention.

Parameters:: channel – The channel to mention.
Returns:: The channel name.

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

Create a mock DM/IM.

Parameters:: users – List of users to include (ID strings or User objects).
Returns:: The stream ID.

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

Create a mock IM.

This is an alias for create_dm.

Parameters:: users – List of users to include (ID strings or User objects).
Returns:: The stream ID.

async create_channel(name: str, description: str = '', public: bool = False, **kwargs: Any) → str | None[source]¶

Create a mock channel/room.

Parameters:

name – The room name.
description – The room description.
public – Whether the room is public.
**kwargs – Additional options: - read_only: Whether the room is read-only.

Returns:

The stream ID.

async create_room(name: str, description: str = '', public: bool = False, **kwargs: Any) → str | None[source]¶

Create a mock room.

This is an alias for create_channel.

Parameters:

name – The room name.
description – The room description.
public – Whether the room is public.
**kwargs – Additional options.

Returns:

The stream ID.

reset() → None[source]¶

Reset all mock data and tracking.

Useful for cleaning up between tests.

Telegram¶

Telegram-specific configuration.

This module provides the Telegram-specific configuration class.

pydantic model chatom.telegram.config.TelegramConfig[source]¶

Bases: BackendConfig

Configuration for the Telegram backend.

bot_token¶: The Telegram Bot API token (from @BotFather).

api_url¶: Custom Telegram Bot API URL (for local API servers).

timeout¶: HTTP request timeout in seconds.

connect_timeout¶: Connection timeout in seconds.

read_timeout¶: Read timeout in seconds.

write_timeout¶: Write timeout in seconds.

pool_timeout¶: Connection pool timeout in seconds.

field bot_token: SecretStr = SecretStr('')¶: Telegram Bot API token from @BotFather.

field connect_timeout: float = 5.0¶: Connection timeout in seconds.

field read_timeout: float = 5.0¶: Read timeout in seconds.

field write_timeout: float = 5.0¶: Write timeout in seconds.

field pool_timeout: float = 1.0¶: Connection pool timeout in seconds.

property bot_token_str: str¶: Get the bot token as a plain string.

property has_token: bool¶: Check if a bot token is configured.

Telegram-specific User model.

This module provides the Telegram-specific User class.

pydantic model chatom.telegram.user.TelegramUser[source]¶

Bases: User

Telegram-specific user with additional Telegram fields.

first_name¶: User’s first name.

last_name¶: User’s last name.

username¶: User’s Telegram username (without @).

language_code¶: IETF language tag of the user’s language.

is_premium¶: Whether the user has Telegram Premium.

added_to_attachment_menu¶: Whether the user added the bot to attachment menu.

field first_name: str = ''¶: User’s first name.

field last_name: str = ''¶: User’s last name.

field username: str = ''¶: User’s Telegram username (without @).

field language_code: str = ''¶: IETF language tag of the user’s language.

field is_premium: bool = False¶: Whether the user has Telegram Premium.

field added_to_attachment_menu: bool = False¶: Whether the user added the bot to attachment menu.

property full_name: str¶: Get the user’s full name.

classmethod from_telegram_user(user: Any) → TelegramUser[source]¶

Create a TelegramUser from a python-telegram-bot User object.

Parameters:: user – A telegram.User object.
Returns:: A TelegramUser instance.

Telegram-specific Channel model.

This module provides the Telegram-specific Channel class.

pydantic model chatom.telegram.channel.TelegramChannel[source]¶

Bases: Channel

Telegram-specific channel (chat) with additional Telegram fields.

In Telegram, “channels” are called “chats” and can be private (1:1), groups, supergroups, or broadcast channels.

chat_type¶: Telegram chat type.

description¶: Chat description/bio.

invite_link¶: Primary invite link for the chat.

has_protected_content¶: Whether messages are protected from forwarding.

is_forum¶: Whether this supergroup has forum topics enabled.

chat_photo_url¶: URL of the chat’s photo if available.

field chat_type: TelegramChatType = TelegramChatType.PRIVATE¶: Telegram chat type.

field description: str = ''¶: Chat description/bio.

field invite_link: str = ''¶: Primary invite link for the chat.

field has_protected_content: bool = False¶: Whether messages are protected from forwarding.

field is_forum: bool = False¶: Whether this supergroup has forum topics enabled.

field chat_photo_url: str = ''¶: URL of the chat’s photo if available.

field username: str = ''¶: Public @username of the chat (without @), if set.

property is_group_chat: bool¶: Check if this is a group or supergroup chat.

property is_broadcast_channel: bool¶: Check if this is a broadcast channel.

classmethod from_telegram_chat(chat: Any) → TelegramChannel[source]¶

Create a TelegramChannel from a python-telegram-bot Chat object.

Parameters:: chat – A telegram.Chat object.
Returns:: A TelegramChannel instance.

class chatom.telegram.channel.TelegramChatType(value)[source]¶

Bases: str, Enum

Telegram chat types.

Telegram-specific Message model.

This module provides the Telegram-specific Message class.

pydantic model chatom.telegram.message.TelegramMessage[source]¶

Bases: Message

Telegram-specific message with additional Telegram fields.

message_id¶: Telegram’s integer message ID within the chat.

chat_id¶: The chat ID this message belongs to.

message_thread_id¶: Thread (topic) ID in a forum supergroup.

reply_to_message_id¶: ID of the message being replied to.

forward_origin¶: Information about the original forwarded message.

entities¶: List of special entities (mentions, URLs, etc.) from the API.

has_protected_content¶: Whether the message is protected from forwarding.

field message_id: int = 0¶: Telegram’s integer message ID within the chat.

field chat_id: str = ''¶: The chat ID this message belongs to.

field message_thread_id: int | None = None¶: Thread (topic) ID in a forum supergroup.

field reply_to_message_id: int | None = None¶: ID of the message being replied to.

field forward_origin: dict[str, Any] | None = None¶: Information about the original forwarded message.

field entities: list[dict[str, Any]] [Optional]¶: List of message entities from the API.

field has_protected_content: bool = False¶: Whether the message is protected from forwarding.

to_formatted() → FormattedMessage[source]¶

Convert this Telegram message to a FormattedMessage.

Returns:: The formatted message representation.
Return type:: FormattedMessage

classmethod from_formatted(formatted: FormattedMessage, backend: str = '', **kwargs: Any) → TelegramMessage[source]¶

Create a TelegramMessage from a FormattedMessage.

Parameters:

formatted – The FormattedMessage to convert.
backend – Target backend (ignored, always uses HTML format).
**kwargs – Additional message attributes.

Returns:

A new TelegramMessage instance.

Return type:

TelegramMessage

classmethod from_api_response(data: dict[str, Any]) → TelegramMessage[source]¶

Create a TelegramMessage from a Telegram API response dict.

Parameters:: data – The API response data (message dict).
Returns:: A TelegramMessage instance.

classmethod from_telegram_message(msg: Any) → TelegramMessage[source]¶

Create a TelegramMessage from a python-telegram-bot Message object.

Parameters:: msg – A telegram.Message object.
Returns:: A TelegramMessage instance.

mentions_user(user: User) → bool[source]¶

Check if this message mentions a specific user.

Parameters:: user – The User to check for.
Returns:: True if the message mentions the user.

Telegram-specific Presence model.

This module provides the Telegram-specific Presence class.

Note: Telegram has very limited presence support. Bots cannot see user online/offline status. Only “last seen” approximations are available for some users, and even that is privacy-controlled.

pydantic model chatom.telegram.presence.TelegramPresence[source]¶

Bases: Presence

Telegram-specific presence.

Telegram provides limited presence information. Bots can only see a user’s status if the user has not restricted that in privacy settings.

last_seen_approximate¶: Approximate last seen description (e.g. “recently”, “within a week”, “within a month”, “long time ago”).

field last_seen_approximate: str = ''¶: Approximate last seen description.

Telegram-specific mention utilities.

This module registers Telegram-specific mention formatting. Telegram uses HTML-style mentions: <a href=”tg://user?id=123”>Name</a> or @username for public usernames.

chatom.telegram.mention.mention_channel(channel: Channel) → str[source]¶

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

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

chatom.telegram.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.telegram.mention.mention_user(user: User) → str[source]¶

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

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

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

chatom.telegram.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'

Telegram backend implementation for chatom.

pydantic model chatom.telegram.backend.TelegramBackend[source]¶

Bases: BackendBase

Telegram backend implementation using python-telegram-bot.

Uses the Telegram Bot API via python-telegram-bot library for all API interactions. Supports sending/receiving messages, reactions, file uploads, and real-time message streaming via polling.

name¶: The backend identifier (‘telegram’).

display_name¶: Human-readable name.

format¶: Telegram uses the Bot API HTML subset for rich text.

capabilities¶: Telegram-specific capabilities.

config¶: Telegram-specific configuration.

Example

>>> config = TelegramConfig(bot_token="123456:ABC-DEF...")
>>> backend = TelegramBackend(config=config)
>>> await backend.connect()
>>> msg = await backend.send_message("-100123456789", "Hello!")

name: ClassVar[str] = 'telegram'¶

display_name: ClassVar[str] = 'Telegram'¶

format: ClassVar[Format] = 'telegram-html'¶

user_class¶: alias of TelegramUser

channel_class¶: alias of TelegramChannel

presence_class¶: alias of TelegramPresence

field capabilities: BackendCapabilities | None = BackendCapabilities(capabilities=frozenset({<Capability.VIDEOS: 'videos'>, <Capability.AUDIO: 'audio'>, <Capability.CODE_BLOCKS: 'code_blocks'>, <Capability.FILES: 'files'>, <Capability.IMAGES: 'images'>, <Capability.USER_MENTIONS: 'user_mentions'>, <Capability.MARKDOWN: 'markdown'>, <Capability.PLAINTEXT: 'plaintext'>, <Capability.HTML: 'html'>, <Capability.REPLIES: 'replies'>, <Capability.EMOJI_REACTIONS: 'emoji_reactions'>}), max_message_length=4000, max_attachment_size=26214400, max_attachments=10, max_embeds=10, max_reactions=20)¶

field config: TelegramConfig [Optional]¶

property bot_user_id: str | None¶: Get the bot’s user ID.

property bot_user_name: str | None¶: Get the bot’s username.

async connect() → None[source]¶

Connect to Telegram by initializing the Bot and verifying credentials.

Raises:

ImportError – If python-telegram-bot is not installed.
RuntimeError – If the token is missing or invalid.

async disconnect() → None[source]¶: Disconnect from Telegram.

Fetch a user from Telegram.

Note: Telegram Bot API does not support arbitrary user lookup. Users can only be fetched if the bot has interacted with them via getChatMember in a known chat. ID-based lookup checks the cache first. name/handle lookup is cache-only.

Parameters:

identifier – A TelegramUser object or user ID string.
id – User ID.
name – Display name to search for (cache only).
email – Email (not supported by Telegram).
handle – Username to search for (cache only).

Returns:

The user if found, None otherwise.

Fetch a channel (chat) from Telegram.

Uses the getChat API to fetch chat information by ID. Name-based lookup checks cache only.

Parameters:

identifier – A TelegramChannel object or chat ID string.
id – Chat ID.
name – Chat name to search for (cache only).

Returns:

The channel if found, None otherwise.

Fetch messages from a Telegram chat.

Note: The Telegram Bot API does not support fetching message history. Bots can only receive messages in real-time via updates/webhooks. This method returns an empty list as message history is not available.

Parameters:

channel – The channel to fetch messages from.
limit – Maximum number of messages.
before – Fetch messages before this message.
after – Fetch messages after this message.

Returns:

Empty list (Telegram Bot API limitation).

async send_message(channel: str | Channel, content: str, **kwargs: Any) → TelegramMessage[source]¶

Send a message to a Telegram chat.

Uses the sendMessage API. Supports HTML parse mode by default.

Parameters:

channel – The chat to send to (ID string or Channel object).
content – The message content (HTML formatted). kwargs: Telegram thread, reply, parse mode, notification, and content-protection options.

Returns:

The sent message.

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

Upload a file to a Telegram chat.

Uses send_photo for image MIME types and send_document for everything else.

async download_attachment(attachment: Any, *, message: Message | None = None) → bytes[source]¶

Download an attachment’s bytes from Telegram.

Telegram media is referenced by file_id (stored as the attachment id), which is resolved to a temporary download via getFile.

async edit_message(message: str | Message, content: str, channel: str | Channel | None = None, **kwargs: Any) → TelegramMessage[source]¶

Edit a Telegram message.

Uses the editMessageText API.

Parameters:

message – The message to edit (ID string or TelegramMessage).
content – The new content.
channel – The chat containing the message (required if message is str).
**kwargs – Additional options.

Returns:

The edited message.

async delete_message(message: str | Message, channel: str | Channel | None = None) → None[source]¶

Delete a Telegram message.

Uses the deleteMessage API.

Parameters:

message – The message to delete (ID string or TelegramMessage).
channel – The chat containing the message (required if message is str).

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

Forward a message to another Telegram chat.

Telegram has native forward support via the forwardMessage API. If include_attribution is False and source info is available, uses native forwarding. Otherwise sends a new message with attribution.

Parameters:

message – The message to forward (TelegramMessage object).
to_channel – The destination chat.
include_attribution – If True, include info about original source.
prefix – Optional text to prepend.
**kwargs – Additional options.

Returns:

The forwarded message.

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

Add a reaction to a message.

Uses the setMessageReaction API (Telegram Bot API 7.0+).

Parameters:

message – The message to react to.
emoji – The emoji to react with (Unicode emoji).
channel – The chat containing the message.

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

Remove a reaction from a message.

Telegram’s API sets the full reaction list, so removing means setting an empty list.

Parameters:

message – The message to remove reaction from.
emoji – The emoji to remove (unused - clears all bot reactions).
channel – The chat containing the message.

mention_user(user: User) → str[source]¶

Format a user mention for Telegram.

Parameters:: user – The user to mention.
Returns:: Telegram user mention string.

mention_channel(channel: Channel) → str[source]¶

Format a channel mention for Telegram.

Parameters:: channel – The channel to mention.
Returns:: Telegram channel mention string.

async get_bot_info() → User | None[source]¶

Get information about the connected bot.

Returns:: The bot’s User object.

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

Set bot presence.

Note: Telegram bots don’t have traditional presence/status. This is a no-op for API compatibility.

Parameters:

status – Status string (ignored).
status_text – Status text (ignored).

async get_presence(user: str | User) → Presence | None[source]¶

Get a user’s presence.

Note: Telegram bots have very limited access to user presence. Returns None as presence is not reliably available.

Parameters:: user – The user to check.
Returns:: None (Telegram limitation).

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

Create a DM with a user.

In Telegram, DMs are just private chats identified by the user’s ID. The bot must have been contacted by the user first.

Parameters:: users – List of users (only first is used for 1:1 DM).
Returns:: The chat ID (same as user ID for private chats).

async discover_chats(timeout: float = 5.0) → list[TelegramChannel][source]¶

Discover chats by consuming recent getUpdates.

Useful for finding chat IDs / usernames when you know the bot has received messages in specific groups or channels.

Parameters:: timeout – Seconds to wait for updates (default 5).
Returns:: List of unique TelegramChannel objects found.

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

Stream incoming messages in real-time using polling.

Uses Telegram’s getUpdates long-polling to receive messages.

Parameters:

channel – Optional chat to filter messages by.
skip_own – If True, skip messages sent by the bot itself.
skip_history – If True, skip old messages.

Yields:

TelegramMessage – Each message as it arrives.

Mock Telegram backend for testing.

This module provides a mock implementation of the Telegram backend for use in testing without requiring an actual Telegram connection.

pydantic model chatom.telegram.testing.MockTelegramBackend[source]¶

Bases: TelegramBackend

Mock Telegram backend for testing.

Stores all data in memory and provides methods to set up mock data for tests.

Example

>>> backend = MockTelegramBackend()
>>> backend.add_mock_user("123", "Alice", "alice")
>>> backend.add_mock_channel("-100456", "general")
>>> await backend.connect()
>>> user = await backend.fetch_user("123")
>>> assert user.name == "Alice"

add_mock_user(id: str, name: str, handle: str = '', *, first_name: str = '', last_name: str = '', username: str = '', is_bot: bool = False, is_premium: bool = False, language_code: str = '') → TelegramUser[source]¶

Add a mock user for testing.

Parameters:

id – The user ID.
name – The user’s display name.
handle – The username (without @).
first_name – First name.
last_name – Last name.
username – Telegram username (defaults to handle).
is_bot – Whether the user is a bot.
is_premium – Whether the user has Telegram Premium.
language_code – User’s language code.

Returns:

The created mock user.

add_mock_channel(id: str, name: str, channel_type: str = 'supergroup', *, topic: str = '', description: str = '', chat_type: TelegramChatType | None = None, is_forum: bool = False) → TelegramChannel[source]¶

Add a mock channel for testing.

Parameters:

id – The chat ID.
name – The chat name/title.
channel_type – Chat type string (“private”, “group”, “supergroup”, “channel”).
topic – The chat topic.
description – The chat description.
chat_type – TelegramChatType enum (overrides channel_type).
is_forum – Whether the chat is a forum supergroup.

Returns:

The created mock channel.

add_mock_message(channel_id: str, user_id: str, content: str, *, message_id: str | None = None, timestamp: datetime | None = None, edited: bool = False) → str[source]¶

Add a mock message for testing.

Parameters:

channel_id – The chat containing the message.
user_id – The author’s user ID.
content – The message content.
message_id – Optional message ID (auto-generated if not provided).
timestamp – Message timestamp.
edited – Whether the message was edited.

Returns:

The message ID.

set_mock_presence(user_id: str, status: PresenceStatus = PresenceStatus.ONLINE, *, last_seen_approximate: str = '') → TelegramPresence[source]¶

Set mock presence for a user.

Parameters:

user_id – The user ID.
status – The presence status.
last_seen_approximate – Approximate last seen description.

Returns:

The created mock presence.

property sent_messages: list[TelegramMessage]¶: Get all messages sent through this backend.

property edited_messages: list[TelegramMessage]¶: Get all messages edited through this backend.

property deleted_messages: list[dict[str, str]]¶: Get all message IDs deleted through this backend.

get_sent_messages() → list[TelegramMessage][source]¶: Get all messages sent through this backend (copy).

get_edited_messages() → list[TelegramMessage][source]¶: Get all messages edited through this backend (copy).

get_deleted_messages() → list[dict[str, str]][source]¶: Get all messages deleted through this backend (copy).

get_reactions() → list[dict[str, str]][source]¶: Get all reactions added/removed through this backend (copy).

get_presence_updates() → list[dict[str, Any]][source]¶: Get all presence updates (copy).

property created_dms: list[list[str]]¶: Get all DMs created through this backend.

clear() → None[source]¶: Clear all mock data and tracking stores.

async connect() → None[source]¶: Mock connect - always succeeds.

async disconnect() → None[source]¶: Mock disconnect.

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[source]¶: Fetch a mock user.

async fetch_channel(identifier: str | Channel | None = None, *, id: str | None = None, name: str | None = None) → Channel | None[source]¶: Fetch a mock channel.

async fetch_messages(channel: str | Channel, limit: int = 100, before: str | Message | datetime | None = None, after: str | Message | datetime | None = None) → list[Message][source]¶: Fetch mock messages from a channel.

async send_message(channel: str | Channel, content: str, **kwargs: Any) → TelegramMessage[source]¶: Send a mock message.

async edit_message(message: str | Message, content: str, channel: str | Channel | None = None, **kwargs: Any) → TelegramMessage[source]¶: Edit a mock message.

async delete_message(message: str | Message, channel: str | Channel | None = None) → None[source]¶: Delete a mock message.

async forward_message(message: str | Message, to_channel: str | Channel, *, include_attribution: bool = True, prefix: str | None = None, **kwargs: Any) → TelegramMessage[source]¶: Forward a mock message.

async set_presence(status: str, status_text: str | None = None, **kwargs: Any) → None[source]¶: Set mock presence.

async get_presence(user: str | User) → Presence | None[source]¶: Get mock presence for a user.

async add_reaction(message: str | Message, emoji: str, channel: str | Channel | None = None) → None[source]¶: Add a mock reaction.

async remove_reaction(message: str | Message, emoji: str, channel: str | Channel | None = None) → None[source]¶: Remove a mock reaction.

async create_dm(users: list[str | User]) → str | None[source]¶: Create a mock DM channel.