CMA：迭代修复失败测试

入门教程：Agent 自动修复测试 Bug

🌐 查看英文原文 | 源码 Notebook

迭代：执行 → 观察 → 修复

这是入口 notebook。你将通过做 Agent 最普遍的事情来学习 Managed Agents API：尝试某些操作，读取发生了什么，再试一次。我们上传一个包含两个植入 bug 的小型包，告诉 Agent 让测试通过，然后观察它执行循环——运行测试、读取 traceback、编辑代码、重新运行、重复直到通过。

在此过程中，你将看到本 cookbook 其余部分所基于的每个 API 形状：agent / environment / session、文件挂载、事件流和 archive 调用。通过本 notebook，你将完成驱动 Agent 端到端运行所需的一切。

概念

三个需要了解的资源：

  • Agent，可复用的配置（模型、系统提示、工具）
  • Environment，容器模板（包、网络）
  • Session，绑定 agent 和 environment，挂载 Agent 需要的任何文件，并产生事件流

你创建一次 agent 和 environment，然后在多个 session 中复用它们。每个 session 是一个独立的运行。

import os
from pathlib import Path

from anthropic import Anthropic

MODEL = os.environ.get("COOKBOOK_MODEL", "claude-sonnet-4-6")

client = Anthropic()
FIXTURE = Path("example_data") / "iterate"

1. 创建 agent

系统提示是故意精简的。我们希望 Agent 自己发现迭代循环，而不是遵循逐步脚本——测试输出使任务足够明显，无需进一步指导。

agent_toolset_20260401 是内置工具集：bash、read、write、edit、glob、grep、web_fetch 和 web_search。将 permission_policy 设置为 always_allow 让 Agent 无需往返确认即可运行它们。

agent = client.beta.agents.create(
    name="cookbook-iterate",
    model=MODEL,
    system=(
        "You are a debugging agent. Your job is to make failing tests pass. "
        "Run the tests, read the failures, fix the code, repeat until green. "
        "Stop when every assertion passes."
    ),
    tools=[
        {
            "type": "agent_toolset_20260401",
            "default_config": {
                "enabled": True,
                "permission_policy": {"type": "always_allow"},
            },
        }
    ],
)

2. 创建 environment

Environment 是一个容器模板。type: cloud 在 Anthropic 的托管沙盒中运行。networking: limited 阻止任意出站流量——本 notebook 完全不需要网络访问，所以我们保持锁定。

env = client.beta.environments.create(
    name="cookbook-iterate-env",
    config={"type": "cloud", "networking": {"type": "limited"}},
)

3. 上传失败的测试

通过 Files API 上传文件以获取 ID。我们将在步骤 4 中将它们挂载到 session 上。

calc.py 有两个植入的 bug，test_calc.py 有三个捕获它们的断言。其中一个失败（test_mean）是其他两个的下游，这巧妙地教会 Agent 不要过度修复：mean() 内部调用 add 和 divide，所以一旦它们被修复，test_mean 就会自行通过，无需对 mean() 进行任何直接编辑。

calc_file = client.beta.files.upload(
    file=("calc.py", (FIXTURE / "calc.py").read_bytes(), "text/x-python")
)
test_file = client.beta.files.upload(
    file=("test_calc.py", (FIXTURE / "test_calc.py").read_bytes(), "text/x-python")
)
print(f"uploaded: {calc_file.id}, {test_file.id}")

4. 创建 session

Session 绑定 agent 和 environment，挂载 Agent 需要的任何文件，并启动一个新的容器。resources= 是你在 Agent 启动之前将数据放入容器的方式——orchestrate notebook 展示了如何使用相同的字段克隆 GitHub 仓库而不是挂载单个文件。

文件挂载在 /mnt/session/uploads/<mount_path> 下，这是只读的。Agent 必须将文件复制到可写目录（如 /mnt/user 或 /tmp）才能编辑它们，而你想稍后检索的任何内容都放在 /mnt/session/outputs/ 中。

session = client.beta.sessions.create(
    environment_id=env.id,
    agent={"type": "agent", "id": agent.id, "version": agent.version},
    resources=[
        {"type": "file", "file_id": calc_file.id, "mount_path": "calc.py"},
        {"type": "file", "file_id": test_file.id, "mount_path": "test_calc.py"},
    ],
    title="Get the tests green",
)
print(f"session: {session.id}")

5. 驱动 Agent 并观察其工作

两个步骤：发送带有任务的 user.message 事件，然后读取事件流直到 Agent 到达 end_turn。

该流是一个 server-sent event 连接。我们使用它（而不是轮询，参见末尾的侧边栏）因为 Agent 将花费约 30 秒进行迭代，我们希望实时看到每一轮。

两个需要内化的模式：

  1. 先打开流，再发送。 with 块打开 SSE 连接；你在块内 send 的任何内容都保证可观察。先发送后打开有丢失在竞争窗口中触发的事件的风险。
  2. 在 session.status_idle 且 stop_reason.type == "end_turn" 时退出。 Session 在等待输入时变为空闲——无论是一轮结束时还是自定义工具调用需要响应时。stop_reason.type 消除歧义；end_turn 是我们的退出信号。gate notebook 展示了同一循环的 requires_action 侧。
