Connect and send through a backend¶
Install Chatom and the SDKs for the platforms you use:
pip install chatom
pip install slack-sdk discord.py symphony-bdk-python python-telegram-bot
Create a backend¶
Each adapter accepts its own configuration model:
from chatom.discord import DiscordBackend, DiscordConfig
from chatom.slack import SlackBackend, SlackConfig
from chatom.symphony import SymphonyBackend, SymphonyConfig
from chatom.telegram import TelegramBackend, TelegramConfig
slack = SlackBackend(config=SlackConfig(bot_token="xoxb-..."))
discord = DiscordBackend(config=DiscordConfig(token="..."))
symphony = SymphonyBackend(
config=SymphonyConfig(
host="company.symphony.com",
bot_username="chatom-bot",
bot_private_key_path="bot-private-key.pem",
)
)
telegram = TelegramBackend(config=TelegramConfig(bot_token="..."))
Pass secrets from environment variables or a secret manager in production.
Use the common lifecycle¶
The rest of the workflow is backend-independent:
async def announce(backend, channel_name: str, content: str):
await backend.connect()
try:
channel = await backend.fetch_channel(name=channel_name)
return await backend.send_message(channel, content)
finally:
await backend.disconnect()
chatom.BackendBase.fetch_channel() accepts an identifier or named keyword. Methods that accept a channel generally accept either a channel ID or a chatom.Channel.
Use the synchronous facade¶
For a synchronous caller, use chatom.BackendBase.sync and close it when finished:
backend.sync.connect()
try:
channel = backend.sync.fetch_channel(name="operations")
backend.sync.send_message(channel, "Deployment complete")
finally:
backend.sync.disconnect()
backend.sync.close()
Check an optional operation¶
from chatom import Capability
if backend.capabilities.supports(Capability.EMOJI_REACTIONS):
await backend.add_reaction(channel, message, "thumbsup")
Consult the backend’s chatom.BackendCapabilities before relying on reactions, presence, editing, files, threads, or interactions.
Read and search messages¶
from datetime import datetime, timedelta, timezone
since = datetime.now(timezone.utc) - timedelta(hours=1)
messages = await backend.fetch_messages(channel, limit=100, after=since)
async for message in backend.read_messages(channel, limit=20):
print(message.author_name, message.content)
if backend.capabilities.supports(Capability.MESSAGE_SEARCH):
matches = await backend.search_messages("deployment", channel=channel)
History is returned newest-first. Bounds accept message IDs, chatom.Message objects, or timezone-aware datetimes.
Edit, delete, and react¶
if backend.capabilities.supports(Capability.EDITING):
message = await backend.edit_message(message, "Corrected content")
if backend.capabilities.supports(Capability.EMOJI_REACTIONS):
await backend.add_reaction(message, "thumbsup")
await backend.remove_reaction(message, "thumbsup")
if backend.capabilities.supports(Capability.DELETING):
await backend.delete_message(message)
Upload and download files¶
uploaded = await backend.upload_file(
channel,
report_bytes,
filename="report.csv",
content_type="text/csv",
content="Latest report",
)
for attachment in incoming_message.attachments:
data = await backend.download_attachment(attachment, message=incoming_message)
Send a direct message¶
user = await backend.fetch_user(email="alice@example.com")
message = await backend.send_dm(user, "The report is ready")
Listen for events¶
async for message in backend.listen(channel=channel):
if message.content == "ping":
await backend.reply_in_thread(message, "pong")
Use chatom.BackendBase.stream_interactions() for button, menu, and modal events. Streaming support depends on the selected backend and its connection mode.
Manage presence¶
if backend.capabilities.supports(Capability.PRESENCE):
await backend.set_presence("online", "Ready")
presence = await backend.get_presence(user)