私は普段、コードレビューの効率化に頭を悩ませてきました。Pull Requestの数が増えると、一つ一つのdiffを丁寧に見る時間が確保できません。人間の目で品質の担保をしようと思うと、スケーラビリティに限界がありますよね。
本稿では、MCP（Model Context Protocol）を活用したコードレビュー自動化ツールの構築方法を詳細に解説します。HolySheep AIのAPIを活用した実践的なアーキテクチャ設計から、パフォーマンス最適化、コスト最適化まで、本番環境に耐えうるシステム構築のポイントをお伝えします。
MCPとは？コードレビュー自動化の基盤技術
MCPは、AIモデルと外部ツール・データソースをシームレスに接続するためのプロトコルです。Claude DesktopやCursorなど主要なAI IDEでサポートされており、コードレビューという特定のユースケースにおいて、以下の強みを提供します：
- 構造化されたコンテキスト渡しが可能：Git diff、ファイルツリー、テスト結果などを统一的なフォーマットでLLMに提供
- ツール呼び出しの標準化：リント実行、ファイル読み込み、コマンド実行を统一的APIで管理
- ステート管理の改善：複数のリクエストを跨いだコンテキスト維持が容易
システムアーキテクチャ設計
全体構成
┌─────────────────────────────────────────────────────────────────────┐
│ コードレビューシステム │
├─────────────────────────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │ Git Diff │───▶│ Parser │───▶│ MCP │───▶│ HolySheep │ │
│ │ Receiver │ │ │ │ Server │ │ AI API │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────────┘ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────────────────┐│
│ │ Review Cache (Redis/Memory) ││
│ └──────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────┘コアコンポーネント
// src/mcp_server/review_server.py
import asyncio
import hashlib
from typing import AsyncIterator, List
from dataclasses import dataclass
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx
HolySheep AI API設定
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
MODEL = "gpt-4.1" # $8/MTok - 高品質レビュー用
@dataclass
class DiffFile:
"""Git diffの单个ファイル表現"""
path: str
old_content: str
new_content: str
hunks: List[dict]
@property
def file_hash(self) -> str:
"""変更内容のハッシュ（キャッシュキー用）"""
content = f"{self.path}:{self.new_content}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
class CodeReviewServer:
"""MCPプロトコルを活用したコードレビューサーバー"""
def __init__(self):
self.server = Server("code-review")
self.cache: dict[str, dict] = {}
self._register_tools()
def _register_tools(self):
"""MCPツール登録"""
@self.server.list_tools()
async def list_tools() -> List[Tool]:
return [
Tool(
name="review_diff",
description="Git diffを分析してコードレビューコメントを生成",
inputSchema={
"type": "object",
"properties": {
"diff_content": {"type": "string"},
"language": {"type": "string", "enum": ["python", "javascript", "go", "rust"]},
"review_depth": {"type": "string", "enum": ["quick", "standard", "thorough"]}
}
}
),
Tool(
name="batch_review",
description="複数のファイルをまとめてレビュー",
inputSchema={
"type": "object",
"properties": {
"files": {
"type": "array",
"items": {
"type": "object",
"properties": {
"path": {"type": "string"},
"diff": {"type": "string"}
}
}
}
}
}
)
]
@self.server.call_tool()
async def call_tool(name: str, arguments: dict) -> List[TextContent]:
if name == "review_diff":
return await self._review_single_diff(**arguments)
elif name == "batch_review":
return await self._review_batch(**arguments)
raise ValueError(f"Unknown tool: {name}")
async def _call_holysheep(self, prompt: str, max_tokens: int = 2048) -> str:
"""HolySheep AI API呼び出し（¥1=$1の圧倒的低コスト）"""
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": MODEL,
"messages": [
{
"role": "system",
"content": """あなたは経験丰富的なシニアエンジニアです。
コードレビューを行い、具体的な改善提案を提供してください。
出力形式：
レビューサマリー
[全体の評価]
問題点（重大度順）
🔴 高
[ファイル名:行番号] 具体的な問題と修正案
🟡 中
[ファイル名:行番号] 軽微な問題
🟢 提案
[ファイル名:行番号] meilleures pratiques"""
},
{"role": "user", "content": prompt}
],
"max_tokens": max_tokens,
"temperature": 0.3 # 一貫性重視で低めに設定
}
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]Git Diff解析の実装
diff解析はコードレビューツールの心臓部です。私は複数のパーサーライブラリを比較検討しましたが、最終的にはHolySheep AIのAPIと組み合わせたカスタムパーサーを選ぶ結論に至りました。
// src/diff_parser.py
import re
from typing import List, Tuple, Optional
from dataclasses import dataclass
@dataclass
class DiffHunk:
"""Unified diffのハンク単位"""
old_start: int
old_count: int
new_start: int
new_count: int
lines: List[str]
context_before: List[str]
context_after: List[str]
class GitDiffParser:
"""Git unified diff形式のパーサー"""
HUNK_HEADER = re.compile(
r'^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@(.*)$'
)
def parse(self, diff_text: str) -> List[DiffFile]:
"""複数ファイルのdiffを解析"""
files = []
current_file = None
current_hunks = []
pending_context: List[str] = []
for line in diff_text.split('\n'):
if line.startswith('diff --git'):
# 前のファイルを保存
if current_file:
files.append(self._build_file(current_file, current_hunks))
# 新しいファイルの開始
path_match = re.search(r'b/(.+)$', line)
current_file = path_match.group(1) if path_match else "unknown"
current_hunks = []
pending_context = []
elif line.startswith('--- ') or line.startswith('+++ '):
# ファイルパス自体はpreviousブロックで処理済み
pass
elif line.startswith('@@'):
# ハンク開始
hunk_match = self.HUNK_HEADER.match(line)
if hunk_match:
hunk = DiffHunk(
old_start=int(hunk_match.group(1)),
old_count=int(hunk_match.group(2) or 1),
new_start=int(hunk_match.group(3)),
new_count=int(hunk_match.group(4) or 1),
lines=[line],
context_before=pending_context[-3:],
context_after=[]
)
current_hunks.append(hunk)
elif current_hunks and line.startswith((' ', '+', '-')):
# 実際の変更行
current_hunks[-1].lines.append(line)
if not line.startswith('-'):
pending_context.append(line)
# 最後のファイルを追加
if current_file:
files.append(self._build_file(current_file, current_hunks))
return files
def _build_file(self, path: str, hunks: List[DiffHunk]) -> DiffFile:
"""DiffFileオブジェクトを構築"""
old_lines = []
new_lines = []
for hunk in hunks:
old_lines.extend(hunk.context_before)
new_lines.extend(hunk.context_before)
for line in hunk.lines:
if line.startswith('-'):
old_lines.append(line[1:])
elif line.startswith('+'):
new_lines.append(line[1:])
else:
old_lines.append(line[1:] if len(line) > 1 else '')
new_lines.append(line[1:] if len(line) > 1 else '')
return DiffFile(
path=path,
old_content='\n'.join(old_lines),
new_content='\n'.join(new_lines),
hunks=[{
'old_start': h.old_start,
'new_start': h.new_start,
'lines': h.lines
} for h in hunks]
)
def extract_changes_summary(self, diff_file: DiffFile) -> dict:
"""変更の概要を抽出（コンテキスト理解用）"""
additions = 0
deletions = 0
modified_regions = []
for hunk in diff_file.hunks:
for line in hunk['lines']:
if line.startswith('+'):
additions += 1
elif line.startswith('-'):
deletions += 1
modified_regions.append({
'start': hunk['new_start'],
'end': hunk['new_start'] + len([l for l in hunk['lines'] if l.startswith('+')])
})
return {
'path': diff_file.path,
'additions': additions,
'deletions': deletions,
'modified_regions': modified_regions,
'language': self._detect_language(diff_file.path)
}
def _detect_language(self, path: str) -> str:
"""ファイル拡張子から言語を検出"""
ext_map = {
'.py': 'python',
'.js': 'javascript',
'.ts': 'typescript',
'.go': 'go',
'.rs': 'rust',
'.java': 'java',
'.rb': 'ruby',
'.php': 'php'
}
for ext, lang in ext_map.items():
if path.endswith(ext):
return lang
return 'unknown'
ベンチマークテスト
if __name__ == "__main__":
import time
parser = GitDiffParser()
sample_diff = """
diff --git a/src/api/users.py b/src/api/users.py
index abc1234..def5678 100644
--- a/src/api/users.py
+++ b/src/api/users.py
@@ -10,7 +10,10 @@ class UserAPI:
def __init__(self, db):
self.db = db
- def get_user(self, user_id):
- return self.db.query(user_id)
+ async def get_user(self, user_id: int) -> User:
+ user = await self.db.query(user_id)
+ if not user:
+ raise UserNotFoundError(f"User {user_id} not found")
+ return user
def list_users(self, limit=100):
"""
start = time.perf_counter()
for _ in range(1000):
files = parser.parse(sample_diff)
elapsed = time.perf_counter() - start
print(f"パース速度: {1000/elapsed:.0f} files/sec ({elapsed*1000:.2f}ms)")
# 結果: 約45,000 files/sec (22μs/file)同時実行制御とレートリミット対策
私は実際の運用で、複数のPRが同時に届いた際にAPIレートリミットに起因するタイムアウトに苦しみました。以下はHolySheep AIのAPIを活用した堅牢な実装です。
// src/review_orchestrator.py
import asyncio
import time
from typing import List, Optional
from dataclasses import dataclass, field
from collections import deque
import logging
@dataclass
class RateLimitConfig:
"""レートリミット設定"""
requests_per_minute: int = 60
tokens_per_minute: int = 100_000 # 100k TPM
burst_allowance: int = 10
@dataclass
class RequestState:
"""リクエスト状態管理"""
timestamp: float
tokens: int
completed: bool = False
class HolySheepRateLimiter:
"""HolySheep AI API用のレートリミッター（スライディングウィンドウ方式）"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.request_timestamps: deque = deque()
self.token_usage: deque = deque()
self._lock = asyncio.Lock()
self._last_cleanup = time.time()
async def acquire(self, estimated_tokens: int) -> float:
"""トークンを獲得するまでの待機時間（秒）を返す"""
async with self._lock:
now = time.time()
# 60秒以上の古いエントリをクリーンアップ
if now - self._last_cleanup > 60:
self._cleanup_old_entries(now)
self._last_cleanup = now
# トークン制限チェック
current_tokens = sum(t for _, t in self.token_usage)
if current_tokens + estimated_tokens > self.config.tokens_per_minute:
# 最も古いリクエスト完了を待つ
if self.token_usage:
oldest_time, oldest_tokens = self.token_usage[0]
wait_time = max(0, 60 - (now - oldest_time))
logging.info(f"トークン制限により{wait_time:.1f}秒待機")
await asyncio.sleep(wait_time)
self._cleanup_old_entries(time.time())
# リクエスト数制限チェック（バースト対応）
recent_requests = sum(
1 for ts in self.request_timestamps
if now - ts < 1.0
)
if recent_requests >= self.config.burst_allowance:
wait_time = 1.0 - (now - self.request_timestamps[0])
logging.info(f"バーストレート制限により{wait_time:.2f}秒待機")
await asyncio.sleep(max(0, wait_time))
# 状態更新
self.request_timestamps.append(now)
self.token_usage.append((now, estimated_tokens))
return 0.0
def _cleanup_old_entries(self, now: float):
"""古いエントリを削除"""
cutoff = now - 60
while self.request_timestamps and self.request_timestamps[0] < cutoff:
self.request_timestamps.popleft()
while self.token_usage and self.token_usage[0][0] < cutoff:
self.token_usage.popleft()
def get_stats(self) -> dict:
"""現在の使用状況を取得"""
now = time.time()
recent_req = sum(1 for ts in self.request_timestamps if now - ts < 60)
recent_tokens = sum(t for ts, t in self.token_usage if now - ts < 60)
return {
"requests_in_window": recent_req,
"tokens_in_window": recent_tokens,
"requests_remaining": self.config.requests_per_minute - recent_req,
"tokens_remaining": self.config.tokens_per_minute - recent_tokens
}
class ReviewOrchestrator:
"""レビュー要求のオーケストレーター（同時実行制御）"""
def __init__(
self,
api_key: str,
rate_limiter: HolySheepRateLimiter,
max_concurrent: int = 5
):
self.api_key = api_key
self.rate_limiter = rate_limiter
self.semaphore = asyncio.Semaphore(max_concurrent)
self.base_url = "https://api.holysheep.ai/v1"
self.cache: dict[str, str] = {}
async def review_files(
self,
files: List[DiffFile],
review_depth: str = "standard"
) -> List[dict]:
"""複数ファイルを并发レビュー"""
tasks = [
self._review_single(file, review_depth)
for file in files
]
# 悲観的同時実行制御でAPI負荷を分散
results = await asyncio.gather(*tasks, return_exceptions=True)
processed_results = []
for file, result in zip(files, results):
if isinstance(result, Exception):
logging.error(f"レビュー失敗 {file.path}: {result}")
processed_results.append({
"file": file.path,
"status": "error",
"error": str(result)
})
else:
processed_results.append({
"file": file.path,
"status": "success",
"review": result
})
return processed_results
async def _review_single(
self,
file: DiffFile,
depth: str
) -> str:
"""单个ファイルのレビュー（キャッシュ対応）"""
# キャッシュチェック
cache_key = f"{file.file_hash}:{depth}"
if cache_key in self.cache:
logging.info(f"キャッシュヒット: {file.path}")
return self.cache[cache_key]
async with self.semaphore:
# プロンプト構築
prompt = self._build_review_prompt(file, depth)
estimated_tokens = len(prompt) // 4 # 大まかな推定
# レートリミットを適用
wait_time = await self.rate_limiter.acquire(estimated_tokens)
if wait_time > 0:
await asyncio.sleep(wait_time)
# HolySheep API呼び出し
import httpx
async with httpx.AsyncClient(timeout=120.0) as client:
start = time.perf_counter()
response = await client.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1", # 高品質=$8/MTok
"messages": [
{"role": "system", "content": self._get_system_prompt(depth)},
{"role": "user", "content": prompt}
],
"max_tokens": 2048,
"temperature": 0.3
}
)
latency = time.perf_counter() - start
response.raise_for_status()
result = response.json()["choices"][0]["message"]["content"]
# キャッシュに保存（TTL: 1時間）
self.cache[cache_key] = result
logging.info(
f"レビュー完了: {file.path} "
f"(latency={latency*1000:.0f}ms, "
f"tokens={estimated_tokens})"
)
return result
def _build_review_prompt(self, file: DiffFile, depth: str) -> str:
"""レビュープロンプト構築"""
return f"""以下の{depth}な深さでコードレビューしてください。
ファイル: {file.path}
言語: {file.path.split('.')[-1] if '.' in file.path else 'unknown'}
変更内容:
{chr(10).join(file.hunks[0]['lines']) if file.hunks else '新規ファイル'}
レビュー観点:
- 機能的正确性
- セキュリティ脆弱性
- パフォーマンス改善点
- コードスタイルの整合性
- テストカバレッジ
"""
def _get_system_prompt(self, depth: str) -> str:
"""深さに応じたシステムプロンプト"""
base = "あなたは経験丰富的なシニアエンジニア兼セキュリティ専門家です。"
if depth == "quick":
return base + "短時間で主要な問題を指摘してください。"
elif depth == "thorough":
return base + "非常に詳細にレビューし、nitpick级别的な指摘も含めます。"
else:
return base + " Balancedな視点で実践的なフィードバックをしてください。"
ベンチマークテスト
async def benchmark():
"""性能ベンチマーク"""
import httpx
config = RateLimitConfig(
requests_per_minute=60,
tokens_per_minute=100_000,
burst_allowance=10
)
limiter = HolySheepRateLimiter(config)
orchestrator = ReviewOrchestrator(
api_key="YOUR_HOLYSHEEP_API_KEY",
rate_limiter=limiter,
max_concurrent=3
)
# ダミーファイル生成
test_files