Skip to main content

utils.agent_utils

Cloud agent runtime provisioning helpers.

Reusable logic for launching cloud agent-runtimes from a :class:~datalayer_core.client.client.DatalayerClient. Shared by the eval examples and the GitHub Actions integration so credit/time-reservation math, environment burning-rate lookup, and create_runtime error handling are not duplicated across consumers.

resolve_environment_burning_rate​

def resolve_environment_burning_rate(client: Any,
                                     environment_name: str) -> float

Return the positive burning rate for an environment.

Parameters​

client : DatalayerClient An authenticated client able to list environments. environment_name : str The environment to look up.

Returns​

float The environment's positive burning rate.

Raises​

RuntimeError If the environment cannot be listed, is not found, or has no positive burning rate.

compute_time_reservation_minutes​

def compute_time_reservation_minutes(*, credits_limit: float,
                                     burning_rate: float) -> int

Compute a time reservation (minutes) from a credits budget.

create_runtime charges burning_rate * 60 * time_reservation credits, so this returns the smallest whole-minute reservation whose cost is at least credits_limit (minimum 1 minute).

Raises​

ValueError If burning_rate is not positive.

create_cloud_agent_runtime​

def create_cloud_agent_runtime(
        client: Any,
        *,
        environment_name: str,
        name: Optional[str] = None,
        agent_spec_id: Optional[str] = None,
        agent_spec: Optional[dict[str, Any]] = None,
        credits_limit: Optional[float] = None,
        time_reservation: Optional[int] = None,
        billing_entity_uid: Optional[str] = None,
        billing_entity_type: Optional[str] = None,
        billing_entity_handle: Optional[str] = None) -> Any

Create a cloud agent runtime via the core client.

Either time_reservation (in minutes) or credits_limit must be provided. When only credits_limit is given, the time reservation is derived from the environment's burning rate.

Parameters​

client : DatalayerClient An authenticated client. environment_name : str The runtime environment to launch in. name : Optional[str] Optional runtime name. agent_spec_id : Optional[str] Registered agentspec id (ignored when agent_spec is provided). agent_spec : Optional[dict[str, Any]] Inline agentspec payload (takes precedence over agent_spec_id). credits_limit : Optional[float] Target credits budget used to derive time_reservation when the latter is not supplied. time_reservation : Optional[int] Explicit time reservation in minutes. billing_entity_uid : Optional[str] Optional billing entity UID used for runtime billing attribution. billing_entity_type : Optional[str] Optional billing entity type (user, organization, team). billing_entity_handle : Optional[str] Optional billing entity handle.

Returns​

Any The created runtime object (exposes pod_name and ingress).

Raises​

ValueError If neither time_reservation nor credits_limit is provided. RuntimeError If runtime creation fails or returns no pod_name.

terminate_cloud_agent_runtime​

def terminate_cloud_agent_runtime(client: Any,
                                  runtime_or_pod_name: Any,
                                  *,
                                  raise_on_error: bool = False) -> bool

Terminate a cloud runtime created for agent execution.

Parameters​

client : DatalayerClient An authenticated client exposing terminate_runtime. runtime_or_pod_name : Any Runtime object (with pod_name) or raw pod-name string. raise_on_error : bool When True, raise :class:RuntimeError if termination fails.

Returns​

bool True when the runtime was terminated, otherwise False.

teardown_agent_execution_resources​

def teardown_agent_execution_resources(
        client: Any,
        *,
        execution_target: str,
        cloud_runtime_or_pod_name: Any = None,
        local_base_url: Optional[str] = None,
        local_agent_name: Optional[str] = None,
        token: Optional[str] = None,
        local_runtime: Any = None) -> dict[str, bool]

Teardown resources used by agent execution.

Handles both cloud and local cleanup using a single API so consumers (examples, GitHub Actions) don't duplicate teardown logic.