with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session_id=session.id,
        events=[
            {
                "type": "user.message",
                "content": [
                    {
                        "type": "text",
                        "text": (
                            "The tests in /mnt/session/uploads/test_calc.py are "
                            "failing. Copy both files into /mnt/user, iterate "
                            "on calc.py until every test passes, then write the "
                            "final calc.py to /mnt/session/outputs/calc.py. "
                            "pytest isn't installed here, run the assertions "
                            "directly with `python3 -c ...` instead."
                        ),
                    }
                ],
            }
        ],
    )
    print("--- iterate loop ---")
    for ev in stream:
        match ev.type:
            case "agent.message":
                for b in ev.content:
                    if b.type == "text":
                        print(b.text, end="")
            case "agent.tool_use":
                print(f"\n[{ev.name}]")
            case "session.status_idle" if ev.stop_reason and ev.stop_reason.type == "end_turn":
                break
            case "session.status_terminated":
                break

那个 match ev.type: 块是标准的流式传输模式。本 cookbook 中的所有其他 notebook 都从 utilities.py 导入它作为 stream_until_end_turn，而不是重复循环。我们在下面的验证步骤中使用它。

wait_for_idle_status 是来自 utilities.py 的第二个辅助函数。它吸收了步骤 7 中 callout 描述的竞争：即使流已经产生了 session.status_idle，session 记录上的服务端 status 字段可能短暂地仍为 running，立即调用 archive() 会返回 400。该辅助函数只是轮询 sessions.retrieve 直到字段稳定。流式传输后紧接着归档的代码需要它。

from utilities import stream_until_end_turn, wait_for_idle_status

6. 验证

不要轻信 Agent 的话。独立地再运行每个断言一次并打印最终的 calc.py。如果 Agent 在最后一次循环内运行和轮次结束之间过度修复或回退了某些内容，这会捕获它。

client.beta.sessions.events.send(
    session_id=session.id,
    events=[
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": (
                        "Re-run every assertion from "
                        "/mnt/session/uploads/test_calc.py one more time "
                        "against your final calc.py with `python3 -c ...` "
                        "to confirm they all pass, then cat the final "
                        "/mnt/session/outputs/calc.py."
                    ),
                }
            ],
        }
    ],
)
stream_until_end_turn(client, session.id)

7. 清理

归档是你将 session、environment 或 agent 标记为已完成的方式。它会拆除任何活动的容器，停止资源计入你的工作空间配额，并将其从默认列表视图中隐藏，但它会保留记录、配置和事件历史以供审计和任何想稍后通过 ID 检索资源的人使用。如果你想完全删除记录，大多数资源还暴露了一个单独的 delete 端点（operate notebook 详细介绍了资源生命周期），但 archive 几乎总是你在运行结束时想要的。

wait_for_idle_status(client, session.id)
client.beta.sessions.archive(session.id)
client.beta.environments.archive(env.id)
client.beta.agents.archive(agent.id)
print("archived")

侧边栏：轮询代替流式传输

当你想要实时查看 Agent 将花费超过几秒钟的任务进度时，步骤 5 中的流式传输模式是正确的选择。对于较短的任务，或者对于不希望长时间保持 HTTP 连接的生产代码，你可以改用 events.list 轮询来做同样的事情：

client.beta.sessions.events.send(session_id=..., events=[...])
while True:
    time.sleep(2)
    events = client.beta.sessions.events.list(session.id).data
    last = events[-1] if events else None
    if last and last.type == "session.status_terminated":
        break
    if (
        last
        and last.type == "session.status_idle"
        and last.stop_reason
        and last.stop_reason.type == "end_turn"
    ):
        break
# walk events to print agent.message text

权衡：

当你想观察 Agent 工作时，流式传输胜出。每次工具调用、每条部分消息、每次状态转换都在服务器发射后立即到达，这正是你在开发新工作流时想要的。代价是一个长期的 SSE 连接：你的进程必须在轮次期间保持活跃，不能在间隔中暂停和恢复，而且错误时刻的网络故障可能让你在流中途断开，没有干净的方法恢复你离开的地方。

轮询在相反的情况下胜出。它是无状态的，在进程重启后存活，并与不想保持连接打开的 webhook 处理器和队列工作器干净地组合。代价是延迟和隐藏的进度：在你再次轮询之前你什么也看不到，所以反馈受限于你的轮询间隔，一个长轮次在完成之前看起来像是在沉默。

在 Agent 可能运行数分钟且你的处理器无法保持连接打开的生产设置中，轮询模式（或其生产表亲——gate notebook 中展示的 session.status_idled webhook）就是你想要的。

接下来去哪里

迭代循环是 Agent 循环采取的最简单形状。此目录中的四个配套 notebook 基于相同的 API 形状，展示了你可以驱动的其他工作流：

  • CMA_orchestrate_issue_to_pr.ipynb — 多轮 Agent 在更长的工具链中携带状态：读取 issue、编写修复、打开 PR、从 CI 失败中恢复、处理审查评论并合并。
  • CMA_explore_unfamiliar_codebase.ipynb — 将 Agent 放入它从未见过的仓库的接地模式，带有一个植入的过期文档陷阱。还展示了 sessions.resources.add 用于向运行中的 session 推送更多上下文。
  • CMA_gate_human_in_the_loop.ipynb — 用于人工参与工作流的自定义工具 decide() 和 escalate() 往返。涵盖了 requires_action 空闲弹跳和并行工具调用去重。
  • CMA_operate_in_production.ipynb — 生产设置故事：基于 vault 的 MCP 凭证、用于无需长期连接的 HITL 的 session.status_idled webhook，以及资源生命周期 CRUD 动词。
Tip

中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。