Python Slack client + MCP server — fetch messages, threads, normalize markup

slack-mcp-client.md

Slack MCP Client (Quickstart + Auto Channel Discovery)

Copy/paste implementation for a Slack MCP server where channel discovery is the default path.

If no channel IDs are provided, it will:

1. list all channels the bot can see,
2. keep only channels the bot is a member of,
3. pull messages from those channels.

This is the intended onboarding behavior: no manual channel IDs needed.

---

1) One-minute setup

1. Sign in at https://api.slack.com/apps
2. Open the shared app: https://api.slack.com/apps/A097CS08VT5
3. In OAuth & Permissions, ensure bot token scopes include:
   • channels:read
   • groups:read
   • channels:history
   • groups:history
4. Reinstall app to workspace after scope changes
5. Copy bot token (xoxb-...) and set:

export SLACK_BOT_TOKEN='xoxb-your-token'

Install deps:

pip install slack-sdk mcp

---

2) slack_client.py

"""Slack client: fetch messages, normalize text, handle retries."""

from __future__ import annotations

import os
import random
import re
import time
from datetime import datetime, timezone
from html import unescape
from typing import Any, Dict, Iterable, List, Optional

from slack_sdk.errors import SlackApiError
from slack_sdk.web import WebClient

MAX_RETRIES = 5
INITIAL_BACKOFF_SECONDS = 1.0
MAX_BACKOFF_SECONDS = 30.0
DEFAULT_SINCE_HOURS = 24


class SlackClient:
    """Fetch Slack messages with thread support and automatic retries."""

    _link_pat = re.compile(r"<(http[^>|]+)(?:\|([^>]+))?>")
    _mail_pat = re.compile(r"<mailto:([^>|]+)(?:\|[^>]+)?>")
    _chan_pat = re.compile(r"<#C[^>|]+?\|([^>]+)>")
    _user_pat = re.compile(r"<@([UW][A-Z0-9]+)>")

    def __init__(self, token: Optional[str] = None):
        token = token or os.environ.get("SLACK_BOT_TOKEN")
        if not token:
            raise EnvironmentError("SLACK_BOT_TOKEN not set")
        self.client = WebClient(token=token)

    def test_connection(self) -> Dict[str, Any]:
        """Fail-fast auth check to catch bad/stale tokens early."""
        response = self.client.auth_test()
        return {
            "ok": response.get("ok", False),
            "team": response.get("team"),
            "team_id": response.get("team_id"),
            "user": response.get("user"),
            "user_id": response.get("user_id"),
            "bot_id": response.get("bot_id"),
        }

    def normalize_slack_text(self, text: str) -> str:
        if not text:
            return ""
        text = unescape(text)

        def _repl_link(m: re.Match) -> str:
            url, label = m.group(1), m.group(2)
            return f"{label} ({url})" if label else url

        text = self._link_pat.sub(_repl_link, text)
        text = self._mail_pat.sub(r"\1", text)
        text = self._chan_pat.sub(r"#\1", text)
        text = self._user_pat.sub(r"@\1", text)
        return " ".join(text.split())

    @staticmethod
    def _sleep_with_jitter(base_seconds: float) -> None:
        time.sleep(base_seconds * random.uniform(0.5, 1.5))

    def _call_with_retries(self, func, *args, **kwargs):
        delay = INITIAL_BACKOFF_SECONDS
        last_error: Optional[Exception] = None

        for _ in range(MAX_RETRIES):
            try:
                return func(*args, **kwargs)
            except SlackApiError as exc:
                last_error = exc
                status = getattr(exc.response, "status_code", None)
                headers = getattr(exc.response, "headers", {}) or {}

                if status == 429:
                    retry_after = headers.get("Retry-After")
                    wait = float(retry_after) if retry_after else delay
                    self._sleep_with_jitter(min(wait, MAX_BACKOFF_SECONDS))
                elif status and status >= 500:
                    self._sleep_with_jitter(min(delay, MAX_BACKOFF_SECONDS))
                else:
                    break
            except Exception as exc:
                last_error = exc
                self._sleep_with_jitter(min(delay, MAX_BACKOFF_SECONDS))

            delay = min(delay * 2, MAX_BACKOFF_SECONDS)

        if last_error:
            raise last_error
        raise RuntimeError("Slack API call failed")

    def discover_channels(self) -> List[Dict[str, Any]]:
        """Live discovery of channels the bot is currently a member of."""
        channels: List[Dict[str, Any]] = []
        cursor = None

        while True:
            response = self._call_with_retries(
                self.client.conversations_list,
                types="public_channel,private_channel",
                exclude_archived=True,
                limit=200,
                cursor=cursor,
            )
            for ch in response.get("channels", []):
                if ch.get("is_member"):
                    channels.append(
                        {
                            "id": ch["id"],
                            "name": ch["name"],
                            "is_private": ch.get("is_private", False),
                        }
                    )
            cursor = (response.get("response_metadata") or {}).get("next_cursor")
            if not cursor:
                break

        return channels

    def fetch_messages(
        self, channel_id: str, oldest: float, latest: float
    ) -> List[Dict[str, Any]]:
        messages: List[Dict[str, Any]] = []
        cursor = None

        while True:
            response = self._call_with_retries(
                self.client.conversations_history,
                channel=channel_id,
                oldest=str(oldest),
                latest=str(latest),
                inclusive=True,
                limit=200,
                cursor=cursor,
            )
            messages.extend(response.get("messages", []))
            cursor = (response.get("response_metadata") or {}).get("next_cursor")
            if not cursor:
                break

        return messages

    def fetch_replies(self, channel_id: str, thread_ts: str) -> List[Dict[str, Any]]:
        replies: List[Dict[str, Any]] = []
        cursor = None

        while True:
            response = self._call_with_retries(
                self.client.conversations_replies,
                channel=channel_id,
                ts=thread_ts,
                limit=200,
                cursor=cursor,
            )
            page = response.get("messages", [])
            if replies:
                replies.extend(page)
            else:
                replies.extend(page[1:] if len(page) > 1 else [])
            cursor = (response.get("response_metadata") or {}).get("next_cursor")
            if not cursor:
                break

        return replies

    def iter_messages_with_threads(
        self, channel_id: str, oldest: float, latest: float
    ) -> Iterable[Dict[str, Any]]:
        for message in self.fetch_messages(channel_id, oldest, latest):
            yield message
            if message.get("thread_ts") == message.get("ts"):
                for reply in self.fetch_replies(channel_id, message["thread_ts"]):
                    yield reply

    def resolve_time_range(
        self, since_hours: Optional[int] = None
    ) -> tuple[float, float]:
        now = datetime.now(timezone.utc)
        hours = since_hours if since_hours is not None else DEFAULT_SINCE_HOURS
        oldest = now.timestamp() - (hours * 3600)
        return oldest, now.timestamp()

