はじめに

第3章では、Trelloのカードデータを活用してタスク分析エージェントを構築し、完了率やボトルネックを分析しました。これにより、AIがチームのパフォーマンスを評価し、プロセス改善のインサイトを提供できるようになりました。今回は、TrelloのWebhook機能を利用して、カードの移動や期限変更をリアルタイムで検知するリアルタイム管理AIを構築します。

この第4章では、MCPサーバーにWebhookを統合し、Trelloボードの更新を即座に捕捉します。AIはこれを利用して、変更内容をコメントとして記録したり、重要な更新をチームに通知したりできます。コード例とステップごとのガイドで、リアルタイム管理AIの構築を体験しましょう。さあ、始めましょう！

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

リアルタイム管理AIは、Trelloボードの変更を監視し、プロジェクトの動的な管理を支援するエージェントです。MCPサーバーとTrelloのWebhookを組み合わせることで、以下のような機能を実現できます：

変更検知：カードの作成、移動、期限変更をリアルタイムで捕捉。
コメント生成：変更内容を要約し、Trelloにコメントとして投稿。
通知：重要な変更をチームに通知（例：Slackやメール）。

ユースケース

プロジェクト管理：カードがDoneに移動した際に自動でコメントを追加。
チームコラボレーション：期限変更を即座にチームに通知。
プロセス監視：意図しない変更（例：期限の削除）を検出。

開発環境の準備

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

Python 3.8以降、mcpライブラリ、requestsライブラリ、Claude Desktop：これまでと同じ。
python-dotenv：環境変数の管理（既にインストール済み）。
ngrok：ローカルサーバーを公開し、Webhookを受信。
Trelloボード：Webhookテスト用のデータ。

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

pip install requests python-dotenv

Trelloのセットアップ

ボード準備：
- 第3章のTrelloボード（例：MCP-Project）を使用。
- リスト（To Do、In Progress、Done）とカードを準備。
- テスト用にカードを移動（例：To DoからIn Progress）または期限を変更。
Webhookの設定：
- TrelloのDeveloper PortalでWebhookを作成。
- イベント：updateCardを選択（カードの移動や期限変更を捕捉）。
- Webhook URL：後でngrokから取得（例：https://abc123.ngrok.io）。
- WebhookのCallback URLを記録（セキュリティ用）。
ngrokの設定：

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

TRELLO_API_KEY=your_api_key
TRELLO_TOKEN=your_token
TRELLO_BOARD_ID=your_board_id
TRELLO_TO_DO_LIST_ID=your_to_do_list_id
TRELLO_IN_PROGRESS_LIST_ID=your_in_progress_list_id
TRELLO_DONE_LIST_ID=your_done_list_id

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

以下のMCPサーバーは、Trelloの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 TrelloRealtimeServer(MCPServer):
    def __init__(self, host, port, api_key, token, board_id, to_do_list_id, in_progress_list_id, done_list_id):
        super().__init__(host, port)
        self.api_key = api_key
        self.token = token
        self.board_id = board_id
        self.to_do_list_id = to_do_list_id
        self.in_progress_list_id = in_progress_list_id
        self.done_list_id = done_list_id
        self.base_url = "https://api.trello.com/1"
        self.params = {"key": api_key, "token": token}
        self.latest_update = None
        self.register_resource("get_latest_update", self.get_latest_update)
        self.register_tool("add_comment", self.add_comment)
        self.start_webhook_server()

    def get_latest_update(self, params):
        try:
            if self.latest_update:
                action = self.latest_update["action"]
                update_info = {
                    "card_id": action["data"]["card"]["id"],
                    "card_name": action["data"]["card"]["name"],
                    "action_type": action["type"],
                    "list_before": action["data"].get("listBefore", {}).get("name", ""),
                    "list_after": action["data"].get("listAfter", {}).get("name", ""),
                    "date": action["date"]
                }
                return {"status": "success", "update_info": update_info}
            return {"status": "success", "update_info": None, "message": "更新なし"}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def add_comment(self, params):
        try:
            card_id = params.get("card_id", "")
            message = params.get("message", "")
            if not card_id or not message:
                return {"status": "error", "message": "カードIDとコメントが必要です"}
            
            url = f"{self.base_url}/cards/{card_id}/actions/comments"
            payload = {"text": message}
            response = requests.post(url, params=self.params, json=payload)
            response.raise_for_status()
            comment = response.json()
            return {"status": "success", "comment_id": comment["id"]}
        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)
                self.server.parent.latest_update = json.loads(post_data.decode("utf-8"))
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"Webhook received")

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

