当前位置：首页 > news >正文

手写一个AI代码审查员：Claude Agent SDK + MCP 深度实战

news 2026/5/22 0:47:01

引言

2026年5月，Anthropic做了一件意味深长的事：把 Claude Code SDK 改名为Claude Agent SDK。

改名背后是一个判断——这不再是"帮你写代码的工具"，而是一个能自主读代码、分析逻辑、修改文件、跑测试、甚至提PR的AI Agent编排框架。Python 和 TypeScript 双语言支持，原生集成 MCP 协议，Docker 一键部署。

本文从 Quickstart 开始，逐步构建一个完整的代码审查Agent：自动扫描代码 → 发现Bug → 修复 → 跑测试 → 生成审查报告。最后接入 MCP，让Agent直接操作 GitHub Issue 和数据库。

一、为什么是 Claude Agent SDK，不是直接调 API？

很多人第一反应："我直接用requests调 Claude API 不就完了？"

两种范式的区别：

方式	你做什么	Claude做什么
直接调 API	写 Prompt、解析返回、自己读文件、自己改代码、自己跑测试	只生成文本
Agent SDK	给一个目标	自己读文件、自己判断改哪里、自己编辑、自己验证

Agent SDK 的核心是Agentic Loop：

你给目标 → Claude 自主选择工具(Read/Edit/Bash) → 观察结果 → 决定下一步 → 直到目标达成

这意味着你可以把整个代码审查流程交给它，而不需要写一堆 if-else 来编排。下面直接上代码。

二、环境准备

# 创建项目 mkdir code-review-agent && cd code-review-agent # Python 虚拟环境 python3 -m venv .venv && source .venv/bin/activate # Windows: .venv\Scripts\activate # 安装 SDK pip install claude-agent-sdk

在项目根目录创建.env：

ANTHROPIC_API_KEY=sk-ant-api03-your-key-here

注意：Agent SDK 也支持 AWS Bedrock、Google Vertex AI、Azure AI Foundry。生产环境建议走 Bedrock，API Key 只在本地开发用。

三、第一个Agent：5分钟跑通自动修Bug

创建一个有 Bug 的文件utils.py：

# utils.py — 故意埋了两个Bug def calculate_average(numbers): total = 0 for num in numbers: total += num return total / len(numbers) # Bug 1: 空列表会 ZeroDivisionError def get_user_name(user): return user["name"].upper() # Bug 2: None 会 TypeError

创建 Agentagent.py：

