Skip to main content

client.agent_client

Agent client for agent-runtimes.

Extends the account-only datalayer_core.DatalayerClient with runtime creation, environments, snapshots, Ray, evals, code execution, and local/remote agent lifecycle helpers — migrated out of datalayer-core.

LocalAgentRuntime Objects​

@dataclass
class LocalAgentRuntime()

Handle to a running local agent-runtimes server.

chat_endpoint​

@property
def chat_endpoint() -> str

Vercel AI chat endpoint for this runtime's agent.

terminate​

def terminate() -> None

Terminate the underlying server process (if any).

find_free_port​

def find_free_port(host: str = DEFAULT_LOCAL_HOST) -> int

Return a free TCP port bound on host.

build_agent_runtime_env​

def build_agent_runtime_env() -> tuple[dict[str, str], list[str]]

Build subprocess env with Bedrock-to-AWS variable mappings.

wait_for_local_runtime​

def wait_for_local_runtime(base_url: str, timeout_seconds: int = 25) -> None

Block until the local runtime /health endpoint responds.

start_local_agent_runtime​

def start_local_agent_runtime(
        *,
        agent_spec_id: str,
        agent_name: str = DEFAULT_LOCAL_AGENT_NAME,
        host: str = DEFAULT_LOCAL_HOST,
        port: Optional[int] = None,
        protocol: str = DEFAULT_LOCAL_PROTOCOL,
        log_level: str = DEFAULT_LOCAL_LOG_LEVEL,
        wait: bool = True,
        disable_tool_approvals: bool = False) -> LocalAgentRuntime

Launch a local agent-runtimes server as a subprocess.

terminate_local_agent_runtime​

def terminate_local_agent_runtime(runtime: LocalAgentRuntime) -> None

Terminate a local runtime process, escalating to kill if needed.

ensure_local_agent​

def ensure_local_agent(*,
                       base_url: str,
                       agent_name: str,
                       token: str,
                       agent_spec_id: str,
                       agent_library: str = "pydantic-ai",
                       transport: str = DEFAULT_LOCAL_PROTOCOL,
                       enable_skills: bool = True,
                       description: Optional[str] = None,
                       timeout: int = 120,
                       disable_tool_approvals: bool = False) -> None

Ensure a local agent with the expected transport is registered.

delete_local_agents​

def delete_local_agents(*, base_url: str, token: str) -> tuple[int, int]

Delete all locally-registered agents.

delete_local_agent​

def delete_local_agent(*, base_url: str, token: str, agent_name: str) -> bool

Delete a single locally-registered agent by id or name.

extract_vercel_stream_text​

def extract_vercel_stream_text(raw: str) -> str

Extract concatenated text deltas from a Vercel AI SSE stream.

extract_vercel_stream_usage​

def extract_vercel_stream_usage(raw: str) -> dict[str, Any]

Extract best-effort pydantic usage metadata from a Vercel AI SSE stream.

run_local_agent_chat​

def run_local_agent_chat(*,
                         base_url: str,
                         agent_name: str,
                         token: str,
                         prompt: str,
                         timeout: int = 300) -> dict[str, Any]

Send a single prompt to a local agent via the Vercel AI endpoint.

build_agent_runtimes_base_url​

def build_agent_runtimes_base_url(ingress: str) -> str

Derive cloud agent-runtimes base URL from a runtime ingress.

runtime_route_candidates​

def runtime_route_candidates(*,
                             agent_name: Optional[str] = None,
                             agent_spec_id: Optional[str] = None,
                             pod_name: Optional[str] = None) -> list[str]

Build ordered and de-duplicated Vercel AI route candidates.

run_cloud_agent_chat​

def run_cloud_agent_chat(*,
                         ingress: str,
                         token: str,
                         prompt: str,
                         route_candidates: list[str],
                         timeout: int = 300) -> dict[str, Any]

Send a single prompt to a cloud runtime agent via Vercel AI.

AgentClient Objects​

class AgentClient(_BaseDatalayerClient, RuntimesMixin, EnvironmentsMixin,
                  EvalsMixin, EventsMixin, RayMixin, SandboxSnapshotsMixin)

Datalayer client with runtime creation, code execution, and agent lifecycle.

Supports both remote (cloud runtime) and local (agent-runtimes server) agent execution via the start_local_agent_runtime, ensure_local_agent, run_local_agent_chat, and run_cloud_agent_chat helpers exposed as methods.

