托管代理生产环境设置教程：从MCP工具集到推理地理固定

本教程面向有编程基础的开发者，深入讲解在生产环境中部署托管代理的关键机制。内容涵盖MCP工具集与Vaults的集成、Webhooks驱动的人工介入工作流、资源生命周期管理以及推理地理固定。通过端到端示例，展示如何为终端用户创建Vault、附加GitHub MCP凭证、运行代理会话，并利用Webhooks实现生产级的人机协同。同时，详细解析了资源归档与清理的最佳实践，并提供了常见错误排查与生产优化建议，帮助开发者构建安全、合规、可扩展的AI代理应用。

技术背景

在生产环境中运行托管代理，不仅需要实现代理循环本身，还需处理凭证管理、工具集成、人工介入、资源生命周期和合规性等外围机制。随着代理从原型走向真实用户，单租户的硬编码令牌和长连接流式模式无法满足多用户隔离、可扩展性和审计需求。MCP工具集通过标准协议连接外部SaaS服务，Vaults提供按用户隔离的凭证存储，Webhooks实现异步人工审批，推理地理固定满足数据驻留合规要求。这些机制共同构成了生产级代理应用的基础设施，确保安全性、可维护性和合规性。

核心原理

核心原理包括：1) MCP工具集：代理通过标准协议直接调用外部MCP服务器上的工具，无需应用层转发，Anthropic代理调用并返回结果。2) Vaults：按用户存储凭据的容器，会话创建时通过ID引用，代理调用工具时自动匹配凭据，令牌对代理不可见，审计日志与Vault绑定。3) Webhooks：注册事件回调，如session.status_idled和session.budget_reached，服务器异步处理，无需保持长连接。4) 推理地理固定：在代理上设置inference_geo，强制模型请求从指定区域服务，并在会话创建和每轮对话中验证。5) 资源生命周期：所有资源遵循list、retrieve、update、archive、delete五动词模式，archive保留审计记录但拆除实时容器。

架构分析

系统架构采用分层设计：客户端通过Anthropic SDK与API交互，API管理代理、环境、会话、Vaults和凭据。代理定义包含MCP服务器配置和工具集，会话创建时关联Vault和代理。运行时，代理在沙箱环境中执行，通过MCP协议调用外部服务，凭据由Vault提供。Webhooks作为异步事件通道，将状态变更推送给外部服务器，实现人工介入。资源生命周期管理通过统一API动词实现，确保可审计性和配额控制。地理固定作为代理属性，在请求路由时强制执行。

完整实战教程

操作：在生产环境中运行托管代理

大多数其他托管代理指南侧重于代理循环本身，即让代理对某个测试夹具执行有用的操作。本指南则关注该循环周围的机制，即在你将托管代理应用部署到真实用户之前所需的部分：

MCP工具集 而非自定义工具，当你的代理需要与SaaS API通信，而不想每次调用都通过你的应用程序往返时。
Vaults 用于保存每个终端用户的凭证，这样每个用户的GitHub / Linear / Slack令牌就能与其他用户分开，并且你的审计日志保持干净。
Webhooks 用于驱动人工介入的工作流，而无需在整个过程中保持长时间运行的HTTP连接。
资源生命周期 动词（列出、检索、更新、归档、删除）用于管理工作区随时间积累的内容。
推理地理固定（inference_geo）当合规要求模型请求在特定区域运行时。

我们将构建一个端到端流程，涵盖前四个，然后以地理固定结束：为虚构的终端用户创建Vault，附加GitHub MCP凭证，运行使用服务端凭证的代理会话，展示你注册的Webhook处理器以从真实生产服务器驱动同一会话，并逐步介绍之后用于清理的管理动词。

此笔记本需要环境中的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 工具集与保险库

MCP 工具集是第三种扩展模式，前两种分别是自定义工具（gate 笔记本）和 resources= 挂载（orchestrate 笔记本）。MCP 工具集将代理指向一个实现了模型上下文协议的外部服务器。代理直接在沙箱内调用该服务器上的工具，无需经过你的应用程序往返，Anthropic 代理这些调用，服务器响应，代理继续执行。绝大多数公共 SaaS API（GitHub、Slack、Linear、Stripe、Notion、Salesforce、Asana 等）要么已经有 MCP 服务器，要么可以在一个下午内封装成一个，它们都是很好的 MCP 候选。

经验法则：如果服务可以通过公共互联网使用 bearer token 访问，那么 MCP 工具集就能工作。如果只能从你自己的网络内部访问，则改用自定义工具，这正是 gate 笔记本所涵盖的内容。

保险库回答了“我把令牌放在哪里？”这个问题。在会话创建时硬编码单个令牌适用于单租户设置，但一旦你有最终用户，它就会崩溃。每个用户需要自己的 GitHub 凭据，并且你需要将它们彼此隔离。保险库是一个按用户存储凭据的容器，你只需注册一次，然后在为该用户创建的每个会话中通过 ID 引用它。你不需要运行自己的秘密存储，不需要在每个请求中传递令牌，审计跟踪与保险库绑定，因此你始终知道代理代表哪个最终用户行事。

1. 为终端用户创建保险库

保险库有一个 display_name，会显示在控制台中，还有一个 metadata 字典，你通常会在其中存储内部用户 ID，以便将保险库映射回你自己数据库中的记录。

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 与代理调用该服务器时使用的令牌配对。对于 GitHub Copilot MCP 服务器，静态持有者令牌（即你的 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. 在会话中引用保险库

在 sessions.create 上传递 vault_ids=[vault.id]，API 会在每次工具调用时查找匹配的 MCP 服务器 URL。代理永远不会看到令牌本身，你也不必在请求中传递它。代理定义只是像往常一样列出 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. 以最终用户身份运行一轮对话

