はじめに
第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のセットアップ
-
Slackアプリの拡張：
- 第1章のアプリ（例：
MCP-Bot）を使用。 - スコープを追加：
chat:write（メッセージ送信）、channels:manage（チャンネル作成）。 - アプリを再インストールし、Bot User OAuth Tokenを更新。
- テスト用チャンネル（例：
#mcp-project）にボットを招待。
- 第1章のアプリ（例：
-
環境変数：
第1章の.envファイル（SLACK_TOKEN、SLACK_CHANNEL_ID）を再利用：SLACK_TOKEN=your_bot_token SLACK_CHANNEL_ID=your_channel_id
コード例：タスク自動化用MCPサーバー
以下のMCPサーバーは、Slackからメッセージを取得し、メッセージ送信やチャンネル作成の機能を提供します。
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が設定済み。
サーバーのテスト
サーバーが正しく動作するか確認します：
-
サーバー起動：
python slack_task_server.pyコンソールに「SlackタスクMCPサーバーを起動中: http://localhost:8123」と表示。
-
メッセージ送信のテスト：
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))期待されるレスポンス：
{ "jsonrpc": "2.0", "result": { "status": "success", "message_id": "1617187200.000300" }, "id": 1 } -
チャンネル作成のテスト：
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))期待されるレスポンス：
{ "jsonrpc": "2.0", "result": { "status": "success", "channel_id": "C123456" }, "id": 2 }
Claude Desktopとの接続
サーバーをClaude Desktopに接続します：
-
設定ファイルの編集：
Claude Desktopの設定ファイル（例：claude_desktop_config.json）に以下を追加：{ "mcp_servers": [ { "name": "SlackTaskServer", "url": "http://localhost:8123", "auth": "none" } ] } -
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に興味がある方は、ぜひお楽しみに！
役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の章でまたお会いしましょう！