GitHubMCPでプルリクエスト自動作成

やりたいこと

あるアプリのエラースタックトレースをGitHubMCP実行APIに渡すと、エラー調査とプルリクエスト作成を勝手にやってくれる。(Datadog等のログシステム経由でAPI叩ければ、ほぼ自動化できそう)

事前準備

・GitHub アカウント取得
・GitHub PAT(Personal Access token)取得
・GitHub リポジトリ作成
・Amazon BedRock用意
　※AWSコンソールからモデル使用申請(すぐ終わったと記憶)
　今回は「us.anthropic.claude-sonnet-4-20250514-v1:0」を使用

【１．サンプルアプリ準備】

requirements.txt

fastapi
uvicorn

sample.py

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/")
def exec1(id: int = None):
    print(id)
    result = 2 + id
    return {"sts": result}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

【２．GitHubMCP実行API準備】

requirements.txt

fastapi
uvicorn
strands-agents
mcp

main.py

import os
import httpx
import uvicorn
import asyncio
import logging
from typing import Dict, Any
from fastapi import FastAPI, Body, HTTPException, BackgroundTasks
from mcp.client.streamable_http import streamable_http_client
from strands.tools.mcp import MCPClient
from strands import Agent

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

SYSTEM_PROMPT = """
あなたはエラー調査用のエージェントです。スタックトレースから原因コードを特定し、
GitHub の MCP ツールを利用して、関連するリポジトリを検索・参照し、必要なら Issue を作成、
修正 PR を起票してください。日本語で簡潔に記録し、実行したツール名と入力をログに残してください。
"""

def mcp_transport():
    headers = {
        "Authorization": f"Bearer {os.environ.get('GITHUB_MCP_PAT','')}",
    }
    return streamable_http_client(
        url="https://api.githubcopilot.com/mcp/",
        http_client=httpx.AsyncClient(headers=headers, timeout=500.0)
    )

app = FastAPI()

@app.post("/mcp-sample")
async def triage(payload: Dict[str, Any] = Body(...), background_tasks: BackgroundTasks = None):
    """
    Webhookからスタックトレース等がPOSTされる想定のエンドポイント。
    フロントには即 200 を返し、裏でエージェント実行を続ける。
    """
    stack = payload.get('stack_trace', '')
    if not stack:
        raise HTTPException(status_code=400, detail="stacktrace not found")

    # 裏処理を登録（レスポンスはすぐ返る）
    background_tasks.add_task(run_agent_job, stack)

    # 受け付けたことだけ返す（必要に応じてジョブIDを生成して返しても良い）
    return {"status": "accepted"}

def exec_agent(agent: Agent, stack: str):
    # 同期関数：Agent の実行本体
    agent(f"スタックトレース:\n{stack}\n対象リポジトリ: totoaoao2全て")

async def run_agent_job(stack: str):
    """
    バックグラウンドで動く実処理。
    """
    try:
        mcp = MCPClient(transport_callable=mcp_transport)
        with mcp:
            tools = mcp.list_tools_sync()
            agent = Agent(
                tools=tools,
                system_prompt=(
                    "MCPツールを使用して、スタックトレースからコード特定→Issue作成→修正PRを実施せよ。\n"
                    "各ツールの引数は適切に設定し、結果と根拠をログに残すこと。"
                ),
            )

            # 非同期実行
            await asyncio.to_thread(exec_agent, agent, stack)

        logger.info("MCP agent job completed successfully.")

    except Exception:
        logger.exception("MCP agent job failed.")


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

【３．Claude Codeインストール】

>npm install -g @anthropic-ai/claude-cli

【４．環境変数セット】

>export CLAUDE_CODE_USE_BEDROCK=1
>export ANTHROPIC_MODEL='us.anthropic.claude-sonnet-4-20250514-v1:0'　※モデルは任意
>export AWS_ACCESS_KEY_ID=xxx
>export AWS_SECRET_ACCESS_KEY=xxx
>export GITHUB_MCP_PAT=xxx

【５．実行】

