An EnvironmentWorker claims work items from a self-hosted CMA environment, downloads the agent's skills, runs a session tool runner per claimed item, and keeps its lease alive via parallel heartbeats—terminating only on control-plane stop or unrecoverable failure. The worker is configured via factory method client.beta.environments.work.worker() with credentials (environment_key), an optional tools factory to bind context-aware tools per session, and a working directory for tool path resolution. A CMA (Computer Management Agent) environment is a self-hosted server that queues agent sessions as work items for workers to claim and process. By running agent tools locally and returning results to the CMA environment, an EnvironmentWorker keeps compute off Anthropic's infrastructure.
EnvironmentWorker (in src/anthropic/lib/environments/_worker.py) polls a self-hosted CMA environment for work items and, for each claimed session item, builds a per-session AgentToolContext, downloads the session agent's skills, runs a SessionToolRunner, and heartbeats the work-item lease in parallel — then force-stops the item and loops to the next one.[1] EnvironmentWorker is async-only: run() loops forever and must be cancelled or wrapped in asyncio.wait_for to stop.[1]
The preferred way to build an EnvironmentWorker is via client.beta.environments.work.worker(environment_id=..., environment_key=...), which is equivalent to calling the constructor directly.[1] A single environment_key is the worker's only credential: a Bearer-only scoped sub-client is built once per call (one for polling, one for heartbeat/force-stop; the session tool runner builds its own internally), with the parent client's X-Api-Key cleared on every request.[1]
The tools parameter accepts either a fixed Sequence[BetaAnyRunnableTool] or a factory Callable[[AgentToolContext], Sequence[BetaAnyRunnableTool]] invoked once per claimed session; it defaults to beta_agent_toolset_20260401(env). Use the factory form to bind tools that need the workdir or session ID to the right session.[1] The workdir parameter defaults to os.getcwd() captured at construction time, so a chdir between constructing the worker and serving a session does not change where tools resolve paths.[1]
EnvironmentWorker.handle_item() runs the same per-work-item flow for a single already-claimed item. Called with no arguments, it reads the ANTHROPIC_* environment variables that ant worker poll --on-work sets on the spawned process.[1] The internal _require helper resolves a parameter value by falling back to the named environment variable (e.g. ANTHROPIC_ENVIRONMENT_KEY), raising a descriptive ValueError if neither the argument nor the variable is set.[1]
The heartbeat interval defaults to 30 s (_HEARTBEAT_DEFAULT) and is dynamically adjusted to min(ttl_seconds / 2, 30) once the server reports the real TTL; the assumed TTL before the first response is 90 s (_HEARTBEAT_TTL_DEFAULT).[1] Each heartbeat call is bounded by anyio.fail_after(interval) so a network blackhole cannot leave the loop awaiting while the lease TTL expires; TimeoutError is treated as a transient error alongside the SDK's TRANSIENT_ERRORS.[1] The _heartbeat_loop sets a stop event and returns when the control plane reports state == 'stopping' or 'stopped', the lease is not extended, a permanent (non-transient) heartbeat failure occurs, or transient failures persist longer than the lease TTL without a successful heartbeat — preventing two runners from serving the same work.[1]
agent_toolset (which pulls in host-only modules such as subprocess and tarfile) is imported lazily — never at module level — so EnvironmentWorker can be exposed on the generated work resource without dragging those imports into import anthropic.[1]
Example of running a long-lived EnvironmentWorker daemon and the single-item handle_item variant:
from anthropic import AsyncAnthropic
client = AsyncAnthropic()
# Long-running daemon: poll for work, serve each session, loop.
await client.beta.environments.work.worker(
environment_id=environment_id,
environment_key=environment_key,
workdir="/workspace",
).run()
# Already-claimed item (e.g. inside `ant worker poll --on-work ...`):
await client.beta.environments.work.worker(workdir="/workspace").handle_item()
Canonical EnvironmentWorker usage that adds a custom tool alongside the standard agent toolset:
await client.beta.environments.work.worker(
environment_id=os.environ["ANTHROPIC_ENVIRONMENT_ID"],
environment_key=os.environ["ANTHROPIC_ENVIRONMENT_KEY"],
workdir="/workspace",
tools=lambda env: [*beta_agent_toolset_20260401(env), deploy],
).run()
When invoked from an ant worker poll --on-work script (where credentials are already in the environment), calling handle_item() with no arguments is sufficient:
await client.beta.environments.work.worker(workdir="/workspace", tools=tools).handle_item()
When iterating the poller manually, pass all IDs and the key explicitly to handle_item:
await client.beta.environments.work.worker(workdir="/workspace", tools=tools).handle_item(
work_id=work.id,
environment_id=work.environment_id,
session_id=work.data.id,
environment_key=environment_key,
)
Sources