CMA：Issue 到 PR 全流程

从 Issue 创建到提交 PR 的完整自动化

🌐 查看英文原文 | 源码 Notebook

编排：从 issue 到合并的 PR

本 notebook 引导 Agent 通过一个现实的端到端循环：读取模糊的 bug 报告、找到 bug、修复它、打开 PR、通过 CI、处理审查反馈并合并。真正的维护者工作流从来不是线性的，这就是练习的重点——Agent 必须在多种不同工具类型（读取 JSON、grep 代码、编辑文件、运行模拟 CLI、解析 CI 输出）之间携带状态，同时从两个链中意外中恢复：CI 失败和要求 docstring 的审查机器人。

状态通过链条流动：issue 正文 → 文件路径 → 修复 diff → PR 编号 → CI 输出 → 审查评论 → 最终合并。fixture 中的 gh-mock CLI 将所有内容持久化在 .gh-state/ 中，以便每个步骤都能看到前一步做了什么，在没有任何网络访问的情况下模拟真实的 GitHub 工作流。

除了 iterate notebook 之外，这还教会你：

跨长链条的多轮引导。 Session 文件系统和对话历史在轮次之间持久化，因此每条用户消息都从上一条消息停止的地方继续。我们利用这一点在最后验证最终状态。
链中恢复。 Agent 必须读取 CI 失败或审查评论并进行调整，而不是盲目重试。

fixture 位于 example_data/orchestrate/ 中，包含一个模拟 gh CLI、一个 issue JSON 文件和一个带有植入 bug 的 src/ + tests/ 布局。

import io
import os
import zipfile
from pathlib import Path

from anthropic import Anthropic
from utilities import stream_until_end_turn, wait_for_idle_status

MODEL = os.environ.get("COOKBOOK_MODEL", "claude-sonnet-4-6")
GH_TOKEN = os.environ.get("GITHUB_TOKEN")  # only needed for the github_repository sidebar

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

1. 打包 fixture

模拟仓库打包了几个文件：src/url_utils.py 包含实际的 bug，src/blog.py 是一个使 bug 更容易观察的调用者，tests/test_urls.py 在 bug 修复之前会失败，而 gh-mock CLI 加 issue_42.json 提供了 Agent 将驱动的类 GitHub 工作流。我们在内存中压缩目录并作为单个文件资源上传——有关如何改为挂载真实 GitHub 仓库，请参见末尾的侧边栏。

buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
    for f in FIXTURE.rglob("*"):
        if f.is_file() and f.name != "README.md":
            zf.write(f, f.relative_to(FIXTURE))
buf.seek(0)
fixture_zip = client.beta.files.upload(file=("repo.zip", buf, "application/zip"))
print(f"fixture: {fixture_zip.id}")

2. Agent + environment + session

Environment 声明 pytest 为 pip 依赖，以便 Agent 可以在其 CI 循环中实际运行测试套件。这是本 cookbook 中第一个需要网络访问来安装包的 notebook，因此在原本 limited 的网络配置旁边设置了 allow_package_managers: True。

agent = client.beta.agents.create(
    name="cookbook-orchestrate",
    model=MODEL,
    system=(
        "You are a maintainer bot. You read issues via `./gh-mock`, explore "
        "the codebase, write fixes, and shepherd PRs through CI and review. "
        "When CI fails or a reviewer requests changes, read what they said "
        "and address it, don't just retry blindly.\n\n"
        "Work in /mnt/user. All gh-mock commands run from there."
    ),
    tools=[
        {
            "type": "agent_toolset_20260401",
            "default_config": {
                "enabled": True,
                "permission_policy": {"type": "always_allow"},
            },
        }
    ],
)

env = client.beta.environments.create(
    name="cookbook-orchestrate-env",
    config={
        "type": "cloud",
        "networking": {"type": "limited", "allow_package_managers": True},
        "packages": {"pip": ["pytest"]},
    },
)

session = client.beta.sessions.create(
    environment_id=env.id,
    agent={"type": "agent", "id": agent.id, "version": agent.version},
    resources=[{"type": "file", "file_id": fixture_zip.id, "mount_path": "repo.zip"}],
    title="Issue #42 → PR",
)
print(f"session: {session.id}")

3. 运行完整链条

一条指令启动整个循环。需要关注的两个恢复点是 CI 失败和审查评论。如果 Agent 的第一次修复不完整——例如，如果它只处理了 é 而遗漏了 ü——gh-mock pr checks 将以非零退出，带有 Agent 需要读取和迭代的 pytest 输出。然后，一旦 CI 变绿，如果 slugify() 缺少 docstring，审查机器人将阻止合并，给 Agent 在最终合并步骤之前再一次适应的机会。

client.beta.sessions.events.send(
    session_id=session.id,
    events=[
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": (
                        "Unpack /mnt/session/uploads/repo.zip into /mnt/user "
                        "and ship a fix for issue #42 end-to-end. Read the "
                        "./gh-mock script first to see what subcommands it "
                        "supports; use those to view the issue, open a PR, "
                        "run CI, handle review feedback, and merge. Show me "
                        "the final PR state when you're done."
                    ),
                }
            ],
        }
    ],
)

print("=== full orchestrate chain ===")
stream_until_end_turn(client, session.id)

4. 多轮验证

Session 是有状态的：容器文件系统和对话历史在轮次之间持久化，因此后续只需发送另一个 user.message。我们在这里利用这一点独立验证最终状态——模拟 CLI 将 PR 状态持久化在 .gh-state/pr_101.json 中，因此打印该文件是确认 PR 最终以 CI 绿色且在合并前至少有一个审查批准而合并的最简单方法。

client.beta.sessions.events.send(
    session_id=session.id,
    events=[
        {
            "type": "user.message",
            "content": [
                {
                    "type": "text",
                    "text": (
                        "Print the contents of /mnt/user/.gh-state/pr_101.json so "
                        "I can see the final state, CI status, and reviews."
                    ),
                }
            ],
        }
    ],
)
stream_until_end_turn(client, session.id)

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")

侧边栏：挂载真实 GitHub 仓库

上面的 fixture 是模拟的，这样 notebook 可以离线运行，你不需要任何 GitHub 凭证来尝试它。对于针对真实仓库的真实工作，将上面的 {"type": "file", ...} 挂载替换为 {"type": "github_repository", ...} 资源，API 将在 session 启动时将仓库克隆到容器中。

这与 iterate notebook 中的文件挂载使用相同的 resources= 字段——该列表接受混合类型，因此你也可以在同一调用中克隆仓库并挂载单独的配置文件。Agent 的 bash/read/grep 工具将工作树视为普通目录；唯一的区别是 API 处理克隆而不是你。

session = client.beta.sessions.create(
    environment_id=env.id,
    agent={"type": "agent", "id": agent.id, "version": agent.version},
    resources=[
        {
            "type": "github_repository",
            "url": "https://github.com/anthropics/claude-cookbooks",
            "mount_path": "/workspace/cookbook",
            "authorization_token": GH_TOKEN,
            "checkout": {"type": "branch", "name": "main"},
        }
    ],
    title="Repo explorer",
)

私有仓库和克隆认证都需要 GITHUB_TOKEN。克隆在 session 创建时发生一次；后续轮次在相同的工作树上工作，无需重新克隆。

Tip

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