SlackでAIを強化する | 第4章：リアルタイム管理AIを強化：Webhook連携

## はじめに

第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対応の設定。

インストールコマンド（必要に応じて）：
```bash
pip install requests python-dotenv
```

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

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

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

```python
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`が設定済み。
- チャンネルにボットが招待済み。

## サーバーのテスト

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

1. **ngrok起動**：
   ```bash
   ngrok http 8125
   ```
   ngrok URL（例：`https://abc123.ngrok.io`）を記録し、SlackアプリのEvent Subscriptionsに設定（例：`https://abc123.ngrok.io/webhook`）。

2. **サーバー起動**：
   ```bash
   python slack_realtime_server.py
   ```
   コンソールに「SlackリアルタイムMCPサーバーを起動中: http://localhost:8125」と「Webhookサーバーを起動中: http://localhost:8125」が表示。

3. **イベント検知のテスト**：
   - SlackのUIでチャンネル（例：`#mcp-project`）にメッセージを投稿（例：「タスク確認」）またはリアクション（例：👍）を追加。
   - サーバーがイベントをWebhook経由で検知。

4. **最新イベント取得のテスト**：
   Pythonでリクエストを送信：
   ```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))
   ```
   期待されるレスポンス：
   ```json
   {
     "jsonrpc": "2.0",
     "result": {
       "status": "success",
       "event_info": {
         "type": "message",
         "user": "U123456",
         "text": "タスク確認",
         "channel": "C123456",
         "timestamp": "1617187200.000100"
       }
     },
     "id": 1
   }
   ```

5. **自動応答のテスト**：
   ```python
   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に接続します：

1. **設定ファイルの編集**：
   Claude Desktopの設定ファイル（例：`claude_desktop_config.json`）に以下を追加：
   ```json
   {
     "mcp_servers": [
       {
         "name": "SlackRealtimeServer",
         "url": "http://localhost:8125",
         "auth": "none"
       }
     ]
   }
   ```

2. **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の未来に興味がある方は、ぜひお楽しみに！

---

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