自托管沙箱

1. 创建自托管环境

   在控制台：Workspace > Environments > New > Self-hosted

   或通过 API：

   cURLcURLPythonTypescriptCsharpGoJavaPhpRuby

   curl -sS --fail-with-body https://api.anthropic.com/v1/environments \
     -H "x-api-key: $ANTHROPIC_API_KEY" \
     -H "anthropic-version: 2023-06-01" \
     -H "anthropic-beta: managed-agents-2026-04-01" \
     -H "content-type: application/json" \
     -d '{
       "name": "self-hosted",
       "config": {"type": "self_hosted"}
     }'
   

   ant beta:environments create \
     --name self-hosted \
     --config '{"type": "self_hosted"}'
   

   client = anthropic.Anthropic()
   
   environment = client.beta.environments.create(
       name="self-hosted", config={"type": "self_hosted"}
   )
   print(environment.id)
   

   const client = new Anthropic();
   
   const environment = await client.beta.environments.create({
     name: "self-hosted",
     config: { type: "self_hosted" }
   });
   console.log(environment.id);
   

   using Anthropic.Models.Beta.Environments;
   
   var client = new AnthropicClient();
   
   var environment = await client.Beta.Environments.Create(
       new EnvironmentCreateParams
       {
           Name = "self-hosted",
           Config = new BetaSelfHostedConfigParams(),
       }
   );
   Console.WriteLine(environment.ID);
   

   package main
   
   import (
   	"context"
   	"fmt"
   
   	"github.com/anthropics/anthropic-sdk-go"
   )
   
   func main() {
   	client := anthropic.NewClient()
   
   	environment, err := client.Beta.Environments.New(context.Background(), anthropic.BetaEnvironmentNewParams{
   		Name: "self-hosted",
   		Config: anthropic.BetaEnvironmentNewParamsConfigUnion{
   			OfSelfHosted: &anthropic.BetaSelfHostedConfigParams{},
   		},
   	})
   	if err != nil {
   		panic(err)
   	}
   	fmt.Println(environment.ID)
   }
   

   import com.anthropic.models.beta.environments.BetaSelfHostedConfigParams;
   import com.anthropic.models.beta.environments.EnvironmentCreateParams;
   
   var client = AnthropicOkHttpClient.fromEnv();
   
   var environment = client.beta().environments().create(
       EnvironmentCreateParams.builder()
           .name("self-hosted")
           .config(BetaSelfHostedConfigParams.builder().build())
           .build()
   );
   IO.println(environment.id());
   

   $client = new Anthropic\Client();
   
   $environment = $client->beta->environments->create(
       name: 'self-hosted',
       config: ['type' => 'self_hosted'],
   );
   echo $environment->id, PHP_EOL;
   

   client = Anthropic::Client.new
   
   environment = client.beta.environments.create(
     name: "self-hosted",
     config: {type: :self_hosted}
   )
   puts environment.id
   

2. 设置环境密钥

   在控制台中，打开环境并单击生成环境密钥。然后在工作者主机上导出环境 ID 和密钥：

   export ANTHROPIC_ENVIRONMENT_KEY="sk-ant-oat01-..."
   export ANTHROPIC_ENVIRONMENT_ID="env_..."
   

