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

## はじめに

第3章では、**GitHub**のコミットデータを活用してコード分析エージェントを構築しました。変更頻度や貢献者を分析することで、AIが開発プロセスのインサイトを提供し、コード品質向上を支援できるようになりました。今回は、GitHubのWebhook機能を利用して、IssueやPull Request（PR）のイベントをリアルタイムで検知する**リアルタイム管理AI**を構築します。

この第4章では、MCPサーバーにGitHubのWebhookを統合し、Issue作成やPRマージなどのイベントを即座に捕捉します。AIはこれを利用して、自動コメントを追加したり、イベントを他のツール（例：Slack）に通知したりできます。コード例とステップごとのガイドで、リアルタイム管理AIの構築を体験しましょう。さあ、始めましょう！

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

リアルタイム管理AIは、GitHubのリポジトリイベント（Issue作成、PRマージ、コメント追加など）を監視し、開発プロセスを動的に管理するエージェントです。MCPサーバーとGitHubのWebhookを組み合わせることで、以下のような機能を実現できます：
- **イベント検知**：新しいIssue、PR、コメントをリアルタイムで捕捉。
- **自動応答**：特定のイベント（例：Issue作成）に基づいてコメントやラベルを追加。
- **外部連携**：イベントをSlackやメールに通知し、コラボレーションを強化。

### ユースケース
- プロジェクト管理：AIが「バグ」Issue作成時に自動で「bug」ラベルを付ける。
- チーム通知：重要なイベント（例：PRマージ）にSlackで通知。
- プロセス監視：意図しないイベント（例：未レビューのPRマージ）を検出。

## 開発環境の準備

第3章の環境を基に、以下の追加準備を行います：
- **Python 3.8以降**、**mcpライブラリ**、**requestsライブラリ**、**Claude Desktop**：これまでと同じ。
- **python-dotenv**：環境変数の管理（既にインストール済み）。
- **ngrok**：ローカルサーバーを公開し、Webhookを受信。
- **GitHubリポジトリ**：Webhook対応の設定。

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

### GitHubのセットアップ
1. **GitHubトークンの確認**：
   - 第3章のトークンを使用。
   - スコープを確認：`repo`（Issue、PR操作）、`admin:repo_hook`（Webhook管理）。
2. **Webhookの設定**：
   - リポジトリ（例：`yourusername/mcp-test-repo`）の「Settings」→「Webhooks」に移動。
   - 新しいWebhookを追加：
     - **Payload URL**：ngrokで生成するURL（例：`https://abc123.ngrok.io/webhook`）。
     - **Content type**：`application/json`。
     - **Events**：`Issues`（Issue関連）、`Pull requests`（PR関連）、`Push`（コミット）。
     - **Active**：チェック。
   - シークレット（オプション）：Webhook署名検証用の文字列を設定（例：`mysecret`）。
3. **リポジトリ準備**：
   - IssueやPRを作成し、イベントを生成（例：Issueを開く、PRにコメント）。
4. **ngrokの設定**：
   - ngrokをインストール（`brew install ngrok`または公式サイト）。
   - ローカルサーバーを公開：
     ```bash
     ngrok http 8130
     ```
   - 生成されたURLを記録（例：`https://abc123.ngrok.io`）。
   - GitHubのWebhook設定にPayload URLを設定（例：`https://abc123.ngrok.io/webhook`）。
5. **環境変数**：
   第3章の`.env`ファイルに以下を追加：
   ```bash
   GITHUB_TOKEN=your_token
   GITHUB_REPO=yourusername/mcp-test-repo
   GITHUB_WEBHOOK_SECRET=mysecret
   ```

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

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

```python
from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import hmac
import hashlib
import threading

class GitHubRealtimeServer(MCPServer):
    def __init__(self, host, port, token, repo, webhook_secret):
        super().__init__(host, port)
        self.token = token
        self.repo = repo
        self.webhook_secret = webhook_secret
        self.base_url = "https://api.github.com"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github.v3+json"
        }
        self.latest_event = None
        self.register_resource("get_latest_event", self.get_latest_event)
        self.register_tool("add_issue_comment", self.add_issue_comment)
        self.start_webhook_server()

    def verify_webhook_signature(self, body, signature):
        try:
            computed_sig = "sha256=" + hmac.new(
                self.webhook_secret.encode("utf-8"),
                body.encode("utf-8"),
                hashlib.sha256
            ).hexdigest()
            return hmac.compare_digest(computed_sig, signature)
        except Exception as e:
            return False

    def add_issue_comment(self, params):
        try:
            issue_number = params.get("issue_number", 0)
            comment = params.get("comment", "")
            if not issue_number or not comment:
                return {"status": "error", "message": "Issue番号とコメントが必要です"}
            
            url = f"{self.base_url}/repos/{self.repo}/issues/{issue_number}/comments"
            payload = {"body": comment}
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            return {"status": "success", "comment_id": response.json()["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_latest_event(self, params):
        try:
            if self.latest_event:
                event_type = self.latest_event.get("event", "")
                event = self.latest_event.get("payload", {})
                event_info = {
                    "event_type": event_type,
                    "action": event.get("action", ""),
                    "issue_number": event.get("issue", {}).get("number", 0),
                    "pr_number": event.get("pull_request", {}).get("number", 0),
                    "user": event.get("sender", {}).get("login", "unknown"),
                    "created_at": event.get("issue", {}).get("created_at", "") or event.get("pull_request", {}).get("created_at", "")
                }
                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)
                signature = self.headers.get("X-Hub-Signature-256", "")

                # Webhook署名検証
                if not self.server.parent.verify_webhook_signature(post_data.decode("utf-8"), signature):
                    self.send_response(401)
                    self.end_headers()
                    self.wfile.write(b"Invalid signature")
                    return

                data = json.loads(post_data.decode("utf-8"))
                event_type = self.headers.get("X-GitHub-Event", "")

                # イベント処理
                if event_type in ["issues", "pull_request"]:
                    self.server.parent.latest_event = {"event": event_type, "payload": data}

                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"Webhook received")

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

if __name__ == "__main__":
    load_dotenv()
    server = GitHubRealtimeServer(
        host="localhost",
        port=8130,
        token=os.getenv("GITHUB_TOKEN"),
        repo=os.getenv("GITHUB_REPO"),
        webhook_secret=os.getenv("GITHUB_WEBHOOK_SECRET")
    )
    print("GitHubリアルタイムMCPサーバーを起動中: http://localhost:8130")
    server.start()
```

