API overview

This bot showcases how to use most of the features in the library. Check the commands section and handlers section to see the implementation of each command and handler. The code shown here can be found the examples folder.

This bot uses additional libraries. If you cloned the repository, install them with:

uv sync --group examples

Bot code:

import os

from examples.commands import (
    AboutCommand,
    AttachmentCommand,
    BroadcastCommand,
    CloseCommand,
    DeleteCommand,
    DeleteLocalAttachmentCommand,
    EditCommand,
    EditNotifierCommand,
    HelpCommand,
    LinkPreviewCommand,
    PingCommand,
    PollCommand,
    ReactCommand,
    ReceiptCommand,
    RegexTriggeredCommand,
    ReplyCommand,
    StylesCommand,
    TriggeredCommand,
    TypingCommand,
    TypingIndicatorToggleCommand,
    UpdateContactCommand,
    UpdateGroupCommand,
)
from examples.handlers import (
    DeletionNotifierHandler,
    FilteredReactionHandler,
    GroupUpdateNotifierHandler,
    ReactionDetailsHandler,
    TypingIndicatorHandler,
    WelcomeHandler,
)
from signalbot import Config, SignalBot


def main() -> None:
    phone_number = os.environ["PHONE_NUMBER"]

    # Replace the recipient with your own phone number or group ID to
    # receive the welcome message and the broadcast message.
    contact_phone_number = os.environ.get("CONTACT_PHONE_NUMBER")

    bot = SignalBot(Config(phone_number=phone_number))

    bot.register(WelcomeHandler(recipient=contact_phone_number))

    # By default the handlers are enabled for all contacts and all groups
    bot.register(HelpCommand())
    bot.register(PingCommand())
    bot.register(ReplyCommand())
    bot.register(RegexTriggeredCommand())
    bot.register(ReactCommand())
    bot.register(EditCommand())
    bot.register(EditNotifierCommand())
    bot.register(DeleteCommand())
    bot.register(DeleteLocalAttachmentCommand())
    bot.register(StylesCommand())
    bot.register(LinkPreviewCommand())
    bot.register(CloseCommand())
    bot.register(PollCommand())
    bot.register(ReceiptCommand())
    bot.register(AboutCommand())
    bot.register(ReactionDetailsHandler())
    bot.register(FilteredReactionHandler())
    bot.register(DeletionNotifierHandler())
    bot.register(GroupUpdateNotifierHandler())

    # Disabled by default; toggle it with the enable-typing-indicator /
    # disable-typing-indicator commands.
    typing_indicator_handler = TypingIndicatorHandler()
    bot.register(typing_indicator_handler)
    bot.register(TypingIndicatorToggleCommand(typing_indicator_handler))

    # The handler will only trigger for group messages
    bot.register(AttachmentCommand(), contacts=False)
    bot.register(UpdateGroupCommand(), contacts=False)

    # The handler will only trigger for private messages, since updating a
    # contact's metadata doesn't apply to groups
    bot.register(UpdateContactCommand(), groups=False)

    # Replace with the phone numbers or group IDs that should receive the
    # broadcast message
    broadcast_recipients = [contact_phone_number] if contact_phone_number else []
    bot.register(BroadcastCommand(recipients=broadcast_recipients))

    # The handler will only trigger the group named "My Group"
    bot.register(TypingCommand(), groups=["My Group"], contacts=False)

    # The handler will only trigger for the contact "+490123456789"
    bot.register(TriggeredCommand(), contacts=["+490123456789"], groups=False)

    bot.start()


if __name__ == "__main__":
    main()

Commands¶

AboutCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class AboutCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "about: 📋 Show signal-cli-rest-api version information."

    @text_triggered("about")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        about = await context.bot.general.about()
        await context.send(
            SendMessage(text=f"signal-cli-rest-api version: {about.version}")
        )

AttachmentCommand

from anyio import Path

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class AttachmentCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "friday: 🦀 Send an image."

    @text_triggered("friday")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.send(
            SendMessage(
                text="https://www.youtube.com/watch?v=pU2SdH1HBuk",
                attachments=[Path(__file__).parent / "image.jpeg"],
            )
        )

BroadcastCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class BroadcastCommand(DataMessageHandler):
    def __init__(self, recipients: list[str]) -> None:
        self.recipients = recipients

    def help_message(self) -> str:
        return "broadcast: 📢 Send the same message to multiple recipients at once."

    @text_triggered("broadcast")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        if self.recipients == []:
            await context.send(
                SendMessage(
                    text="No recipients specified for broadcast.",
                )
            )
            return

        await context.bot.messages.send_multiple(
            SendMessage(
                text="📢 Broadcast message!",
            ),
            recipients=[*self.recipients, context.message.source_or_group_id()],
        )

CloseCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class CloseCommand(DataMessageHandler):
    """Demonstrates that `request_stop()` is safe to call from a handler,
    even though the handler itself runs on one of the tasks being shut down.
    """

    def help_message(self) -> str:
        return "close: 🛑 Gracefully shut the bot down."

    @text_triggered("close")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.send(SendMessage(text="Shutting down..."))
        context.bot.request_stop()

UpdateContactCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    UpdateContact,
    text_triggered,
)

FIVE_MINUTES_IN_SECONDS = 5 * 60


class UpdateContactCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "expiration: ⏳ Toggle disappearing messages between off and 5 minutes."

    @text_triggered("expiration")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        if context.message.expires_in_seconds == FIVE_MINUTES_IN_SECONDS:
            new_expiration = 0
            status = "disabled"
        else:
            new_expiration = FIVE_MINUTES_IN_SECONDS
            status = "set to 5 minutes"

        await context.update_contact(
            UpdateContact(expiration_in_seconds=new_expiration)
        )
        await context.send(SendMessage(text=f"Disappearing messages {status}."))

DeleteCommand & DeleteLocalAttachmentCommand

import asyncio

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class DeleteCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "delete: 🗑️ Delete a message."

    @text_triggered("delete")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        sent_message = await context.send(
            SendMessage(text="This message will be deleted in two seconds.")
        )
        await asyncio.sleep(2)
        await context.remote_delete(sent_message)


class DeleteLocalAttachmentCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "delete-attachment: 🗑️ Delete the local copy of an attachment."

    @text_triggered("delete-attachment")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        attachments = context.message.attachments
        if attachments is None or len(attachments) == 0:
            await context.send(SendMessage(text="Please send an attachment to delete."))
            return

        for attachment in attachments:
            attachment_path = await attachment.local_path()

            if attachment_path is None:
                continue

            if await attachment_path.exists():
                await context.send(SendMessage(text=f"Received file {attachment_path}"))

            await context.delete_attachment(attachment)

            if not await attachment_path.exists():
                await context.send(SendMessage(text=f"Deleted file {attachment_path}"))

EditCommand & EditNotifierCommand

import asyncio

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    EditMessage,
    SendMessage,
    text_triggered,
)


class EditCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "edit: ✏️ Edit a message."

    @text_triggered("edit")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        sent_message = await context.send(
            SendMessage(text="This message will be edited in two seconds.")
        )
        await asyncio.sleep(2)
        await context.edit(
            SendMessage(text="This message has been edited."),
            sent_message,
        )


class EditNotifierCommand(DataMessageHandler):
    """`EditMessage` is a `DataMessage` subclass, so edits are dispatched to the
    same `handle_data_message` method as regular messages — check `isinstance`
    to tell them apart.
    """

    def help_message(self) -> str:
        return "Message edited: 📝 Notifies when someone edits a sent message."

    async def handle_data_message(self, context: DataMessageContext) -> None:
        if not isinstance(context.message, EditMessage):
            return

        await context.send(
            SendMessage(text=f"You edited your message to: {context.message.text}")
        )

UpdateGroupCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    UpdateGroup,
    text_triggered,
)


class UpdateGroupCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "set-group-description: 📝 Update this group's description."

    @text_triggered("set-group-description")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        # group_id_or_name is filled in by context.update_group() with the
        # group this message came from.
        await context.update_group(UpdateGroup(description="Managed by signalbot 🤖"))
        await context.send(SendMessage(text="Updated the group description."))

HelpCommand

from typing import Protocol, runtime_checkable

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    SignalBot,
    text_triggered,
)


@runtime_checkable
class HasHelpMessage(Protocol):
    def help_message(self) -> str: ...


def build_help_messages(bot: SignalBot) -> tuple[str, str]:
    commands = []
    handlers = []
    for registered, _, _, _ in bot.handlers:
        if not isinstance(registered, HasHelpMessage):
            continue
        if isinstance(registered, DataMessageHandler):
            commands.append(registered.help_message())
        else:
            handlers.append(registered.help_message())

    command_sections = []
    if commands:
        entries = "\n".join(f"  {entry}" for entry in commands)
        command_sections.append(f"Commands:\n{entries}")

    handler_sections = []
    if handlers:
        entries = "\n".join(f"  {entry}" for entry in handlers)
        handler_sections.append(f"Handlers:\n{entries}")

    return "\n".join(command_sections), "\n".join(handler_sections)


class HelpCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "help: 🆘 Shows information about available commands."

    @text_triggered("help")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        commands_msg, handlers_msg = build_help_messages(context.bot)
        await context.send(SendMessage(text=commands_msg))
        await context.send(SendMessage(text=handlers_msg))

