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

引言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在项目根目录创建.envANTHROPIC_API_KEYsk-ant-api03-your-key-here注意Agent SDK 也支持 AWS Bedrock、Google Vertex AI、Azure AI Foundry。生产环境建议走 BedrockAPI Key 只在本地开发用。三、第一个Agent5分钟跑通自动修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.pyimport asyncio from claude_agent_sdk import ( query, ClaudeAgentOptions, AssistantMessage, ResultMessage ) async def main(): async for message in query( prompt审查 utils.py找出所有会导致崩溃的Bug并修复, optionsClaudeAgentOptions( allowed_tools[Read, Edit, Glob], permission_modeacceptEdits, # 自动批准文件编辑 ), ): 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.pyAgent 会自动 1.Readutils.py— 理解代码逻辑 2.分析— 识别空列表和 None 的边界情况 3.Edit— 添加if not numbers: return 0和if user is None: return Unknown4. 返回结果这就是 Agent SDK 和传统 API 的本质区别Claude 直接操作文件系统你只管定义目标。四、构建代码审查Agent从玩具到工具上面的例子修了两个 Bug但真正的代码审查要复杂得多。我们来构建一个实用的审查 Agent。4.1 审查标准配置创建review_config.pyREVIEW_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( promptprompt, optionsClaudeAgentOptions( allowed_tools[Read, Glob, Grep], system_prompt你是资深Python代码审查专家。只输出JSON格式结果不做任何修改。, permission_modeacceptEdits, max_turns5, # 限制最多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 / fcode_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, encodingutf-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( promptprompt, optionsClaudeAgentOptions( allowed_tools[Read, Edit, Glob, Grep, Bash], permission_modeacceptEdits, max_turns10, ), ): # 处理消息... pass这里的关键是max_turns10— Agent SDK 支持多轮自主循环Agent 修复 → 跑测试 → 失败 → 再修复 → 再测试直到通过或达到轮次上限。五、接入 MCP给 Agent 装上手和脚截至这一步Agent 还只能在本地文件系统里打转。MCPModel Context Protocol让它能操作外部系统——GitHub、数据库、Slack、Jira。5.1 配置 MCP Server在ClaudeAgentOptions中声明 MCP Serveroptions ClaudeAgentOptions( allowed_tools[Read, Edit, Glob, Grep, Bash], permission_modeacceptEdits, 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( fhttps://erp.internal/api/orders/{order_id}, headers{Authorization: Bearer xxx}, timeout10, ) 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}, timeout10, ) return resp.json() server.run()配置到 Agent SDKmcp_servers{ erp: { command: python, args: [erp_mcp_server.py], }, }现在 Agent 可以做到审查order_service.py如果发现有SQL注入风险查询ERP验证是否影响线上订单然后在工单系统创建修复任务给 zhangsanAgent 会Read 代码 → 发现 SQL 注入 → 调用query_order验证影响范围 → 调用create_issue创建工单 → Edit 修复代码。一个目标四个工具自动协同。六、Hooks在关键时刻拦截 AgentHooks 让你在 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_modeacceptEdits, hooks{ pre_tool_call: pre_tool_hook, post_tool_call: post_tool_hook, }, )Hooks 的本质是信任但验证——你相信 Agent 能做对但在关键路径上设卡。七、多Agent协作审查员 修复员 测试员单个 Agent 审查大项目会超时。参考现实世界的 Code Review 流程拆成三个专职 Agentimport asyncio async def pipeline(file_path: str): 三阶段流水线审查 → 修复 → 验证 # Agent 1: 审查员只读模式 issues [] async for message in query( promptf审查 {file_path}列出所有问题不修改代码, optionsClaudeAgentOptions( allowed_tools[Read, Grep], permission_modeacceptEdits, 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( promptf根据以下审查意见修复 {file_path}\n\n \n.join(issues), optionsClaudeAgentOptions( allowed_tools[Read, Edit], permission_modeacceptEdits, system_prompt你只负责修复代码按审查意见逐一修改。, ), ): pass # 静默修复 # Agent 3: 测试员 async for message in query( promptf运行 {file_path} 的测试套件报告结果, optionsClaudeAgentOptions( allowed_tools[Bash, Read], permission_modeacceptEdits, ), ): 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, recursiveTrue) # 逐个文件走流水线 for f in py_files: await pipeline(f) asyncio.run(main())三个 Agent 各司其职比一个大而全的 Agent 更可控、更快。system_prompt约束每个 Agent 的行为边界避免审查员忍不住改代码、修复员忍不住写测试。八、生产部署Docker CI/CDAgent 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/checkoutv4 - 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-scriptv7 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 });每次提 PRClaude Agent 自动审查、修复、跑测试然后把报告贴在 PR 评论区。不用任何人手动触发。九、常见坑与最佳实践9.1 API Thinking 模式兼容性如果遇到thinking.type.enabled is not supported错误说明 SDK 版本过旧。Agent SDK v0.2.111 才支持 Claude Opus 4.7 的thinking.type.adaptivepip install --upgrade claude-agent-sdk9.2 成本控制Agent SDK 的 Agentic Loop 比单次 API 调用消耗更多 Token。三条省钱法则options ClaudeAgentOptions( max_turns8, # 限制最大轮次防止死循环 modelclaude-sonnet-4-6, # 非关键任务用 Sonnet审查才用 Opus permission_modeacceptEdits, )任务类型推荐模型预估成本代码审查单文件Sonnet 4.6~$0.05复杂重构Opus 4.7~$0.20自动修复测试Sonnet 4.6修 Sonnet 4.6测~$0.159.3 安全边界permission_modebypassPermissions只在 Docker 沙箱内使用。本地开发必须用acceptEdits或default配合 Hooks 拦截危险操作。# 生产环境的安全配置 options ClaudeAgentOptions( allowed_tools[Read, Edit, Glob, Grep], # 禁止 Bash permission_modeacceptEdits, hooks{pre_tool_call: security_hook}, # 审计拦截 max_turns10, )总结Claude Agent SDK 代表了 2026 年 AI 开发的范式转移从我调API → AI返回文本 → 我解析 → 我执行变成了我给目标 → AI自己读/写/跑/验证。本文从 5 分钟 Quickstart 到多 Agent 协作流水线覆盖了 Agent SDK 的核心链路基础query()ClaudeAgentOptions→ 第一个自动修Bug的 Agent实战CodeReviewAgent类 → 项目级代码审查输出 Markdown 报告扩展MCP Server → 让 Agent 操作 GitHub、ERP、数据库安全Hooks → 审计日志 危险操作拦截生产Docker GitHub Actions → 每次 PR 自动触发 AI Code ReviewAgent 不会取代开发者但会用 Agent SDK 的开发者正在取代不用的人。本文代码基于 claude-agent-sdk v0.2.111Python 3.12测试于 2026-05-21。