Bot

The Bot class is the central object of every FastAPI Interactions app. It owns the FastAPI instance, the command registry, and the interactions endpoint.

Usage

from fastapi_interactions import Bot

bot = Bot(
    app_id=123456789,
    public_key="your_public_key",
    bot_token="Bot your_token",
)

bot.load_commands("commands")

app = bot.app

Reference

fastapi_interactions.bot.Bot

Source code in fastapi_interactions/bot.py

class Bot:
    def __init__(
        self,
        app_id: int,
        public_key: str,
        bot_token: str,
        interactions_path: str = "/interactions",
    ):
        """Initialize a new Discord HTTP webhook bot.

        Args:
            app_id (int): The Discord Application ID.
            public_key (str): Public key used to validate incoming webhook signatures
                from Discord.
            bot_token (str): Bot token used for authenticated API calls to Discord
                (command registration, message sending, etc.).
            interactions_path (str, optional): The route path where the bot listens
                for interaction webhooks. Defaults to "/interactions".
        """
        self.app_id: int = app_id
        self.public_key: str = public_key
        self.bot_token: str = bot_token
        self.interactions_path: str = interactions_path
        self.base_url: str = f"https://discord.com/api/v10/applications/{app_id}"
        self.commands: dict[str, Command] = {}

        self.http = httpx.AsyncClient()

        self.app = FastAPI(openapi_url=None)
        self._register_routes()

    async def process_interactions(self, request: Request):
        payload = json.loads(request.state.raw_body)

        if payload["type"] == InteractionType.PING:
            return await PongResponse()

        if payload["type"] == InteractionType.APPLICATION_COMMAND:
            """Construct Context"""
            try:
                interaction = Interaction.model_validate(payload)
                application_command = ApplicationCommandData.model_validate(
                    interaction.data
                )
            except (ValidationError, Exception):
                # print(e.errors())
                return await MessageResponse("Unexpected error", ephemeral=True)

            context = Context(
                interaction=interaction, options=application_command, http=self.http
            )

            return await self.dispatch(
                command_name=application_command.name, ctx=context
            )

        return await MessageResponse("Unsupported interaction received", ephemeral=True)

    def _register_routes(self) -> None:
        """Set up middleware and the interactions endpoint.

        Registers the signature verification middleware and configures the route handler
        for the interactions endpoint. Called during initialization.
        """
        self.app.add_middleware(VerifySignatureMiddleware, public_key=self.public_key)
        self.app.add_api_route(
            self.interactions_path, endpoint=self.process_interactions, methods=["POST"]
        )

    def attach_router(self, router: CommandRouter) -> None:
        """Attach a CommandRouter to the bot.

        Registers a CommandRouter with this Discord webhook bot, enabling it to
        handle slash commands, message components, modals, and other interactions
        defined within the router.

        Args:
            router (CommandRouter): The router instance containing command
                registrations and handler mappings.

        Raises:
            TypeError: If the provided router is not an instance of CommandRouter.
        """
        if not isinstance(router, CommandRouter):
            raise TypeError(f"Expected a CommandRouter, got {type(router).__name__!r}")
        self.commands.update(router.commands)
        logger.info(
            f"Router {router.name!r} attached with {len(router)} commands - {str(router)}"
        )

    def __load_routers_from_module(self, module: types.ModuleType) -> None:
        """Discover and register routers from a module.

        Checks for an explicit __routers__ list first. If not found, scans the module
        for all CommandRouter instances and registers them automatically.

        Args:
            module: The module to scan for routers.

        """
        routers = getattr(module, "__routers__", None)

        if routers is None:
            routers = [
                obj for obj in vars(module).values() if isinstance(obj, CommandRouter)
            ]

        if not routers:
            logger.warning(f"No routers configured in {module.__name__!r}")
            return

        logger.debug(f"{len(routers)} routers discovered in {module.__name__!r}")
        for router in routers:
            self.attach_router(router)

    def load_extension(self, path: str) -> None:
        """Load extension(s) from a module or package.

        Dynamically imports the given Python path and registers all `CommandRouter`
        instances by calling the internal `__load_routers_from_module` method.

        Behavior:
            - If `path` points to a **module**: Loads routers from that single module.
            - If `path` points to a **package**: Recursively discovers and loads
              all non-package modules within it (and its subpackages).

        Args:
            path (str): Dot-separated import path to a module or package.

        Raises:
            ModuleNotFoundError: If the module or package does not exist.
            ImportError: If an error occurs while importing any module.

        Examples:
            Load a single module:
            ```python
            bot.load_extension("my_bot.extensions.moderation")
            ```
            Load all modules from a package(recursive):
            ```python
            bot.load_extension('my_bot.extensions')
            ```
        """
        module = importlib.import_module(path)
        if hasattr(module, "__path__"):
            # This is a package, lets recurisvely find modules
            logger.info(f"Scanning packages {path!r} for extensions")
            walked_packages = pkgutil.walk_packages(
                module.__path__, module.__name__ + "."
            )
            for _, module_name, is_package in walked_packages:
                if not is_package:
                    imported = importlib.import_module(module_name)
                    self.__load_routers_from_module(imported)
        else:
            self.__load_routers_from_module(module)

    def sync_commands(self) -> None:
        """Sync all registered commands with Discord's API.

        Sends a bulk overwrite of the bot's slash commands using the Discord
        Application Commands endpoint.

        Raises:
            Exception: If the API request fails.
        """
        global_payloads = []
        guild_payloads: dict[int, list] = {}

        for command in self.commands.values():
            if command.guild_id is not None:
                guild_payloads.setdefault(command.guild_id, []).append(
                    command.meta.as_payload()
                )
            else:
                global_payloads.append(command.meta.as_payload())

        if global_payloads:
            self.__put__commands(f"{self.base_url}/commands", global_payloads)

        for guild_id, payload in guild_payloads.items():
            self.__put__commands(f"{self.base_url}/guilds/{guild_id}/commands", payload)

    def __put__commands(self, url: str, payload: list) -> None:
        """Send a batch of commands to Discord's API.

        Makes a PUT request to the specified Discord endpoint with the command payload.
        Handles both global and guild-scoped command registration.

        Args:
            url: The Discord API endpoint (global or guild-scoped).
            payload: List of command definitions to register.

        Raises:
            Exception: If the HTTP response status is not 200.
        """
        headers = {"Authorization": f"Bot {self.bot_token}"}
        with httpx.Client() as client:
            response = client.put(url, headers=headers, json=payload)
        if response.status_code != 200:
            raise Exception(
                {"error": "registering commands failed", "data": response.json()}
            )
        logger.info(f"Synced {len(payload)} commands to {url!r}")

    async def dispatch(self, command_name: str, ctx: Context) -> InteractionResponse:
        """Invoke a command and return its response.

        Looks up the command by name and invokes its callback with the provided context.
        Option values are automatically bound to the callback's parameters.

        Args:
            command_name (str): The name of the command to invoke.
            ctx (Context): The interaction context.

        Returns:
            An InteractionResponse instance
        """
        command = self.commands.get(command_name)
        if command is None:
            return await MessageResponse("Unknown command", ephemeral=True)

        result = await call_with_options(command.callback, ctx)

        if not isinstance(result, InteractionResponse):
            result = MessageResponse(str(result))

        return await result

    def delete_all_commands(self, guild_id: Snowflake = None) -> None:
        """
        Delete all application commands by replacing the command set with an empty list.

        This performs a bulk overwrite operation. If ``guild_id`` is not
        provided, all global commands are deleted. Otherwise, all commands
        registered for the specified guild are deleted.

        Parameters
        ----------
        guild_id : Snowflake, optional
            The guild ID to target. If ``None``, the global command scope
            is used.

        Returns
        -------
        None
        """
        if not guild_id:
            url = f"{self.base_url}/commands"
        else:
            url = f"{self.base_url}/guilds/{guild_id}/commands"

        self.__put__commands(url, [])