3. 运行工作者

   EnvironmentWorker 获取分配给环境的会话，下载技能，在工作目录中执行工具调用，并将结果发回。使用您在步骤 2 中生成的环境密钥进行身份验证。

   ℹ️

   Note

   技能可以包含智能体可能直接运行的可执行文件。SDK 会自动将下载的技能文件在沙箱中标记为可执行。如果您手动实现技能下载，您有责任设置可执行权限。

   PythonTypescriptCsharpGoJavaPhpRuby

   import asyncio
   import os
   from anthropic import AsyncAnthropic
   from anthropic.lib.environments import EnvironmentWorker
   
   
   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",
           ).run()
   
   
   asyncio.run(main())
   

   import Anthropic from "@anthropic-ai/sdk";
   import { EnvironmentWorker } from "@anthropic-ai/sdk/helpers/beta/environments";
   
   const environmentKey = process.env.ANTHROPIC_ENVIRONMENT_KEY!;
   const environmentId = process.env.ANTHROPIC_ENVIRONMENT_ID!;
   const client = new Anthropic({ authToken: environmentKey });
   const ctrl = new AbortController();
   process.once("SIGTERM", () => ctrl.abort());
   
   await new EnvironmentWorker({
     client,
     environmentId,
     environmentKey,
     workdir: "/workspace",
     signal: ctrl.signal
   }).run();
   

   // EnvironmentWorker 目前在 C# SDK 中不可用。请参阅上面的常驻（ant CLI）选项卡。
   

   package main
   
   import (
   	"context"
   	"log"
   	"os"
   	"os/signal"
   	"syscall"
   
   	"github.com/anthropics/anthropic-sdk-go"
   	"github.com/anthropics/anthropic-sdk-go/lib/environments"
   	"github.com/anthropics/anthropic-sdk-go/option"
   )
   
   func main() {
   	environmentKey := os.Getenv("ANTHROPIC_ENVIRONMENT_KEY")
   	environmentID := os.Getenv("ANTHROPIC_ENVIRONMENT_ID")
   
   	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
   	defer stop()
   
   	client := anthropic.NewClient(option.WithAuthToken(environmentKey))
   
   	worker := environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{
   		EnvironmentID:  environmentID,
   		EnvironmentKey: environmentKey,
   		Workdir:        "/workspace",
   	})
   	if err := worker.Run(ctx); err != nil {
   		log.Fatalf("worker: %v", err)
   	}
   }
   
   

   // EnvironmentWorker 目前在 Java SDK 中不可用。请参阅上面的常驻（ant CLI）选项卡。
   

   // EnvironmentWorker 目前在 PHP SDK 中不可用。请参阅上面的常驻（ant CLI）选项卡。
   

   # EnvironmentWorker 目前在 Ruby SDK 中不可用。请参阅上面的常驻（ant CLI）选项卡。
   

1. 订阅会话 webhook

   在控制台中，定义一个侦听 session.status_run_started 事件的 webhook 端点。有关更多详细信息，请参阅 webhook 文档。

2. 创建自托管环境

   在控制台：Workspace > Environments > New > Self-hosted

   或通过 API：

   cURLcURLPythonTypescriptCsharpGoJavaPhpRuby

   curl -sS --fail-with-body https://api.anthropic.com/v1/environments \
     -H "x-api-key: $ANTHROPIC_API_KEY" \
     -H "anthropic-version: 2023-06-01" \
     -H "anthropic-beta: managed-agents-2026-04-01" \
     -H "content-type: application/json" \
     -d '{
       "name": "self-hosted",
       "config": {"type": "self_hosted"}
     }'
   

   ant beta:environments create \
     --name self-hosted \
     --config '{"type": "self_hosted"}'
   

   client = anthropic.Anthropic()
   
   environment = client.beta.environments.create(
       name="self-hosted", config={"type": "self_hosted"}
   )
   print(environment.id)
   

   const client = new Anthropic();
   
   const environment = await client.beta.environments.create({
     name: "self-hosted",
     config: { type: "self_hosted" }
   });
   console.log(environment.id);
   

   using Anthropic.Models.Beta.Environments;
   
   var client = new AnthropicClient();
   
   var environment = await client.Beta.Environments.Create(
       new EnvironmentCreateParams
       {
           Name = "self-hosted",
           Config = new BetaSelfHostedConfigParams(),
       }
   );
   Console.WriteLine(environment.ID);
   

   package main
   
   import (
   	"context"
   	"fmt"
   
   	"github.com/anthropics/anthropic-sdk-go"
   )
   
   func main() {
   	client := anthropic.NewClient()
   
   	environment, err := client.Beta.Environments.New(context.Background(), anthropic.BetaEnvironmentNewParams{
   		Name: "self-hosted",
   		Config: anthropic.BetaEnvironmentNewParamsConfigUnion{
   			OfSelfHosted: &anthropic.BetaSelfHostedConfigParams{},
   		},
   	})
   	if err != nil {
   		panic(err)
   	}
   	fmt.Println(environment.ID)
   }
   

   import com.anthropic.models.beta.environments.BetaSelfHostedConfigParams;
   import com.anthropic.models.beta.environments.EnvironmentCreateParams;
   
   var client = AnthropicOkHttpClient.fromEnv();
   
   var environment = client.beta().environments().create(
       EnvironmentCreateParams.builder()
           .name("self-hosted")
           .config(BetaSelfHostedConfigParams.builder().build())
           .build()
   );
   IO.println(environment.id());
   

   $client = new Anthropic\Client();
   
   $environment = $client->beta->environments->create(
       name: 'self-hosted',
       config: ['type' => 'self_hosted'],
   );
   echo $environment->id, PHP_EOL;
   

   client = Anthropic::Client.new
   
   environment = client.beta.environments.create(
     name: "self-hosted",
     config: {type: :self_hosted}
   )
   puts environment.id
   

