はじめに

第3章では、Slackの会話データを活用して会話分析エージェントを構築しました。メッセージ頻度やエンゲージメントを分析することで、AIがチームのコミュニケーションを評価し、プロセス改善を支援できるようになりました。今回は、SlackのWebhook機能を利用して、メッセージやイベントをリアルタイムで検知するリアルタイム管理AIを構築します。

この第4章では、MCPサーバーにSlackのWebhookを統合し、チャンネルへの新しいメッセージやリアクションを即座に捕捉します。AIはこれを利用して、自動応答を生成したり、重要なイベントを他のツール（例：Notion）に記録したりできます。コード例とステップごとのガイドで、リアルタイム管理AIの構築を体験しましょう。さあ、始めましょう！

リアルタイム管理AIとは？

リアルタイム管理AIは、Slackのメッセージやイベント（リアクション、チャンネル作成など）を監視し、チームの動的なコミュニケーションを管理するエージェントです。MCPサーバーとSlackのWebhookを組み合わせることで、以下のような機能を実現できます：

イベント検知：新しいメッセージ、リアクション、チャンネル変更をリアルタイムで捕捉。
自動応答：特定のキーワードやイベントに基づいてメッセージを送信。
外部連携：イベントをNotionやメールに記録し、コラボレーションを強化。

ユースケース

タスク管理：AIが「タスク追加」メッセージを検知し、Notionにタスクを登録。
チーム通知：重要なメッセージ（例：@メンション）にリアクションや通知を追加。
プロセス監視：意図しないイベント（例：公開チャンネルの誤作成）を検出。

開発環境の準備

第3章の環境を基に、以下の追加準備を行います：

Python 3.8以降、mcpライブラリ、requestsライブラリ、Claude Desktop：これまでと同じ。
python-dotenv：環境変数の管理（既にインストール済み）。
ngrok：ローカルサーバーを公開し、Webhookを受信。
Slackアプリ：Webhook対応の設定。

インストールコマンド（必要に応じて）：

pip install requests python-dotenv

Slackのセットアップ

Slackアプリの拡張：
- 第3章のアプリ（例：MCP-Bot）を使用。
- スコープを確認：channels:history、channels:read、reactions:read、chat:write。
- Event Subscriptionsを有効化：
  - Slack APIのアプリ設定で「Event Subscriptions」をオン。
  - スコープを追加：message.channels（チャンネルメッセージ）、reaction_added（リアクション）。
  - Webhook URLを後で設定（ngrokで生成）。
- アプリを再インストールし、トークンを更新。
チャンネル準備：
- テスト用チャンネル（例：#mcp-project）を使用。
- メッセージやリアクションを投稿してイベントを生成。
ngrokの設定：
- ngrokをインストール（brew install ngrokまたは公式サイト）。
- ローカルサーバーを公開：
```
ngrok http 8125
```
- 生成されたURLを記録（例：https://abc123.ngrok.io）。
- SlackアプリのEvent SubscriptionsにWebhook URLを設定（例：https://abc123.ngrok.io/webhook）。

環境変数：
第3章の.envファイルに以下を確認：

SLACK_TOKEN=your_bot_token
SLACK_CHANNEL_ID=your_channel_id

コード例：リアルタイム管理用MCPサーバー

以下のMCPサーバーは、SlackのWebhookを介してイベントを検知し、自動応答やイベント記録の機能を提供します。

from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import threading

