셀프 호스팅 샌드박스
셀프 호스팅 샌드박스 (Self-hosted sandboxes)
기본적으로 Managed Agents는 도구와 코드를 Anthropic 관리형 클라우드 샌드박스 안에서 실행해요. 셀프 호스팅 샌드박스는 오케스트레이션을 Anthropic 쪽에 두되 도구 실행을 여러분이 제어하는 인프라로 옮겨서, 에이전트의 코드, 파일시스템, 네트워크 이그레스가 여러분의 환경을 결코 떠나지 않게 해요.
출처: 문서
본문
기본적으로 Managed Agents는 도구와 코드를 Anthropic 관리형 클라우드 샌드박스 안에서 실행해요. 셀프 호스팅 샌드박스는 오케스트레이션을 Anthropic 쪽에 두되 도구 실행을 여러분이 제어하는 인프라로 옮겨서, 에이전트의 코드, 파일시스템, 네트워크 이그레스가 여러분의 환경을 결코 떠나지 않게 해요.
도구 실행은 여러분의 호스트에 남아요: 에이전트가 읽고 쓰는 파일시스템, 그것이 생성하는 프로세스, 도달할 수 있는 네트워크가 모두 여러분의 통제 아래 있어요. 도구 입력과 출력은 여전히 Anthropic 제어 평면(Claude가 실행되는 곳)으로 흘러 모델이 결과를 보고 다음에 무엇을 할지 결정할 수 있어요. 에이전트의 스킬과 세션에 붙은 메모리 스토어의 내용은 Anthropic이 저장하고 세션 동안 여러분의 샌드박스로 복사돼요. 에이전트가 메모리 파일에 한 변경은 스토어로 다시 동기화돼요. 전체 데이터 흐름 경계는 보안 모델을 참고하세요.
클라우드 환경과의 차이 (How it differs from cloud environments)
| 클라우드 환경 (Cloud environment) | 셀프 호스팅 샌드박스 (Self-hosted sandbox) | |
|---|---|---|
| 도구가 실행되는 곳 | Anthropic 관리형 샌드박스 | 여러분의 인프라 |
| 네트워크 도달 범위 | Anthropic의 이그레스 컨트롤 | 여러분의 네트워크 정책 |
| 파일·GitHub 리포지토리 마운트 | Anthropic이 관리 | 여러분이 관리 |
| 메모리 스토어 | Anthropic이 /mnt/memory/에 마운트 |
/mnt/memory/로 다운로드되고 SDK 워커가 동기화 |
| 수명주기 | Anthropic이 관리 | 여러분이 관리 |
셀프 호스팅은 에이전트가 네트워크 경계를 벗어날 수 없는 데이터를 다뤄야 하거나, 공개적으로 라우팅되지 않는 내부 서비스에 닿아야 하거나, 여러분 조직의 컴플라이언스·감사 컨트롤 아래 실행되어야 할 때 잘 맞아요.
Zero Data Retention과 HIPAA BAA 자격에 대해서는 API 및 데이터 보존을 참고하세요.
MCP 터널과 결합할 때 (When to combine with MCP tunnels)
셀프 호스팅은 에이전트의 코드가 실행되는 곳을 제어해요. MCP 터널은 Anthropic이 네트워크 안의 MCP 서버에 어떻게 닿는지를 제어해요. 둘은 독립적이에요: Anthropic 클라우드 샌드박스에서 실행되는 세션도 터널을 통해 개인 MCP 서버에 닿을 수 있고, 셀프 호스팅 세션도 터널 또는 공개 MCP 서버를 사용할 수 있어요. 실행과 도구 접근이 모두 여러분 경계 안에 머물게 하려면 둘 다 사용하세요. 터널을 실행하지 않고 네트워크 안의 MCP 서버에서 도구를 에이전트에 주려면, 서버를 여러분의 워커가 서빙하는 커스텀 도구로 감싸 수도 있어요.
환경 워커 (Environment worker)
환경 워커(environment worker)는 여러분의 인프라에서 실행하는 프로세스예요. Anthropic에서 도구 실행 요청을 받아 로컬에서 실행해요. self_hosted 환경은 작업 큐로 동작해요: 세션이 그 환경에 배정되면 Anthropic이 세션을 작업 항목으로 큐에 넣어요. 여러분의 워커가 그 큐에서 작업 항목을 가져와 각각에 대한 실행 컨텍스트를 생성하고, 에이전트의 스킬(에이전트에게 도메인 특화 전문성을 주는 재사용 가능한 파일시스템 기반 리소스)을 다운로드하고, 도구 호출을 실행하고, 결과를 다시 게시해요.
작업 항목은 환경의 큐를 폴링해 가져와요: 지속적으로 폴링하는 항상 켜진(always-on) 워커 또는 session.status_run_started에서 깨어나 폴링을 시작하는 웹훅 트리거 핸들러에요.
CLI와 SDK 모두 사전 구축된 워커를 제공해요. ant CLI는 항상 켜진 패턴만 지원하고, SDK는 항상 켜진 것과 웹훅 트리거 둘 다 지원해요. 둘 다 구성 가능해요: CLI 플래그는 reference의 셀프 호스팅 워커를, SDK 옵션은 이 페이지의 SDK 헬퍼를 참고하세요. 더 많은 제어가 필요하면 Environments Work 엔드포인트를 직접 호출해 자체 워커를 구현하세요.
샌드박스 파일시스템 (Sandbox filesystem)
/workspace: 도구 실행과 스킬 다운로드의 시스템 기본 작업 디렉터리예요. CLI의--workdir플래그는 현재 디렉터리를 기본으로 해요. 시스템 기본값과 맞추려면--workdir /workspace를 전달하세요. 스킬은<workdir>/skills/<name>/에 다운로드돼요. 다른 작업 디렉터리를 사용한다면 Claude가 스킬 파일을 찾도록 에이전트의 시스템 프롬프트를 업데이트하세요.- 산출물 (Outputs): 셀프 호스팅 환경에서는 세션의 시스템 프롬프트가 Anthropic 관리형 샌드박스에서 사용되는
/mnt/session/outputs지시를 생략하므로, 최종 산출물은 에이전트가 여러분의 샌드박스 파일시스템 어디에 쓰든 그 자리에 놓여요. 보통 작업 디렉터리 아래예요. /mnt/memory/: 세션에 붙은 메모리 스토어가 SDK 워커에 의해 여기에 구체화되며, 스토어당 한 디렉터리가 스토어의mount_path에 생겨요(예:/mnt/memory/user-preferences/). 워커는 세션을 가져올 때 이 디렉터리들을 만들고 세션이 끝나면 제거해요. 메모리 스토어 사용을 참고하세요.
시작하기 전에 (Before you begin)
다음이 필요해요:
- 기존 에이전트. 없다면 먼저 퀵스타트를 완료하고 에이전트 ID를 기록하세요.
- Linux 호스트에 정확히 그 경로의
/bin/bash. 워커의 bash 도구가PATH를 조회하지 않고 직접 호출해요. TypeScript SDK는 추가로PATH에unzip과tar, Node.js 22 이상이 필요해요. Python과 Go SDK는 아카이브 추출에 표준 라이브러리를 사용해 추가 바이너리 요구 사항이 없어요. - 워커 호스트의
antCLI 또는 Anthropic SDK(Python, TypeScript, Go). - 자격 증명: 환경 키(다음 단계에서 Console로 생성)가 워커를 그 큐에 인증하고, 여러분의 Claude API 키는 워커 호스트 밖에서 세션을 만들고 큐 통계를 읽어요. 키 생성은 Console 전용이에요. 가져온 작업 항목은 워커가 메모리 스토어를 마운트하는 데 쓰는 세션별
secret도 싣는데, 여러분이 생성하진 않지만 샌드박스-세션당 패턴에서는 그 secret을 직접 샌드박스로 전달해요(세션당 샌드박스 하나 실행 참고). - 메모리 스토어용 준비된 호스트. 이 환경의 세션이 메모리 스토어를 붙이면 워커 시작 전에
/mnt/memory를 준비하세요. 호스트 준비를 참고하세요.
Claude Platform on AWS의 셀프 호스팅 환경에서는 세션에 메모리 스토어를 붙일 수 없어요.
또는 API로:
<CodeGroup defaultLanguage="CLI">
```bash cURL
curl -sS --fail-with-body https://api.anthropic.com/v1/environments \
-H "x-api-key: $ANTHR...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"}
}'
```
<MultiFileExample language="cli" label="CLI">
```bash CLI
ant apply environment.yaml
```
<File filename="environment.yaml">
```yaml
# yaml-language-server: $schema=https://platform.claude.com/schemas/ant/beta/environment.json
name: self-hosted
config:
type: self_hosted
```
</File>
</MultiFileExample>
```python Python
client = anthropic.Anthropic()
environment = client.beta.environments.create(
name="self-hosted", config={"type": "self_hosted"}
)
print(environment.id)
```
```typescript TypeScript
const client = new Anthropic();
const environment = await client.beta.environments.create({
name: "self-hosted",
config: { type: "self_hosted" }
});
console.log(environment.id);
```
```csharp C#
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);
```
```go Go
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)
```
```java Java
import com.anthropic.models.beta.environments.BetaSelfHostedConfigParams;
import com.anthropic.models.beta.environments.EnvironmentCreateParams;
void main() {
var client = AnthropicOkHttpClient.fromEnv();
var environment = client.beta().environments().create(
EnvironmentCreateParams.builder()
.name("self-hosted")
.config(BetaSelfHostedConfigParams.builder().build())
.build()
);
IO.println(environment.id());
}
```
```php PHP
$client = new Anthropic\Client();
$environment = $client->beta->environments->create(
name: 'self-hosted',
config: ['type' => 'self_hosted'],
);
echo $environment->id, PHP_EOL;
```
```ruby Ruby
client = Anthropic::Client.new
environment = client.beta.environments.create(
name: "self-hosted",
config: {type: :self_hosted}
)
puts environment.id
```
</CodeGroup>
```bash
export ANTHROPIC_ENVIRONMENT_KEY="«redacted:sk-…»..."
export ANTHROPIC_ENVIRONMENT_ID="env_..."
```
워커 실행하기 (Run a worker)
항상 켜진(always-on) 방식을 가장 단순한 설정으로 선택하세요: 장기 실행 프로세스가 큐를 지속적으로 폴링하고 단지 아웃바운드 HTTPS만 필요해요. 유휴 폴러를 실행하지 않으려면 웹훅 트리거를 선택하세요. 이는 Anthropic이 닿을 수 있는 웹훅 엔드포인트가 필요해요(엔드포인트 설정과 서명 검증은 웹훅 참고).
<Tabs>
<Tab title="curl (Linux/WSL)">
Linux 환경에서는 릴리스 바이너리를 직접 다운로드하세요.
```bash
VERSION=1.35.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 releases 페이지](https://github.com/anthropics/anthropic-cli/releases)에서 찾을 수 있어요.
</Tab>
<Tab title="Homebrew (macOS)">
```bash
brew install anthropics/tap/ant
```
</Tab>
</Tabs>
</Step>
<Step title="워커 실행 (Run the worker)">
**인프로세스 (In-process)**
`ant beta:worker poll`은 환경에 배정된 작업 항목을 가져오고, 스킬을 다운로드하고, 작업 디렉터리에서 도구 호출을 실행하고, 결과를 다시 게시해요. `ANTHROPIC_ENVIRONMENT_KEY`와 `ANTHROPIC_ENVIRONMENT_ID`를 환경에서 읽어요.
```bash
ant beta:worker poll --workdir "/workspace"
```
워커는 SIGTERM이나 SIGINT에서 깨끗하게 종료해요: 진행 중인 도구 호출을 취소하고, 그 오류 결과를 게시하고, 멈추기 전에 작업 항목을 해제해요.
**세션당 샌드박스 (Sandbox per session)**
더 강한 격리(새 파일시스템, 리소스 제한, 세션별 네트워크 컨트롤)가 필요하면 각 세션을 자체 샌드박스에서 실행하세요. `ant`가 설치되고 `ant beta:worker run`이 엔트리포인트인 이미지를 구축하세요. 기본 이미지는 `/bin/bash`를 제공해야 하고, `curl`은 빌드 시에만 사용돼요. 샌드박스가 시작되면 환경 변수에서 세션 세부 정보를 읽고, 그 세션을 처리하고, 종료해요:
```text
FROM your-base-image
ARG ANT_VERSION=1.35.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"]
```
그런 다음 세션 세부 정보를 새 샌드박스로 전달하는 스폰 스크립트를 작성하세요. 폴러는 `ANTHROPIC_SESSION_ID`, `ANTHROPIC_WORK_ID`, `ANTHROPIC_ENVIRONMENT_ID`, `ANTHROPIC_ENVIRONMENT_KEY`를 스크립트의 환경에 주입하고, 가져온 작업 항목을 JSON으로 표준 입력에 써요(Anthropic이 발급했을 때 작업 항목의 세션별 `secret` 포함). `ANTHROPIC_BASE_URL`은 선택 사항이며 폴러 호스트에 설정된 경우에만 통과돼요. 기본 API 엔드포인트를 재정의해요. 예시에서 `/host/outputs`는 여러분이 선택하는 호스트 디렉터리이며, 샌드박스의 작업 디렉터리(`/workspace`)에 바인드 마운트되어 샌드박스 종료 후 세션 산출물을 검색할 수 있어요. 셀프 호스팅 환경에서는 에이전트가 `/mnt/session/outputs`가 아니라 작업 디렉터리 아래에 산출물을 쓰므로([샌드박스 파일시스템](https://platform.claude.com/docs/en/managed-agents/self-hosted-sandboxes#sandbox-filesystem) 참고), 작업 디렉터리를 마운트하는 것이 그것을 잡아내는 방법이에요. 마운트는 다운로드된 `skills/` 트리와 에이전트가 만든 중간 파일도 잡아요.
```bash
#!/bin/bash
# spawn.sh: called once per claimed work item
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 run` 엔트리포인트는 [메모리 스토어](https://platform.claude.com/docs/en/managed-agents/self-hosted-sandboxes#use-memory-stores)를 마운트하지 않아요. 이 환경의 세션이 메모리 스토어를 붙인다면 폴러를 유지하되, [세션당 샌드박스 하나 실행](https://platform.claude.com/docs/en/managed-agents/self-hosted-sandboxes#run-one-sandbox-per-session)에 표시된 대로 세션당 이미지를 SDK 워커 중심으로 구축하고 스폰 스크립트를 확장해 작업 항목의 `secret`을 샌드박스로 전달하세요.
스크립트를 가리키는 폴러를 시작하세요:
```bash
ant beta:worker poll --on-work ./spawn.sh
```
</Step>
</Steps>
<CodeGroup exclude="shell">
```python Python
import asyncio
import contextlib
import os
import signal
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:
worker = EnvironmentWorker(
client,
environment_id=environment_id,
environment_key=environment_key,
workdir="/workspace",
)
task = asyncio.create_task(worker.run())
# Cancelling the task, rather than killing the process, lets the worker stop its
# in-flight work item and upload changed memory files before it exits.
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())
```
```typescript TypeScript
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 controller = new AbortController();
// Aborting on either signal lets the worker upload changed memory files and remove its
// store directories before the process exits.
process.once("SIGINT", () => controller.abort());
process.once("SIGTERM", () => controller.abort());
await new EnvironmentWorker({
client,
environmentId,
environmentKey,
workdir: "/workspace",
signal: controller.signal
}).run();
```
```csharp C#
// EnvironmentWorker is not currently available in the C# SDK. See the Always-on (ant CLI) tab.
```
```go Go
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)
}
}
```
```java Java
// EnvironmentWorker is not currently available in the Java SDK. See the Always-on (ant CLI) tab.
```
```php PHP
// EnvironmentWorker is not currently available in the PHP SDK. See the Always-on (ant CLI) tab.
```
```ruby Ruby
# EnvironmentWorker is not currently available in the Ruby SDK. See the Always-on (ant CLI) tab.
```
</CodeGroup>
</Step>
</Steps>
<Step title="웹훅 서명 키 내보내기 (Export the webhook signing key)">
[시작하기 전에](https://platform.claude.com/docs/en/managed-agents/self-hosted-sandboxes#before-you-begin)의 환경 ID와 키 외에 핸들러 호스트에 웹훅 서명 키를 내보내 핸들러가 들어오는 페이로드를 검증할 수 있게 하세요. Python 핸들러에서 서명 검증에는 webhooks extra가 필요해요: `pip install "anthropic[webhooks]"`.
```bash
export ANTHROPIC_WEBHOOK_SIGNING_KEY="whsec_..."
```
</Step>
<Step title="웹훅 핸들러 구현 (Implement the webhook handler)">
`EnvironmentWorker`는 작업 항목을 가져오고, 스킬을 다운로드하고, 작업 디렉터리에서 도구 호출을 실행하고, 결과를 다시 게시하고, 종료해요. `session.status_run_started`가 발화할 때 호출하세요.
가져온 작업 항목을 이 핸들러처럼 `handle_item()`에 직접 넘길 때는 그것에 붙은 [메모리 스토어](https://platform.claude.com/docs/en/managed-agents/self-hosted-sandboxes#use-memory-stores)를 세션이 마운트할 수 있도록 작업 항목의 `secret`을 `work_secret`(TypeScript는 `workSecret`, Go는 `WorkSecret`)으로 함께 전달하세요. 이 핸들러 같은 것은 모든 가져온 항목을 한 프로세스의 한 호스트에서 실행하므로, 같은 메모리 스토어를 붙이는 두 세션은 그것을 동시에 통과할 수 없어요([호스트 준비](https://platform.claude.com/docs/en/managed-agents/self-hosted-sandboxes#prepare-the-host) 참고). 세션이 스토어를 공유한다면 대신 [세션당 샌드박스 하나](https://platform.claude.com/docs/en/managed-agents/self-hosted-sandboxes#run-one-sandbox-per-session)를 실행하세요.
<CodeGroup exclude="shell">
```python Python
import asyncio
import os
import anthropic
import standardwebhooks # installed by the anthropic[webhooks] extra
environment_key = os.environ["ANTHROPIC_ENVIRONMENT_KEY"]
environment_id = os.environ["ANTHROPIC_ENVIRONMENT_ID"]
client = anthropic.AsyncAnthropic(
auth_token=environment_key,
)
# Cancelled by shutdown() so an in-flight work item can upload changed memory files and
# remove its store directories before the process exits.
inflight: set[asyncio.Task[None]] = set()
# Await this from the host's shutdown hook, such as an ASGI lifespan shutdown (the code after
# `yield` in a FastAPI lifespan), which uvicorn runs on SIGTERM. uvicorn lets open requests
# finish before that hook runs, so set --timeout-graceful-shutdown to bound the wait.
async def shutdown() -> None:
for task in inflight:
task.cancel()
await asyncio.gather(*inflight, return_exceptions=True)
async def handle(raw: bytes, headers: dict[str, str]) -> tuple[dict[str, str], int]:
try:
event = client.beta.webhooks.unwrap(raw.decode(), headers=headers)
except standardwebhooks.WebhookVerificationError:
return {"error": "signature verification failed"}, 401
if event.data.type != "session.status_run_started":
return {"status": "ignored"}, 200
task = asyncio.create_task(run_queued_work())
inflight.add(task)
task.add_done_callback(inflight.discard)
try:
# Shielded: a dropped or timed-out delivery must not cancel the item; shutdown() does.
await asyncio.shield(task)
except asyncio.CancelledError:
return {"status": "shutting down"}, 503
return {"status": "ok"}, 200
async def run_queued_work() -> None:
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,
# The per-session secret is what lets the worker mount the session's memory stores.
work_secret=work.secret,
)
```
```typescript TypeScript
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
});
// Call shutdown.abort() from the host's SIGTERM/SIGINT handler, alongside closing the server,
// then wait for in-flight handle() calls before exiting: the abort lets a running work item
// upload changed memory files and remove its store directories first.
export const shutdown = new AbortController();
export async function handle(req: Request): Promise<Response> {
// Never acknowledge a delivery whose work will not run here; a 503 makes the sender retry.
if (shutdown.signal.aborted) {
return Response.json({ status: "shutting down" }, { status: 503 });
}
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,
signal: shutdown.signal
})) {
await client.beta.environments.work.worker({ workdir: "/workspace" }).handleItem({
workId: work.id,
environmentId,
sessionId: work.data.id,
environmentKey,
// The per-session secret is what lets the worker mount the session's memory stores.
workSecret: work.secret ?? undefined,
signal: shutdown.signal
});
}
// The poller and handleItem return quietly on abort, so a drain cut short lands here.
if (shutdown.signal.aborted) {
return Response.json({ status: "shutting down" }, { status: 503 });
}
return Response.json({ status: "ok" });
}
```
```csharp C#
// EnvironmentWorker is not currently available in the C# SDK.
// To handle work items directly, see the Environments Work endpoints.
```
```go Go
package main
import (
"context"
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"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"
"github.com/anthropics/anthropic-sdk-go/packages/param"
)
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",
})
// Cancelled on SIGINT or SIGTERM (set in main) so an in-flight work item can
// upload changed memory files and remove its store directories before exit.
shutdown context.Context
)
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
}
// The Go SDK does not provide a RunOne convenience: drain pending items
// with WorkPoller and run each one with HandleItem.
// Detach from r.Context(): the session can outlive the webhook delivery timeout.
// The process-wide shutdown context still ends the item cleanly on SIGTERM.
ctx := shutdown
poller := environments.NewWorkPoller(ctx, client, environments.WorkPollerOptions{
EnvironmentID: environmentID,
EnvironmentKey: environmentKey,
BlockMs: param.Null[int64](),
ReclaimOlderThanMs: param.NewOpt[int64](2000),
Drain: true,
AutoStop: param.NewOpt(false),
})
defer poller.Close()
for poller.Next() {
item := poller.Current()
if err := worker.HandleItem(ctx, environments.HandleItemOptions{
WorkID: item.ID,
EnvironmentID: item.EnvironmentID,
SessionID: item.Data.ID,
EnvironmentKey: environmentKey,
// The per-session secret is what lets the worker mount the session's memory stores.
WorkSecret: item.Secret,
}); 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() {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
shutdown = ctx
server := &http.Server{Addr: ":8080"}
http.HandleFunc("POST /webhook", handle)
go func() {
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
slog.Error("http server", "err", err)
os.Exit(1)
}
}()
// On a signal, stop accepting deliveries and return only after in-flight
// handlers, and therefore their work items' memory teardown, have finished.
<-ctx.Done()
if err := server.Shutdown(context.Background()); err != nil {
slog.Error("http shutdown", "err", err)
}
}
```
```java Java
// EnvironmentWorker is not currently available in the Java SDK.
// To handle work items directly, see the Environments Work endpoints.
```
```php PHP
// EnvironmentWorker is not currently available in the PHP SDK.
// To handle work items directly, see the Environments Work endpoints.
```
```ruby Ruby
# EnvironmentWorker is not currently available in the Ruby SDK.
# To handle work items directly, see the Environments Work endpoints.
```
</CodeGroup>
</Step>
</Steps>
SDK 헬퍼 (SDK helpers)
SDK는 서로 다른 제어 수준의 세 가지 헬퍼를 제공해요. EnvironmentWorker가 대부분의 사용 사례를 다루고, 자체 세션별 프로세스를 시작하거나 이미 가져온 세션에 대해 도구를 실행해야 할 때 저수준 헬퍼로 내려가세요.
-
EnvironmentWorker: 기본 제공 워커. 폴링, 설정, 실행을 처음부터 끝까지 처리해요..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]())을 전달하고, 파라미터를 생략하면 기본 999ms 롱폴을 사용해요.reclaim_older_than_ms: 이 밀리초 안에 가져왔지만 승인되지 않은 작업 항목을 다시 가져와요.auto_stop(TypeScript는autoStop, Go는AutoStop): 루프 본문이 작업 항목을 마치면 그 각각에 대해 정지 신호를 게시할지. 작업 항목을 실행하는 무엇이든 스스로 정지를 게시할 때 끄세요:handle_item()이 정지를 게시하므로, 이 페이지의 웹훅 핸들러처럼 가져온 항목을handle_item()에 넘길 때 false로 설정하고, 정지 호출을 소유하는 시작한 샌드박스도 그래요.
-
client.beta.sessions.events.tool_runner(): 세션 ID와 도구 목록이 주어지면 단일 세션에 대해 도구 호출을 실행해요. 이미 작업을 가져왔고 실행 계층만 필요할 때 사용하세요.
자체 세션별 프로세스를 시작하려 할 때(가져온 세션마다 샌드박스를 띄우는 것 등) 작업 폴러를 직접 사용하세요:
# The work poller is an SDK helper (Python, TypeScript, Go), not a raw
# endpoint. From the shell, use `ant beta:worker poll --on-work` instead;
# see the Always-on (ant CLI) tab.
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}")
# Replace `docker run` with your own sandbox launcher. Forward the environment
# key (never your API key) and the work item's per-session secret: the worker
# inside needs the secret to mount the session's memory stores.
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())
import { spawn } from "node:child_process";
import { once } from "node:events";
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 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
];
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> {
console.log(`claimed session ${work.data.id}`);
// Replace `docker run` with your own sandbox launcher. Forward the environment
// key (never your API key) and the work item's per-session secret: the worker
// inside needs the secret to mount the session's memory stores.
const env = {
...process.env,
ANTHROPIC_WORK_ID: work.id,
ANTHROPIC_SESSION_ID: work.data.id,
ANTHROPIC_WORK_SECRET: work.secret ?? ""
};
const forward = SANDBOX_ENV.flatMap((name) => ["-e", name]);
const launcher = spawn(
"docker",
["run", "--rm", "--detach", ...forward, "your-sdk-worker-image"],
{ env, stdio: "inherit" }
);
await once(launcher, "close");
}
const poller = new WorkPoller({
client,
environmentId,
environmentKey,
autoStop: false // the launched sandbox owns the stop call
});
for await (const work of poller) {
await launchContainer(work);
}
// A work-polling helper is not currently available in the C# SDK.
// To claim work directly, see the Environments Work endpoints.
package main
import (
"context"
"fmt"
"log"
"os"
"os/exec"
"github.com/anthropics/anthropic-sdk-go"
"github.com/anthropics/anthropic-sdk-go/lib/environments"
"github.com/anthropics/anthropic-sdk-go/option"
"github.com/anthropics/anthropic-sdk-go/packages/param"
)
var sandboxEnv = []string{
"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
}
func launchContainer(ctx context.Context, work *anthropic.BetaSelfHostedWork) error {
fmt.Printf("claimed session %s\n", work.Data.ID)
// Replace `docker run` with your own sandbox launcher. Forward the environment
// key (never your API key) and the work item's per-session secret: the worker
// inside needs the secret to mount the session's memory stores.
args := []string{"run", "--rm", "--detach"}
for _, name := range sandboxEnv {
args = append(args, "-e", name)
}
launcher := exec.CommandContext(ctx, "docker", append(args, "your-sdk-worker-image")...)
launcher.Env = append(os.Environ(),
"ANTHROPIC_WORK_ID="+work.ID,
"ANTHROPIC_SESSION_ID="+work.Data.ID,
"ANTHROPIC_WORK_SECRET="+work.Secret,
)
launcher.Stdout, launcher.Stderr = os.Stdout, os.Stderr
return launcher.Run()
}
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,
AutoStop: param.NewOpt(false), // the launched sandbox owns the stop call
})
defer poller.Close()
for work, err := range poller.All() {
if err != nil {
log.Fatal(err)
}
if err := launchContainer(ctx, work); err != nil {
log.Fatal(err)
}
}
}
// A work-polling helper is not currently available in the Java SDK.
// To claim work directly, see the Environments Work endpoints.
// A work-polling helper is not currently available in the PHP SDK.
// To claim work directly, see the Environments Work endpoints.
# A work-polling helper is not currently available in the Ruby SDK.
# To claim work directly, see the Environments Work endpoints.
샌드박스를 시작하는 무엇이든 작업 항목의 secret을 세션, 작업, 환경 식별자와 함께 그것으로 전달해야 해요(예: ANTHROPIC_WORK_SECRET으로). 그러면 안의 워커가 세션의 메모리 스토어를 마운트할 수 있어요. 세션당 샌드박스 하나 실행을 참고하세요.
**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 팩토리를 전달해 도구 목록을 맞춤 설정하세요:
new EnvironmentWorker({
client,
environmentId,
environmentKey,
tools: (ctx) => [betaBashTool(ctx), myCustomTool]
});
// EnvironmentWorker is not currently available in the C# SDK.
// To answer custom tool calls directly, see the session event stream.
worker := environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{
EnvironmentID: environmentID,
EnvironmentKey: environmentKey,
ToolsFunc: func(env *agenttoolset.AgentToolContext) []anthropic.BetaTool {
return []anthropic.BetaTool{agenttoolset.BetaBashTool(env), myCustomTool}
},
})
// EnvironmentWorker is not currently available in the Java SDK.
// To answer custom tool calls directly, see the session event stream.
// EnvironmentWorker is not currently available in the PHP SDK.
// To answer custom tool calls directly, see the session event stream.
# EnvironmentWorker is not currently available in the Ruby SDK.
# To answer custom tool calls directly, see the session event stream.
work.poller()와 tool_runner() 사용 시: tools로 도구 목록을 client.beta.sessions.events.tool_runner()에 전달하세요. 그 목록을 만들려면 AgentToolContext를 직접 설정하고 beta_agent_toolset_20260401(env)를 호출하세요:
async with AgentToolContext(
workdir="/workspace", client=client, session_id=work.data.id
) as env:
# skills downloaded to /workspace/skills/
```typescript TypeScript
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 is not currently available in the C# SDK.
env := &agenttoolset.AgentToolContext{Workdir: "/workspace"}
if err := env.SetupSkills(ctx, client, work.Data.ID); err != nil {
panic(err)
}
// skills downloaded to /workspace/skills/<name>/
tools := agenttoolset.BetaAgentToolset20260401(env)
// AgentToolContext is not currently available in the Java SDK.
// AgentToolContext is not currently available in the PHP SDK.
# AgentToolContext is not currently available in the Ruby SDK.
워커가 연결됐는지 확인 (Verify the worker is connected)
별도 셸에서 ANTHROPIC_API_KEY를 Claude API 키(환경 키 아님)로 설정하고 workers_polling이 최소 1인지 확인하세요:
ant beta:environments:work stats --environment-id "$ANTHROPIC_ENVIRONMENT_ID"
workers_polling이 0에 머물면 워커가 큐에 닿지 못하는 것이에요. ANTHROPIC_ENVIRONMENT_KEY와 ANTHROPIC_ENVIRONMENT_ID가 워커 호스트에 설정됐는지 확인하세요. 전체 stats 응답과 다른 언어 예시는 큐 깊이 읽기를 참고하세요.
세션 시작하기 (Start a session)
워커가 실행되면 그 환경을 대상으로 하는 세션을 만드세요. AGENT_ID를 시작하기 전에에서 기록한 에이전트 ID로 설정하세요. 세션은 환경의 작업 큐에 들어가 워커가 가져갈 때까지 거기서 기다려요. 연결된 워커가 없으면 세션은 실패하는 대신 큐에 머물러요.
Anthropic은 셀프 호스팅 샌드박스에 파일이나 GitHub 리포지토리를 마운트하지 않아요. 세션별 파일을 사용할 수 있게 하려면 세션 metadata 필드에 파일 참조(예: S3 경로나 커밋 SHA)를 전달하세요. 가져온 작업 항목은 세션의 메타데이터를 싣지 않지만 세션 ID는 싣어요. 여러분의 스폰 스크립트나 --on-work 핸들러가 세션(GET /v1/sessions/{session_id})을 가져와 metadata 필드를 읽고, 도구 실행이 시작되기 전에 파일을 작업 디렉터리로 스테이징해요.
ant beta:sessions create \
--agent "$AGENT_ID" \
--environment-id "$ANTHROPIC_ENVIRONMENT_ID" \
--metadata '{"input_file": "s3://my-bucket/data.csv"}'
session = client.beta.sessions.create(
agent=agent.id,
environment_id=environment.id,
metadata={"input_file": "s3://my-bucket/data.csv"},
)
const session = await client.beta.sessions.create({
agent: agent.id,
environment_id: environment.id,
metadata: { input_file: "s3://my-bucket/data.csv" }
});
var session = await client.Beta.Sessions.Create(new()
{
Agent = agent.ID,
EnvironmentID = environment.ID,
Metadata = new Dictionary<string, string> { ["input_file"] = "s3://my-bucket/data.csv" },
});
session, err := client.Beta.Sessions.New(ctx, anthropic.BetaSessionNewParams{
Agent: anthropic.BetaSessionNewParamsAgentUnion{OfString: anthropic.String(agent.ID)},
EnvironmentID: environment.ID,
Metadata: map[string]string{
"input_file": "s3://my-bucket/data.csv",
},
})
if err != nil {
panic(err)
}
var session = client.beta().sessions().create(SessionCreateParams.builder()
.agent(agent.id())
.environmentId(environment.id())
.metadata(SessionCreateParams.Metadata.builder()
.putAdditionalProperty("input_file", JsonValue.from("s3://my-bucket/data.csv"))
.build())
.build());
$session = $client->beta->sessions->create(
agent: $agent->id,
environmentID: $environment->id,
metadata: ['input_file' => 's3://my-bucket/data.csv'],
);
session = client.beta.sessions.create(
agent: agent.id,
environment_id: environment.id,
metadata: {input_file: "s3://my-bucket/data.csv"}
)
Environment env_... is a self-hosted environment. `resources` are not supported with self-hosted environments.
디플로이먼트가 셀프 호스팅 환경을 대상으로 할 때도 같은 규칙을 따라요.
reference의 셀프 호스팅 워커에서 CLI 플래그 전체 목록, SDK 헬퍼에서 SDK 헬퍼 옵션을 참고하세요.
메모리 스토어 사용하기 (Use memory stores)
셀프 호스팅 환경의 세션은 클라우드 환경의 세션과 똑같이 메모리 스토어를 붙여요. 세션을 만들 때 resources에 나열하세요. 세션에 메모리 스토어 붙이기에 표시된 대로요. 세션은 최대 8개의 메모리 스토어를 받아요. 셀프 호스팅 환경에서는 Anthropic 인프라가 아니라 SDK 워커가 각 스토어를 에이전트에게 구체화하므로, 메모리 스토어는 Python, TypeScript, Go SDK의 EnvironmentWorker(또는 그 handle_item() 메서드)가 필요해요.
ant CLI 워커(ant beta:worker poll과 ant beta:worker run)는 메모리 스토어를 마운트하지 않아요. CLI 폴러와 메모리 스토어를 결합하려면 세션당 샌드박스 하나 실행에 설명된 대로 세션당 샌드박스 안에서 SDK 워커를 실행하세요.
Claude Platform on AWS의 셀프 호스팅 환경에서는 세션에 메모리 스토어를 붙일 수 없어요.
워커가 메모리를 어떻게 처리하나 (How the worker handles memory)
워커가 세션에 메모리 스토어가 붙은 작업 항목을 가져오면:
- 각 붙은 스토어를 작업 항목의 세션별
secret으로 인증해 호스트의mount_path로 다운로드해요.mount_path는 클라우드 세션이 사용하는 것과 같은/mnt/memory/아래 디렉터리예요(예: "User Preferences"라는 스토어는/mnt/memory/user-preferences/), 그리고 세션의 시스템 프롬프트가 에이전트에게 그것을 설명해요. - 그 디렉터리들을 파일 도구의 허용 루트에,
access: "read_only"로 붙은 스토어의 디렉터리는 읽기 전용 루트에 추가해, 에이전트가 작업 디렉터리에서 쓰는 것과 같은read,write,edit,glob,grep도구로 메모리에서 작업하게 해요. - 도구 호출 후 로컬과 원격 변경을 동기화 간격(기본 15초)당 최대 한 번 조정해요: 스토어에서 바뀐 메모리는 디스크에 쓰고, 에이전트가 바꾼 파일은 스토어에 업로드해요.
- 세션이 끝나면 최종 동기화를 실행하고, 아직 보류 중인 업로드를 최대 30초까지 플러시하고, 만든 디렉터리를 제거해요. 세션이 실행되는 동안 취소된 워커는 최종 동기화를 건너뛰지만 종료 전에 변경된 파일을 업로드하고 디렉터리를 제거해요.
Anthropic 쪽 메모리 스토어가 진실의 원천으로 남아요. 메모리 버전, redaction, Console에서 메모리 보기·편집은 클라우드 세션과 같게 동작하고, 에이전트의 메모리 읽기·쓰기는 이벤트 스트림에 평범한 도구 이벤트로 나타나요. 각 워커가 간격으로 동기화하기 때문에, 한 세션에서 쓴 변경이 다른 실행 중인 세션에 보이려면 둘 다 동기화한 뒤여야 해요. 기본 간격에서 보통 1분보다 잘 아래예요. 클라우드 샌드박스의 세션은 서로의 변경을 거의 즉시 봐요.
각 스토어 디렉터리에는 디렉터리를 그 스토어에 묶는 .anthropic-memory-store라는 마커 파일이 있어요. 그대로 두세요: 워커는 마커가 없거나 변경된 디렉터리를 동기화하지 않아요.
호스트 준비 (Prepare the host)
셀프 호스팅 샌드박스의 메모리 스토어는 워커 호스트에 POSIX 파일시스템이 필요해요(시작하기 전에의 Linux 호스트). 워커가 메모리 파일을 열 때 O_NOFOLLOW를 요구하므로 Windows 호스트는 지원되지 않아요. 대소문자 구분 파일시스템이 권장되는데, 대소문자만 다른 메모리 경로가 충돌하지 않도록 해요.
워커를 시작하기 전에 부모 디렉터리를 만들고 워커가 실행되는 사용자가 쓸 수 있게 하세요:
sudo mkdir -p /mnt/memory && sudo chown "$USER" /mnt/memory
스토어별 디렉터리를 직접 만들지 마세요. 워커는 세션이 시작될 때 각 스토어의 mount_path 디렉터리(예: /mnt/memory/user-preferences)를 만들고, 그 경로에 무언가 이미 존재하면 세션의 작업을 시작하지 않으며, 세션이 끝나면 디렉터리를 제거해요. 두 가지 운영 규칙이 따라와요:
- 세션들이 같은 스토어를 붙일 때 파일시스템당 세션 하나 실행. 두 세션은 한 호스트에서 같은 스토어를 동시에 마운트할 수 없어요. 둘 다 같은 경로가 필요하기 때문이에요. 세션당 샌드박스 하나 실행에 설명된 대로 각 세션에 자체 샌드박스를 주는 것이 이 규칙을 충족해요.
- 워커를 우아하게 멈추세요. 세션이 실행되는 동안 워커를 멈출 때,
EnvironmentWorker는 취소될 때만 세션의 변경된 메모리 파일을 업로드하고 스토어 디렉터리를 제거해요. 죽은 프로세스는 정리를 실행하지 않고, 워커는 신호 핸들러를 스스로 설치하지 않아요. SIGTERM과 SIGINT를 그것을 실행하는 프로세스의 취소에 연결하세요: TypeScript에서는 워커에 전달하는signal을 중단하고, Go에서는 컨텍스트를 취소하고, Python에서는run()이나handle_item()을 실행하는 작업을 취소하세요. 워커가 프로세스일 때 이를 신호 핸들러에서 하세요(이 페이지의 독립 워커처럼). 워커가 웹훅 핸들러 안에서 실행될 때는 서버의 자체 종료 훅에서 하세요(서버의 신호를 가로채면 안 돼요). 그런 다음 SIGTERM으로 워커를 멈추고 하드 킬 전에 최소 30초의 종료 시간을 주세요. 최종 업로드가 그만큼 걸릴 수 있기 때문이에요. 워커가 정리 전에 죽으면, 그 스토어를 붙이는 다음 세션 전에/mnt/memory/아래 남은 스토어 디렉터리를 제거하세요. 그 안의 동기화되지 않은 편집은 손실돼요.
세션당 샌드박스 하나 실행 (Run one sandbox per session)
워커 실행의 샌드박스-세션당 패턴은 각 세션에 새 파일시스템을 주는데, 호스트 준비가 세션이 같은 스토어를 붙일 때 요구하는 것이에요. 호스트의 폴러로는 ant beta:worker poll --on-work(또는 SDK의 작업 폴러)를 유지하세요.
거기에 표시된 ant beta:worker run 엔트리포인트는 메모리 스토어를 마운트하지 않으므로, 대신 SDK 워커 중심으로 세션당 이미지를 구축하세요: 그 엔트리포인트는 EnvironmentWorker를 구성하고 handle_item()(TypeScript는 handleItem, Go는 HandleItem)을 호출하며, ANTHROPIC_* 변수에서 세션, 작업, 환경 식별자를 읽고 ANTHROPIC_WORK_SECRET에서 작업 항목의 세션별 secret을 읽어요. work_secret(TypeScript는 workSecret, Go는 WorkSecret)으로 secret을 명시적으로 전달할 수도 있어요.
async def main() -> None: async with AsyncAnthropic(auth_token=os.environ["ANTHROPIC_ENVIRONMENT_KEY"]) as client: worker = EnvironmentWorker(client, workdir="/workspace") # With no arguments, handle_item() reads the ANTHROPIC_* variables the spawn # script forwarded, including ANTHROPIC_WORK_SECRET. task = asyncio.create_task(worker.handle_item()) # Cancelling the task when the container is stopped lets the worker upload # changed memory files and remove the store directories before it exits. 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())
```typescript TypeScript
import Anthropic from "@anthropic-ai/sdk";
import { EnvironmentWorker } from "@anthropic-ai/sdk/helpers/beta/environments";
const client = new Anthropic({ authToken: process.env.ANTHROPIC_ENVIRONMENT_KEY });
const controller = new AbortController();
// Aborting when the container is stopped lets the worker upload changed memory
// files and remove the store directories before it exits.
process.once("SIGTERM", () => controller.abort());
process.once("SIGINT", () => controller.abort());
// With no arguments, handleItem() reads the ANTHROPIC_* variables the spawn
// script forwarded, including ANTHROPIC_WORK_SECRET.
await new EnvironmentWorker({
client,
workdir: "/workspace",
signal: controller.signal
}).handleItem();
// EnvironmentWorker is not currently available in the C# SDK.
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() {
// Cancelling the context when the container is stopped lets the worker upload
// changed memory files and remove the store directories before it exits.
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
client := anthropic.NewClient(option.WithAuthToken(os.Getenv("ANTHROPIC_ENVIRONMENT_KEY")))
worker := environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{
Workdir: "/workspace",
})
// With zero-value options, HandleItem reads the ANTHROPIC_* variables the spawn
// script forwarded, including ANTHROPIC_WORK_SECRET.
if err := worker.HandleItem(ctx, environments.HandleItemOptions{}); err != nil {
log.Fatalf("worker: %v", err)
}
}
// EnvironmentWorker is not currently available in the Java SDK.
// EnvironmentWorker is not currently available in the PHP SDK.
# EnvironmentWorker is not currently available in the Ruby SDK.
ant beta:worker poll --on-work는 만드는 스크립트에 ANTHROPIC_WORK_SECRET을 설정하지 않으므로, 스폰 스크립트는 표준 입력의 작업 항목 JSON에서 secret을 읽어 샌드박스로 전달해야 해요:
#!/bin/bash
# spawn.sh: called once per claimed work item
# The claimed work item arrives as JSON on stdin. Its secret is the
# per-session credential that the memory store endpoints require.
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가 필요해요(호스트 준비 참고). 각 샌드박스가 세션 하나를 서빙하고 이후 폐기되므로 남은 디렉터리 정리가 필요 없고, 메모리 디렉터리를 호스트에 바인드 마운트할 필요도 없어요: 워커가 샌드박스 종료 전에 그 내용을 스토어에 업로드해요. 세션이 끝나기 전에 컨테이너를 멈추면 업로드가 여전히 실행되도록 엔트리포인트가 취소로 바꾸는 신호(호스트 준비 참고)를 보내고, 죽이지 마세요. 컨테이너에 업로드를 끝낼 시간도 주세요: Docker는 기본적으로 10초 후에 정지 신호 뒤에 SIGKILL이 따라오므로, 호스트 준비가 요구하는 최소 30초로 docker run에 --stop-timeout을 주거나 오케스트레이터의 종료 유예 기간을 올려 그 한도를 높이세요.
동기화 구성 (Configure sync)
두 개의 EnvironmentWorker 옵션이 메모리 동작을 제어해요:
memory_sync_interval(Python, 초 단위; TypeScript는memorySyncIntervalMs, 밀리초; Go는MemorySyncInterval, 지속 시간): 세션이 실행되는 동안 붙은 스토어가 서버와 얼마나 자주 조정하는지. 기본 15초, 최소 5초. 간격이 짧을수록 다른 세션이 오래된 메모리를 보는 창이 좁아지지만 메모리 스토어 요청 비용이 커져요. Python의None, TypeScript의null, Go의 음수 지속 시간은 메모리 지원을 완전히 비활성화해요: 워커가 스토어를 다운로드하지도 동기화하지도 않고, 메모리 스토어가 붙은 세션이 그 시스템 프롬프트가 여전히 그것을 설명해도 메모리 없이 실행돼요. 그러니 붙은 메모리 스토어가 없는 세션의 워커에서만 메모리 지원을 비활성화하세요. 메모리 지원이 켜져 있는 동안, 붙은 스토어가 있는 세션에 세션별secret없이 도착한 작업 항목은 메모리 없이 실행되는 대신 실패해요(메모리 마운트 문제 해결 참고).memory_sync_deletions(TypeScript는memorySyncDeletions, Go는MemorySyncDeletions): 에이전트가 로컬에서 삭제한 파일도 스토어에서 삭제할지. Python과 TypeScript에서 값은"enabled"(기본값),"log_only","disabled"중 하나이고, Go에서는 상수environments.MemorySyncDeletionsEnabled(제로 값),environments.MemorySyncDeletionsLogOnly,environments.MemorySyncDeletionsDisabled중 하나예요. enabled면 워커가 이후 동기화에서 파일이 여전히 사라진 것을 확인하면 스토어에서 그 메모리를 삭제해요. log-only 모드에서는 같은 검사를 실행하지만 지웠을 것을 로그만 남겨, enabled 모드를 신뢰하기 전에 워커가 무엇을 지울지 지켜볼 수 있어요. disabled면 스토어에서 결코 삭제하지 않아요. 업로드와 다운로드는 이 설정의 영향을 받지 않아요.
이 옵션들을 워커를 구성하는 곳에서 설정하세요. EnvironmentWorker 생성자를 통해서든, Python과 TypeScript에서는 웹훅 핸들러가 사용하는 client.beta.environments.work.worker() 팩토리를 통해서든요.
예를 들어, 10초마다 동기화하고 워커가 지울 삭제만 로그로 남기려면:
const worker = new EnvironmentWorker({
client,
environmentId,
environmentKey,
workdir: "/workspace",
memorySyncIntervalMs: 10_000,
memorySyncDeletions: "log_only"
});
// EnvironmentWorker is not currently available in the C# SDK.
worker := environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{
EnvironmentID: environmentID,
EnvironmentKey: environmentKey,
Workdir: "/workspace",
MemorySyncInterval: 10 * time.Second,
MemorySyncDeletions: environments.MemorySyncDeletionsLogOnly,
})
// EnvironmentWorker is not currently available in the Java SDK.
// EnvironmentWorker is not currently available in the PHP SDK.
# EnvironmentWorker is not currently available in the Ruby SDK.
읽기 전용 스토어와 충돌 (Read-only stores and conflicts)
access: "read_only"로 붙은 스토어의 경우 write와 edit 도구가 그 디렉터리 안의 파일 변경을 거부하고, 워커는 그것에서 결코 업로드하지 않아요. bash를 통해 또는 샌드박스에서 서빙하는 커스텀 도구나 MCP 서버를 통해 만들어진 변경은 로컬에서 차단되지 않아요: 그것들은 스토어에 결코 동기화되지 않고, 그 메모리에 대한 다음 원격 변경이 그것들을 덮어써요. 세션 중 로컬 복사본 자체가 변하지 않게 해야 한다면 그 에이전트에 대해 bash 도구를 비활성화하고 샌드박스 파일시스템에 쓰는 커스텀 도구를 주지 마세요. 스토어 경로를 읽기 전용으로 마운트하지 마세요. 워커 자신이 디렉터리를 만들고 다운로드한 메모리를 그 안에 써야 하기 때문이에요.
충돌은 스토어에 유리하게 해결돼요. 에이전트가 세션이 마지막으로 동기화한 이후 스토어에서도 바뀐 메모리 파일을 변경하면, 워커가 다음 동기화에서 스토어 버전을 유지하고 로컬 파일을 그것으로 덮어쓰며 경고를 로그해요. write와 edit 도구 자체는 성공하고 오류가 에이전트에 닿지 않아요. 에이전트의 변경이 여전히 적용된다면 동기화 후 파일을 다시 읽고 다시 변경할 수 있어요.
메모리 마운트 문제 해결 (Troubleshoot memory mounts)
워커는 마운트와 백그라운드 동기화 실패를 세션에 보고하는 대신 로그해요. 읽기 전용 거부만 도구 오류로 에이전트에 닿아요(읽기 전용 스토어와 충돌 참고). 워커가 세션을 가져올 때 메모리 스토어를 마운트할 수 없으면 작업 항목이 실패해요: 세션은 오류 이벤트를 발행하지 않고 idle로 남아요.
| 증상 (Symptom) | 원인 (Cause) | 해결책 (Fix) |
|---|---|---|
워커 로그에 the work item carried no sessions token(Go에서는 ErrSessionMemoryNoToken 오류)이 있고 작업 항목이 실패해요. |
작업 항목의 세션별 secret이 워커에 닿지 않았어요: 셀프 호스팅 샌드박스의 메모리 스토어가 조직에 대해 활성화되지 않았거나, 스폰 스크립트가 secret을 샌드박스로 전달하지 않았어요. |
샌드박스-세션당 패턴에서 세션당 샌드박스 하나 실행에 표시된 대로 ANTHROPIC_WORK_SECRET을 샌드박스로 전달하세요. 워커가 한 프로세스에서 폴링과 세션 실행을 하고 그래도 이 로그가 나오면 지원에 연락하세요. |
워커 로그에 something already exists at the memory store's path. |
정리 전에 워커가 죽은 이전 세션에서 남은 디렉터리. | 로그 줄이 이름 붙이는 남은 디렉터리를 제거하세요. 동기화되지 않은 편집은 손실돼요. |
워커 로그에 cannot create the memory store's folder와 the worker host must make this mount path writable. |
워커가 실행되는 사용자가 /mnt/memory 아래에 디렉터리를 만들 수 없어요. |
/mnt/memory를 만들고 그 사용자에게 chown하세요. 호스트 준비 참고. |
워커가 작업 항목을 가져온 직후 세션은 오류 이벤트 없이 requires_action stop reason으로 idle에 앉아 있어요. |
워커가 앞선 이유 중 하나로 메모리 스토어를 마운트할 수 없어 작업 항목을 실패시켰어요. | 호스트에서 원인을 고치고 user.interrupt 이벤트를 보내세요. 세션의 작업이 다시 큐에 들어가고 다음에 가져가는 워커가 마운트를 다시 시도해요. |
샌드박스에서 커스텀 도구 서빙 (Serve custom tools from your sandbox)
커스텀 도구는 여러분의 코드가 실행하는 도구예요: 에이전트가 agent.custom_tool_use 이벤트를 발행하고 일치하는 user.custom_tool_result를 기다려요. 워커가 그 코드일 수 있고, 샌드박스 안에서 실행되므로 도구는 샌드박스에 대해 구성한 내부 서비스, 자격 증명, 네트워크 이그레스에만 닿고 그 이상은 아니에요. 환경 키가 커스텀 도구 결과를 게시할 권한을 부여하므로 Claude API 키는 워커 호스트 밖에 머물러요.
```json
{
"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"]
}
}
```
<CodeGroup exclude="shell">
```python Python
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."""
# Runs on the worker host: call anything the sandbox can reach.
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())
```
```typescript TypeScript
import Anthropic from "@anthropic-ai/sdk";
import { EnvironmentWorker } from "@anthropic-ai/sdk/helpers/beta/environments";
import { betaTool } from "@anthropic-ai/sdk/helpers/beta/json-schema";
import { betaAgentToolset20260401 } from "@anthropic-ai/sdk/tools/agent-toolset/node";
const getOrderStatus = betaTool({
name: "get_order_status",
description: "Look up an order in the internal fulfillment system by order ID.",
inputSchema: {
type: "object",
properties: { order_id: { type: "string", description: "The order ID" } },
required: ["order_id"]
},
// Runs on the worker host: call anything the sandbox can reach.
run: async ({ order_id }) => `Order ${order_id}: shipped`
});
const environmentKey = process.env.ANTHROPIC_ENVIRONMENT_KEY!;
const environmentId = process.env.ANTHROPIC_ENVIRONMENT_ID!;
const client = new Anthropic({ authToken: environmentKey });
const controller = new AbortController();
process.once("SIGTERM", () => controller.abort());
await new EnvironmentWorker({
client,
environmentId,
environmentKey,
workdir: "/workspace",
signal: controller.signal,
tools: (ctx) => [...betaAgentToolset20260401(ctx), getOrderStatus]
}).run();
```
```csharp C#
// EnvironmentWorker is not currently available in the C# SDK.
// To answer custom tool calls directly, see the session event stream.
```
```go Go
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"
"github.com/anthropics/anthropic-sdk-go/toolrunner"
"github.com/anthropics/anthropic-sdk-go/tools/agenttoolset"
)
type orderStatusInput struct {
OrderID string `json:"order_id"`
}
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()
getOrderStatus := toolrunner.NewBetaTool(
"get_order_status",
"Look up an order in the internal fulfillment system by order ID.",
anthropic.BetaToolInputSchemaParam{
Properties: map[string]any{
"order_id": map[string]any{"type": "string", "description": "The order ID"},
},
Required: []string{"order_id"},
},
// Runs on the worker host: call anything the sandbox can reach.
func(ctx context.Context, input orderStatusInput) (anthropic.BetaToolResultBlockParamContentUnion, error) {
return anthropic.BetaToolResultBlockParamContentUnion{
OfText: &anthropic.BetaTextBlockParam{Text: "Order " + input.OrderID + ": shipped"},
}, nil
},
)
client := anthropic.NewClient(option.WithAuthToken(environmentKey))
worker := environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{
EnvironmentID: environmentID,
EnvironmentKey: environmentKey,
Workdir: "/workspace",
ToolsFunc: func(env *agenttoolset.AgentToolContext) []anthropic.BetaTool {
return append(agenttoolset.BetaAgentToolset20260401(env), getOrderStatus)
},
})
if err := worker.Run(ctx); err != nil {
log.Fatalf("worker: %v", err)
}
}
```
```java Java
// EnvironmentWorker is not currently available in the Java SDK.
// To answer custom tool calls directly, see the session event stream.
```
```php PHP
// EnvironmentWorker is not currently available in the PHP SDK.
// To answer custom tool calls directly, see the session event stream.
```
```ruby Ruby
# EnvironmentWorker is not currently available in the Ruby SDK.
# To answer custom tool calls directly, see the session event stream.
```
</CodeGroup>
워커는 등록된 도구만 응답해요. 에이전트에 선언됐지만 어떤 워커나 클라이언트에도 등록되지 않은 커스텀 도구는 누군가 그 결과를 게시할 때까지 requires_action stop reason으로 세션을 멈춰 둬요. 이벤트 흐름은 커스텀 도구 호출 처리를 참고하세요.
MCP 서버를 커스텀 도구로 감싸기 (Wrap an MCP server as custom tools)
MCP 커넥터는 Anthropic 쪽에서 MCP 서버에 연결하므로, 서버는 직접 또는 MCP 터널을 통해 Anthropic이 닿을 수 있는 HTTP 엔드포인트를 노출해야 해요. 네트워크 안에서만 닿을 수 있는 서버를 사용하려면 대신 워커를 MCP 클라이언트로 만들고 서버의 도구를 커스텀 도구로 선언하세요. MCP 서버는 네트워크 밖에서 인바운드 연결이 필요 없어요. Anthropic은 에이전트에 선언한 도구 정의, 각 호출의 입력, 워커가 다시 게시하는 결과를 받아요. 런타임에서 모델은 감싼 도구를 다른 커스텀 도구처럼 호출해요:
- 에이전트가
agent.custom_tool_use이벤트를 발행해요. - 샌드박스 안의 워커가 열려 있는 MCP 세션을 통해 네트워크의 서버에 호출을 전달해요.
- 워커가 서버의 응답을
user.custom_tool_result로 게시해요.
SDK의 클라이언트 측 MCP 헬퍼가 서버의 도구를 워커가 받아들이는 실행 가능한 도구로 변환해요. Anthropic SDK와 함께 MCP SDK를 설치하세요(pip install "anthropic[mcp]" "mcp>=1.24", npm install @modelcontextprotocol/sdk, go get github.com/modelcontextprotocol/go-sdk). 예시는 인증 없이 연결해요. 자격 증명을 보내려면 MCP 트랜스포트에 넘기는 HTTP 클라이언트나 요청 옵션(http_client (Python), requestInit (TypeScript), HTTPClient (Go))을 구성하세요.
<CodeGroup exclude="shell">
```python Python
import asyncio
from typing import Any, cast
from anthropic import AsyncAnthropic
from anthropic.types.beta import BetaManagedAgentsCustomToolParams
from mcp import ClientSession, types
# Requires mcp >= 1.24, which renamed streamablehttp_client to 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:
# The MCP fields map one to one onto a custom tool declaration. The cast
# hands the schema dictionary to the SDK's typed parameter unchanged.
return {
"type": "custom",
"name": tool.name,
"description": tool.description or tool.name,
"input_schema": cast(Any, tool.inputSchema),
}
async def main() -> None:
# Run this wherever you create agents, not on the worker host: it
# authenticates with your Claude API key (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())
```
```typescript TypeScript
import Anthropic from "@anthropic-ai/sdk";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
const MCP_SERVER_URL = "http://mcp.internal.example.com:8000/mcp";
// Run this wherever you create agents, not on the worker host: it
// authenticates with your Claude API key (ANTHROPIC_API_KEY).
const client = new Anthropic();
const mcpClient = new Client({ name: "declare-agent-tools", version: "1.0.0" });
await mcpClient.connect(new StreamableHTTPClientTransport(new URL(MCP_SERVER_URL)));
const { tools } = await mcpClient.listTools();
const agent = await client.beta.agents.create({
name: "Internal tools agent",
model: "claude-opus-5-5",
tools: [
{ type: "agent_toolset_20260401" },
// The MCP fields map one to one onto a custom tool declaration.
...tools.map((tool) => ({
type: "custom" as const,
name: tool.name,
description: tool.description || tool.name,
input_schema: tool.inputSchema
}))
]
});
console.log(agent.id);
await mcpClient.close();
```
```csharp C#
// See the Python, TypeScript, and Go tabs. Declaring custom tools from
// C# works the same way once you list the server's tools with an MCP client.
```
```go Go
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"github.com/anthropics/anthropic-sdk-go"
mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp"
)
const mcpServerURL = "http://mcp.internal.example.com:8000/mcp"
// toCustomTool maps one MCP tool definition onto a custom tool declaration.
// The fields map one to one: the typed parameter carries `properties` and
// `required`, and every other JSON Schema keyword the server emits travels in
// ExtraFields so the declared schema matches the server's schema.
func toCustomTool(tool *mcpsdk.Tool) (anthropic.BetaAgentNewParamsToolUnion, error) {
raw, err := json.Marshal(tool.InputSchema)
if err != nil {
return anthropic.BetaAgentNewParamsToolUnion{}, err
}
var schema map[string]any
if err := json.Unmarshal(raw, &schema); err != nil {
return anthropic.BetaAgentNewParamsToolUnion{}, err
}
inputSchema := anthropic.BetaManagedAgentsCustomToolInputSchemaParam{ExtraFields: map[string]any{}}
for keyword, value := range schema {
switch keyword {
case "type":
// The parameter type always marshals "type": "object".
case "properties":
properties, _ := value.(map[string]any)
inputSchema.Properties = properties
case "required":
entries, _ := value.([]any)
for _, entry := range entries {
if name, isString := entry.(string); isString {
inputSchema.Required = append(inputSchema.Required, name)
}
}
default:
inputSchema.ExtraFields[keyword] = value
}
}
description := tool.Description
if description == "" {
description = tool.Name
}
return anthropic.BetaAgentNewParamsToolUnion{
OfCustom: &anthropic.BetaManagedAgentsCustomToolParams{
Type: anthropic.BetaManagedAgentsCustomToolParamsTypeCustom,
Name: tool.Name,
Description: description,
InputSchema: inputSchema,
},
}, nil
}
func main() {
ctx := context.Background()
// Run this wherever you create agents, not on the worker host: it
// authenticates with your Claude API key (ANTHROPIC_API_KEY).
client := anthropic.NewClient()
mcpClient := mcpsdk.NewClient(&mcpsdk.Implementation{Name: "declare-agent-tools", Version: "1.0.0"}, nil)
session, err := mcpClient.Connect(ctx, &mcpsdk.StreamableClientTransport{Endpoint: mcpServerURL}, nil)
if err != nil {
log.Fatalf("connect to MCP server: %v", err)
}
defer session.Close()
listed, err := session.ListTools(ctx, nil)
if err != nil {
log.Fatalf("list MCP tools: %v", err)
}
tools := []anthropic.BetaAgentNewParamsToolUnion{
{OfAgentToolset20260401: &anthropic.BetaManagedAgentsAgentToolset20260401Params{
Type: anthropic.BetaManagedAgentsAgentToolset20260401ParamsTypeAgentToolset20260401,
}},
}
for _, tool := range listed.Tools {
custom, err := toCustomTool(tool)
if err != nil {
log.Fatalf("convert MCP tool %s: %v", tool.Name, err)
}
tools = append(tools, custom)
}
agent, err := client.Beta.Agents.New(ctx, anthropic.BetaAgentNewParams{
Name: "Internal tools agent",
Model: anthropic.BetaManagedAgentsModelConfigParams{ID: anthropic.BetaManagedAgentsModelClaudeOpus5_5},
Tools: tools,
})
if err != nil {
log.Fatalf("create agent: %v", err)
}
fmt.Println(agent.ID)
}
```
```java Java
// See the Python, TypeScript, and Go tabs. Declaring custom tools from
// Java works the same way once you list the server's tools with an MCP client.
```
```php PHP
// See the Python, TypeScript, and Go tabs. Declaring custom tools from
// PHP works the same way once you list the server's tools with an MCP client.
```
```ruby Ruby
# See the Python, TypeScript, and Go tabs. Declaring custom tools from
# Ruby works the same way once you list the server's tools with an MCP client.
```
</CodeGroup>
<CodeGroup exclude="shell">
```python Python
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
# Requires mcp >= 1.24, which renamed streamablehttp_client to 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"]
# Connect to the MCP server once at startup and keep the session open for
# the life of the worker. The timeout turns a hung tool call into an error
# result instead of a stalled call.
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())
```
```typescript TypeScript
import Anthropic from "@anthropic-ai/sdk";
import { EnvironmentWorker } from "@anthropic-ai/sdk/helpers/beta/environments";
import {
mcpTools,
type MCPCallToolResultLike,
type MCPClientLike
} from "@anthropic-ai/sdk/helpers/beta/mcp";
import { betaAgentToolset20260401 } from "@anthropic-ai/sdk/tools/agent-toolset/node";
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
const MCP_SERVER_URL = "http://mcp.internal.example.com:8000/mcp";
const environmentKey = process.env.ANTHROPIC_ENVIRONMENT_KEY!;
const environmentId = process.env.ANTHROPIC_ENVIRONMENT_ID!;
const client = new Anthropic({ authToken: environmentKey });
const controller = new AbortController();
process.once("SIGTERM", () => controller.abort());
// Connect to the MCP server once at startup and keep the connection open for
// the life of the worker.
const mcpClient = new Client({ name: "sandbox-worker", version: "1.0.0" });
await mcpClient.connect(new StreamableHTTPClientTransport(new URL(MCP_SERVER_URL)));
const { tools } = await mcpClient.listTools();
// The MCP SDK's callTool return type still includes a legacy result shape that
// mcpTools does not accept; narrow it. Drop this once MCPClientLike widens.
const mcpClientForTools: MCPClientLike = {
callTool: (params) => mcpClient.callTool(params) as Promise<MCPCallToolResultLike>
};
await new EnvironmentWorker({
client,
environmentId,
environmentKey,
workdir: "/workspace",
signal: controller.signal,
tools: (ctx) => [...betaAgentToolset20260401(ctx), ...mcpTools(tools, mcpClientForTools)]
}).run();
```
```csharp C#
// EnvironmentWorker is not currently available in the C# SDK.
```
```go Go
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/mcp"
"github.com/anthropics/anthropic-sdk-go/option"
"github.com/anthropics/anthropic-sdk-go/tools/agenttoolset"
mcpsdk "github.com/modelcontextprotocol/go-sdk/mcp"
)
const mcpServerURL = "http://mcp.internal.example.com:8000/mcp"
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))
// Connect to the MCP server once at startup and keep the session open for
// the life of the worker.
mcpClient := mcpsdk.NewClient(&mcpsdk.Implementation{Name: "sandbox-worker", Version: "1.0.0"}, nil)
session, err := mcpClient.Connect(ctx, &mcpsdk.StreamableClientTransport{Endpoint: mcpServerURL}, nil)
if err != nil {
log.Fatalf("connect to MCP server: %v", err)
}
defer session.Close()
listed, err := session.ListTools(ctx, nil)
if err != nil {
log.Fatalf("list MCP tools: %v", err)
}
mcpTools, err := mcp.NewBetaTools(listed.Tools, session)
if err != nil {
log.Fatalf("convert MCP tools: %v", err)
}
worker := environments.NewEnvironmentWorker(client, environments.EnvironmentWorkerOptions{
EnvironmentID: environmentID,
EnvironmentKey: environmentKey,
Workdir: "/workspace",
ToolsFunc: func(env *agenttoolset.AgentToolContext) []anthropic.BetaTool {
return append(agenttoolset.BetaAgentToolset20260401(env), mcpTools...)
},
})
if err := worker.Run(ctx); err != nil {
log.Fatalf("worker: %v", err)
}
}
```
```java Java
// EnvironmentWorker is not currently available in the Java SDK.
```
```php PHP
// EnvironmentWorker is not currently available in the PHP SDK.
```
```ruby Ruby
# EnvironmentWorker is not currently available in the Ruby SDK.
```
</CodeGroup>
MCP 서버를 감쌀 때 다음을 염두에 두세요:
- 도구는 선언되지 실행 시점에 발견되지 않아요. 워커는 시작 시 MCP 서버의 도구를 한 번 나열하고 실행 중인 세션에 도구를 추가할 수 없어요. 서버의 도구가 바뀌면 에이전트 구성 업데이트로 에이전트나 유휴 세션에서 다시 선언하고 워커를 재시작하세요.
- 이름과 설명이 Managed Agents API에 맞아야 해요. 커스텀 도구 이름은 에이전트당 고유하며 문자, 숫자, 밑줄, 하이픈(1–128자)을 사용하고, 비어 있지 않은 설명이 필요하며, 에이전트의
tools배열은 최대 128개 항목을 받아요(감싼 각 도구가 항목 하나, 내장 도구세트가 하나 더). API는 도구 이름을 재사용하는 선언,bash나read같은 내장 에이전트 도구 이름을 따는 커스텀 도구, 예약된mcp__접두사를 사용하는 선언을 거부해요. MCP 헬퍼는 서버의 이름과 설명을 유지하므로 필요한 곳에서 이름을 바꾸거나 다듬으세요. 두 서버가 같은 도구 이름을 노출하면 접두사가 붙은 이름 아래에서 래퍼를 직접 정의하고 그것이 서버의 원래 도구 이름을 호출하게 하세요. - 대부분 스키마는 변경 없이 통과돼요. API는 MCP 서버가 흔히 발행하는
additionalProperties,title같은 JSON Schema 키워드를 받아줘요.$ref같은 참조 키워드는 커스텀 도구의input_schema어디에도 거부하므로, pydantic 같은 생성기가$defs로 분리한 스키마는 인라인하세요. 최상위oneOf,anyOf,allOf와 문자, 숫자, 밑줄, 점, 하이픈(1–64자) 밖의 속성 이름도 거부해요. - 도구 실패는 오류 도구 결과로 표면화돼요. MCP 서버가 도구 오류를 보고하면 워커가 모델이 반응할 수 있는 오류 도구 결과를 게시해요. 오디오 블록과 리소스 링크 같은 도구 결과 등가물이 없는 MCP 콘텐츠도 오류로 표면화돼요. Python 워커 예시가
read_timeout_seconds로 하는 것처럼 더 빠르고 명확한 실패를 위해 MCP 클라이언트에 타임아웃을 설정하세요. 없으면 TypeScript MCP SDK의 기본 요청 타임아웃(약 1분)이 발화하거나 워커 자체의 백스톱(Python에서 약 2분 반, Go에서 2분)이 발화할 때만 멈춘 호출이 오류 결과가 돼요. Go에서는 워커가 120초 기본값을 넘긴 도구 호출을 취소하고 오류 결과를 게시해요. - 운영하거나 신뢰하는 서버를 감싸세요. 감싼 도구의 이름, 설명, 결과는 다른 도구처럼 모델의 컨텍스트에 들어가요. 신뢰할 수 없는 입력이 에이전트가 워커 호스트의
bash를 포함한 다른 도구로 하는 일에 영향을 줄 수 있어요. 에이전트가 사용하도록 의도한 도구만 선언하세요. - 권한 정책은 커스텀 도구에 적용되지 않아요. 권한 정책은 내장 및 MCP 도구세트를 관장해요. 워커는 모델이 만드는 감싼 모든 도구 호출을 실행하므로, 승인 단계가 필요하면 자체 도구 코드에 두세요.
모니터링과 운영 (Monitoring and operations)
이 호출들은 여러분의 Claude API 키로 인증된 모니터링 또는 운영 도구에서 실행되어 워커 플릿을 관찰하고 관리해요. 워커 헬퍼 안에서 가져오기와 유지(keep-alive) 루프가 처리되므로 그 엔드포인트를 직접 호출하지 않아요.
큐 깊이 읽기 (Read queue depth)
work.stats는 환경의 큐 상태를 반환해요:
depth는 가져와지길 기다리는 항목의 수예요. 이 값으로 워커 플릿을 확장하거나 백로그에 경보를 걸어요.pending은 워커가 가져왔지만 아직 승인되지 않은 항목의 수예요. 워커 헬퍼는 각 항목을 처리하기 전에 승인하므로 정상 운영에서 이 값은 0에 가까워요. 지속적인 비제로 값은 워커가 가져오기와 승인 사이에서 멈췄다는 뜻이에요.oldest_queued_at은 큐에 여전히 있는(가져와지길 기다리거나 가져왔지만 아직 승인되지 않은) 가장 오래된 항목의 타임스탬프로, 없으면null이에요.workers_polling은 지난 30초 안에 폴링한 워커의 수예요. 생존(liveness) 경보에 사용하세요.
ant beta:environments:work stats --environment-id "$ANTHROPIC_ENVIRONMENT_ID"
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}")
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();
const stats = await client.beta.environments.work.stats(process.env.ANTHROPIC_ENVIRONMENT_ID!);
console.log(`depth=${stats.depth} pending=${stats.pending}`);
using Anthropic;
var client = new AnthropicClient();
var environmentId = Environment.GetEnvironmentVariable("ANTHROPIC_ENVIRONMENT_ID")!;
var stats = await client.Beta.Environments.Work.Stats(environmentId);
Console.WriteLine($"depth={stats.Depth} pending={stats.Pending}");
package main
import (
"context"
"fmt"
"os"
"github.com/anthropics/anthropic-sdk-go"
)
func main() {
client := anthropic.NewClient()
environmentID := os.Getenv("ANTHROPIC_ENVIRONMENT_ID")
stats, err := client.Beta.Environments.Work.Stats(
context.Background(),
environmentID,
anthropic.BetaEnvironmentWorkStatsParams{},
)
if err != nil {
panic(err)
}
fmt.Printf("depth=%d pending=%d\n", stats.Depth, stats.Pending)
}
import com.anthropic.client.AnthropicClient;
import com.anthropic.client.okhttp.AnthropicOkHttpClient;
import com.anthropic.models.beta.environments.work.BetaSelfHostedWorkQueueStats;
void main() {
AnthropicClient client = AnthropicOkHttpClient.fromEnv();
BetaSelfHostedWorkQueueStats stats = client.beta()
.environments()
.work()
.stats(System.getenv("ANTHROPIC_ENVIRONMENT_ID"));
IO.println("depth=" + stats.depth() + " pending=" + stats.pending());
}
<?php
use Anthropic\Client;
$client = new Client();
$stats = $client->beta->environments->work->stats(getenv('ANTHROPIC_ENVIRONMENT_ID'));
printf("depth=%d pending=%d\n", $stats->depth, $stats->pending);
require "anthropic"
client = Anthropic::Client.new
stats = client.beta.environments.work.stats(ENV.fetch("ANTHROPIC_ENVIRONMENT_ID"))
puts "depth=#{stats.depth} pending=#{stats.pending}"
{
"type": "work_queue_stats",
"depth": 0,
"pending": 0,
"oldest_queued_at": null,
"workers_polling": 0
}
세션을 우아하게 멈추기 (Stop a session gracefully)
특정 세션을 처리하는 워커에 종료를 요청하려면 work.stop을 사용하세요. 기본적으로 작업 항목은 stopping으로 이동해요: 워커가 다음 임대 하트비트에서 알게 되고, 세션의 진행 중 도구 호출을 취소하고, 종료를 확인하며, 그 시점에 작업 항목이 stopped가 돼요. 요청 본문에 force: true(CLI에서는 --force)를 전달하면 워커의 확인을 기다리는 대신 작업 항목을 즉시 stopped로 표시해요.
이 호출들은 워커 호스트가 아닌 운영 도구에서 실행되므로 ANTHROPIC_WORK_ID가 자동으로 설정되지 않아요. 다음 예시를 실행하기 전에 대상 작업 항목의 ID로 설정하세요. 작업 항목의 ID를 찾으려면 Environments Work 엔드포인트로 환경의 작업 항목을 나열하세요.
ant beta:environments:work stop \
--environment-id "$ANTHROPIC_ENVIRONMENT_ID" \
--work-id "$ANTHROPIC_WORK_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)
import Anthropic from "@anthropic-ai/sdk";
const client = new Anthropic();
const work = await client.beta.environments.work.stop(process.env.ANTHROPIC_WORK_ID!, {
environment_id: process.env.ANTHROPIC_ENVIRONMENT_ID!
});
console.log(work.state);
using Anthropic;
var client = new AnthropicClient();
var work = await client.Beta.Environments.Work.Stop(
Environment.GetEnvironmentVariable("ANTHROPIC_WORK_ID")!,
new()
{
EnvironmentID = Environment.GetEnvironmentVariable("ANTHROPIC_ENVIRONMENT_ID")!
}
);
Console.WriteLine(work.State);
package main
import (
"context"
"fmt"
"os"
"github.com/anthropics/anthropic-sdk-go"
)
func main() {
client := anthropic.NewClient()
work, err := client.Beta.Environments.Work.Stop(
context.Background(),
os.Getenv("ANTHROPIC_WORK_ID"),
anthropic.BetaEnvironmentWorkStopParams{
EnvironmentID: os.Getenv("ANTHROPIC_ENVIRONMENT_ID"),
},
)
if err != nil {
panic(err)
}
fmt.Println(work.State)
}
import com.anthropic.client.AnthropicClient;
import com.anthropic.client.okhttp.AnthropicOkHttpClient;
import com.anthropic.models.beta.environments.work.BetaSelfHostedWork;
import com.anthropic.models.beta.environments.work.BetaSelfHostedWorkStopRequest;
import com.anthropic.models.beta.environments.work.WorkStopParams;
void main() {
AnthropicClient client = AnthropicOkHttpClient.fromEnv();
BetaSelfHostedWork work = client.beta().environments().work().stop(
WorkStopParams.builder()
.environmentId(System.getenv("ANTHROPIC_ENVIRONMENT_ID"))
.workId(System.getenv("ANTHROPIC_WORK_ID"))
.betaSelfHostedWorkStopRequest(BetaSelfHostedWorkStopRequest.builder().build())
.build()
);
IO.println(work.state());
}
<?php
use Anthropic\Client;
$client = new Client();
$work = $client->beta->environments->work->stop(
getenv('ANTHROPIC_WORK_ID'),
environmentID: getenv('ANTHROPIC_ENVIRONMENT_ID'),
);
echo $work->state . "\n";
require "anthropic"
client = Anthropic::Client.new
work = client.beta.environments.work.stop(
ENV.fetch("ANTHROPIC_WORK_ID"),
environment_id: ENV.fetch("ANTHROPIC_ENVIRONMENT_ID")
)
puts work.state