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 = logging.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. Returns a typed subclass if the room type is registered (e.g. Space for m.space rooms).

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.
    Returns a typed subclass if the room type is registered (e.g. `Space` for m.space rooms).

    ## Example

    ```python
    room = bot.get_room("!abc123:matrix.org")

    if room:
        print(room.name)
    ```
    """
    if matrix_room := self.client.rooms.get(room_id):
        return make_room(matrix_room, self.client)
    return None

get_rooms

get_rooms()

Retrieve a list of all rooms the bot is aware of.

This method returns a list of Room objects for all rooms currently known to the client. This includes both regular rooms and spaces; spaces are returned as Space instances.

Example

rooms = bot.get_rooms()

for room in rooms:
    print(room.name)

Source code in matrix/bot.py

def get_rooms(self) -> list[Room]:
    """Retrieve a list of all rooms the bot is aware of.

    This method returns a list of `Room` objects for all rooms currently
    known to the client. This includes both regular rooms and spaces;
    spaces are returned as `Space` instances.

    ## Example

    ```python
    rooms = bot.get_rooms()

    for room in rooms:
        print(room.name)
    ```
    """
    rooms = []

    for matrix_room in self.client.rooms.values():
        rooms.append(make_room(matrix_room, self.client))

    return rooms

get_space

get_space(space_id)

Retrieve a Space instance by its Matrix room ID.

Returns the Space object corresponding to space_id if it exists in the client's known rooms and is a space. Returns None otherwise.

Example

space = bot.get_space("!space123:matrix.org")

if space:
    print(space.name)

Source code in matrix/bot.py

def get_space(self, space_id: str) -> Space | None:
    """Retrieve a `Space` instance by its Matrix room ID.

    Returns the `Space` object corresponding to `space_id` if it exists in
    the client's known rooms and is a space. Returns `None` otherwise.

    ## Example

    ```python
    space = bot.get_space("!space123:matrix.org")

    if space:
        print(space.name)
    ```
    """
    room = self.get_room(space_id)
    return room if isinstance(room, Space) else None

get_spaces

get_spaces()

Retrieve a list of all spaces the bot is aware of.

This method returns a list of Space objects for all rooms currently known to the client that are identified as spaces.

Example

spaces = bot.get_spaces()

for space in spaces:
    print(space.name)

Source code in matrix/bot.py

def get_spaces(self) -> list[Space]:
    """Retrieve a list of all spaces the bot is aware of.

    This method returns a list of `Space` objects for all rooms currently
    known to the client that are identified as spaces.

    ## Example

    ```python
    spaces = bot.get_spaces()

    for space in spaces:
        print(space.name)
    ```
    """
    return [room for room in self.get_rooms() if isinstance(room, Space)]

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 matrix_call(
            self.client.login(self.config.password),
            error_message="Failed to log in",
        )
        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)