init

__init__(app_id: int, public_key: str, bot_token: str, interactions_path: str = '/interactions')

Initialize a new Discord HTTP webhook bot.

Parameters:

Name	Type	Description	Default
`app_id`	`int`	The Discord Application ID.	required
`public_key`	`str`	Public key used to validate incoming webhook signatures from Discord.	required
`bot_token`	`str`	Bot token used for authenticated API calls to Discord (command registration, message sending, etc.).	required
`interactions_path`	`str`	The route path where the bot listens for interaction webhooks. Defaults to "/interactions".	`'/interactions'`

Source code in fastapi_interactions/bot.py

def __init__(
    self,
    app_id: int,
    public_key: str,
    bot_token: str,
    interactions_path: str = "/interactions",
):
    """Initialize a new Discord HTTP webhook bot.

    Args:
        app_id (int): The Discord Application ID.
        public_key (str): Public key used to validate incoming webhook signatures
            from Discord.
        bot_token (str): Bot token used for authenticated API calls to Discord
            (command registration, message sending, etc.).
        interactions_path (str, optional): The route path where the bot listens
            for interaction webhooks. Defaults to "/interactions".
    """
    self.app_id: int = app_id
    self.public_key: str = public_key
    self.bot_token: str = bot_token
    self.interactions_path: str = interactions_path
    self.base_url: str = f"https://discord.com/api/v10/applications/{app_id}"
    self.commands: dict[str, Command] = {}

    self.http = httpx.AsyncClient()

    self.app = FastAPI(openapi_url=None)
    self._register_routes()

