Skip to content

AsyncRuntime

Controls one runtime through asynchronous lifecycle and I/O operations.

Async · Sync counterpart: Runtime

class AsyncRuntime:

This type is created by the SDK; do not construct it directly.

runtime = await client.runtimes.create(...)
runtime = await client.runtimes.get(...)
NameType or valueDescription
cached_infoRuntimeInfoReturn the most recently fetched runtime metadata.
egressAsyncEgressManagerReturn the egress policy manager for this runtime.
secretsAsyncRuntimeSecretManagerReturn the secret-stub manager for this runtime.
filesAsyncFileManagerReturn the file manager for this runtime.
checkpointsAsyncRuntimeCheckpointManagerReturn the checkpoint manager for this runtime.
idstrReturn the runtime UUID.
namestrReturn the runtime display name.
heartbeatstatusAlias for status().

Refresh and return runtime metadata from the API.

async def refresh_info() -> RuntimeInfo
  • RuntimeInfo

Start the runtime and optionally wait until it is ready.

async def start(*, wait: bool = True, timeout: float = 240.0, poll_interval: float = 0.5) -> None
NameTypeDefaultDescription
waitboolTrueWait for the operation to reach its target state.
timeoutfloat240.0Maximum operation time in seconds.
poll_intervalfloat0.5Seconds between status checks.

Pause the runtime.

async def pause() -> None

Resume the runtime and optionally wait until it is ready.

async def resume(*, wait: bool = True, timeout: float = 240.0, poll_interval: float = 0.5) -> None
NameTypeDefaultDescription
waitboolTrueWait for the operation to reach its target state.
timeoutfloat240.0Maximum operation time in seconds.
poll_intervalfloat0.5Seconds between status checks.

Shut down the runtime.

async def stop() -> None

Delete the runtime.

async def delete() -> None

Update the runtime resource configuration.

async def update_resources(resources: PatchRuntimeResources | Mapping[str, object]) -> RuntimeInfo
NameTypeDefaultDescription
resourcesPatchRuntimeResources | Mapping[str, object]Structured resource requests and limits.
  • RuntimeInfo

Change the runtime memory limit in MiB.

async def resize_memory(new_memory_mib: int) -> RuntimeInfo
NameTypeDefaultDescription
new_memory_mibintNew runtime memory limit in MiB.
  • RuntimeInfo

Fetch the current runtime lifecycle status.

async def status() -> RuntimeStatus
  • RuntimeStatus

Wait until the runtime reaches the running state.

async def wait_until_ready(*, timeout: float = 240.0, poll_interval: float = 0.5) -> RuntimeInfo
NameTypeDefaultDescription
timeoutfloat240.0Maximum operation time in seconds.
poll_intervalfloat0.5Seconds between status checks.
  • RuntimeInfo

Run a command and return its buffered output.

async def exec(command: str | Sequence[str], timeout: float | None = None, env: Mapping[str, str] | None = None, stdin: bytes | str | None = None, cwd: str | None = None, max_output_bytes: int | None = None) -> CommandResult
NameTypeDefaultDescription
commandstr | Sequence[str]Shell command or argument sequence to execute.
timeoutfloat | NoneNoneOptional execution timeout in seconds.
envMapping[str, str] | NoneNoneEnvironment variables added to the command.
stdinbytes | str | NoneNoneBytes or text written to standard input.
cwdstr | NoneNoneWorking directory used for the command.
max_output_bytesint | NoneNoneMaximum buffered output size.
  • CommandResult — The command exit code and buffered standard output and error.
  • ApiError — If command execution cannot be started or completed.
>>> result = await runtime.exec("python --version")
>>> print(result.stdout_text)
Python 3.12.0

Run a command in a background worker and return a future.

def exec_detached(command: str | Sequence[str], timeout: float | None = None, env: Mapping[str, str] | None = None, stdin: bytes | str | None = None, cwd: str | None = None, max_output_bytes: int | None = None) -> asyncio.Task[CommandResult]
NameTypeDefaultDescription
commandstr | Sequence[str]Shell command or argument sequence to execute.
timeoutfloat | NoneNoneMaximum operation time in seconds.
envMapping[str, str] | NoneNoneEnvironment variables added to the command.
stdinbytes | str | NoneNoneBytes or text written to standard input.
cwdstr | NoneNoneWorking directory used for the command.
max_output_bytesint | NoneNoneMaximum buffered command output size.
  • asyncio.Task[CommandResult]

Create a runtime from one of this runtime’s checkpoints.

async def restore(checkpoint_id: str, *, name: str | None = None, **options: Any) -> AsyncRuntime
NameTypeDefaultDescription
checkpoint_idstrCheckpoint UUID used by the operation.
namestr | NoneNoneOptional resource display name.
optionsAny{}Additional runtime creation options.
  • AsyncRuntime