LoginSignup
1
1

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 | 日本品質 × ベトナム開発、最高のバランスを。

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

Posted at

はじめに

第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対応の設定。

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

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または公式サイト）。
    • ローカルサーバーを公開： 
      ngrok http 8130
    • 生成されたURLを記録（例：https://abc123.ngrok.io）。
    • GitHubのWebhook設定にPayload URLを設定（例：https://abc123.ngrok.io/webhook）。
  5. 環境変数：
    第3章の.envファイルに以下を追加： 
    GITHUB_TOKEN=your_token
GITHUB_REPO=yourusername/mcp-test-repo
GITHUB_WEBHOOK_SECRET=mysecret

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

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

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起動：

    ngrok http 8130

    ngrok URL（例：https://abc123.ngrok.io）を記録し、GitHubリポジトリのWebhook設定に設定（例：https://abc123.ngrok.io/webhook）。

  2. サーバー起動：

    python github_realtime_server.py

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

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

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

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

    期待されるレスポンス：

    {
  "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. 自動コメントのテスト：

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

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

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

1
1
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
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?