3. 设置凭据

   在控制台中，打开环境并单击生成环境密钥。在您的处理器主机上导出所有三个：

   export ANTHROPIC_ENVIRONMENT_KEY="sk-ant-oat01-..."
   export ANTHROPIC_ENVIRONMENT_ID="env_..."
   export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."
   

4. 实现 webhook 处理器

   EnvironmentWorker 获取会话，下载技能，在工作目录中执行工具调用，发回结果，然后退出。在 session.status_run_started 触发时调用它。

   ℹ️

   Note

   技能可以包含智能体可能直接运行的可执行文件。SDK 会自动将下载的技能文件在沙箱中标记为可执行。如果您手动实现技能下载，您有责任设置可执行权限。

   PythonTypescriptCsharpGoJavaPhpRuby

   import os
   import anthropic
   
   environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"]
   environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"]
   client = anthropic.AsyncAnthropic(
       auth_token=environment_key,
   )
   
   
   async def handle(raw: bytes, headers: dict[str, str]) -> dict:
       event = client.beta.webhooks.unwrap(raw.decode(), headers=headers)
       if event.data.type != "session.status_run_started":
           return {"status": "ignored"}
       async for work in client.beta.environments.work.poller(
           environment_id=environment_id,
           environment_key=environment_key,
           block_ms=None,
           reclaim_older_than_ms=2000,
           drain=True,
           auto_stop=False,
       ):
           await client.beta.environments.work.worker(workdir="/workspace").handle_item(
               work_id=work.id,
               environment_id=environment_id,
               session_id=work.data.id,
               environment_key=environment_key,
           )
       return {"status": "ok"}
   

   import Anthropic from "@anthropic-ai/sdk";
   
   const environmentKey = process.env.ANTHROPIC_ENVIRONMENT_KEY!;
   const environmentId = process.env.ANTHROPIC_ENVIRONMENT_ID!;
   const client = new Anthropic({
     authToken: environmentKey
   });
   
   export async function handle(req: Request): Promise<Response> {
     const body = await req.text();
     let event;
     try {
       event = client.beta.webhooks.unwrap(body, { headers: Object.fromEntries(req.headers) });
     } catch {
       return new Response("signature verification failed", { status: 401 });
     }
     if (event.data.type !== "session.status_run_started") {
       return Response.json({ status: "ignored" });
     }
   
     for await (const work of client.beta.environments.work.poller({
       environmentId,
       environmentKey,
       blockMs: null,
       reclaimOlderThanMs: 2000,
       drain: true,
       autoStop: false
     })) {
       await client.beta.environments.work.worker({ workdir: "/workspace" }).handleItem({
         workId: work.id,
         environmentId,
         sessionId: work.data.id,
         environmentKey
       });
     }
     return Response.json({ status: "ok" });
   }
   

   // EnvironmentWorker 目前在 C# SDK 中不可用。
   // 要直接处理工作项，请参阅介绍中链接的环境工作端点。
   

   package main
   
   import (
   	"context"
   	"encoding/json"
   	"io"
   	"log/slog"
   	"net/http"
   	"os"
   
   	"github.com/anthropics/anthropic-sdk-go"
   	"github.com/anthropics/anthropic-sdk-go/lib/environments"
   	"github.com/anthropics/anthropic-sdk-go/option"
   )
   
   var (
   	environmentKey = os.Getenv("ANTHROPIC_ENVIRONMENT_KEY")
   	environmentID  = os.Getenv("ANTHROPIC_ENVIRONMENT_ID")
   	client         = anthropic.NewClient(
   		option.WithAuthToken(environmentKey),
   		option.WithWebhookKey(os.Getenv("ANTHROPIC_WEBHOOK_SIGNING_KEY")),
   	)
   	worker = environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{
   		Workdir: "/workspace",
   	})
   )
   
   func handle(w http.ResponseWriter, r *http.Request) {
   	body, err := io.ReadAll(r.Body)
   	if err != nil {
   		http.Error(w, "bad request", http.StatusBadRequest)
   		return
   	}
   	event, err := client.Beta.Webhooks.Unwrap(body, r.Header)
   	if err != nil {
   		http.Error(w, "signature verification failed", http.StatusUnauthorized)
   		return
   	}
   	if event.Data.Type != "session.status_run_started" {
   		json.NewEncoder(w).Encode(map[string]string{"status": "ignored"})
   		return
   	}
   
   	// Go SDK 不提供 RunOne 便利方法：使用 WorkPoller 获取项，
   	// 然后使用 HandleItem 运行它。
   	// 从 r.Context() 分离：会话可能比 webhook 交付超时更长。
   	ctx := context.Background()
   	poller := environments.NewWorkPoller(ctx, client, environments.WorkPollerOptions{
   		EnvironmentID:  environmentID,
   		EnvironmentKey: environmentKey,
   		Drain:          true,
   	})
   	defer poller.Close()
   	if poller.Next() {
   		item := poller.Current()
   		if err := worker.HandleItem(ctx, environments.HandleItemOptions{
   			WorkID:         item.ID,
   			EnvironmentID:  item.EnvironmentID,
   			SessionID:      item.Data.ID,
   			EnvironmentKey: environmentKey,
   		}); err != nil {
   			slog.Error("handle work item", "work_id", item.ID, "err", err)
   			http.Error(w, "internal error", http.StatusInternalServerError)
   			return
   		}
   	}
   	if err := poller.Err(); err != nil {
   		slog.Error("poll work queue", "err", err)
   		http.Error(w, "internal error", http.StatusInternalServerError)
   		return
   	}
   	json.NewEncoder(w).Encode(map[string]string{"status": "ok"})
   }
   
   func main() {
   	http.HandleFunc("POST /webhook", handle)
   	if err := http.ListenAndServe(":8080", nil); err != nil {
   		slog.Error("http server", "err", err)
   		os.Exit(1)
   	}
   }
   
   

   // EnvironmentWorker 目前在 Java SDK 中不可用。
   // 要直接处理工作项，请参阅介绍中链接的环境工作端点。
   

   // EnvironmentWorker 目前在 PHP SDK 中不可用。
   // 要直接处理工作项，请参阅介绍中链接的环境工作端点。
   

   # EnvironmentWorker 目前在 Ruby SDK 中不可用。
   # 要直接处理工作项，请参阅介绍中链接的环境工作端点。
   