__load_routers_from_module

__load_routers_from_module(module: ModuleType) -> None

Discover and register routers from a module.

Checks for an explicit routers list first. If not found, scans the module for all CommandRouter instances and registers them automatically.

Parameters:

Name	Type	Description	Default
`module`	`ModuleType`	The module to scan for routers.	required

Source code in fastapi_interactions/bot.py

def __load_routers_from_module(self, module: types.ModuleType) -> None:
    """Discover and register routers from a module.

    Checks for an explicit __routers__ list first. If not found, scans the module
    for all CommandRouter instances and registers them automatically.

    Args:
        module: The module to scan for routers.

    """
    routers = getattr(module, "__routers__", None)

    if routers is None:
        routers = [
            obj for obj in vars(module).values() if isinstance(obj, CommandRouter)
        ]

    if not routers:
        logger.warning(f"No routers configured in {module.__name__!r}")
        return

    logger.debug(f"{len(routers)} routers discovered in {module.__name__!r}")
    for router in routers:
        self.attach_router(router)

putcommands

__put__commands(url: str, payload: list) -> None

Send a batch of commands to Discord's API.

Makes a PUT request to the specified Discord endpoint with the command payload. Handles both global and guild-scoped command registration.

Parameters:

Name	Type	Description	Default
`url`	`str`	The Discord API endpoint (global or guild-scoped).	required
`payload`	`list`	List of command definitions to register.	required

Raises:

Type	Description
`Exception`	If the HTTP response status is not 200.

Source code in fastapi_interactions/bot.py

def __put__commands(self, url: str, payload: list) -> None:
    """Send a batch of commands to Discord's API.

    Makes a PUT request to the specified Discord endpoint with the command payload.
    Handles both global and guild-scoped command registration.

    Args:
        url: The Discord API endpoint (global or guild-scoped).
        payload: List of command definitions to register.

    Raises:
        Exception: If the HTTP response status is not 200.
    """
    headers = {"Authorization": f"Bot {self.bot_token}"}
    with httpx.Client() as client:
        response = client.put(url, headers=headers, json=payload)
    if response.status_code != 200:
        raise Exception(
            {"error": "registering commands failed", "data": response.json()}
        )
    logger.info(f"Synced {len(payload)} commands to {url!r}")

attach_router

attach_router(router: CommandRouter) -> None

Attach a CommandRouter to the bot.

Registers a CommandRouter with this Discord webhook bot, enabling it to handle slash commands, message components, modals, and other interactions defined within the router.

Parameters:

Name	Type	Description	Default
`router`	`CommandRouter`	The router instance containing command registrations and handler mappings.	required

Raises:

Type	Description
`TypeError`	If the provided router is not an instance of CommandRouter.

