slack_client.py•9.31 kB
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
from typing import Dict, List, Optional, Any
import logging
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SlackClient:
"""Client for interacting with Slack API with read-only capabilities."""
def __init__(self, token: str):
"""Initialize the Slack client with the provided token."""
self.client = WebClient(token=token)
def get_channels(self, exclude_archived: bool = True, limit: int = 100, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get a list of all channels in the workspace.
Args:
exclude_archived: Whether to exclude archived channels
limit: Maximum number of channels to return (default 100)
cursor: Pagination cursor for getting additional channels
Returns:
Dictionary containing channels and pagination info
"""
try:
response = self.client.conversations_list(
exclude_archived=exclude_archived,
limit=limit,
cursor=cursor
)
next_cursor = response.get("response_metadata", {}).get("next_cursor", "")
return {
"channels": response["channels"],
"has_more": bool(next_cursor), # True if next_cursor exists and is not empty
"next_cursor": next_cursor
}
except SlackApiError as e:
logger.error(f"Error getting channels: {e}")
raise
def find_channel_by_name(self, channel_name: str) -> Optional[str]:
"""
Find a channel ID by its name.
Args:
channel_name: Name of the channel (without #)
Returns:
Channel ID if found, None otherwise
"""
try:
# Fetch all channels
channels = self.get_channels(limit=1000)["channels"]
# Look for the channel by name
for channel in channels:
if channel["name"] == channel_name:
return channel["id"]
# Channel not found
logger.error(f"Channel with name '{channel_name}' not found")
return None
except SlackApiError as e:
logger.error(f"Error finding channel by name: {e}")
raise
def get_channel_info(self, channel_id: Optional[str] = None, channel_name: Optional[str] = None) -> Dict[str, Any]:
"""
Get information about a specific channel.
Args:
channel_id: ID of the channel
channel_name: Name of the channel (alternative to channel_id)
Returns:
Channel information
"""
try:
# If channel_name is provided but not channel_id, look up the ID
if channel_id is None and channel_name is not None:
channel_id = self.find_channel_by_name(channel_name)
if not channel_id:
raise ValueError(f"Channel with name '{channel_name}' not found")
# Get channel info by ID
response = self.client.conversations_info(channel=channel_id)
return response["channel"]
except SlackApiError as e:
logger.error(f"Error getting channel info: {e}")
raise
def get_messages(self, channel_name: str, limit: int = 100, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get messages from a channel.
Args:
channel_name: Name of the channel (without # prefix)
limit: Maximum number of messages to return (default 100)
cursor: Pagination cursor for getting additional messages
Returns:
Dictionary containing messages and pagination info
"""
try:
# Find channel ID by name
channel_id = self.find_channel_by_name(channel_name)
if not channel_id:
raise ValueError(f"Channel with name '{channel_name}' not found")
response = self.client.conversations_history(
channel=channel_id,
limit=limit,
cursor=cursor
)
return {
"messages": response["messages"],
"has_more": response.get("has_more", False),
"next_cursor": response.get("response_metadata", {}).get("next_cursor")
}
except SlackApiError as e:
logger.error(f"Error getting messages: {e}")
raise
def get_users(self, limit: int = 100, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get a list of all users in the workspace.
Args:
limit: Maximum number of users to return (default 100)
cursor: Pagination cursor for getting additional users
Returns:
Dictionary containing users and pagination info
"""
try:
response = self.client.users_list(
limit=limit,
cursor=cursor
)
return {
"members": response["members"],
"has_more": response.get("has_more", False),
"next_cursor": response.get("response_metadata", {}).get("next_cursor")
}
except SlackApiError as e:
logger.error(f"Error getting users: {e}")
raise
def get_user_info(self, user_id: str) -> Dict[str, Any]:
"""
Get information about a specific user.
Args:
user_id: ID of the user
Returns:
User information
"""
try:
response = self.client.users_info(user=user_id)
return response["user"]
except SlackApiError as e:
logger.error(f"Error getting user info: {e}")
raise
def list_threads(self, channel_name: str, limit: int = 100, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
List threads in a channel with pagination support.
Returns only the first message of each thread (thread parent message) in the specified channel.
Args:
channel_name: Name of the channel to get threads from (without # prefix)
limit: Maximum number of threads to return (default 100)
cursor: Pagination cursor for getting additional threads
Returns:
Dictionary containing threads array and pagination info
"""
try:
# Find channel ID by name
channel_id = self.find_channel_by_name(channel_name)
if not channel_id:
raise ValueError(f"Channel with name '{channel_name}' not found")
# Get messages from the channel that have thread replies
response = self.client.conversations_history(
channel=channel_id,
limit=limit,
cursor=cursor
)
# Filter out messages that don't have replies (not thread parents)
threads = [msg for msg in response["messages"] if msg.get("reply_count", 0) > 0]
return {
"threads": threads,
"has_more": response.get("has_more", False),
"next_cursor": response.get("response_metadata", {}).get("next_cursor")
}
except SlackApiError as e:
logger.error(f"Error listing threads: {e}")
raise
def get_thread_messages(self, thread_ts: str, channel_name: str, limit: int = 100, cursor: Optional[str] = None) -> Dict[str, Any]:
"""
Get all messages in a thread with pagination support.
Args:
thread_ts: Timestamp of the thread parent message
channel_name: Name of the channel containing the thread (without # prefix)
limit: Maximum number of messages to return (default 100)
cursor: Pagination cursor for getting additional thread messages
Returns:
Dictionary containing thread messages array and pagination info
"""
try:
# Find channel ID by name
channel_id = self.find_channel_by_name(channel_name)
if not channel_id:
raise ValueError(f"Channel with name '{channel_name}' not found")
# Get replies to the thread
response = self.client.conversations_replies(
channel=channel_id,
ts=thread_ts,
limit=limit,
cursor=cursor
)
return {
"messages": response["messages"],
"has_more": response.get("has_more", False),
"next_cursor": response.get("response_metadata", {}).get("next_cursor")
}
except SlackApiError as e:
logger.error(f"Error getting thread messages: {e}")
raise