自行託管的沙箱
在自行託管的沙箱中執行 Claude Managed Agents 工作階段,將工具執行、檔案與網路出口流量保留在您自己的基礎架構中。
預設情況下,Managed Agents 會在 Anthropic 管理的雲端沙箱內執行工具與程式碼。「Self-hosted sandboxes」(自行託管的沙箱)將編排(orchestration)保留在 Anthropic 端,但將工具執行移至您所控制的基礎架構中,因此代理程式的程式碼、檔案系統與網路出口流量永遠不會離開您的環境。
工具執行會留在您的主機上:代理程式讀寫的檔案系統、它所產生的行程,以及它能連線的網路,全都在您的控制之下。工具的輸入與輸出仍會流向 Anthropic 的控制平面(Claude 執行之處),讓模型能看到結果並決定下一步該做什麼。代理程式的 skills(技能)以及附加至工作階段的任何 memory stores(記憶儲存區)的內容由 Anthropic 儲存,並在工作階段期間複製到您的沙箱中;代理程式對記憶檔案所做的變更會同步回儲存區。完整的資料流邊界請參閱安全模型。
與雲端環境的差異
| 雲端環境 | 自行託管的沙箱 | |
|---|---|---|
| 工具執行位置 | Anthropic 管理的沙箱 | 您的基礎架構 |
| 網路可達範圍 | Anthropic 的出口流量控制 | 您的網路政策 |
| 檔案與 GitHub 儲存庫掛載 | 由 Anthropic 管理 | 由您管理 |
| 記憶儲存區 | 由 Anthropic 掛載於 /mnt/memory/ | 下載至 /mnt/memory/ 並由 SDK worker 同步 |
| 生命週期 | 由 Anthropic 管理 | 由您管理 |
當代理程式需要處理不能離開您網路邊界的資料、連線至無法公開路由的內部服務,或在您組織自身的合規與稽核控制下執行時,自行託管是很合適的選擇。
關於零資料保留(Zero Data Retention)與 HIPAA BAA 的適用資格,請參閱 API 與資料保留。
何時與 MCP 通道搭配使用
自行託管控制的是代理程式的程式碼在哪裡執行。MCP 通道控制的是 Anthropic 如何連線至您網路中的 MCP 伺服器。兩者彼此獨立:在 Anthropic 雲端沙箱中執行的工作階段仍可透過通道連線至私有 MCP 伺服器,而自行託管的工作階段則可使用經通道連線或公開的 MCP 伺服器。當您希望執行與工具存取都留在您的邊界內時,請同時使用兩者。若要在不執行通道的情況下,讓代理程式取得您網路內 MCP 伺服器的工具,您也可以將該伺服器包裝為自訂工具,由您的 worker 提供服務。
環境 worker
「Environment worker」(環境 worker)是您在自己的基礎架構上執行的行程。它接收來自 Anthropic 的工具執行請求並在本機執行。self_hosted 環境的作用如同一個工作佇列:當某個 session(工作階段)被指派給它時,Anthropic 會將該工作階段作為工作項目加入佇列。您的 worker 從該佇列認領工作項目、為每個項目產生一個執行上下文、下載代理程式的 skills(可重複使用、以檔案系統為基礎的資源,賦予代理程式特定領域的專業能力)、執行工具呼叫,並將結果回傳。
工作項目是透過輪詢環境的佇列來認領的:可以是持續輪詢的常駐 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 能找到技能檔案。- 輸出: 在自行託管的環境中,工作階段的系統提示會省略 Anthropic 管理的沙箱所使用的
/mnt/session/outputs指示,因此最終交付成果會落在代理程式於您沙箱檔案系統中寫入的任何位置,通常位於工作目錄之下。 /mnt/memory/: 附加至工作階段的記憶儲存區由 SDK worker 在此具體化,每個儲存區一個目錄,位於該儲存區的mount_path(例如/mnt/memory/user-preferences/)。worker 在認領工作階段時建立這些目錄,並在工作階段結束時移除它們;請參閱使用記憶儲存區。
開始之前
您需要:
- 一個現有的代理程式。 若您還沒有,請先完成快速入門並記下其代理程式 ID。
- 一台 Linux 主機,且
/bin/bash位於該確切路徑。worker 的 bash 工具會直接呼叫它,而不查詢PATH。TypeScript SDK 另外要求PATH上有unzip與tar,以及 Node.js 22 或更新版本;Python 與 Go SDK 使用其標準函式庫進行封存檔解壓縮,沒有額外的二進位檔需求。 antCLI 或 Anthropic SDK(Python、TypeScript 或 Go),安裝於 worker 主機上。- 憑證: 環境金鑰(在後續步驟中於 Console 產生)用於向其佇列驗證 worker 的身分;您的 Claude API 金鑰則用於從 worker 主機外部建立工作階段與讀取佇列統計資料。金鑰產生僅能在 Console 進行。已認領的工作項目還會攜帶一個每工作階段專屬的
secret,worker 用它來掛載記憶儲存區;您不需要產生它,但在每工作階段一個沙箱的模式中,您需要自行將它轉送進沙箱(請參閱每個工作階段執行一個沙箱)。 - 若使用記憶儲存區,需要一台已準備好的主機。 若此環境上的工作階段會附加記憶儲存區,請在啟動 worker 之前於 worker 主機上準備好
/mnt/memory;請參閱準備主機。
建立自行託管的環境
在 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_..."
執行 worker
若要最簡單的設定,請選擇常駐模式:一個長時間執行的行程持續輪詢佇列,且只需要對外的 HTTPS 連線。若要避免執行閒置的輪詢器,請選擇 webhook 觸發模式;它需要一個 Anthropic 能連線的 webhook 端點(端點設定與簽章驗證請參閱 Webhooks)。
安裝 ant CLI
在 worker 主機上執行此操作。
對於 Linux 環境,請直接下載發行版二進位檔。
VERSION=1.27.0 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.27.0 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"]接著撰寫一個產生(spawn)腳本,將工作階段詳細資訊轉送進全新的沙箱。輪詢器會將
ANTHROPIC_SESSION_ID、ANTHROPIC_WORK_ID、ANTHROPIC_ENVIRONMENT_ID與ANTHROPIC_ENVIRONMENT_KEY注入腳本的環境中,並將已認領的工作項目以 JSON 格式寫入腳本的標準輸入,其中包含工作項目的每工作階段secret(若 Anthropic 有核發)。ANTHROPIC_BASE_URL為選用,僅在輪詢器主機上有設定時才會傳遞;它會覆寫預設的 API 端點。在範例中,/host/outputs是您選擇的主機目錄;它以 bind-mount 方式掛載至沙箱的工作目錄(/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-imageant beta:worker run進入點不會掛載記憶儲存區。若此環境上的工作階段會附加記憶儲存區,請保留輪詢器,但以 SDK worker 為核心建置每工作階段的映像檔,並擴充產生腳本以將工作項目的secret轉送進沙箱,如每個工作階段執行一個沙箱所示。啟動指向該腳本的輪詢器:
ant beta:worker poll --on-work ./spawn.sh
SDK 輔助工具
SDK 提供三個不同控制層級的輔助工具。EnvironmentWorker 涵蓋大多數使用情境;當您需要啟動自己的每工作階段行程,或對已認領的工作階段執行工具時,請改用較低層級的輔助工具。
EnvironmentWorker: 開箱即用的 worker。端到端處理輪詢、設定與執行。.run():無限期執行,在工作階段到達時接手處理。.handle_item():處理單一已認領的工作項目後結束。可明確傳入工作、工作階段與環境識別碼,或讓它讀取ant beta:worker poll --on-work為其產生的行程所設定的ANTHROPIC_*變數。若要讓工作階段掛載其記憶儲存區,還需將工作項目的secret以work_secret(TypeScript 中為workSecret,Go 中為WorkSecret)傳入,或設定ANTHROPIC_WORK_SECRET;ant beta:worker poll --on-work不會設定該變數,因此請從它寫入您腳本標準輸入的工作項目 JSON 中讀取 secret,如每個工作階段執行一個沙箱所示。memory_sync_interval(TypeScript 中為memorySyncIntervalMs,Go 中為MemorySyncInterval)與memory_sync_deletions(memorySyncDeletions、MemorySyncDeletions):工作階段執行期間,附加的記憶儲存區與伺服器進行協調的頻率,以及代理程式在本機刪除的檔案是否也從儲存區中刪除。單位、預設值以及如何停用記憶支援,請參閱設定同步。
work.poller(): 代您輪詢工作佇列,並將每個已認領的工作階段交給您。當您想自行決定每個工作階段的處理方式時使用,例如啟動沙箱而非在行程內執行工具。drain:佇列清空後是否停止輪詢,而非等待新工作。block_ms:在返回前等待工作到達的時間,以毫秒為單位。必須介於 1 到 999 之間(每次輪詢的等待時間;輔助工具會自動重新輪詢)。傳入null(Python 中為None,Go 中為param.Null[int64]())進行非阻塞檢查;省略此參數則使用預設的 999 毫秒長輪詢。reclaim_older_than_ms:重新認領已被認領但在此毫秒數內從未被確認的工作項目。auto_stop(TypeScript 中為autoStop,Go 中為AutoStop):您的迴圈主體處理完每個工作項目後,是否為其發送停止訊號。只要執行工作項目的程式會自行發送停止訊號,就請關閉它:handle_item()會這麼做,因此當您像本頁的 webhook 處理程式一樣將已認領的項目交給handle_item()時,請將它設為 false;您啟動的、自行負責停止呼叫的沙箱也是如此。
client.beta.sessions.events.tool_runner(): 給定工作階段 ID 與工具清單,為單一工作階段執行工具呼叫。當您已認領工作且只需要執行層時使用。
當您想啟動自己的每工作階段行程時,請直接使用工作輪詢器,例如為每個已認領的工作階段啟動一個沙箱:
import asyncio
import os
from anthropic import AsyncAnthropic
from anthropic.types.beta.environments import BetaSelfHostedWork
SANDBOX_ENV = (
"ANTHROPIC_ENVIRONMENT_ID",
"ANTHROPIC_ENVIRONMENT_KEY",
"ANTHROPIC_WORK_ID",
"ANTHROPIC_SESSION_ID",
"ANTHROPIC_WORK_SECRET",
"ANTHROPIC_BASE_URL", # forwarded only when set on this host
)
async def launch_container(work: BetaSelfHostedWork) -> None:
print(f"claimed session {work.data.id}")
# 請將 `docker run` 替換為您自己的沙箱啟動器。轉發環境
# 金鑰(絕非您的 API 金鑰)以及工作項目的每個工作階段密鑰:內部的 worker
# 需要該密鑰才能掛載工作階段的記憶體儲存區。
env = os.environ | {
"ANTHROPIC_WORK_ID": work.id,
"ANTHROPIC_SESSION_ID": work.data.id,
"ANTHROPIC_WORK_SECRET": work.secret or "",
}
forward = [arg for name in SANDBOX_ENV for arg in ("-e", name)]
launcher = await asyncio.create_subprocess_exec(
"docker", "run", "--rm", "--detach", *forward, "your-sdk-worker-image", env=env
)
await launcher.wait()
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())無論由什麼啟動沙箱,都必須將已認領工作項目的 secret 連同工作階段、工作與環境識別碼一起轉送進沙箱(例如作為 ANTHROPIC_WORK_SECRET),讓沙箱內的 worker 能掛載工作階段的記憶儲存區;請參閱每個工作階段執行一個沙箱。
AgentToolContext 是工具呼叫的執行上下文。它定義工作目錄與路徑政策,並可下載工作階段的技能。檔案工具(read、write、edit、glob、grep)被限制在工作目錄加上 allowed_roots(TypeScript 中為 allowedRoots,Go 中為 AllowedRoots)所列的任何目錄內,而 write 與 edit 另外會拒絕 read_only_roots(readOnlyRoots、ReadOnlyRoots)之下的路徑。EnvironmentWorker 會自行將工作階段的記憶儲存區目錄加入這些清單。此限制僅是檔案工具的防護措施,而非沙箱;它不會約束 bash。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)驗證 worker 已連線
在另一個 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)。已認領的工作項目不會攜帶工作階段的 metadata,但會攜帶工作階段 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 輔助工具。
使用記憶儲存區
自行託管環境上的工作階段附加記憶儲存區的方式與雲端環境上的工作階段完全相同:建立工作階段時將它們列於 resources 中,如將記憶儲存區附加至工作階段所示。一個工作階段最多可接受 8 個記憶儲存區。在自行託管的環境中,是由 SDK worker 而非 Anthropic 的基礎架構為代理程式具體化每個儲存區,因此該處的記憶儲存區需要 Python、TypeScript 或 Go SDK 的 EnvironmentWorker(或其 handle_item() 方法)。
ant CLI worker(ant beta:worker poll 與 ant beta:worker run)不會掛載記憶儲存區。若要將 CLI 輪詢器與記憶儲存區搭配使用,請依每個工作階段執行一個沙箱所述,在每工作階段的沙箱內執行 SDK worker。
在 Claude Platform on AWS 上,記憶儲存區無法附加至自行託管環境上的工作階段。
worker 如何處理記憶
當 worker 認領一個其工作階段附加了記憶儲存區的工作項目時,它會:
- 將每個附加的儲存區下載至 worker 主機上的
mount_path,並以工作項目的每工作階段secret進行驗證。mount_path與雲端工作階段所使用的/mnt/memory/下的目錄相同(例如,名為「User Preferences」的儲存區為/mnt/memory/user-preferences/),且工作階段的系統提示會向代理程式描述它。 - 將這些目錄加入檔案工具的允許根目錄,並將以
access: "read_only"附加的儲存區目錄加入其唯讀根目錄,讓代理程式能以在工作目錄中使用的相同read、write、edit、glob與grep工具處理記憶。 - 在工具呼叫後協調本機與遠端的變更,每個同步間隔(預設 15 秒)最多一次:儲存區中已變更的記憶會寫入磁碟,而代理程式變更的檔案會上傳至儲存區。
- 在工作階段結束時執行最終同步,最多花 30 秒清空任何仍待處理的上傳,然後移除它所建立的目錄。在工作階段執行期間被取消的 worker 會略過最終同步,但仍會在結束前上傳已變更的檔案並移除目錄。
Anthropic 端的記憶儲存區仍是唯一真實來源。記憶版本、遮蔽(redaction),以及在 Console 中檢視或編輯記憶的運作方式與雲端工作階段相同,而代理程式的記憶讀寫會以一般工具事件的形式出現在事件串流中。由於每個 worker 依間隔同步,在一個工作階段中寫入的變更只有在兩者都同步後,才會對另一個執行中的工作階段可見,在預設間隔下通常遠低於一分鐘;雲端沙箱上的工作階段則幾乎能立即看到彼此的變更。
每個儲存區目錄都包含一個名為 .anthropic-memory-store 的標記檔案,將該目錄與其儲存區綁定。請保留它:worker 不會同步標記遺失或遭更動的目錄。
準備主機
自行託管沙箱上的記憶儲存區需要 worker 主機(開始之前中的 Linux 主機)上有 POSIX 檔案系統;不支援 Windows 主機,因為 worker 在開啟記憶檔案時需要 O_NOFOLLOW。建議使用區分大小寫的檔案系統,以免僅大小寫不同的記憶路徑發生衝突。
在啟動 worker 之前,建立父目錄並讓 worker 執行時所使用的使用者可寫入:
sudo mkdir -p /mnt/memory && sudo chown "$USER" /mnt/memory請勿自行建立每個儲存區的目錄。worker 會在工作階段啟動時建立每個儲存區的 mount_path 目錄(例如 /mnt/memory/user-preferences),若該路徑已存在任何內容則拒絕啟動該工作階段的工作,並在工作階段結束時移除該目錄。由此衍生出兩條操作規則:
- 當工作階段附加相同儲存區時,每個檔案系統執行一個工作階段。 兩個工作階段無法同時在一台主機上掛載相同的儲存區,因為兩者需要相同的路徑。依每個工作階段執行一個沙箱所述,為每個工作階段提供各自的沙箱即可滿足此規則。
- 優雅地停止 worker。 當您在工作階段執行期間停止 worker 時,
EnvironmentWorker只有在被取消而非被強制終止的情況下,才會上傳工作階段已變更的記憶檔案並移除其儲存區目錄:被強制終止的行程不會執行任何清理,且 worker 本身不會安裝訊號處理程式。請在執行它的行程中將 SIGTERM 與 SIGINT 連接至取消動作:在 TypeScript 中中止您傳給 worker 的signal,在 Go 中取消 context,在 Python 中取消執行run()或handle_item()的 task。當您的 worker 本身就是該行程時(如本頁的獨立 worker),請從訊號處理程式執行此操作;當 worker 在 webhook 處理程式內執行時,則從您伺服器自身的關閉掛鉤執行,因為它不得接管伺服器的訊號。接著以 SIGTERM 停止 worker,並在任何強制終止前給予至少 30 秒的結束時間,因為最終上傳可能需要那麼久。若 worker 在其清理執行前被強制終止,請在下一個附加該儲存區的工作階段之前,移除/mnt/memory/下殘留的儲存區目錄;其中任何尚未同步的編輯都會遺失。
每個工作階段執行一個沙箱
執行 worker 中的「每個工作階段一個沙箱」模式會為每個工作階段提供全新的檔案系統,這正是當多個工作階段附加同一個儲存區時,準備主機 所要求的。請在主機上保留 ant beta:worker poll --on-work(或 SDK 的工作輪詢器)作為輪詢器。
該處所示的 ant beta:worker run 進入點不會掛載記憶儲存區,因此請改以 SDK worker 為核心建構每個工作階段的映像檔:其進入點會建構 EnvironmentWorker 並呼叫 handle_item()(TypeScript 中為 handleItem,Go 中為 HandleItem),它會從 ANTHROPIC_* 變數讀取工作階段、工作與環境識別碼,並從 ANTHROPIC_WORK_SECRET 讀取工作項目的每工作階段 secret。您也可以將 secret 明確地以 work_secret(TypeScript 中為 workSecret,Go 中為 WorkSecret)傳入。
import asyncio
import contextlib
import os
import signal
from anthropic import AsyncAnthropic
from anthropic.lib.environments import EnvironmentWorker
async def main() -> None:
async with AsyncAnthropic(auth_token=os.environ["ANTHROPIC_ENVIRONMENT_KEY"]) as client:
worker = EnvironmentWorker(client, workdir="/workspace")
# 不帶引數時,handle_item() 會讀取 spawn 指令碼所轉發的
# ANTHROPIC_* 變數,包括 ANTHROPIC_WORK_SECRET。
task = asyncio.create_task(worker.handle_item())
# 在容器停止時取消任務,可讓 worker 在結束前上傳
# 已變更的記憶檔案並移除儲存目錄。
loop = asyncio.get_running_loop()
for signum in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(signum, task.cancel)
with contextlib.suppress(asyncio.CancelledError):
await task
asyncio.run(main())ant beta:worker poll --on-work 不會為其產生的腳本設定 ANTHROPIC_WORK_SECRET,因此產生腳本會從其標準輸入上的工作項目 JSON 讀取 secret,並將其傳入沙箱:
#!/bin/bash
# spawn.sh:每個已認領的工作項目呼叫一次
# 已認領的工作項目以 JSON 形式透過 stdin 傳入。其 secret 是
# 記憶體儲存端點所需的每個工作階段憑證。
ANTHROPIC_WORK_SECRET="$(jq -r '.secret // empty')"
export ANTHROPIC_WORK_SECRET
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 \
-e ANTHROPIC_WORK_SECRET \
-v "/host/outputs/$ANTHROPIC_SESSION_ID":/workspace \
your-sdk-worker-image如果您改用 SDK 的工作輪詢器來領取工作,請以相同方式將每個已領取項目的 secret 傳入您啟動的沙箱。僅將其傳入服務該工作階段的沙箱,且絕不要將其記錄到日誌中。
沙箱映像檔也需要一個可寫入的 /mnt/memory(請參閱準備主機)。由於每個沙箱只服務一個工作階段並在之後被丟棄,因此沒有殘留目錄需要清理,記憶目錄也不需要綁定掛載到主機:worker 會在沙箱結束前將其內容上傳到儲存區。如果您在工作階段結束前停止容器,請傳送一個進入點會轉換為取消的訊號(請參閱準備主機),而不是直接終止它,以便上傳仍能執行。同時也要給容器足夠的時間完成上傳:Docker 預設會在停止訊號後 10 秒送出 SIGKILL,因此請使用 docker run 的 --stop-timeout 或您的編排器的終止寬限期,將該限制提高到至少「準備主機」所要求的 30 秒。
設定同步
兩個 EnvironmentWorker 選項控制記憶行為:
memory_sync_interval(Python,以秒為單位;TypeScript 中為memorySyncIntervalMs,以毫秒為單位;Go 中為MemorySyncInterval,為一個 duration):工作階段執行期間,已附加的儲存區與伺服器協調的頻率。預設為 15 秒;最小值為 5 秒。較短的間隔會縮小另一個工作階段看到過時記憶的時間窗口,代價是更多的記憶儲存區請求。Python 中的None、TypeScript 中的null或 Go 中的負 duration 會完全停用記憶支援:worker 既不下載也不同步儲存區,而附加了記憶儲存區的工作階段會在沒有它們的情況下執行,即使其系統提示仍然描述了它們,因此請僅在其工作階段不附加任何記憶儲存區的 worker 上停用記憶支援。在記憶支援啟用時,對於附加了儲存區的工作階段,若工作項目抵達時沒有每工作階段的secret,則會失敗而非在沒有記憶的情況下執行(請參閱疑難排解記憶掛載)。memory_sync_deletions(TypeScript 中為memorySyncDeletions,Go 中為MemorySyncDeletions):代理在本機刪除的檔案是否也從儲存區中刪除。在 Python 和 TypeScript 中,其值為"enabled"(預設)、"log_only"或"disabled"之一;在 Go 中則為常數environments.MemorySyncDeletionsEnabled(零值)、environments.MemorySyncDeletionsLogOnly或environments.MemorySyncDeletionsDisabled之一。啟用時,一旦後續同步確認檔案仍然不存在,worker 就會從儲存區中刪除該記憶;在僅記錄模式下,它會執行相同的檢查,但只記錄它本來會刪除的內容,這讓您可以在信任啟用模式之前觀察您的 worker 會刪除什麼;停用時,它永遠不會從儲存區中刪除。上傳和下載不受此設定影響。
請在您建構 worker 的地方設定這些選項,無論是透過 EnvironmentWorker 建構函式,或是在 Python 和 TypeScript 中透過 webhook 處理程式所使用的 client.beta.environments.work.worker() 工廠函式。
例如,要每 10 秒同步一次,並且只記錄 worker 本來會執行的刪除:
worker = EnvironmentWorker(
client,
environment_id=environment_id,
environment_key=environment_key,
workdir="/workspace",
memory_sync_interval=10, # seconds
memory_sync_deletions="log_only",
)唯讀儲存區與衝突
對於以 access: "read_only" 附加的儲存區,write 和 edit 工具會拒絕變更其目錄內的檔案,且 worker 永遠不會從中上傳任何內容。透過 bash,或透過您從沙箱提供的自訂工具或 MCP 伺服器所做的變更,在本機不會被阻擋:它們永遠不會同步到儲存區,且對該記憶的下一次遠端變更會覆寫它們。如果您需要本機副本本身在工作階段期間保持不變,請為該代理停用 bash 工具,並且不要給它任何會寫入沙箱檔案系統的自訂工具;不要以唯讀方式掛載儲存區路徑,因為 worker 本身必須建立該目錄並將下載的記憶寫入其中。
衝突的解決以儲存區為準。當代理變更了一個自工作階段上次同步以來在儲存區中也已變更的記憶檔案時,worker 會在下一次同步時保留儲存區的版本,以其覆寫本機檔案,並記錄一則警告;write 和 edit 工具本身會成功,且不會有錯誤傳達給代理。如果代理的變更仍然適用,它可以在同步後重新讀取檔案並再次進行變更。
疑難排解記憶掛載
worker 會記錄掛載和背景同步失敗,而不是將其回報給工作階段;只有唯讀拒絕會以工具錯誤的形式傳達給代理(請參閱唯讀儲存區與衝突)。如果在 worker 領取工作階段時無法掛載記憶儲存區,worker 會使該工作項目失敗:工作階段不會發出錯誤事件並保持閒置。
| 症狀 | 原因 | 修正方式 |
|---|---|---|
worker 日誌包含 the work item carried no sessions token(在 Go 中為 ErrSessionMemoryNoToken 錯誤),且工作項目失敗。 | 工作項目的每工作階段 secret 未送達 worker:您的組織未啟用自託管沙箱上的記憶儲存區,或您的產生腳本未將 secret 轉送到沙箱中。 | 在每個工作階段一個沙箱的模式中,請如每個工作階段執行一個沙箱所示,將 ANTHROPIC_WORK_SECRET 轉送到沙箱中。如果 worker 在同一個程序中輪詢並執行工作階段,但仍記錄此訊息,請聯絡支援。 |
worker 日誌包含 something already exists at the memory store's path。 | 先前工作階段殘留的目錄,通常是其 worker 在拆除程序執行前就被終止的工作階段。 | 移除日誌行所指名的殘留目錄。其中尚未同步的編輯將會遺失。 |
worker 日誌包含 cannot create the memory store's folder 和 the worker host must make this mount path writable。 | worker 執行所用的使用者無法在 /mnt/memory 下建立目錄。 | 建立 /mnt/memory 並將其 chown 給該使用者;請參閱準備主機。 |
工作階段在 worker 領取後不久即處於 idle 狀態,停止原因為 requires_action,且沒有錯誤事件。 | worker 因上述原因之一無法掛載記憶儲存區,而使工作項目失敗。 | 在主機上修正原因,然後傳送一個 user.interrupt 事件:工作階段的工作會再次排入佇列,下一個領取它的 worker 會重試掛載。 |
從您的沙箱提供自訂工具
自訂工具是由您自己的程式碼執行的工具:代理發出一個 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 伺服器包裝為自訂工具
MCP 連接器從 Anthropic 端連線到 MCP 伺服器,因此伺服器必須公開一個 Anthropic 可以直接或透過 MCP 通道存取的 HTTP 端點。若要使用只有您的網路可以存取的伺服器,請改讓 worker 成為 MCP 用戶端,並將伺服器的工具宣告為自訂工具。MCP 伺服器不需要來自您網路外部的入站連線;Anthropic 會收到您在代理上宣告的工具定義、每次呼叫的輸入,以及您的 worker 回傳的結果。在執行時,模型會像呼叫任何其他自訂工具一樣呼叫被包裝的工具:
- 代理發出一個
agent.custom_tool_use事件。 - worker 在您的沙箱內,透過其開啟的 MCP 工作階段將呼叫轉送到您網路上的伺服器。
- worker 將伺服器的回應發布為
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 伺服器時,請記住以下幾點:
- 工具是宣告的,而非在執行時探索的。 worker 在啟動時列出 MCP 伺服器的工具一次,且無法向執行中的工作階段新增工具。當伺服器的工具變更時,請在代理上或透過更新代理設定在閒置的工作階段上再次宣告它們,並重新啟動 worker。
- 名稱和描述必須符合 Managed Agents API。 自訂工具名稱在每個代理中是唯一的,並使用字母、數字、底線和連字號(1–128 個字元);需要非空的描述;且代理的
tools陣列最多接受 128 個項目(每個被包裝的工具是一個項目,內建工具集是另一個)。API 會拒絕重複使用工具名稱、以內建代理工具(例如bash或read)命名自訂工具,或使用保留的mcp__前綴的宣告。MCP 輔助工具會保留伺服器的名稱和描述,因此請在需要時重新命名或修剪。當兩個伺服器公開相同的工具名稱時,請自行以帶前綴的名稱定義包裝器,並讓它呼叫伺服器的原始工具名稱。 - 大多數結構描述會原封不動地通過。 API 接受 MCP 伺服器通常發出的 JSON Schema 關鍵字,例如
additionalProperties和title。它會拒絕自訂工具input_schema中任何位置的參照關鍵字(例如$ref),因此請將 pydantic 等產生器分解到$defs中的結構描述內嵌。它也會拒絕頂層的oneOf、anyOf和allOf,以及字母、數字、底線、點和連字號(1–64 個字元)以外的屬性名稱。 - 工具失敗會以錯誤工具結果的形式呈現。 當 MCP 伺服器回報工具錯誤時,worker 會發布一個模型可以回應的錯誤工具結果。沒有對應工具結果的 MCP 內容(例如音訊區塊和資源連結)也會以錯誤的形式呈現。請在 MCP 用戶端上設定逾時,以獲得更快、更清楚的失敗,如 Python worker 範例使用
read_timeout_seconds所做的那樣。若沒有設定,掛起的呼叫只有在 TypeScript MCP SDK 的預設請求逾時觸發時(約一分鐘),或在 worker 自己的後備機制觸發時才會變成錯誤結果:Python 中約兩分半鐘,Go 中為兩分鐘,此時 worker 會取消超過其 120 秒預設值的工具呼叫並發布錯誤結果。 - 包裝您營運或信任的伺服器。 被包裝工具的名稱、描述和結果會像任何其他工具一樣進入模型的上下文:不受信任的輸入可能影響代理使用其他工具(包括 worker 主機上的
bash)所做的事情。請只宣告您打算讓代理使用的工具。 - 權限政策不適用於自訂工具。 權限政策管理內建和 MCP 工具集;worker 會執行模型發出的每個被包裝工具呼叫,因此請將任何核准步驟放在您自己的工具程式碼中。
監控與營運
這些呼叫從您的監控或營運工具執行,以您的 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 將其關閉。預設情況下,工作項目會移至 stopping:worker 在其下一次租約心跳時注意到,取消工作階段進行中的工具呼叫,並確認關閉,此時工作項目變為 stopped。在請求主體中傳入 force: true(使用 CLI 時,傳入 --force)可立即將工作項目標記為 stopped,而不等待 worker 的確認。
由於這些呼叫從您的營運工具而非 worker 主機執行,ANTHROPIC_WORK_ID 不會自動設定。請在執行以下範例之前將其設定為目標工作項目的 ID。若要找到工作項目的 ID,請透過 Environments Work 端點列出環境的工作項目。
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)後續步驟
Was this page helpful?