如何使用 Claude 构建你的第一个 AI Agent：从首次 API 调用到自主系统

1while True:
2    user_input = input("> ")
3    response = call_claude(user_input)
4    print(response)

1import anthropic
2
3client = anthropic.Anthropic()
4
5response = client.messages.create(
6    model="claude-sonnet-4-6",
7    max_tokens=4096,
8    system="""你是一个代码审查 Agent。
9
10规则：
11- 阅读整个 diff 后再评论
12- 只标记真正的 bug，而不是风格偏好
13- 如果没有任何问题，回复 "LGTM" 并停止
14- 永远不要建议你没有在脑子里测试过的更改
15- 输出格式：{file, line, issue, fix} 的 JSON 数组""",
16    messages=[{"role": "user", "content": diff_content}]
17)

1# 强制 JSON 输出，告诉 Claude 确切的结构
2system = """只返回有效的 JSON。不要 Markdown。不要解释。
3Schema:
4{
5  "status": "pass" | "fail",
6  "issues": [{"file": str, "line": int, "issue": str}],
7  "summary": str
8}"""

1tools = [{
2    "name": "run_sql",
3    "description": "对数据库执行只读 SQL 查询",
4    "input_schema": {
5        "type": "object",
6        "properties": {
7            "query": {
8                "type": "string",
9                "description": "要执行的 SQL SELECT 查询"
10            }
11        },
12        "required": ["query"]
13    }
14},
15{
16    "name": "write_file",
17    "description": "将内容写入磁盘上的文件",
18    "input_schema": {
19        "type": "object",
20        "properties": {
21            "path": {"type": "string"},
22            "content": {"type": "string"}
23        },
24        "required": ["path", "content"]
25    }
26}]

1import json
2from pathlib import Path
3
4def run_agent(task: str, max_attempts: int = 5):
5    state = {"task": task, "attempts": [], "done": False}
6
7    for i in range(max_attempts):
8        # 从状态构建上下文
9        context = build_prompt(state)
10
11        # 使用工具调用 Claude
12        result = call_claude(context, tools)
13
14        # 执行任何工具调用
15        output = execute_tools(result)
16
17        # 验证结果
18        check = verify(output)
19
20        # 更新状态
21        state["attempts"].append({
22            "attempt": i + 1,
23            "action": result.summary,
24            "passed": check.passed,
25            "reason": check.reason
26        })
27
28        if check.passed:
29            state["done"] = True
30            break
31
32    # 为下一次运行保存状态
33    Path("state.json").write_text(json.dumps(state, indent=2))
34    return state

1# CLAUDE.md
2
3## 项目
4任务管理 API。Python 3.12，FastAPI，PostgreSQL。
5
6## 规则
7- 所有响应：{data, error, meta} 模式
8- 每个新端点都需要测试
9- 提交信息：type(scope): description
10- 永远不要使用 print() 记录日志。使用 structlog。
11
12## 已知问题
13- Auth 中间件期望 x-auth-token，而不是 Authorization
14- 测试套件全量运行需要 45 秒。迭代时使用 --filter。

1# learnings.md
2
3- 支付 API 期望幂等键在 header 中，而不是 body
4- PostgreSQL NOTIFY 需要在连接池中显式 LISTEN
5- 速率限制器按 key 计数，而不是按 IP。测试需要唯一的 key。

1def verify(output):
2    # 运行测试套件
3    result = subprocess.run(
4        ["pytest", "tests/", "-x", "--tb=short"],
5        capture_output=True, text=True
6    )
7    return {
8        "passed": result.returncode == 0,
9        "reason": result.stdout if result.returncode != 0 else "所有测试通过"
10    }

1# 审查者提示词 - 与构建者分离
2reviewer_system = """你是一个严格的代码审查者。
3你唯一的任务是发现问题。
4
5检查：
6- 代码是否符合规范？
7- 是否有未捕获的边界情况？
8- 所有测试是否真的测试了正确的东西？
9
10如果一切正确，回复：{"passed": true}
11如果有任何错误，回复：{"passed": false, "issues": [...]}
12
13不要提出改进建议。只标记真正的 bug。"""

1import anthropic, subprocess, json
2from pathlib import Path
3
4client = anthropic.Anthropic()
5CLAUDE_MD = Path("CLAUDE.md").read_text()
6LEARNINGS = Path("learnings.md").read_text()
7
8SYSTEM = f"""你是一个编码 Agent。
9阅读 issue，编写修复，运行测试。
10
11项目上下文：
12{CLAUDE_MD}
13
14已知问题：
15{LEARNINGS}
16
17规则：
18- 在更改任何内容之前阅读完整代码库
19- 为每次更改编写测试
20- 如果测试失败，修复代码，而不是测试
21- 当所有测试通过时停止"""
22
23TOOLS = [
24    read_file_tool,
25    write_file_tool,
26    run_command_tool,
27    search_codebase_tool,
28]
29
30def run(issue_text, max_attempts=5):
31    messages = [{"role": "user", "content": issue_text}]
32
33    for attempt in range(max_attempts):
34        # 调用 Claude
35        response = client.messages.create(
36            model="claude-sonnet-4-6",
37            max_tokens=8192,
38            system=SYSTEM,
39            tools=TOOLS,
40            messages=messages
41        )
42
43        # 执行工具调用
44        messages = handle_tool_use(response, messages)
45
46        # 验证：运行测试
47        test_result = subprocess.run(
48            ["pytest", "-x", "--tb=short"],
49            capture_output=True, text=True
50        )
51
52        if test_result.returncode == 0:
53            print(f"在 {attempt + 1} 次尝试后完成")
54            return True
55
56        # 将失败反馈到循环中
57        messages.append({
58            "role": "user",
59            "content": f"测试失败：\n{test_result.stdout}\n修复并重试。"
60        })
61
62    return False