---

3) mcp_server.py (auto-discovery default)

"""MCP server exposing Slack as pull-based tools with auto discovery."""

from __future__ import annotations

import logging
from typing import Any, Dict, List, Optional

from mcp.server.fastmcp import FastMCP

from slack_client import SlackClient

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

mcp = FastMCP("slack-client")


def _startup_check(client: SlackClient) -> None:
    """Validate token and discover channels at startup for easy onboarding."""
    identity = client.test_connection()
    if not identity.get("ok"):
        raise RuntimeError("Slack auth.test failed")
    logger.info(
        "Slack connected: team=%s bot=%s user_id=%s",
        identity.get("team"),
        identity.get("user"),
        identity.get("user_id"),
    )
    discovered = client.discover_channels()
    logger.info("Startup discovery: %d channels available", len(discovered))


@mcp.tool()
def get_slack_channels(include_private: Optional[bool] = True) -> Dict[str, Any]:
    """List channels the bot can currently read."""
    client = SlackClient()
    channels = client.discover_channels()
    if not include_private:
        channels = [ch for ch in channels if not ch.get("is_private", False)]
    return {"channels": channels}


@mcp.tool()
def pull_slack_messages(
    channel_ids: Optional[List[str]] = None,
    since_hours: Optional[int] = None,
) -> Dict[str, Any]:
    """Fetch messages.

    Default behavior:
      - If channel_ids is omitted, discover channels automatically.
      - If channel_ids is provided, only fetch those channels.
    """
    client = SlackClient()

    if channel_ids:
        channels = [{"id": cid, "name": cid} for cid in channel_ids]
    else:
        channels = client.discover_channels()

    oldest, latest = client.resolve_time_range(since_hours)

    results: List[Dict[str, Any]] = []
    for ch in channels:
        count = 0
        for msg in client.iter_messages_with_threads(ch["id"], oldest, latest):
            text = client.normalize_slack_text(msg.get("text", ""))
            if not text:
                continue
            results.append(
                {
                    "channel": ch["name"],
                    "user": msg.get("user", ""),
                    "text": text,
                    "ts": msg.get("ts"),
                    "thread_ts": msg.get("thread_ts"),
                }
            )
            count += 1
        logger.info("Fetched %d messages from #%s", count, ch["name"])

    return {"messages": results, "total": len(results), "channels": len(channels)}


if __name__ == "__main__":
    client = SlackClient()
    _startup_check(client)
    mcp.run()

---

4) MCP config

{
  "mcpServers": {
    "slack-client": {
      "command": "python",
      "args": ["path/to/mcp_server.py"],
      "env": {
        "SLACK_BOT_TOKEN": "xoxb-your-token-here"
      }
    }
  }
}

---

5) Quick onboarding test

1. Start MCP server
2. Call get_slack_channels() and verify expected channels appear
3. Call pull_slack_messages() without channel_ids
4. Confirm it fetches from discovered channels

If manual channel_ids works but default discovery returns empty, the token likely lacks read/list scopes or the bot is not in those channels.

---

6) Common errors and diagnosis

• invalid_auth / token_revoked:
  • wrong or stale token
• missing_scope:
  • add required scopes in Slack app, reinstall app, retry
• not_in_channel:
  • bot is not a member of that channel (especially private channels)
• empty discovery list:
  • token is valid but bot has not been invited to channels yet

Use auth.test first, then get_slack_channels(), then pull_slack_messages().