默认情况下,Managed Agents 在 Anthropic 托管的云沙箱中执行工具和代码。自托管沙箱将编排保留在 Anthropic 一侧,但将工具执行移至您控制的基础设施中,因此智能体的代码、文件系统和网络出口永远不会离开您的环境。
工具执行保留在您的主机上:智能体读写的文件系统、它派生的进程以及它可以访问的网络都在您的控制之下。工具输入和输出仍会流向 Anthropic 的控制平面(Claude 在此运行),以便模型能够查看结果并决定下一步操作。有关完整的数据流边界,请参阅安全模型。
| 云环境 | 自托管沙箱 | |
|---|---|---|
| 工具运行位置 | Anthropic 托管的沙箱 | 您的基础设施 |
| 网络访问范围 | Anthropic 的出口控制 | 您的网络策略 |
| 文件和 GitHub 仓库挂载 | 由 Anthropic 管理 | 由您管理 |
| 生命周期 | 由 Anthropic 管理 | 由您管理 |
当智能体需要操作不能离开您网络边界的数据、访问无法公开路由的内部服务,或在您组织自己的合规和审计控制下运行时,自托管是一个很好的选择。
有关零数据保留(Zero Data Retention)和 HIPAA BAA 资格,请参阅 API 和数据保留。
自托管控制的是智能体代码的执行位置。MCP 隧道控制的是 Anthropic 如何访问您网络中的 MCP 服务器。两者相互独立:在 Anthropic 云沙箱中运行的会话仍可通过隧道访问私有 MCP 服务器,而自托管会话可以使用隧道式或公共 MCP 服务器。当您希望执行和工具访问都保留在您的边界内时,可同时使用两者。如果要让智能体使用您网络内 MCP 服务器的工具而不运行隧道,您也可以将服务器封装为自定义工具,由您的 worker 提供服务。
环境 worker 是您在自己的基础设施上运行的进程。它从 Anthropic 接收工具执行请求并在本地运行。self_hosted 环境充当工作队列:当会话被分配到该环境时,Anthropic 会将该会话作为工作项加入队列。您的 worker 从该队列中认领工作项,为每个工作项派生一个执行上下文,下载智能体的技能(可复用的、基于文件系统的资源,为智能体提供特定领域的专业知识),运行工具调用,并将结果回传。
工作项通过轮询环境的队列来认领:可以使用持续轮询的常驻 worker,也可以使用在 session.status_run_started 时唤醒并开始轮询的 webhook 触发式处理程序。
CLI 和 SDK 都附带了预构建的 worker。ant CLI 仅支持常驻模式;SDK 同时支持常驻和 webhook 触发两种模式。两者均可配置:有关 CLI 标志,请参阅参考文档中的自托管 worker;有关 SDK 选项,请参阅本页的 SDK 辅助工具。如需更多控制,可直接调用 Environments Work 端点并实现您自己的 worker。
/workspace: 工具执行和技能下载的系统默认工作目录。CLI 的 --workdir 标志默认为当前目录;传递 --workdir /workspace 以匹配系统默认值。技能会下载到 <workdir>/skills/<name>/。如果您使用不同的工作目录,请更新智能体的系统提示,以便 Claude 能够找到技能文件。/mnt/session/outputs 指令,因此最终交付物会落在智能体在您的沙箱文件系统中写入的任何位置,通常在工作目录下。您需要:
/bin/bash 位于该确切路径。worker 的 bash 工具会直接调用它,而不查询 PATH。TypeScript SDK 还需要 PATH 中有 unzip 和 tar,以及 Node.js 22 或更高版本;Python 和 Go SDK 使用其标准库进行归档提取,没有额外的二进制文件要求。ant CLI 或 Anthropic SDK(Python、TypeScript 或 Go)。创建自托管环境
在 Console 中:Workspace > Environments > New > Self-hosted
或通过 API:
client = anthropic.Anthropic()
environment = client.beta.environments.create(
name="self-hosted", config={"type": "self_hosted"}
)
print(environment.id)生成环境密钥
在 Console 中,打开该环境并点击 Generate environment key。无论您是通过 Console 还是 API 创建的环境,密钥生成都仅限 Console。然后在 worker 主机上导出环境 ID 和密钥:
export ANTHROPIC_ENVIRONMENT_KEY="sk-ant-oat01-..."
export ANTHROPIC_ENVIRONMENT_ID="env_..."选择常驻模式以获得最简单的设置:一个长期运行的进程持续轮询队列,只需要出站 HTTPS。选择 webhook 触发模式可避免运行空闲的轮询器;它需要一个 Anthropic 可以访问的 webhook 端点(有关端点设置和签名验证,请参阅 Webhooks)。
安装 ant CLI
在 worker 主机上运行此命令。
对于 Linux 环境,直接下载发布的二进制文件。
VERSION=1.22.1
OS=$(uname -s | tr '[:upper:]' '[:lower:]')
case $(uname -m) in
x86_64) ARCH=amd64 ;;
aarch64) ARCH=arm64 ;;
esac
curl -fsSL "https://github.com/anthropics/anthropic-cli/releases/download/v${VERSION}/ant_${VERSION}_${OS}_${ARCH}.tar.gz" \
| sudo tar -xz -C /usr/local/bin ant您可以在 GitHub 发布页面上找到所有版本。
运行 worker
进程内运行
ant beta:worker poll 认领分配给该环境的工作项,下载技能,在工作目录中执行工具调用,并将结果回传。它从环境变量中读取 ANTHROPIC_ENVIRONMENT_KEY 和 ANTHROPIC_ENVIRONMENT_ID。
ant beta:worker poll \
--workdir "/workspace"worker 在收到 SIGTERM 或 SIGINT 时会干净退出:它会取消任何正在进行的工具调用,发布其错误结果,并在停止前释放工作项。
每个会话一个沙箱
如果您需要更强的隔离(全新的文件系统、资源限制或按会话的网络控制),请在每个会话自己的沙箱中运行。构建一个安装了 ant 并以 ant beta:worker run 作为入口点的镜像。基础镜像必须提供 /bin/bash;curl 仅在构建时使用。当沙箱启动时,它从环境变量中读取会话详细信息,处理该会话,然后退出:
FROM your-base-image
ARG ANT_VERSION=1.22.1
ARG TARGETARCH
RUN ARCH=$([ "$TARGETARCH" = "arm64" ] && echo arm64 || echo amd64) && \
curl -fsSL "https://github.com/anthropics/anthropic-cli/releases/download/v${ANT_VERSION}/ant_${ANT_VERSION}_linux_${ARCH}.tar.gz" \
| tar -xz -C /usr/local/bin ant
WORKDIR /workspace
VOLUME /workspace
ENTRYPOINT ["ant", "beta:worker", "run"]然后编写一个派生脚本,将会话详细信息转发到新的沙箱中。轮询器会将 ANTHROPIC_SESSION_ID、ANTHROPIC_WORK_ID、ANTHROPIC_ENVIRONMENT_ID 和 ANTHROPIC_ENVIRONMENT_KEY 注入到脚本的环境中。ANTHROPIC_BASE_URL 是可选的,仅当它在轮询器主机上已设置时才会传递;它会覆盖默认的 API 端点。在示例中,/host/outputs 是您选择的主机目录;它被绑定挂载到沙箱的工作目录(/workspace),以便您在沙箱退出后检索会话交付物。在自托管环境中,智能体将交付物写入工作目录下而非 /mnt/session/outputs(参见沙箱文件系统),因此挂载工作目录才能捕获它们;该挂载还会获取下载的 skills/ 目录树以及智能体创建的任何中间文件。
#!/bin/bash
# spawn.sh:每个已认领的工作项调用一次
mkdir -p "/host/outputs/$ANTHROPIC_SESSION_ID"
exec docker run --rm \
-e ANTHROPIC_SESSION_ID -e ANTHROPIC_ENVIRONMENT_KEY \
-e ANTHROPIC_WORK_ID -e ANTHROPIC_ENVIRONMENT_ID -e ANTHROPIC_BASE_URL \
-v "/host/outputs/$ANTHROPIC_SESSION_ID":/workspace \
your-image启动指向该脚本的轮询器:
ant beta:worker poll \
--on-work ./spawn.shSDK 提供了三个不同控制级别的辅助工具。EnvironmentWorker 涵盖了大多数用例;当您需要启动自己的按会话进程或针对已认领的会话运行工具时,可降级使用较低级别的辅助工具。
EnvironmentWorker: 开箱即用的 worker。端到端处理轮询、设置和执行。
.run():无限期运行,在会话到达时拾取它们。.handle_item():处理单个已认领的工作项并退出。显式传递工作、会话和环境标识符,或让它读取 ant beta:worker poll --on-work 为其派生的进程设置的 ANTHROPIC_* 变量。work.poller(): 代表您轮询工作队列,并将每个已认领的会话交给您。当您想要决定每个会话的处理方式时使用此方法,例如启动沙箱而不是在进程内运行工具。
drain:是否在队列为空后停止轮询,而不是等待新工作。block_ms:等待工作到达的时长(毫秒),超时后返回。必须在 1 到 999 之间(单次轮询等待;辅助工具会自动重新轮询)。传递 null(Python 中为 None,Go 中为 param.Null[int64]())进行非阻塞检查;省略该参数则使用默认的 999 毫秒长轮询。reclaim_older_than_ms:重新认领在此毫秒数内已被认领但从未确认的工作项。auto_stop:是否在您的循环体处理完每个工作项后为其发布停止信号。Go 轮询器没有退出选项,始终会发布停止信号,因此请在循环体中阻塞直到会话完成,而不是分离。client.beta.sessions.events.tool_runner(): 给定会话 ID 和工具列表,为单个会话运行工具调用。当您已经认领了工作且只需要执行层时使用。当您想要启动自己的按会话进程时(例如为每个已认领的会话启动一个沙箱),直接使用工作轮询器:
import asyncio
import os
from anthropic import AsyncAnthropic
from anthropic.types.beta.environments import BetaSelfHostedWork
async def launch_container(work: BetaSelfHostedWork) -> None:
# 替换为您自己的每会话沙箱启动器。将
# ANTHROPIC_ENVIRONMENT_KEY 传入启动的沙箱,切勿传入
# 您的 API 密钥。
print(f"claimed session {work.data.id}")
async def main() -> None:
environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"]
environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"]
async with AsyncAnthropic(auth_token=environment_key) as client:
async for work in client.beta.environments.work.poller(
environment_id=environment_id,
environment_key=environment_key,
auto_stop=False, # the launched sandbox owns the stop call
):
await launch_container(work)
asyncio.run(main())AgentToolContext 是工具调用的执行上下文。它定义了工作目录和路径策略,并可以下载会话的技能。beta_agent_toolset_20260401(env) 接受一个 AgentToolContext 并返回标准工具实现(bash、read、write、edit、glob、grep)。
使用 EnvironmentWorker 时: 两者都会自动管理。传递 tools 工厂以自定义工具列表:
EnvironmentWorker(client, ..., tools=lambda env: [beta_bash_tool(env), my_custom_tool])使用 work.poller() 和 tool_runner() 时: 将工具列表作为 tools 传递给 client.beta.sessions.events.tool_runner()。要构建该列表,请自行设置 AgentToolContext 并调用 beta_agent_toolset_20260401(env):
from anthropic.lib.tools.agent_toolset import (
AgentToolContext,
beta_agent_toolset_20260401,
)
async with AgentToolContext(
workdir="/workspace", client=client, session_id=work.data.id
) as env:
# skills 已下载到 /workspace/skills/<name>/
tools = beta_agent_toolset_20260401(env)在另一个 shell 中,将 ANTHROPIC_API_KEY 设置为您的 Claude API 密钥(而非环境密钥),确认 workers_polling 至少为 1:
ant beta:environments:work stats --environment-id "$ANTHROPIC_ENVIRONMENT_ID"如果 workers_polling 保持为 0,则表示 worker 未连接到队列:请确认 worker 主机上已设置 ANTHROPIC_ENVIRONMENT_KEY 和 ANTHROPIC_ENVIRONMENT_ID。有关完整的统计响应和其他语言示例,请参阅读取队列深度。
worker 运行后,创建一个指向该环境的会话。将 AGENT_ID 设置为您在开始之前中记下的智能体 ID。会话进入环境的工作队列并在那里等待,直到有 worker 认领它;如果没有 worker 连接,会话会保持排队状态而不会失败。
Anthropic 不会将文件或 GitHub 仓库挂载到自托管沙箱中。要使会话特定的文件可用,请在会话的 metadata 字段中传递文件引用(例如 S3 路径或提交 SHA)。已认领的工作项不携带会话的元数据,但携带会话 ID:您的派生脚本或 --on-work 处理程序检索会话(GET /v1/sessions/{session_id})以读取 metadata 字段,然后在工具执行开始之前将文件暂存到工作目录中。
session = client.beta.sessions.create(
agent=agent.id,
environment_id=environment.id,
metadata={"input_file": "s3://my-bucket/data.csv"},
)有关 CLI 标志的完整列表,请参阅参考文档中的自托管 worker;有关 SDK 辅助工具选项,请参阅 SDK 辅助工具。
自定义工具是由您自己的代码执行的工具:智能体发出 agent.custom_tool_use 事件并等待匹配的 user.custom_tool_result。worker 可以充当该代码,并且由于它在您的沙箱内运行,该工具可以访问您为沙箱配置的内部服务、凭据和网络出口,仅此而已。环境密钥授权发布自定义工具结果,因此您的 Claude API 密钥无需存放在 worker 主机上。
在智能体上声明工具
向智能体的 tools 添加一个 custom 条目,其 name 与您的 worker 注册的工具匹配。有关完整的声明格式,请参阅自定义工具。
{
"type": "custom",
"name": "get_order_status",
"description": "Look up an order in the internal fulfillment system by order ID.",
"input_schema": {
"type": "object",
"properties": {
"order_id": { "type": "string", "description": "The order ID" }
},
"required": ["order_id"]
}
}向 worker 注册实现
通过 worker 的 tools 工厂(参见 SDK 辅助工具)传递该工具,与内置工具集一起:
import asyncio
import os
from anthropic import AsyncAnthropic, beta_async_tool
from anthropic.lib.environments import EnvironmentWorker
from anthropic.lib.tools.agent_toolset import beta_agent_toolset_20260401
@beta_async_tool
async def get_order_status(order_id: str) -> str:
"""Look up an order in the internal fulfillment system by order ID."""
# 在 worker 主机上运行:可调用沙箱能够访问的任何内容。
return f"Order {order_id}: shipped"
async def main() -> None:
environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"]
environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"]
async with AsyncAnthropic(auth_token=environment_key) as client:
await EnvironmentWorker(
client,
environment_id=environment_id,
environment_key=environment_key,
workdir="/workspace",
tools=lambda env: [*beta_agent_toolset_20260401(env), get_order_status],
).run()
asyncio.run(main())worker 只响应向其注册的工具。如果某个自定义工具在智能体上声明了但未向任何 worker 或客户端注册,会话将以 requires_action 停止原因暂停,直到有某个组件发布其结果;有关事件流程,请参阅处理自定义工具调用。
MCP 连接器从 Anthropic 一侧连接到 MCP 服务器,因此服务器必须暴露一个 Anthropic 可以访问的 HTTP 端点,无论是直接访问还是通过 MCP 隧道。要使用只有您的网络可以访问的服务器,请让 worker 充当 MCP 客户端,并将服务器的工具声明为自定义工具。MCP 服务器不需要来自您网络外部的入站连接;Anthropic 接收您在智能体上声明的工具定义、每次调用的输入以及您的 worker 回传的结果。在运行时,模型像调用任何其他自定义工具一样调用封装的工具:
agent.custom_tool_use 事件。user.custom_tool_result 发布。SDK 的客户端 MCP 辅助工具将服务器的工具转换为 worker 接受的可运行工具;请在 Anthropic SDK 之外安装 MCP SDK(pip install "anthropic[mcp]" "mcp>=1.24"、npm install @modelcontextprotocol/sdk、go get github.com/modelcontextprotocol/go-sdk)。示例在不进行身份验证的情况下连接;要发送凭据,请配置您传递给 MCP 传输层的 HTTP 客户端或请求选项(Python 中为 http_client,TypeScript 中为 requestInit,Go 中为 HTTPClient)。
在智能体上声明服务器的工具
列出 MCP 服务器的工具,并将每个工具声明为 custom 工具;MCP 的 name、description 和 inputSchema 一一对应到自定义工具的字段。如果服务器对其工具列表进行分页,请声明每一页;worker 必须列出相同的页面。
import asyncio
from typing import Any, cast
from anthropic import AsyncAnthropic
from anthropic.types.beta import BetaManagedAgentsCustomToolParams
from mcp import ClientSession, types
# 需要 mcp >= 1.24,该版本将 streamablehttp_client 重命名为 streamable_http_client。
from mcp.client.streamable_http import streamable_http_client
MCP_SERVER_URL = "http://mcp.internal.example.com:8000/mcp"
def to_custom_tool(tool: types.Tool) -> BetaManagedAgentsCustomToolParams:
# MCP 字段与自定义工具声明一一对应。cast
# 将 schema 字典原样传递给 SDK 的类型化参数。
return {
"type": "custom",
"name": tool.name,
"description": tool.description or tool.name,
"input_schema": cast(Any, tool.inputSchema),
}
async def main() -> None:
# 请在您创建代理的位置运行此代码,而不是在工作主机上:
# 它使用您的 Claude API 密钥(ANTHROPIC_API_KEY)进行身份验证。
async with (
streamable_http_client(MCP_SERVER_URL) as (read, write, _),
ClientSession(read, write) as mcp_session,
AsyncAnthropic() as client,
):
await mcp_session.initialize()
listed = await mcp_session.list_tools()
agent = await client.beta.agents.create(
name="Internal tools agent",
model="claude-opus-5",
tools=[
{"type": "agent_toolset_20260401"},
*[to_custom_tool(tool) for tool in listed.tools],
],
)
print(agent.id)
asyncio.run(main())从 worker 提供工具
在启动时连接到同一个 MCP 服务器,使用 MCP 辅助工具转换其工具,并将它们与内置工具集一起注册。在 worker 的整个生命周期内保持一个 MCP 会话打开。
import asyncio
import os
from datetime import timedelta
from anthropic import AsyncAnthropic
from anthropic.lib.environments import EnvironmentWorker
from anthropic.lib.tools.agent_toolset import beta_agent_toolset_20260401
from anthropic.lib.tools.mcp import async_mcp_tool
from mcp import ClientSession
# 需要 mcp >= 1.24,该版本将 streamablehttp_client 重命名为 streamable_http_client。
from mcp.client.streamable_http import streamable_http_client
MCP_SERVER_URL = "http://mcp.internal.example.com:8000/mcp"
async def main() -> None:
environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"]
environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"]
# 在启动时连接一次 MCP 服务器,并在工作进程的整个生命周期内
# 保持会话打开。超时设置会将挂起的工具调用转为错误
# 结果,而不是让调用一直停滞。
async with (
streamable_http_client(MCP_SERVER_URL) as (read, write, _),
ClientSession(read, write, read_timeout_seconds=timedelta(seconds=60)) as mcp_session,
AsyncAnthropic(auth_token=environment_key) as client,
):
await mcp_session.initialize()
listed = await mcp_session.list_tools()
mcp_tools = [async_mcp_tool(tool, mcp_session) for tool in listed.tools]
await EnvironmentWorker(
client,
environment_id=environment_id,
environment_key=environment_key,
workdir="/workspace",
tools=lambda env: [*beta_agent_toolset_20260401(env), *mcp_tools],
).run()
asyncio.run(main())封装 MCP 服务器时,请注意以下几点:
tools 数组最多包含 128 个条目(每个封装的工具是一个条目,内置工具集是另一个条目)。API 会拒绝重复使用工具名称的声明、以内置智能体工具(如 bash 或 read)命名自定义工具的声明,或使用保留的 mcp__ 前缀的声明。MCP 辅助工具保留服务器的名称和描述,因此请在需要时重命名或裁剪。当两个服务器暴露相同的工具名称时,请自行以带前缀的名称定义封装器,并让它调用服务器的原始工具名称。additionalProperties 和 title。它拒绝自定义工具 input_schema 中任何位置的引用关键字(如 $ref),因此请内联那些由 pydantic 等生成器提取到 $defs 中的 schema。它还拒绝顶层的 oneOf、anyOf 和 allOf,以及超出字母、数字、下划线、点和连字符范围的属性名称(1–64 个字符)。read_timeout_seconds 所示。如果没有设置超时,挂起的调用只有在 TypeScript MCP SDK 的默认请求超时触发(约一分钟)或 worker 自身的后备机制触发时才会变成错误结果:Python 中约为两分半钟,Go 中为两分钟——Go worker 会取消超过其 120 秒默认值的工具调用并发布错误结果。bash)的方式。仅声明您打算让智能体使用的工具。这些调用从您的监控或运维工具中运行,使用您的 Claude API 密钥进行身份验证,以观察和管理 worker 集群。认领和保活循环在 worker 辅助工具内部处理,因此您无需直接调用这些端点。
work.stats 返回环境的队列状态:
depth 是等待被认领的项目数量。根据此值扩展您的 worker 集群或对积压发出警报。pending 是已被 worker 认领但尚未确认的项目数量。worker 辅助工具在处理每个项目之前会先确认它,因此在正常运行中此值保持接近零;持续的非零值意味着某个 worker 在认领和确认之间停滞了。oldest_queued_at 是队列中最早项目的时间戳(等待被认领或已认领但尚未确认),如果没有则为 null。workers_polling 是过去 30 秒内进行过轮询的 worker 数量。使用此值进行存活性警报。import os
import anthropic
client = anthropic.Anthropic()
stats = client.beta.environments.work.stats(os.environ["ANTHROPIC_ENVIRONMENT_ID"])
print(f"depth={stats.depth} pending={stats.pending}"){
"type": "work_queue_stats",
"depth": 0,
"pending": 0,
"oldest_queued_at": null,
"workers_polling": 0
}使用 work.stop 请求处理特定会话的工作进程将其关闭。默认情况下,工作项会进入 stopping 状态:工作进程在下一次租约心跳时会注意到这一变化,取消该会话正在进行的工具调用,并确认关闭,此时工作项变为 stopped 状态。在请求正文中传递 force: true(使用 CLI 时传递 --force),可立即将工作项标记为 stopped,而无需等待工作进程的确认。
由于这些调用是从您的运维工具而非工作进程主机发起的,因此 ANTHROPIC_WORK_ID 不会被自动设置。在运行以下示例之前,请将其设置为目标工作项的 ID。要查找工作项的 ID,请通过环境工作端点列出该环境的工作项。
import os
import anthropic
client = anthropic.Anthropic()
work = client.beta.environments.work.stop(
os.environ["ANTHROPIC_WORK_ID"],
environment_id=os.environ["ANTHROPIC_ENVIRONMENT_ID"],
)
print(work.state)自托管沙箱环境的责任共担模型。
创建会话以运行您的智能体并开始执行任务。
将 Claude 安全地连接到在您的私有网络中运行的 MCP 服务器,无需开放入站端口或将服务暴露到公共互联网。
Was this page helpful?