SlackでAIを強化する | 第2章：タスクを自動化：AIによるメッセージ操作

## はじめに

第1章では、**Slack**と**Model Context Protocol（MCP）**の基本を学び、Slackからメッセージデータを取得するMCPサーバーを構築しました。これにより、AIがチャンネルの最新メッセージを取得できるようになりました。今回は、この基盤を進化させ、Slackにメッセージを送信したり、ボットを介して自動応答したりする機能を実装します。これにより、**タスクを自動化**するエージェントを構築します。

この第2章では、MCPサーバーを通じてAIがメッセージを投稿したり、特定のキーワードに反応して自動応答したり、チャンネルを作成したりできるようにします。たとえば、ユーザーが「タスク追加」と入力すると、AIが確認メッセージを送信し、Notionにタスクを記録できます。コード例とステップごとのガイドで、コミュニケーション自動化の可能性を体感しましょう。さあ、始めましょう！

## タスク自動化エージェントとは？

タスク自動化エージェントは、Slackのメッセージやチャンネルを操作し、チームコミュニケーションを効率化するAIです。MCPサーバーを介して、以下のような機能を実現できます：
- **メッセージ送信**：AIがチャンネルやユーザーにメッセージを投稿。
- **自動応答**：特定のキーワードやイベントに基づいてボットが反応。
- **チャンネル管理**：新しいチャンネルを自動作成。

### ユースケース
- タスク管理：AIが「タスク追加」メッセージを検知し、NotionやTrelloにタスクを登録。
- サポートボット：AIがFAQに自動応答し、カスタマーサポートを効率化。
- チーム通知：AIが重要なメッセージを要約し、特定のチャンネルに投稿。

## 開発環境の準備

第1章の環境を基に、以下の準備を行います：
- **Python 3.8以降**、**mcpライブラリ**、**requestsライブラリ**、**Claude Desktop**：第1章と同じ。
- **python-dotenv**：環境変数の管理（既にインストール済み）。
- **Slackアプリ**：メッセージ送信とチャンネル管理用のスコープを追加。

### Slackのセットアップ
1. **Slackアプリの拡張**：
   - 第1章のアプリ（例：`MCP-Bot`）を使用。
   - スコープを追加：`chat:write`（メッセージ送信）、`channels:manage`（チャンネル作成）。
   - アプリを再インストールし、Bot User OAuth Tokenを更新。
   - テスト用チャンネル（例：`#mcp-project`）にボットを招待。
2. **環境変数**：
   第1章の`.env`ファイル（`SLACK_TOKEN`、`SLACK_CHANNEL_ID`）を再利用：
   ```bash
   SLACK_TOKEN=your_bot_token
   SLACK_CHANNEL_ID=your_channel_id
   ```

## コード例：タスク自動化用MCPサーバー

以下のMCPサーバーは、Slackからメッセージを取得し、メッセージ送信やチャンネル作成の機能を提供します。

```python
from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests

class SlackTaskServer(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.register_resource("get_messages", self.get_messages)
        self.register_tool("send_message", self.send_message)
        self.register_tool("create_channel", self.create_channel)

    def get_messages(self, params):
        try:
            url = f"{self.base_url}/conversations.history"
            query = {"channel": self.channel_id, "limit": params.get("limit", 10)}
            response = requests.get(url, headers=self.headers, params=query)
            response.raise_for_status()
            messages = response.json()["messages"]
            message_list = [
                {
                    "text": msg["text"],
                    "user": msg.get("user", "unknown"),
                    "timestamp": msg["ts"]
                }
                for msg in messages
            ]
            return {"status": "success", "messages": message_list}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    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 create_channel(self, params):
        try:
            name = params.get("name", "")
            if not name:
                return {"status": "error", "message": "チャンネル名が必要です"}
            
            url = f"{self.base_url}/conversations.create"
            payload = {
                "name": name,
                "is_private": params.get("is_private", False)
            }
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            channel = response.json()["channel"]
            return {"status": "success", "channel_id": channel["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = SlackTaskServer(
        host="localhost",
        port=8123,
        token=os.getenv("SLACK_TOKEN"),
        channel_id=os.getenv("SLACK_CHANNEL_ID")
    )
    print("SlackタスクMCPサーバーを起動中: http://localhost:8123")
    server.start()
```

