Skip to content
microvms-agentd

AgentVm

View as MarkdownOpen in ChatGPTOpen in ClaudeOpen in Claude CodeOpen in Cursor
@final
class AgentVm

One VM with coding agents in it: the sandbox plus the specs it is built for.

The sequence is the CLI’s agent-up and agent-prompt, one method per step: find_image or build_artifact + your upload + build_image; launch; install_access; prompt or prompt_sync; terminate. sandbox and session reach the same VM for suspend, resume, cp-shaped transfers, and any other exec. Each step that calls AWS or the VM has an awaitable _async twin that drives the same future.

Constructor

Section titled “Constructor”
AgentVm(region: Region, agents: SupportsGetItem[int, AgentSpec] | None = None)

Resolves credentials for region and returns a VM with nothing built or launched.

agents defaults to Claude Code alone. An empty list or a repeated agent is refused by the core before any AWS call.

Properties

Section titled “Properties”

agents

Section titled “agents”
agents: list[AgentSpec]

The specs this VM carries, in profile order.

region

Section titled “region”
region: Region

sandbox

Section titled “sandbox”
sandbox: Sandbox

The sandbox this VM drives, for suspend, resume, and the lifecycle getters. The same lock: a call here and a call there cannot interleave.

session

Section titled “session”
session: Session | None

The session, once launch has run.

Methods

Section titled “Methods”

adopt

Section titled “adopt”
@staticmethod
def adopt(
    region: Region,
    microvm_id: str,
    endpoint: str,
    agent_token: str,
    agents: SupportsGetItem[int, AgentSpec] | None = None,
    *,
    port: SupportsIndex | None = None,
) -> AgentVm

An agent VM for a VM another process launched; see Sandbox.adopt.

agents states what the VM carries and defaults to Claude Code alone; read installed_agents(session) first when the adopting process does not know.

adopt_async

Section titled “adopt_async”
@staticmethod
def adopt_async(
    region: Region,
    microvm_id: str,
    endpoint: str,
    agent_token: str,
    agents: SupportsGetItem[int, AgentSpec] | None = None,
    *,
    port: SupportsIndex | None = None,
) -> AgentVm

The awaitable twin of AgentVm.adopt.

build_artifact

Section titled “build_artifact”
def build_artifact(
    *,
    binary: SupportsGetItem[int, SupportsIndex],
    build_role_arn: str,
    size: SizeClass | None = None,
) -> bytes

The artifact bytes to upload to s3://<bucket>/<image_name>.zip before build_image.

build_image

Section titled “build_image”
def build_image(
    *,
    binary: SupportsGetItem[int, SupportsIndex],
    code_artifact_uri: str,
    build_role_arn: str,
    size: SizeClass | None = None,
) -> Image

Builds the image and waits for it to become usable. code_artifact_uri is where you uploaded build_artifact’s bytes. Several minutes, server-side.

build_image_async

Section titled “build_image_async”
def build_image_async(
    *,
    binary: SupportsGetItem[int, SupportsIndex],
    code_artifact_uri: str,
    build_role_arn: str,
    size: SizeClass | None = None,
) -> Image

The awaitable twin of build_image. A lifecycle transition: cancelling the awaitable leaves the build running to completion.

create_async

Section titled “create_async”
@staticmethod
def create_async(
    region: Region,
    agents: SupportsGetItem[int, AgentSpec] | None = None,
) -> AgentVm

The awaitable twin of AgentVm(region, agents).

detach

Section titled “detach”
def detach() -> Detached

Hands the VM off to another process; see Sandbox.detach. The adopter passes the same agents to AgentVm.adopt.

dockerfile

Section titled “dockerfile”
def dockerfile() -> str

The Dockerfile build_image will send: the client’s agentd stanza plus the agent layers. Read it to see what the image will contain; nothing in it is a secret.

ensure_image

Section titled “ensure_image”
def ensure_image(
    *,
    binary: SupportsGetItem[int, SupportsIndex],
    build_role_arn: str,
    s3_bucket: str,
    size: SizeClass | None = None,
    s3_key_prefix: str | None = None,
) -> EnsuredImage

Builds or reuses this VM’s image, named per image_name: returned at once when ready, waited on while building, deleted and rebuilt when failed, and uploaded to s3://<s3_bucket>/<s3_key_prefix>/<name>/artifact.zip only when a build is needed.

ensure_image_async