list_environments​

@lru_cache
def list_environments() -> list[EnvironmentModel]

List all available environments.

Returns​

list[Environment] A list of available environments.

create_runtime​

def create_runtime(name: Optional[str] = None,
                   environment: str = DEFAULT_ENVIRONMENT,
                   time_reservation: Minutes = DEFAULT_TIME_RESERVATION,
                   snapshot_name: Optional[str] = None,
                   agent_spec_id: Optional[str] = None,
                   agent_spec: Optional[dict[str, Any]] = None,
                   billing_entity_uid: Optional[str] = None,
                   billing_entity_type: Optional[str] = None,
                   billing_entity_handle: Optional[str] = None,
                   api_key: Optional[str] = None) -> RuntimeService

Create a new runtime (kernel) for code execution.

Parameters​

name : str, optional Name of the runtime to create. environment : str, optional Environment type (e.g., "ai-agents-env"). Type of resources needed (cpu, gpu, etc.). time_reservation : Minutes, optional Time reservation in minutes for the runtime. Defaults to 10 minutes. snapshot_name : Optional[str], optional Name of the snapshot to create from. If provided, the runtime will be created from this snapshot.

Returns​

Runtime A runtime object for code execution.

list_runtimes​

def list_runtimes() -> list[RuntimeService]

List all running runtimes.

Returns​

list[Runtime] List of Runtime objects representing active runtimes.

terminate_runtime​

def terminate_runtime(runtime: Union[RuntimeService, str],
                      api_key: Optional[str] = None) -> bool

Terminate a running Runtime.

Parameters​

runtime : Union[Runtime, str] Runtime object or pod name string to terminate.

Returns​

bool True if termination was successful, False otherwise.

get_runtime​

def get_runtime(runtime: Union[RuntimeService, str]) -> RuntimeService

Get a single running Runtime by pod name.

Parameters​

runtime : Union[Runtime, str] Runtime object or pod name string to fetch.

Returns​

Runtime The Runtime object matching the pod name.

Raises​

RuntimeError If the runtime cannot be retrieved.

update_runtime​

def update_runtime(runtime: Union[RuntimeService, str],
                   capabilities: list[str]) -> bool

Update a running Runtime's capabilities.

Parameters​

runtime : Union[Runtime, str] Runtime object or pod name string to update. capabilities : list[str] New capabilities to apply to the runtime.

Returns​

bool True if the update succeeded.

Raises​

RuntimeError If the update fails.

check_runtime_health​

def check_runtime_health(
        runtime: Union[RuntimeService, str],
        probe_code: str = "print('datalayer runtime health probe')",
        timeout: float = 20.0,
        api_key: Optional[str] = None) -> dict[str, Any]

Check runtime reachability and execute a probe on the sandbox.

Parameters​

runtime : Union[RuntimeService, str] Runtime object or runtime identifier (pod name/uid/name). probe_code : str Python code to execute as health probe on the sandbox. timeout : float Probe execution timeout in seconds. api_key : Optional[str] Optional API key override used for runtime lookup.

Returns​

dict[str, Any] Health result with success flag and diagnostics.

create_snapshot​

def create_snapshot(runtime: Optional["RuntimeService"] = None,
                    pod_name: Optional[str] = None,
                    name: Optional[str] = None,
                    description: Optional[str] = None,
                    stop: bool = True) -> "SandboxSnapshotModel"

Create a snapshot of the current runtime state.

Parameters​

runtime : Optional[Runtime] The runtime object to create a snapshot from. pod_name : Optional[str] The pod name of the runtime. name : Optional[str] Name for the new snapshot. description : Optional[str] Description for the new snapshot. stop : bool Whether to stop the runtime after creating snapshot.

Returns​

SandboxSnapshotModel The created snapshot object.

list_snapshots​

def list_snapshots() -> list[SandboxSnapshotModel]

List all snapshots.

Returns​

list[SandboxSnapshotModel] A list of snapshots associated with the user.

delete_snapshot​

def delete_snapshot(
        snapshot: Union[str, SandboxSnapshotModel]) -> dict[str, str]

Delete a specific snapshot.

Parameters​

snapshot : Union[str, SandboxSnapshotModel] Snapshot object or UID string to delete.

Returns​

dict[str, str] The result of the deletion operation.

start_local_agent_runtime​

