Managed Agents自托管沙箱

自托管沙箱中的自定义工具

Copy page



从自托管沙箱 worker 提供自定义工具，并将您网络内的 MCP 服务器封装为自定义工具，无需运行隧道。

自定义工具是由您自己的代码执行的工具：智能体发出 agent.custom_tool_use 事件，并等待匹配的 user.custom_tool_result。您的 worker 可以充当这段代码。由于它在您的沙箱内运行，该工具可以访问您为沙箱配置的内部服务、凭据和网络出口（network egress），仅此而已。

环境密钥授权发布自定义工具结果，因此您的 Claude API 密钥无需放在 worker 主机上。

提供自定义工具

1. 1

   在智能体上声明工具

   在智能体的 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"]
     }
   }

   
2. 2

   向 worker 注册实现

   通过 worker 的 tools 工厂（参阅 EnvironmentWorker）传入该工具，并与内置工具集一起使用：

   PythonTypeScriptC#GoJavaPHPRuby

   

   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 服务器封装为自定义工具

MCP 连接器从 Anthropic 一侧连接到 MCP 服务器。因此，服务器必须公开一个 Anthropic 可以访问的 HTTP 端点，可以是直接访问，也可以通过 MCP 隧道访问。

要使用只有您的网络才能访问的服务器，请改为让 worker 充当 MCP 客户端，并将服务器的工具声明为自定义工具。MCP 服务器不需要来自您网络外部的入站连接。Anthropic 会接收您在智能体上声明的工具定义、每次调用的输入，以及您的 worker 回传的结果。

在运行时，模型会像调用任何其他自定义工具一样调用封装的工具：

1. 智能体发出 agent.custom_tool_use 事件。
2. 位于您沙箱内的 worker 通过其已打开的 MCP 会话，将调用转发到您网络上的服务器。
3. worker 将服务器的响应作为 user.custom_tool_result 发布。

安装 MCP SDK

SDK 的客户端 MCP 辅助函数会将服务器的工具转换为 worker 可接受的可运行工具。请在 Anthropic SDK 之外安装一个 MCP SDK：pip install "anthropic[mcp]" "mcp>=1.24"。

示例在不进行身份验证的情况下连接。要发送凭据，请配置您传递给 MCP 传输层的 http_client。

声明并提供工具

1. 1

   在智能体上声明服务器的工具

   列出 MCP 服务器的工具，并将每个工具声明为 custom 工具。MCP 的 name、description 和 inputSchema 与自定义工具的字段一一对应。如果服务器对其工具列表进行分页，请声明每一页；worker 必须列出相同的页面。

   PythonTypeScriptC#GoJavaPHPRuby

   

   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-5",
               tools=[
                   {"type": "agent_toolset_20260401"},
                   *[to_custom_tool(tool) for tool in listed.tools],
               ],
           )
           print(agent.id)
   
   
   asyncio.run(main())

2. 2

   从 worker 提供工具

   在启动时连接到同一个 MCP 服务器，使用 async_mcp_tool 转换其工具，并将它们与 beta_agent_toolset_20260401 一起注册。在 worker 的整个生命周期内保持一个 MCP 会话处于打开状态。

   PythonTypeScriptC#GoJavaPHPRuby

   

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

限制与行为

工具需要声明，而非在运行时发现

worker 在启动时列出一次 MCP 服务器的工具，并且无法向正在运行的会话添加工具。当服务器的工具发生变化时：

1. 重新声明这些工具，可以在智能体上声明，也可以通过更新智能体配置在空闲会话上声明。
2. 重启 worker。

声明必须符合 Managed Agents API 的要求

MCP 辅助工具会保留服务器的名称和描述，大多数 schema 会原样传递。如果某个声明违反了以下规则之一，请进行重命名、精简或内联：

有两种情况需要额外处理：

• 两个服务器公开了相同的工具名称： 自行以带前缀的名称定义封装器，并让它调用服务器的原始工具名称。
• pydantic 等生成器将 schema 提取到 $defs 中： 在声明工具之前内联这些 schema。

工具失败以错误工具结果的形式呈现

当 MCP 服务器报告工具错误时，worker 会发布一个模型可以做出响应的错误工具结果。没有对应工具结果形式的 MCP 内容（例如音频块和资源链接）也会以错误形式呈现。

在 MCP 客户端上设置超时，可以更快、更清晰地暴露失败，就像 Python worker 示例中使用 read_timeout_seconds 那样。有关未设置超时时会发生什么，请参阅封装的 MCP 工具调用挂起。

仅封装您运营或信任的服务器

封装工具的名称、描述和结果会像任何其他工具一样进入模型的上下文。它们属于不受信任的输入，可能会影响智能体如何使用其他工具，包括 worker 主机上的 bash。请仅声明您希望智能体使用的工具。

权限策略不适用

权限策略管控内置工具集和 MCP 工具集。worker 会执行模型发起的每一个封装工具调用，因此请将任何审批步骤放在您自己的工具代码中。