Run a managed agent's tools locally with EnvironmentWorker
mainTo run tools for a managed agent in a self-hosted environment, use client.beta.environments.work.worker(...) (which returns an EnvironmentWorker). This worker polls the environment for work, sets up a workdir, downloads agent skills, and executes tools in response to agent.tool_use or agent.custom_tool_use events.
Key features:
- Polling: Use
.run()to start a loop that polls for work. - Single Item: Use
.handle_item(...)to process a specific work item already claimed. - Environment Variables: If no arguments are passed to
handle_item(), it readsANTHROPIC_WORK_ID,ANTHROPIC_ENVIRONMENT_ID,ANTHROPIC_SESSION_ID, andANTHROPIC_ENVIRONMENT_KEYfrom the environment. - Async Only: This is built on
anyioand works withasyncioortrio.
import os, asyncio
from anthropic import AsyncAnthropic
from anthropic.lib.tools import beta_async_tool
from anthropic.lib.tools.agent_toolset import beta_agent_toolset_20260401
client = AsyncAnthropic()
@beta_async_tool
async def deploy(target: str) -> str:
...
# Start a worker that loops forever
await client.beta.environments.work.worker(
environment_id=os.environ["ANTHROPIC_ENVIRONMENT_ID"],
environment_key=os.environ["ANTHROPIC_ENVIRONMENT_KEY"],
workdir="/workspace",
# Use a factory to bind the toolset to the session's AgentToolContext
tools=lambda env: [*beta_agent_toolset_20260401(env), deploy],
).run()