Source code for camel.toolkits.slack_toolkit

# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========

from __future__ import annotations

import json
import logging
import os
from typing import TYPE_CHECKING, List, Optional

from camel.toolkits.base import BaseToolkit

if TYPE_CHECKING:
    from ssl import SSLContext

    from slack_sdk import WebClient

from camel.toolkits import FunctionTool

logger = logging.getLogger(__name__)




class SlackToolkit(BaseToolkit):
    r"""A class representing a toolkit for Slack operations.

    This class provides methods for Slack operations such as creating a new
    channel, joining an existing channel, leaving a channel.
    """

    def _login_slack(
        self,
        slack_token: Optional[str] = None,
        ssl: Optional[SSLContext] = None,
    ) -> WebClient:
        r"""Authenticate using the Slack API.

        Args:
            slack_token (str, optional): The Slack API token.
                If not provided, it attempts to retrieve the token from
                the environment variable SLACK_BOT_TOKEN or SLACK_USER_TOKEN.
            ssl (SSLContext, optional): SSL context for secure connections.
                Defaults to `None`.

        Returns:
            WebClient: A WebClient object for interacting with Slack API.

        Raises:
            ImportError: If slack_sdk package is not installed.
            KeyError: If SLACK_BOT_TOKEN or SLACK_USER_TOKEN
                environment variables are not set.
        """
        try:
            from slack_sdk import WebClient
        except ImportError as e:
            raise ImportError(
                "Cannot import slack_sdk. Please install the package with \
                `pip install slack_sdk`."
            ) from e
        if not slack_token:
            slack_token = os.environ.get("SLACK_BOT_TOKEN") or os.environ.get(
                "SLACK_USER_TOKEN"
            )
            if not slack_token:
                raise KeyError(
                    "SLACK_BOT_TOKEN or SLACK_USER_TOKEN environment "
                    "variable not set."
                )

        client = WebClient(token=slack_token, ssl=ssl)
        logger.info("Slack login successful.")
        return client



    def create_slack_channel(
        self, name: str, is_private: Optional[bool] = True
    ) -> str:
        r"""Creates a new slack channel, either public or private.

        Args:
            name (str): Name of the public or private channel to create.
            is_private (bool, optional): Whether to create a private channel
                instead of a public one. Defaults to `True`.

        Returns:
            str: JSON string containing information about Slack
                channel created.

        Raises:
            SlackApiError: If there is an error during get slack channel
                information.
        """
        from slack_sdk.errors import SlackApiError

        try:
            slack_client = self._login_slack()
            response = slack_client.conversations_create(
                name=name, is_private=is_private
            )
            channel_id = response["channel"]["id"]
            response = slack_client.conversations_archive(channel=channel_id)
            return str(response)
        except SlackApiError as e:
            return f"Error creating conversation: {e.response['error']}"




    def join_slack_channel(self, channel_id: str) -> str:
        r"""Joins an existing Slack channel.

        Args:
            channel_id (str): The ID of the Slack channel to join.

        Returns:
            str: A confirmation message indicating whether join successfully
                or an error message.

        Raises:
            SlackApiError: If there is an error during get slack channel
                information.
        """
        from slack_sdk.errors import SlackApiError

        try:
            slack_client = self._login_slack()
            response = slack_client.conversations_join(channel=channel_id)
            return str(response)
        except SlackApiError as e:
            return f"Error creating conversation: {e.response['error']}"




    def leave_slack_channel(self, channel_id: str) -> str:
        r"""Leaves an existing Slack channel.

        Args:
            channel_id (str): The ID of the Slack channel to leave.

        Returns:
            str: A confirmation message indicating whether leave successfully
                or an error message.

        Raises:
            SlackApiError: If there is an error during get slack channel
                information.
        """
        from slack_sdk.errors import SlackApiError

        try:
            slack_client = self._login_slack()
            response = slack_client.conversations_leave(channel=channel_id)
            return str(response)
        except SlackApiError as e:
            return f"Error creating conversation: {e.response['error']}"




    def get_slack_channel_information(self) -> str:
        r"""Retrieve Slack channels and return relevant information in JSON
            format.

        Returns:
            str: JSON string containing information about Slack channels.

        Raises:
            SlackApiError: If there is an error during get slack channel
                information.
        """
        from slack_sdk.errors import SlackApiError

        try:
            slack_client = self._login_slack()
            response = slack_client.conversations_list()
            conversations = response["channels"]
            # Filtering conversations and extracting required information
            filtered_result = [
                {
                    key: conversation[key]
                    for key in ("id", "name", "created", "num_members")
                }
                for conversation in conversations
                if all(
                    key in conversation
                    for key in ("id", "name", "created", "num_members")
                )
            ]
            return json.dumps(filtered_result, ensure_ascii=False)
        except SlackApiError as e:
            return f"Error creating conversation: {e.response['error']}"




    def get_slack_channel_message(self, channel_id: str) -> str:
        r"""Retrieve messages from a Slack channel.

        Args:
            channel_id (str): The ID of the Slack channel to retrieve messages
                from.

        Returns:
            str: JSON string containing filtered message data.

        Raises:
            SlackApiError: If there is an error during get
                slack channel message.
        """
        from slack_sdk.errors import SlackApiError

        try:
            slack_client = self._login_slack()
            result = slack_client.conversations_history(channel=channel_id)
            messages = result["messages"]
            filtered_messages = [
                {key: message[key] for key in ("user", "text", "ts")}
                for message in messages
                if all(key in message for key in ("user", "text", "ts"))
            ]
            return json.dumps(filtered_messages, ensure_ascii=False)
        except SlackApiError as e:
            return f"Error retrieving messages: {e.response['error']}"




    def send_slack_message(
        self,
        message: str,
        channel_id: str,
        user: Optional[str] = None,
    ) -> str:
        r"""Send a message to a Slack channel.

        Args:
            message (str): The message to send.
            channel_id (str): The ID of the Slack channel to send message.
            user (Optional[str]): The user ID of the recipient.
                Defaults to `None`.

        Returns:
            str: A confirmation message indicating whether the message was sent
                successfully or an error message.

        Raises:
            SlackApiError: If an error occurs while sending the message.
        """
        from slack_sdk.errors import SlackApiError

        try:
            slack_client = self._login_slack()
            if user:
                response = slack_client.chat_postEphemeral(
                    channel=channel_id, text=message, user=user
                )
            else:
                response = slack_client.chat_postMessage(
                    channel=channel_id, text=message
                )
            return str(response)
        except SlackApiError as e:
            return f"Error creating conversation: {e.response['error']}"




    def delete_slack_message(
        self,
        time_stamp: str,
        channel_id: str,
    ) -> str:
        r"""Delete a message to a Slack channel.

        Args:
            time_stamp (str): Timestamp of the message to be deleted.
            channel_id (str): The ID of the Slack channel to delete message.

        Returns:
            str: A confirmation message indicating whether the message
                was delete successfully or an error message.

        Raises:
            SlackApiError: If an error occurs while sending the message.
        """
        from slack_sdk.errors import SlackApiError

        try:
            slack_client = self._login_slack()
            response = slack_client.chat_delete(
                channel=channel_id, ts=time_stamp
            )
            return str(response)
        except SlackApiError as e:
            return f"Error creating conversation: {e.response['error']}"




    def get_tools(self) -> List[FunctionTool]:
        r"""Returns a list of FunctionTool objects representing the
        functions in the toolkit.

        Returns:
            List[FunctionTool]: A list of FunctionTool objects
                representing the functions in the toolkit.
        """
        return [
            FunctionTool(self.create_slack_channel),
            FunctionTool(self.join_slack_channel),
            FunctionTool(self.leave_slack_channel),
            FunctionTool(self.get_slack_channel_information),
            FunctionTool(self.get_slack_channel_message),
            FunctionTool(self.send_slack_message),
            FunctionTool(self.delete_slack_message),
        ]