Section titled “ensure_image_async”
def ensure_image_async(
    *,
    binary: SupportsGetItem[int, SupportsIndex],
    build_role_arn: str,
    s3_bucket: str,
    size: SizeClass | None = None,
    s3_key_prefix: str | None = None,
) -> EnsuredImage

The awaitable twin of ensure_image. Cancellation stops the local wait and releases this sandbox without recording an image. A build already accepted by the service keeps running; ensure the same inputs again to join it.

find_image

Section titled “find_image”
def find_image(
    *,
    binary: SupportsGetItem[int, SupportsIndex],
    build_role_arn: str,
    size: SizeClass | None = None,
) -> str | None

The ARN of an existing image named per image_name, or None when there is none.

find_image_async

Section titled “find_image_async”
def find_image_async(
    *,
    binary: SupportsGetItem[int, SupportsIndex],
    build_role_arn: str,
    size: SizeClass | None = None,
) -> str | None

The awaitable twin of find_image.

from_name

Section titled “from_name”
@staticmethod
def from_name(
    region: Region,
    name: str,
    registry: NameRegistry,
    agents: SupportsGetItem[int, AgentSpec] | None = None,
    *,
    port: SupportsIndex | None = None,
) -> AgentVm

Adopts the agent VM registered as name in registry; see Sandbox.from_name.

from_name_async

Section titled “from_name_async”
@staticmethod
def from_name_async(
    region: Region,
    name: str,
    registry: NameRegistry,
    agents: SupportsGetItem[int, AgentSpec] | None = None,
    *,
    port: SupportsIndex | None = None,
) -> AgentVm

The awaitable twin of AgentVm.from_name.

image_name

Section titled “image_name”
def image_name(
    *,
    binary: SupportsGetItem[int, SupportsIndex],
    build_role_arn: str,
    size: SizeClass | None = None,
) -> str

The image name for these specs and this daemon binary: agent-vm-<agents>-<hash12>.

Content-addressed, so an unchanged binary, spec set and size name the image a previous run built; find_image looks it up, and ensure_image builds or reuses it.

install_access

Section titled “install_access”
def install_access(
    *,
    token: BearerToken | None = None,
    ttl_seconds: SupportsFloat | SupportsIndex | None = None,
) -> BearerToken

Mints a token (or takes yours) and installs Bedrock access for this VM’s agents.

Returns the token used, so expires_at says when to call this again. Re-runnable on a running VM: that call is the credential refresh.

install_access_async

Section titled “install_access_async”
def install_access_async(
    *,
    token: BearerToken | None = None,
    ttl_seconds: SupportsFloat | SupportsIndex | None = None,
) -> BearerToken

The awaitable twin of install_access.

launch

Section titled “launch”
def launch(
    *,
    image_identifier: str,
    execution_role_arn: str | None = None,
    agent_token: str | None = None,
    client_token: str | None = None,
    max_idle_sec: SupportsIndex | None = None,
    suspended_sec: SupportsIndex | None = None,
    auto_resume: bool = False,
    max_duration_sec: SupportsIndex | None = None,
    image_version: str | None = None,
    egress_network_connectors: SupportsGetItem[int, str] | None = None,
    log_group: str | None = None,
    log_stream: str | None = None,
    disable_logging: bool = False,
    launch_env: dict[str, str] | None = None,
    shell: bool = False,
    ready_timeout: SupportsFloat | SupportsIndex | None = None,
) -> Session

Launches with egress and waits for the daemon to answer.

Egress is not optional: neither agent reaches Bedrock without it. The managed internet connector is the default; egress_network_connectors replaces it with customer-managed VPC connectors, whose VPC must route to Bedrock. The idle knobs default to the core’s figures (ten-minute idle and suspended windows, a one-hour ceiling); a multi-hour session raises max_duration_sec and polls health from outside to stay awake.

image_identifier is an image ARN or a bare image name; the core resolves a name with one ListMicrovmImages read and raises PreconditionError for a name no image carries.

launch_env, shell and ready_timeout mean what they mean on Sandbox.run: ready_timeout bounds the wait for RUNNING, and the wait for the daemon after it is session_constants()["defaultReadyTimeoutSeconds"].

launch_async

