Command

A Command wraps an async callback together with its metadata (name, description, usage, cooldown, and checks). Commands are normally created via the @bot.command() decorator rather than instantiated directly.

from matrix import Bot, Context

bot = Bot()

@bot.command("greet", description="Greet a user by name", usage="<name>")
async def greet(ctx: Context, name: str):
    await ctx.reply(f"Hello, {name}!")

matrix.command.Command

Command(func, *, name=None, description=None, prefix=None, parent=None, usage=None, cooldown=None)

Represents a command that can be executed with a context and arguments.

Source code in matrix/command.py

def __init__(
    self,
    func: Callback,
    *,
    name: Optional[str] = None,
    description: Optional[str] = None,
    prefix: Optional[str] = None,
    parent: Optional[str] = None,
    usage: Optional[str] = None,
    cooldown: Optional[tuple[int, float]] = None,
):
    if name is not None and not isinstance(name, str):
        raise TypeError("Name must be a string.")

    self.name: str = name or func.__name__
    self.callback = func
    self.checks: List[Callback] = []

    self.description: str = description or ""
    self.prefix: str = prefix or ""
    self.parent: str = parent or ""
    self.usage: str = usage or self._build_usage()
    self.help: str = self._build_help()

    self._before_invoke_callback: Optional[Callback] = None
    self._after_invoke_callback: Optional[Callback] = None
    self._on_error: Optional[ErrorCallback] = None
    self._error_handlers: dict[type[Exception], ErrorCallback] = {}

    self.cooldown_rate: Optional[int] = None
    self.cooldown_period: Optional[float] = None
    self.cooldown_calls: DefaultDict[str, deque[float]] = defaultdict(deque)

    if cooldown:
        self.set_cooldown(*cooldown)

name instance-attribute

name = name or __name__

checks instance-attribute

checks = []

description instance-attribute

description = description or ''

prefix instance-attribute

prefix = prefix or ''

parent instance-attribute

parent = parent or ''

usage instance-attribute

usage = usage or _build_usage()

help instance-attribute

help = _build_help()

cooldown_rate instance-attribute

cooldown_rate = None

cooldown_period instance-attribute

cooldown_period = None

cooldown_calls instance-attribute

cooldown_calls = defaultdict(deque)

callback property writable

callback

Returns the coroutine function for this command.

check

check(func)

Register a check callback

Example

@bot.command("secret")
async def secret(ctx: Context) -> None:
    await ctx.reply("Access granted!")

@secret.check
async def is_allowed(ctx: Context) -> bool:
    return ctx.sender == "@admin:matrix.org"

Source code in matrix/command.py

def check(self, func: Callback) -> None:
    """
    Register a check callback

    ## Example

    ```python
    @bot.command("secret")
    async def secret(ctx: Context) -> None:
        await ctx.reply("Access granted!")

    @secret.check
    async def is_allowed(ctx: Context) -> bool:
        return ctx.sender == "@admin:matrix.org"
    ```
    """
    if not inspect.iscoroutinefunction(func):
        raise TypeError("Checks must be coroutine")

    self.checks.append(func)

set_cooldown

set_cooldown(rate, period)

Source code in matrix/command.py

def set_cooldown(self, rate: int, period: float) -> None:
    self.cooldown_rate = rate
    self.cooldown_period = period

    async def cooldown_function(ctx: "Context") -> bool:
        if ctx is None or not hasattr(ctx, "sender"):
            return False

        if self.cooldown_period is None or self.cooldown_rate is None:
            return False

        now = monotonic()
        user_id = ctx.sender
        calls = self.cooldown_calls[user_id]

        while calls and now - calls[0] >= self.cooldown_period:
            calls.popleft()

        if len(calls) >= self.cooldown_rate:
            retry = self.cooldown_period - (now - calls[0])
            raise CooldownError(self, cooldown_function, retry)

        calls.append(now)
        return True

    self.checks.append(cooldown_function)