SDK 在不同级别提供三个辅助程序：

• EnvironmentWorker：开箱即用的工作者。端到端处理轮询、设置和执行。
  • .run()：无限期运行，随会话到达而获取。在 SIGTERM 时干净退出。
  • .handle_item()：获取一个待处理会话，处理它，然后退出。
• work.poller()：代表您轮询工作队列并向您提供每个获取的会话。当您想决定每个会话发生什么时使用此选项，例如启动容器而不是在进程内运行工具。
  • drain：队列为空时是否停止轮询，而不是等待新工作。
  • block_ms：在返回之前等待工作到达的时间（毫秒）。必须在 1 到 999 之间。设置为 None 进行非阻塞检查。
  • reclaim_older_than_ms：重新获取已租给停止响应的工作者的工作项。
  • auto_stop：迭代器退出后是否在工作项上发布停止信号。
• tool_runner()：为单个会话运行工具调用。当您已获取工作并且只需要执行层时使用。

EnvironmentWorker 覆盖大多数用例。当您想启动自己的每会话进程时，直接使用工作轮询器，例如为每个获取的会话启动容器：

# 轮询端点需要环境密钥（Bearer 身份验证），而不是您的
# API 密钥。使用下面的 SDK 选项之一。

# 轮询端点需要环境密钥（Bearer 身份验证），而不是您的
# API 密钥。使用下面的 SDK 选项之一。

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,  # 启动的容器拥有停止调用
        ):
            await launch_container(work)