### コードの説明
- **get_messages**：第1章から再利用。チャンネルのメッセージを取得。
- **send_message**：指定したチャンネルにメッセージを送信。テキストとチャンネルIDを指定。
- **create_channel**：新しいチャンネルを作成。チャンネル名とプライバシー設定を指定。
- **register_resource/tool**：メッセージ取得をリソース、メッセージ送信・チャンネル作成をツールとして登録。

### 前提条件
- Slackアプリに`chat:write`と`channels:manage`スコープが付与されている。
- チャンネルが存在し、ボットが招待されている。
- `.env`ファイルに正しい`SLACK_TOKEN`と`SLACK_CHANNEL_ID`が設定済み。

## サーバーのテスト

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

1. **サーバー起動**：
   ```bash
   python slack_task_server.py
   ```
   コンソールに「SlackタスクMCPサーバーを起動中: http://localhost:8123」と表示。

2. **メッセージ送信のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8123"
   payload = {
       "jsonrpc": "2.0",
       "method": "send_message",
       "params": {
           "text": "タスクが追加されました！",
           "channel": os.getenv("SLACK_CHANNEL_ID")
       },
       "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",
       "message_id": "1617187200.000300"
     },
     "id": 1
   }
   ```

3. **チャンネル作成のテスト**：
   ```python
   payload = {
       "jsonrpc": "2.0",
       "method": "create_channel",
       "params": {
           "name": "mcp-task",
           "is_private": False
       },
       "id": 2
   }
   response = requests.post(url, json=payload)
   print(json.dumps(response.json(), indent=2, ensure_ascii=False))
   ```
   期待されるレスポンス：
   ```json
   {
     "jsonrpc": "2.0",
     "result": {
       "status": "success",
       "channel_id": "C123456"
     },
     "id": 2
   }
   ```

## Claude Desktopとの接続

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

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

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   チャンネル #mcp-project に「タスクが追加されました！」を送信してください。
   ```
   レスポンス例：
   ```
   チャンネル #mcp-project に「タスクが追加されました！」を送信しました。
   ```

   別のプロンプト：
   ```
   新しいチャンネル「mcp-task」を作成してください。
   ```
   レスポンス例：
   ```
   チャンネル「mcp-task」を作成しました。
   ```

## 実装のコツと注意点

- **エラーハンドリング**：無効なチャンネルIDや権限エラーを適切に処理。
- **レートリミティング**：Slack APIの制限（例：`chat.postMessage`は1秒に1リクエスト）に注意。
- **セキュリティ**：本番環境では、`auth: none`を避け、トークン認証を導入。
- **テスト**：テスト用チャンネルを作成し、本番データに影響を与えない。
- **拡張性**：大量のメッセージを処理する場合、キャッシュ（例：Redis）を検討。

## 試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- 特定のキーワード（例：「タスク追加」）に反応して自動応答する機能。
- メッセージにスレッドを追加するツール。
- 作成したチャンネルに初期メッセージを投稿する機能。

## まとめと次のステップ

この第2章では、Slackにメッセージを送信し、チャンネルを作成するMCPサーバーを構築し、コミュニケーション自動化エージェントを実現しました。AIがメッセージ操作を処理できるようになり、チームの効率が向上しました。

次の第3章では、Slackの会話データを活用して**会話分析エージェント**を構築します。たとえば、AIがメッセージのトピックやエンゲージメントを分析し、チームのコミュニケーションを評価します。会話分析AIに興味がある方は、ぜひお楽しみに！

---

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