Section titled “launch_async”
def launch_async(
    *,
    image_identifier: str,
    execution_role_arn: str | None = None,
    agent_token: str | None = None,
    client_token: str | None = None,
    max_idle_sec: SupportsIndex | None = None,
    suspended_sec: SupportsIndex | None = None,
    auto_resume: bool = False,
    max_duration_sec: SupportsIndex | None = None,
    image_version: str | None = None,
    egress_network_connectors: SupportsGetItem[int, str] | None = None,
    log_group: str | None = None,
    log_stream: str | None = None,
    disable_logging: bool = False,
    launch_env: dict[str, str] | None = None,
    shell: bool = False,
    ready_timeout: SupportsFloat | SupportsIndex | None = None,
) -> Session

The awaitable twin of launch, with its keywords. A lifecycle transition: cancelling the awaitable leaves the launch running to completion, so the VM it starts is still this object’s to terminate.

prompt

Section titled “prompt”
def prompt(
    agent: str,
    task: str,
    *,
    timeout_sec: SupportsFloat | SupportsIndex | None = None,
    exec_id: str | None = None,
    permission_mode: str = 'agent-default',
    reap_group_on_exit: bool = False,
) -> ExecHandle

Starts one task for agent and returns its handle. Does not wait.

prompt_async

Section titled “prompt_async”
def prompt_async(
    agent: str,
    task: str,
    *,
    timeout_sec: SupportsFloat | SupportsIndex | None = None,
    exec_id: str | None = None,
    permission_mode: str = 'agent-default',
    reap_group_on_exit: bool = False,
) -> ExecHandle

The awaitable twin of prompt. Cancelling it before the daemon answers may leave the task started; pass an exec_id to reach it with session.exec(exec_id).

prompt_sync

Section titled “prompt_sync”
def prompt_sync(
    agent: str,
    task: str,
    *,
    timeout: SupportsFloat | SupportsIndex = ...,
    exec_id: str | None = None,
    permission_mode: str = 'agent-default',
    reap_group_on_exit: bool = False,
) -> ExecResult

Start, wait, ack: one task’s whole result. timeout defaults to 900 seconds, because agent tasks run minutes, and is also the daemon-side budget.

prompt_sync_async

Section titled “prompt_sync_async”
def prompt_sync_async(
    agent: str,
    task: str,
    *,
    timeout: SupportsFloat | SupportsIndex = ...,
    exec_id: str | None = None,
    permission_mode: str = 'agent-default',
    reap_group_on_exit: bool = False,
) -> ExecResult

The awaitable twin of prompt_sync. Cancelling it stops the wait; the task keeps running in the VM until its own timeout.

terminate

Section titled “terminate”
def terminate(
    *,
    delete_image: bool = False,
    delete_log_group: bool = False,
    delete_attempts: SupportsIndex | None = None,
    delete_backoff: SupportsFloat | SupportsIndex | None = None,
    wait_for_terminated: bool | SupportsFloat | SupportsIndex = ...,
) -> TeardownReport

Tears down, best-effort, never raising; see Sandbox.terminate.

terminate_async

Section titled “terminate_async”
def terminate_async(
    *,
    delete_image: bool = False,
    delete_log_group: bool = False,
    delete_attempts: SupportsIndex | None = None,
    delete_backoff: SupportsFloat | SupportsIndex | None = None,
    wait_for_terminated: bool | SupportsFloat | SupportsIndex = ...,
) -> TeardownReport

The awaitable twin of terminate; see Sandbox.terminate_async.

Special methods

Section titled “Special methods”

__aenter__

Section titled “__aenter__”
def __aenter__() -> AgentVm

async with AgentVm(...): returns the VM itself.

__aexit__

Section titled “__aexit__”
def __aexit__(
    exc_type: Any | None = None,
    exc_value: Any | None = None,
    traceback: Any | None = None,
) -> bool

__exit__’s teardown, awaited; it runs to completion even when the task awaiting it is cancelled.

__enter__

Section titled “__enter__”
def __enter__() -> AgentVm

__exit__

Section titled “__exit__”
def __exit__(
    exc_type: Any | None = None,
    exc_value: Any | None = None,
    traceback: Any | None = None,
) -> bool

Tears down on the way out, whatever happened inside the block. Returns False, so an exception raised inside the block propagates.

__repr__

Section titled “__repr__”
def __repr__() -> str

Provenance

Section titled “Provenance”

This page is generated from bindings/microvms-py/microvms.pyi by Griffe 2.3.0, run from site/scripts/gen-reference.mjs on every pnpm run sync, so an edit made here is overwritten by the next run.

To change the page, change the doc comment in the binding’s Rust source, then regenerate the declarations with mise run stubs. mise run stubs:check fails when the committed file no longer matches the source.