mcp_slack

Instructions

Implementation Reference

• main.py:158-181 (handler)

  The handler function for the 'get_recent_slack_messages' tool. It fetches recent messages from a specified Slack channel using the Slack API, retrieves a user name mapping, formats the messages, and returns them as a string.

  @mcp.tool()
  async def get_recent_slack_messages(channel_id: str, limit: int = 5) -> str:
      """Fetch recent messages from a Slack channel."""
      params = {
          "channel": channel_id,
          "limit": limit
      }
      data = await make_slack_request("conversations.history", params)
  
      if not data or not data.get("ok"):
          return "Unable to fetch Slack messages."
  
      messages = data.get("messages", [])
      if not messages:
          return "No messages found in the channel."
  
      # 🔁 Get user ID to name map
      user_map = await get_user_name_map()
  
      # Format with name
      formatted = [
          format_slack_message(msg, user_map) for msg in messages
      ]
      return "\n---\n".join(formatted)

• main.py:45-53 (helper)

  Helper function that fetches and caches user ID to name mappings from Slack users.list API, used by get_recent_slack_messages to display user names.

  async def get_user_name_map() -> dict[str, str]:
      """Fetch and return a mapping of user_id to user name."""
      data = await make_slack_request("users.list")
      if not data or not data.get("ok"):
          return {}
      return {
          user["id"]: user.get("real_name") or user.get("name")
          for user in data["members"]
      }

• main.py:34-40 (helper)

  Helper function to format individual Slack messages with timestamps and resolved user names, used by get_recent_slack_messages.

  def format_slack_message(msg: dict, user_map: dict[str, str]) -> str:
      """Format Slack message into a readable string with user name."""
      user_id = msg.get("user", "unknown user")
      user_name = user_map.get(user_id, user_id)  # fallback to user ID if name not found
      text = msg.get("text", "")
      ts = msg.get("ts", "")
      return f"[{ts}] {user_name}: {text}"

• main.py:19-32 (helper)

  Core helper for making authenticated requests to the Slack API, used by get_recent_slack_messages and other tools.

  async def make_slack_request(method: str, params: dict[str, Any] | None = None) -> dict[str, Any] | None:
      """Make a request to the Slack Web API with proper error handling."""
      headers = {
          "Authorization": f"Bearer {SLACK_TOKEN}",
          "Content-Type": "application/x-www-form-urlencoded"
      }
      async with httpx.AsyncClient() as client:
          try:
              response = await client.post(f"{SLACK_API_BASE}/{method}", data=params, headers=headers, timeout=10.0)
              response.raise_for_status()
              return response.json()
          except Exception as e:
              print(f"Slack API error: {e}")
              return None