现在，代理对 GitHub 执行的所有操作都通过 vault 的凭据进行。在您自己的系统中进行审计非常简单：您确切知道是哪个最终用户在执行操作，因为 vault 通过您在步骤 1 中设置的元数据与他们关联。

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. 用于生产环境 HITL 的 Webhooks

门控笔记本中的流式模式在开发期间很方便，因为所有事情都在一个进程中发生，但它会在人工审核期间保持 HTTP 连接打开，这无法扩展，并且在进程重启后无法存活。生产模式则是在控制台中注册一个 webhook，该 webhook 在 session.status_idled 事件时触发，该事件表示代理已完成或正在等待工具结果。

当 webhook 触发时，你的服务器检查事件，将任何待处理的升级请求呈现给审核者，并在人工完成后 POST 回 user.custom_tool_result。会话只是保持空闲状态，直到你响应，你的服务器无需保持长连接。

如果你的会话使用 budget 运行（参见 CMA_cap_session_spend.ipynb），还要注册 session.budget_reached 事件。当会话的累计成本达到上限时，该事件触发一次，这是监督进程需要决定每个会话是提高上限还是保持暂停的信号。它携带会话 ID，因此处理程序与空闲事件的处理程序形状相同：查找会话，做出决定，执行操作。

Webhook 注册是控制台中的一次性步骤，位于 设置 → Webhooks 下。你会获得一个 whsec_... 签名密钥，该密钥仅在创建时显示一次；请将其存储在密钥管理器中。

下面的代码块是参考实现，不是笔记本单元格。 将其复制到你的服务器中，它依赖于 FastAPI，而 cookbook 并未安装该库，并且它不会作为本笔记本流程的一部分运行。与门控笔记本中的代理定义配合使用，足以从生产服务器端到端驱动门控工作流。

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)

    elif event["event_type"] == "session.budget_reached":
        # Session paused at its spend cap with its work intact. Raise
        # the budget to let it continue, or leave it paused and page
        # an operator.
        review_budget(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. 固定推理地理位置

当合规性要求模型请求必须从特定地理位置提供服务时，请将其固定在代理上，而不是依赖工作区的默认设置：

agent = client.beta.agents.create(
    name="residency-pinned agent",
    model={"id": MODEL, "inference_geo": "us"},
    ...
)

可接受值为 "global" 和 "us"，受工作区的 allowed_inference_geos 限制。若未设置，每个请求在服务时解析为工作区的 default_inference_geo，因此之后对该默认值的更改会影响已在运行的会话。若已设置，该固定值会在代理保存时、会话创建时以及每一轮对话中验证：缩小工作区允许列表会阻止新的会话使用现在不被允许的代理，并拒绝正在运行的会话的下一轮对话，而不是将其视为旧有会话而放行。这种严格性使得该固定值可用作驻留控制。sessions.create 上的 agent_with_overrides 可将单个会话移动到另一个允许的地理区域，而无需编辑共享代理。完整的演练（包括清除固定值的更新语义）见 CMA_pin_inference_geo.ipynb。

7. 资源生命周期：列表、检索、更新、归档

API 中的每个资源（代理、环境、会话、保险库、凭据）都遵循相同的五动词模式：list、retrieve、update、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")

8. 清理

凭据和保险库都有自己的归档端点。归档保险库不会自动归档其凭据，因此如果你想彻底清理，请先处理凭据。

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

其他示例笔记本

本笔记本是生产形态的收尾之作。如果你还没有运行过它所围绕的工作流笔记本，建议先运行它们：

CMA_iterate_fix_failing_tests.ipynb — 入门笔记本。通过一个失败测试套件的观察-修复循环，介绍代理、环境、会话、文件挂载和流式事件循环。
CMA_orchestrate_issue_to_pr.ipynb — 多轮代理，通过模拟的 gh CLI 将一个问题一路推进到合并的 PR，并在中途从 CI 失败和审查评论中恢复。
CMA_explore_unfamiliar_codebase.ipynb — 接地模式，包含一个故意放置的过时文档陷阱。还展示了 sessions.resources.add 用于向运行中的会话推送更多上下文。
CMA_gate_human_in_the_loop.ipynb — 自定义工具 decide() 和 escalate() 的往返，用于人在回路工作流，并与上面的 webhook 参考块配对。

常见问题

常见错误：1) 未设置GITHUB_TOKEN导致脚本退出，需确保环境变量存在。2) Vault或凭据ID错误导致会话创建失败，检查ID是否正确。3) Webhook签名验证失败，确认签名密钥与请求头匹配。4) 会话未空闲时归档导致冲突，需先等待idle状态。5) 地理固定值不在allowed_inference_geos中，导致会话被拒绝。排查方法：使用API的list和retrieve端点验证资源状态；检查Webhook日志；使用调试模式查看事件流。生产优化：使用密钥管理器存储令牌和Webhook密钥；实现重试机制；监控会话预算和资源配额；定期归档旧资源。

企业应用扩展

可扩展方向：1) 集成更多MCP服务器，如Slack、Stripe等，扩展代理能力。2) 实现完整的OAuth流程，支持动态令牌刷新。3) 构建自定义Webhook处理服务，支持多事件类型和复杂审批流程。4) 利用资源版本控制，实现代理配置的灰度发布。5) 结合监控和日志系统，实现全链路追踪。6) 探索多区域地理固定，满足全球合规需求。7) 开发自动化清理脚本，定期归档过期资源。

总结

内容来源与验证

内容等级：PREMIUM_A
质量等级：S_CANDIDATE
难度：高级
验证等级：V1
原文作者：pauly-ant
原文日期：2026-04-08
来源许可：MIT
来源版本：fbb9bbdb187d69f4d5e4c2cb42f68168cd558b2a
查看原文

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员