if __name__ == "__main__":
    load_dotenv()
    server = TrelloRealtimeServer(
        host="localhost",
        port=8114,
        api_key=os.getenv("TRELLO_API_KEY"),
        token=os.getenv("TRELLO_TOKEN"),
        board_id=os.getenv("TRELLO_BOARD_ID"),
        to_do_list_id=os.getenv("TRELLO_TO_DO_LIST_ID"),
        in_progress_list_id=os.getenv("TRELLO_IN_PROGRESS_LIST_ID"),
        done_list_id=os.getenv("TRELLO_DONE_LIST_ID")
    )
    print("TrelloリアルタイムMCPサーバーを起動中: http://localhost:8114")
    server.start()

コードの説明

get_latest_update：Webhookで捕捉した最新のカード更新情報を取得（カードID、名前、アクションタイプ、リスト移動、更新日）。
add_comment：指定したカードにコメントを追加。
start_webhook_server：ローカルWebhookサーバーを起動し、Trelloからの更新を受信。
WebhookHandler：Webhookリクエストを処理し、更新データを保存。

前提条件

Trelloボードにリストとカードが存在。
TrelloのWebhookが設定済みで、ngrok URLが正しく登録。
.envファイルに正しいTRELLO_API_KEY、TRELLO_TOKEN、TRELLO_BOARD_ID、TRELLO_TO_DO_LIST_ID、TRELLO_IN_PROGRESS_LIST_ID、TRELLO_DONE_LIST_IDが設定済み。

サーバーのテスト

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

ngrok起動：
```
ngrok http 8115
```
ngrok URL（例：https://abc123.ngrok.io）をTrelloのWebhook設定に登録。
サーバー起動：
```
python trello_realtime_server.py
```
コンソールに「TrelloリアルタイムMCPサーバーを起動中: http://localhost:8114」と「Webhookサーバーを起動中: http://localhost:8115」が表示。
Webhookのテスト：
- TrelloのUIでカードを移動（例：To DoからIn Progress）または期限を変更。
- サーバーのコンソールにWebhook受信ログが表示。

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

import requests
import json

url = "http://localhost:8114"
payload = {
    "jsonrpc": "2.0",
    "method": "get_latest_update",
    "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",
    "update_info": {
      "card_id": "card126",
      "card_name": "コードレビュー",
      "action_type": "updateCard",
      "list_before": "To Do",
      "list_after": "In Progress",
      "date": "2025-04-21T10:00:00Z"
    }
  },
  "id": 1
}

コメント追加のテスト：

payload = {
    "jsonrpc": "2.0",
    "method": "add_comment",
    "params": {
      "card_id": "card126",
      "message": "カードがIn Progressに移動しました"
    },
    "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": "TrelloRealtimeServer",
      "url": "http://localhost:8114",
      "auth": "none"
    }
  ]
}

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

最新のカード更新を教えてください。

レスポンス例：

最新のカード更新：
- カード：コードレビュー
- アクション：To DoからIn Progressに移動
- 日時：2025-04-21 10:00

別のプロンプト：

カード「card126」に移動のコメントを追加してください。

レスポンス例：

カード「コードレビュー」に「カードがIn Progressに移動しました」をコメントしました。

実装のコツと注意点

Webhookの安定性：ngrokの無料枠ではセッションが切れる場合がある。本番環境では固定URL（例：AWS Lambda）を使用。
レートリミティング：Trello APIの制限（通常100リクエスト/10秒）に注意。
セキュリティ：本番環境では、Webhookリクエストの検証を強化（例：TrelloのCallback URLを使用）。
テスト：テスト用ボードを作成し、本番データに影響を与えない。
拡張性：大量の更新を処理する場合、キャッシュやキュー（例：Redis）を検討。

試してみよう：挑戦課題

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

カードの期限変更を検知し、自動でコメントを追加。
更新データをSlackに通知するツール。
特定アクション（例：createCard）のみをフィルタリングする機能。

まとめと次のステップ

この第4章では、TrelloのWebhookを活用してリアルタイム管理AIを構築しました。カードの更新をリアルタイムで捕捉し、AIがコメントや通知を生成できるようになりました。

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

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