預設情況下,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 觸發模式。兩者都可設定:請參閱參考文件中的自託管 worker 以了解 CLI 旗標,以及本頁的 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():處理單一已認領的工作項目並結束。明確傳遞 work、session 和 environment 識別碼,或讓它讀取 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 路徑或 commit 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"},
)請參閱參考文件中的自託管 worker 以了解完整的 CLI 旗標清單,以及 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:
# 請在您建立代理程式的地方執行,而非在 worker 主機上:
# 它會使用您的 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 伺服器一次,並在 worker 的整個生命週期中
# 保持工作階段開啟。逾時設定會將卡住的工具呼叫轉為錯誤
# 結果,而非停滯的呼叫。
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 中的結構描述內聯化。它也會拒絕頂層的 oneOf、anyOf 和 allOf,以及超出字母、數字、底線、句點和連字號範圍的屬性名稱(1–64 個字元)。read_timeout_seconds 所示。若未設定,掛起的呼叫只有在 TypeScript MCP SDK 的預設請求逾時觸發(約一分鐘)或 worker 自己的後備機制觸發時才會變成錯誤結果:Python 中約兩分半鐘,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 要求處理特定工作階段的 worker 將其關閉。預設情況下,work item(工作項目)會進入 stopping 狀態:worker 會在下一次 lease heartbeat(租約心跳)時察覺,取消該工作階段正在執行中的工具呼叫,並確認關閉,此時 work item 會變為 stopped。在請求主體中傳遞 force: true(使用 CLI 時,傳遞 --force)可立即將 work item 標記為 stopped,而無需等待 worker 的確認。
由於這些呼叫是從您的維運工具執行,而非從 worker 主機執行,因此 ANTHROPIC_WORK_ID 不會自動設定。在執行以下範例之前,請將其設定為目標 work item 的 ID。若要尋找 work item 的 ID,請透過 Environments Work 端點列出環境的 work item。
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?