API Documentation

This is the API documentation of all the classes and functions relevant for Plugin development. The rest of the code deals with the internal workings of Slack Machine and is very much an implementation detail and subject to change. Therefore it is not documented.

Plugin classes

The following classes form the basis for Plugin development.

`machine.plugins.base.MachineBasePlugin`

Base class for all Slack Machine plugins

The purpose of this class is two-fold:

It acts as a marker-class so Slack Machine can recognize plugins as such
It provides a lot of common functionality and convenience methods for plugins to interact with channels and users

Attributes:

Name	Type	Description
`settings`	`CaseInsensitiveDict`	Slack Machine settings object that contains all settings that were defined through `local_settings.py` Plugin developers can use any settings that are defined by the user, and ask users to add new settings specifically for their plugin.

Source code in machine/plugins/base.py

class MachineBasePlugin:
    """Base class for all Slack Machine plugins

    The purpose of this class is two-fold:

    1. It acts as a marker-class so Slack Machine can recognize plugins as such
    2. It provides a lot of common functionality and convenience methods for plugins to
       interact with channels and users

    :var settings: Slack Machine settings object that contains all settings that
        were defined through ``local_settings.py`` Plugin developers can use any
        settings that are defined by the user, and ask users to add new settings
        specifically for their plugin.
    """

    _client: SlackClient
    storage: PluginStorage
    settings: CaseInsensitiveDict
    _fq_name: str

    def __init__(self, client: SlackClient, settings: CaseInsensitiveDict, storage: PluginStorage):
        self._client = client
        self.storage = storage
        self.settings = settings
        self._fq_name = f"{self.__module__}.{self.__class__.__name__}"

    async def init(self) -> None:
        """Initialize plugin

        This method can be implemented by concrete plugin classes. It will be called **once**
        for each plugin, when that plugin is first loaded. You can refer to settings via
        ``self.settings``, and access storage through ``self.storage``, but the Slack client has
        not been initialized yet, so you cannot send or process messages during initialization.

        :return: None
        """
        return None

    @property
    def users(self) -> dict[str, User]:
        """Dictionary of all users in the Slack workspace

        :return: a dictionary of all users in the Slack workspace, where the key is the user id and
            the value is a [`User`][machine.models.user.User] object
        """
        return self._client.users

    @property
    def users_by_email(self) -> dict[str, User]:
        """Dictionary of all users in the Slack workspace by email

        **Note**: not every user might have an email address in their profile, so this
        dictionary might not contain all users in the Slack workspace

        :return: a dictionary of all users in the Slack workspace, where the key is the email and
            the value is a [`User`][machine.models.user.User] object
        """
        return self._client.users

    @property
    def channels(self) -> dict[str, Channel]:
        """List of all channels in the Slack workspace

        This is a list of all channels in the Slack workspace that the bot is aware of. This
        includes all public channels, all private channels the bot is a member of and all DM
        channels the bot is a member of.

        :return: a list of all channels in the Slack workspace, where each channel is a
            :py:class:`~machine.models.channel.Channel` object
        """
        return self._client.channels

    @property
    def web_client(self) -> AsyncWebClient:
        """Slack SDK web client to access the [Slack Web API][slack-web-api]

        This property references an instance of [`AsyncWebClient`][async-web-client]

        [slack-web-api]: https://api.slack.com/web
        [async-web-client]: https://slack.dev/python-slack-sdk/api-docs/slack_sdk/web/async_client.html#slack_sdk.web.async_client.AsyncWebClient
        """  # noqa: E501
        return self._client.web_client

    def find_channel_by_name(self, channel_name: str) -> Channel | None:
        """Find a channel by its name, irrespective of a preceding pound symbol. This does not include DMs.

        :param channel_name: The name of the channel to retrieve.
        :return: The channel if found, None otherwise.
        """
        if channel_name.startswith("#"):
            channel_name = channel_name[1:]
        for c in self.channels.values():
            if c.name_normalized and channel_name.lower() == c.name_normalized.lower():
                return c
        return None

    def get_user_by_id(self, user_id: str) -> User | None:
        """Get a user by their ID.

        :param user_id: The ID of the user to retrieve.
        :return: The user if found, None otherwise.
        """
        return self.users.get(user_id)

    def get_user_by_email(self, email: str) -> User | None:
        """Get a user by their email address.

        :param email: The email address of the user to retrieve.
        :return: The user if found, None otherwise.
        """
        return self._client.get_user_by_email(email)

    @property
    def bot_info(self) -> dict[str, Any]:
        """Information about the bot user in Slack

        This will return a dictionary with information about the bot user in Slack that represents
        Slack Machine

        :return: Bot user
        """
        return self._client.bot_info

    def at(self, user: User) -> str:
        """Create a mention of the provided user

        Create a mention of the provided user in the form of ``<@[user_id]>``. This method is
        convenient when you want to include mentions in your message. This method does not send
        a message, but should be used together with methods like
        :py:meth:`~machine.plugins.base.MachineBasePlugin.say`

        :param user: user your want to mention
        :return: user mention
        """
        return user.fmt_mention()

    async def say(
        self,
        channel: Channel | str,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        thread_ts: str | None = None,
        ephemeral_user: User | str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a message to a channel

        Send a message to a channel using the WebAPI. Allows for rich formatting using
        `blocks`_ and/or `attachments`_. You can provide blocks and attachments as Python dicts or
        you can use the `convenient classes`_ that the underlying slack client provides.
        Can also reply in-thread and send ephemeral messages, visible to only one user.
        Ephemeral messages and threaded messages are mutually exclusive, and ``ephemeral_user``
        takes precedence over ``thread_ts``
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
        `chat.postEphemeral`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :param ephemeral_user: optional user name or id if the message needs to visible
            to a specific user only
        :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
            if `ephemeral_user` is True.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
        """
        return await self._client.send(
            channel,
            text=text,
            attachments=attachments,
            blocks=blocks,
            thread_ts=thread_ts,
            ephemeral_user=ephemeral_user,
            **kwargs,
        )

    async def say_scheduled(
        self,
        when: datetime,
        channel: Channel | str,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        thread_ts: str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a message to a channel

        This is the scheduled version of :py:meth:`~machine.plugins.base.MachineBasePlugin.say`.
        It behaves the same, but will send the message at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        return await self._client.send_scheduled(
            when,
            channel,
            text=text,
            attachments=attachments,
            blocks=blocks,
            thread_ts=thread_ts,
            **kwargs,
        )

    async def update(
        self,
        channel: Channel | str,
        ts: str,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        ephemeral_user: User | str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Update an existing message

        Update an existing message using the WebAPI. Allows for rich formatting using
        [blocks] and/or [attachments]. You can provide blocks and attachments as Python dicts or
        you can use the [convenient classes] that the underlying slack client provides.
        Can also update in-thread and ephemeral messages, visible to only one user.
        Any extra kwargs you provide, will be passed on directly to the [`chat.update`][chat_update] request.

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes
        [chat_update]: https://api.slack.com/methods/chat.update

        :param channel: [`Channel`][machine.models.channel.Channel] object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param ts: timestamp of the message to be updated.
        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :param ephemeral_user: optional user name or id if the message needs to visible
            to a specific user only
        :return: Dictionary deserialized from [`chat.update`](https://api.slack.com/methods/chat.update) request


        """
        return await self._client.update(
            channel,
            ts=ts,
            text=text,
            attachments=attachments,
            blocks=blocks,
            ephemeral_user=ephemeral_user,
            **kwargs,
        )

    async def delete(
        self,
        channel: Channel | str,
        ts: str,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Delete an existing message

        Delete an existing message using the WebAPI.
        Any extra kwargs you provide, will be passed on directly to the [`chat.delete`][chat_delete] request.

        [chat_delete]: https://api.slack.com/methods/chat.delete

        :param channel: [`Channel`][machine.models.channel.Channel] object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param ts: timestamp of the message to be deleted.
        :return: Dictionary deserialized from [`chat.delete`](https://api.slack.com/methods/chat.delete) request


        """
        return await self._client.delete(
            channel,
            ts=ts,
            **kwargs,
        )

    async def react(self, channel: Channel | str, ts: str, emoji: str) -> AsyncSlackResponse:
        """React to a message in a channel

        Add a reaction to a message in a channel. What message to react to, is determined by the
        combination of the channel and the timestamp of the message.

        :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
            message to. Can be public or private (group) channel, or DM channel.
        :param ts: timestamp of the message to react to
        :param emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)
        :return: Dictionary deserialized from `reactions.add`_ request.

        .. _reactions.add: https://api.slack.com/methods/reactions.add
        """
        return await self._client.react(channel, ts, emoji)

    async def send_dm(
        self,
        user: User | str,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a Direct Message

        Send a Direct Message to a user by opening a DM channel and sending a message to it. Allows
        for rich formatting using `blocks`_ and/or `attachments`_. You can provide blocks and
        attachments as Python dicts or you can use the `convenient classes`_ that the underlying
        slack client provides.
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param user: :py:class:`~machine.models.user.User` object or id of user to send DM to.
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :return: Dictionary deserialized from `chat.postMessage`_ request.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        """
        return await self._client.send_dm(user, text, attachments=attachments, blocks=blocks, **kwargs)

    async def send_dm_scheduled(
        self,
        when: datetime,
        user: User | str,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a Direct Message

        This is the scheduled version of
        :py:meth:`~machine.plugins.base.MachineBasePlugin.send_dm`. It behaves the same, but
        will send the DM at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param user: :py:class:`~machine.models.user.User` object or id of user to send DM to.
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        return await self._client.send_dm_scheduled(
            when, user, text=text, attachments=attachments, blocks=blocks, **kwargs
        )

    def emit(self, event: str, **kwargs: Any) -> None:
        """Emit an event

        Emit an event that plugins can listen for. You can include arbitrary data as keyword
        arguments.

        :param event: name of the event
        :param **kwargs: any data you want to emit with the event
        :return: None
        """
        ee.emit(event, self, **kwargs)

    async def pin_message(self, channel: Channel | str, ts: str) -> AsyncSlackResponse:
        """Pin message

        Pin a message in a channel

        :param channel: channel to pin the message in
        :param ts: timestamp of the message to pin
        :return: response from the Slack Web API
        """
        return await self._client.pin_message(channel, ts)

    async def unpin_message(self, channel: Channel | str, ts: str) -> AsyncSlackResponse:
        """Unpin message

        Unpin a message that was previously pinned in a channel

        :param channel: channel where the message is pinned that needs to be unpinned
        :param ts: timestamp of the message to unpin
        :return: response from the Slack Web API
        """
        return await self._client.unpin_message(channel, ts)

    async def set_topic(self, channel: Channel | str, topic: str, **kwargs: Any) -> AsyncSlackResponse:
        """Set channel topic

        Set or update topic for the channel

        :param channel: channel where topic needs to be set or updated
        :param topic: topic for the channel (slack does not support formatting for topics)
        :return: response from the Slack Web API
        """
        return await self._client.set_topic(channel, topic, **kwargs)

`users: dict[str, User]` `property`

Dictionary of all users in the Slack workspace

Returns:

Type	Description
`dict[str, User]`	a dictionary of all users in the Slack workspace, where the key is the user id and the value is a `User` object

`users_by_email: dict[str, User]` `property`

Dictionary of all users in the Slack workspace by email

Note: not every user might have an email address in their profile, so this dictionary might not contain all users in the Slack workspace

Returns:

Type	Description
`dict[str, User]`	a dictionary of all users in the Slack workspace, where the key is the email and the value is a `User` object

`channels: dict[str, Channel]` `property`

List of all channels in the Slack workspace

This is a list of all channels in the Slack workspace that the bot is aware of. This includes all public channels, all private channels the bot is a member of and all DM channels the bot is a member of.

Returns:

Type	Description
`dict[str, Channel]`	a list of all channels in the Slack workspace, where each channel is a :py:class:`~machine.models.channel.Channel` object

`web_client: AsyncWebClient` `property`

Slack SDK web client to access the Slack Web API

This property references an instance of AsyncWebClient

`bot_info: dict[str, Any]` `property`

Information about the bot user in Slack

This will return a dictionary with information about the bot user in Slack that represents Slack Machine

Returns:

Type	Description
`dict[str, Any]`	Bot user

`init() -> None` `async`

Initialize plugin

This method can be implemented by concrete plugin classes. It will be called once for each plugin, when that plugin is first loaded. You can refer to settings via self.settings, and access storage through self.storage, but the Slack client has not been initialized yet, so you cannot send or process messages during initialization.

Returns:

Type	Description
`None`	None

Source code in machine/plugins/base.py

async def init(self) -> None:
    """Initialize plugin

    This method can be implemented by concrete plugin classes. It will be called **once**
    for each plugin, when that plugin is first loaded. You can refer to settings via
    ``self.settings``, and access storage through ``self.storage``, but the Slack client has
    not been initialized yet, so you cannot send or process messages during initialization.

    :return: None
    """
    return None

`find_channel_by_name(channel_name: str) -> Channel | None`

Find a channel by its name, irrespective of a preceding pound symbol. This does not include DMs.

Parameters:

Name	Type	Description	Default
`channel_name`	`str`	The name of the channel to retrieve.	required

Returns:

Type	Description
`Channel \| None`	The channel if found, None otherwise.

Source code in machine/plugins/base.py

def find_channel_by_name(self, channel_name: str) -> Channel | None:
    """Find a channel by its name, irrespective of a preceding pound symbol. This does not include DMs.

    :param channel_name: The name of the channel to retrieve.
    :return: The channel if found, None otherwise.
    """
    if channel_name.startswith("#"):
        channel_name = channel_name[1:]
    for c in self.channels.values():
        if c.name_normalized and channel_name.lower() == c.name_normalized.lower():
            return c
    return None

`get_user_by_id(user_id: str) -> User | None`

Get a user by their ID.

Parameters:

Name	Type	Description	Default
`user_id`	`str`	The ID of the user to retrieve.	required

Returns:

Type	Description
`User \| None`	The user if found, None otherwise.

Source code in machine/plugins/base.py

def get_user_by_id(self, user_id: str) -> User | None:
    """Get a user by their ID.

    :param user_id: The ID of the user to retrieve.
    :return: The user if found, None otherwise.
    """
    return self.users.get(user_id)

`get_user_by_email(email: str) -> User | None`

Get a user by their email address.

Parameters:

Name	Type	Description	Default
`email`	`str`	The email address of the user to retrieve.	required

Returns:

Type	Description
`User \| None`	The user if found, None otherwise.

Source code in machine/plugins/base.py

def get_user_by_email(self, email: str) -> User | None:
    """Get a user by their email address.

    :param email: The email address of the user to retrieve.
    :return: The user if found, None otherwise.
    """
    return self._client.get_user_by_email(email)

`at(user: User) -> str`

Create a mention of the provided user

Create a mention of the provided user in the form of <@[user_id]>. This method is convenient when you want to include mentions in your message. This method does not send a message, but should be used together with methods like :py:meth:~machine.plugins.base.MachineBasePlugin.say

Parameters:

Name	Type	Description	Default
`user`	`User`	user your want to mention	required

Returns:

Type	Description
`str`	user mention

Source code in machine/plugins/base.py

def at(self, user: User) -> str:
    """Create a mention of the provided user

    Create a mention of the provided user in the form of ``<@[user_id]>``. This method is
    convenient when you want to include mentions in your message. This method does not send
    a message, but should be used together with methods like
    :py:meth:`~machine.plugins.base.MachineBasePlugin.say`

    :param user: user your want to mention
    :return: user mention
    """
    return user.fmt_mention()

`say(channel: Channel | str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, ephemeral_user: User | str | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

Send a message to a channel

Send a message to a channel using the WebAPI. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Can also reply in-thread and send ephemeral messages, visible to only one user. Ephemeral messages and threaded messages are mutually exclusive, and ephemeral_user takes precedence over thread_ts Any extra kwargs you provide, will be passed on directly to the chat.postMessage or chat.postEphemeral_ request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
`channel`	`Channel \| str`	:py:class:`~machine.models.channel.Channel` object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
`text`	`str \| None`	message text	`None`
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`
`thread_ts`	`str \| None`	optional timestamp of thread, to send a message in that thread	`None`
`ephemeral_user`	`User \| str \| None`	optional user name or id if the message needs to visible to a specific user only	`None`

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `chat.postMessage` request, or `chat.postEphemeral` if `ephemeral_user` is True. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral

Source code in machine/plugins/base.py

async def say(
    self,
    channel: Channel | str,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    thread_ts: str | None = None,
    ephemeral_user: User | str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a message to a channel

    Send a message to a channel using the WebAPI. Allows for rich formatting using
    `blocks`_ and/or `attachments`_. You can provide blocks and attachments as Python dicts or
    you can use the `convenient classes`_ that the underlying slack client provides.
    Can also reply in-thread and send ephemeral messages, visible to only one user.
    Ephemeral messages and threaded messages are mutually exclusive, and ``ephemeral_user``
    takes precedence over ``thread_ts``
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
    `chat.postEphemeral`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :param ephemeral_user: optional user name or id if the message needs to visible
        to a specific user only
    :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
        if `ephemeral_user` is True.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
    """
    return await self._client.send(
        channel,
        text=text,
        attachments=attachments,
        blocks=blocks,
        thread_ts=thread_ts,
        ephemeral_user=ephemeral_user,
        **kwargs,
    )

`say_scheduled(when: datetime, channel: Channel | str, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

Schedule a message to a channel

This is the scheduled version of :py:meth:~machine.plugins.base.MachineBasePlugin.say. It behaves the same, but will send the message at the scheduled time.

Parameters:

Name	Type	Description	Default
`when`	`datetime`	when you want the message to be sent, as :py:class:`datetime.datetime` instance	required
`channel`	`Channel \| str`	:py:class:`~machine.models.channel.Channel` object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
`text`	`str`	message text	required
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`
`thread_ts`	`str \| None`	optional timestamp of thread, to send a message in that thread	`None`

Returns:

Type	Description
`AsyncSlackResponse`	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/base.py

async def say_scheduled(
    self,
    when: datetime,
    channel: Channel | str,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    thread_ts: str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a message to a channel

    This is the scheduled version of :py:meth:`~machine.plugins.base.MachineBasePlugin.say`.
    It behaves the same, but will send the message at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    return await self._client.send_scheduled(
        when,
        channel,
        text=text,
        attachments=attachments,
        blocks=blocks,
        thread_ts=thread_ts,
        **kwargs,
    )

`update(channel: Channel | str, ts: str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, ephemeral_user: User | str | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

Update an existing message

Update an existing message using the WebAPI. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Can also update in-thread and ephemeral messages, visible to only one user. Any extra kwargs you provide, will be passed on directly to the chat.update request.

Parameters:

Name	Type	Description	Default
`channel`	`Channel \| str`	`Channel` object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
`ts`	`str`	timestamp of the message to be updated.	required
`text`	`str \| None`	message text	`None`
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see [attachments])	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see [blocks])	`None`
`thread_ts`		optional timestamp of thread, to send a message in that thread	required
`ephemeral_user`	`User \| str \| None`	optional user name or id if the message needs to visible to a specific user only	`None`

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `chat.update` request

Source code in machine/plugins/base.py

async def update(
    self,
    channel: Channel | str,
    ts: str,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    ephemeral_user: User | str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Update an existing message

    Update an existing message using the WebAPI. Allows for rich formatting using
    [blocks] and/or [attachments]. You can provide blocks and attachments as Python dicts or
    you can use the [convenient classes] that the underlying slack client provides.
    Can also update in-thread and ephemeral messages, visible to only one user.
    Any extra kwargs you provide, will be passed on directly to the [`chat.update`][chat_update] request.

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes
    [chat_update]: https://api.slack.com/methods/chat.update

    :param channel: [`Channel`][machine.models.channel.Channel] object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param ts: timestamp of the message to be updated.
    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :param ephemeral_user: optional user name or id if the message needs to visible
        to a specific user only
    :return: Dictionary deserialized from [`chat.update`](https://api.slack.com/methods/chat.update) request


    """
    return await self._client.update(
        channel,
        ts=ts,
        text=text,
        attachments=attachments,
        blocks=blocks,
        ephemeral_user=ephemeral_user,
        **kwargs,
    )

`delete(channel: Channel | str, ts: str, **kwargs: Any) -> AsyncSlackResponse` `async`

Delete an existing message

Delete an existing message using the WebAPI. Any extra kwargs you provide, will be passed on directly to the chat.delete request.

Parameters:

Name	Type	Description	Default
`channel`	`Channel \| str`	`Channel` object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
`ts`	`str`	timestamp of the message to be deleted.	required

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `chat.delete` request

Source code in machine/plugins/base.py

async def delete(
    self,
    channel: Channel | str,
    ts: str,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Delete an existing message

    Delete an existing message using the WebAPI.
    Any extra kwargs you provide, will be passed on directly to the [`chat.delete`][chat_delete] request.

    [chat_delete]: https://api.slack.com/methods/chat.delete

    :param channel: [`Channel`][machine.models.channel.Channel] object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param ts: timestamp of the message to be deleted.
    :return: Dictionary deserialized from [`chat.delete`](https://api.slack.com/methods/chat.delete) request


    """
    return await self._client.delete(
        channel,
        ts=ts,
        **kwargs,
    )

`react(channel: Channel | str, ts: str, emoji: str) -> AsyncSlackResponse` `async`

React to a message in a channel

Add a reaction to a message in a channel. What message to react to, is determined by the combination of the channel and the timestamp of the message.

Parameters:

Name	Type	Description	Default
`channel`	`Channel \| str`	:py:class:`~machine.models.channel.Channel` object or id of channel to send message to. Can be public or private (group) channel, or DM channel.	required
`ts`	`str`	timestamp of the message to react to	required
`emoji`	`str`	what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)	required

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `reactions.add`_ request. .. _reactions.add: https://api.slack.com/methods/reactions.add

Source code in machine/plugins/base.py

async def react(self, channel: Channel | str, ts: str, emoji: str) -> AsyncSlackResponse:
    """React to a message in a channel

    Add a reaction to a message in a channel. What message to react to, is determined by the
    combination of the channel and the timestamp of the message.

    :param channel: :py:class:`~machine.models.channel.Channel` object or id of channel to send
        message to. Can be public or private (group) channel, or DM channel.
    :param ts: timestamp of the message to react to
    :param emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)
    :return: Dictionary deserialized from `reactions.add`_ request.

    .. _reactions.add: https://api.slack.com/methods/reactions.add
    """
    return await self._client.react(channel, ts, emoji)

`send_dm(user: User | str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

Send a Direct Message

Send a Direct Message to a user by opening a DM channel and sending a message to it. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
`user`	`User \| str`	:py:class:`~machine.models.user.User` object or id of user to send DM to.	required
`text`	`str \| None`	message text	`None`
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `chat.postMessage`_ request. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage

Source code in machine/plugins/base.py

async def send_dm(
    self,
    user: User | str,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a Direct Message

    Send a Direct Message to a user by opening a DM channel and sending a message to it. Allows
    for rich formatting using `blocks`_ and/or `attachments`_. You can provide blocks and
    attachments as Python dicts or you can use the `convenient classes`_ that the underlying
    slack client provides.
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param user: :py:class:`~machine.models.user.User` object or id of user to send DM to.
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :return: Dictionary deserialized from `chat.postMessage`_ request.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    """
    return await self._client.send_dm(user, text, attachments=attachments, blocks=blocks, **kwargs)

`send_dm_scheduled(when: datetime, user: User | str, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

Schedule a Direct Message

This is the scheduled version of :py:meth:~machine.plugins.base.MachineBasePlugin.send_dm. It behaves the same, but will send the DM at the scheduled time.

Parameters:

Name	Type	Description	Default
`when`	`datetime`	when you want the message to be sent, as :py:class:`datetime.datetime` instance	required
`user`	`User \| str`	:py:class:`~machine.models.user.User` object or id of user to send DM to.	required
`text`	`str`	message text	required
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`

Returns:

Type	Description
`AsyncSlackResponse`	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/base.py

async def send_dm_scheduled(
    self,
    when: datetime,
    user: User | str,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a Direct Message

    This is the scheduled version of
    :py:meth:`~machine.plugins.base.MachineBasePlugin.send_dm`. It behaves the same, but
    will send the DM at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param user: :py:class:`~machine.models.user.User` object or id of user to send DM to.
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    return await self._client.send_dm_scheduled(
        when, user, text=text, attachments=attachments, blocks=blocks, **kwargs
    )

`emit(event: str, **kwargs: Any) -> None`

Emit an event

Emit an event that plugins can listen for. You can include arbitrary data as keyword arguments.

Parameters:

Name	Type	Description	Default
`event`	`str`	name of the event	required
`**kwargs`	`Any`	any data you want to emit with the event	`{}`

Returns:

Type	Description
`None`	None

Source code in machine/plugins/base.py

def emit(self, event: str, **kwargs: Any) -> None:
    """Emit an event

    Emit an event that plugins can listen for. You can include arbitrary data as keyword
    arguments.

    :param event: name of the event
    :param **kwargs: any data you want to emit with the event
    :return: None
    """
    ee.emit(event, self, **kwargs)

`pin_message(channel: Channel | str, ts: str) -> AsyncSlackResponse` `async`

Pin message

Pin a message in a channel

Parameters:

Name	Type	Description	Default
`channel`	`Channel \| str`	channel to pin the message in	required
`ts`	`str`	timestamp of the message to pin	required

Returns:

Type	Description
`AsyncSlackResponse`	response from the Slack Web API

Source code in machine/plugins/base.py

async def pin_message(self, channel: Channel | str, ts: str) -> AsyncSlackResponse:
    """Pin message

    Pin a message in a channel

    :param channel: channel to pin the message in
    :param ts: timestamp of the message to pin
    :return: response from the Slack Web API
    """
    return await self._client.pin_message(channel, ts)

`unpin_message(channel: Channel | str, ts: str) -> AsyncSlackResponse` `async`

Unpin message

Unpin a message that was previously pinned in a channel

Parameters:

Name	Type	Description	Default
`channel`	`Channel \| str`	channel where the message is pinned that needs to be unpinned	required
`ts`	`str`	timestamp of the message to unpin	required

Returns:

Type	Description
`AsyncSlackResponse`	response from the Slack Web API

Source code in machine/plugins/base.py

async def unpin_message(self, channel: Channel | str, ts: str) -> AsyncSlackResponse:
    """Unpin message

    Unpin a message that was previously pinned in a channel

    :param channel: channel where the message is pinned that needs to be unpinned
    :param ts: timestamp of the message to unpin
    :return: response from the Slack Web API
    """
    return await self._client.unpin_message(channel, ts)

`set_topic(channel: Channel | str, topic: str, **kwargs: Any) -> AsyncSlackResponse` `async`

Set channel topic

Set or update topic for the channel

Parameters:

Name	Type	Description	Default
`channel`	`Channel \| str`	channel where topic needs to be set or updated	required
`topic`	`str`	topic for the channel (slack does not support formatting for topics)	required

Returns:

Type	Description
`AsyncSlackResponse`	response from the Slack Web API

Source code in machine/plugins/base.py

async def set_topic(self, channel: Channel | str, topic: str, **kwargs: Any) -> AsyncSlackResponse:
    """Set channel topic

    Set or update topic for the channel

    :param channel: channel where topic needs to be set or updated
    :param topic: topic for the channel (slack does not support formatting for topics)
    :return: response from the Slack Web API
    """
    return await self._client.set_topic(channel, topic, **kwargs)

`machine.plugins.message.Message`

A message that was received by the bot

This class represents a message that was received by the bot and passed to one or more plugins. It contains the message (text) itself, and metadata about the message, such as the sender of the message, the channel the message was sent to.

The Message class also contains convenience methods for replying to the message in the right channel, replying to the sender, etc.

Source code in machine/plugins/message.py

class Message:
    """A message that was received by the bot

    This class represents a message that was received by the bot and passed to one or more
    plugins. It contains the message (text) itself, and metadata about the message, such as the
    sender of the message, the channel the message was sent to.

    The ``Message`` class also contains convenience methods for replying to the message in the
    right channel, replying to the sender, etc.
    """

    # TODO: create proper class for msg_event
    def __init__(self, client: SlackClient, msg_event: dict[str, Any]):
        self._client = client
        self._msg_event = msg_event

    @property
    def sender(self) -> User:
        """The sender of the message

        :return: the User the message was sent by
        """
        return self._client.users[self._msg_event["user"]]

    @property
    def channel(self) -> Channel:
        """The channel the message was sent to

        :return: the Channel the message was sent to
        """
        return self._client.channels[self._msg_event["channel"]]

    @property
    def is_dm(self) -> bool:
        channel_id = self._msg_event["channel"]
        return not (channel_id.startswith("C") or channel_id.startswith("G"))

    @property
    def text(self) -> str:
        """The body of the actual message

        :return: the body (text) of the actual message
        """
        return self._msg_event["text"]

    @property
    def at_sender(self) -> str:
        """The sender of the message formatted as mention

        :return: a string representation of the sender of the message, formatted as `mention`_,
            to be used in messages

        .. _mention: https://api.slack.com/docs/message-formatting#linking_to_channels_and_users
        """
        return self.sender.fmt_mention()

    async def say(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        thread_ts: str | None = None,
        ephemeral: bool = False,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Send a new message to the channel the original message was received in

        Send a new message to the channel the original message was received in, using the WebAPI.
        Allows for rich formatting using `blocks`_ and/or `attachments`_. You can provide blocks
        and attachments as Python dicts or you can use the `convenient classes`_ that the
        underlying slack client provides.
        Can also reply to a thread and send an ephemeral message only visible to the sender of the
        original message. Ephemeral messages and threaded messages are mutually exclusive, and
        ``ephemeral`` takes precedence over ``thread_ts``
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
        `chat.postEphemeral`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :param ephemeral: ``True/False`` wether to send the message as an ephemeral message, only
            visible to the sender of the original message
        :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
            if `ephemeral` is True.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
        """
        ephemeral_user = self.sender.id if ephemeral else None

        return await self._client.send(
            self.channel.id,
            text=text,
            attachments=attachments,
            blocks=blocks,
            thread_ts=thread_ts,
            ephemeral_user=ephemeral_user,
            **kwargs,
        )

    async def say_scheduled(
        self,
        when: datetime,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        thread_ts: str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a message

        This is the scheduled version of :py:meth:`~machine.plugins.base.Message.say`.
        It behaves the same, but will send the message at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param thread_ts: optional timestamp of thread, to send a message in that thread
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        return await self._client.send_scheduled(
            when,
            self.channel.id,
            text=text,
            attachments=attachments,
            blocks=blocks,
            thread_ts=thread_ts,
            **kwargs,
        )

    async def reply(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        in_thread: bool = False,
        ephemeral: bool = False,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Reply to the sender of the original message

        Reply to the sender of the original message with a new message, mentioning that user. Rich
        formatting using `blocks`_ and/or `attachments`_ is possible. You can provide blocks
        and attachments as Python dicts or you can use the `convenient classes`_ that the
        underlying slack client provides.
        Can also reply to a thread and send an ephemeral message only visible to the sender of the
        original message. In the case of in-thread response, the sender of the original message
        will not be mentioned. Ephemeral messages and threaded messages are mutually exclusive,
        and ``ephemeral`` takes precedence over ``in_thread``
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
        `chat.postEphemeral`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param in_thread: ``True/False`` wether to reply to the original message in-thread
        :param ephemeral: ``True/False`` wether to send the message as an ephemeral message, only
            visible to the sender of the original message
        :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
            if `ephemeral` is True.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
        """
        if in_thread and not ephemeral:
            return await self.say(text, attachments=attachments, blocks=blocks, thread_ts=self.ts, **kwargs)
        else:
            text = self._create_reply(text)
            return await self.say(text, attachments=attachments, blocks=blocks, ephemeral=ephemeral, **kwargs)

    async def reply_scheduled(
        self,
        when: datetime,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        in_thread: bool = False,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a reply and send it

        This is the scheduled version of :py:meth:`~machine.plugins.base.Message.reply`.
        It behaves the same, but will send the reply at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :param in_thread: ``True/False`` wether to reply to the original message in-thread
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        if in_thread:
            return await self.say_scheduled(
                when, text, attachments=attachments, blocks=blocks, thread_ts=self.ts, **kwargs
            )
        else:
            text = cast(str, self._create_reply(text))
            return await self.say_scheduled(when, text, attachments=attachments, blocks=blocks, **kwargs)

    async def reply_dm(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Reply to the sender of the original message with a DM

        Reply in a Direct Message to the sender of the original message by opening a DM channel and
        sending a message to it. Allows for rich formatting using `blocks`_ and/or `attachments`_.
        You can provide blocks and attachments as Python dicts or you can use the
        `convenient classes`_ that the underlying slack client provides.
        Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ request.

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        .. _convenient classes:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :return: Dictionary deserialized from `chat.postMessage`_ request.

        .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
        """
        return await self._client.send_dm(self.sender.id, text, attachments=attachments, blocks=blocks, **kwargs)

    async def reply_dm_scheduled(
        self,
        when: datetime,
        text: str,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Schedule a DM reply and send it

        This is the scheduled version of :py:meth:`~machine.plugins.base.Message.reply_dm`.
        It behaves the same, but will send the DM at the scheduled time.

        :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        :param text: message text
        :param attachments: optional attachments (see `attachments`_)
        :param blocks: optional blocks (see `blocks`_)
        :return: None

        .. _attachments: https://api.slack.com/docs/message-attachments
        .. _blocks: https://api.slack.com/reference/block-kit/blocks
        """
        return await self._client.send_dm_scheduled(
            when, self.sender.id, text=text, attachments=attachments, blocks=blocks, **kwargs
        )

    async def react(self, emoji: str) -> AsyncSlackResponse:
        """React to the original message

        Add a reaction to the original message

        :param emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)
        :return: Dictionary deserialized from `reactions.add`_ request.

        .. _reactions.add: https://api.slack.com/methods/reactions.add
        """
        return await self._client.react(self.channel.id, self._msg_event["ts"], emoji)

    def _create_reply(self, text: str | None) -> str | None:
        if not self.is_dm and text is not None:
            return f"{self.at_sender}: {text}"
        else:
            return text

    @property
    def ts(self) -> str:
        """The timestamp of the message

        :return: the timestamp of the message
        """
        return self._msg_event["ts"]

    @property
    def in_thread(self) -> bool:
        """Is message in a thread

        :return: bool
        """
        return "thread_ts" in self._msg_event

    async def pin_message(self) -> AsyncSlackResponse:
        """Pin message

        Pin the current message in the channel it was posted in
        """
        return await self._client.pin_message(self.channel, self.ts)

    def __str__(self) -> str:
        if self.channel.is_im:
            message = f"Message '{self.text}', sent by user @{self.sender.profile.real_name} in DM"
        else:
            message = (
                f"Message '{self.text}', sent by user @{self.sender.profile.real_name} in channel #{self.channel.name}"
            )
        return message

    def __repr__(self) -> str:
        return f"Message(text={self.text!r}, sender={self.sender.profile.real_name!r}, channel={self.channel.name!r})"

`sender: User` `property`

The sender of the message

Returns:

Type	Description
`User`	the User the message was sent by

`channel: Channel` `property`

The channel the message was sent to

Returns:

Type	Description
`Channel`	the Channel the message was sent to

`text: str` `property`

The body of the actual message

Returns:

Type	Description
`str`	the body (text) of the actual message

`at_sender: str` `property`

The sender of the message formatted as mention

Returns:

Type	Description
`str`	a string representation of the sender of the message, formatted as `mention`_, to be used in messages .. _mention: https://api.slack.com/docs/message-formatting#linking_to_channels_and_users

`ts: str` `property`

The timestamp of the message

Returns:

Type	Description
`str`	the timestamp of the message

`in_thread: bool` `property`

Is message in a thread

Returns:

Type	Description
`bool`	bool

`say(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, ephemeral: bool = False, **kwargs: Any) -> AsyncSlackResponse` `async`

Send a new message to the channel the original message was received in

Send a new message to the channel the original message was received in, using the WebAPI. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Can also reply to a thread and send an ephemeral message only visible to the sender of the original message. Ephemeral messages and threaded messages are mutually exclusive, and ephemeral takes precedence over thread_ts Any extra kwargs you provide, will be passed on directly to the chat.postMessage or chat.postEphemeral_ request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
`text`	`str \| None`	message text	`None`
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`
`thread_ts`	`str \| None`	optional timestamp of thread, to send a message in that thread	`None`
`ephemeral`	`bool`	`True/False` wether to send the message as an ephemeral message, only visible to the sender of the original message	`False`

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `chat.postMessage` request, or `chat.postEphemeral` if `ephemeral` is True. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral

Source code in machine/plugins/message.py

async def say(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    thread_ts: str | None = None,
    ephemeral: bool = False,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Send a new message to the channel the original message was received in

    Send a new message to the channel the original message was received in, using the WebAPI.
    Allows for rich formatting using `blocks`_ and/or `attachments`_. You can provide blocks
    and attachments as Python dicts or you can use the `convenient classes`_ that the
    underlying slack client provides.
    Can also reply to a thread and send an ephemeral message only visible to the sender of the
    original message. Ephemeral messages and threaded messages are mutually exclusive, and
    ``ephemeral`` takes precedence over ``thread_ts``
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
    `chat.postEphemeral`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :param ephemeral: ``True/False`` wether to send the message as an ephemeral message, only
        visible to the sender of the original message
    :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
        if `ephemeral` is True.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
    """
    ephemeral_user = self.sender.id if ephemeral else None

    return await self._client.send(
        self.channel.id,
        text=text,
        attachments=attachments,
        blocks=blocks,
        thread_ts=thread_ts,
        ephemeral_user=ephemeral_user,
        **kwargs,
    )

`say_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

Schedule a message

This is the scheduled version of :py:meth:~machine.plugins.base.Message.say. It behaves the same, but will send the message at the scheduled time.

Parameters:

Name	Type	Description	Default
`when`	`datetime`	when you want the message to be sent, as :py:class:`datetime.datetime` instance	required
`text`	`str`	message text	required
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`
`thread_ts`	`str \| None`	optional timestamp of thread, to send a message in that thread	`None`

Returns:

Type	Description
`AsyncSlackResponse`	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/message.py

async def say_scheduled(
    self,
    when: datetime,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    thread_ts: str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a message

    This is the scheduled version of :py:meth:`~machine.plugins.base.Message.say`.
    It behaves the same, but will send the message at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param thread_ts: optional timestamp of thread, to send a message in that thread
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    return await self._client.send_scheduled(
        when,
        self.channel.id,
        text=text,
        attachments=attachments,
        blocks=blocks,
        thread_ts=thread_ts,
        **kwargs,
    )

`reply(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, in_thread: bool = False, ephemeral: bool = False, **kwargs: Any) -> AsyncSlackResponse` `async`

Reply to the sender of the original message

Reply to the sender of the original message with a new message, mentioning that user. Rich formatting using blocks and/or attachments is possible. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Can also reply to a thread and send an ephemeral message only visible to the sender of the original message. In the case of in-thread response, the sender of the original message will not be mentioned. Ephemeral messages and threaded messages are mutually exclusive, and ephemeral takes precedence over in_thread Any extra kwargs you provide, will be passed on directly to the chat.postMessage or chat.postEphemeral_ request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
`text`	`str \| None`	message text	`None`
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`
`in_thread`	`bool`	`True/False` wether to reply to the original message in-thread	`False`
`ephemeral`	`bool`	`True/False` wether to send the message as an ephemeral message, only visible to the sender of the original message	`False`

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `chat.postMessage` request, or `chat.postEphemeral` if `ephemeral` is True. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral

Source code in machine/plugins/message.py

async def reply(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    in_thread: bool = False,
    ephemeral: bool = False,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Reply to the sender of the original message

    Reply to the sender of the original message with a new message, mentioning that user. Rich
    formatting using `blocks`_ and/or `attachments`_ is possible. You can provide blocks
    and attachments as Python dicts or you can use the `convenient classes`_ that the
    underlying slack client provides.
    Can also reply to a thread and send an ephemeral message only visible to the sender of the
    original message. In the case of in-thread response, the sender of the original message
    will not be mentioned. Ephemeral messages and threaded messages are mutually exclusive,
    and ``ephemeral`` takes precedence over ``in_thread``
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ or
    `chat.postEphemeral`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param in_thread: ``True/False`` wether to reply to the original message in-thread
    :param ephemeral: ``True/False`` wether to send the message as an ephemeral message, only
        visible to the sender of the original message
    :return: Dictionary deserialized from `chat.postMessage`_ request, or `chat.postEphemeral`_
        if `ephemeral` is True.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    .. _chat.postEphemeral: https://api.slack.com/methods/chat.postEphemeral
    """
    if in_thread and not ephemeral:
        return await self.say(text, attachments=attachments, blocks=blocks, thread_ts=self.ts, **kwargs)
    else:
        text = self._create_reply(text)
        return await self.say(text, attachments=attachments, blocks=blocks, ephemeral=ephemeral, **kwargs)

`reply_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, in_thread: bool = False, **kwargs: Any) -> AsyncSlackResponse` `async`

Schedule a reply and send it

This is the scheduled version of :py:meth:~machine.plugins.base.Message.reply. It behaves the same, but will send the reply at the scheduled time.

Parameters:

Name	Type	Description	Default
`when`	`datetime`	when you want the message to be sent, as :py:class:`datetime.datetime` instance	required
`text`	`str`	message text	required
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`
`in_thread`	`bool`	`True/False` wether to reply to the original message in-thread	`False`

Returns:

Type	Description
`AsyncSlackResponse`	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/message.py

async def reply_scheduled(
    self,
    when: datetime,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    in_thread: bool = False,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a reply and send it

    This is the scheduled version of :py:meth:`~machine.plugins.base.Message.reply`.
    It behaves the same, but will send the reply at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :param in_thread: ``True/False`` wether to reply to the original message in-thread
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    if in_thread:
        return await self.say_scheduled(
            when, text, attachments=attachments, blocks=blocks, thread_ts=self.ts, **kwargs
        )
    else:
        text = cast(str, self._create_reply(text))
        return await self.say_scheduled(when, text, attachments=attachments, blocks=blocks, **kwargs)

`reply_dm(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

Reply to the sender of the original message with a DM

Reply in a Direct Message to the sender of the original message by opening a DM channel and sending a message to it. Allows for rich formatting using blocks and/or attachments. You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

.. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks .. _convenient classes: https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

Parameters:

Name	Type	Description	Default
`text`	`str \| None`	message text	`None`
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `chat.postMessage`_ request. .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage

Source code in machine/plugins/message.py

async def reply_dm(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Reply to the sender of the original message with a DM

    Reply in a Direct Message to the sender of the original message by opening a DM channel and
    sending a message to it. Allows for rich formatting using `blocks`_ and/or `attachments`_.
    You can provide blocks and attachments as Python dicts or you can use the
    `convenient classes`_ that the underlying slack client provides.
    Any extra kwargs you provide, will be passed on directly to the `chat.postMessage`_ request.

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    .. _convenient classes:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :return: Dictionary deserialized from `chat.postMessage`_ request.

    .. _chat.postMessage: https://api.slack.com/methods/chat.postMessage
    """
    return await self._client.send_dm(self.sender.id, text, attachments=attachments, blocks=blocks, **kwargs)

`reply_dm_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

Schedule a DM reply and send it

This is the scheduled version of :py:meth:~machine.plugins.base.Message.reply_dm. It behaves the same, but will send the DM at the scheduled time.

Parameters:

Name	Type	Description	Default
`when`	`datetime`	when you want the message to be sent, as :py:class:`datetime.datetime` instance	required
`text`	`str`	message text	required
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see `attachments`_)	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see `blocks`_)	`None`

Returns:

Type	Description
`AsyncSlackResponse`	None .. _attachments: https://api.slack.com/docs/message-attachments .. _blocks: https://api.slack.com/reference/block-kit/blocks

Source code in machine/plugins/message.py

async def reply_dm_scheduled(
    self,
    when: datetime,
    text: str,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Schedule a DM reply and send it

    This is the scheduled version of :py:meth:`~machine.plugins.base.Message.reply_dm`.
    It behaves the same, but will send the DM at the scheduled time.

    :param when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
    :param text: message text
    :param attachments: optional attachments (see `attachments`_)
    :param blocks: optional blocks (see `blocks`_)
    :return: None

    .. _attachments: https://api.slack.com/docs/message-attachments
    .. _blocks: https://api.slack.com/reference/block-kit/blocks
    """
    return await self._client.send_dm_scheduled(
        when, self.sender.id, text=text, attachments=attachments, blocks=blocks, **kwargs
    )

`react(emoji: str) -> AsyncSlackResponse` `async`

React to the original message

Add a reaction to the original message

Parameters:

Name	Type	Description	Default
`emoji`	`str`	what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)	required

Returns:

Type	Description
`AsyncSlackResponse`	Dictionary deserialized from `reactions.add`_ request. .. _reactions.add: https://api.slack.com/methods/reactions.add

Source code in machine/plugins/message.py

async def react(self, emoji: str) -> AsyncSlackResponse:
    """React to the original message

    Add a reaction to the original message

    :param emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)
    :return: Dictionary deserialized from `reactions.add`_ request.

    .. _reactions.add: https://api.slack.com/methods/reactions.add
    """
    return await self._client.react(self.channel.id, self._msg_event["ts"], emoji)

`pin_message() -> AsyncSlackResponse` `async`

Pin message

Pin the current message in the channel it was posted in

Source code in machine/plugins/message.py

async def pin_message(self) -> AsyncSlackResponse:
    """Pin message

    Pin the current message in the channel it was posted in
    """
    return await self._client.pin_message(self.channel, self.ts)

`machine.plugins.command.Command`

A Slack command that was received by the bot

This class represents a Slack command that was received by the bot and passed to a plugin. It contains the text that was included when the command was invoked, and metadata about the command, such as the user that invoked the command, the channel the command was invoked in.

The Command class also contains convenience methods for sending messages in the right channel, opening modals etc.

Source code in machine/plugins/command.py

class Command:
    """A Slack command that was received by the bot

    This class represents a Slack command that was received by the bot and passed to a plugin.
    It contains the text that was included when the command was invoked, and metadata about
    the command, such as the user that invoked the command, the channel the command was invoked
    in.

    The `Command` class also contains convenience methods for sending messages in the right
    channel, opening modals etc.
    """

    # TODO: create proper class for cmd_event
    def __init__(self, client: SlackClient, cmd_payload: dict[str, Any]):
        self._client = client
        self._cmd_payload = cmd_payload
        self._webhook_client = AsyncWebhookClient(self._cmd_payload["response_url"])

    @property
    def sender(self) -> User:
        """The sender of the message

        :return: the User the message was sent by
        """
        return self._client.users[self._cmd_payload["user_id"]]

    @property
    def channel(self) -> Channel:
        """The channel the message was sent to

        :return: the Channel the message was sent to
        """
        return self._client.channels[self._cmd_payload["channel_id"]]

    @property
    def is_dm(self) -> bool:
        channel_id = self._cmd_payload["channel_id"]
        return not (channel_id.startswith("C") or channel_id.startswith("G"))

    @property
    def text(self) -> str:
        """The body of the actual message

        :return: the body (text) of the actual message
        """
        return self._cmd_payload["text"]

    @property
    def command(self) -> str:
        """The command that was invoked

        :return: the command that was invoked
        """
        return self._cmd_payload["command"]

    @property
    def response_url(self) -> str:
        """The response url associated with the command

        This is a unique url for this specific command invocation.
        It can be used for sending messages in response to the command.
        This can only be used 5 times within 30 minutes of receiving the payload.

        :return: the response url associated with the command
        """
        return self._cmd_payload["response_url"]

    @property
    def trigger_id(self) -> str:
        """The trigger id associated with the command

        The trigger id can be used to trigger modals

        :return: the trigger id associated with the command
        """
        return self._cmd_payload["trigger_id"]

    async def say(
        self,
        text: str | None = None,
        attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
        blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
        ephemeral: bool = True,
        **kwargs: Any,
    ) -> WebhookResponse:
        """Send a new message to the channel the command was invoked in

        Send a new message to the channel the command was invoked in, using the response_url as a webhook.
        Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
        and attachments as Python dicts or you can use the [convenient classes] that the
        underlying slack client provides.
        This will send an ephemeral message by default, only visible to the user that invoked the command.
        You can set `ephemeral` to `False` to make the message visible to everyone in the channel
        Any extra kwargs you provide, will be passed on directly to `AsyncWebhookClient.send()`

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :param ephemeral: `True/False` wether to send the message as an ephemeral message, only
            visible to the sender of the original message
        :return: Dictionary deserialized from `AsyncWebhookClient.send()`

        """
        response_type = "ephemeral" if ephemeral else "in_channel"

        return await self._webhook_client.send(
            text=text, attachments=attachments, blocks=blocks, response_type=response_type, **kwargs
        )

`sender: User` `property`

The sender of the message

Returns:

Type	Description
`User`	the User the message was sent by

`channel: Channel` `property`

The channel the message was sent to

Returns:

Type	Description
`Channel`	the Channel the message was sent to

`text: str` `property`

The body of the actual message

Returns:

Type	Description
`str`	the body (text) of the actual message

`command: str` `property`

The command that was invoked

Returns:

Type	Description
`str`	the command that was invoked

`response_url: str` `property`

The response url associated with the command

This is a unique url for this specific command invocation. It can be used for sending messages in response to the command. This can only be used 5 times within 30 minutes of receiving the payload.

Returns:

Type	Description
`str`	the response url associated with the command

`trigger_id: str` `property`

The trigger id associated with the command

The trigger id can be used to trigger modals

Returns:

Type	Description
`str`	the trigger id associated with the command

`say(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, ephemeral: bool = True, **kwargs: Any) -> WebhookResponse` `async`

Send a new message to the channel the command was invoked in

Send a new message to the channel the command was invoked in, using the response_url as a webhook. Allows for rich formatting using blocks and/or attachments . You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. This will send an ephemeral message by default, only visible to the user that invoked the command. You can set ephemeral to False to make the message visible to everyone in the channel Any extra kwargs you provide, will be passed on directly to AsyncWebhookClient.send()

Parameters:

Name	Type	Description	Default
`text`	`str \| None`	message text	`None`
`attachments`	`Sequence[Attachment] \| Sequence[dict[str, Any]] \| None`	optional attachments (see [attachments])	`None`
`blocks`	`Sequence[Block] \| Sequence[dict[str, Any]] \| None`	optional blocks (see [blocks])	`None`
`ephemeral`	`bool`	`True/False` wether to send the message as an ephemeral message, only visible to the sender of the original message	`True`

Returns:

Type	Description
`WebhookResponse`	Dictionary deserialized from `AsyncWebhookClient.send()`

Source code in machine/plugins/command.py

async def say(
    self,
    text: str | None = None,
    attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None,
    blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None,
    ephemeral: bool = True,
    **kwargs: Any,
) -> WebhookResponse:
    """Send a new message to the channel the command was invoked in

    Send a new message to the channel the command was invoked in, using the response_url as a webhook.
    Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
    and attachments as Python dicts or you can use the [convenient classes] that the
    underlying slack client provides.
    This will send an ephemeral message by default, only visible to the user that invoked the command.
    You can set `ephemeral` to `False` to make the message visible to everyone in the channel
    Any extra kwargs you provide, will be passed on directly to `AsyncWebhookClient.send()`

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :param ephemeral: `True/False` wether to send the message as an ephemeral message, only
        visible to the sender of the original message
    :return: Dictionary deserialized from `AsyncWebhookClient.send()`

    """
    response_type = "ephemeral" if ephemeral else "in_channel"

    return await self._webhook_client.send(
        text=text, attachments=attachments, blocks=blocks, response_type=response_type, **kwargs
    )

`machine.plugins.block_action.BlockAction`

A Slack block action that was received by the bot

This class represents a block action that was received by the bot and passed to a plugin. Block actions are actions that are triggered by interactions with blocks in Slack messages and modals. This class contains metadata about the block action, such as the action that happened that triggered this handler, the user that triggered the action, the state of the block when the action was triggered, the payload that was received when the action was triggered.

Source code in machine/plugins/block_action.py

class BlockAction:
    """A Slack block action that was received by the bot

    This class represents a block action that was received by the bot and passed to a plugin.
    Block actions are actions that are triggered by interactions with blocks in Slack messages and modals.
    This class contains metadata about the block action, such as the action that happened that triggered this handler,
    the user that triggered the action, the state of the block when the action was triggered, the payload that was
    received when the action was triggered.
    """

    payload: BlockActionsPayload
    triggered_action: Action

    def __init__(self, client: SlackClient, payload: BlockActionsPayload, triggered_action: Action):
        self._client = client
        self.payload = payload  #: blablab
        """The payload that was received by the bot when the action was triggered that this plugin method listens for"""
        self.triggered_action = triggered_action
        """The action that triggered this plugin method"""
        self._webhook_client = AsyncWebhookClient(self.payload.response_url) if self.payload.response_url else None

    @property
    def user(self) -> User:
        """The user that triggered the action

        :return: the user that triggered the action
        """
        return self._client.users[self.payload.user.id]

    @property
    def channel(self) -> Optional[Channel]:
        """The channel the action was triggered in

        :return: the channel the action was triggered in or None if the action was triggered in a modal
        """
        if self.payload.channel is None:
            return None
        return self._client.channels[self.payload.channel.id]

    @property
    def state(self) -> Optional[State]:
        """The state of the block when the action was triggered

        :return: the state of the block when the action was triggered
        """
        return self.payload.state

    @property
    def response_url(self) -> Optional[str]:
        """The response URL for the action

        :return: the response URL for the action or None if the action was triggered in a modal
        """
        return self.payload.response_url

    @property
    def trigger_id(self) -> str:
        """The trigger id associated with the action

        The trigger id can be user ot open a modal

        :return: the trigger id for the action
        """
        return self.payload.trigger_id

    async def say(
        self,
        text: Optional[str] = None,
        attachments: Union[Sequence[Attachment], Sequence[dict[str, Any]], None] = None,
        blocks: Union[Sequence[Block], Sequence[dict[str, Any]], None] = None,
        ephemeral: bool = True,
        replace_original: bool = False,
        delete_original: bool = False,
        **kwargs: Any,
    ) -> Optional[WebhookResponse]:
        """Send a new message to the channel the block action was triggered in

        Send a new message to the channel the block action was triggered in, using the response_url as a webhook.
        If the block action happened in a modal, the response_url will be None and this method will not send a message
        but instead log a warning.
        Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
        and attachments as Python dicts or you can use the [convenient classes] that the
        underlying slack client provides.
        This will send an ephemeral message by default, only visible to the user that triggered the action.
        You can set `ephemeral` to `False` to make the message visible to everyone in the channel.
        By default, Slack replaces the original message in which the action was triggered. This method overrides this
        behavior. If you want your message to replace the original, set replace_original to True.
        Any extra kwargs you provide, will be passed on directly to `AsyncWebhookClient.send()`

        [attachments]: https://api.slack.com/docs/message-attachments
        [blocks]: https://api.slack.com/reference/block-kit/blocks
        [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        :param text: message text
        :param attachments: optional attachments (see [attachments])
        :param blocks: optional blocks (see [blocks])
        :param ephemeral: `True/False` wether to send the message as an ephemeral message, only
            visible to the user that initiated the action
        :param replace_original: `True/False` whether the message that contains the block from which the action was
            triggered should be replaced by this message
        :param delete_original: `True/False` whether the message that contains the block from which the action was
            triggered should be deleted. No other parameters should be provided.
        :return: Dictionary deserialized from `AsyncWebhookClient.send()`

        """
        if self._webhook_client is None:
            logger.warning(
                "response_url is None, cannot send message. This likely means the action was triggered in a modal."
            )
            return None

        response_type = "ephemeral" if ephemeral else "in_channel"
        return await self._webhook_client.send(
            text=text,
            attachments=attachments,
            blocks=blocks,
            response_type=response_type,
            replace_original=replace_original,
            delete_original=delete_original,
            **kwargs,
        )

`payload: BlockActionsPayload = payload` `instance-attribute`

The payload that was received by the bot when the action was triggered that this plugin method listens for

`triggered_action: Action = triggered_action` `instance-attribute`

The action that triggered this plugin method

`user: User` `property`

The user that triggered the action

Returns:

Type	Description
`User`	the user that triggered the action

`channel: Optional[Channel]` `property`

The channel the action was triggered in

Returns:

Type	Description
`Optional[Channel]`	the channel the action was triggered in or None if the action was triggered in a modal

`state: Optional[State]` `property`

The state of the block when the action was triggered

Returns:

Type	Description
`Optional[State]`	the state of the block when the action was triggered

`response_url: Optional[str]` `property`

The response URL for the action

Returns:

Type	Description
`Optional[str]`	the response URL for the action or None if the action was triggered in a modal

`trigger_id: str` `property`

The trigger id associated with the action

The trigger id can be user ot open a modal

Returns:

Type	Description
`str`	the trigger id for the action

`say(text: Optional[str] = None, attachments: Union[Sequence[Attachment], Sequence[dict[str, Any]], None] = None, blocks: Union[Sequence[Block], Sequence[dict[str, Any]], None] = None, ephemeral: bool = True, replace_original: bool = False, delete_original: bool = False, **kwargs: Any) -> Optional[WebhookResponse]` `async`

Send a new message to the channel the block action was triggered in

Send a new message to the channel the block action was triggered in, using the response_url as a webhook. If the block action happened in a modal, the response_url will be None and this method will not send a message but instead log a warning. Allows for rich formatting using blocks and/or attachments . You can provide blocks and attachments as Python dicts or you can use the convenient classes that the underlying slack client provides. This will send an ephemeral message by default, only visible to the user that triggered the action. You can set ephemeral to False to make the message visible to everyone in the channel. By default, Slack replaces the original message in which the action was triggered. This method overrides this behavior. If you want your message to replace the original, set replace_original to True. Any extra kwargs you provide, will be passed on directly to AsyncWebhookClient.send()

Parameters:

Name	Type	Description	Default
`text`	`Optional[str]`	message text	`None`
`attachments`	`Union[Sequence[Attachment], Sequence[dict[str, Any]], None]`	optional attachments (see [attachments])	`None`
`blocks`	`Union[Sequence[Block], Sequence[dict[str, Any]], None]`	optional blocks (see [blocks])	`None`
`ephemeral`	`bool`	`True/False` wether to send the message as an ephemeral message, only visible to the user that initiated the action	`True`
`replace_original`	`bool`	`True/False` whether the message that contains the block from which the action was triggered should be replaced by this message	`False`
`delete_original`	`bool`	`True/False` whether the message that contains the block from which the action was triggered should be deleted. No other parameters should be provided.	`False`

Returns:

Type	Description
`Optional[WebhookResponse]`	Dictionary deserialized from `AsyncWebhookClient.send()`

Source code in machine/plugins/block_action.py

async def say(
    self,
    text: Optional[str] = None,
    attachments: Union[Sequence[Attachment], Sequence[dict[str, Any]], None] = None,
    blocks: Union[Sequence[Block], Sequence[dict[str, Any]], None] = None,
    ephemeral: bool = True,
    replace_original: bool = False,
    delete_original: bool = False,
    **kwargs: Any,
) -> Optional[WebhookResponse]:
    """Send a new message to the channel the block action was triggered in

    Send a new message to the channel the block action was triggered in, using the response_url as a webhook.
    If the block action happened in a modal, the response_url will be None and this method will not send a message
    but instead log a warning.
    Allows for rich formatting using [blocks] and/or [attachments] . You can provide blocks
    and attachments as Python dicts or you can use the [convenient classes] that the
    underlying slack client provides.
    This will send an ephemeral message by default, only visible to the user that triggered the action.
    You can set `ephemeral` to `False` to make the message visible to everyone in the channel.
    By default, Slack replaces the original message in which the action was triggered. This method overrides this
    behavior. If you want your message to replace the original, set replace_original to True.
    Any extra kwargs you provide, will be passed on directly to `AsyncWebhookClient.send()`

    [attachments]: https://api.slack.com/docs/message-attachments
    [blocks]: https://api.slack.com/reference/block-kit/blocks
    [convenient classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    :param text: message text
    :param attachments: optional attachments (see [attachments])
    :param blocks: optional blocks (see [blocks])
    :param ephemeral: `True/False` wether to send the message as an ephemeral message, only
        visible to the user that initiated the action
    :param replace_original: `True/False` whether the message that contains the block from which the action was
        triggered should be replaced by this message
    :param delete_original: `True/False` whether the message that contains the block from which the action was
        triggered should be deleted. No other parameters should be provided.
    :return: Dictionary deserialized from `AsyncWebhookClient.send()`

    """
    if self._webhook_client is None:
        logger.warning(
            "response_url is None, cannot send message. This likely means the action was triggered in a modal."
        )
        return None

    response_type = "ephemeral" if ephemeral else "in_channel"
    return await self._webhook_client.send(
        text=text,
        attachments=attachments,
        blocks=blocks,
        response_type=response_type,
        replace_original=replace_original,
        delete_original=delete_original,
        **kwargs,
    )

Decorators

These are the decorators you can use to have Slack Machine respond to specific things (events, messages, etc.)

`machine.plugins.decorators`

`process(slack_event_type: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

Process Slack events of a specific type

This decorator will enable a Plugin method to process Slack events_ of a specific type. The Plugin method will be called for each event of the specified type that the bot receives. The received event will be passed to the method when called.

.. _Slack events: https://api.slack.com/events

Parameters:

Name	Type	Description	Default
`slack_event_type`	`str`	type of event the method needs to process. Can be any event supported by the RTM API	required

Returns:

Type	Description
`Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`	wrapped method

Source code in machine/plugins/decorators.py

def process(slack_event_type: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Process Slack events of a specific type

    This decorator will enable a Plugin method to process `Slack events`_ of a specific type. The
    Plugin method will be called for each event of the specified type that the bot receives.
    The received event will be passed to the method when called.

    .. _Slack events: https://api.slack.com/events

    :param slack_event_type: type of event the method needs to process. Can be any event supported
        by the RTM API
    :return: wrapped method
    """

    def process_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        fn.metadata.plugin_actions.process.append(slack_event_type)
        return fn

    return process_decorator

`listen_to(regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

Listen to messages matching a regex pattern

This decorator will enable a Plugin method to listen to messages that match a regex pattern. The Plugin method will be called for each message that matches the specified regex pattern. The received :py:class:~machine.plugins.base.Message will be passed to the method when called. Named groups can be used in the regex pattern, to catch specific parts of the message. These groups will be passed to the method as keyword arguments when called.

Parameters:

Name	Type	Description	Default
`regex`	`str`	regex pattern to listen for	required
`flags`	`RegexFlag \| int`	regex flags to apply when matching	`IGNORECASE`
`handle_message_changed`	`bool`	if changed messages should trigger the decorated function	`False`

Returns:

Type	Description
`Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`	wrapped method

Source code in machine/plugins/decorators.py

def listen_to(
    regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False
) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Listen to messages matching a regex pattern

    This decorator will enable a Plugin method to listen to messages that match a regex pattern.
    The Plugin method will be called for each message that matches the specified regex pattern.
    The received :py:class:`~machine.plugins.base.Message` will be passed to the method when called.
    Named groups can be used in the regex pattern, to catch specific parts of the message. These
    groups will be passed to the method as keyword arguments when called.

    :param regex: regex pattern to listen for
    :param flags: regex flags to apply when matching
    :param handle_message_changed: if changed messages should trigger the decorated function
    :return: wrapped method
    """

    def listen_to_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        fn.metadata.plugin_actions.listen_to.append(MatcherConfig(re.compile(regex, flags), handle_message_changed))
        return fn

    return listen_to_decorator

`respond_to(regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

Listen to messages mentioning the bot and matching a regex pattern

This decorator will enable a Plugin method to listen to messages that are directed to the bot (ie. message starts by mentioning the bot) and match a regex pattern. The Plugin method will be called for each message that mentions the bot and matches the specified regex pattern. The received :py:class:~machine.plugins.base.Message will be passed to the method when called. Named groups can be used in the regex pattern, to catch specific parts of the message. These groups will be passed to the method as keyword arguments when called.

Parameters:

Name	Type	Description	Default
`regex`	`str`	regex pattern to listen for	required
`flags`	`RegexFlag \| int`	regex flags to apply when matching	`IGNORECASE`
`handle_message_changed`	`bool`	if changed messages should trigger the decorated function	`False`

Returns:

Type	Description
`Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`	wrapped method

Source code in machine/plugins/decorators.py

def respond_to(
    regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False
) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Listen to messages mentioning the bot and matching a regex pattern

    This decorator will enable a Plugin method to listen to messages that are directed to the bot
    (ie. message starts by mentioning the bot) and match a regex pattern.
    The Plugin method will be called for each message that mentions the bot and matches the
    specified regex pattern. The received :py:class:`~machine.plugins.base.Message` will be passed
    to the method when called. Named groups can be used in the regex pattern, to catch specific
    parts of the message. These groups will be passed to the method as keyword arguments when
    called.

    :param regex: regex pattern to listen for
    :param flags: regex flags to apply when matching
    :param handle_message_changed: if changed messages should trigger the decorated function
    :return: wrapped method
    """

    def respond_to_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        fn.metadata.plugin_actions.respond_to.append(MatcherConfig(re.compile(regex, flags), handle_message_changed))
        return fn

    return respond_to_decorator

`command(slash_command: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

Respond to a slash command

This decorator will enable a Plugin method to respond to slash commands

Parameters:

Name	Type	Description	Default
`slash_command`	`str`	the slash command to respond to	required

Returns:

Type	Description
`Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`	wrapped method

Source code in machine/plugins/decorators.py

def command(slash_command: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Respond to a slash command

    This decorator will enable a Plugin method to respond to slash commands

    :param slash_command: the slash command to respond to
    :return: wrapped method
    """

    def command_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        normalized_slash_command = f"/{slash_command}" if not slash_command.startswith("/") else slash_command
        fn.metadata.plugin_actions.commands.append(
            CommandConfig(command=normalized_slash_command, is_generator=inspect.isasyncgenfunction(f))
        )
        return fn

    return command_decorator

`action(action_id: Union[re.Pattern[str], str, None] = None, block_id: Union[re.Pattern[str], str, None] = None) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

Respond to block actions

This decorator will enable a Plugin method to be triggered when certain block actions are received. The Plugin method will be called when a block action event is received for which the action_id and block_id match the provided values. action_id and block_id can be strings, in which case the incoming action_id and block_id must match exactly, or regex patterns, in which case the incoming action_id and block_id must match the regex pattern.

Both action_id and block_id are optional, but at least one of them must be provided.

Parameters:

Name	Type	Description	Default
`action_id`	`Union[Pattern[str], str, None]`	the action_id to respond to, can be a string or regex pattern	`None`
`block_id`	`Union[Pattern[str], str, None]`	the block_id to respond to, can be a string or regex pattern	`None`

Returns:

Type	Description
`Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`	wrapped method

Source code in machine/plugins/decorators.py

def action(
    action_id: Union[re.Pattern[str], str, None] = None, block_id: Union[re.Pattern[str], str, None] = None
) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Respond to block actions

    This decorator will enable a Plugin method to be triggered when certain block actions are
    received. The Plugin method will be called when a block action event is received for which
    the action_id and block_id match the provided values. action_id and block_id can be strings,
    in which case the incoming action_id and block_id must match exactly, or regex patterns, in
    which case the incoming action_id and block_id must match the regex pattern.

    Both action_id and block_id are optional, but **at least one of them must be provided**.

    :param action_id: the action_id to respond to, can be a string or regex pattern
    :param block_id: the block_id to respond to, can be a string or regex pattern
    :return: wrapped method
    """

    def action_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        if action_id is None and block_id is None:
            raise ValueError("At least one of action_id or block_id must be provided")
        fn.metadata.plugin_actions.actions.append(ActionConfig(action_id=action_id, block_id=block_id))
        return fn

    return action_decorator

`schedule(year: int | str | None = None, month: int | str | None = None, day: int | str | None = None, week: int | str | None = None, day_of_week: int | str | None = None, hour: int | str | None = None, minute: int | str | None = None, second: int | str | None = None, start_date: datetime | str | None = None, end_date: datetime | str | None = None, timezone: tzinfo | str | None = None) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

Schedule a function to be executed according to a crontab-like schedule

The decorated function will be executed according to the schedule provided. Slack Machine uses APScheduler under the hood for scheduling. For more information on the interpretation of the provided parameters, see :class:CronTrigger<apscheduler:apscheduler.triggers.cron.CronTrigger>

Parameters:

Name	Type	Description	Default
`year`	`int\|str`	4-digit year	`None`
`month`	`int\|str`	month (1-12)	`None`
`day`	`int\|str`	day of the (1-31)	`None`
`week`	`int\|str`	ISO week (1-53)	`None`
`day_of_week`	`int\|str`	number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)	`None`
`hour`	`int\|str`	hour (0-23)	`None`
`minute`	`int\|str`	minute (0-59)	`None`
`second`	`int\|str`	second (0-59)	`None`
`start_date`	`datetime\|str`	earliest possible date/time to trigger on (inclusive)	`None`
`end_date`	`datetime\|str`	latest possible date/time to trigger on (inclusive)	`None`
`timezone`	`datetime.tzinfo\|str`	time zone to use for the date/time calculations (defaults to scheduler timezone)	`None`

Source code in machine/plugins/decorators.py

def schedule(
    year: int | str | None = None,
    month: int | str | None = None,
    day: int | str | None = None,
    week: int | str | None = None,
    day_of_week: int | str | None = None,
    hour: int | str | None = None,
    minute: int | str | None = None,
    second: int | str | None = None,
    start_date: datetime | str | None = None,
    end_date: datetime | str | None = None,
    timezone: tzinfo | str | None = None,
) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Schedule a function to be executed according to a crontab-like schedule

    The decorated function will be executed according to the schedule provided. Slack Machine uses
    APScheduler under the hood for scheduling. For more information on the interpretation of the
    provided parameters, see :class:`CronTrigger<apscheduler:apscheduler.triggers.cron.CronTrigger>`

    :param int|str year: 4-digit year
    :param int|str month: month (1-12)
    :param int|str day: day of the (1-31)
    :param int|str week: ISO week (1-53)
    :param int|str day_of_week: number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
    :param int|str hour: hour (0-23)
    :param int|str minute: minute (0-59)
    :param int|str second: second (0-59)
    :param datetime|str start_date: earliest possible date/time to trigger on (inclusive)
    :param datetime|str end_date: latest possible date/time to trigger on (inclusive)
    :param datetime.tzinfo|str timezone: time zone to use for the date/time calculations (defaults
        to scheduler timezone)
    """
    kwargs = locals()

    def schedule_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        fn.metadata.plugin_actions.schedule = kwargs
        return fn

    return schedule_decorator

`on(event: str) -> Callable[[Callable[P, R]], Callable[P, R]]`

Listen for an event

The decorated function will be called whenever a plugin (or Slack Machine itself) emits an event with the given name.

Parameters:

Name	Type	Description	Default
`event`	`str`	name of the event to listen for. Event names are global	required

Source code in machine/plugins/decorators.py

def on(event: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Listen for an event

    The decorated function will be called whenever a plugin (or Slack Machine itself) emits an
    event with the given name.

    :param event: name of the event to listen for. Event names are global
    """

    def on_decorator(f: Callable[P, R]) -> Callable[P, R]:
        ee.add_listener(event, f)
        return f

    return on_decorator

`required_settings(settings: list[str] | str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

Specify a required setting for a plugin or plugin method

The settings specified with this decorator will be added to the required settings for the plugin. If one or more settings have not been defined by the user, the plugin will not be loaded and a warning will be written to the console upon startup.

Parameters:

Name	Type	Description	Default
`settings`	`list[str] \| str`	settings that are required (can be list of strings, or single string)	required

Source code in machine/plugins/decorators.py

def required_settings(settings: list[str] | str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Specify a required setting for a plugin or plugin method

    The settings specified with this decorator will be added to the required settings for the
    plugin. If one or more settings have not been defined by the user, the plugin will not be
    loaded and a warning will be written to the console upon startup.

    :param settings: settings that are required (can be list of strings, or single string)
    """

    def required_settings_decorator(f_or_cls: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        casted_f_or_cls = cast(DecoratedPluginFunc, f_or_cls)
        casted_f_or_cls.metadata = getattr(f_or_cls, "metadata", Metadata())
        if isinstance(settings, list):
            casted_f_or_cls.metadata.required_settings.extend(settings)
        elif isinstance(settings, str):
            casted_f_or_cls.metadata.required_settings.append(settings)
        return casted_f_or_cls

    return required_settings_decorator

`require_any_role(required_roles: list[str]) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]`

Specify required roles for a plugin method

To use the plugin method where this decorator is applied, the user must have at least one of the listed roles.

Parameters:

Name	Type	Description	Default
`required_roles`	`list[str]`	list of roles required to use the plugin method	required

Source code in machine/plugins/decorators.py

def require_any_role(
    required_roles: list[str],
) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]:
    """Specify required roles for a plugin method

    To use the plugin method where this decorator is applied, the user must have
    at least one of the listed roles.

    :param required_roles: list of roles required to use the plugin method
    """

    def middle(func: Callable[..., Awaitable[None]]) -> Callable[..., Awaitable[None]]:
        async def wrapper(self: MachineBasePlugin, msg: Message, **kwargs: Any) -> None:
            if await matching_roles_by_user_id(self, msg.sender.id, required_roles):
                logger.debug(f"User {msg.sender} has one of the required roles {required_roles}")
                return await func(self, msg, **kwargs)
            else:
                logger.debug(f"User {msg.sender} does not have any of the required roles {required_roles}")
                ee.emit(
                    "unauthorized-access",
                    self,
                    message=msg,
                    required_roles=required_roles,
                    combinator=RoleCombinator.ANY,
                )
                await msg.say("I'm sorry, but you don't have access to that command", ephemeral=True)
                return None

        # Copy any existing docs and metadata from container function to
        # generated function
        wrapper.__doc__ = func.__doc__
        casted_wrapper = cast(DecoratedPluginFunc, wrapper)
        casted_wrapper.metadata = getattr(func, "metadata", Metadata())
        return casted_wrapper

    return middle

`require_all_roles(required_roles: list[str]) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]`

Specify required roles for a plugin method

To use the plugin method where this decorator is applied, the user must have all of the listed roles.

Parameters:

Name	Type	Description	Default
`required_roles`	`list[str]`	list of roles required to use the plugin method	required

Source code in machine/plugins/decorators.py

def require_all_roles(
    required_roles: list[str],
) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]:
    """Specify required roles for a plugin method

    To use the plugin method where this decorator is applied, the user must have
    all of the listed roles.

    :param required_roles: list of roles required to use the plugin method
    """

    def middle(func: Callable[..., Awaitable[None]]) -> Callable[..., Awaitable[None]]:
        async def wrapper(self: MachineBasePlugin, msg: Message, **kwargs: Any) -> None:
            if await matching_roles_by_user_id(self, msg.sender.id, required_roles) == len(required_roles):
                logger.debug(f"User {msg.sender} has all of the required roles {required_roles}")
                return await func(self, msg, **kwargs)
            else:
                logger.debug(f"User {msg.sender} does not have all of the required roles {required_roles}")
                ee.emit(
                    "unauthorized-access",
                    self,
                    message=msg,
                    required_roles=required_roles,
                    combinator=RoleCombinator.ALL,
                )
                await msg.say("I'm sorry, but you don't have access to that command", ephemeral=True)
                return None

        # Copy any existing docs and metadata from container function to
        # generated function
        wrapper.__doc__ = func.__doc__
        casted_wrapper = cast(DecoratedPluginFunc, wrapper)
        casted_wrapper.metadata = getattr(func, "metadata", Metadata())
        return casted_wrapper

    return middle

Models

These classes represent base objects from the Slack API

`machine.models.user.User`

Bases: BaseModel

User model that represents a user object from the Slack API

Source code in machine/models/user.py

class User(BaseModel):
    """
    User model that represents a user object from the Slack API
    """

    model_config = ConfigDict(frozen=True)

    id: str
    team_id: Optional[str] = None
    name: str
    deleted: Optional[bool] = None
    profile: Profile
    is_bot: bool
    updated: int
    is_app_user: bool
    color: Optional[str] = None
    real_name: Optional[str] = None
    tz: Optional[str] = None
    tz_label: Optional[str] = None
    tz_offset: Optional[int] = None
    is_admin: Optional[bool] = None
    is_owner: Optional[bool] = None
    is_primary_owner: Optional[bool] = None
    is_restricted: Optional[bool] = None
    is_ultra_restricted: Optional[bool] = None
    is_stranger: Optional[bool] = None
    has_2fa: Optional[bool] = None
    locale: Optional[str] = None

    def fmt_mention(self) -> str:
        return f"<@{self.id}>"

`machine.models.channel.Channel`

Bases: BaseModel

Channel model that represents a channel object from the Slack API

Source code in machine/models/channel.py

class Channel(BaseModel):
    """
    Channel model that represents a channel object from the Slack API
    """

    model_config = ConfigDict(frozen=True)

    id: str
    name: Optional[str] = None
    created: int
    creator: Optional[str] = None
    is_archived: bool
    is_general: Optional[bool] = None
    name_normalized: Optional[str] = None
    is_shared: Optional[bool] = None
    is_org_shared: bool
    is_member: Optional[bool] = None
    is_private: Optional[bool] = None
    is_mpim: Optional[bool] = None
    is_channel: Optional[bool] = None
    is_group: Optional[bool] = None
    is_im: Optional[bool] = None
    user: Optional[str] = None
    topic: Optional[PurposeTopic] = None
    purpose: Optional[PurposeTopic] = None
    previous_names: Optional[List[str]] = None

    @property
    def identifier(self) -> str:
        if self.name:
            return self.name
        else:
            return self.id

Storage

Storage is exposed to plugins through the self.storage field. The following class implements the interface plugins can use to interact with the storage backend.

`machine.storage.PluginStorage`

Class providing access to persistent storage for plugins

This class is the main access point for plugins to work with persistent storage. It is accessible from plugins using self.storage. Data is serialized before sending it to the storage backend, and deserialized upon retrieval. Serialization is done by dill_, so pretty much any Python object can be stored and retrieved.

.. _Dill: https://pypi.python.org/pypi/dill

Source code in machine/storage/__init__.py

class PluginStorage:
    """Class providing access to persistent storage for plugins

    This class is the main access point for plugins to work with persistent storage. It is
    accessible from plugins using ``self.storage``. Data is serialized before sending it to
    the storage backend, and deserialized upon retrieval. Serialization is done by `dill`_, so
    pretty much any Python object can be stored and retrieved.

    .. _Dill: https://pypi.python.org/pypi/dill
    """

    def __init__(self, fq_plugin_name: str, storage_backend: MachineBaseStorage):
        self._fq_plugin_name = fq_plugin_name
        self._storage = storage_backend

    def _gen_unique_key(self, key: str) -> str:
        return f"{self._fq_plugin_name}:{key}"

    def _namespace_key(self, key: str, shared: bool = False) -> str:
        return key if shared else self._gen_unique_key(key)

    async def set(self, key: str, value: Any, expires: int | timedelta | None = None, shared: bool = False) -> None:
        """Store or update a value by key

        :param key: the key under which to store the data
        :param value: the data to store
        :param expires: optional number of seconds after which the data is expired
        :param shared: ``True/False`` wether this data should be shared by other plugins.  Use with
            care, because it pollutes the global namespace of the storage.
        """
        expires = int(expires.total_seconds()) if isinstance(expires, timedelta) else expires
        namespaced_key = self._namespace_key(key, shared)
        pickled_value = dill.dumps(value)
        await self._storage.set(namespaced_key, pickled_value, expires)

    async def get(self, key: str, shared: bool = False) -> Any | None:
        """Retrieve data by key

        :param key: key for the data to retrieve
        :param shared: ``True/False`` wether to retrieve data from the shared (global) namespace.
        :return: the data, or ``None`` if the key cannot be found/has expired
        """
        namespaced_key = self._namespace_key(key, shared)
        value = await self._storage.get(namespaced_key)
        if value:
            return dill.loads(value)
        else:
            return None

    async def has(self, key: str, shared: bool = False) -> bool:
        """Check if the key exists in storage

        Note: this class implements ``__contains__`` so instead of calling
        ``self.storage.has(...)``, you can also use: ``key in self.storage``. This will check the
        *namespaced* version of the key, so it's the same as:
        ``self.storage.has('key', shared=False)``

        :param key: key to check
        :param shared: ``True/False`` wether to check in the shared (global) namespace
        :return: ``True/False`` wether the key exists. Can only return ``True`` if the key has not
            expired.
        """
        namespaced_key = self._namespace_key(key, shared)
        return await self._storage.has(namespaced_key)

    async def delete(self, key: str, shared: bool = False) -> None:
        """Remove a key and its data from storage

        :param key: key to remove
        :param shared: ``True/False`` wether the key to remove should be in the shared (global)
            namespace
        """
        namespaced_key = self._namespace_key(key, shared)
        await self._storage.delete(namespaced_key)

    async def get_storage_size(self) -> int:
        """Calculate the total size of the storage

        :return: the total size of the storage in bytes (integer)
        """
        return await self._storage.size()

    async def get_storage_size_human(self) -> str:
        """Calculate the total size of the storage in human readable format

        :return: the total size of the storage in a human readable string, rounded to the nearest
            applicable division. eg. B for Bytes, KiB for Kilobytes, MiB for Megabytes etc.
        """
        size = await self.get_storage_size()
        return sizeof_fmt(size)

`set(key: str, value: Any, expires: int | timedelta | None = None, shared: bool = False) -> None` `async`

Store or update a value by key

Parameters:

Name	Type	Description	Default
`key`	`str`	the key under which to store the data	required
`value`	`Any`	the data to store	required
`expires`	`int \| timedelta \| None`	optional number of seconds after which the data is expired	`None`
`shared`	`bool`	`True/False` wether this data should be shared by other plugins. Use with care, because it pollutes the global namespace of the storage.	`False`

Source code in machine/storage/__init__.py

async def set(self, key: str, value: Any, expires: int | timedelta | None = None, shared: bool = False) -> None:
    """Store or update a value by key

    :param key: the key under which to store the data
    :param value: the data to store
    :param expires: optional number of seconds after which the data is expired
    :param shared: ``True/False`` wether this data should be shared by other plugins.  Use with
        care, because it pollutes the global namespace of the storage.
    """
    expires = int(expires.total_seconds()) if isinstance(expires, timedelta) else expires
    namespaced_key = self._namespace_key(key, shared)
    pickled_value = dill.dumps(value)
    await self._storage.set(namespaced_key, pickled_value, expires)

`get(key: str, shared: bool = False) -> Any | None` `async`

Retrieve data by key

Parameters:

Name	Type	Description	Default
`key`	`str`	key for the data to retrieve	required
`shared`	`bool`	`True/False` wether to retrieve data from the shared (global) namespace.	`False`

Returns:

Type	Description
`Any \| None`	the data, or `None` if the key cannot be found/has expired

Source code in machine/storage/__init__.py

async def get(self, key: str, shared: bool = False) -> Any | None:
    """Retrieve data by key

    :param key: key for the data to retrieve
    :param shared: ``True/False`` wether to retrieve data from the shared (global) namespace.
    :return: the data, or ``None`` if the key cannot be found/has expired
    """
    namespaced_key = self._namespace_key(key, shared)
    value = await self._storage.get(namespaced_key)
    if value:
        return dill.loads(value)
    else:
        return None

`has(key: str, shared: bool = False) -> bool` `async`

Check if the key exists in storage

Note: this class implements __contains__ so instead of calling self.storage.has(...), you can also use: key in self.storage. This will check the namespaced version of the key, so it's the same as: self.storage.has('key', shared=False)

Parameters:

Name	Type	Description	Default
`key`	`str`	key to check	required
`shared`	`bool`	`True/False` wether to check in the shared (global) namespace	`False`

Returns:

Type	Description
`bool`	`True/False` wether the key exists. Can only return `True` if the key has not expired.

Source code in machine/storage/__init__.py

async def has(self, key: str, shared: bool = False) -> bool:
    """Check if the key exists in storage

    Note: this class implements ``__contains__`` so instead of calling
    ``self.storage.has(...)``, you can also use: ``key in self.storage``. This will check the
    *namespaced* version of the key, so it's the same as:
    ``self.storage.has('key', shared=False)``

    :param key: key to check
    :param shared: ``True/False`` wether to check in the shared (global) namespace
    :return: ``True/False`` wether the key exists. Can only return ``True`` if the key has not
        expired.
    """
    namespaced_key = self._namespace_key(key, shared)
    return await self._storage.has(namespaced_key)

`delete(key: str, shared: bool = False) -> None` `async`

Remove a key and its data from storage

Parameters:

Name	Type	Description	Default
`key`	`str`	key to remove	required
`shared`	`bool`	`True/False` wether the key to remove should be in the shared (global) namespace	`False`

Source code in machine/storage/__init__.py

async def delete(self, key: str, shared: bool = False) -> None:
    """Remove a key and its data from storage

    :param key: key to remove
    :param shared: ``True/False`` wether the key to remove should be in the shared (global)
        namespace
    """
    namespaced_key = self._namespace_key(key, shared)
    await self._storage.delete(namespaced_key)

`get_storage_size() -> int` `async`

Calculate the total size of the storage

Returns:

Type	Description
`int`	the total size of the storage in bytes (integer)

Source code in machine/storage/__init__.py

async def get_storage_size(self) -> int:
    """Calculate the total size of the storage

    :return: the total size of the storage in bytes (integer)
    """
    return await self._storage.size()

`get_storage_size_human() -> str` `async`

Calculate the total size of the storage in human readable format

Returns:

Type	Description
`str`	the total size of the storage in a human readable string, rounded to the nearest applicable division. eg. B for Bytes, KiB for Kilobytes, MiB for Megabytes etc.

Source code in machine/storage/__init__.py

async def get_storage_size_human(self) -> str:
    """Calculate the total size of the storage in human readable format

    :return: the total size of the storage in a human readable string, rounded to the nearest
        applicable division. eg. B for Bytes, KiB for Kilobytes, MiB for Megabytes etc.
    """
    size = await self.get_storage_size()
    return sizeof_fmt(size)

New Storage Backends can be implemented by extending the following class:

`machine.storage.backends.base.MachineBaseStorage`

Bases: ABC

Base class for storage backends

Extending classes should implement the five methods in this base class. Slack Machine takes care of a lot of details regarding the persistent storage of data. So storage backends do not have to deal with the following, because Slack Machine takes care of these:

Serialization/Deserialization of data
Namespacing of keys (so data stored by different plugins doesn't clash)

Source code in machine/storage/backends/base.py

class MachineBaseStorage(ABC):
    """Base class for storage backends

    Extending classes should implement the five methods in this base class. Slack Machine takes
    care of a lot of details regarding the persistent storage of data. So storage backends
    **do not** have to deal with the following, because Slack Machine takes care of these:

    - Serialization/Deserialization of data
    - Namespacing of keys (so data stored by different plugins doesn't clash)
    """

    settings: Mapping[str, Any]

    def __init__(self, settings: Mapping[str, Any]):
        self.settings = settings

    async def init(self) -> None:  # noqa: B027 (no-op by design)
        """Initialize the storage backend"""
        pass

    @abstractmethod
    async def get(self, key: str) -> bytes | None:
        """Retrieve data by key

        :param key: key for which to retrieve data
        :return: the raw data for the provided key, as (byte)string. Should return ``None`` when
            the key is unknown or the data has expired.
        """
        ...

    @abstractmethod
    async def set(self, key: str, value: bytes, expires: int | None = None) -> None:
        """Store data by key

        :param key: the key under which to store the data
        :param value: data as (byte)string
        :param expires: optional expiration time in seconds, after which the data should not be
            returned any more.
        """
        ...

    @abstractmethod
    async def delete(self, key: str) -> None:
        """Delete data by key

        :param key: key for which to delete the data
        """
        ...

    @abstractmethod
    async def has(self, key: str) -> bool:
        """Check if the key exists

        :param key: key to check
        :return: ``True/False`` wether the key exists
        """
        ...

    @abstractmethod
    async def size(self) -> int:
        """Calculate the total size of the storage

        :return: total size of storage in bytes (integer)
        """
        ...

    @abstractmethod
    async def close(self) -> None:
        """Close the storage backend"""
        ...

`init() -> None` `async`

Initialize the storage backend

Source code in machine/storage/backends/base.py

async def init(self) -> None:  # noqa: B027 (no-op by design)
    """Initialize the storage backend"""
    pass

`get(key: str) -> bytes | None` `abstractmethod` `async`

Retrieve data by key

Parameters:

Name	Type	Description	Default
`key`	`str`	key for which to retrieve data	required

Returns:

Type	Description
`bytes \| None`	the raw data for the provided key, as (byte)string. Should return `None` when the key is unknown or the data has expired.

Source code in machine/storage/backends/base.py

@abstractmethod
async def get(self, key: str) -> bytes | None:
    """Retrieve data by key

    :param key: key for which to retrieve data
    :return: the raw data for the provided key, as (byte)string. Should return ``None`` when
        the key is unknown or the data has expired.
    """
    ...

`set(key: str, value: bytes, expires: int | None = None) -> None` `abstractmethod` `async`

Store data by key

Parameters:

Name	Type	Description	Default
`key`	`str`	the key under which to store the data	required
`value`	`bytes`	data as (byte)string	required
`expires`	`int \| None`	optional expiration time in seconds, after which the data should not be returned any more.	`None`

Source code in machine/storage/backends/base.py

@abstractmethod
async def set(self, key: str, value: bytes, expires: int | None = None) -> None:
    """Store data by key

    :param key: the key under which to store the data
    :param value: data as (byte)string
    :param expires: optional expiration time in seconds, after which the data should not be
        returned any more.
    """
    ...

`delete(key: str) -> None` `abstractmethod` `async`

Delete data by key

Parameters:

Name	Type	Description	Default
`key`	`str`	key for which to delete the data	required

Source code in machine/storage/backends/base.py

@abstractmethod
async def delete(self, key: str) -> None:
    """Delete data by key

    :param key: key for which to delete the data
    """
    ...

`has(key: str) -> bool` `abstractmethod` `async`

Check if the key exists

Parameters:

Name	Type	Description	Default
`key`	`str`	key to check	required

Returns:

Type	Description
`bool`	`True/False` wether the key exists

Source code in machine/storage/backends/base.py

@abstractmethod
async def has(self, key: str) -> bool:
    """Check if the key exists

    :param key: key to check
    :return: ``True/False`` wether the key exists
    """
    ...

`size() -> int` `abstractmethod` `async`

Calculate the total size of the storage

Returns:

Type	Description
`int`	total size of storage in bytes (integer)

Source code in machine/storage/backends/base.py

@abstractmethod
async def size(self) -> int:
    """Calculate the total size of the storage

    :return: total size of storage in bytes (integer)
    """
    ...

`close() -> None` `abstractmethod` `async`

Close the storage backend

Source code in machine/storage/backends/base.py

@abstractmethod
async def close(self) -> None:
    """Close the storage backend"""
    ...

API Documentation

Plugin classes

machine.plugins.base.MachineBasePlugin

users: dict[str, User] property

users_by_email: dict[str, User] property

channels: dict[str, Channel] property

web_client: AsyncWebClient property

bot_info: dict[str, Any] property

init() -> None async

find_channel_by_name(channel_name: str) -> Channel | None

get_user_by_id(user_id: str) -> User | None

get_user_by_email(email: str) -> User | None

at(user: User) -> str

say_scheduled(when: datetime, channel: Channel | str, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, **kwargs: Any) -> AsyncSlackResponse async

update(channel: Channel | str, ts: str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, ephemeral_user: User | str | None = None, **kwargs: Any) -> AsyncSlackResponse async

delete(channel: Channel | str, ts: str, **kwargs: Any) -> AsyncSlackResponse async

react(channel: Channel | str, ts: str, emoji: str) -> AsyncSlackResponse async

send_dm(user: User | str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

send_dm_scheduled(when: datetime, user: User | str, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

emit(event: str, **kwargs: Any) -> None

pin_message(channel: Channel | str, ts: str) -> AsyncSlackResponse async

unpin_message(channel: Channel | str, ts: str) -> AsyncSlackResponse async

set_topic(channel: Channel | str, topic: str, **kwargs: Any) -> AsyncSlackResponse async

machine.plugins.message.Message

sender: User property

channel: Channel property

text: str property

at_sender: str property

ts: str property

in_thread: bool property

say(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, ephemeral: bool = False, **kwargs: Any) -> AsyncSlackResponse async

say_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, **kwargs: Any) -> AsyncSlackResponse async

reply(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, in_thread: bool = False, ephemeral: bool = False, **kwargs: Any) -> AsyncSlackResponse async

reply_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, in_thread: bool = False, **kwargs: Any) -> AsyncSlackResponse async

reply_dm(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

reply_dm_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse async

react(emoji: str) -> AsyncSlackResponse async

pin_message() -> AsyncSlackResponse async

machine.plugins.command.Command

sender: User property

channel: Channel property

text: str property

command: str property

response_url: str property

trigger_id: str property

say(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, ephemeral: bool = True, **kwargs: Any) -> WebhookResponse async

machine.plugins.block_action.BlockAction

payload: BlockActionsPayload = payload instance-attribute

triggered_action: Action = triggered_action instance-attribute

user: User property

channel: Optional[Channel] property

state: Optional[State] property

response_url: Optional[str] property

trigger_id: str property

Decorators

machine.plugins.decorators

process(slack_event_type: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

listen_to(regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

respond_to(regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

command(slash_command: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

action(action_id: Union[re.Pattern[str], str, None] = None, block_id: Union[re.Pattern[str], str, None] = None) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

on(event: str) -> Callable[[Callable[P, R]], Callable[P, R]]

required_settings(settings: list[str] | str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

require_any_role(required_roles: list[str]) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]

require_all_roles(required_roles: list[str]) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]

Models

machine.models.user.User

machine.models.channel.Channel

Storage

machine.storage.PluginStorage

set(key: str, value: Any, expires: int | timedelta | None = None, shared: bool = False) -> None async

get(key: str, shared: bool = False) -> Any | None async

has(key: str, shared: bool = False) -> bool async

delete(key: str, shared: bool = False) -> None async

get_storage_size() -> int async

get_storage_size_human() -> str async

machine.storage.backends.base.MachineBaseStorage

init() -> None async

get(key: str) -> bytes | None abstractmethod async

set(key: str, value: bytes, expires: int | None = None) -> None abstractmethod async

`machine.plugins.base.MachineBasePlugin`

`users: dict[str, User]` `property`

`users_by_email: dict[str, User]` `property`

`channels: dict[str, Channel]` `property`

`web_client: AsyncWebClient` `property`

`bot_info: dict[str, Any]` `property`

`init() -> None` `async`

`find_channel_by_name(channel_name: str) -> Channel | None`

`get_user_by_id(user_id: str) -> User | None`

`get_user_by_email(email: str) -> User | None`

`at(user: User) -> str`

`say_scheduled(when: datetime, channel: Channel | str, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

`update(channel: Channel | str, ts: str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, ephemeral_user: User | str | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

`delete(channel: Channel | str, ts: str, **kwargs: Any) -> AsyncSlackResponse` `async`

`react(channel: Channel | str, ts: str, emoji: str) -> AsyncSlackResponse` `async`

`send_dm(user: User | str, text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

`send_dm_scheduled(when: datetime, user: User | str, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

`emit(event: str, **kwargs: Any) -> None`

`pin_message(channel: Channel | str, ts: str) -> AsyncSlackResponse` `async`

`unpin_message(channel: Channel | str, ts: str) -> AsyncSlackResponse` `async`

`set_topic(channel: Channel | str, topic: str, **kwargs: Any) -> AsyncSlackResponse` `async`

`machine.plugins.message.Message`

`sender: User` `property`

`channel: Channel` `property`

`text: str` `property`

`at_sender: str` `property`

`ts: str` `property`

`in_thread: bool` `property`

`say(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, ephemeral: bool = False, **kwargs: Any) -> AsyncSlackResponse` `async`

`say_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, thread_ts: str | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

`reply(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, in_thread: bool = False, ephemeral: bool = False, **kwargs: Any) -> AsyncSlackResponse` `async`

`reply_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, in_thread: bool = False, **kwargs: Any) -> AsyncSlackResponse` `async`

`reply_dm(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

`reply_dm_scheduled(when: datetime, text: str, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, **kwargs: Any) -> AsyncSlackResponse` `async`

`react(emoji: str) -> AsyncSlackResponse` `async`

`pin_message() -> AsyncSlackResponse` `async`

`machine.plugins.command.Command`

`sender: User` `property`

`channel: Channel` `property`

`text: str` `property`

`command: str` `property`

`response_url: str` `property`

`trigger_id: str` `property`

`say(text: str | None = None, attachments: Sequence[Attachment] | Sequence[dict[str, Any]] | None = None, blocks: Sequence[Block] | Sequence[dict[str, Any]] | None = None, ephemeral: bool = True, **kwargs: Any) -> WebhookResponse` `async`

`machine.plugins.block_action.BlockAction`

`payload: BlockActionsPayload = payload` `instance-attribute`

`triggered_action: Action = triggered_action` `instance-attribute`

`user: User` `property`

`channel: Optional[Channel]` `property`

`state: Optional[State]` `property`

`response_url: Optional[str]` `property`

`trigger_id: str` `property`

`machine.plugins.decorators`

`process(slack_event_type: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

`listen_to(regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

`respond_to(regex: str, flags: re.RegexFlag | int = re.IGNORECASE, handle_message_changed: bool = False) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

`command(slash_command: str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

`action(action_id: Union[re.Pattern[str], str, None] = None, block_id: Union[re.Pattern[str], str, None] = None) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

`on(event: str) -> Callable[[Callable[P, R]], Callable[P, R]]`

`required_settings(settings: list[str] | str) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]`

`require_any_role(required_roles: list[str]) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]`

`require_all_roles(required_roles: list[str]) -> Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]`

`machine.models.user.User`

`machine.models.channel.Channel`

`machine.storage.PluginStorage`

`set(key: str, value: Any, expires: int | timedelta | None = None, shared: bool = False) -> None` `async`

`get(key: str, shared: bool = False) -> Any | None` `async`

`has(key: str, shared: bool = False) -> bool` `async`

`delete(key: str, shared: bool = False) -> None` `async`

`get_storage_size() -> int` `async`

`get_storage_size_human() -> str` `async`

`machine.storage.backends.base.MachineBaseStorage`

`init() -> None` `async`

`get(key: str) -> bytes | None` `abstractmethod` `async`

`set(key: str, value: bytes, expires: int | None = None) -> None` `abstractmethod` `async`

`delete(key: str) -> None` `abstractmethod` `async`

`has(key: str) -> bool` `abstractmethod` `async`

`size() -> int` `abstractmethod` `async`

`close() -> None` `abstractmethod` `async`