before_invoke

before_invoke(func)

Registers a coroutine to be called before the command is invoked.

Example

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

@ping.before_invoke
async def before_ping(ctx: Context) -> None:
    print(f"ping invoked by {ctx.sender}")

Source code in matrix/command.py

def before_invoke(self, func: Callback) -> None:
    """
    Registers a coroutine to be called before the command is invoked.

    ## Example

    ```python
    @bot.command("ping")
    async def ping(ctx: Context) -> None:
        await ctx.reply("Pong!")

    @ping.before_invoke
    async def before_ping(ctx: Context) -> None:
        print(f"ping invoked by {ctx.sender}")
    ```
    """

    if not inspect.iscoroutinefunction(func):
        raise TypeError("The hook must be a coroutine.")

    self._before_invoke_callback = func

after_invoke

after_invoke(func)

Registers a coroutine to be called after the command is invoked.

Example

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

@ping.after_invoke
async def after_ping(ctx: Context) -> None:
    print(f"ping completed for {ctx.sender}")

Source code in matrix/command.py

def after_invoke(self, func: Callback) -> None:
    """
    Registers a coroutine to be called after the command is invoked.

    ## Example

    ```python
    @bot.command("ping")
    async def ping(ctx: Context) -> None:
        await ctx.reply("Pong!")

    @ping.after_invoke
    async def after_ping(ctx: Context) -> None:
        print(f"ping completed for {ctx.sender}")
    ```
    """

    if not inspect.iscoroutinefunction(func):
        raise TypeError("The hook must be a coroutine.")

    self._after_invoke_callback = func

error

error(exception=None)

Decorator used to register an error handler for this command.

Example

@bot.command("div")
async def div(ctx: Context, a: int, b: int) -> None:
    await ctx.reply(f"{a / b}")

@div.error(ZeroDivisionError)
async def div_error(ctx: Context, error: ZeroDivisionError) -> None:
    await ctx.reply("Cannot divide by zero!")

@div.error(MissingArgumentError)
async def div_missing(ctx: Context, error: MissingArgumentError) -> None:
    await ctx.reply(f"Missing argument: {error}")

Source code in matrix/command.py

def error(self, exception: Optional[type[Exception]] = None) -> Callable:
    """
    Decorator used to register an error handler for this command.

    ## Example

    ```python
    @bot.command("div")
    async def div(ctx: Context, a: int, b: int) -> None:
        await ctx.reply(f"{a / b}")

    @div.error(ZeroDivisionError)
    async def div_error(ctx: Context, error: ZeroDivisionError) -> None:
        await ctx.reply("Cannot divide by zero!")

    @div.error(MissingArgumentError)
    async def div_missing(ctx: Context, error: MissingArgumentError) -> None:
        await ctx.reply(f"Missing argument: {error}")
    ```
    """

    def wrapper(func: ErrorCallback) -> Callable:
        if not inspect.iscoroutinefunction(func):
            raise TypeError("The error handler must be a coroutine.")

        if exception:
            self._error_handlers[exception] = func
        else:
            self._on_error = func
        return func

    return wrapper

on_error async

on_error(ctx, error)

Executes the registered error handler if present.

Source code in matrix/command.py

async def on_error(self, ctx: "Context", error: Exception) -> None:
    """
    Executes the registered error handler if present.
    """

    if handler := self._error_handlers.get(type(error)):
        await handler(ctx, error)
        return

    await ctx.bot.on_command_error(ctx, error)

    if self._on_error:
        await self._on_error(ctx, error)
    else:
        await ctx.send_help()

    ctx.logger.exception("error while executing command '%s'", self)

invoke async

invoke(ctx)

Source code in matrix/command.py

async def invoke(self, ctx: "Context") -> None:
    parsed_args = self._parse_arguments(ctx)
    await self.callback(ctx, *parsed_args)