CMA：生产环境运维
端到端的生产环境 Agent 部署
🌐 查看英文原文 | 源码 Notebook
运维：在生产环境中运行 Managed Agents
其他大多数 Managed Agents cookbook 专注于 agent 循环本身，即让 agent 针对某个固定任务完成有用的工作。而本篇关注的是围绕该循环的周边机制——在将 Managed Agents 应用部署到真实用户面前之前需要的各个组件：
- MCP 工具集——替代自定义工具，适用于 agent 需要与 SaaS API 通信且不希望每次调用都经过应用层往返的场景。
- Vault（凭证保险库）——用于存储每个终端用户的凭证，使每个用户的 GitHub / Linear / Slack token 相互隔离，审计链路清晰。
- Webhook——用于驱动人工介入（human-in-the-loop）流程，无需全程保持长连接 HTTP 连接。
- 资源生命周期——通过动词式操作（list、retrieve、update、archive、delete）管理工作空间中长期积累的资源。
我们将构建一个端到端流程，串联上述四个部分：为一个虚拟终端用户创建 vault，为其绑定 GitHub MCP 凭证，运行一个使用该凭证的 agent 会话（在服务端），展示你在生产服务器上注册的 webhook 处理程序，并梳理用于后续清理的管理操作。
运行本 notebook 需要在环境中设置 GITHUB_TOKEN。
import os
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")
client = Anthropic()
GH_TOKEN = os.environ.get("GITHUB_TOKEN")
if not GH_TOKEN:
raise SystemExit("Set GITHUB_TOKEN to run this notebook.")概念：MCP 工具集和 Vault
MCP 工具集是第三种扩展模式，前两种分别是自定义工具（gate notebook 中介绍）和 resources= 挂载（orchestrate notebook 中介绍）。MCP 工具集将 agent 指向一个实现了 Model Context Protocol 的外部服务器。agent 直接从沙箱内部调用该服务器上的工具，无需经过你的应用层往返——Anthropic 负责代理调用，服务器响应后 agent 继续执行。绝大多数公共 SaaS API（GitHub、Slack、Linear、Stripe、Notion、Salesforce、Asana 等）要么已有 MCP 服务器，要么可以在半天内封装出一个，它们都是很好的 MCP 候选。
经验法则：如果服务可以通过公共互联网使用 bearer token 访问，MCP 工具集就能工作。如果只能从你自己的网络内部访问，就使用自定义工具——这正是 gate notebook 所覆盖的内容。
Vault 解决的是”token 放在哪里”的问题。在会话创建时硬编码单个 token 适用于单租户场景，但一旦有了终端用户就不行了。每个用户需要自己的 GitHub 凭证，而且你需要将它们相互隔离。Vault 是一个按用户的凭证容器，你只需注册一次，之后在为该用户创建的每个会话中通过 ID 引用即可。你不需要运行自己的密钥存储，不需要在每次请求中传递 token，审计链路与 vault 绑定，因此你始终知道 agent 代表哪个终端用户在操作。
1. 为终端用户创建 Vault
Vault 有一个在 Console 中显示的 display_name，以及一个 metadata 字典（通常存放你的内部用户 ID），以便将 vault 映射回你自己数据库中的记录。
vault = client.beta.vaults.create(
display_name="Cookbook demo user",
metadata={"internal_user_id": "u_demo_001", "team": "engineering"},
)
print(f"vault: {vault.id}")2. 绑定 MCP 凭证
凭证存在于 vault 之下。每个凭证将一个 MCP 服务器 URL 与 agent 调用该服务器时使用的 token 配对。对于 GitHub Copilot MCP 服务器，最简单的形式是静态 bearer token（即你的 GitHub PAT）。API 还支持带有刷新机制的完整 OAuth 流程，适用于需要它的服务，两种形式都通过 auth= 处理。
credential = client.beta.vaults.credentials.create(
vault_id=vault.id,
display_name="GitHub Copilot",
auth={
"type": "static_bearer",
"mcp_server_url": "https://api.githubcopilot.com/mcp/",
"token": GH_TOKEN,
},
)
print(f"credential: {credential.id}")3. 在会话中引用 Vault
在 sessions.create 时传入 vault_ids=[vault.id]，API 会在每次工具调用时查找匹配的 MCP 服务器 URL。agent 永远不会看到 token 本身，你也不需要在请求中传递它。agent 定义只需照常列出 MCP 服务器，凭证的关联在会话创建时完成。
agent = client.beta.agents.create(
name="cookbook-operate",
model=MODEL,
system="You navigate GitHub repositories on behalf of the logged-in user.",
mcp_servers=[
{
"type": "url",
"name": "github",
"url": "https://api.githubcopilot.com/mcp/",
}
],
tools=[
{
"type": "mcp_toolset",
"mcp_server_name": "github",
"default_config": {
"enabled": True,
"permission_policy": {"type": "always_allow"},
},
}
],
)
env = client.beta.environments.create(
name="cookbook-operate-env",
config={"type": "cloud", "networking": {"type": "unrestricted"}},
)
session = client.beta.sessions.create(
environment_id=env.id,
agent={"type": "agent", "id": agent.id, "version": agent.version},
vault_ids=[vault.id],
title="Operate demo",
)
print(f"session: {session.id}")4. 以终端用户身份运行一轮对话
agent 针对 GitHub 的所有操作现在都会通过 vault 中的凭证进行。在你自己的系统中做审计很简单：你确切知道是哪个终端用户在操作，因为 vault 通过你在步骤 1 中设置的 metadata 与之关联。
client.beta.sessions.events.send(
session_id=session.id,
events=[
{
"type": "user.message",
"content": [
{
"type": "text",
"text": "List the three most recently updated repos in the anthropics org.",
}
],
}
],
)
print("--- vault-backed MCP call ---")
stream_until_end_turn(client, session.id)5. 用于生产环境人工介入的 Webhook
gate notebook 中的流式模式在开发阶段很方便，因为一切都在一个进程中完成。但它在人工审核期间会保持 HTTP 连接打开——这无法扩展，也无法在进程重启后存活。生产环境的做法是在 Console 中注册一个 webhook，触发条件为 session.status_idled，这个信号表示 agent 要么已完成，要么正在等待工具结果。
当 webhook 触发时，你的服务器检查事件，将待处理的升级请求推送到审核者面前，并在人工完成后 POST user.custom_tool_result 回去。会话只是安静地等待你的响应，你这边没有任何长连接。
Webhook 注册是 Console 中的一次性步骤，位于 Settings → Webhooks 下。你会获得一个 whsec_... 签名密钥，仅在创建时显示一次；请将其存储到你的密钥管理器中。
下面的代码块是参考实现，不是 notebook 单元格。 将其复制到你自己的服务器中——它依赖 FastAPI（本 cookbook 未安装），不作为本 notebook 流程的一部分运行。配合 gate notebook 中的 agent 定义，它足以从生产服务器端到端地驱动 gate 工作流。
import hmac
import hashlib
import json
from fastapi import FastAPI, Header, HTTPException, Request
app = FastAPI()
WEBHOOK_SECRET = "whsec_..." # from Console, load from your secrets manager
def verify(body: bytes, sig: str) -> bool:
expected = hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, sig)
@app.post("/webhooks/anthropic")
async def receive(req: Request, x_anthropic_signature: str = Header()):
body = await req.body()
if not verify(body, x_anthropic_signature):
raise HTTPException(401)
event = json.loads(body)
session_id = event["resource_id"]
if event["event_type"] == "session.status_idled":
# Agent went idle. Either it's done OR it called escalate()
# and is waiting on a user.custom_tool_result. Look at the
# latest events to decide.
events = client.beta.sessions.events.list(session_id=session_id)
pending = [
e for e in events.data
if e.type == "agent.custom_tool_use" and e.name == "escalate"
and not has_result(events, e.id)
]
if pending:
for tu in pending:
enqueue_for_review(session_id, tu.id, tu.input)
else:
finalize_run(session_id)
return {"ok": True}
# Called from your review UI once the human decides
def submit_review(session_id: str, custom_tool_use_id: str, decision: str):
client.beta.sessions.events.send(
session_id=session_id,
events=[{
"type": "user.custom_tool_result",
"custom_tool_use_id": custom_tool_use_id,
"content": [{"type": "text",
"text": json.dumps({"human_decision": decision})}],
}],
)响应 agent 的代码与 gate notebook 中 Part A 循环的代码完全相同。唯一变化的是你的服务器如何得知有工作要做：不再是本地循环拉取事件，而是 webhook 按你的节奏推送通知。
6. 资源生命周期：list、retrieve、update、archive
API 中的每个资源——agents、environments、sessions、vaults、credentials——都暴露相同的五种动词模式：list、retrieve、update、archive 和（部分资源的）delete。我们将在 agents 上演示完整集合，然后列出其他资源上可用的动词作为快速参考。
archive 与 delete 的区别： archive 保留记录以供审计和检索，但拆除任何活跃容器并停止该资源占用工作空间配额。delete 则完全删除记录。对于大多数工作流，archive 是正确的选择；仅在你确实需要删除记录时（例如测试清理）才使用 delete。
listed = client.beta.agents.list(limit=5)
print(f"workspace has at least {len(listed.data)} agents")
retrieved = client.beta.agents.retrieve(agent.id)
print(f"retrieved agent: name={retrieved.name} version={retrieved.version}")
# Updating an agent produces a new version. Pass the current
# version to confirm you're updating from a known state, a
# concurrent update from another process will reject yours rather
# than silently overwriting.
updated = client.beta.agents.update(
agent.id,
version=agent.version,
system="You navigate GitHub repositories. Be terse.",
)
print(f"updated to version {updated.version}")
# Every historical version is queryable.
versions = client.beta.agents.versions.list(agent_id=agent.id)
print(f"agent has {len(versions.data)} versions")7. 清理
凭证和 vault 有各自的 archive 端点。归档 vault 不会自动归档其下的凭证，因此如果你想要彻底清理，请先处理凭证。
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)
client.beta.vaults.credentials.archive(credential.id, vault_id=vault.id)
client.beta.vaults.archive(vault.id)
print("archived")其他 Cookbook
本 notebook 是面向生产环境的收尾篇。它所包裹的工作流 notebook 如果你还没跑过，值得先跑一遍：
CMA_iterate_fix_failing_tests.ipynb— 入门 notebook。通过一个”执行-观察-修复”循环（针对一个失败的测试套件）介绍 agents、environments、sessions、文件挂载和流式事件循环。CMA_orchestrate_issue_to_pr.ipynb— 多轮 agent，通过模拟 gh CLI 将一个 issue 一路推进到合并的 PR，包含从 CI 失败和 review 评论中的链路中恢复。CMA_explore_unfamiliar_codebase.ipynb— 接地（grounding）模式，包含一个预设的过时文档陷阱。还展示了sessions.resources.add用于向运行中的会话推送更多上下文。CMA_gate_human_in_the_loop.ipynb— 用于人工介入工作流的自定义工具decide()和escalate()往返，与上方的 webhook 参考代码块配套使用。
中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。