LoginSignup
2
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@BNR-Longin株式会社BinaryTech | 日本品質 × ベトナム開発、最高のバランスを。

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

Posted at

はじめに

第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のセットアップ

  1. 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で生成）。
    • アプリを再インストールし、トークンを更新。
  2. チャンネル準備：
    • テスト用チャンネル（例：#mcp-project）を使用。
    • メッセージやリアクションを投稿してイベントを生成。
  3. ngrokの設定：
    • ngrokをインストール（brew install ngrokまたは公式サイト）。
    • ローカルサーバーを公開： 
      ngrok http 8125
    • 生成されたURLを記録（例：https://abc123.ngrok.io）。
    • SlackアプリのEvent SubscriptionsにWebhook URLを設定（例：https://abc123.ngrok.io/webhook）。
  4. 環境変数：
    第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が設定済み。
  • チャンネルにボットが招待済み。

サーバーのテスト

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

  1. ngrok起動：

    ngrok http 8125

    ngrok URL（例：https://abc123.ngrok.io）を記録し、SlackアプリのEvent Subscriptionsに設定（例：https://abc123.ngrok.io/webhook）。

  2. サーバー起動：

    python slack_realtime_server.py

    コンソールに「SlackリアルタイムMCPサーバーを起動中: http://localhost:8125」と「Webhookサーバーを起動中: http://localhost:8125」が表示。

  3. イベント検知のテスト：

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

  4. 最新イベント取得のテスト：
    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
}

  5. 自動応答のテスト：

    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）に以下を追加：

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

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

2
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?