Skip to content

SlackRetrieveTool

autogen.tools.experimental.SlackRetrieveTool #

SlackRetrieveTool(*, bot_token, channel_id)

Bases: Tool

Retrieves messages from a Slack channel.

Initialize the SlackRetrieveTool.

PARAMETER DESCRIPTION
bot_token

Bot User OAuth Token starting with "xoxb-".

TYPE: str

channel_id

Channel ID where messages will be sent.

TYPE: str

Source code in autogen/tools/experimental/messageplatform/slack/slack.py
def __init__(self, *, bot_token: str, channel_id: str) -> None:
    """
    Initialize the SlackRetrieveTool.

    Args:
        bot_token: Bot User OAuth Token starting with "xoxb-".
        channel_id: Channel ID where messages will be sent.
    """

    async def slack_retrieve_messages(
        bot_token: Annotated[str, Depends(on(bot_token))],
        channel_id: Annotated[str, Depends(on(channel_id))],
        messages_since: Annotated[
            Union[str, None],
            "Date to retrieve messages from (ISO format) OR Slack message ID. If None, retrieves latest messages.",
        ] = None,
        maximum_messages: Annotated[
            Union[int, None], "Maximum number of messages to retrieve. If None, retrieves all messages since date."
        ] = None,
    ) -> Any:
        """
        Retrieves messages from a Discord channel.

        Args:
            bot_token: The bot token to use for Discord. (uses dependency injection)
            channel_id: The ID of the channel. (uses dependency injection)
            messages_since: ISO format date string OR Slack message ID, to retrieve messages from. If None, retrieves latest messages.
            maximum_messages: Maximum number of messages to retrieve. If None, retrieves all messages since date.
        """
        try:
            web_client = WebClient(token=bot_token)

            # Convert ISO datetime to Unix timestamp if needed
            oldest = None
            if messages_since:
                if "." in messages_since:  # Likely a Slack message ID
                    oldest = messages_since
                else:  # Assume ISO format
                    try:
                        dt = datetime.fromisoformat(messages_since.replace("Z", "+00:00"))
                        oldest = str(dt.timestamp())
                    except ValueError as e:
                        return f"Invalid date format. Please provide either a Slack message ID or ISO format date (e.g., '2025-01-25T00:00:00Z'). Error: {e}"

            messages = []
            cursor = None

            while True:
                try:
                    # Prepare API call parameters
                    params = {
                        "channel": channel_id,
                        "limit": min(1000, maximum_messages) if maximum_messages else 1000,
                    }
                    if oldest:
                        params["oldest"] = oldest
                    if cursor:
                        params["cursor"] = cursor

                    # Make API call
                    response = web_client.conversations_history(**params)  # type: ignore[arg-type]

                    if not response["ok"]:
                        return f"Message retrieval failed, Slack response error: {response['error']}"

                    # Add messages to our list
                    messages.extend(response["messages"])

                    # Check if we've hit our maximum
                    if maximum_messages and len(messages) >= maximum_messages:
                        messages = messages[:maximum_messages]
                        break

                    # Check if there are more messages
                    if not response["has_more"]:
                        break

                    cursor = response["response_metadata"]["next_cursor"]

                except SlackApiError as e:
                    return f"Message retrieval failed on pagination, Slack API error: {e.response['error']}"

            return {
                "message_count": len(messages),
                "messages": messages,
                "start_time": oldest or "latest",
            }

        except SlackApiError as e:
            return f"Message retrieval failed, Slack API exception: {e.response['error']} (See https://api.slack.com/automation/cli/errors#{e.response['error']})"
        except Exception as e:
            return f"Message retrieval failed, exception: {e}"

    super().__init__(
        name="slack_retrieve",
        description="Retrieves messages from a Slack channel based datetime/message ID and/or number of latest messages.",
        func_or_tool=slack_retrieve_messages,
    )

name property #

name

description property #

description

func property #

func

tool_schema property #

tool_schema

Get the schema for the tool.

This is the preferred way of handling function calls with OpeaAI and compatible frameworks.

function_schema property #

function_schema

Get the schema for the function.

This is the old way of handling function calls with OpenAI and compatible frameworks. It is provided for backward compatibility.

realtime_tool_schema property #

realtime_tool_schema

Get the schema for the tool.

This is the preferred way of handling function calls with OpeaAI and compatible frameworks.

register_for_llm #

register_for_llm(agent)

Registers the tool for use with a ConversableAgent's language model (LLM).

This method registers the tool so that it can be invoked by the agent during interactions with the language model.

PARAMETER DESCRIPTION
agent

The agent to which the tool will be registered.

TYPE: ConversableAgent

Source code in autogen/tools/tool.py
def register_for_llm(self, agent: "ConversableAgent") -> None:
    """Registers the tool for use with a ConversableAgent's language model (LLM).

    This method registers the tool so that it can be invoked by the agent during
    interactions with the language model.

    Args:
        agent (ConversableAgent): The agent to which the tool will be registered.
    """
    agent.register_for_llm()(self)

register_for_execution #

register_for_execution(agent)

Registers the tool for direct execution by a ConversableAgent.

This method registers the tool so that it can be executed by the agent, typically outside of the context of an LLM interaction.

PARAMETER DESCRIPTION
agent

The agent to which the tool will be registered.

TYPE: ConversableAgent

Source code in autogen/tools/tool.py
def register_for_execution(self, agent: "ConversableAgent") -> None:
    """Registers the tool for direct execution by a ConversableAgent.

    This method registers the tool so that it can be executed by the agent,
    typically outside of the context of an LLM interaction.

    Args:
        agent (ConversableAgent): The agent to which the tool will be registered.
    """
    agent.register_for_execution()(self)

register_tool #

register_tool(agent)

Register a tool to be both proposed and executed by an agent.

Equivalent to calling both register_for_llm and register_for_execution with the same agent.

Note: This will not make the agent recommend and execute the call in the one step. If the agent recommends the tool, it will need to be the next agent to speak in order to execute the tool.

PARAMETER DESCRIPTION
agent

The agent to which the tool will be registered.

TYPE: ConversableAgent

Source code in autogen/tools/tool.py
def register_tool(self, agent: "ConversableAgent") -> None:
    """Register a tool to be both proposed and executed by an agent.

    Equivalent to calling both `register_for_llm` and `register_for_execution` with the same agent.

    Note: This will not make the agent recommend and execute the call in the one step. If the agent
    recommends the tool, it will need to be the next agent to speak in order to execute the tool.

    Args:
        agent (ConversableAgent): The agent to which the tool will be registered.
    """
    self.register_for_llm(agent)
    self.register_for_execution(agent)