Bot

The Bot class is the heart of every matrix.py application. It manages the connection to a Matrix homeserver, registers commands and event handlers, and drives the main event loop.

from matrix import Bot, Context

bot = Bot()


@bot.command("ping")
async def ping(ctx: Context):
    await ctx.reply("Pong!")


bot.start(config="config.yml")

matrix.bot.Bot

Bot(*, help_=None)

Bases: Registry

The base class defining a Matrix bot.

This class manages the connection to a Matrix homeserver, listens for events, and dispatches them to registered handlers. It also supports a command system with decorators for easy registration.

Source code in matrix/bot.py

def __init__(
    self,
    *,
    help_: Optional[HelpCommand] = None,
) -> None:
    super().__init__(self.__class__.__name__)

    self._config: Config | None = None
    self._client: AsyncClient | None = None
    self._synced: asyncio.Event = asyncio.Event()
    self._help: HelpCommand | None = help_

    self.extensions: dict[str, Extension] = {}
    self.scheduler: Scheduler = Scheduler()
    self.log: logging.Logger = logging.getLogger(__name__)
    self.start_at: float | None = None

extensions instance-attribute

extensions = {}

scheduler instance-attribute

scheduler = Scheduler()

log instance-attribute

log = getLogger(__name__)

start_at instance-attribute

start_at = None

client property

client

config property

config

help property

help

get_room

get_room(room_id)

Retrieve a Room instance by its Matrix room ID.

Returns the Room object corresponding to room_id if it exists in the client's known rooms. Returns None if the room cannot be found.

Example

room = bot.get_room("!abc123:matrix.org")
if room:
    print(room.name)

Source code in matrix/bot.py

def get_room(self, room_id: str) -> Room | None:
    """Retrieve a `Room` instance by its Matrix room ID.

    Returns the `Room` object corresponding to `room_id` if it exists in
    the client's known rooms. Returns `None` if the room cannot be found.

    ## Example

    ```python
    room = bot.get_room("!abc123:matrix.org")
    if room:
        print(room.name)
    ```
    """
    if matrix_room := self.client.rooms.get(room_id):
        return Room(matrix_room=matrix_room, client=self.client)
    return None

load_extension

load_extension(extension)

Source code in matrix/bot.py

def load_extension(self, extension: Extension) -> None:
    self.log.debug(f"Loading extension: '{extension.name}'")

    if extension.name in self.extensions:
        raise AlreadyRegisteredError(extension)

    for cmd in extension._commands.values():
        if isinstance(cmd, Group):
            self.register_group(cmd)
        else:
            self.register_command(cmd)

    for event_type, handlers in extension._event_handlers.items():
        self._event_handlers[event_type].extend(handlers)

    for hook_name, handlers in extension._hook_handlers.items():
        self._hook_handlers[hook_name].extend(handlers)

    self._checks.extend(extension._checks)
    self._error_handlers.update(extension._error_handlers)
    self._command_error_handlers.update(extension._command_error_handlers)

    for job in extension._scheduler.jobs:
        self.scheduler.scheduler.add_job(
            job.func,
            trigger=job.trigger,
            name=job.name,
        )

    self.extensions[extension.name] = extension
    extension.load(self)
    self.log.debug("loaded extension '%s'", extension.name)

unload_extension

unload_extension(ext_name)

Source code in matrix/bot.py

def unload_extension(self, ext_name: str) -> None:
    self.log.debug("Unloading extension: '%s'", ext_name)

    extension = self.extensions.pop(ext_name, None)
    if extension is None:
        raise ValueError(f"No extension named '{ext_name}' is loaded")

    for cmd_name in extension._commands:
        self._commands.pop(cmd_name, None)

    for event_type, handlers in extension._event_handlers.items():
        for handler in handlers:
            self._event_handlers[event_type].remove(handler)

    for check in extension._checks:
        self._checks.remove(check)

    for exc_type in extension._error_handlers:
        self._error_handlers.pop(exc_type, None)

    for exc_type in extension._command_error_handlers:
        self._command_error_handlers.pop(exc_type, None)

    for job in extension._scheduler.jobs:
        bot_job = next((j for j in self.scheduler.jobs if j.func is job.func), None)
        if bot_job:
            bot_job.remove()

    extension.unload()
    self.log.debug("unloaded extension '%s'", ext_name)

on_ready async

on_ready()

Override this in a subclass.

Source code in matrix/bot.py

async def on_ready(self) -> None:
    """Override this in a subclass."""
    pass

on_error async

on_error(error)

Override this in a subclass.

Source code in matrix/bot.py

async def on_error(self, error: Exception) -> None:
    """Override this in a subclass."""
    self.log.exception("Unhandled error: '%s'", error)

on_command async

on_command(_ctx)

Override this in a subclass.

Source code in matrix/bot.py

async def on_command(self, _ctx: Context) -> None:
    """Override this in a subclass."""
    pass

on_command_error async

on_command_error(_ctx, error)

Override this in a subclass.

Source code in matrix/bot.py

async def on_command_error(self, _ctx: Context, error: Exception) -> None:
    """Override this in a subclass."""
    self.log.exception("Unhandled error: '%s'", error)

start

start(*, config)

Synchronous entry point for running the bot.

This is a convenience wrapper that allows running the bot like a script using a blocking call. It internally calls :meth:run within :func:asyncio.run, and ensures the client is closed gracefully on interruption.

Source code in matrix/bot.py

def start(self, *, config: Config | str) -> None:
    """
    Synchronous entry point for running the bot.

    This is a convenience wrapper that allows running the bot like a
    script using a blocking call. It internally calls :meth:`run` within
    :func:`asyncio.run`, and ensures the client is closed gracefully
    on interruption.
    """
    self._load_config(config)

    try:
        asyncio.run(self.run())
    except KeyboardInterrupt:
        self.log.info("bot interrupted by user")
    finally:
        asyncio.run(self.client.close())

run async

run()

Log in to the Matrix homeserver and begin syncing events.

This method should be used within an asynchronous context, typically via :func:asyncio.run. It handles authentication, calls the :meth:on_ready hook, and starts the long-running sync loop for receiving events.

Source code in matrix/bot.py

async def run(self) -> None:
    """
    Log in to the Matrix homeserver and begin syncing events.

    This method should be used within an asynchronous context,
    typically via :func:`asyncio.run`. It handles authentication,
    calls the :meth:`on_ready` hook, and starts the long-running
    sync loop for receiving events.
    """
    self.client.user = self.config.username

    self.start_at = time.time()
    self.log.info("starting – timestamp=%s", self.start_at)

    if self.config.token:
        self.client.access_token = self.config.token
    else:
        login_resp = await self.client.login(self.config.password)
        self.log.info("logged in: %s", login_resp)

    sync_task = asyncio.create_task(self.client.sync_forever(timeout=30_000))

    await self._wait_until_synced()
    await self._on_ready()

    self.scheduler.start()
    await sync_task

on_message async

on_message(room, event)

Source code in matrix/bot.py

async def on_message(self, room: Room, event: Event) -> None:
    await self._process_commands(room, event)