LinkPreviewCommand

from anyio import Path

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    LinkPreview,
    SendMessage,
    text_triggered,
)


class LinkPreviewCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "link-preview: 🧽 Send a link preview."

    @text_triggered("link-preview")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.send(
            SendMessage(
                text="This is the link preview for https://www.youtube.com/watch?v=pU2SdH1HBuk",
                link_preview=LinkPreview(
                    description="A link preview description",
                    title="A link preview title",
                    url="https://www.youtube.com/watch?v=pU2SdH1HBuk",
                    thumbnail=Path(__file__).parent / "image.jpeg",
                ),
            )
        )

TriggeredCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class TriggeredCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "command-1, command-2 or command-3: 😤😤😤 Decorator example."

    # add case_sensitive=True for case sensitive triggers
    @text_triggered("command-1", "Command-2", "CoMmAnD-3")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.send(SendMessage(text="Multi command trigger"))

PingCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class PingCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "ping: 🏓 Listen for a ping and send a pong reply."

    @text_triggered("ping")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.send(SendMessage(text="pong"))

PollCommand

from signalbot import CreatePoll, DataMessageContext, DataMessageHandler, text_triggered


class PollCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "poll: 🗳️ Create a poll."

    @text_triggered("poll")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.create_poll(
            CreatePoll(
                question="Cats or dogs?",
                answers=["Cats", "Dogs"],
                allow_multiple_selections=False,
            )
        )

ReactCommand

from signalbot import DataMessageContext, DataMessageHandler, text_triggered


class ReactCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "send-reaction: 🎉 Send a reaction to a message."

    @text_triggered("send-reaction")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.react("🎉")

ReceiptCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    ReceiptType,
    text_triggered,
)


class ReceiptCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "receipt: 👀 Send a read receipt back for this message."

    @text_triggered("receipt")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.send_receipt(ReceiptType.READ)

RegexTriggeredCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    regex_triggered,
)


class RegexTriggeredCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "^[\\w\\.-]+@gmail\\.com$: 😤 Regular expression decorator example."

    @regex_triggered(r"^[\w\.-]+@gmail\.com$")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.send(SendMessage(text="Detected a Gmail address!"))

ReplyCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class ReplyCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "reply: 💬 Reply to a message."

    @text_triggered("reply")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.reply(SendMessage(text="This is a reply."))

StylesCommand

from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    TextMode,
    text_triggered,
)


class StylesCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "styles: 🎨 Demonstrates different text styles."

    @text_triggered("styles")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.send(
            SendMessage(text="**Bold style**", text_mode=TextMode.STYLED)
        )
        await context.send(
            SendMessage(text="*Italic style*", text_mode=TextMode.STYLED)
        )
        await context.send(
            SendMessage(text="~Strikethrough style~", text_mode=TextMode.STYLED)
        )
        await context.send(
            SendMessage(text="||Spoiler style||", text_mode=TextMode.STYLED)
        )
        await context.send(
            SendMessage(text="`Monospaced style`", text_mode=TextMode.STYLED)
        )

TypingCommand

import asyncio

from examples.handlers import TypingIndicatorHandler
from signalbot import (
    DataMessageContext,
    DataMessageHandler,
    SendMessage,
    text_triggered,
)


class TypingCommand(DataMessageHandler):
    def help_message(self) -> str:
        return "typing: ⌨️ Demonstrates typing indicator for a few seconds."

    @text_triggered("typing")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        await context.start_typing()
        seconds = 5
        await asyncio.sleep(seconds)
        await context.stop_typing()
        await context.send(SendMessage(text=f"Typed for {seconds}s"))


class TypingIndicatorToggleCommand(DataMessageHandler):
    """Turns `TypingIndicatorHandler`'s notifications on or off. Needs a reference
    to that handler's instance, so it's constructed with it in `examples/bot.py`.
    """

    def __init__(self, typing_indicator_handler: TypingIndicatorHandler) -> None:
        self._typing_indicator_handler = typing_indicator_handler

    def help_message(self) -> str:
        return (
            "enable-typing-indicator / disable-typing-indicator: ⌨️ Turns the "
            "typing indicator notifier on or off (starts disabled)."
        )

    @text_triggered("enable-typing-indicator", "disable-typing-indicator")
    async def handle_data_message(self, context: DataMessageContext) -> None:
        text = context.message.text
        if text is None:
            return

        enable = text.strip().lower() == "enable-typing-indicator"
        self._typing_indicator_handler.enabled = enable
        state = "enabled" if enable else "disabled"
        await context.send(SendMessage(text=f"Typing indicator notifier {state}"))

