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:

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

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.
storage	PluginStorage	Plugin storage object that allows plugins to store and retrieve data

Source code in src/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

    Attributes:
        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.
        storage: Plugin storage object that allows plugins to store and retrieve data
    """

    _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__}"

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

        Returns:
            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

        Returns:
            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.

        Returns:
            a list of all channels in the Slack workspace, where each channel is a
                [`Channel`][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](https://api.slack.com/web)

        Returns:
            an instance of [`AsyncWebClient`](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

    @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

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

    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

    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.

        Args:
            channel_name: The name of the channel to retrieve.

        Returns:
            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.

        Args:
            user_id: The ID of the user to retrieve.

        Returns:
            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.

        Args:
            email: The email address of the user to retrieve.

        Returns:
            The user if found, None otherwise.
        """
        return self._client.get_user_by_email(email)

    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
        [`say()`][machine.plugins.base.MachineBasePlugin.say].

        Args:
            user: user your want to mention

        Returns:
            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 [convenience 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
        [convenience classes]:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
        [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
        [chat.postEphemeral]: https://api.slack.com/methods/chat.postEphemeral

        Args:
            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.
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments)
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
            thread_ts: optional timestamp of thread, to send a message in that thread
            ephemeral_user: optional user name or id if the message needs to visible
                to a specific user only

        Returns:
            Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response,
                or [chat.postEphemeral](https://api.slack.com/methods/chat.postEphemeral) if `ephemeral_user` is set.
        """
        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 [`say()`][machine.plugins.base.MachineBasePlugin.say].
        It behaves the same, but will send the message at the scheduled time.

        Args:
            when: when you want the message to be sent, as [`datetime`][datetime.datetime] instance
            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.
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
            thread_ts: optional timestamp of thread, to send a message in that thread
        Returns:
            Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
                response.
        """
        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 [convenience 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
        [convenience classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes
        [chat_update]: https://api.slack.com/methods/chat.update

        Args:
            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.
            ts: timestamp of the message to be updated.
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
            ephemeral_user: optional user name or id if the message needs to visible
                to a specific user only

        Returns:
            Dictionary deserialized from [`chat.update`](https://api.slack.com/methods/chat.update) response
        """
        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

        Args:
            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.
            ts: timestamp of the message to be deleted.

        Returns:
            Dictionary deserialized from [`chat.delete`](https://api.slack.com/methods/chat.delete) response
        """
        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.

        Args:
            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.
            ts: timestamp of the message to react to
            emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)

        Returns:
            Dictionary deserialized from [reactions.add](https://api.slack.com/methods/reactions.add) response.
        """
        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 [convenience 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
        [convenience classes]:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
        [chat.postMessage]: https://api.slack.com/methods/chat.postMessage

        Args:
            user: [`User`][machine.models.user.User] object or id of user to send DM to.
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

        Returns:
            Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
        """
        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
        [`send_dm()`][machine.plugins.base.MachineBasePlugin.send_dm]. It behaves the same, but
        will send the DM at the scheduled time.

        Args:
            when: when you want the message to be sent, as [`datetime`][datetime.datetime] instance
            user: [`User`][machine.models.user.User] object or id of user to send DM to.
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

        Returns:
            Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
                response.
        """
        return await self._client.send_dm_scheduled(
            when, user, text=text, attachments=attachments, blocks=blocks, **kwargs
        )

    async def open_im(self, users: User | str | list[User | str]) -> str:
        """Open a DM channel with one or more users

        Open a DM channel with one or more users. If the DM channel already exists, the existing channel id
        will be returned. If the DM channel does not exist, a new channel will be created and the
        id of the new channel will be returned.

        Args:
            users: [`User`][machine.models.user.User] object or id of user to open DM with, or a list of user objects
                or user ids.

        Returns:
            id of the DM channel
        """
        return await self._client.open_im(users)

    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.

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

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

        Pin a message in a channel

        Args:
            channel: channel to pin the message in
            ts: timestamp of the message to pin

        Returns:
            Dictionary deserialized from [pins.add](https://api.slack.com/methods/pins.add) response.
        """
        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

        Args:
            channel: channel where the message is pinned that needs to be unpinned
            ts: timestamp of the message to unpin

        Returns:
            Dictionary deserialized from [pins.remove](https://api.slack.com/methods/pins.remove) response.
        """
        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

        Args:
            channel: channel where topic needs to be set or updated
            topic: topic for the channel (slack does not support formatting for topics)

        Returns:
            Dictionary deserialized from [conversations.setTopic](https://api.slack.com/methods/conversations.setTopic)
                response.
        """
        return await self._client.set_topic(channel, topic, **kwargs)

    async def open_modal(self, trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse:
        """Open a modal dialog

        Open a modal dialog in response to a user action. The modal dialog can be used to collect
        information from the user, or to display information to the user.

        Args:
            trigger_id: trigger id is provided by Slack when a user action is performed, such as a slash command
                or a button click
            view: view definition for the modal dialog

        Returns:
            Dictionary deserialized from [views.open](https://api.slack.com/methods/views.open) response.
        """
        return await self._client.web_client.views_open(trigger_id=trigger_id, view=view, **kwargs)

    async def push_modal(self, trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse:
        """Push a new view onto the stack of a modal that was already opened

        Push a new view onto the stack of a modal that was already opened by a open_modal call. At most 3 views can be
        active in a modal at the same time. For more information on the lifecycle of modals, refer to the
        [relevant Slack documentation](https://api.slack.com/surfaces/modals)

        Args:
            trigger_id: trigger id is provided by Slack when a user action is performed, such as a slash command
                or a button click
            view: view definition for the modal dialog

        Returns:
            Dictionary deserialized from [views.push](https://api.slack.com/methods/views.push) response.
        """
        return await self._client.push_modal(trigger_id=trigger_id, view=view, **kwargs)

    async def update_modal(
        self,
        view: dict | View,
        view_id: str | None = None,
        external_id: str | None = None,
        hash: str | None = None,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Update a modal dialog

        Update a modal dialog that was previously opened. You can update the view by providing the view_id or the
        external_id of the modal. external_id has precedence over view_id, but at least one needs to be provided.
        You can also provide a hash of the view that you want to update to prevent race conditions.

        Args:
            view: view definition for the modal dialog
            view_id: id of the view to update
            external_id: external id of the view to update
            hash: hash of the view to update
        Returns:
            Dictionary deserialized from [views.update](https://api.slack.com/methods/views.update) response.
        """
        return await self._client.update_modal(view=view, view_id=view_id, external_id=external_id, hash=hash, **kwargs)

    async def publish_home_tab(
        self, user: User | str, view: dict | View, hash: str | None = None, **kwargs: Any
    ) -> AsyncSlackResponse:
        """Publish a view to the home tab of a user

        Publish a view to the home tab of a user. The view will be visible to the user when they open the home tab of
        your Slack app. This method can be used both to publish a new view for the home tab or update an existing view.
        You can provide a hash of the view that you want to update to prevent race conditions.

        Warning:
            Be careful with the use of this method, as you might be overwriting the user's home tab that was set by
            another Slack Machine plugin enabled in your bot.

        Args:
            user: user for whom to publish or update the home tab
            view: view definition for the home tab
            hash: hash of the view to update

        Returns:
            Dictionary deserialized from [views.publish](https://api.slack.com/methods/views.publish) response.
        """
        return await self._client.publish_home_tab(user=user, view=view, hash=hash, **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 Channel object

web_client: AsyncWebClient property

Slack SDK web client to access the Slack Web API

Returns:

Type	Description
AsyncWebClient	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.

Source code in src/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

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 src/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.

    Args:
        channel_name: The name of the channel to retrieve.

    Returns:
        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 src/machine/plugins/base.py

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

    Args:
        user_id: The ID of the user to retrieve.

    Returns:
        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 src/machine/plugins/base.py

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

    Args:
        email: The email address of the user to retrieve.

    Returns:
        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 say().

Parameters:

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

Returns:

Type	Description
str	user mention

Source code in src/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
    [`say()`][machine.plugins.base.MachineBasePlugin.say].

    Args:
        user: user your want to mention

    Returns:
        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 convenience 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.

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
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 response, or chat.postEphemeral if ephemeral_user is set.

Source code in src/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 [convenience 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
    [convenience classes]:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
    [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
    [chat.postEphemeral]: https://api.slack.com/methods/chat.postEphemeral

    Args:
        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.
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments)
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
        thread_ts: optional timestamp of thread, to send a message in that thread
        ephemeral_user: optional user name or id if the message needs to visible
            to a specific user only

    Returns:
        Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response,
            or [chat.postEphemeral](https://api.slack.com/methods/chat.postEphemeral) if `ephemeral_user` is set.
    """
    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 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 datetime instance	required
channel	Channel | str	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: Dictionary deserialized from chat.scheduleMessage response.

Source code in src/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 [`say()`][machine.plugins.base.MachineBasePlugin.say].
    It behaves the same, but will send the message at the scheduled time.

    Args:
        when: when you want the message to be sent, as [`datetime`][datetime.datetime] instance
        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.
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
        thread_ts: optional timestamp of thread, to send a message in that thread
    Returns:
        Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
            response.
    """
    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 convenience 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
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 response

Source code in src/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 [convenience 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
    [convenience classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes
    [chat_update]: https://api.slack.com/methods/chat.update

    Args:
        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.
        ts: timestamp of the message to be updated.
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
        ephemeral_user: optional user name or id if the message needs to visible
            to a specific user only

    Returns:
        Dictionary deserialized from [`chat.update`](https://api.slack.com/methods/chat.update) response
    """
    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 response

Source code in src/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

    Args:
        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.
        ts: timestamp of the message to be deleted.

    Returns:
        Dictionary deserialized from [`chat.delete`](https://api.slack.com/methods/chat.delete) response
    """
    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	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 response.

Source code in src/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.

    Args:
        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.
        ts: timestamp of the message to react to
        emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)

    Returns:
        Dictionary deserialized from [reactions.add](https://api.slack.com/methods/reactions.add) response.
    """
    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 convenience classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

Parameters:

Name	Type	Description	Default
user	User | str	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 response.

Source code in src/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 [convenience 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
    [convenience classes]:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
    [chat.postMessage]: https://api.slack.com/methods/chat.postMessage

    Args:
        user: [`User`][machine.models.user.User] object or id of user to send DM to.
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

    Returns:
        Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
    """
    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 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 datetime instance	required
user	User | str	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	Dictionary deserialized from chat.scheduleMessage response.

Source code in src/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
    [`send_dm()`][machine.plugins.base.MachineBasePlugin.send_dm]. It behaves the same, but
    will send the DM at the scheduled time.

    Args:
        when: when you want the message to be sent, as [`datetime`][datetime.datetime] instance
        user: [`User`][machine.models.user.User] object or id of user to send DM to.
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

    Returns:
        Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
            response.
    """
    return await self._client.send_dm_scheduled(
        when, user, text=text, attachments=attachments, blocks=blocks, **kwargs
    )

open_im(users: User | str | list[User | str]) -> str async

Open a DM channel with one or more users

Open a DM channel with one or more users. If the DM channel already exists, the existing channel id will be returned. If the DM channel does not exist, a new channel will be created and the id of the new channel will be returned.

Parameters:

Name	Type	Description	Default
users	User | str | list[User | str]	User object or id of user to open DM with, or a list of user objects or user ids.	required

Returns:

Type	Description
str	id of the DM channel

Source code in src/machine/plugins/base.py

async def open_im(self, users: User | str | list[User | str]) -> str:
    """Open a DM channel with one or more users

    Open a DM channel with one or more users. If the DM channel already exists, the existing channel id
    will be returned. If the DM channel does not exist, a new channel will be created and the
    id of the new channel will be returned.

    Args:
        users: [`User`][machine.models.user.User] object or id of user to open DM with, or a list of user objects
            or user ids.

    Returns:
        id of the DM channel
    """
    return await self._client.open_im(users)

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	{}

Source code in src/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.

    Args:
        event: name of the event
        **kwargs: any data you want to emit with the event
    """
    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	Dictionary deserialized from pins.add response.

Source code in src/machine/plugins/base.py

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

    Pin a message in a channel

    Args:
        channel: channel to pin the message in
        ts: timestamp of the message to pin

    Returns:
        Dictionary deserialized from [pins.add](https://api.slack.com/methods/pins.add) response.
    """
    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	Dictionary deserialized from pins.remove response.

Source code in src/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

    Args:
        channel: channel where the message is pinned that needs to be unpinned
        ts: timestamp of the message to unpin

    Returns:
        Dictionary deserialized from [pins.remove](https://api.slack.com/methods/pins.remove) response.
    """
    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	Dictionary deserialized from conversations.setTopic response.

Source code in src/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

    Args:
        channel: channel where topic needs to be set or updated
        topic: topic for the channel (slack does not support formatting for topics)

    Returns:
        Dictionary deserialized from [conversations.setTopic](https://api.slack.com/methods/conversations.setTopic)
            response.
    """
    return await self._client.set_topic(channel, topic, **kwargs)

open_modal(trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse async

Open a modal dialog

Open a modal dialog in response to a user action. The modal dialog can be used to collect information from the user, or to display information to the user.

Parameters:

Name	Type	Description	Default
trigger_id	str	trigger id is provided by Slack when a user action is performed, such as a slash command or a button click	required
view	dict | View	view definition for the modal dialog	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from views.open response.

Source code in src/machine/plugins/base.py

async def open_modal(self, trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse:
    """Open a modal dialog

    Open a modal dialog in response to a user action. The modal dialog can be used to collect
    information from the user, or to display information to the user.

    Args:
        trigger_id: trigger id is provided by Slack when a user action is performed, such as a slash command
            or a button click
        view: view definition for the modal dialog

    Returns:
        Dictionary deserialized from [views.open](https://api.slack.com/methods/views.open) response.
    """
    return await self._client.web_client.views_open(trigger_id=trigger_id, view=view, **kwargs)

push_modal(trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse async

Push a new view onto the stack of a modal that was already opened

Push a new view onto the stack of a modal that was already opened by a open_modal call. At most 3 views can be active in a modal at the same time. For more information on the lifecycle of modals, refer to the relevant Slack documentation

Parameters:

Name	Type	Description	Default
trigger_id	str	trigger id is provided by Slack when a user action is performed, such as a slash command or a button click	required
view	dict | View	view definition for the modal dialog	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from views.push response.

Source code in src/machine/plugins/base.py

async def push_modal(self, trigger_id: str, view: dict | View, **kwargs: Any) -> AsyncSlackResponse:
    """Push a new view onto the stack of a modal that was already opened

    Push a new view onto the stack of a modal that was already opened by a open_modal call. At most 3 views can be
    active in a modal at the same time. For more information on the lifecycle of modals, refer to the
    [relevant Slack documentation](https://api.slack.com/surfaces/modals)

    Args:
        trigger_id: trigger id is provided by Slack when a user action is performed, such as a slash command
            or a button click
        view: view definition for the modal dialog

    Returns:
        Dictionary deserialized from [views.push](https://api.slack.com/methods/views.push) response.
    """
    return await self._client.push_modal(trigger_id=trigger_id, view=view, **kwargs)

update_modal(view: dict | View, view_id: str | None = None, external_id: str | None = None, hash: str | None = None, **kwargs: Any) -> AsyncSlackResponse async

Update a modal dialog

Update a modal dialog that was previously opened. You can update the view by providing the view_id or the external_id of the modal. external_id has precedence over view_id, but at least one needs to be provided. You can also provide a hash of the view that you want to update to prevent race conditions.

Parameters:

Name	Type	Description	Default
view	dict | View	view definition for the modal dialog	required
view_id	str | None	id of the view to update	None
external_id	str | None	external id of the view to update	None
hash	str | None	hash of the view to update	None

Returns: Dictionary deserialized from views.update response.

Source code in src/machine/plugins/base.py

async def update_modal(
    self,
    view: dict | View,
    view_id: str | None = None,
    external_id: str | None = None,
    hash: str | None = None,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Update a modal dialog

    Update a modal dialog that was previously opened. You can update the view by providing the view_id or the
    external_id of the modal. external_id has precedence over view_id, but at least one needs to be provided.
    You can also provide a hash of the view that you want to update to prevent race conditions.

    Args:
        view: view definition for the modal dialog
        view_id: id of the view to update
        external_id: external id of the view to update
        hash: hash of the view to update
    Returns:
        Dictionary deserialized from [views.update](https://api.slack.com/methods/views.update) response.
    """
    return await self._client.update_modal(view=view, view_id=view_id, external_id=external_id, hash=hash, **kwargs)

publish_home_tab(user: User | str, view: dict | View, hash: str | None = None, **kwargs: Any) -> AsyncSlackResponse async

Publish a view to the home tab of a user

Publish a view to the home tab of a user. The view will be visible to the user when they open the home tab of your Slack app. This method can be used both to publish a new view for the home tab or update an existing view. You can provide a hash of the view that you want to update to prevent race conditions.

Warning

Be careful with the use of this method, as you might be overwriting the user's home tab that was set by another Slack Machine plugin enabled in your bot.

Parameters:

Name	Type	Description	Default
user	User | str	user for whom to publish or update the home tab	required
view	dict | View	view definition for the home tab	required
hash	str | None	hash of the view to update	None

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from views.publish response.

Source code in src/machine/plugins/base.py

async def publish_home_tab(
    self, user: User | str, view: dict | View, hash: str | None = None, **kwargs: Any
) -> AsyncSlackResponse:
    """Publish a view to the home tab of a user

    Publish a view to the home tab of a user. The view will be visible to the user when they open the home tab of
    your Slack app. This method can be used both to publish a new view for the home tab or update an existing view.
    You can provide a hash of the view that you want to update to prevent race conditions.

    Warning:
        Be careful with the use of this method, as you might be overwriting the user's home tab that was set by
        another Slack Machine plugin enabled in your bot.

    Args:
        user: user for whom to publish or update the home tab
        view: view definition for the home tab
        hash: hash of the view to update

    Returns:
        Dictionary deserialized from [views.publish](https://api.slack.com/methods/views.publish) response.
    """
    return await self._client.publish_home_tab(user=user, view=view, hash=hash, **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 src/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

        Returns:
            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

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

    @property
    def is_dm(self) -> bool:
        """Is the message a direct message

        Returns:
            `True` if the message was _not_ sent in a channel or group, `False` otherwise
        """
        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

        Returns:
            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

        Returns:
            a string representation of the sender of the message, formatted as
                [mention](https://api.slack.com/docs/message-formatting#linking_to_channels_and_users),
                to be used in messages
        """
        return self.sender.fmt_mention()

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

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

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

        Returns:
            the message is in a thread
        """
        return "thread_ts" in self._msg_event

    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 [convenience 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
        [convenience classes]:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
        [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
        [chat.postEphemeral]: https://api.slack.com/methods/chat.postEphemeral

        Args:
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
            thread_ts: optional timestamp of thread, to send a message in that thread
            ephemeral: `True/False` wether to send the message as an ephemeral message, only
                visible to the sender of the original message

        Returns:
            Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response,
                or [chat.postEphemeral](https://api.slack.com/methods/chat.postEphemeral) if `ephemeral` is True.


        """
        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 [`say()`][machine.plugins.message.Message.say].
        It behaves the same, but will send the message at the scheduled time.

        Args:
            when: when you want the message to be sent, as [`datetime`][datetime.datetime] instance
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
            thread_ts: optional timestamp of thread, to send a message in that thread

        Returns:
            Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
                response.
        """
        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 [convenience 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
        [convenience classes]:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
        [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
        [chat.postEphemeral]: https://api.slack.com/methods/chat.postEphemeral

        Args:
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks)
            in_thread: `True/False` wether to reply to the original message in-thread
            ephemeral: `True/False` wether to send the message as an ephemeral message, only
                visible to the sender of the original message

        Returns:
            Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response,
                or [chat.postEphemeral](https://api.slack.com/methods/chat.postEphemeral) if `ephemeral` is True.
        """
        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 [`reply()`][machine.plugins.message.Message.reply].
        It behaves the same, but will send the reply at the scheduled time.

        Args:
            when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
            in_thread: `True/False` wether to reply to the original message in-thread

        Returns:
            Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
                response.
        """
        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
        [convenience 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
        [convenience classes]:
            https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
        [chat.postMessage]: https://api.slack.com/methods/chat.postMessage

        Args:
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

        Returns:
            Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
        """
        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 [`reply_dm()`][machine.plugins.message.Message.reply_dm].
        It behaves the same, but will send the DM at the scheduled time.

        Args:
            when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

        Returns:
            Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
                response.
        """
        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

        Args:
            emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)

        Returns:
            Dictionary deserialized from [reactions.add](https://api.slack.com/methods/reactions.add) response.
        """
        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

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

        Pin the current message in the channel it was posted in

        Returns:
            Dictionary deserialized from [pins.add](https://api.slack.com/methods/pins.add) response.
        """
        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

is_dm: bool property

Is the message a direct message

Returns:

Type	Description
bool	True if the message was not sent in a channel or group, False otherwise

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

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	the message is in a thread

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 convenience 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.

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 response, or chat.postEphemeral if ephemeral is True.

Source code in src/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 [convenience 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
    [convenience classes]:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
    [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
    [chat.postEphemeral]: https://api.slack.com/methods/chat.postEphemeral

    Args:
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
        thread_ts: optional timestamp of thread, to send a message in that thread
        ephemeral: `True/False` wether to send the message as an ephemeral message, only
            visible to the sender of the original message

    Returns:
        Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response,
            or [chat.postEphemeral](https://api.slack.com/methods/chat.postEphemeral) if `ephemeral` is True.


    """
    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 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 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	Dictionary deserialized from chat.scheduleMessage response.

Source code in src/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 [`say()`][machine.plugins.message.Message.say].
    It behaves the same, but will send the message at the scheduled time.

    Args:
        when: when you want the message to be sent, as [`datetime`][datetime.datetime] instance
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
        thread_ts: optional timestamp of thread, to send a message in that thread

    Returns:
        Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
            response.
    """
    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 convenience 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.

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 response, or chat.postEphemeral if ephemeral is True.

Source code in src/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 [convenience 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
    [convenience classes]:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
    [chat.postMessage]: https://api.slack.com/methods/chat.postMessage
    [chat.postEphemeral]: https://api.slack.com/methods/chat.postEphemeral

    Args:
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks)
        in_thread: `True/False` wether to reply to the original message in-thread
        ephemeral: `True/False` wether to send the message as an ephemeral message, only
            visible to the sender of the original message

    Returns:
        Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response,
            or [chat.postEphemeral](https://api.slack.com/methods/chat.postEphemeral) if `ephemeral` is True.
    """
    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 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	Dictionary deserialized from chat.scheduleMessage response.

Source code in src/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 [`reply()`][machine.plugins.message.Message.reply].
    It behaves the same, but will send the reply at the scheduled time.

    Args:
        when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
        in_thread: `True/False` wether to reply to the original message in-thread

    Returns:
        Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
            response.
    """
    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 convenience classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

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 response.

Source code in src/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
    [convenience 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
    [convenience classes]:
        https://github.com/slackapi/python-slackclient/tree/master/slack/web/classes
    [chat.postMessage]: https://api.slack.com/methods/chat.postMessage

    Args:
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

    Returns:
        Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
    """
    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 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	Dictionary deserialized from chat.scheduleMessage response.

Source code in src/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 [`reply_dm()`][machine.plugins.message.Message.reply_dm].
    It behaves the same, but will send the DM at the scheduled time.

    Args:
        when: when you want the message to be sent, as :py:class:`datetime.datetime` instance
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

    Returns:
        Dictionary deserialized from [chat.scheduleMessage](https://api.slack.com/methods/chat.scheduleMessage)
            response.
    """
    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 response.

Source code in src/machine/plugins/message.py

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

    Add a reaction to the original message

    Args:
        emoji: what emoji to react with (should be a string, like 'angel', 'thumbsup', etc.)

    Returns:
        Dictionary deserialized from [reactions.add](https://api.slack.com/methods/reactions.add) response.
    """
    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

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from pins.add response.

Source code in src/machine/plugins/message.py

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

    Pin the current message in the channel it was posted in

    Returns:
        Dictionary deserialized from [pins.add](https://api.slack.com/methods/pins.add) response.
    """
    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 src/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 user that invoked the command

        Returns:
            the User that invoked the command
        """
        return self._client.users[self._cmd_payload["user_id"]]

    @property
    def channel(self) -> Channel:
        """The channel the command was invoked in

        Returns:
            the Channel the command was invoked in
        """
        return self._client.channels[self._cmd_payload["channel_id"]]

    @property
    def is_dm(self) -> bool:
        """Whether the command was invoked in a DM

        Returns:
            `True` if the message was _not_ invoked in a channel or group, `False` otherwise
        """
        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 command (i.e. anything after the command itself)

        Returns:
            the body (text) of the command
        """
        return self._cmd_payload["text"]

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

        Returns:
            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.

        Returns:
            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

        Note:
            The `trigger_id` is only valid for 3 seconds after the modal was submitted.

            You can use [`open_modal`][machine.plugins.command.Command.open_modal] to open a modal instead of using
            the `trigger_id` directly.

        Returns:
            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 [convenience 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
        [convenience classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        Args:
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
            ephemeral: `True/False` wether to send the message as an ephemeral message, only
                visible to the sender of the original message

        Returns:
            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
        )

    async def open_modal(
        self,
        view: dict | View,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Open a modal in response to the command

        Open a modal in response to the command, using the trigger_id that was returned when the command was invoked.
        Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_open()`

        Note:
            You have to call this method within 3 seconds of receiving the command payload.

        Args:
            view: the view to open

        Returns:
            Dictionary deserialized from `AsyncWebClient.views_open()`
        """
        return await self._client.open_modal(self.trigger_id, view, **kwargs)

sender: User property

The user that invoked the command

Returns:

Type	Description
User	the User that invoked the command

channel: Channel property

The channel the command was invoked in

Returns:

Type	Description
Channel	the Channel the command was invoked in

is_dm: bool property

Whether the command was invoked in a DM

Returns:

Type	Description
bool	True if the message was not invoked in a channel or group, False otherwise

text: str property

The body of the command (i.e. anything after the command itself)

Returns:

Type	Description
str	the body (text) of the command

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

Note

The trigger_id is only valid for 3 seconds after the modal was submitted.

You can use open_modal to open a modal instead of using the trigger_id directly.

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 convenience 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 src/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 [convenience 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
    [convenience classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    Args:
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
        ephemeral: `True/False` wether to send the message as an ephemeral message, only
            visible to the sender of the original message

    Returns:
        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
    )

open_modal(view: dict | View, **kwargs: Any) -> AsyncSlackResponse async

Open a modal in response to the command

Open a modal in response to the command, using the trigger_id that was returned when the command was invoked. Any extra kwargs you provide, will be passed on directly to AsyncWebClient.views_open()

Note

You have to call this method within 3 seconds of receiving the command payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to open	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from AsyncWebClient.views_open()

Source code in src/machine/plugins/command.py

async def open_modal(
    self,
    view: dict | View,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Open a modal in response to the command

    Open a modal in response to the command, using the trigger_id that was returned when the command was invoked.
    Any extra kwargs you provide, will be passed on directly to `AsyncWebClient.views_open()`

    Note:
        You have to call this method within 3 seconds of receiving the command payload.

    Args:
        view: the view to open

    Returns:
        Dictionary deserialized from `AsyncWebClient.views_open()`
    """
    return await self._client.open_modal(self.trigger_id, view, **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.

Attributes:

Name	Type	Description
payload	BlockActionsPayload	The payload that was received by the bot when the action was triggered that this plugin method listens for
triggered_action	Action	The action that triggered this plugin method

Source code in src/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.

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

    payload: BlockActionsPayload
    triggered_action: Action

    def __init__(self, client: SlackClient, payload: BlockActionsPayload, triggered_action: Action):
        self._client = client
        self.payload = payload
        """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

        Returns:
            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

        Returns:
            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

        Returns:
            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

        Returns:
            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 used to open a modal

        Note:
            The `trigger_id` is only valid for 3 seconds after the modal was submitted.

            You can use [`open_modal`][machine.plugins.block_action.BlockAction.open_modal] to open a modal instead of
            using the `trigger_id` directly.

        Returns:
            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 [convenience 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
        [convenience classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        Args:
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/docs/message-attachments))
            ephemeral: `True/False` wether to send the message as an ephemeral message, only
                visible to the user that initiated the action
            replace_original: `True/False` whether the message that contains the block from which the action was
                triggered should be replaced by this message
            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.

        Returns:
            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,
        )

    async def send_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:
        """Send a DM to the user that triggered the block action

        Send a Direct Message to the user that triggered the block action 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 [convenience 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
        [convenience classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

        Args:
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

        Returns:
            Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
        """
        return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

    async def open_modal(
        self,
        view: dict | View,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Open a modal in response to the block action

        Open a modal in response to the block action, using the trigger_id that was returned when the block action was
        triggered.
        Any extra kwargs you provide, will be passed on directly to
        [views.open](https://api.slack.com/methods/views.open)

        Note:
            You have to call this method within 3 seconds of receiving the block action payload.

        Args:
            view: the view to open

        Returns:
            Dictionary deserialized from [views.open](https://api.slack.com/methods/views.open)
        """
        return await self._client.open_modal(self.trigger_id, view, **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 used to open a modal

Note

The trigger_id is only valid for 3 seconds after the modal was submitted.

You can use open_modal to open a modal instead of using the trigger_id directly.

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 convenience 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 src/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 [convenience 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
    [convenience classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    Args:
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/docs/message-attachments))
        ephemeral: `True/False` wether to send the message as an ephemeral message, only
            visible to the user that initiated the action
        replace_original: `True/False` whether the message that contains the block from which the action was
            triggered should be replaced by this message
        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.

    Returns:
        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,
    )

send_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

Send a DM to the user that triggered the block action

Send a Direct Message to the user that triggered the block action 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 convenience classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

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 response.

Source code in src/machine/plugins/block_action.py

async def send_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:
    """Send a DM to the user that triggered the block action

    Send a Direct Message to the user that triggered the block action 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 [convenience 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
    [convenience classes]: https://github.com/slackapi/python-slack-sdk/tree/main/slack/web/classes

    Args:
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

    Returns:
        Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
    """
    return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

open_modal(view: dict | View, **kwargs: Any) -> AsyncSlackResponse async

Open a modal in response to the block action

Open a modal in response to the block action, using the trigger_id that was returned when the block action was triggered. Any extra kwargs you provide, will be passed on directly to views.open

Note

You have to call this method within 3 seconds of receiving the block action payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to open	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from views.open

Source code in src/machine/plugins/block_action.py

async def open_modal(
    self,
    view: dict | View,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Open a modal in response to the block action

    Open a modal in response to the block action, using the trigger_id that was returned when the block action was
    triggered.
    Any extra kwargs you provide, will be passed on directly to
    [views.open](https://api.slack.com/methods/views.open)

    Note:
        You have to call this method within 3 seconds of receiving the block action payload.

    Args:
        view: the view to open

    Returns:
        Dictionary deserialized from [views.open](https://api.slack.com/methods/views.open)
    """
    return await self._client.open_modal(self.trigger_id, view, **kwargs)

machine.plugins.modals.ModalSubmission

A Slack modal submission that was received by the bot

This class represents a modal submission that was received by the bot and passed to a plugin.

Attributes:

Name	Type	Description
payload	ViewSubmissionPayload	The payload that was received by the bot when the modal was submitted

Source code in src/machine/plugins/modals.py

class ModalSubmission:
    """A Slack modal submission that was received by the bot

    This class represents a modal submission that was received by the bot and passed to a plugin.

    Attributes:
        payload: The payload that was received by the bot when the modal was submitted
    """

    payload: ViewSubmissionPayload

    def __init__(self, client: SlackClient, payload: ViewSubmissionPayload):
        self._client = client
        self.payload = payload

    @property
    def user(self) -> User:
        """The user that submitted the modal

        Returns:
            the user that submitted the modal
        """
        return self._client.users[self.payload.user.id]

    @property
    def view(self) -> View:
        """The view that was submitted including the state of all the elements in the view

        Returns:
            the view that was submitted
        """
        return self.payload.view

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

        The trigger id can be used to open another modal

        Note:
            The trigger id is only valid for 3 seconds after the modal was submitted.

            You can use [`open_modal()`][machine.plugins.modals.ModalSubmission.open_modal] to open a modal instead of
            using the `trigger_id` directly.

        Returns:
            the trigger id for the modal
        """
        return self.payload.trigger_id

    async def open_modal(
        self,
        view: dict | SlackSDKView,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Open another modal in response to the modal submission

        Open another modal in response to modal submission, using the trigger_id that was returned when the modal was
        submitted.
        Any extra kwargs you provide, will be passed on directly to
        [views.open](https://api.slack.com/methods/views.open)

        Note:
            You have to call this method within 3 seconds of receiving the modal submission payload.

        Args:
            view: the view to open

        Returns:
            Dictionary deserialized from [views.open](https://api.slack.com/methods/views.open)
        """
        return await self._client.open_modal(self.trigger_id, view, **kwargs)

    async def push_modal(
        self,
        view: dict | SlackSDKView,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Push a new modal view in response to the modal submission

        Push a new modal view on top of the view stack in response to modal submission, using the trigger_id that was
        returned when the modal was submitted.
        Any extra kwargs you provide, will be passed on directly to
        [views.push](https://api.slack.com/methods/views.push)

        Note:
            You have to call this method within 3 seconds of receiving the modal submission payload.

        Args:
            view: the view to push

        Returns:
            Dictionary deserialized from [views.push](https://api.slack.com/methods/views.push)
        """
        return await self._client.push_modal(self.trigger_id, view, **kwargs)

    async def update_modal(
        self,
        view: dict | SlackSDKView,
        **kwargs: Any,
    ) -> AsyncSlackResponse:
        """Update the modal view in response to the modal submission

        Update the modal view in response to modal submission, using the trigger_id that was returned when the modal was
        submitted.
        Any extra kwargs you provide, will be passed on directly to
        [views.update](https://api.slack.com/methods/views.update)

        Note:
            You have to call this method within 3 seconds of receiving the modal submission payload.

        Args:
            view: the view to update

        Returns:
            Dictionary deserialized from [views.update](https://api.slack.com/methods/views.update)
        """
        return await self._client.update_modal(view, self.payload.view.id, self.payload.view.external_id, **kwargs)

    async def send_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:
        """Send a DM to the user that submitted the modal

        Send a Direct Message to the user that submitted the modal 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 [convenience classes] that the
        underlying slack client provides.
        Any extra kwargs you provide, will be passed on directly to the
        [chat.postMessage](https://api.slack.com/methods/chat.postMessage) request.

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

        Args:
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

        Returns:
            Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
        """
        return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

user: User property

The user that submitted the modal

Returns:

Type	Description
User	the user that submitted the modal

view: View property

The view that was submitted including the state of all the elements in the view

Returns:

Type	Description
View	the view that was submitted

trigger_id: str property

The trigger id associated with the submitted modal

The trigger id can be used to open another modal

Note

The trigger id is only valid for 3 seconds after the modal was submitted.

You can use open_modal() to open a modal instead of using the trigger_id directly.

Returns:

Type	Description
str	the trigger id for the modal

open_modal(view: dict | SlackSDKView, **kwargs: Any) -> AsyncSlackResponse async

Open another modal in response to the modal submission

Open another modal in response to modal submission, using the trigger_id that was returned when the modal was submitted. Any extra kwargs you provide, will be passed on directly to views.open

Note

You have to call this method within 3 seconds of receiving the modal submission payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to open	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from views.open

Source code in src/machine/plugins/modals.py

async def open_modal(
    self,
    view: dict | SlackSDKView,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Open another modal in response to the modal submission

    Open another modal in response to modal submission, using the trigger_id that was returned when the modal was
    submitted.
    Any extra kwargs you provide, will be passed on directly to
    [views.open](https://api.slack.com/methods/views.open)

    Note:
        You have to call this method within 3 seconds of receiving the modal submission payload.

    Args:
        view: the view to open

    Returns:
        Dictionary deserialized from [views.open](https://api.slack.com/methods/views.open)
    """
    return await self._client.open_modal(self.trigger_id, view, **kwargs)

push_modal(view: dict | SlackSDKView, **kwargs: Any) -> AsyncSlackResponse async

Push a new modal view in response to the modal submission

Push a new modal view on top of the view stack in response to modal submission, using the trigger_id that was returned when the modal was submitted. Any extra kwargs you provide, will be passed on directly to views.push

Note

You have to call this method within 3 seconds of receiving the modal submission payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to push	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from views.push

Source code in src/machine/plugins/modals.py

async def push_modal(
    self,
    view: dict | SlackSDKView,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Push a new modal view in response to the modal submission

    Push a new modal view on top of the view stack in response to modal submission, using the trigger_id that was
    returned when the modal was submitted.
    Any extra kwargs you provide, will be passed on directly to
    [views.push](https://api.slack.com/methods/views.push)

    Note:
        You have to call this method within 3 seconds of receiving the modal submission payload.

    Args:
        view: the view to push

    Returns:
        Dictionary deserialized from [views.push](https://api.slack.com/methods/views.push)
    """
    return await self._client.push_modal(self.trigger_id, view, **kwargs)

update_modal(view: dict | SlackSDKView, **kwargs: Any) -> AsyncSlackResponse async

Update the modal view in response to the modal submission

Update the modal view in response to modal submission, using the trigger_id that was returned when the modal was submitted. Any extra kwargs you provide, will be passed on directly to views.update

Note

You have to call this method within 3 seconds of receiving the modal submission payload.

Parameters:

Name	Type	Description	Default
view	dict | View	the view to update	required

Returns:

Type	Description
AsyncSlackResponse	Dictionary deserialized from views.update

Source code in src/machine/plugins/modals.py

async def update_modal(
    self,
    view: dict | SlackSDKView,
    **kwargs: Any,
) -> AsyncSlackResponse:
    """Update the modal view in response to the modal submission

    Update the modal view in response to modal submission, using the trigger_id that was returned when the modal was
    submitted.
    Any extra kwargs you provide, will be passed on directly to
    [views.update](https://api.slack.com/methods/views.update)

    Note:
        You have to call this method within 3 seconds of receiving the modal submission payload.

    Args:
        view: the view to update

    Returns:
        Dictionary deserialized from [views.update](https://api.slack.com/methods/views.update)
    """
    return await self._client.update_modal(view, self.payload.view.id, self.payload.view.external_id, **kwargs)

send_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

Send a DM to the user that submitted the modal

Send a Direct Message to the user that submitted the modal 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 convenience classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

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 response.

Source code in src/machine/plugins/modals.py

async def send_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:
    """Send a DM to the user that submitted the modal

    Send a Direct Message to the user that submitted the modal 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 [convenience classes] that the
    underlying slack client provides.
    Any extra kwargs you provide, will be passed on directly to the
    [chat.postMessage](https://api.slack.com/methods/chat.postMessage) request.

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

    Args:
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))

    Returns:
        Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
    """
    return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

machine.plugins.modals.ModalClosure

A Slack modal closure that was received by the bot

This class represents the closure (cancellation) of a modal that was received by the bot and passed to a plugin.

Attributes:

Name	Type	Description
payload	ViewClosedPayload	The payload that was received by the bot when the modal was closed

Source code in src/machine/plugins/modals.py

class ModalClosure:
    """A Slack modal closure that was received by the bot

    This class represents the closure (cancellation) of a modal that was received by the bot and passed to a plugin.

    Attributes:
        payload: The payload that was received by the bot when the modal was closed
    """

    payload: ViewClosedPayload

    def __init__(self, client: SlackClient, payload: ViewClosedPayload):
        self._client = client
        self.payload = payload

    @property
    def user(self) -> User:
        """The user that closed the modal

        Returns:
            the user that closed the modal
        """
        return self._client.users[self.payload.user.id]

    @property
    def view(self) -> View:
        """The view that was closed including the state of all the elements in the view when it was closed

        Returns:
            the view that was closed
        """
        return self.payload.view

    async def send_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:
        """Send a DM to the user that closed the modal

        Send a Direct Message to the user that closed the modal 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 [convenience classes] that the
        underlying slack client provides.
        Any extra kwargs you provide, will be passed on directly to the
        [chat.postMessage](https://api.slack.com/methods/chat.postMessage) request.

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

        Args:
            text: message text
            attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
            blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
        Returns:
            Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
        """
        return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **kwargs)

user: User property

The user that closed the modal

Returns:

Type	Description
User	the user that closed the modal

view: View property

The view that was closed including the state of all the elements in the view when it was closed

Returns:

Type	Description
View	the view that was closed

send_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

Send a DM to the user that closed the modal

Send a Direct Message to the user that closed the modal 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 convenience classes that the underlying slack client provides. Any extra kwargs you provide, will be passed on directly to the chat.postMessage request.

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: Dictionary deserialized from chat.postMessage response.

Source code in src/machine/plugins/modals.py

async def send_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:
    """Send a DM to the user that closed the modal

    Send a Direct Message to the user that closed the modal 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 [convenience classes] that the
    underlying slack client provides.
    Any extra kwargs you provide, will be passed on directly to the
    [chat.postMessage](https://api.slack.com/methods/chat.postMessage) request.

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

    Args:
        text: message text
        attachments: optional attachments (see [attachments](https://api.slack.com/docs/message-attachments))
        blocks: optional blocks (see [blocks](https://api.slack.com/reference/block-kit/blocks))
    Returns:
        Dictionary deserialized from [chat.postMessage](https://api.slack.com/methods/chat.postMessage) response.
    """
    return await self._client.send_dm(self.user.id, text, attachments=attachments, blocks=blocks, **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.

Parameters:

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

Returns:

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

Source code in src/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

    Args:
        slack_event_type: type of event the method needs to process. Can be any event supported
            by the [Events API](https://api.slack.com/events?filter=Events)

    Returns:
        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 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 src/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 [`Message`][machine.plugins.message.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.

    Args:
        regex: regex pattern to listen for
        flags: regex flags to apply when matching
        handle_message_changed: if changed messages should trigger the decorated function

    Returns:
        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 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 src/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 [`Message`][machine.plugins.message.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.

    Args:
        regex: regex pattern to listen for
        flags: regex flags to apply when matching
        handle_message_changed: if changed messages should trigger the decorated function

    Returns:
        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 src/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

    Args:
        slash_command: the slash command to respond to

    Returns:
        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 src/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**.

    Args:
        action_id: the action_id to respond to, can be a string or regex pattern
        block_id: the block_id to respond to, can be a string or regex pattern

    Returns:
        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

modal(callback_id: Union[re.Pattern[str], str]) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Respond to modal submissions

This decorator will enable a Plugin method to be triggered when certain modals are submitted. The Plugin method will be called when a modal submission event is received for which the callback_id matches the provided value. The callback_id can be a string or a regex pattern.

Parameters:

Name	Type	Description	Default
callback_id	Union[Pattern[str], str]	the callback id to respond to, can be a string or regex pattern	required

Returns:

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

Source code in src/machine/plugins/decorators.py

def modal(callback_id: Union[re.Pattern[str], str]) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Respond to modal submissions

    This decorator will enable a Plugin method to be triggered when certain modals are submitted.
    The Plugin method will be called when a modal submission event is received for which the
    `callback_id` matches the provided value. The `callback_id` can be a string or a regex pattern.

    Args:
        callback_id: the callback id to respond to, can be a string or regex pattern

    Returns:
        wrapped method
    """

    def modal_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        is_generator = inspect.isasyncgenfunction(f)
        fn.metadata.plugin_actions.modal_submissions.append(
            ModalConfig(callback_id=callback_id, is_generator=is_generator)
        )
        return fn

    return modal_decorator

modal_closed(callback_id: Union[re.Pattern[str], str]) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]

Respond to modal closures

This decorator will enable a Plugin method to be triggered when certain modals are closed. The Plugin method will be called when a modal closure event is received for which the callback_id matches the provided value. The callback_id can be a string or a regex pattern.

Note

In order to receive modal close events, the modal must have the notify_on_close property set to True.

Parameters:

Name	Type	Description	Default
callback_id	Union[Pattern[str], str]	the callback id to respond to, can be a string or regex pattern	required

Returns:

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

Source code in src/machine/plugins/decorators.py

def modal_closed(callback_id: Union[re.Pattern[str], str]) -> Callable[[Callable[P, R]], DecoratedPluginFunc[P, R]]:
    """Respond to modal closures

    This decorator will enable a Plugin method to be triggered when certain modals are closed.
    The Plugin method will be called when a modal closure event is received for which the
    `callback_id` matches the provided value. The `callback_id` can be a string or a regex pattern.

    Note:
        In order to receive modal close events, the modal must have the `notify_on_close` property set to `True`.

    Args:
        callback_id: the callback id to respond to, can be a string or regex pattern

    Returns:
        wrapped method
    """

    def modal_closed_decorator(f: Callable[P, R]) -> DecoratedPluginFunc[P, R]:
        fn = cast(DecoratedPluginFunc, f)
        fn.metadata = getattr(f, "metadata", Metadata())
        is_generator = inspect.isasyncgenfunction(f)
        if is_generator:
            raise ValueError("Modal closed handlers cannot be async generators")
        fn.metadata.plugin_actions.modal_closures.append(ModalConfig(callback_id=callback_id))
        return fn

    return modal_closed_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 CronTrigger

Parameters:

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

Returns:

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

Source code in src/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 [`CronTrigger`][apscheduler.triggers.cron.CronTrigger]

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

    Returns:
        wrapped method
    """
    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

Returns:

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

Source code in src/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.

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

    Returns:
        wrapped method
    """

    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

Returns:

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

Source code in src/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.

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

    Returns:
        wrapped method
    """

    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

Returns:

Type	Description
Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]	wrapped method

Source code in src/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.

    Args:
        required_roles: list of roles required to use the plugin method

    Returns:
        wrapped 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

Returns:

Type	Description
Callable[[Callable[..., Awaitable[None]]], Callable[..., Awaitable[None]]]	wrapped method

Source code in src/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.

    Args:
        required_roles: list of roles required to use the plugin method

    Returns:
        wrapped 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 src/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:
        """Format the user as a mention"""
        return f"<@{self.id}>"

fmt_mention() -> str

Format the user as a mention

Source code in src/machine/models/user.py

def fmt_mention(self) -> str:
    """Format the user as a mention"""
    return f"<@{self.id}>"

machine.models.channel.Channel

Bases: BaseModel

Channel model that represents a channel object from the Slack API

Source code in src/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:
        """Return the name of the channel if it exists, otherwise return the id"""
        if self.name:
            return self.name
        else:
            return self.id

identifier: str property

Return the name of the channel if it exists, otherwise return the id

machine.models.interactive.BlockActionsPayload

Bases: TypedModel

Source code in src/machine/models/interactive.py

class BlockActionsPayload(TypedModel):
    type: Literal["block_actions"]
    user: User
    api_app_id: str
    token: str
    container: Container
    trigger_id: str
    team: Team
    enterprise: Optional[str]
    is_enterprise_install: bool
    channel: Optional[Channel] = None
    message: Optional[Message] = None
    view: Optional[View] = None
    state: Optional[State] = None
    response_url: Optional[str] = None
    actions: list[Action]

    @model_validator(mode="after")
    def validate_view_or_message(self) -> BlockActionsPayload:
        if self.view is None and self.message is None:
            raise ValueError("Either view or message must be present!")
        if self.message is not None:
            if self.channel is None:
                raise ValueError("channel must be present when message is present!")
            if self.state is None:
                raise ValueError("state must be present when message is present!")
            if self.response_url is None:
                raise ValueError("response_url must be present when message is present!")
        return self

machine.models.interactive.Action = Annotated[Union[RadioButtonsAction, ButtonAction, CheckboxAction, DatepickerAction, StaticSelectAction, ChannelSelectAction, ConversationSelectAction, UserSelectAction, ExternalSelectAction, MultiStaticSelectAction, MultiChannelSelectAction, MultiConversationSelectAction, MultiUserSelectAction, MultiExternalSelectAction, TimepickerAction, UrlAction, OverflowAction, PlainTextInputAction, RichTextInputAction], Field(discriminator='type')] module-attribute

machine.models.interactive.ViewSubmissionPayload

Bases: TypedModel

Source code in src/machine/models/interactive.py

class ViewSubmissionPayload(TypedModel):
    type: Literal["view_submission"]
    team: Team
    user: User
    view: View
    enterprise: Optional[str]
    api_app_id: str
    token: str
    trigger_id: str
    response_urls: list[ResponseUrlForView]
    is_enterprise_install: bool

machine.models.interactive.View

Bases: BaseModel

Source code in src/machine/models/interactive.py

class View(BaseModel):
    id: str
    team_id: str
    type: Literal["modal", "home"]
    blocks: list[Block]
    private_metadata: str
    callback_id: str
    state: State
    hash: str
    title: Text
    clear_on_close: bool
    notify_on_close: bool
    close: Optional[Text]
    submit: Optional[Text]
    previous_view_id: Optional[str]
    root_view_id: str
    app_id: str
    external_id: str
    app_installed_team_id: str
    bot_id: str

machine.models.interactive.State

Bases: BaseModel

Source code in src/machine/models/interactive.py

class State(BaseModel):
    values: dict[str, dict[str, Values]]

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.

Source code in src/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

        Args:
            key: the key under which to store the data
            value: the data to store
            expires: optional number of seconds after which the data is expired
            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

        Args:
            key: key for the data to retrieve
            shared: `True/False` wether to retrieve data from the shared (global) namespace.

        Returns:
            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)`

        Args:
            key: key to check
            shared: `True/False` wether to check in the shared (global) namespace

        Returns:
            `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

        Args:
            key: key to remove
            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

        Returns:
            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

        Returns:
            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 src/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

    Args:
        key: the key under which to store the data
        value: the data to store
        expires: optional number of seconds after which the data is expired
        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 src/machine/storage/__init__.py

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

    Args:
        key: key for the data to retrieve
        shared: `True/False` wether to retrieve data from the shared (global) namespace.

    Returns:
        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 src/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)`

    Args:
        key: key to check
        shared: `True/False` wether to check in the shared (global) namespace

    Returns:
        `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 src/machine/storage/__init__.py

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

    Args:
        key: key to remove
        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 src/machine/storage/__init__.py

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

    Returns:
        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 src/machine/storage/__init__.py

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

    Returns:
        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 src/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

        To be implemented by subclasses if initialization is required. This method is called once after instantiation.
        """
        pass

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

        Args:
            key: key for which to retrieve data

        Returns:
            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

        Args:
            key: the key under which to store the data
            value: data as (byte)string
            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

        Args:
            key: key to check

        Returns:
            `True/False` wether the key exists
        """
        ...

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

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

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

init() -> None async

Initialize the storage backend

To be implemented by subclasses if initialization is required. This method is called once after instantiation.

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

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

    To be implemented by subclasses if initialization is required. This method is called once after instantiation.
    """
    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 src/machine/storage/backends/base.py

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

    Args:
        key: key for which to retrieve data

    Returns:
        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 src/machine/storage/backends/base.py

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

    Args:
        key: the key under which to store the data
        value: data as (byte)string
        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

:param key: key for which to delete the data

Source code in src/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 src/machine/storage/backends/base.py

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

    Args:
        key: key to check

    Returns:
        `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 src/machine/storage/backends/base.py

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

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

close() -> None abstractmethod async

Close the storage backend

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

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