def start_local_agent_runtime(
        *,
        agent_spec_id: str,
        agent_name: str = DEFAULT_LOCAL_AGENT_NAME,
        host: str = "127.0.0.1",
        port: Optional[int] = None,
        protocol: str = "vercel-ai",
        log_level: str = "info",
        wait: bool = True,
        disable_tool_approvals: bool = False) -> LocalAgentRuntime

Launch a local agent-runtimes server as a subprocess.

Parameters​

agent_spec_id : str Agentspec id to boot the runtime with. agent_name : str Registered agent name/id served by the runtime. host : str Host interface to bind to. port : Optional[int] Port to bind to. A free port is selected when omitted. protocol : str Transport protocol exposed by the runtime (e.g. vercel-ai). log_level : str Log level for the runtime process. wait : bool Whether to block until the runtime reports healthy. disable_tool_approvals : bool Whether to disable tool approvals on the launched runtime.

Returns​

LocalAgentRuntime Handle pointing at the running server.

ensure_local_agent​

def ensure_local_agent(*,
                       base_url: str,
                       agent_name: str,
                       agent_spec_id: str,
                       token: Optional[str] = None,
                       transport: str = "vercel-ai",
                       enable_skills: bool = True,
                       description: Optional[str] = None,
                       timeout: int = 120,
                       disable_tool_approvals: bool = False) -> None

Ensure a local agent with the expected transport is registered.

Parameters​

base_url : str Local agent-runtimes base URL. agent_name : str Agent name/id to register. agent_spec_id : str Agentspec id backing the agent. token : Optional[str] Bearer token; falls back to this client's API key when omitted. transport : str Transport protocol to register (e.g. vercel-ai). enable_skills : bool Whether to enable skills for the registered agent. description : Optional[str] Optional description for the agent. timeout : int Registration request timeout in seconds. disable_tool_approvals : bool Whether to disable tool approvals for the agent.

delete_local_agent​

def delete_local_agent(*,
                       base_url: str,
                       agent_name: str,
                       token: Optional[str] = None) -> bool

Delete a single locally-registered agent by id or name.

Parameters​

base_url : str Local agent-runtimes base URL. agent_name : str Agent id or name to delete. token : Optional[str] Bearer token; falls back to this client's API key when omitted.

Returns​

bool True when a matching agent was found and delete accepted.

delete_local_agents​

def delete_local_agents(*,
                        base_url: str,
                        token: Optional[str] = None) -> tuple[int, int]

Delete all locally-registered agents.

Parameters​

base_url : str Local agent-runtimes base URL. token : Optional[str] Bearer token; falls back to this client's API key when omitted.

Returns​

tuple[int, int] (total_agents, deleted_agents).

run_local_agent_chat​

def run_local_agent_chat(*,
                         base_url: str,
                         agent_name: str,
                         prompt: str,
                         token: Optional[str] = None,
                         timeout: int = 300) -> dict[str, Any]

Send a single prompt to a local agent via the Vercel AI endpoint.

Parameters​

base_url : str Local agent-runtimes base URL. agent_name : str Registered agent name/id to target. prompt : str Prompt to send. token : Optional[str] Bearer token; falls back to this client's API key when omitted. timeout : int Per-request timeout in seconds.

Returns​

dict[str, Any] Structured chat result (status/output/failure_cause).

run_cloud_agent_chat​

def run_cloud_agent_chat(*,
                         ingress: str,
                         prompt: str,
                         route_candidates: Optional[list[str]] = None,
                         agent_name: Optional[str] = None,
                         agent_spec_id: Optional[str] = None,
                         pod_name: Optional[str] = None,
                         token: Optional[str] = None,
                         timeout: int = 300) -> dict[str, Any]

Send a single prompt to a cloud runtime agent via Vercel AI.

Parameters​

ingress : str Runtime ingress URL. prompt : str Prompt to send. route_candidates : Optional[list[str]] Explicit ordered route candidates. When omitted they are derived from agent_name/agent_spec_id/pod_name. agent_name : Optional[str] Agent name used to derive route candidates. agent_spec_id : Optional[str] Agentspec id used to derive route candidates. pod_name : Optional[str] Runtime pod name used to derive route candidates. token : Optional[str] Bearer token; falls back to this client's API key when omitted. timeout : int Per-request timeout in seconds.

Returns​

dict[str, Any] Structured chat result (status/output/failure_cause).