import asyncio from claude_agent_sdk import ( query, ClaudeAgentOptions, AssistantMessage, ResultMessage ) async def main(): async for message in query( prompt="审查 utils.py，找出所有会导致崩溃的Bug并修复", options=ClaudeAgentOptions( allowed_tools=["Read", "Edit", "Glob"], permission_mode="acceptEdits", # 自动批准文件编辑 ), ): if isinstance(message, AssistantMessage): for block in message.content: if hasattr(block, "text"): print(block.text) elif hasattr(block, "name"): print(f"[Tool] {block.name}") elif isinstance(message, ResultMessage): print(f"[Done] {message.subtype}") asyncio.run(main())

运行：

python agent.py

Agent 会自动： 1.Readutils.py— 理解代码逻辑 2.分析— 识别空列表和 None 的边界情况 3.Edit— 添加if not numbers: return 0和if user is None: return "Unknown"4. 返回结果

这就是 Agent SDK 和传统 API 的本质区别：Claude 直接操作文件系统，你只管定义目标。

四、构建代码审查Agent：从玩具到工具

上面的例子修了两个 Bug，但真正的代码审查要复杂得多。我们来构建一个实用的审查 Agent。

4.1 审查标准配置

创建review_config.py：

REVIEW_RULES = """ ## 审查标准（按优先级） 1. **安全漏洞**：SQL注入、XSS、命令注入、硬编码密钥 2. **空指针/None引用**：所有外部输入必须做None检查 3. **资源泄露**：文件句柄、数据库连接、网络socket未关闭 4. **并发安全**：共享状态是否加锁、是否有竞态条件 5. **错误处理**：异常是否被静默吞掉 6. **类型安全**：关键函数是否有类型标注 """

4.2 审查Agent核心代码

import asyncio import json from datetime import datetime from pathlib import Path from claude_agent_sdk import ( query, ClaudeAgentOptions, AssistantMessage, ResultMessage ) from review_config import REVIEW_RULES class CodeReviewAgent: def __init__(self, project_path: str): self.project_path = Path(project_path) self.report = [] async def review_file(self, file_path: str): """审查单个文件""" full_path = self.project_path / file_path if not full_path.exists(): print(f"文件不存在: {full_path}") return prompt = f"""审查文件 {file_path}，按以下标准逐项检查： {REVIEW_RULES} 输出格式（JSON）： {{ "file": "{file_path}", "issues": [ {{ "severity": "critical|high|medium|low", "line_range": "10-15", "category": "安全漏洞|空指针|资源泄露|并发安全|错误处理|类型安全", "description": "问题描述", "fix_suggestion": "修复建议（含代码示例）" }} ], "summary": "整体评价，50字以内" }} 如果文件没有问题，issues为空数组。 """ async for message in query( prompt=prompt, options=ClaudeAgentOptions( allowed_tools=["Read", "Glob", "Grep"], system_prompt="你是资深Python代码审查专家。只输出JSON格式结果，不做任何修改。", permission_mode="acceptEdits", max_turns=5, # 限制最多5轮交互，防止审查过长 ), ): if isinstance(message, AssistantMessage): for block in message.content: if hasattr(block, "text"): # 尝试解析JSON结果 text = block.text try: result = json.loads(text) self.report.append(result) except json.JSONDecodeError: print(text) # 打印思考过程 elif isinstance(message, ResultMessage): print(f"审查完成: {file_path}") async def review_project(self, glob_pattern: str = "**/*.py"): """审查整个项目""" py_files = list(self.project_path.glob(glob_pattern)) # 排除测试文件和虚拟环境 py_files = [ f for f in py_files if ".venv" not in str(f) and "test_" not in f.name ] print(f"发现 {len(py_files)} 个Python文件，开始审查...") for py_file in py_files: rel_path = py_file.relative_to(self.project_path) await self.review_file(str(rel_path)) self.generate_report() def generate_report(self): """生成审查报告""" report_path = self.project_path / f"code_review_{datetime.now():%Y%m%d_%H%M%S}.md" total_issues = sum(len(r.get("issues", [])) for r in self.report) critical = sum( 1 for r in self.report for i in r.get("issues", []) if i["severity"] == "critical" ) lines = [ f"# 代码审查报告", f"审查时间：{datetime.now():%Y-%m-%d %H:%M}", f"审查文件数：{len(self.report)}", f"发现问题数：{total_issues}（严重：{critical}）", "", "## 问题详情", "", ] for file_report in self.report: issues = file_report.get("issues", []) if not issues: continue lines.append(f"### {file_report['file']}") for issue in issues: icon = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🟢"} lines.append( f"- {icon.get(issue['severity'], '⚪')} " f"[{issue['category']}] {issue['description']} " f"（行 {issue['line_range']}）" ) lines.append(f" - 修复建议：{issue['fix_suggestion']}") lines.append("") report_content = "\n".join(lines) report_path.write_text(report_content, encoding="utf-8") print(f"报告已保存: {report_path}") print(report_content) async def main(): agent = CodeReviewAgent("./my_project") await agent.review_project() asyncio.run(main())

4.3 进阶：让Agent自动修复+跑测试验证

审查只读，修复才创造价值。把permission_mode调成"acceptEdits"并加上Bash工具：

async def review_and_fix(self, file_path: str): """审查并自动修复，修复后跑测试验证""" prompt = f"""对 {file_path} 执行以下流程： 1. 审查代码，找出所有Bug和安全问题 2. 逐一修复 3. 运行 pytest {file_path.replace('.py', '_test.py')} 验证修复 4. 如果测试不通过，继续修复直到通过 """ async for message in query( prompt=prompt, options=ClaudeAgentOptions( allowed_tools=["Read", "Edit", "Glob", "Grep", "Bash"], permission_mode="acceptEdits", max_turns=10, ), ): # 处理消息... pass

这里的关键是max_turns=10— Agent SDK 支持多轮自主循环，Agent 修复 → 跑测试 → 失败 → 再修复 → 再测试，直到通过或达到轮次上限。

五、接入 MCP：给 Agent 装上手和脚

截至这一步，Agent 还只能在本地文件系统里打转。MCP（Model Context Protocol）让它能操作外部系统——GitHub、数据库、Slack、Jira。

5.1 配置 MCP Server

在ClaudeAgentOptions中声明 MCP Server：

options = ClaudeAgentOptions( allowed_tools=["Read", "Edit", "Glob", "Grep", "Bash"], permission_mode="acceptEdits", mcp_servers={ "github": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-github"], "env": {"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxx"} }, "postgres": { "command": "python", "args": ["-m", "mcp_server_postgres"], "env": {"DATABASE_URL": "postgresql://localhost:5432/mydb"} }, }, )

5.2 写一个自定义 MCP Server

官方的 GitHub MCP Server 很好，但企业内部系统（ERP、工单、配置中心）需要自己写。10 行代码搞定：

# erp_mcp_server.py from mcp.server import Server import requests server = Server("erp-tools") @server.tool() async def query_order(order_id: str) -> dict: """查询ERP系统订单详情""" resp = requests.get( f"https://erp.internal/api/orders/{order_id}", headers={"Authorization": "Bearer xxx"}, timeout=10, ) return resp.json() @server.tool() async def create_issue(title: str, assignee: str, description: str) -> dict: """在内部工单系统创建Issue""" resp = requests.post( "https://ticket.internal/api/issues", json={"title": title, "assignee": assignee, "description": description}, timeout=10, ) return resp.json() server.run()

配置到 Agent SDK：

mcp_servers={ "erp": { "command": "python", "args": ["erp_mcp_server.py"], }, }

现在 Agent 可以做到：

"审查order_service.py，如果发现有SQL注入风险，查询ERP验证是否影响线上订单，然后在工单系统创建修复任务给 @zhangsan"

Agent 会：Read 代码 → 发现 SQL 注入 → 调用query_order验证影响范围 → 调用create_issue创建工单 → Edit 修复代码。一个目标，四个工具，自动协同。

六、Hooks：在关键时刻拦截 Agent

Hooks 让你在 Agent 调用工具前后插入自定义逻辑。最实用的三个场景：

from claude_agent_sdk import ClaudeAgentOptions, HookContext async def pre_tool_hook(ctx: HookContext): """工具调用前：审计日志 + 敏感操作拦截""" print(f"[AUDIT] {ctx.tool_name} → {ctx.tool_input}") # 拦截危险命令 if ctx.tool_name == "Bash": dangerous = ["rm -rf", "DROP TABLE", "ALTER TABLE", "shutdown"] cmd = str(ctx.tool_input).lower() if any(d.lower() in cmd for d in dangerous): return {"blocked": True, "reason": "危险命令已拦截"} return {"blocked": False} async def post_tool_hook(ctx: HookContext): """工具调用后：记录耗时 + 结果校验""" print(f"[DONE] {ctx.tool_name} → 耗时 {ctx.duration_ms}ms") # 如果 Edit 的结果文件为空，回滚 if ctx.tool_name == "Edit" and ctx.result: file_path = ctx.tool_input.get("file_path", "") if file_path and not open(file_path).read().strip(): print(f"警告：{file_path} 编辑后为空文件！") options = ClaudeAgentOptions( allowed_tools=["Read", "Edit", "Bash"], permission_mode="acceptEdits", hooks={ "pre_tool_call": pre_tool_hook, "post_tool_call": post_tool_hook, }, )

Hooks 的本质是信任但验证——你相信 Agent 能做对，但在关键路径上设卡。

七、多Agent协作：审查员 + 修复员 + 测试员

单个 Agent 审查大项目会超时。参考现实世界的 Code Review 流程，拆成三个专职 Agent：

import asyncio async def pipeline(file_path: str): """三阶段流水线：审查 → 修复 → 验证""" # Agent 1: 审查员（只读模式） issues = [] async for message in query( prompt=f"审查 {file_path}，列出所有问题（不修改代码）", options=ClaudeAgentOptions( allowed_tools=["Read", "Grep"], permission_mode="acceptEdits", system_prompt="你只负责发现和描述问题，不做任何代码修改。", ), ): if isinstance(message, AssistantMessage): for block in message.content: if hasattr(block, "text"): issues.append(block.text) if not issues: print(f"{file_path}: 未发现问题") return # Agent 2: 修复员 async for message in query( prompt=f"根据以下审查意见修复 {file_path}：\n\n" + "\n".join(issues), options=ClaudeAgentOptions( allowed_tools=["Read", "Edit"], permission_mode="acceptEdits", system_prompt="你只负责修复代码，按审查意见逐一修改。", ), ): pass # 静默修复 # Agent 3: 测试员 async for message in query( prompt=f"运行 {file_path} 的测试套件，报告结果", options=ClaudeAgentOptions( allowed_tools=["Bash", "Read"], permission_mode="acceptEdits", ), ): if isinstance(message, AssistantMessage): for block in message.content: if hasattr(block, "text"): print(f"[测试] {block.text}") async def main(): import glob py_files = glob.glob("src/**/*.py", recursive=True) # 逐个文件走流水线 for f in py_files: await pipeline(f) asyncio.run(main())

三个 Agent 各司其职，比一个大而全的 Agent 更可控、更快。system_prompt约束每个 Agent 的行为边界，避免审查员忍不住改代码、修复员忍不住写测试。

八、生产部署：Docker + CI/CD

Agent SDK 支持一键 Docker 化。在.claude/目录下创建部署配置：

# Dockerfile FROM python:3.12-slim RUN pip install claude-agent-sdk COPY . /app WORKDIR /app ENV ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY} CMD ["python", "agent.py"]

GitHub Actions 集成 —— 每次 PR 自动触发审查：

# .github/workflows/ai-review.yml name: AI Code Review on: pull_request: types: [opened, synchronize] jobs: review: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Run Claude Review Agent env: ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} run: | pip install claude-agent-sdk python agent.py - name: Post Review Comment uses: actions/github-script@v7 with: script: | const fs = require('fs'); const report = fs.readFileSync('code_review_report.md', 'utf8'); github.rest.issues.createComment({ ...context.repo, issue_number: context.issue.number, body: report });

每次提 PR，Claude Agent 自动审查、修复、跑测试，然后把报告贴在 PR 评论区。不用任何人手动触发。

九、常见坑与最佳实践

9.1 API Thinking 模式兼容性

如果遇到thinking.type.enabled is not supported错误，说明 SDK 版本过旧。Agent SDK v0.2.111+ 才支持 Claude Opus 4.7 的thinking.type.adaptive：

pip install --upgrade claude-agent-sdk

9.2 成本控制

Agent SDK 的 Agentic Loop 比单次 API 调用消耗更多 Token。三条省钱法则：

options = ClaudeAgentOptions( max_turns=8, # 限制最大轮次，防止死循环 model="claude-sonnet-4-6", # 非关键任务用 Sonnet，审查才用 Opus permission_mode="acceptEdits", )

任务类型	推荐模型	预估成本
代码审查（单文件）	Sonnet 4.6	~$0.05
复杂重构	Opus 4.7	~$0.20
自动修复+测试	Sonnet 4.6（修） + Sonnet 4.6（测）	~$0.15

9.3 安全边界

permission_mode="bypassPermissions"只在 Docker 沙箱内使用。本地开发必须用"acceptEdits"或"default"，配合 Hooks 拦截危险操作。

# 生产环境的安全配置 options = ClaudeAgentOptions( allowed_tools=["Read", "Edit", "Glob", "Grep"], # 禁止 Bash permission_mode="acceptEdits", hooks={"pre_tool_call": security_hook}, # 审计+拦截 max_turns=10, )