วันนี้ผมจะมาแชร์ประสบการณ์จริงในการสร้างระบบ Code Review อัตโนมัติด้วย MCP Server และ GitHub API ซึ่งเป็นโปรเจกต์ที่ทีมของผมพัฒนาขึ้นมาเพื่อลดภาระงานการตรวจโค้ดแบบ Manual ลงอย่างมาก
เริ่มต้น: ปัญหาที่ทำให้เราต้องหาทางออก
ตอนแรกทีมของผมมีปัญหาหลายอย่างกับการ Review Code:
- PR ค้างนานเพราะ Senior Developer ไม่มีเวลาตรวจทุกตัว
- Bug ที่ตรวจจับได้ง่ายก็ยังหลุดเข้า Production อยู่เรื่อยๆ
- การตรวจ Code ใช้เวลาวันละ 2-3 ชั่วโมง ซึ่งเป็นเวลาที่เสียไปโดยไม่จำเป็น
ผมลองใช้วิธีต่างๆ จนกระทั่งเจอปัญหานี้ตอนที่พยายามเชื่อมต่อ GitHub API โดยตรง:
ConnectionError: timeout occurred while connecting to github.com
HTTPSConnectionPool(host='api.github.com', port=443):
Max retries exceeded with url: /repos/owner/repo/pulls (Caused by
ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at
0x7f...>, 'Connection timed out after 30 seconds'))หลังจากแก้ปัญหานี้ได้ ผมตัดสินใจสร้าง MCP Server ขึ้นมาเพื่อจัดการเรื่องนี้ทั้งหมด โดยใช้ HolySheep AI เป็น Engine หลักในการวิเคราะห์โค้ด ซึ่งมีความเร็วตอบสนองน้อยกว่า 50ms และราคาถูกกว่าวิธีอื่นมาก
MCP Server คืออะไร และทำไมต้องใช้?
MCP (Model Context Protocol) Server คือ Bridge ที่เชื่อมต่อระหว่าง AI Model กับ External Tools ต่างๆ อย่าง GitHub API ให้สามารถทำงานร่วมกันได้อย่างมีประสิทธิภาพ ซึ่งช่วยให้เราสามารถ:
- ดึง Pull Request จาก GitHub มาวิเคราะห์โดยอัตโนมัติ
- ให้ AI ตรวจสอบ Code Quality, Security Issues และ Best Practices
- แสดงผล Review กลับไปที่ GitHub PR โดยตรง
การติดตั้ง MCP Server
ก่อนอื่นให้เราสร้าง Project และติดตั้ง Dependencies กันก่อน:
# สร้างโฟลเดอร์สำหรับโปรเจกต์
mkdir mcp-github-reviewer
cd mcp-github-reviewer
สร้าง Virtual Environment
python -m venv venv
source venv/bin/activate # สำหรับ Linux/Mac
หรือ venv\Scripts\activate # สำหรับ Windows
ติดตั้ง Dependencies
pip install fastapi uvicorn httpx python-dotenv
pip install mcp PyJWTสร้าง GitHub API Integration
ต่อไปเราจะสร้าง Module สำหรับเชื่อมต่อกับ GitHub API:
import httpx
import os
from typing import List, Dict, Optional
class GitHubAPIClient:
def __init__(self, token: str):
self.token = token
self.base_url = "https://api.github.com"
self.headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github.v3+json",
"X-GitHub-Api-Version": "2022-11-28"
}
async def get_pull_request(self, owner: str, repo: str, pr_number: int) -> Dict:
"""ดึงข้อมูล Pull Request"""
async with httpx.AsyncClient(timeout=30.0) as client:
url = f"{self.base_url}/repos/{owner}/{repo}/pulls/{pr_number}"
response = await client.get(url, headers=self.headers)
if response.status_code == 401:
raise Exception("401 Unauthorized: GitHub Token ไม่ถูกต้องหรือหมดอายุ")
elif response.status_code == 404:
raise Exception(f"404 Not Found: PR #{pr_number} ไม่พบใน Repository")
response.raise_for_status()
return response.json()
async def get_pr_files(self, owner: str, repo: str, pr_number: int) -> List[Dict]:
"""ดึงไฟล์ที่ถูกแก้ไขใน Pull Request"""
async with httpx.AsyncClient(timeout=60.0) as client:
url = f"{self.base_url}/repos/{owner}/{repo}/pulls/{pr_number}/files"
response = await client.get(url, headers=self.headers)
response.raise_for_status()
return response.json()
async def get_file_content(self, owner: str, repo: str, path: str, ref: str) -> str:
"""ดึงเนื้อหาของไฟล์"""
async with httpx.AsyncClient(timeout=30.0) as client:
url = f"{self.base_url}/repos/{owner}/{repo}/contents/{path}"
params = {"ref": ref}
response = await client.get(url, headers=self.headers, params=params)
response.raise_for_status()
import base64
data = response.json()
return base64.b64decode(data["content"]).decode("utf-8")
async def create_review_comment(self, owner: str, repo: str, pr_number: int,
body: str, commit_id: str, path: str,
line: Optional[int] = None) -> Dict:
"""สร้าง Comment บน Pull Request"""
async with httpx.AsyncClient(timeout=30.0) as client:
url = f"{self.base_url}/repos/{owner}/{repo}/pulls/{pr_number}/comments"
comment_data = {
"body": body,
"commit_id": commit_id,
"path": path
}
if line:
comment_data["line"] = line
comment_data["side"] = "RIGHT"
response = await client.post(url, headers=self.headers, json=comment_data)
response.raise_for_status()
return response.json()สร้าง MCP Server สำหรับ Code Review
ต่อไปจะเป็นหัวใจหลักของระบบ นั่นคือ MCP Server ที่ใช้ HolySheep AI ในการวิเคราะห์โค้ด:
import httpx
from mcp.server import Server
from mcp.types import Tool, CallToolResult
from pydantic import AnyUrl
import json
ตั้งค่า HolySheep AI
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("YOUR_HOLYSHEEP_API_KEY")
server = Server("github-code-reviewer")
github_client = GitHubAPIClient(os.getenv("GITHUB_TOKEN"))
@server.list_tools()
async def list_tools() -> List[Tool]:
"""ประกาศ Tools ที่ MCP Server รองรับ"""
return [
Tool(
name="review_pull_request",
description="วิเคราะห์ Pull Request และให้คำแนะนำ",
inputSchema={
"type": "object",
"properties": {
"owner": {"type": "string", "description": "ชื่อ Owner ของ Repository"},
"repo": {"type": "string", "description": "ชื่อ Repository"},
"pr_number": {"type": "integer", "description": "หมายเลข Pull Request"}
},
"required": ["owner", "repo", "pr_number"]
}
),
Tool(
name="get_pr_diff",
description="ดึงการเปลี่ยนแปลงของ Pull Request",
inputSchema={
"type": "object",
"properties": {
"owner": {"type": "string"},
"repo": {"type": "string"},
"pr_number": {"type": "integer"}
},
"required": ["owner", "repo", "pr_number"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
"""ประมวลผล Tool Calls"""
if name == "review_pull_request":
return await review_pull_request(
arguments["owner"],
arguments["repo"],
arguments["pr_number"]
)
elif name == "get_pr_diff":
return await get_pr_diff(
arguments["owner"],
arguments["repo"],
arguments["pr_number"]
)
raise ValueError(f"Unknown tool: {name}")
async def review_pull_request(owner: str, repo: str, pr_number: int) -> CallToolResult:
"""วิเคราะห์ Pull Request ด้วย HolySheep AI"""
# ดึงข้อมูล PR และไฟล์ที่เปลี่ยนแปลง
pr_data = await github_client.get_pull_request(owner, repo, pr_number)
pr_files = await github_client.get_pr_files(owner, repo, pr_number)
# รวบรวมข้อมูลสำหรับวิเคราะห์
files_content = []
for file in pr_files[:10]: # จำกัด 10 ไฟล์แรกเพื่อประหยัด Cost
files_content.append({
"filename": file["filename"],
"status": file["status"],
"additions": file["additions"],
"deletions": file["deletions"],
"patch": file.get("patch", "")
})
# ส่งไปวิเคราะห์ที่ HolySheep AI
prompt = f"""คุณคือ Senior Code Reviewer ที่มีประสบการณ์มากกว่า 10 ปี
จงวิเคราะห์ Pull Request นี้และให้คำแนะนำ:
Title: {pr_data['title']}
Description: {pr_data.get('body', 'No description')}
Files Changed:
{json.dumps(files_content, indent=2, ensure_ascii=False)}
ให้คะแนนในแต่ละหมวด:
1. Code Quality (1-10)
2. Security Issues (1-10, สูง = ปลอดภัย)
3. Performance Impact (1-10)
4. Overall Recommendation (Approve/Request Changes/Comment)
พร้อมรายละเอียดปัญหาที่พบ (ถ้ามี) ในรูปแบบ:
- [file] บรรทัดที่ X: ปัญหา - คำแนะนำ
"""
try:
async with httpx.AsyncClient(timeout=120.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": "gpt-4.1",
"messages": [
{"role": "system", "content": "You are an expert code reviewer."},
{"role": "user", "content": prompt}
],
"temperature": 0.3
}
)
if response.status_code == 401:
return CallToolResult(
content=[{"type": "text", "text": "❌ ข้อผิดพลาด: HolySheep API Key ไม่ถูกต้อง"}],
isError=True
)
response.raise_for_status()
result = response.json()
review_text = result["choices"][0]["message"]["content"]
return CallToolResult(
content=[{"type": "text", "text": review_text}]
)
except httpx.TimeoutException:
return CallToolResult(
content=[{"type": "text", "text": "❌ ConnectionError: timeout - HolySheheep AI ไม่ตอบสนองภายใน 120 วินาที"}],
isError=True
)
async def get_pr_diff(owner: str, repo: str, pr_number: int) -> CallToolResult:
"""ดึง Diff ของ Pull Request"""
pr_files = await github_client.get_pr_files(owner, repo, pr_number)
diff_text = f"## Pull Request #{pr_number}\n\n"
diff_text += f"Total Files: {len(pr_files)}\n\n"
for file in pr_files:
diff_text += f"### {file['filename']} ({file['status']})\n"
diff_text += f"Additions: +{file['additions']} | Deletions: -{file['deletions']}\n\n"
if file.get('patch'):
diff_text += f"``diff\n{file['patch']}\n``\n\n"
return CallToolResult(content=[{"type": "text", "text": diff_text}])สร้าง FastAPI Server สำหรับ Webhook
เพื่อให้ระบบทำงานอัตโนมัติเมื่อมี PR ใหม่ เราจะสร้าง Webhook Endpoint:
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
import hmac
import hashlib
app = FastAPI(title="GitHub Code Reviewer API")
class ReviewRequest(BaseModel):
owner: str
repo: str
pr_number: int
@app.post("/webhook/github")
async def github_webhook(request: Request):
"""Webhook สำหรับรับ Events จาก GitHub"""
# ตรวจสอบ GitHub Webhook Signature
signature = request.headers.get("X-Hub-Signature-256")
if signature:
body = await request.body()
secret = os.getenv("GITHUB_WEBHOOK_SECRET", "").encode()
expected_signature = "sha256=" + hmac.new(
secret, body, hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected_signature):
raise HTTPException(status_code=401, detail="Invalid signature")
# ดึงข้อมูล Event
event = request.headers.get("X-GitHub-Event", "ping")
payload = await request.json()
if event == "pull_request":
action = payload.get("action")
pr = payload.get("pull_request")
if action in ["opened", "reopened", "synchronize"]:
# Trigger Review อัตโนมัติ
from mcp.server.stdio import create_server
import asyncio
# เรียก MCP Server เพื่อทำ Review
result = await review_pull_request(
payload["repository"]["owner"]["login"],
payload["repository"]["name"],
pr["number"]
)
# ส่ง Review กลับไปที่ GitHub
for content in result.content:
if hasattr(content, 'text'):
await github_client.create_review_comment(
owner=payload["repository"]["owner"]["login"],
repo=payload["repository"]["name"],
pr_number=pr["number"],
body=content.text,
commit_id=pr["head"]["sha"],
path="", # General comment
line=None
)
return {"status": "review_completed", "pr": pr["number"]}