Slack MCP Server

from fastmcp.server import FastMCP from typing import Dict, List, Optional, Any from pydantic import BaseModel, Field, validator # Define Pydantic models for request/response schema class ChannelListRequest(BaseModel): exclude_archived: bool = Field(default=True, description="Whether to exclude archived channels") cursor: Optional[str] = Field(default=None, description="Pagination cursor for getting additional channels") limit: int = Field(default=50, ge=1, le=200, description="Maximum number of channels to return") class ChannelInfoRequest(BaseModel): channel_id: Optional[str] = Field(default=None, description="ID of the channel to get info for") channel_name: Optional[str] = Field(default=None, description="Name of the channel to get info for (alternative to channel_id)") @validator('channel_id', 'channel_name') def validate_channel_identifier(cls, v, values): # Ensure that at least one of channel_id or channel_name is provided if 'channel_id' in values and values['channel_id'] is None and 'channel_name' in values and values['channel_name'] is None: raise ValueError("Either channel_id or channel_name must be provided") return v class MessageRequest(BaseModel): channel_name: str = Field(description="Name of the channel to get messages from (without # prefix)") limit: int = Field(default=50, ge=1, le=200, description="Maximum number of messages to return") cursor: Optional[str] = Field(default=None, description="Pagination cursor for getting additional messages") class ThreadListRequest(BaseModel): channel_name: str = Field(description="Name of the channel to get threads from (without # prefix)") limit: int = Field(default=50, ge=1, le=200, description="Maximum number of threads to return") cursor: Optional[str] = Field(default=None, description="Pagination cursor for getting additional threads") class ThreadMessagesRequest(BaseModel): thread_ts: str = Field(description="Timestamp of the thread parent message") channel_name: str = Field(description="Name of the channel containing the thread (without # prefix)") limit: int = Field(default=50, ge=1, le=200, description="Maximum number of messages to return") cursor: Optional[str] = Field(default=None, description="Pagination cursor for getting additional thread messages") class UserListRequest(BaseModel): cursor: Optional[str] = Field(default=None, description="Pagination cursor for getting additional users") limit: int = Field(default=50, ge=1, le=200, description="Maximum number of users to return") class UserInfoRequest(BaseModel): user_id: str = Field(description="ID of the user to get info for") def setup_endpoints(app: FastMCP, slack_client): """Setup Slack MCP endpoints on the provided router.""" @app.tool("list_channels") def list_channels(request: ChannelListRequest): """ List channels in the workspace with pagination support. This endpoint supports pagination through the cursor parameter. If the result set is large, the response will include a next_cursor which should be passed in subsequent requests to get the next page. Args: exclude_archived: Whether to exclude archived channels limit: Maximum number of channels to return per page (default 50, max 200) cursor: Pagination cursor for getting additional channels Returns: Dictionary with channels array and pagination info (has_more and next_cursor) { "channels": [...], # Array of channel objects "has_more": true, # Whether more channels exist "next_cursor": "dXNlcjpVMDYz" # Cursor to pass for next page } """ try: result = slack_client.get_channels( exclude_archived=request.exclude_archived, limit=request.limit, cursor=request.cursor ) return result except Exception as e: return {"error": str(e)} @app.tool("get_channel_info") def get_channel_info(request: ChannelInfoRequest): """ Get detailed information about a specific channel. The channel can be specified either by its ID or by its name (without the # prefix). Args: channel_id: ID of the channel to get info for channel_name: Name of the channel to get info for (alternative to channel_id) Returns: Channel object with detailed information """ try: channel = slack_client.get_channel_info( channel_id=request.channel_id, channel_name=request.channel_name ) return channel except Exception as e: return {"error": str(e)} @app.tool("get_messages") def get_messages(request: MessageRequest): """ Get messages from a channel with pagination support. This endpoint supports pagination through the cursor parameter. If the result set is large, the response will include a next_cursor which should be passed in subsequent requests to get the next page. Args: channel_name: Name of the channel to get messages from (without # prefix) limit: Maximum number of messages to return per page (default 50, max 200) cursor: Pagination cursor for getting additional messages Returns: Dictionary containing messages array and pagination info { "messages": [...], # Array of message objects "has_more": true, # Whether more messages exist "next_cursor": "dXNlcjpVMDYz" # Cursor to pass for next page } """ try: messages = slack_client.get_messages( channel_name=request.channel_name, limit=request.limit, cursor=request.cursor ) return messages except Exception as e: return {"error": str(e)} @app.tool("list_users") def list_users(request: UserListRequest): """ List users in the workspace with pagination support. This endpoint supports pagination through the cursor parameter. If the result set is large, the response will include a next_cursor which should be passed in subsequent requests to get the next page. Args: limit: Maximum number of users to return per page (default 50, max 200) cursor: Pagination cursor for getting additional users Returns: Dictionary with members array and pagination info (has_more and next_cursor) { "members": [...], # Array of user objects "has_more": true, # Whether more users exist "next_cursor": "dXNlcjpVMDYz" # Cursor to pass for next page } """ try: result = slack_client.get_users( limit=request.limit, cursor=request.cursor ) return result except Exception as e: return {"error": str(e)} @app.tool("get_user_info") def get_user_info(request: UserInfoRequest): """ Get detailed information about a specific user. Returns: User object with detailed information """ try: user = slack_client.get_user_info(user_id=request.user_id) return user except Exception as e: return {"error": str(e)} @app.tool("list_threads") def list_threads(request: ThreadListRequest): """ List threads in a channel with pagination support. Returns only the first message of each thread (thread parent message) in the specified channel. This endpoint supports pagination through the cursor parameter. If the result set is large, the response will include a next_cursor which should be passed in subsequent requests to get the next page. Args: channel_name: Name of the channel to get threads from (without # prefix) limit: Maximum number of threads to return per page (default 50, max 200) cursor: Pagination cursor for getting additional threads Returns: Dictionary containing threads array and pagination info { "threads": [...], # Array of thread parent messages "has_more": true, # Whether more threads exist "next_cursor": "dXNlcjpVMDYz" # Cursor to pass for next page } """ try: threads = slack_client.list_threads( channel_name=request.channel_name, limit=request.limit, cursor=request.cursor ) return threads except Exception as e: return {"error": str(e)} @app.tool("get_thread_messages") def get_thread_messages(request: ThreadMessagesRequest): """ Get all messages in a thread with pagination support. Returns all messages belonging to the specified thread. This endpoint supports pagination through the cursor parameter. If the result set is large, the response will include a next_cursor which should be passed in subsequent requests to get the next page. 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 per page (default 50, max 200) cursor: Pagination cursor for getting additional thread messages Returns: Dictionary containing thread messages array and pagination info { "messages": [...], # Array of thread messages "has_more": true, # Whether more thread messages exist "next_cursor": "dXNlcjpVMDYz" # Cursor to pass for next page } """ try: messages = slack_client.get_thread_messages( thread_ts=request.thread_ts, channel_name=request.channel_name, limit=request.limit, cursor=request.cursor ) return messages except Exception as e: return {"error": str(e)}