Source code in fastapi_interactions/bot.py

def attach_router(self, router: CommandRouter) -> None:
    """Attach a CommandRouter to the bot.

    Registers a CommandRouter with this Discord webhook bot, enabling it to
    handle slash commands, message components, modals, and other interactions
    defined within the router.

    Args:
        router (CommandRouter): The router instance containing command
            registrations and handler mappings.

    Raises:
        TypeError: If the provided router is not an instance of CommandRouter.
    """
    if not isinstance(router, CommandRouter):
        raise TypeError(f"Expected a CommandRouter, got {type(router).__name__!r}")
    self.commands.update(router.commands)
    logger.info(
        f"Router {router.name!r} attached with {len(router)} commands - {str(router)}"
    )

delete_all_commands

delete_all_commands(guild_id: Snowflake = None) -> None

Delete all application commands by replacing the command set with an empty list.

This performs a bulk overwrite operation. If guild_id is not provided, all global commands are deleted. Otherwise, all commands registered for the specified guild are deleted.

Parameters

guild_id : Snowflake, optional The guild ID to target. If None, the global command scope is used.

Returns

None

Source code in fastapi_interactions/bot.py

def delete_all_commands(self, guild_id: Snowflake = None) -> None:
    """
    Delete all application commands by replacing the command set with an empty list.

    This performs a bulk overwrite operation. If ``guild_id`` is not
    provided, all global commands are deleted. Otherwise, all commands
    registered for the specified guild are deleted.

    Parameters
    ----------
    guild_id : Snowflake, optional
        The guild ID to target. If ``None``, the global command scope
        is used.

    Returns
    -------
    None
    """
    if not guild_id:
        url = f"{self.base_url}/commands"
    else:
        url = f"{self.base_url}/guilds/{guild_id}/commands"

    self.__put__commands(url, [])

dispatch `async`

dispatch(command_name: str, ctx: Context) -> InteractionResponse

Invoke a command and return its response.

Looks up the command by name and invokes its callback with the provided context. Option values are automatically bound to the callback's parameters.

Parameters:

Name	Type	Description	Default
`command_name`	`str`	The name of the command to invoke.	required
`ctx`	`Context`	The interaction context.	required

Returns:

Type	Description
`InteractionResponse`	An InteractionResponse instance

Source code in fastapi_interactions/bot.py

async def dispatch(self, command_name: str, ctx: Context) -> InteractionResponse:
    """Invoke a command and return its response.

    Looks up the command by name and invokes its callback with the provided context.
    Option values are automatically bound to the callback's parameters.

    Args:
        command_name (str): The name of the command to invoke.
        ctx (Context): The interaction context.

    Returns:
        An InteractionResponse instance
    """
    command = self.commands.get(command_name)
    if command is None:
        return await MessageResponse("Unknown command", ephemeral=True)

    result = await call_with_options(command.callback, ctx)

    if not isinstance(result, InteractionResponse):
        result = MessageResponse(str(result))

    return await result

load_extension

load_extension(path: str) -> None

Load extension(s) from a module or package.

Dynamically imports the given Python path and registers all CommandRouter instances by calling the internal __load_routers_from_module method.

Behavior

If path points to a module: Loads routers from that single module.
If path points to a package: Recursively discovers and loads all non-package modules within it (and its subpackages).

Parameters:

Name	Type	Description	Default
`path`	`str`	Dot-separated import path to a module or package.	required

Raises:

Type	Description
`ModuleNotFoundError`	If the module or package does not exist.
`ImportError`	If an error occurs while importing any module.

Examples:

Load a single module:

bot.load_extension("my_bot.extensions.moderation")

Load all modules from a package(recursive):

bot.load_extension('my_bot.extensions')

Source code in fastapi_interactions/bot.py