１．GitHubMCP実行API起動

>pip install -r requirements.txt
>python main.py

２．GitHubMCP実行APIへサンプルアプリのエラースタックトレース※を渡す(API叩く)

※idがundefined時のエラースタックトレース

Traceback (most recent call last):\n  File \"/usr/local/lib/python3.11/site-packages/uvicorn/protocols/http/h11_impl.py\", line 410, in run_asgi\n    result = await app(  # type: ignore[func-returns-value]\n             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/usr/local/lib/python3.11/site-packages/uvicorn/middleware/proxy_headers.py\", line 60, in __call__\n    return await self.app(scope, receive, send)\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/usr/local/lib/python3.11/site-packages/fastapi/applications.py\", line 1135, in __call__\n    await super().__call__(scope, receive, send)\n  File \"/usr/local/lib/python3.11/site-packages/starlette/applications.py\", line 107, in __call__\n    await self.middleware_stack(scope, receive, send)\n  File \"/usr/local/lib/python3.11/site-packages/starlette/middleware/errors.py\", line 186, in __call__\n    raise exc\n  File \"/usr/local/lib/python3.11/site-packages/starlette/middleware/errors.py\", line 164, in __call__\n    await self.app(scope, receive, _send)\n  File \"/usr/local/lib/python3.11/site-packages/starlette/middleware/exceptions.py\", line 63, in __call__\n    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)\n  File \"/usr/local/lib/python3.11/site-packages/starlette/_exception_handler.py\", line 53, in wrapped_app\n    raise exc\n  File \"/usr/local/lib/python3.11/site-packages/starlette/_exception_handler.py\", line 42, in wrapped_app\n    await app(scope, receive, sender)\n  File \"/usr/local/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py\", line 18, in __call__\n    await self.app(scope, receive, send)\n  File \"/usr/local/lib/python3.11/site-packages/starlette/routing.py\", line 716, in __call__\n    await self.middleware_stack(scope, receive, send)\n  File \"/usr/local/lib/python3.11/site-packages/starlette/routing.py\", line 736, in app\n    await route.handle(scope, receive, send)\n  File \"/usr/local/lib/python3.11/site-packages/starlette/routing.py\", line 290, in handle\n    await self.app(scope, receive, send)\n  File \"/usr/local/lib/python3.11/site-packages/fastapi/routing.py\", line 115, in app\n    await wrap_app_handling_exceptions(app, request)(scope, receive, send)\n  File \"/usr/local/lib/python3.11/site-packages/starlette/_exception_handler.py\", line 53, in wrapped_app\n    raise exc\n  File \"/usr/local/lib/python3.11/site-packages/starlette/_exception_handler.py\", line 42, in wrapped_app\n    await app(scope, receive, sender)\n  File \"/usr/local/lib/python3.11/site-packages/fastapi/routing.py\", line 101, in app\n    response = await f(request)\n               ^^^^^^^^^^^^^^^^\n  File \"/usr/local/lib/python3.11/site-packages/fastapi/routing.py\", line 355, in app\n    raw_response = await run_endpoint_function(\n                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/usr/local/lib/python3.11/site-packages/fastapi/routing.py\", line 245, in run_endpoint_function\n    return await run_in_threadpool(dependant.call, **values)\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/usr/local/lib/python3.11/site-packages/starlette/concurrency.py\", line 32, in run_in_threadpool\n    return await anyio.to_thread.run_sync(func)\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/usr/local/lib/python3.11/site-packages/anyio/to_thread.py\", line 63, in run_sync\n    return await get_async_backend().run_sync_in_worker_thread(\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py\", line 2502, in run_sync_in_worker_thread\n    return await future\n           ^^^^^^^^^^^^\n  File \"/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py\", line 986, in run\n    result = context.run(func, *args)\n             ^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/app/mcp-sample/sample.py\", line 9, in exec1\n    result = 2 + id\n             ~~^~~~\nTypeError: unsupported operand type(s) for +: 'int' and 'NoneType'

３．APIがissue、prを作成し始める

４．完了