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