Function main

examples/managed-agents-worker-dispatch.py:48–69

()

async def main() -> None:
    logging.basicConfig(level=logging.INFO)
    client = AsyncAnthropic()

    # No agent/session creation and no polling here — an upstream poller already
    # claimed the item. Build the worker with just a `tools` factory: it is
    # invoked once per claimed session with that session's `AgentToolContext`,
    # so the built-in toolset binds to the right per-session workdir.
    # (`worker(...)` returns an `EnvironmentWorker`; you can also construct one
    # directly with `EnvironmentWorker(client, ...)` from `anthropic.lib.environments`.)
    worker = client.beta.environments.work.worker(
        workdir="/workspace",
        tools=lambda env: [*beta_agent_toolset_20260401(env), current_time],
    )

    # handle_item() with no arguments reads ANTHROPIC_WORK_ID /
    # ANTHROPIC_ENVIRONMENT_ID / ANTHROPIC_SESSION_ID / ANTHROPIC_ENVIRONMENT_KEY
    # from the environment, then runs the per-item flow: set up the workdir +
    # skills, run the session's tool calls while heartbeating the lease, and
    # force-stop the work item on exit. It returns once the item is done.
    await worker.handle_item()
    print("work item complete")


if __name__ == "__main__":