""" title: Slack Thread Reader author: peter.lyoo author_url: https://github.com/peterlyoo version: 0.1.0 """ import re import requests from pydantic import BaseModel, Field from typing import List class Tools: """ A utility class for interacting with Slack, specifically for parsing Slack message links and fetching messages in a thread. This class provides methods to extract channel and message details from a Slack link, interact with the Slack API, and filter messages. Need slack permissions for channels:history, groups:history, im:history, mpim:history and the Bot should be integrated in the private channels you want to fetch. """ class Valves(BaseModel): SLACK_BOT_USER_TOKEN: str = Field( default="xoxb-", description="Slack Bot User Token used for authentication with the Slack API.", ) def __init__(self): self.citation = True self.valves = self.Valves() def _process_slack_link(self, slack_link: str) -> List[str]: """ Processes a Slack link to fetch and return all messages in the thread. :param slack_link: The Slack URL containing the channel ID and message timestamp. :type slack_link: str :return: Messages from the Slack thread, formatted as 'user: message'. :rtype: list[str] """ # Extract channel ID and timestamp from the Slack link match = re.search(r"archives/([^/]+)/p(\d+)", slack_link) if not match: return [] channel_id = match.group(1) message_ts = f"{match.group(2)[:10]}.{match.group(2)[10:]}" url = "https://slack.com/api/conversations.replies" headers = { "Authorization": f"Bearer {self.valves.SLACK_BOT_USER_TOKEN}", "Content-Type": "application/json", } params = {"channel": channel_id, "ts": message_ts} try: # Fetch messages from Slack API response = requests.get(url, headers=headers, params=params) if response.status_code != 200: print(f"HTTP error: {response.status_code}, {response.text}") return [] data = response.json() if not data.get("ok"): print(f"Slack API error: {data.get('error')}") return [] # Filter and format messages messages = data.get("messages", []) results = [] for msg in messages: text = msg.get("text", "") user = msg.get("user", "unknown") results.append(f"{user}: {text}") return "\n".join(results) except requests.RequestException as e: print(f"Request failed: {e}") return []