def load_extension(self, path: str) -> None:
    """Load extension(s) from a module or package.

    Dynamically imports the given Python path and registers all `CommandRouter`
    instances by calling the internal `__load_routers_from_module` method.

    Behavior:
        - If `path` points to a **module**: Loads routers from that single module.
        - If `path` points to a **package**: Recursively discovers and loads
          all non-package modules within it (and its subpackages).

    Args:
        path (str): Dot-separated import path to a module or package.

    Raises:
        ModuleNotFoundError: If the module or package does not exist.
        ImportError: If an error occurs while importing any module.

    Examples:
        Load a single module:
        ```python
        bot.load_extension("my_bot.extensions.moderation")
        ```
        Load all modules from a package(recursive):
        ```python
        bot.load_extension('my_bot.extensions')
        ```
    """
    module = importlib.import_module(path)
    if hasattr(module, "__path__"):
        # This is a package, lets recurisvely find modules
        logger.info(f"Scanning packages {path!r} for extensions")
        walked_packages = pkgutil.walk_packages(
            module.__path__, module.__name__ + "."
        )
        for _, module_name, is_package in walked_packages:
            if not is_package:
                imported = importlib.import_module(module_name)
                self.__load_routers_from_module(imported)
    else:
        self.__load_routers_from_module(module)

sync_commands

sync_commands() -> None

Sync all registered commands with Discord's API.

Sends a bulk overwrite of the bot's slash commands using the Discord Application Commands endpoint.

Raises:

Type	Description
`Exception`	If the API request fails.

Source code in fastapi_interactions/bot.py

def sync_commands(self) -> None:
    """Sync all registered commands with Discord's API.

    Sends a bulk overwrite of the bot's slash commands using the Discord
    Application Commands endpoint.

    Raises:
        Exception: If the API request fails.
    """
    global_payloads = []
    guild_payloads: dict[int, list] = {}

    for command in self.commands.values():
        if command.guild_id is not None:
            guild_payloads.setdefault(command.guild_id, []).append(
                command.meta.as_payload()
            )
        else:
            global_payloads.append(command.meta.as_payload())

    if global_payloads:
        self.__put__commands(f"{self.base_url}/commands", global_payloads)

    for guild_id, payload in guild_payloads.items():
        self.__put__commands(f"{self.base_url}/guilds/{guild_id}/commands", payload)

fastapi_interactions.bot.call_with_options `async`

call_with_options(callback: callable, ctx: Context) -> Any

Invoke a command callback with option values bound to parameters.

Inspects the callback's signature and matches each parameter name to a registered option in the interaction context. Values are passed as keyword arguments to the callback. Parameters with default values are optional; missing required parameters raise an error.

Parameters:

Name	Type	Description	Default
`callback`	`callable`	The async command function to invoke.	required
`ctx`	`Context`	The interaction context containing option values.	required

Returns:

Type	Description
`Any`	The awaited result of the callback invocation.

Raises:

Type	Description
`TypeError`	If a required parameter has no matching option value in the context.

Source code in fastapi_interactions/bot.py

async def call_with_options(callback: callable, ctx: Context) -> Any:
    """Invoke a command callback with option values bound to parameters.

    Inspects the callback's signature and matches each parameter name to a registered
    option in the interaction context. Values are passed as keyword arguments to the
    callback. Parameters with default values are optional; missing required parameters
    raise an error.

    Args:
        callback: The async command function to invoke.
        ctx: The interaction context containing option values.

    Returns:
        The awaited result of the callback invocation.

    Raises:
        TypeError: If a required parameter has no matching option value in the context.
    """
    sig = inspect.signature(callback)
    kwargs = {}

    params = list(sig.parameters.values())[1:]
    for param in params:
        value = ctx.get_option_value(param.name)

        if value is not None:
            kwargs[param.name] = value
        elif param.default is not inspect.Parameter.empty:
            kwargs[param.name] = param.default
        else:
            raise TypeError(
                f"Command {callback.__name__!r} has required parameter "
                f"{param.name!r} but no matching option was provided"
            )

    return await callback(ctx, **kwargs)

Bot

Usage

Reference

fastapi_interactions.bot.Bot

__init__

__load_routers_from_module

__put__commands

attach_router

delete_all_commands

Parameters

Returns

dispatch async

load_extension

sync_commands

fastapi_interactions.bot.call_with_options async

init

putcommands

dispatch `async`

fastapi_interactions.bot.call_with_options `async`