Skip to content

SignalBot

MIN_SIGNAL_CLI_REST_API_VERSION module-attribute 

MIN_SIGNAL_CLI_REST_API_VERSION = Version('0.95.0')

The minimum required version of signal-cli-rest-api for this version of signalbot.

SignalBot 

SignalBot(config: Config | Mapping | Path | str)

SignalBot is the main class for the bot. It provides methods to register handlers, start the bot, and interact with messages.

Parameters:

Name Type Description Default
config Config | Mapping | Path | str

the configuration for the bot.

 required

Example config: 

{
    phone_number: "+49123456789"
}

Source code in src/signalbot/bot.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __init__(self, config: Config | Mapping | Path | str) -> None:
    """Initialization for the SignalBot.

    Args:
        config: the configuration for the bot.

    Example config:
    ```python
    {
        phone_number: "+49123456789"
    }
    ```
    """
    self.config = load_config(config)
    self._logger = _initialize_logger(self.config.logging_level)

    self.init_task = None
    self._shutdown_task: asyncio.Task | None = None

    components = build_components(self.config, self._logger)
    self._signal = components.signal
    self._event_loop = components.event_loop
    self.scheduler = components.scheduler
    self.storage = components.storage

    self._init_actions()

attachments instance-attribute 

attachments: AttachmentActions

Delete local attachment copies.

config instance-attribute 

config: Config = load_config(config)

The configuration for the bot.

contacts instance-attribute 

contacts: ContactActions

Update contact metadata.

general instance-attribute 

general: GeneralActions

Miscellaneous signal-cli-rest-api info.

groups instance-attribute 

groups: GroupRegistry

Cache of the groups the bot is a member of, with lookup helpers. Only populated after SignalBot.start() is called and SignalBot.init_task is done. Group actions are available via groups.actions.

handlers property 

handlers: HandlerList

A list of registered handlers with their filters.

Warning

Only available after SignalBot.start() is called and SignalBot.init_task is done.

init_task instance-attribute 

init_task: Task | None = None

The initialization async task for the bot.

Warning

Only available after SignalBot.start() is called.

messages instance-attribute 

messages: MessageActions

Send, edit, or delete messages, and manage typing indicators.

polls instance-attribute 

polls: PollActions

Create polls.

reactions instance-attribute 

reactions: ReactionActions

React to messages.

receipts instance-attribute 

receipts: ReceiptActions

Send read/viewed receipts.

scheduler instance-attribute 

scheduler: AsyncIOScheduler = components.scheduler

The scheduler for running scheduled tasks.

storage instance-attribute 

storage: SQLiteStorage | RedisStorage = components.storage

The storage backend used by the bot.

close async 

close() -> None

Close the shared HTTP session.

Warning

Not safe to call while the pipeline's producer/consumer tasks may still be making requests through this session — it pulls the connection out from under them mid-request. SignalBot.stop() avoids this by cancelling and awaiting those tasks first; call this directly only once you know none of them are still running.

Source code in src/signalbot/bot.py 
233
234
235
236
237
238
239
240
241
242
243
244
async def close(self) -> None:
    """Close the shared HTTP session.

    Warning:
        Not safe to call while the pipeline's producer/consumer tasks may
        still be making requests through this session — it pulls the
        connection out from under them mid-request.
        [SignalBot.stop()][signalbot.bot.SignalBot.stop] avoids this by
        cancelling and awaiting those tasks first; call this directly only
        once you know none of them are still running.
    """
    await self._signal.close()

register 

register(
    handler: AnyHandler,
    *,
    contacts: list[str] | bool = True,
    groups: list[str] | bool = True,
    f: Callable[[ReceivedMessage], bool] | None = None
) -> None

Register a handler with optional contact/group filters.

Parameters:

Name Type Description Default
handler AnyHandler

Handler instance to register.

 required
contacts list[str] | bool

Allowed contacts or True for all.

 True
groups list[str] | bool

Allowed groups or True for all.

 True
f Callable[[ReceivedMessage], bool] | None

Optional function to further filter messages.

 None
Source code in src/signalbot/bot.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def register(
    self,
    handler: AnyHandler,
    *,
    contacts: list[str] | bool = True,
    groups: list[str] | bool = True,
    f: Callable[[ReceivedMessage], bool] | None = None,
) -> None:
    """Register a handler with optional contact/group filters.

    Args:
        handler: Handler instance to register.
        contacts: Allowed contacts or True for all.
        groups: Allowed groups or True for all.
        f: Optional function to further filter messages.
    """
    self._pipeline.register(handler, contacts=contacts, groups=groups, f=f)

request_stop 

request_stop() -> None

Schedule SignalBot.stop() without blocking the caller.

The non-blocking way to trigger shutdown from within a message handler; see SignalBot.stop() for why calling it directly also works, just not without blocking the handler until shutdown completes.

Source code in src/signalbot/bot.py 
221
222
223
224
225
226
227
228
229
230
231
def request_stop(self) -> None:
    """Schedule [SignalBot.stop()][signalbot.bot.SignalBot.stop] without
    blocking the caller.

    The non-blocking way to trigger shutdown from within a message
    handler; see [SignalBot.stop()][signalbot.bot.SignalBot.stop] for why
    calling it directly also works, just not without blocking the handler
    until shutdown completes.
    """
    # Keep a hard reference to the task so it isn't garbage-collected mid-run
    self._shutdown_task = self._event_loop.create_task(self.stop())

start 

start(*, run_forever: bool = True) -> None

Start the bot event loop and scheduler.

Parameters:

Name Type Description Default
run_forever bool

Whether to start the event loop or only add the task to it.

 True
Source code in src/signalbot/bot.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def start(self, *, run_forever: bool = True) -> None:
    """Start the bot event loop and scheduler.

    Args:
        run_forever: Whether to start the event loop or only add the task to it.
    """
    self.init_task = self._event_loop.create_task(
        rerun_on_exception(self._async_post_init, logger=self._logger),
    )

    if run_forever:
        self._install_sigint_sigterm_handlers()
        self.scheduler.start()

        self._event_loop.run_forever()

stop async 

stop() -> None

Gracefully stop the bot: cancel background tasks, close the shared HTTP session, and stop the event loop started by SignalBot.start().

Source code in src/signalbot/bot.py 
246
247
248
249
250
251
252
253
254
255
async def stop(self) -> None:
    """Gracefully stop the bot: cancel background tasks, close the shared
    HTTP session, and stop the event loop started by
    [SignalBot.start()][signalbot.bot.SignalBot.start].
    """
    if self.scheduler.running:
        self.scheduler.shutdown(wait=False)
    await self._pipeline.stop()
    await self.close()
    self._event_loop.stop()

wait_until_ready async 

wait_until_ready() -> None

Wait until the bot has finished connecting and is ready to send messages.

Source code in src/signalbot/bot.py 
257
258
259
260
261
262
263
async def wait_until_ready(self) -> None:
    """Wait until the bot has finished connecting and is ready to send messages."""
    if self.init_task is None:
        error_msg = "Bot is not initialized yet, call .start() first"
        raise SignalBotError(error_msg)

    await self.init_task