asyncio.run(main())

import Anthropic from "@anthropic-ai/sdk";
import { WorkPoller } from "@anthropic-ai/sdk/helpers/beta/environments";
import type { BetaSelfHostedWork } from "@anthropic-ai/sdk/resources/beta/environments";

const environmentKey = process.env.ANTHROPIC_ENVIRONMENT_KEY!;
const environmentId = process.env.ANTHROPIC_ENVIRONMENT_ID!;
const client = new Anthropic({ authToken: environmentKey });

async function launchContainer(work: BetaSelfHostedWork): Promise<void> {
  // 替换为您自己的每会话沙箱启动器。将
  // ANTHROPIC_ENVIRONMENT_KEY 传入启动的容器，永远不要
  // 传入您的 API 密钥。
  console.log(`claimed session ${work.data.id}`);
}

const poller = new WorkPoller({
  client,
  environmentId,
  environmentKey,
  autoStop: false // 启动的容器拥有停止调用
});

for await (const work of poller) {
  await launchContainer(work);
}

// 工作轮询目前在 C# SDK 中不可用。

package main

import (
	"context"
	"fmt"
	"log"
	"os"

	"github.com/anthropics/anthropic-sdk-go"
	"github.com/anthropics/anthropic-sdk-go/lib/environments"
	"github.com/anthropics/anthropic-sdk-go/option"
)

func launchContainer(work *anthropic.BetaSelfHostedWork) {
	// 替换为您自己的每会话沙箱启动器。Go 轮询器
	// 在此函数返回时调用 work.Stop（它没有自动停止
	// 选择退出），因此在此处阻塞直到会话完成，而不是
	// 像 Python 和 TypeScript 选项卡那样分离。
	fmt.Printf("claimed session %s\n", work.Data.ID)
}

func main() {
	environmentID := os.Getenv("ANTHROPIC_ENVIRONMENT_ID")
	environmentKey := os.Getenv("ANTHROPIC_ENVIRONMENT_KEY")

	client := anthropic.NewClient(option.WithAuthToken(environmentKey))

	ctx := context.Background()

	poller := environments.NewWorkPoller(ctx, client, environments.WorkPollerOptions{
		EnvironmentID:  environmentID,
		EnvironmentKey: environmentKey,
	})
	defer poller.Close()

	for work, err := range poller.All() {
		if err != nil {
			log.Fatal(err)
		}
		launchContainer(work)
	}
}

// 工作轮询目前在 Java SDK 中不可用。

// 工作轮询目前在 PHP SDK 中不可用。

# 工作轮询目前在 Ruby SDK 中不可用。

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()： 在运行工具之前自己设置 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:
    # 技能下载到 /workspace/skills/<name>/
    tools = beta_agent_toolset_20260401(env)

import {
  setupSkills,
  betaAgentToolset20260401
} from "@anthropic-ai/sdk/tools/agent-toolset/node";

const ctx = { workdir: "/workspace", client, sessionId: work.data.id };
await setupSkills(ctx);
const tools = betaAgentToolset20260401(ctx);

// AgentToolContext 目前在 C# SDK 中不可用。

env := &agenttoolset.AgentToolContext{Workdir: "/workspace"}
if err := env.SetupSkills(ctx, client, work.Data.ID); err != nil {
	panic(err)
}
// 技能下载到 /workspace/skills/<name>/
tools := agenttoolset.BetaAgentToolset20260401(env)

// AgentToolContext 目前在 Java SDK 中不可用。

// AgentToolContext 目前在 PHP SDK 中不可用。

# AgentToolContext 目前在 Ruby SDK 中不可用。