Handlers¶

GroupUpdateNotifierHandler

from signalbot import GroupUpdateContext, GroupUpdateHandler, SendMessage


class GroupUpdateNotifierHandler(GroupUpdateHandler):
    def help_message(self) -> str:
        return "Group update received: 👥 Notifies when a group's metadata changes."

    async def handle_group_update(self, context: GroupUpdateContext) -> None:
        group_info = context.message.group_info
        await context.send(
            SendMessage(
                text=(
                    f"Group '{group_info.group_name}' was updated "
                    f"(now at revision {group_info.revision})."
                )
            )
        )

ReactionDetailsHandler & FilteredReactionHandler

from examples.timestamps import local_datetime_str_from_timestamp
from signalbot import ReactionContext, ReactionHandler, SendMessage, reaction_triggered


class ReactionDetailsHandler(ReactionHandler):
    def help_message(self) -> str:
        return (
            "Reaction received or removed (any emoji except 👍/❤️): "
            "🎉 Replies with details about the reaction."
        )

    async def handle_reaction(self, context: ReactionContext) -> None:
        reaction = context.message

        if reaction.emoji in ["👍", "❤️"]:
            # ignore thumbs up/heart, handled by FilteredReactionHandler
            return

        if reaction.is_remove:
            await context.send(
                SendMessage(text=f"You removed your {reaction.emoji} reaction")
            )
            return

        message_sent_at = local_datetime_str_from_timestamp(reaction.timestamp)
        await context.send(
            SendMessage(
                text=(
                    f"{context.message.source_name} reacted with {reaction.emoji} "
                    f"on a message that was sent by {reaction.target_author} at "
                    f"{message_sent_at}"
                )
            )
        )


class FilteredReactionHandler(ReactionHandler):
    def help_message(self) -> str:
        return (
            "Reaction received or removed (👍 or ❤️): 🎯 Filtered reaction received "
            "decorator example."
        )

    @reaction_triggered("👍", "❤️")
    async def handle_reaction(self, context: ReactionContext) -> None:
        reaction = context.message
        if reaction.is_remove:
            await context.send(
                SendMessage(text=f"FilteredReactionHandler: {reaction.emoji} removed")
            )
            return

        await context.send(
            SendMessage(text=f"FilteredReactionHandler: {reaction.emoji} received")
        )

DeletionNotifierHandler

from examples.timestamps import local_datetime_str_from_timestamp
from signalbot import RemoteDeleteContext, RemoteDeleteHandler, SendMessage


class DeletionNotifierHandler(RemoteDeleteHandler):
    def help_message(self) -> str:
        return "Remote delete received: 🗑️ Notifies when a message was deleted."

    async def handle_remote_delete(self, context: RemoteDeleteContext) -> None:
        deleted_at = local_datetime_str_from_timestamp(context.message.timestamp)
        message = f"You've deleted a message, which was sent at {deleted_at}."
        await context.send(SendMessage(text=message))

WelcomeHandler

from examples.commands.help import build_help_messages
from signalbot import ReadyContext, ReadyHandler, SendMessage


class WelcomeHandler(ReadyHandler):
    def __init__(self, recipient: str | None) -> None:
        self.recipient = recipient

    async def handle_ready(self, context: ReadyContext) -> None:
        welcome_message = (
            "👋 Welcome! The bot is now connected and ready to receive messages.\n"
        )
        welcome_message += (
            "Send a text message with one of the commands or perform "
            "an action that a handler is listening to."
        )
        commands_msg, handlers_msg = build_help_messages(context.bot)

        if self.recipient is not None:
            await context.bot.messages.send(
                SendMessage(text=welcome_message), self.recipient
            )
            await context.bot.messages.send(
                SendMessage(text=commands_msg), self.recipient
            )
            await context.bot.messages.send(
                SendMessage(text=handlers_msg), self.recipient
            )
        else:
            print(welcome_message)
            print(commands_msg)
            print(handlers_msg)

TypingIndicatorHandler

from signalbot import SendMessage, TypingAction, TypingContext, TypingHandler


class TypingIndicatorHandler(TypingHandler):
    """Notifies when someone starts typing. Starts disabled; toggle it with the
    `enable-typing-indicator` / `disable-typing-indicator` commands in
    `examples/commands/typing.py`.
    """

    def __init__(self) -> None:
        self.enabled = False

    def help_message(self) -> str:
        return "Typing indicator received: ⌨️ Notifies when someone starts typing."

    async def handle_typing(self, context: TypingContext) -> None:
        if not self.enabled:
            return

        if context.message.action != TypingAction.STARTED:
            return

        await context.send(
            SendMessage(text=f"{context.message.source_name} is typing…")
        )