class SlackRealtimeServer(MCPServer):
    def __init__(self, host, port, token, channel_id):
        super().__init__(host, port)
        self.token = token
        self.channel_id = channel_id
        self.base_url = "https://slack.com/api"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
        self.latest_event = None
        self.register_resource("get_latest_event", self.get_latest_event)
        self.register_tool("send_message", self.send_message)
        self.start_webhook_server()

    def send_message(self, params):
        try:
            text = params.get("text", "")
            channel = params.get("channel", self.channel_id)
            if not text:
                return {"status": "error", "message": "メッセージテキストが必要です"}
            
            url = f"{self.base_url}/chat.postMessage"
            payload = {
                "channel": channel,
                "text": text
            }
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return {"status": "success", "message_id": response.json()["ts"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_latest_event(self, params):
        try:
            if self.latest_event:
                event = self.latest_event
                event_info = {
                    "type": event.get("type", ""),
                    "user": event.get("user", "unknown"),
                    "text": event.get("text", ""),
                    "channel": event.get("channel", ""),
                    "timestamp": event.get("ts", "")
                }
                if event["type"] == "reaction_added":
                    event_info["reaction"] = event.get("reaction", "")
                return {"status": "success", "event_info": event_info}
            return {"status": "success", "event_info": None, "message": "イベントなし"}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def start_webhook_server(self):
        class WebhookHandler(BaseHTTPRequestHandler):
            def do_POST(self):
                content_length = int(self.headers["Content-Length"])
                post_data = self.rfile.read(content_length)
                data = json.loads(post_data.decode("utf-8"))
                
                # SlackのURL検証リクエストを処理
                if data.get("type") == "url_verification":
                    self.send_response(200)
                    self.send_header("Content-Type", "application/json")
                    self.end_headers()
                    self.wfile.write(json.dumps({"challenge": data["challenge"]}).encode("utf-8"))
                    return
                
                # イベントデータを処理
                event = data.get("event", {})
                if event.get("type") in ["message", "reaction_added"]:
                    self.server.parent.latest_event = event
                
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"Webhook received")

        server = HTTPServer(("localhost", 8125), WebhookHandler)
        server.parent = self
        threading.Thread(target=server.serve_forever, daemon=True).start()
        print("Webhookサーバーを起動中: http://localhost:8125")

if __name__ == "__main__":
    load_dotenv()
    server = SlackRealtimeServer(
        host="localhost",
        port=8125,
        token=os.getenv("SLACK_TOKEN"),
        channel_id=os.getenv("SLACK_CHANNEL_ID")
    )
    print("SlackリアルタイムMCPサーバーを起動中: http://localhost:8125")
    server.start()

コードの説明

send_message：第2章から再利用。チャンネルにメッセージを送信。
get_latest_event：Webhookで受信した最新イベント（メッセージまたはリアクション）を取得。
start_webhook_server：ローカルWebhookサーバーを起動し、Slackからのイベントを処理。
WebhookHandler：Slackのイベント（メッセージ、リアクション）を受信し、URL検証リクエストを処理。

前提条件

Slackアプリにmessage.channelsとreaction_addedイベントが設定済み。
ngrokでWebhook URLが公開され、Slackアプリに登録済み。
.envファイルに正しいSLACK_TOKENとSLACK_CHANNEL_IDが設定済み。
チャンネルにボットが招待済み。

サーバーのテスト

サーバーが正しく動作するか確認します：

ngrok起動：
```
ngrok http 8125
```
ngrok URL（例：https://abc123.ngrok.io）を記録し、SlackアプリのEvent Subscriptionsに設定（例：https://abc123.ngrok.io/webhook）。
サーバー起動：
```
python slack_realtime_server.py
```
コンソールに「SlackリアルタイムMCPサーバーを起動中: http://localhost:8125」と「Webhookサーバーを起動中: http://localhost:8125」が表示。
イベント検知のテスト：
- SlackのUIでチャンネル（例：#mcp-project）にメッセージを投稿（例：「タスク確認」）またはリアクション（例：👍）を追加。
- サーバーがイベントをWebhook経由で検知。

最新イベント取得のテスト：
Pythonでリクエストを送信：

import requests
import json

url = "http://localhost:8125"
payload = {
    "jsonrpc": "2.0",
    "method": "get_latest_event",
    "params": {},
    "id": 1
}
response = requests.post(url, json=payload)
print(json.dumps(response.json(), indent=2, ensure_ascii=False))

期待されるレスポンス：

{
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "event_info": {
      "type": "message",
      "user": "U123456",
      "text": "タスク確認",
      "channel": "C123456",
      "timestamp": "1617187200.000100"
    }
  },
  "id": 1
}

自動応答のテスト：

payload = {
    "jsonrpc": "2.0",
    "method": "send_message",
    "params": {
        "text": "メッセージを受信しました！",
        "channel": os.getenv("SLACK_CHANNEL_ID")
    },
    "id": 2
}
response = requests.post(url, json=payload)
print(json.dumps(response.json(), indent=2, ensure_ascii=False))

Claude Desktopとの接続

サーバーをClaude Desktopに接続します：

設定ファイルの編集：
Claude Desktopの設定ファイル（例：claude_desktop_config.json）に以下を追加：

{
  "mcp_servers": [
    {
      "name": "SlackRealtimeServer",
      "url": "http://localhost:8125",
      "auth": "none"
    }
  ]
}

Claudeでテスト：
Claude Desktopを起動し、プロンプトを入力：

最新のチャンネルイベントを教えてください。

レスポンス例：

最新のチャンネルイベント：
- タイプ：メッセージ
- ユーザー：U123456
- 内容：タスク確認
- チャンネル：#mcp-project
- タイムスタンプ：2025-04-22 10:00

別のプロンプト：

チャンネルに「メッセージを受信しました！」を送信してください。

レスポンス例：

チャンネル #mcp-project に「メッセージを受信しました！」を送信しました。

実装のコツと注意点

Webhook検証：SlackのURL検証リクエスト（url_verification）を正しく処理。
レートリミティング：Slack APIの制限（例：chat.postMessageは1秒に1リクエスト）に注意。
セキュリティ：本番環境では、Webhookリクエストの署名検証（SlackのX-Slack-Signature）を導入。
テスト：テスト用チャンネルを作成し、本番データに影響を与えない。
拡張性：大量のイベントを処理する場合、キュー（例：Redis、RabbitMQ）を検討。

試してみよう：挑戦課題

以下の機能を追加して、エージェントを強化してみてください：

特定のキーワード（例：「タスク」）を含むメッセージに自動応答する機能。
リアクションイベントを検知し、特定のリアクション（例：🚀）にメッセージで反応。
イベントデータをNotionに記録するツール。

まとめと次のステップ

この第4章では、SlackのWebhookを活用してリアルタイム管理AIを構築しました。メッセージやリアクションをリアルタイムで検知し、AIが自動応答やイベント記録を行えるようになりました。

次の第5章では、MCPサーバーの最適化とコミュニティへの貢献に焦点を当てます。サーバーのパフォーマンス向上、セキュリティ強化、そしてSlack用MCPサーバーをオープンソースとして共有する方法を学びます。コミュニティAIの未来に興味がある方は、ぜひお楽しみに！

役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の章でまたお会いしましょう！