### コードの説明
- **verify_webhook_signature**：GitHubのWebhook署名（`X-Hub-Signature-256`）をHMAC-SHA256で検証。
- **add_issue_comment**：指定したIssueにコメントを追加。Issue番号とコメント内容を指定。
- **get_latest_event**：Webhookで受信した最新イベント（IssueまたはPR）を取得。
- **start_webhook_server**：ローカルWebhookサーバーを起動し、GitHubからのイベントを処理。
- **WebhookHandler**：GitHubのイベント（Issue、PR）を受信し、署名検証を行う。

### 前提条件
- GitHubリポジトリにWebhookが設定済み（`issues`、`pull_request`イベント）。
- ngrokでWebhook URLが公開され、GitHubに登録済み。
- `.env`ファイルに正しい`GITHUB_TOKEN`、`GITHUB_REPO`、`GITHUB_WEBHOOK_SECRET`が設定済み。
- トークンに`repo`と`admin:repo_hook`スコープが付与されている。

## サーバーのテスト

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

1. **ngrok起動**：
   ```bash
   ngrok http 8130
   ```
   ngrok URL（例：`https://abc123.ngrok.io`）を記録し、GitHubリポジトリのWebhook設定に設定（例：`https://abc123.ngrok.io/webhook`）。

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

3. **イベント検知のテスト**：
   - GitHubのUIでリポジトリにIssueを作成（例：「バグ報告」）またはPRを開く。
   - サーバーがイベントをWebhook経由で検知。

4. **最新イベント取得のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8130"
   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": {
         "event_type": "issues",
         "action": "opened",
         "issue_number": 4,
         "pr_number": 0,
         "user": "yourusername",
         "created_at": "2025-04-22T12:00:00Z"
       }
     },
     "id": 1
   }
   ```

5. **自動コメントのテスト**：
   ```python
   payload = {
       "jsonrpc": "2.0",
       "method": "add_issue_comment",
       "params": {
           "issue_number": 4,
           "comment": "Issueを確認しました。詳細を教えてください。"
       },
       "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": "GitHubRealtimeServer",
         "url": "http://localhost:8130",
         "auth": "none"
       }
     ]
   }
   ```

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   最新のリポジトリイベントを教えてください。
   ```
   レスポンス例：
   ```
   最新のリポジトリイベント：
   - イベント：Issue
   - アクション：作成
   - Issue番号：#4
   - ユーザー：yourusername
   - 作成日時：2025-04-22 12:00
   ```

   別のプロンプト：
   ```
   Issue #4に「Issueを確認しました。詳細を教えてください。」とコメントしてください。
   ```
   レスポンス例：
   ```
   Issue #4にコメントを追加しました。
   ```

## 実装のコツと注意点

- **Webhook検証**：GitHubの署名検証（`X-Hub-Signature-256`）を正しく処理。
- **レートリミティング**：GitHub APIの制限（例：5000リクエスト/時間）に注意。
- **セキュリティ**：本番環境では、HTTPSを有効化し、Webhookシークレットを必須化。
- **テスト**：テスト用リポジトリを作成し、本番データに影響を与えない。
- **拡張性**：大量のイベントを処理する場合、キュー（例：Redis、RabbitMQ）を検討。

## 試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- 「バグ」キーワードを含むIssue作成時に「bug」ラベルを自動付与。
- PRマージイベントを検知し、Slackに通知する機能。
- イベントデータをリポジトリのIssueに記録するツール。

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

この第4章では、GitHubのWebhookを活用してリアルタイム管理AIを構築しました。IssueやPRのイベントをリアルタイムで検知し、AIが自動コメントや通知を行えるようになりました。

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

---

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