async_client.py

# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
#
#  *** DO NOT EDIT THIS FILE ***
#
#  1) Modify slack_sdk/web/client.py
#  2) Run `python setup.py codegen`
#
# !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!

"""A Python module for interacting with Slack's Web API."""
import os
from io import IOBase
from typing import Union, Sequence, Optional, Dict, Tuple

import slack_sdk.errors as e
from slack_sdk.models.views import View
from .async_base_client import AsyncBaseClient, AsyncSlackResponse
from .internal_utils import (
    _parse_web_class_objects,
    _update_call_participants,
    _warn_if_text_is_missing,
)


class AsyncWebClient(AsyncBaseClient):
    """A WebClient allows apps to communicate with the Slack Platform's Web API.

    https://api.slack.com/methods

    The Slack Web API is an interface for querying information from
    and enacting change in a Slack workspace.

    This client handles constructing and sending HTTP requests to Slack
    as well as parsing any responses received into a `SlackResponse`.

    Attributes:
        token (str): A string specifying an xoxp or xoxb token.
        base_url (str): A string representing the Slack API base URL.
            Default is 'https://www.slack.com/api/'
        timeout (int): The maximum number of seconds the client will wait
            to connect and receive a response from Slack.
            Default is 30 seconds.

    Methods:
        api_call: Constructs a request and executes the API call to Slack.

    Example of recommended usage:
    ```python
        import os
        from slack_sdk.web.async_client import AsyncWebClient

        client = AsyncWebClient(token=os.environ['SLACK_API_TOKEN'])
        response = client.chat_postMessage(
            channel='#random',
            text="Hello world!")
        assert response["ok"]
        assert response["message"]["text"] == "Hello world!"
    ```

    Example manually creating an API request:
    ```python
        import os
        from slack_sdk.web.async_client import AsyncWebClient

        client = AsyncWebClient(token=os.environ['SLACK_API_TOKEN'])
        response = client.api_call(
            api_method='chat.postMessage',
            json={'channel': '#random','text': "Hello world!"}
        )
        assert response["ok"]
        assert response["message"]["text"] == "Hello world!"
    ```

    Note:
        Any attributes or methods prefixed with _underscores are
        intended to be "private" internal use only. They may be changed or
        removed at anytime.
    """

    async def admin_analytics_getFile(
        self,
        *,
        type: str,
        date: Optional[str] = None,
        metadata_only: Optional[bool] = None,
        **kwargs
    ) -> AsyncSlackResponse:
        """Retrieve analytics data for a given date, presented as a compressed JSON file

        Args:
            date (str): Date to retrieve the analytics data for,
                expressed as YYYY-MM-DD in UTC.
            type (str): The type of analytics to retrieve.
                The options are currently limited to member.
        """
        kwargs.update({"type": type})
        if date is not None:
            kwargs.update({"date": date})
        if metadata_only is not None:
            kwargs.update({"metadata_only": metadata_only})
        return await self.api_call("admin.analytics.getFile", params=kwargs)

    async def admin_apps_approve(
        self, *, app_id: str = None, request_id: str = None, **kwargs
    ) -> AsyncSlackResponse:
        """Approve an app for installation on a workspace.

        Either app_id or request_id is required.
        These IDs can be obtained either directly via the app_requested event,
        or by the admin.apps.requests.list method.

        Args:
            app_id (str): The id of the app to approve. e.g. 'A12345'
            request_id (str): The id of the request to approve. e.g. 'Ar12345'
        Raises:
            SlackRequestError: If neither or both the `app_id` and `request_id` args are specified.
        """
        if app_id:
            kwargs.update({"app_id": app_id})
        elif request_id:
            kwargs.update({"request_id": request_id})
        else:
            raise e.SlackRequestError(
                "The app_id or request_id argument must be specified."
            )

        return await self.api_call("admin.apps.approve", json=kwargs)

    async def admin_apps_approved_list(self, **kwargs) -> AsyncSlackResponse:
        """List approved apps for an org or workspace."""
        return await self.api_call(
            "admin.apps.approved.list", http_verb="GET", params=kwargs
        )

    async def admin_apps_clearResolution(
        self, *, app_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Clear an app resolution

        Args:
            app_id (str): The id of the app whose resolution you want to clear/undo.
        """
        kwargs.update({"app_id": app_id})
        return await self.api_call(
            "admin.apps.clearResolution", http_verb="POST", params=kwargs
        )

    async def admin_apps_requests_list(self, **kwargs) -> AsyncSlackResponse:
        """List app requests for a team/workspace."""
        return await self.api_call(
            "admin.apps.requests.list", http_verb="GET", params=kwargs
        )

    async def admin_apps_restrict(self, **kwargs) -> AsyncSlackResponse:
        """Restrict an app for installation on a workspace."""
        return await self.api_call("admin.apps.restrict", json=kwargs)

    async def admin_apps_restricted_list(self, **kwargs) -> AsyncSlackResponse:
        """List restricted apps for an org or workspace."""
        return await self.api_call(
            "admin.apps.restricted.list", http_verb="GET", params=kwargs
        )

    async def admin_apps_uninstall(
        self,
        *,
        app_id: str,
        enterprise_id: Optional[str] = None,
        team_ids: Optional[Union[str, Sequence[str]]] = None,
        **kwargs
    ) -> AsyncSlackResponse:
        """Uninstall an app from one or many workspaces, or an entire enterprise organization."""
        kwargs.update({"app_id": app_id})
        if enterprise_id is not None:
            kwargs.update({"enterprise_id": enterprise_id})
        if team_ids is not None:
            if isinstance(team_ids, (list, Tuple)):
                kwargs.update({"team_ids": ",".join(team_ids)})
            else:
                kwargs.update({"team_ids": team_ids})
        return await self.api_call(
            "admin.apps.uninstall", http_verb="POST", params=kwargs
        )

    async def admin_auth_policy_getEntities(
        self,
        *,
        policy_name: str,
        cursor: Optional[str] = None,
        entity_type: Optional[str] = None,
        limit: Optional[int] = None,
        **kwargs
    ) -> AsyncSlackResponse:
        """Fetch all the entities assigned to a particular authentication policy by name."""
        kwargs.update({"policy_name": policy_name})
        if cursor is not None:
            kwargs.update({"cursor": cursor})
        if entity_type is not None:
            kwargs.update({"entity_type": entity_type})
        if limit is not None:
            kwargs.update({"limit": limit})
        return await self.api_call(
            "admin.auth.policy.getEntities", http_verb="POST", params=kwargs
        )

    async def admin_auth_policy_assignEntities(
        self,
        *,
        entity_ids: Union[str, Sequence[str]],
        policy_name: str,
        entity_type: str,
        **kwargs
    ) -> AsyncSlackResponse:
        """Assign entities to a particular authentication policy."""
        if isinstance(entity_ids, (list, Tuple)):
            kwargs.update({"entity_ids": ",".join(entity_ids)})
        else:
            kwargs.update({"entity_ids": entity_ids})
        kwargs.update({"policy_name": policy_name})
        kwargs.update({"entity_type": entity_type})
        return await self.api_call(
            "admin.auth.policy.assignEntities", http_verb="POST", params=kwargs
        )

    async def admin_auth_policy_removeEntities(
        self,
        *,
        entity_ids: Union[str, Sequence[str]],
        policy_name: str,
        entity_type: str,
        **kwargs
    ) -> AsyncSlackResponse:
        """Remove specified entities from a specified authentication policy."""
        if isinstance(entity_ids, (list, Tuple)):
            kwargs.update({"entity_ids": ",".join(entity_ids)})
        else:
            kwargs.update({"entity_ids": entity_ids})
        kwargs.update({"policy_name": policy_name})
        kwargs.update({"entity_type": entity_type})
        return await self.api_call(
            "admin.auth.policy.removeEntities", http_verb="POST", params=kwargs
        )

    async def admin_barriers_create(
        self,
        *,
        barriered_from_usergroup_ids: Union[str, Sequence[str]],
        primary_usergroup_id: str,
        restricted_subjects: Union[str, Sequence[str]],
        **kwargs
    ) -> AsyncSlackResponse:
        """Create an Information Barrier"""
        kwargs.update({"primary_usergroup_id": primary_usergroup_id})
        if isinstance(barriered_from_usergroup_ids, (list, Tuple)):
            kwargs.update(
                {"barriered_from_usergroup_ids": ",".join(barriered_from_usergroup_ids)}
            )
        else:
            kwargs.update(
                {"barriered_from_usergroup_ids": barriered_from_usergroup_ids}
            )
        if isinstance(restricted_subjects, (list, Tuple)):
            kwargs.update({"restricted_subjects": ",".join(restricted_subjects)})
        else:
            kwargs.update({"restricted_subjects": restricted_subjects})
        return await self.api_call(
            "admin.barriers.create", http_verb="POST", params=kwargs
        )

    async def admin_barriers_delete(
        self, *, barrier_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Delete an existing Information Barrier"""
        kwargs.update({"barrier_id": barrier_id})
        return await self.api_call(
            "admin.barriers.delete", http_verb="POST", params=kwargs
        )

    async def admin_barriers_update(
        self,
        *,
        barrier_id: str,
        barriered_from_usergroup_ids: Union[str, Sequence[str]],
        primary_usergroup_id: str,
        restricted_subjects: Union[str, Sequence[str]],
        **kwargs
    ) -> AsyncSlackResponse:
        """Update an existing Information Barrier"""
        kwargs.update(
            {"barrier_id": barrier_id, "primary_usergroup_id": primary_usergroup_id}
        )
        if isinstance(barriered_from_usergroup_ids, (list, Tuple)):
            kwargs.update(
                {"barriered_from_usergroup_ids": ",".join(barriered_from_usergroup_ids)}
            )
        else:
            kwargs.update(
                {"barriered_from_usergroup_ids": barriered_from_usergroup_ids}
            )
        if isinstance(restricted_subjects, (list, Tuple)):
            kwargs.update({"restricted_subjects": ",".join(restricted_subjects)})
        else:
            kwargs.update({"restricted_subjects": restricted_subjects})
        return await self.api_call(
            "admin.barriers.update", http_verb="POST", params=kwargs
        )

    async def admin_barriers_list(self, **kwargs) -> AsyncSlackResponse:
        """Get all Information Barriers for your organization"""
        return await self.api_call(
            "admin.barriers.list", http_verb="GET", params=kwargs
        )

    async def admin_conversations_create(
        self, *, is_private: bool, name: str, **kwargs
    ) -> AsyncSlackResponse:
        """Create a public or private channel-based conversation.

        Args:
            is_private (bool): When true, creates a private channel instead of a public channel
            name (str): Name of the public or private channel to create.
            org_wide (bool): When true, the channel will be available org-wide.
                Note: if the channel is not org_wide=true, you must specify a team_id for this channel
            team_id (str): The workspace to create the channel in.
                Note: this argument is required unless you set org_wide=true.

        """
        kwargs.update({"is_private": is_private, "name": name})
        return await self.api_call("admin.conversations.create", json=kwargs)

    async def admin_conversations_delete(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Delete a public or private channel.

        Args:
            channel_id (str): The channel to delete.

        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call("admin.conversations.delete", json=kwargs)

    async def admin_conversations_invite(
        self, *, channel_id: str, user_ids: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Invite a user to a public or private channel.

        Args:
            channel_id (str): The channel that the users will be invited to.
            user_ids (str or list): The users to invite.
        """
        kwargs.update({"channel_id": channel_id})
        if isinstance(user_ids, (list, Tuple)):
            kwargs.update({"user_ids": ",".join(user_ids)})
        else:
            kwargs.update({"user_ids": user_ids})
        # NOTE: the endpoint is unable to handle Content-Type: application/json as of Sep 3, 2020.
        return await self.api_call("admin.conversations.invite", params=kwargs)

    async def admin_conversations_archive(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Archive a public or private channel.

        Args:
            channel_id (str): The channel to archive.
        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call("admin.conversations.archive", json=kwargs)

    async def admin_conversations_unarchive(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Unarchive a public or private channel.

        Args:
            channel_id (str): The channel to unarchive.
        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call("admin.conversations.unarchive", json=kwargs)

    async def admin_conversations_rename(
        self, *, channel_id: str, name: str, **kwargs
    ) -> AsyncSlackResponse:
        """Rename a public or private channel.

        Args:
            channel_id (str): The channel to rename.
            name (str): The name to rename the channel to.
        """
        kwargs.update({"channel_id": channel_id, "name": name})
        return await self.api_call("admin.conversations.rename", json=kwargs)

    async def admin_conversations_search(self, **kwargs) -> AsyncSlackResponse:
        """Search for public or private channels in an Enterprise organization."""
        return await self.api_call("admin.conversations.search", params=kwargs)

    async def admin_conversations_convertToPrivate(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Convert a public channel to a private channel.

        Args:
            channel_id (str): The channel to convert to private.
        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call("admin.conversations.convertToPrivate", json=kwargs)

    async def admin_conversations_setConversationPrefs(
        self, *, channel_id: str, prefs: Union[str, dict], **kwargs
    ) -> AsyncSlackResponse:
        """Set the posting permissions for a public or private channel.

        Args:
            channel_id (str): The channel to set the prefs for
            prefs (str or dict): The prefs for this channel in a stringified JSON format.
        """
        kwargs.update({"channel_id": channel_id, "prefs": prefs})
        return await self.api_call(
            "admin.conversations.setConversationPrefs", json=kwargs
        )

    async def admin_conversations_getConversationPrefs(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Get conversation preferences for a public or private channel.

        Args:
            channel_id (str): The channel to get the preferences for.
        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call(
            "admin.conversations.getConversationPrefs", json=kwargs
        )

    async def admin_conversations_disconnectShared(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Disconnect a connected channel from one or more workspaces.

        Args:
            channel_id (str): The channel to be disconnected from some workspaces.
        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call("admin.conversations.disconnectShared", json=kwargs)

    async def admin_conversations_ekm_listOriginalConnectedChannelInfo(
        self, **kwargs
    ) -> AsyncSlackResponse:
        """List all disconnected channels—i.e.,
        channels that were once connected to other workspaces and then disconnected—and
        the corresponding original channel IDs for key revocation with EKM.
        """
        return await self.api_call(
            "admin.conversations.ekm.listOriginalConnectedChannelInfo", params=kwargs
        )

    async def admin_conversations_restrictAccess_addGroup(
        self, *, channel_id: str, group_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Add an allowlist of IDP groups for accessing a channel.

        Args:
            channel_id (str): The channel to link this group to. e.g. 'C1234567890'
            group_id (str): The IDP Group ID to be an allowlist for the private channel. 'S0604QSJC'
            team_id (str): The workspace where the channel exists.
                This argument is required for channels only tied to one workspace,
                and optional for channels that are shared across an organization.
                e.g 'T1234'
        """
        kwargs.update({"channel_id": channel_id, "group_id": group_id})
        return await self.api_call(
            "admin.conversations.restrictAccess.addGroup",
            http_verb="GET",
            params=kwargs,
        )

    async def admin_conversations_restrictAccess_listGroups(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """List all IDP Groups linked to a channel.

        Args:
            channel_id (str): The channel to link this group to. e.g. 'C1234567890'
            team_id (str): The workspace where the channel exists.
                This argument is required for channels only tied to one workspace,
                and optional for channels that are shared across an organization.
                e.g 'T1234'
        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call(
            "admin.conversations.restrictAccess.listGroups",
            http_verb="GET",
            params=kwargs,
        )

    async def admin_conversations_restrictAccess_removeGroup(
        self, *, channel_id: str, group_id: str, team_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Remove a linked IDP group linked from a private channel.

        Args:
            channel_id (str): The channel to link this group to. e.g. 'C1234567890'
            group_id (str): The IDP Group ID to be an allowlist for the private channel. 'S0604QSJC'
            team_id (str): The workspace where the channel exists.
                This argument is required for channels only tied to one workspace,
                and optional for channels that are shared across an organization.
                e.g 'T1234'
        """
        kwargs.update(
            {"channel_id": channel_id, "group_id": group_id, "team_id": team_id}
        )
        return await self.api_call(
            "admin.conversations.restrictAccess.removeGroup",
            http_verb="GET",
            params=kwargs,
        )

    async def admin_conversations_setTeams(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Set the workspaces in an Enterprise grid org that connect to a channel.

        Args:
            channel_id (str): The encoded channel_id to add or remove to workspaces.

        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call("admin.conversations.setTeams", json=kwargs)

    async def admin_conversations_getTeams(
        self, *, channel_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Set the workspaces in an Enterprise grid org that connect to a channel.

        Args:
            channel_id (str): The channel to determine connected workspaces within the organization for.

        """
        kwargs.update({"channel_id": channel_id})
        return await self.api_call("admin.conversations.getTeams", params=kwargs)

    async def admin_emoji_add(self, **kwargs) -> AsyncSlackResponse:
        """Add an emoji."""
        return await self.api_call("admin.emoji.add", http_verb="GET", params=kwargs)

    async def admin_emoji_addAlias(self, **kwargs) -> AsyncSlackResponse:
        """Add an emoji alias."""
        return await self.api_call(
            "admin.emoji.addAlias", http_verb="GET", params=kwargs
        )

    async def admin_emoji_list(self, **kwargs) -> AsyncSlackResponse:
        """List emoji for an Enterprise Grid organization."""
        return await self.api_call("admin.emoji.list", http_verb="GET", params=kwargs)

    async def admin_emoji_remove(self, **kwargs) -> AsyncSlackResponse:
        """Remove an emoji across an Enterprise Grid organization."""
        return await self.api_call("admin.emoji.remove", http_verb="GET", params=kwargs)

    async def admin_emoji_rename(self, **kwargs) -> AsyncSlackResponse:
        """Rename an emoji."""
        return await self.api_call("admin.emoji.rename", http_verb="GET", params=kwargs)

    async def admin_users_session_reset(
        self, *, user_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Wipes all valid sessions on all devices for a given user.

        Args:
            user_id (str): The ID of the user to wipe sessions for. e.g. 'W12345678'
        """
        kwargs.update({"user_id": user_id})
        return await self.api_call("admin.users.session.reset", json=kwargs)

    async def admin_users_session_invalidate(
        self, *, session_id: str, team_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Invalidate a single session for a user by session_id.
        Args:
            session_id (str): The ID of a session
            team_id (str): ID of the team that the session belongs to
        """
        kwargs.update({"session_id": session_id, "team_id": team_id})
        return await self.api_call("admin.users.session.invalidate", params=kwargs)

    async def admin_users_session_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists all active user sessions for an organization"""
        return await self.api_call("admin.users.session.list", params=kwargs)

    async def admin_teams_settings_setDefaultChannels(
        self, *, team_id: str, channel_ids: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Set the default channels of a workspace.

        Args:
            team_id (str): ID of the team.
            channel_ids (str or list): A list of channel_ids.
                At least one channel is required. e.g. ['C1A2B3C4D', 'C26Z25Y24']
        """
        kwargs.update({"team_id": team_id})
        if isinstance(channel_ids, (list, Tuple)):
            kwargs.update({"channel_ids": ",".join(channel_ids)})
        else:
            kwargs.update({"channel_ids": channel_ids})
        return await self.api_call(
            "admin.teams.settings.setDefaultChannels", http_verb="GET", params=kwargs
        )

    async def admin_users_session_getSettings(
        self, *, user_ids: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Get user-specific session settings—the session duration
        and what happens when the client closes—given a list of users.

        Args:
            user_ids (str or list): The IDs of users you'd like to fetch session settings for.
                Note: if a user does not have any active sessions, they will not be returned in the response.
        """
        if isinstance(user_ids, (list, Tuple)):
            kwargs.update({"user_ids": ",".join(user_ids)})
        else:
            kwargs.update({"user_ids": user_ids})
        return await self.api_call("admin.users.session.getSettings", params=kwargs)

    async def admin_users_session_setSettings(
        self, *, user_ids: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Configure the user-level session settings—the session duration
        and what happens when the client closes—for one or more users.

        Args:
            user_ids (str or list): The list of user IDs to apply the session settings for.
        """
        if isinstance(user_ids, (list, Tuple)):
            kwargs.update({"user_ids": ",".join(user_ids)})
        else:
            kwargs.update({"user_ids": user_ids})
        return await self.api_call("admin.users.session.setSettings", params=kwargs)

    async def admin_users_session_clearSettings(
        self, *, user_ids: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Clear user-specific session settings—the session duration
        and what happens when the client closes—for a list of users.

        Args:
            user_ids (str or list): The list of user IDs to apply the session settings for.
        """
        if isinstance(user_ids, (list, Tuple)):
            kwargs.update({"user_ids": ",".join(user_ids)})
        else:
            kwargs.update({"user_ids": user_ids})
        return await self.api_call("admin.users.session.clearSettings", params=kwargs)

    async def admin_inviteRequests_approve(
        self, *, invite_request_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Approve a workspace invite request.

        team_id is required if your Enterprise Grid org contains more than one workspace.

        Args:
            invite_request_id (str): ID of the request to invite. e.g. 'Ir1234'
        """
        kwargs.update({"invite_request_id": invite_request_id})
        return await self.api_call("admin.inviteRequests.approve", json=kwargs)

    async def admin_inviteRequests_approved_list(self, **kwargs) -> AsyncSlackResponse:
        """List all approved workspace invite requests."""
        return await self.api_call("admin.inviteRequests.approved.list", params=kwargs)

    async def admin_inviteRequests_denied_list(self, **kwargs) -> AsyncSlackResponse:
        """List all denied workspace invite requests."""
        return await self.api_call("admin.inviteRequests.denied.list", params=kwargs)

    async def admin_inviteRequests_deny(
        self, *, invite_request_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Deny a workspace invite request.

        Args:
            invite_request_id (str): ID of the request to invite. e.g. 'Ir1234'
        """
        kwargs.update({"invite_request_id": invite_request_id})
        return await self.api_call("admin.inviteRequests.deny", json=kwargs)

    async def admin_inviteRequests_list(self, **kwargs) -> AsyncSlackResponse:
        """List all pending workspace invite requests."""
        return await self.api_call("admin.inviteRequests.list", params=kwargs)

    async def admin_teams_admins_list(
        self, *, team_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """List all of the admins on a given workspace.

        Args:
            team_id (str): ID of the team.
        """
        kwargs.update({"team_id": team_id})
        return await self.api_call(
            "admin.teams.admins.list", http_verb="GET", params=kwargs
        )

    async def admin_teams_create(
        self, *, team_domain: str, team_name: str, **kwargs
    ) -> AsyncSlackResponse:
        """Create an Enterprise team.

        Args:
            team_domain (str): Team domain. e.g. 'slacksoftballteam'
            team_name (str): Team name. e.g. 'Slack Softball Team'
        """
        kwargs.update({"team_domain": team_domain, "team_name": team_name})
        return await self.api_call("admin.teams.create", json=kwargs)

    async def admin_teams_list(self, **kwargs) -> AsyncSlackResponse:
        """List all teams on an Enterprise organization."""
        return await self.api_call("admin.teams.list", params=kwargs)

    async def admin_teams_owners_list(
        self, *, team_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """List all of the admins on a given workspace.

        Args:
            team_id (str): ID of the team.
        """
        kwargs.update({"team_id": team_id})
        return await self.api_call(
            "admin.teams.owners.list", http_verb="GET", params=kwargs
        )

    async def admin_teams_settings_info(
        self, team_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Fetch information about settings in a workspace

        Args:
            team_id (str): ID of the team.
        """
        kwargs.update({"team_id": team_id})
        return await self.api_call("admin.teams.settings.info", json=kwargs)

    async def admin_teams_settings_setDescription(
        self, *, team_id: str, description: str, **kwargs
    ) -> AsyncSlackResponse:
        """Set the description of a given workspace.

        Args:
            team_id (str): ID of the team.
            description (str): Description of the team.
        """
        kwargs.update({"team_id": team_id, "description": description})
        return await self.api_call("admin.teams.settings.setDescription", json=kwargs)

    async def admin_teams_settings_setDiscoverability(
        self, *, team_id: str, discoverability: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the icon of a workspace.

        Args:
            team_id (str): ID of the team.
            discoverability (str): This workspace's discovery setting.
                It must be set to one of open, invite_only, closed, or unlisted.
        """
        kwargs.update({"team_id": team_id, "discoverability": discoverability})
        return await self.api_call(
            "admin.teams.settings.setDiscoverability", json=kwargs
        )

    async def admin_teams_settings_setIcon(
        self, *, team_id: str, image_url: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the icon of a workspace.

        Args:
            team_id (str): ID of the team.
            image_url (str): Url of the icon.
        """
        kwargs.update({"team_id": team_id, "image_url": image_url})
        return await self.api_call(
            "admin.teams.settings.setIcon", http_verb="GET", params=kwargs
        )

    async def admin_teams_settings_setName(
        self, *, team_id: str, name: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the icon of a workspace.

        Args:
            team_id (str): ID of the team.
            name (str): Name of the team.
        """
        kwargs.update({"team_id": team_id, "name": name})
        return await self.api_call("admin.teams.settings.setName", json=kwargs)

    async def admin_usergroups_addChannels(
        self,
        *,
        team_id: str,
        usergroup_id: str,
        channel_ids: Union[str, Sequence[str]],
        **kwargs
    ) -> AsyncSlackResponse:
        """Add one or more default channels to an IDP group.

        Args:
            team_id (str): The workspace to add default channels in. e.g. 'T1234'
            usergroup_id (str): ID of the IDP group to add default channels for. e.g. 'S1234'
            channel_ids (str or list): Comma separated string of channel IDs. e.g. 'C123,C234' or ['C123', 'C234']
        """
        kwargs.update({"team_id": team_id, "usergroup_id": usergroup_id})
        if isinstance(channel_ids, (list, Tuple)):
            kwargs.update({"channel_ids": ",".join(channel_ids)})
        else:
            kwargs.update({"channel_ids": channel_ids})
        return await self.api_call("admin.usergroups.addChannels", params=kwargs)

    async def admin_usergroups_addTeams(
        self, *, usergroup_id: str, team_ids: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Associate one or more default workspaces with an organization-wide IDP group.

        Args:
            usergroup_id (str): ID of the IDP group. e.g. 'S1234'
            team_ids (str or list): A comma separated list of encoded team (workspace) IDs.
                Each workspace MUST belong to the organization associated with the token.
                e.g. 'T12345678,T98765432' or ['T12345678', 'T98765432']
        """
        kwargs.update({"usergroup_id": usergroup_id})
        if isinstance(team_ids, (list, Tuple)):
            kwargs.update({"team_ids": ",".join(team_ids)})
        else:
            kwargs.update({"team_ids": team_ids})
        return await self.api_call("admin.usergroups.addTeams", params=kwargs)

    async def admin_usergroups_listChannels(
        self, *, usergroup_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Add one or more default channels to an IDP group.

        Args:
            usergroup_id (str): ID of the IDP group to list default channels for. e.g. 'S1234'
        """
        kwargs.update({"usergroup_id": usergroup_id})
        return await self.api_call("admin.usergroups.listChannels", json=kwargs)

    async def admin_usergroups_removeChannels(
        self, *, usergroup_id: str, channel_ids: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Add one or more default channels to an IDP group.

        Args:
            usergroup_id (str): ID of the IDP group. e.g. 'S1234'
            channel_ids (str or list): Comma separated string of channel IDs. e.g. 'C123,C234' or ['C123', 'C234']
        """
        kwargs.update({"usergroup_id": usergroup_id})
        if isinstance(channel_ids, (list, Tuple)):
            kwargs.update({"channel_ids": ",".join(channel_ids)})
        else:
            kwargs.update({"channel_ids": channel_ids})
        return await self.api_call("admin.usergroups.removeChannels", params=kwargs)

    async def admin_users_assign(
        self, *, team_id: str, user_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Add an Enterprise user to a workspace.

        Args:
            team_id (str): ID of the team. e.g. 'T1234'
            user_id (str): ID of the user to add to the workspace.
        """
        kwargs.update({"team_id": team_id, "user_id": user_id})
        return await self.api_call("admin.users.assign", json=kwargs)

    async def admin_users_invite(
        self,
        *,
        team_id: str,
        email: str,
        channel_ids: Union[str, Sequence[str]],
        **kwargs
    ) -> AsyncSlackResponse:
        """Invite a user to a workspace.

        Args:
            team_id (str): ID of the team. e.g. 'T1234'
            email (str): The email address of the person to invite. e.g. 'joe@email.com'
            channel_ids (str or list): A list of channel_ids for this user to join.
                At least one channel is required. e.g. ['C1A2B3C4D', 'C26Z25Y24']
        """
        kwargs.update({"team_id": team_id, "email": email})
        if isinstance(channel_ids, (list, Tuple)):
            kwargs.update({"channel_ids": ",".join(channel_ids)})
        else:
            kwargs.update({"channel_ids": channel_ids})
        return await self.api_call("admin.users.invite", params=kwargs)

    async def admin_users_list(self, *, team_id: str, **kwargs) -> AsyncSlackResponse:
        """List users on a workspace

        Args:
            team_id (str): ID of the team. e.g. 'T1234'
        """
        kwargs.update({"team_id": team_id})
        return await self.api_call("admin.users.list", params=kwargs)

    async def admin_users_remove(
        self, *, team_id: str, user_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Remove a user from a workspace.

        Args:
            team_id (str): ID of the team. e.g. 'T1234'
            user_id (str): The ID of the user to remove. e.g. 'W12345678'
        """
        kwargs.update({"team_id": team_id, "user_id": user_id})
        return await self.api_call("admin.users.remove", json=kwargs)

    async def admin_users_setAdmin(
        self, *, team_id: str, user_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Set an existing guest, regular user, or owner to be an admin user.

        Args:
            team_id (str): ID of the team. e.g. 'T1234'
            user_id (str): The ID of the user to remove. e.g. 'W12345678'
        """
        kwargs.update({"team_id": team_id, "user_id": user_id})
        return await self.api_call("admin.users.setAdmin", json=kwargs)

    async def admin_users_setExpiration(
        self, *, expiration_ts: int, team_id: str, user_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Set an expiration for a guest user.

        Args:
            expiration_ts (int): Timestamp when guest account should be disabled. e.g. '1234567890'
            team_id (str): ID of the team. e.g. 'T1234'
            user_id (str): The ID of the user to set an expiration for. e.g. 'W12345678'
        """
        kwargs.update(
            {"expiration_ts": expiration_ts, "team_id": team_id, "user_id": user_id}
        )
        return await self.api_call("admin.users.setExpiration", json=kwargs)

    async def admin_users_setOwner(
        self, *, team_id: str, user_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Set an existing guest, regular user, or admin user to be a workspace owner.

        Args:
            team_id (str): ID of the team. e.g. 'T1234'
            user_id (str): The ID of the user to remove. e.g. 'W12345678'
        """
        kwargs.update({"team_id": team_id, "user_id": user_id})
        return await self.api_call("admin.users.setOwner", json=kwargs)

    async def admin_users_setRegular(
        self, *, team_id: str, user_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Set an existing guest user, admin user, or owner to be a regular user.

        Args:
            team_id (str): ID of the team. e.g. 'T1234'
            user_id (str): The ID of the user to remove. e.g. 'W12345678'
        """
        kwargs.update({"team_id": team_id, "user_id": user_id})
        return await self.api_call("admin.users.setRegular", json=kwargs)

    async def api_test(self, **kwargs) -> AsyncSlackResponse:
        """Checks API calling code."""
        return await self.api_call("api.test", json=kwargs)

    async def apps_connections_open(
        self, *, app_token: str, **kwargs
    ) -> AsyncSlackResponse:
        """Get a new WSS URL for Socket Mode"""
        kwargs.update({"token": app_token})
        return await self.api_call(
            "apps.connections.open", http_verb="POST", params=kwargs
        )

    async def apps_event_authorizations_list(
        self, event_context: str, **kwargs
    ) -> AsyncSlackResponse:
        """Get a list of authorizations for the given event context.
        Each authorization represents an app installation that the event is visible to.
        Args:
            event_context (str): You'll receive an event_context identifying an event in each event payload sent to your app.
        """
        kwargs.update({"event_context": event_context})
        return await self.api_call("apps.event.authorizations.list", params=kwargs)

    async def apps_uninstall(
        self, client_id: str, client_secret: str, **kwargs
    ) -> AsyncSlackResponse:
        """Uninstalls your app from a workspace.
        Args:
            client_id (str): Issued when you created your application. e.g. '56579136444.26251006572'
            client_secret (str): Issued when you created your application. e.g. 'f25b5ceaf8a3c2a2c4f52bb4f0b0499e'
        """
        kwargs.update({"client_id": client_id, "client_secret": client_secret})
        return await self.api_call("apps.uninstall", params=kwargs)

    async def auth_revoke(self, **kwargs) -> AsyncSlackResponse:
        """Revokes a token."""
        return await self.api_call("auth.revoke", http_verb="GET", params=kwargs)

    async def auth_test(self, **kwargs) -> AsyncSlackResponse:
        """Checks authentication & identity."""
        return await self.api_call("auth.test", json=kwargs)

    async def bots_info(self, **kwargs) -> AsyncSlackResponse:
        """Gets information about a bot user."""
        return await self.api_call("bots.info", http_verb="GET", params=kwargs)

    async def calls_add(
        self, *, external_unique_id: str, join_url: str, **kwargs
    ) -> AsyncSlackResponse:
        """Registers a new Call.

        Args:
            external_unique_id (str): An ID supplied by the 3rd-party Call provider.
                It must be unique across all Calls from that service.
                e.g. '025169F6-E37A-4E62-BB54-7F93A0FC4C1F'
            join_url (str): The URL required for a client to join the Call.
                e.g. 'https://example.com/calls/1234567890'
        """
        kwargs.update({"external_unique_id": external_unique_id, "join_url": join_url})
        _update_call_participants(  # skipcq: PTC-W0039
            kwargs, kwargs.get("users", None)  # skipcq: PTC-W0039
        )  # skipcq: PTC-W0039
        return await self.api_call("calls.add", http_verb="POST", params=kwargs)

    async def calls_end(
        self, *, id: str, **kwargs
    ) -> AsyncSlackResponse:  # skipcq: PYL-W0622
        """Ends a Call.

        Args:
            id (str): id returned when registering the call using the calls.add method.
        """
        kwargs.update({"id": id})
        return await self.api_call("calls.end", http_verb="POST", params=kwargs)

    async def calls_info(
        self, *, id: str, **kwargs
    ) -> AsyncSlackResponse:  # skipcq: PYL-W0622
        """Returns information about a Call.

        Args:
            id (str): id returned when registering the call using the calls.add method.
        """
        kwargs.update({"id": id})
        return await self.api_call("calls.info", http_verb="POST", params=kwargs)

    async def calls_participants_add(
        self,
        *,
        id: str,  # skipcq: PYL-W0622
        users: Union[str, Sequence[Dict[str, str]]],
        **kwargs
    ) -> AsyncSlackResponse:
        """Registers new participants added to a Call.

        Args:
            id (str): id returned when registering the call using the calls.add method.
            users: (list): The list of users to add as participants in the Call.
        """
        kwargs.update({"id": id})
        _update_call_participants(kwargs, users)
        return await self.api_call(
            "calls.participants.add", http_verb="POST", params=kwargs
        )

    async def calls_participants_remove(
        self,
        *,
        id: str,  # skipcq: PYL-W0622
        users: Union[str, Sequence[Dict[str, str]]],
        **kwargs
    ) -> AsyncSlackResponse:
        """Registers participants removed from a Call.

        Args:
            id (str): id returned when registering the call using the calls.add method.
            users: (list): The list of users to remove as participants in the Call.
        """
        kwargs.update({"id": id})
        _update_call_participants(kwargs, users)
        return await self.api_call(
            "calls.participants.remove", http_verb="POST", params=kwargs
        )

    async def calls_update(
        self, *, id: str, **kwargs
    ) -> AsyncSlackResponse:  # skipcq: PYL-W0622
        """Updates information about a Call.

        Args:
            id (str): id returned by the calls.add method.
        """
        kwargs.update({"id": id})
        return await self.api_call("calls.update", http_verb="POST", params=kwargs)

    async def channels_archive(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Archives a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("channels.archive", json=kwargs)

    async def channels_create(self, *, name: str, **kwargs) -> AsyncSlackResponse:
        """Creates a channel.

        Args:
            name (str): The name of the channel. e.g. 'mychannel'
        """
        kwargs.update({"name": name})
        return await self.api_call("channels.create", json=kwargs)

    async def channels_history(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Fetches history of messages and events from a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("channels.history", http_verb="GET", params=kwargs)

    async def channels_info(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Gets information about a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("channels.info", http_verb="GET", params=kwargs)

    async def channels_invite(
        self, *, channel: str, user: str, **kwargs
    ) -> AsyncSlackResponse:
        """Invites a user to a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            user (str): The user id. e.g. 'U1234567890'
        """
        kwargs.update({"channel": channel, "user": user})
        return await self.api_call("channels.invite", json=kwargs)

    async def channels_join(self, *, name: str, **kwargs) -> AsyncSlackResponse:
        """Joins a channel, creating it if needed.

        Args:
            name (str): The channel name. e.g. '#general'
        """
        kwargs.update({"name": name})
        return await self.api_call("channels.join", json=kwargs)

    async def channels_kick(
        self, *, channel: str, user: str, **kwargs
    ) -> AsyncSlackResponse:
        """Removes a user from a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            user (str): The user id. e.g. 'U1234567890'
        """
        kwargs.update({"channel": channel, "user": user})
        return await self.api_call("channels.kick", json=kwargs)

    async def channels_leave(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Leaves a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("channels.leave", json=kwargs)

    async def channels_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists all channels in a Slack team."""
        return await self.api_call("channels.list", http_verb="GET", params=kwargs)

    async def channels_mark(
        self, *, channel: str, ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the read cursor in a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            ts (str): Timestamp of the most recently seen message. e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "ts": ts})
        return await self.api_call("channels.mark", json=kwargs)

    async def channels_rename(
        self, *, channel: str, name: str, **kwargs
    ) -> AsyncSlackResponse:
        """Renames a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            name (str): The new channel name. e.g. 'newchannel'
        """
        kwargs.update({"channel": channel, "name": name})
        return await self.api_call("channels.rename", json=kwargs)

    async def channels_replies(
        self, *, channel: str, thread_ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Retrieve a thread of messages posted to a channel

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            thread_ts (str): The timestamp of an existing message with 0 or more replies.
                e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "thread_ts": thread_ts})
        return await self.api_call("channels.replies", http_verb="GET", params=kwargs)

    async def channels_setPurpose(
        self, *, channel: str, purpose: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the purpose for a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            purpose (str): The new purpose for the channel. e.g. 'My Purpose'
        """
        kwargs.update({"channel": channel, "purpose": purpose})
        return await self.api_call("channels.setPurpose", json=kwargs)

    async def channels_setTopic(
        self, *, channel: str, topic: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the topic for a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            topic (str): The new topic for the channel. e.g. 'My Topic'
        """
        kwargs.update({"channel": channel, "topic": topic})
        return await self.api_call("channels.setTopic", json=kwargs)

    async def channels_unarchive(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Unarchives a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("channels.unarchive", json=kwargs)

    async def chat_delete(
        self, *, channel: str, ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Deletes a message.

        Args:
            channel (str): Channel containing the message to be deleted. e.g. 'C1234567890'
            ts (str): Timestamp of the message to be deleted. e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "ts": ts})
        return await self.api_call("chat.delete", json=kwargs)

    async def chat_deleteScheduledMessage(
        self, *, channel: str, scheduled_message_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Deletes a scheduled message.

        Args:
            channel (str): The channel the scheduled_message is posting to. e.g. 'C1234567890'
            scheduled_message_id (str): scheduled_message_id returned from call to chat.scheduleMessage e.g. 'Q1234ABCD'
        """
        kwargs.update(
            {"channel": channel, "scheduled_message_id": scheduled_message_id}
        )
        return await self.api_call("chat.deleteScheduledMessage", json=kwargs)

    async def chat_getPermalink(
        self, *, channel: str, message_ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Retrieve a permalink URL for a specific extant message

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            message_ts (str): The timestamp. e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "message_ts": message_ts})
        return await self.api_call("chat.getPermalink", http_verb="GET", params=kwargs)

    async def chat_meMessage(
        self, *, channel: str, text: str, **kwargs
    ) -> AsyncSlackResponse:
        """Share a me message into a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            text (str): The message you'd like to share. e.g. 'Hello world'
        """
        kwargs.update({"channel": channel, "text": text})
        return await self.api_call("chat.meMessage", json=kwargs)

    async def chat_postEphemeral(
        self, *, channel: str, user: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sends an ephemeral message to a user in a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            user (str): The id of user who should see the message. e.g. 'U0BPQUNTA'
            text (str): The message you'd like to share. e.g. 'Hello world'
                text is not required when presenting blocks.
            blocks (list): A list of either dict values or `slack_sdk.models.blocks.Block` objects.
                Blocks are required when not presenting text.
                e.g. [{"type": "section", "text": {"type": "plain_text", "text": "Hello world"}}]
        """
        kwargs.update({"channel": channel, "user": user})
        _parse_web_class_objects(kwargs)
        _warn_if_text_is_missing("chat.postEphemeral", kwargs)
        return await self.api_call("chat.postEphemeral", json=kwargs)

    async def chat_postMessage(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Sends a message to a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            text (str): The message you'd like to share. e.g. 'Hello world'
                text is not required when presenting blocks.
            blocks (list): A list of either dict values or `slack_sdk.models.blocks.Block` objects.
                Blocks are required when not presenting text.
                e.g. [{"type": "section", "text": {"type": "plain_text", "text": "Hello world"}}]
        """
        kwargs.update({"channel": channel})
        _parse_web_class_objects(kwargs)
        _warn_if_text_is_missing("chat.postMessage", kwargs)
        return await self.api_call("chat.postMessage", json=kwargs)

    async def chat_scheduleMessage(
        self, *, channel: str, post_at: str, text: str, **kwargs
    ) -> AsyncSlackResponse:
        """Schedules a message.

        Args:
            channel (str): The channel the scheduled_message is posting to. e.g. 'C1234567890'
            post_at (str): Unix EPOCH timestamp of time in future to send the message. e.g. '299876400'
            text (str): The message you'd like to send. e.g. 'Hello world'
        """
        kwargs.update({"channel": channel, "post_at": post_at, "text": text})
        _parse_web_class_objects(kwargs)
        _warn_if_text_is_missing("chat.scheduleMessage", kwargs)
        return await self.api_call("chat.scheduleMessage", json=kwargs)

    async def chat_unfurl(
        self, *, channel: str, ts: str, unfurls: dict, **kwargs
    ) -> AsyncSlackResponse:
        """Provide custom unfurl behavior for user-posted URLs.

        Args:
            channel (str): The Channel ID of the message. e.g. 'C1234567890'
            ts (str): Timestamp of the message to add unfurl behavior to. e.g. '1234567890.123456'
            unfurls (dict): a dict of the specific URLs you're offering an unfurl for.
                e.g. {"https://example.com/": {"text": "Every day is the test."}}
        """
        kwargs.update({"channel": channel, "ts": ts, "unfurls": unfurls})
        return await self.api_call("chat.unfurl", json=kwargs)

    async def chat_update(
        self, *, channel: str, ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Updates a message in a channel.

        Args:
            channel (str): The channel containing the message to be updated. e.g. 'C1234567890'
            ts (str): Timestamp of the message to be updated. e.g. '1234567890.123456'
            text (str): The message you'd like to share. e.g. 'Hello world'
                text is not required when presenting blocks.
            blocks (list): A list of either dict values or `slack_sdk.models.blocks.Block` objects.
                Blocks are required when not presenting text.
                e.g. [{"type": "section", "text": {"type": "plain_text", "text": "Hello world"}}]
        """
        kwargs.update({"channel": channel, "ts": ts})
        _parse_web_class_objects(kwargs)
        _warn_if_text_is_missing("chat.update", kwargs)
        return await self.api_call("chat.update", json=kwargs)

    async def chat_scheduledMessages_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists all scheduled messages."""
        return await self.api_call("chat.scheduledMessages.list", params=kwargs)

    async def conversations_archive(
        self, *, channel: str, **kwargs
    ) -> AsyncSlackResponse:
        """Archives a conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("conversations.archive", json=kwargs)

    async def conversations_close(
        self, *, channel: str, **kwargs
    ) -> AsyncSlackResponse:
        """Closes a direct message or multi-person direct message.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("conversations.close", json=kwargs)

    async def conversations_create(self, *, name: str, **kwargs) -> AsyncSlackResponse:
        """Initiates a public or private channel-based conversation

        Args:
            name (str): The name of the channel. e.g. 'mychannel'
        """
        kwargs.update({"name": name})
        return await self.api_call("conversations.create", json=kwargs)

    async def conversations_history(
        self, *, channel: str, **kwargs
    ) -> AsyncSlackResponse:
        """Fetches a conversation's history of messages and events.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call(
            "conversations.history", http_verb="GET", params=kwargs
        )

    async def conversations_info(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Retrieve information about a conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("conversations.info", http_verb="GET", params=kwargs)

    async def conversations_invite(
        self, *, channel: str, users: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Invites users to a channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            users (str or list): An list of user id's to invite. e.g. ['U2345678901', 'U3456789012']
        """
        kwargs.update({"channel": channel})
        if isinstance(users, (list, Tuple)):
            kwargs.update({"users": ",".join(users)})
        else:
            kwargs.update({"users": users})
        return await self.api_call("conversations.invite", params=kwargs)

    async def conversations_join(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Joins an existing conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("conversations.join", json=kwargs)

    async def conversations_kick(
        self, *, channel: str, user: str, **kwargs
    ) -> AsyncSlackResponse:
        """Removes a user from a conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            user (str): The id of the user to kick. e.g. 'U2345678901'
        """
        kwargs.update({"channel": channel, "user": user})
        return await self.api_call("conversations.kick", json=kwargs)

    async def conversations_leave(
        self, *, channel: str, **kwargs
    ) -> AsyncSlackResponse:
        """Leaves a conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("conversations.leave", json=kwargs)

    async def conversations_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists all channels in a Slack team."""
        return await self.api_call("conversations.list", http_verb="GET", params=kwargs)

    async def conversations_mark(
        self, *, channel: str, ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the read cursor in a channel.

        Args:
            channel (str): Channel or conversation to set the read cursor for e.g. 'C1234567890'
            ts (str): Unique identifier of message to mark as most recently seen in the convo e.g. '1593473566.000200'
        """
        kwargs.update({"channel": channel, "ts": ts})
        return await self.api_call("conversations.mark", json=kwargs)

    async def conversations_members(
        self, *, channel: str, **kwargs
    ) -> AsyncSlackResponse:
        """Retrieve members of a conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call(
            "conversations.members", http_verb="GET", params=kwargs
        )

    async def conversations_open(self, **kwargs) -> AsyncSlackResponse:
        """Opens or resumes a direct message or multi-person direct message."""
        return await self.api_call("conversations.open", json=kwargs)

    async def conversations_rename(
        self, *, channel: str, name: str, **kwargs
    ) -> AsyncSlackResponse:
        """Renames a conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            name (str): The new channel name. e.g. 'newchannel'
        """
        kwargs.update({"channel": channel, "name": name})
        return await self.api_call("conversations.rename", json=kwargs)

    async def conversations_replies(
        self, *, channel: str, ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Retrieve a thread of messages posted to a conversation

        Args:
            channel (str): Conversation ID to fetch thread from. e.g. 'C1234567890'
            ts (str): Unique identifier of a thread's parent message. e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "ts": ts})
        return await self.api_call(
            "conversations.replies", http_verb="GET", params=kwargs
        )

    async def conversations_setPurpose(
        self, *, channel: str, purpose: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the purpose for a conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            purpose (str): The new purpose for the channel. e.g. 'My Purpose'
        """
        kwargs.update({"channel": channel, "purpose": purpose})
        return await self.api_call("conversations.setPurpose", json=kwargs)

    async def conversations_setTopic(
        self, *, channel: str, topic: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the topic for a conversation.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            topic (str): The new topic for the channel. e.g. 'My Topic'
        """
        kwargs.update({"channel": channel, "topic": topic})
        return await self.api_call("conversations.setTopic", json=kwargs)

    async def conversations_unarchive(
        self, *, channel: str, **kwargs
    ) -> AsyncSlackResponse:
        """Reverses conversation archival.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("conversations.unarchive", json=kwargs)

    async def dialog_open(
        self, *, dialog: dict, trigger_id: str, **kwargs
    ) -> AsyncSlackResponse:
        """Open a dialog with a user.

        Args:
            dialog (dict): A dictionary of dialog arguments.
                {
                    "callback_id": "46eh782b0",
                    "title": "Request something",
                    "submit_label": "Request",
                    "state": "Max",
                    "elements": [
                        {
                            "type": "text",
                            "label": "Origin",
                            "name": "loc_origin"
                        },
                        {
                            "type": "text",
                            "label": "Destination",
                            "name": "loc_destination"
                        }
                    ]
                }
            trigger_id (str): The trigger id of a recent message interaction.
                e.g. '12345.98765.abcd2358fdea'
        """
        kwargs.update({"dialog": dialog, "trigger_id": trigger_id})
        return await self.api_call("dialog.open", json=kwargs)

    async def dnd_endDnd(self, **kwargs) -> AsyncSlackResponse:
        """Ends the current user's Do Not Disturb session immediately."""
        return await self.api_call("dnd.endDnd", json=kwargs)

    async def dnd_endSnooze(self, **kwargs) -> AsyncSlackResponse:
        """Ends the current user's snooze mode immediately."""
        return await self.api_call("dnd.endSnooze", json=kwargs)

    async def dnd_info(self, **kwargs) -> AsyncSlackResponse:
        """Retrieves a user's current Do Not Disturb status."""
        return await self.api_call("dnd.info", http_verb="GET", params=kwargs)

    async def dnd_setSnooze(self, *, num_minutes: int, **kwargs) -> AsyncSlackResponse:
        """Turns on Do Not Disturb mode for the current user, or changes its duration.

        Args:
            num_minutes (int): The snooze duration. e.g. 60
        """
        kwargs.update({"num_minutes": num_minutes})
        return await self.api_call("dnd.setSnooze", http_verb="GET", params=kwargs)

    async def dnd_teamInfo(
        self, users: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Retrieves the Do Not Disturb status for users on a team.

        Args:
            users (str or list): User IDs to fetch information e.g. 'U123,U234' or ["U123", "U234"]
        """
        if isinstance(users, (list, Tuple)):
            kwargs.update({"users": ",".join(users)})
        else:
            kwargs.update({"users": users})
        return await self.api_call("dnd.teamInfo", http_verb="GET", params=kwargs)

    async def emoji_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists custom emoji for a team."""
        return await self.api_call("emoji.list", http_verb="GET", params=kwargs)

    async def files_comments_delete(
        self, *, file: str, id: str, **kwargs  # skipcq: PYL-W0622
    ) -> AsyncSlackResponse:
        """Deletes an existing comment on a file.

        Args:
            file (str): The file id. e.g. 'F1234467890'
            id (str): The file comment id. e.g. 'Fc1234567890'
        """
        kwargs.update({"file": file, "id": id})
        return await self.api_call("files.comments.delete", json=kwargs)

    async def files_delete(self, *, file: str, **kwargs) -> AsyncSlackResponse:
        """Deletes a file.

        Args:
            file (str): The file id. e.g. 'F1234467890'
        """
        kwargs.update({"file": file})
        return await self.api_call("files.delete", json=kwargs)

    async def files_info(self, *, file: str, **kwargs) -> AsyncSlackResponse:
        """Gets information about a team file.

        Args:
            file (str): The file id. e.g. 'F1234467890'
        """
        kwargs.update({"file": file})
        return await self.api_call("files.info", http_verb="GET", params=kwargs)

    async def files_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists & filters team files."""
        return await self.api_call("files.list", http_verb="GET", params=kwargs)

    async def files_remote_info(self, **kwargs) -> AsyncSlackResponse:
        """Retrieve information about a remote file added to Slack."""
        return await self.api_call("files.remote.info", http_verb="GET", params=kwargs)

    async def files_remote_list(self, **kwargs) -> AsyncSlackResponse:
        """Retrieve information about a remote file added to Slack."""
        return await self.api_call("files.remote.list", http_verb="GET", params=kwargs)

    async def files_remote_add(
        self, *, external_id: str, external_url: str, title: str, **kwargs
    ) -> AsyncSlackResponse:
        """Adds a file from a remote service.

        Args:
            external_id (str): Creator defined GUID for the file. e.g. '123456'
            external_url (str): URL of the remote file. e.g. 'http://example.com/my_cloud_service_file/abc123'
            title (str): Title of the file being shared. e.g. 'Danger, High Voltage!'
        """
        kwargs.update(
            {"external_id": external_id, "external_url": external_url, "title": title}
        )
        files = None
        # preview_image (file): Preview of the document via multipart/form-data.
        if "preview_image" in kwargs:
            files = {"preview_image": kwargs.pop("preview_image")}

        return await self.api_call(
            # Intentionally using "POST" method over "GET" here
            "files.remote.add",
            http_verb="POST",
            data=kwargs,
            files=files,
        )

    async def files_remote_update(self, **kwargs) -> AsyncSlackResponse:
        """Updates an existing remote file."""
        return await self.api_call(
            "files.remote.update", http_verb="GET", params=kwargs
        )

    async def files_remote_remove(self, **kwargs) -> AsyncSlackResponse:
        """Remove a remote file."""
        return await self.api_call(
            "files.remote.remove", http_verb="GET", params=kwargs
        )

    async def files_remote_share(
        self, *, channels: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Share a remote file into a channel.

        Args:
            channels (str or list): Comma-separated list of channel IDs where the file will be shared.
                e.g. ['C1234567890', 'C2345678901']
        """
        if isinstance(channels, (list, Tuple)):
            kwargs.update({"channels": ",".join(channels)})
        else:
            kwargs.update({"channels": channels})
        return await self.api_call("files.remote.share", http_verb="GET", params=kwargs)

    async def files_revokePublicURL(self, *, file: str, **kwargs) -> AsyncSlackResponse:
        """Revokes public/external sharing access for a file

        Args:
            file (str): The file id. e.g. 'F1234467890'
        """
        kwargs.update({"file": file})
        return await self.api_call("files.revokePublicURL", json=kwargs)

    async def files_sharedPublicURL(self, *, file: str, **kwargs) -> AsyncSlackResponse:
        """Enables a file for public/external sharing.

        Args:
            file (str): The file id. e.g. 'F1234467890'
        """
        kwargs.update({"file": file})
        return await self.api_call("files.sharedPublicURL", json=kwargs)

    async def files_upload(
        self, *, file: Union[str, bytes, IOBase] = None, content: str = None, **kwargs
    ) -> AsyncSlackResponse:
        """Uploads or creates a file.

        Args:
            file (str): Supply a file path.
                when you'd like to upload a specific file. e.g. 'dramacat.gif'
            content (str): Supply content when you'd like to create an
                editable text file containing the specified text. e.g. 'launch plan'
        Raises:
            SlackRequestError: If neither or both the `file` and `content` args are specified.
        """
        if file is None and content is None:
            raise e.SlackRequestError("The file or content argument must be specified.")
        if file is not None and content is not None:
            raise e.SlackRequestError(
                "You cannot specify both the file and the content argument."
            )

        if file:
            if "filename" not in kwargs and isinstance(file, str):
                # use the local filename if filename is missing
                kwargs["filename"] = file.split(os.path.sep)[-1]
            return await self.api_call(
                "files.upload", files={"file": file}, data=kwargs
            )
        data = kwargs.copy()
        data.update({"content": content})
        return await self.api_call("files.upload", data=data)

    async def groups_archive(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Archives a private channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("groups.archive", json=kwargs)

    async def groups_create(self, *, name: str, **kwargs) -> AsyncSlackResponse:
        """Creates a private channel.

        Args:
            name (str): The name of the private group. e.g. 'mychannel'
        """
        kwargs.update({"name": name})
        return await self.api_call("groups.create", json=kwargs)

    async def groups_createChild(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Clones and archives a private channel.

        Args:
            channel (str): The group id. e.g. 'G1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("groups.createChild", http_verb="GET", params=kwargs)

    async def groups_history(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Fetches history of messages and events from a private channel.

        Args:
            channel (str): The group id. e.g. 'G1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("groups.history", http_verb="GET", params=kwargs)

    async def groups_info(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Gets information about a private channel.

        Args:
            channel (str): The group id. e.g. 'G1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("groups.info", http_verb="GET", params=kwargs)

    async def groups_invite(
        self, *, channel: str, user: str, **kwargs
    ) -> AsyncSlackResponse:
        """Invites a user to a private channel.

        Args:
            channel (str): The group id. e.g. 'G1234567890'
            user (str): The user id. e.g. 'U1234567890'
        """
        kwargs.update({"channel": channel, "user": user})
        return await self.api_call("groups.invite", json=kwargs)

    async def groups_kick(
        self, *, channel: str, user: str, **kwargs
    ) -> AsyncSlackResponse:
        """Removes a user from a private channel.

        Args:
            channel (str): The group id. e.g. 'G1234567890'
            user (str): The user id. e.g. 'U1234567890'
        """
        kwargs.update({"channel": channel, "user": user})
        return await self.api_call("groups.kick", json=kwargs)

    async def groups_leave(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Leaves a private channel.

        Args:
            channel (str): The group id. e.g. 'G1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("groups.leave", json=kwargs)

    async def groups_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists private channels that the calling user has access to."""
        return await self.api_call("groups.list", http_verb="GET", params=kwargs)

    async def groups_mark(
        self, *, channel: str, ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the read cursor in a private channel.

        Args:
            channel (str): Private channel to set reading cursor in. e.g. 'C1234567890'
            ts (str): Timestamp of the most recently seen message. e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "ts": ts})
        return await self.api_call("groups.mark", json=kwargs)

    async def groups_open(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Opens a private channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("groups.open", json=kwargs)

    async def groups_rename(
        self, *, channel: str, name: str, **kwargs
    ) -> AsyncSlackResponse:
        """Renames a private channel.

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            name (str): The new channel name. e.g. 'newchannel'
        """
        kwargs.update({"channel": channel, "name": name})
        return await self.api_call("groups.rename", json=kwargs)

    async def groups_replies(
        self, *, channel: str, thread_ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Retrieve a thread of messages posted to a private channel

        Args:
            channel (str): The channel id. e.g. 'C1234567890'
            thread_ts (str): The timestamp of an existing message with 0 or more replies.
                e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "thread_ts": thread_ts})
        return await self.api_call("groups.replies", http_verb="GET", params=kwargs)

    async def groups_setPurpose(
        self, *, channel: str, purpose: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the purpose for a private channel.

        Args:
            channel (str): The channel id. e.g. 'G1234567890'
            purpose (str): The new purpose for the channel. e.g. 'My Purpose'
        """
        kwargs.update({"channel": channel, "purpose": purpose})
        return await self.api_call("groups.setPurpose", json=kwargs)

    async def groups_setTopic(
        self, *, channel: str, topic: str, **kwargs
    ) -> AsyncSlackResponse:
        """Sets the topic for a private channel.

        Args:
            channel (str): The channel id. e.g. 'G1234567890'
            topic (str): The new topic for the channel. e.g. 'My Topic'
        """
        kwargs.update({"channel": channel, "topic": topic})
        return await self.api_call("groups.setTopic", json=kwargs)

    async def groups_unarchive(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Unarchives a private channel.

        Args:
            channel (str): The channel id. e.g. 'G1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("groups.unarchive", json=kwargs)

    async def im_close(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Close a direct message channel.

        Args:
            channel (str): Direct message channel to close. e.g. 'D1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("im.close", json=kwargs)

    async def im_history(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Fetches history of messages and events from direct message channel.

        Args:
            channel (str): Direct message channel to fetch history from. e.g. 'D1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("im.history", http_verb="GET", params=kwargs)

    async def im_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists direct message channels for the calling user."""
        return await self.api_call("im.list", http_verb="GET", params=kwargs)

    async def im_mark(self, *, channel: str, ts: str, **kwargs) -> AsyncSlackResponse:
        """Sets the read cursor in a direct message channel.

        Args:
            channel (str): Direct message channel to set reading cursor in. e.g. 'D1234567890'
            ts (str): Timestamp of the most recently seen message. e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "ts": ts})
        return await self.api_call("im.mark", json=kwargs)

    async def im_open(self, *, user: str, **kwargs) -> AsyncSlackResponse:
        """Opens a direct message channel.

        Args:
            user (str): The user id to open a DM with. e.g. 'W1234567890'
        """
        kwargs.update({"user": user})
        return await self.api_call("im.open", json=kwargs)

    async def im_replies(
        self, *, channel: str, thread_ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Retrieve a thread of messages posted to a direct message conversation

        Args:
            channel (str): Direct message channel to fetch thread from. e.g. 'C1234567890'
            thread_ts (str): The timestamp of an existing message with 0 or more replies.
                e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "thread_ts": thread_ts})
        return await self.api_call("im.replies", http_verb="GET", params=kwargs)

    async def migration_exchange(
        self, *, users: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """For Enterprise Grid workspaces, map local user IDs to global user IDs

        Args:
            users (str or list): A list of user ids, up to 400 per request.
                e.g. ['W1234567890', 'U2345678901', 'U3456789012']
        """
        if isinstance(users, (list, Tuple)):
            kwargs.update({"users": ",".join(users)})
        else:
            kwargs.update({"users": users})
        return await self.api_call("migration.exchange", http_verb="GET", params=kwargs)

    async def mpim_close(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Closes a multiparty direct message channel.

        Args:
            channel (str): Multiparty Direct message channel to close. e.g. 'G1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("mpim.close", json=kwargs)

    async def mpim_history(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Fetches history of messages and events from a multiparty direct message.

        Args:
            channel (str): Multiparty direct message to fetch history for. e.g. 'G1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("mpim.history", http_verb="GET", params=kwargs)

    async def mpim_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists multiparty direct message channels for the calling user."""
        return await self.api_call("mpim.list", http_verb="GET", params=kwargs)

    async def mpim_mark(self, *, channel: str, ts: str, **kwargs) -> AsyncSlackResponse:
        """Sets the read cursor in a multiparty direct message channel.

        Args:
            channel (str): Multiparty direct message channel to set reading cursor in.
                e.g. 'G1234567890'
            ts (str): Timestamp of the most recently seen message.
                e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "ts": ts})
        return await self.api_call("mpim.mark", json=kwargs)

    async def mpim_open(
        self, *, users: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """This method opens a multiparty direct message.

        Args:
            users (str or list): A lists of user ids. The ordering of the users
                is preserved whenever a MPIM group is returned.
                e.g. ['W1234567890', 'U2345678901', 'U3456789012']
        """
        if isinstance(users, (list, Tuple)):
            kwargs.update({"users": ",".join(users)})
        else:
            kwargs.update({"users": users})
        return await self.api_call("mpim.open", params=kwargs)

    async def mpim_replies(
        self, *, channel: str, thread_ts: str, **kwargs
    ) -> AsyncSlackResponse:
        """Retrieve a thread of messages posted to a direct message conversation from a
        multiparty direct message.

        Args:
            channel (str): Multiparty direct message channel to fetch thread from.
                e.g. 'G1234567890'
            thread_ts (str): Unique identifier of a thread's parent message.
                e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel, "thread_ts": thread_ts})
        return await self.api_call("mpim.replies", http_verb="GET", params=kwargs)

    async def oauth_v2_access(
        self,
        *,
        client_id: str,
        client_secret: str,
        code: str,
        redirect_uri: Optional[str] = None,
        **kwargs
    ) -> AsyncSlackResponse:
        """Exchanges a temporary OAuth verifier code for an access token.

        Args:
            client_id (str): Issued when you created your application. e.g. '4b39e9-752c4'
            client_secret (str): Issued when you created your application. e.g. '33fea0113f5b1'
            code (str): The code param returned via the OAuth callback. e.g. 'ccdaa72ad'
            redirect_uri (optional str): Must match the originally submitted URI
                (if one was sent). e.g. 'https://example.com'
        """
        if redirect_uri is not None:
            kwargs.update({"redirect_uri": redirect_uri})
        kwargs.update({"code": code})
        return await self.api_call(
            "oauth.v2.access",
            data=kwargs,
            auth={"client_id": client_id, "client_secret": client_secret},
        )

    async def oauth_access(
        self,
        *,
        client_id: str,
        client_secret: str,
        code: str,
        redirect_uri: Optional[str] = None,
        **kwargs
    ) -> AsyncSlackResponse:
        """Exchanges a temporary OAuth verifier code for an access token.

        Args:
            client_id (str): Issued when you created your application. e.g. '4b39e9-752c4'
            client_secret (str): Issued when you created your application. e.g. '33fea0113f5b1'
            code (str): The code param returned via the OAuth callback. e.g. 'ccdaa72ad'
            redirect_uri (optional str): Must match the originally submitted URI
                (if one was sent). e.g. 'https://example.com'
        """
        if redirect_uri is not None:
            kwargs.update({"redirect_uri": redirect_uri})
        kwargs.update({"code": code})
        return await self.api_call(
            "oauth.access",
            data=kwargs,
            auth={"client_id": client_id, "client_secret": client_secret},
        )

    async def pins_add(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Pins an item to a channel.

        Args:
            channel (str): Channel to pin the item in. e.g. 'C1234567890'
            file (str): File id to pin. e.g. 'F1234567890'
            file_comment (str): File comment to pin. e.g. 'Fc1234567890'
            timestamp (str): Timestamp of message to pin. e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("pins.add", json=kwargs)

    async def pins_list(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Lists items pinned to a channel.

        Args:
            channel (str): Channel to get pinned items for. e.g. 'C1234567890'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("pins.list", http_verb="GET", params=kwargs)

    async def pins_remove(self, *, channel: str, **kwargs) -> AsyncSlackResponse:
        """Un-pins an item from a channel.

        Args:
            channel (str): Channel to pin the item in. e.g. 'C1234567890'
            file (str): File id to pin. e.g. 'F1234567890'
            file_comment (str): File comment to pin. e.g. 'Fc1234567890'
            timestamp (str): Timestamp of message to pin. e.g. '1234567890.123456'
        """
        kwargs.update({"channel": channel})
        return await self.api_call("pins.remove", json=kwargs)

    async def reactions_add(self, *, name: str, **kwargs) -> AsyncSlackResponse:
        """Adds a reaction to an item.

        Args:
            name (str): Reaction (emoji) name. e.g. 'thumbsup'
            channel (str): Channel where the message to add reaction to was posted.
                e.g. 'C1234567890'
            timestamp (str): Timestamp of the message to add reaction to. e.g. '1234567890.123456'
        """
        kwargs.update({"name": name})
        return await self.api_call("reactions.add", json=kwargs)

    async def reactions_get(self, **kwargs) -> AsyncSlackResponse:
        """Gets reactions for an item."""
        return await self.api_call("reactions.get", http_verb="GET", params=kwargs)

    async def reactions_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists reactions made by a user."""
        return await self.api_call("reactions.list", http_verb="GET", params=kwargs)

    async def reactions_remove(self, *, name: str, **kwargs) -> AsyncSlackResponse:
        """Removes a reaction from an item.

        Args:
            name (str): Reaction (emoji) name. e.g. 'thumbsup'
        """
        kwargs.update({"name": name})
        return await self.api_call("reactions.remove", json=kwargs)

    async def reminders_add(
        self, *, text: str, time: str, **kwargs
    ) -> AsyncSlackResponse:
        """Creates a reminder.

        Args:
            text (str): The content of the reminder. e.g. 'eat a banana'
            time (str): When this reminder should happen:
                the Unix timestamp (up to five years from now e.g. '1602288000'),
                the number of seconds until the reminder (if within 24 hours),
                or a natural language description (Ex. 'in 15 minutes' or 'every Thursday')
        """
        kwargs.update({"text": text, "time": time})
        return await self.api_call("reminders.add", json=kwargs)

    async def reminders_complete(
        self, *, reminder: str, **kwargs
    ) -> AsyncSlackResponse:
        """Marks a reminder as complete.

        Args:
            reminder (str): The ID of the reminder to be marked as complete.
                e.g. 'Rm12345678'
        """
        kwargs.update({"reminder": reminder})
        return await self.api_call("reminders.complete", json=kwargs)

    async def reminders_delete(self, *, reminder: str, **kwargs) -> AsyncSlackResponse:
        """Deletes a reminder.

        Args:
            reminder (str): The ID of the reminder. e.g. 'Rm12345678'
        """
        kwargs.update({"reminder": reminder})
        return await self.api_call("reminders.delete", json=kwargs)

    async def reminders_info(self, *, reminder: str, **kwargs) -> AsyncSlackResponse:
        """Gets information about a reminder.

        Args:
            reminder (str): The ID of the reminder. e.g. 'Rm12345678'
        """
        kwargs.update({"reminder": reminder})
        return await self.api_call("reminders.info", http_verb="GET", params=kwargs)

    async def reminders_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists all reminders created by or for a given user."""
        return await self.api_call("reminders.list", http_verb="GET", params=kwargs)

    async def rtm_connect(self, **kwargs) -> AsyncSlackResponse:
        """Starts a Real Time Messaging session."""
        return await self.api_call("rtm.connect", http_verb="GET", params=kwargs)

    async def rtm_start(self, **kwargs) -> AsyncSlackResponse:
        """Starts a Real Time Messaging session."""
        return await self.api_call("rtm.start", http_verb="GET", params=kwargs)

    async def search_all(self, *, query: str, **kwargs) -> AsyncSlackResponse:
        """Searches for messages and files matching a query.

        Args:
            query (str): Search query. May contains booleans, etc.
                e.g. 'pickleface'
        """
        kwargs.update({"query": query})
        return await self.api_call("search.all", http_verb="GET", params=kwargs)

    async def search_files(self, *, query: str, **kwargs) -> AsyncSlackResponse:
        """Searches for files matching a query.

        Args:
            query (str): Search query. May contains booleans, etc.
                e.g. 'pickleface'
        """
        kwargs.update({"query": query})
        return await self.api_call("search.files", http_verb="GET", params=kwargs)

    async def search_messages(self, *, query: str, **kwargs) -> AsyncSlackResponse:
        """Searches for messages matching a query.

        Args:
            query (str): Search query. May contains booleans, etc.
                e.g. 'pickleface'
        """
        kwargs.update({"query": query})
        return await self.api_call("search.messages", http_verb="GET", params=kwargs)

    async def stars_add(self, **kwargs) -> AsyncSlackResponse:
        """Adds a star to an item.

        Args:
            channel (str): Channel to add star to, or channel where the message to add
                star to was posted (used with timestamp). e.g. 'C1234567890'
            file (str): File to add star to. e.g. 'F1234567890'
            file_comment (str): File comment to add star to. e.g. 'Fc1234567890'
            timestamp (str): Timestamp of the message to add star to. e.g. '1234567890.123456'
        """
        return await self.api_call("stars.add", json=kwargs)

    async def stars_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists stars for a user."""
        return await self.api_call("stars.list", http_verb="GET", params=kwargs)

    async def stars_remove(self, **kwargs) -> AsyncSlackResponse:
        """Removes a star from an item.

        Args:
            channel (str): Channel to remove star from, or channel where
                the message to remove star from was posted (used with timestamp). e.g. 'C1234567890'
            file (str): File to remove star from. e.g. 'F1234567890'
            file_comment (str): File comment to remove star from. e.g. 'Fc1234567890'
            timestamp (str): Timestamp of the message to remove star from. e.g. '1234567890.123456'
        """
        return await self.api_call("stars.remove", json=kwargs)

    async def team_accessLogs(self, **kwargs) -> AsyncSlackResponse:
        """Gets the access logs for the current team."""
        return await self.api_call("team.accessLogs", http_verb="GET", params=kwargs)

    async def team_billableInfo(self, **kwargs) -> AsyncSlackResponse:
        """Gets billable users information for the current team."""
        return await self.api_call("team.billableInfo", http_verb="GET", params=kwargs)

    async def team_info(self, **kwargs) -> AsyncSlackResponse:
        """Gets information about the current team."""
        return await self.api_call("team.info", http_verb="GET", params=kwargs)

    async def team_integrationLogs(self, **kwargs) -> AsyncSlackResponse:
        """Gets the integration logs for the current team."""
        return await self.api_call(
            "team.integrationLogs", http_verb="GET", params=kwargs
        )

    async def team_profile_get(self, **kwargs) -> AsyncSlackResponse:
        """Retrieve a team's profile."""
        return await self.api_call("team.profile.get", http_verb="GET", params=kwargs)

    async def usergroups_create(self, *, name: str, **kwargs) -> AsyncSlackResponse:
        """Create a User Group

        Args:
            name (str): A name for the User Group. Must be unique among User Groups.
                e.g. 'My Test Team'
        """
        kwargs.update({"name": name})
        return await self.api_call("usergroups.create", json=kwargs)

    async def usergroups_disable(
        self, *, usergroup: str, **kwargs
    ) -> AsyncSlackResponse:
        """Disable an existing User Group

        Args:
            usergroup (str): The encoded ID of the User Group to disable.
                e.g. 'S0604QSJC'
        """
        kwargs.update({"usergroup": usergroup})
        return await self.api_call("usergroups.disable", json=kwargs)

    async def usergroups_enable(
        self, *, usergroup: str, **kwargs
    ) -> AsyncSlackResponse:
        """Enable a User Group

        Args:
            usergroup (str): The encoded ID of the User Group to enable.
                e.g. 'S0604QSJC'
        """
        kwargs.update({"usergroup": usergroup})
        return await self.api_call("usergroups.enable", json=kwargs)

    async def usergroups_list(self, **kwargs) -> AsyncSlackResponse:
        """List all User Groups for a team"""
        return await self.api_call("usergroups.list", http_verb="GET", params=kwargs)

    async def usergroups_update(
        self, *, usergroup: str, **kwargs
    ) -> AsyncSlackResponse:
        """Update an existing User Group

        Args:
            usergroup (str): The encoded ID of the User Group to update.
                e.g. 'S0604QSJC'
        """
        kwargs.update({"usergroup": usergroup})
        return await self.api_call("usergroups.update", json=kwargs)

    async def usergroups_users_list(
        self, *, usergroup: str, **kwargs
    ) -> AsyncSlackResponse:
        """List all users in a User Group

        Args:
            usergroup (str): The encoded ID of the User Group to update.
                e.g. 'S0604QSJC'
        """
        kwargs.update({"usergroup": usergroup})
        return await self.api_call(
            "usergroups.users.list", http_verb="GET", params=kwargs
        )

    async def usergroups_users_update(
        self, *, usergroup: str, users: Union[str, Sequence[str]], **kwargs
    ) -> AsyncSlackResponse:
        """Update the list of users for a User Group

        Args:
            usergroup (str): The encoded ID of the User Group to update.
                e.g. 'S0604QSJC'
            users (str or list): A list user IDs that represent the entire list of
                users for the User Group. e.g. ['U060R4BJ4', 'U060RNRCZ']
        """
        kwargs.update({"usergroup": usergroup})
        if isinstance(users, (list, Tuple)):
            kwargs.update({"users": ",".join(users)})
        else:
            kwargs.update({"users": users})
        return await self.api_call("usergroups.users.update", params=kwargs)

    async def users_conversations(self, **kwargs) -> AsyncSlackResponse:
        """List conversations the calling user may access."""
        return await self.api_call(
            "users.conversations", http_verb="GET", params=kwargs
        )

    async def users_deletePhoto(self, **kwargs) -> AsyncSlackResponse:
        """Delete the user profile photo"""
        return await self.api_call("users.deletePhoto", http_verb="GET", params=kwargs)

    async def users_getPresence(self, *, user: str, **kwargs) -> AsyncSlackResponse:
        """Gets user presence information.

        Args:
            user (str): User to get presence info on. Defaults to the authed user.
                e.g. 'W1234567890'
        """
        kwargs.update({"user": user})
        return await self.api_call("users.getPresence", http_verb="GET", params=kwargs)

    async def users_identity(self, **kwargs) -> AsyncSlackResponse:
        """Get a user's identity."""
        return await self.api_call("users.identity", http_verb="GET", params=kwargs)

    async def users_info(self, *, user: str, **kwargs) -> AsyncSlackResponse:
        """Gets information about a user.

        Args:
            user (str): User to get info on.
                e.g. 'W1234567890'
        """
        kwargs.update({"user": user})
        return await self.api_call("users.info", http_verb="GET", params=kwargs)

    async def users_list(self, **kwargs) -> AsyncSlackResponse:
        """Lists all users in a Slack team."""
        return await self.api_call("users.list", http_verb="GET", params=kwargs)

    async def users_lookupByEmail(self, *, email: str, **kwargs) -> AsyncSlackResponse:
        """Find a user with an email address.

        Args:
            email (str): An email address belonging to a user in the workspace.
                e.g. 'spengler@ghostbusters.example.com'
        """
        kwargs.update({"email": email})
        return await self.api_call(
            "users.lookupByEmail", http_verb="GET", params=kwargs
        )

    async def users_setPhoto(
        self, *, image: Union[str, IOBase], **kwargs
    ) -> AsyncSlackResponse:
        """Set the user profile photo

        Args:
            image (str): Supply the path of the image you'd like to upload.
                e.g. 'myimage.png'
        """
        return await self.api_call(
            "users.setPhoto", files={"image": image}, data=kwargs
        )

    async def users_setPresence(self, *, presence: str, **kwargs) -> AsyncSlackResponse:
        """Manually sets user presence.

        Args:
            presence (str): Either 'auto' or 'away'.
        """
        kwargs.update({"presence": presence})
        return await self.api_call("users.setPresence", json=kwargs)

    async def users_profile_get(self, **kwargs) -> AsyncSlackResponse:
        """Retrieves a user's profile information."""
        return await self.api_call("users.profile.get", http_verb="GET", params=kwargs)

    async def users_profile_set(self, **kwargs) -> AsyncSlackResponse:
        """Set the profile information for a user."""
        return await self.api_call("users.profile.set", json=kwargs)

    async def views_open(
        self, *, trigger_id: str, view: Union[dict, View], **kwargs
    ) -> AsyncSlackResponse:
        """Open a view for a user.
        See https://api.slack.com/block-kit/surfaces/modals for details.

        Args:
            trigger_id (str): Exchange a trigger to post to the user.
                e.g. '12345.98765.abcd2358fdea'
            view (dict or View): The view payload.
        """
        kwargs.update({"trigger_id": trigger_id})
        if isinstance(view, View):
            kwargs.update({"view": view.to_dict()})
        else:
            kwargs.update({"view": view})
        return await self.api_call("views.open", json=kwargs)

    async def views_push(
        self, *, trigger_id: str, view: Union[dict, View], **kwargs
    ) -> AsyncSlackResponse:
        """Push a view onto the stack of a root view.

        Push a new view onto the existing view stack by passing a view
        payload and a valid trigger_id generated from an interaction
        within the existing modal.

        Read the modals documentation (https://api.slack.com/block-kit/surfaces/modals)
        to learn more about the lifecycle and intricacies of views.

        Args:
            trigger_id (str): Exchange a trigger to post to the user.
                e.g. '12345.98765.abcd2358fdea'
            view (dict or View): The view payload.
        """
        kwargs.update({"trigger_id": trigger_id, "view": view})
        if isinstance(view, View):
            kwargs.update({"view": view.to_dict()})
        else:
            kwargs.update({"view": view})
        return await self.api_call("views.push", json=kwargs)

    async def views_update(
        self,
        *,
        view: Union[dict, View],
        external_id: str = None,
        view_id: str = None,
        **kwargs
    ) -> AsyncSlackResponse:
        """Update an existing view.

        Update a view by passing a new view definition along with the
        view_id returned in views.open or the external_id.

        See the modals documentation (https://api.slack.com/block-kit/surfaces/modals#updating_views)
        to learn more about updating views and avoiding race conditions with the hash argument.

        Args:
            view (dict or View): The view payload.
            external_id (str): A unique identifier of the view set by the developer.
                e.g. 'bmarley_view2'
            view_id (str): A unique identifier of the view to be updated.
                e.g. 'VMM512F2U'
        Raises:
            SlackRequestError: Either view_id or external_id is required.
        """
        if isinstance(view, View):
            kwargs.update({"view": view.to_dict()})
        else:
            kwargs.update({"view": view})
        if external_id:
            kwargs.update({"external_id": external_id})
        elif view_id:
            kwargs.update({"view_id": view_id})
        else:
            raise e.SlackRequestError("Either view_id or external_id is required.")

        return await self.api_call("views.update", json=kwargs)

    async def views_publish(
        self, *, user_id: str, view: Union[dict, View], **kwargs
    ) -> AsyncSlackResponse:
        """Publish a static view for a User.
        Create or update the view that comprises an
        app's Home tab (https://api.slack.com/surfaces/tabs)
        for a specific user.
        Args:
            user_id (str): id of the user you want publish a view to.
                e.g. 'U0BPQUNTA'
            view (dict or View): The view payload.
        """
        kwargs.update({"user_id": user_id})
        if isinstance(view, View):
            kwargs.update({"view": view.to_dict()})
        else:
            kwargs.update({"view": view})
        return await self.api_call("views.publish", json=kwargs)

    async def workflows_stepCompleted(
        self, *, workflow_step_execute_id: str, outputs: dict = None, **kwargs
    ) -> AsyncSlackResponse:
        """Indicate a successful outcome of a workflow step's execution.
        Args:
            workflow_step_execute_id (str): A unique identifier of the workflow step to be updated.
                e.g. 'add_task'
            outputs (dict): A key-value object of outputs from your step.
                e.g. { 'task_name': 'Task Name' }
        """
        kwargs.update({"workflow_step_execute_id": workflow_step_execute_id})
        if outputs:
            kwargs.update({"outputs": outputs})

        return await self.api_call("workflows.stepCompleted", json=kwargs)

    async def workflows_stepFailed(
        self, *, workflow_step_execute_id: str, error: dict, **kwargs
    ) -> AsyncSlackResponse:
        """Indicate an unsuccessful outcome of a workflow step's execution.
        Args:
            workflow_step_execute_id (str): A unique identifier of the workflow step to be updated.
                e.g. 'add_task'
            error (dict): A dict with a message property that contains a human readable error message
                e.g. { message: 'Step failed to execute.' }
        """
        kwargs.update(
            {"workflow_step_execute_id": workflow_step_execute_id, "error": error}
        )
        return await self.api_call("workflows.stepFailed", json=kwargs)

    async def workflows_updateStep(
        self,
        *,
        workflow_step_edit_id: str,
        inputs: dict = None,
        outputs: list = None,
        **kwargs
    ) -> AsyncSlackResponse:
        """Update the configuration for a workflow extension step.
        Args:
            workflow_step_edit_id (str): A unique identifier of the workflow step to be updated.
                e.g. 'add_task'
            inputs (dict): A key-value object of inputs required from a user during step configuration.
                e.g. { 'title': { 'value': 'The Title' }, 'submitter': { 'value': 'The Submitter' } }
            outputs (list): A list of output objects used during step execution.
                e.g. [{ 'type': 'text', 'name': 'title', 'label': 'Title' }]
        """
        kwargs.update({"workflow_step_edit_id": workflow_step_edit_id})

        if inputs:
            kwargs.update({"inputs": inputs})
        if outputs:
            kwargs.update({"outputs": outputs})

        return await self.api_call("workflows.updateStep", json=kwargs)