"""Asynchronous SDK entry points and management interfaces."""
from __future__ import annotations
import asyncio
from typing import Dict, List, Mapping, Optional
from pathlib import Path
from datetime import timedelta
from typing_extensions import Unpack
import httpx
from ._types import (
SDKAxonListParams,
BaseRequestOptions,
LongRequestOptions,
SDKAgentListParams,
SDKAxonCreateParams,
SDKDevboxListParams,
SDKObjectListParams,
SDKAgentCreateParams,
SDKDevboxCreateParams,
SDKObjectCreateParams,
SDKBlueprintListParams,
SDKMcpConfigListParams,
SDKAgentListPublicParams,
SDKBlueprintCreateParams,
SDKMcpConfigCreateParams,
SDKDiskSnapshotListParams,
SDKGatewayConfigListParams,
SDKNetworkPolicyListParams,
SDKGatewayConfigCreateParams,
SDKNetworkPolicyCreateParams,
SDKDevboxCreateFromImageParams,
)
from .._types import Timeout, NotGiven, not_given
from .._client import DEFAULT_MAX_RETRIES, AsyncRunloop
from ._helpers import detect_content_type
from .async_axon import AsyncAxon
from .._constants import DEFAULT_API_POOL_SHARDS, DEFAULT_TRANSFER_POOL_SHARDS, DEFAULT_BACKGROUND_POOL_SHARDS
from .async_agent import AsyncAgent
from .async_devbox import AsyncDevbox
from .async_secret import AsyncSecret
from .async_snapshot import AsyncSnapshot
from .async_blueprint import AsyncBlueprint
from .async_mcp_config import AsyncMcpConfig
from ..types.secret_view import SecretView
from .async_secret_by_id import AsyncSecretById
from ..lib.context_loader import TarFilter, build_directory_tar
from .async_gateway_config import AsyncGatewayConfig
from .async_network_policy import AsyncNetworkPolicy
from .async_storage_object import AsyncStorageObject
from ..types.object_create_params import ContentType
from ..types.agent_devbox_counts_view import AgentDevboxCountsView
from ..types.shared_params.agent_source import Git, Npm, Pip, Object
[docs]
class AsyncDevboxOps:
"""High-level async manager for creating and managing AsyncDevbox instances.
Accessed via ``runloop.devbox`` from :class:`AsyncRunloopSDK`, provides
coroutines to create devboxes from scratch, blueprints, or snapshots, and to
list existing devboxes.
Example:
>>> runloop = AsyncRunloopSDK()
>>> devbox = await runloop.devbox.create(name="my-devbox")
>>> devboxes = await runloop.devbox.list(limit=10)
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize the manager.
:param client: Generated AsyncRunloop client to wrap
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def create(
self,
**params: Unpack[SDKDevboxCreateParams],
) -> AsyncDevbox:
"""Provision a new devbox and wait until it reaches ``running`` state.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKDevboxCreateParams` for available parameters
:return: Wrapper bound to the newly created devbox
:rtype: AsyncDevbox
"""
devbox_view = await self._client.devboxes.create_and_await_running(
**params,
)
return AsyncDevbox(self._client, devbox_view.id)
[docs]
async def create_from_blueprint_id(
self,
blueprint_id: str,
**params: Unpack[SDKDevboxCreateFromImageParams],
) -> AsyncDevbox:
"""Create a devbox from an existing blueprint by identifier.
:param blueprint_id: Blueprint ID to create from
:type blueprint_id: str
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKDevboxCreateFromImageParams` for available parameters
:return: Wrapper bound to the newly created devbox
:rtype: AsyncDevbox
"""
devbox_view = await self._client.devboxes.create_and_await_running(
blueprint_id=blueprint_id,
**params,
)
return AsyncDevbox(self._client, devbox_view.id)
[docs]
async def create_from_blueprint_name(
self,
blueprint_name: str,
**params: Unpack[SDKDevboxCreateFromImageParams],
) -> AsyncDevbox:
"""Create a devbox from the latest blueprint with the given name.
:param blueprint_name: Blueprint name to create from
:type blueprint_name: str
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKDevboxCreateFromImageParams` for available parameters
:return: Wrapper bound to the newly created devbox
:rtype: AsyncDevbox
"""
devbox_view = await self._client.devboxes.create_and_await_running(
blueprint_name=blueprint_name,
**params,
)
return AsyncDevbox(self._client, devbox_view.id)
[docs]
async def create_from_snapshot(
self,
snapshot_id: str,
**params: Unpack[SDKDevboxCreateFromImageParams],
) -> AsyncDevbox:
"""Create a devbox initialized from a snapshot.
:param snapshot_id: Snapshot ID to create from
:type snapshot_id: str
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKDevboxCreateFromImageParams` for available parameters
:return: Wrapper bound to the newly created devbox
:rtype: AsyncDevbox
"""
devbox_view = await self._client.devboxes.create_and_await_running(
snapshot_id=snapshot_id,
**params,
)
return AsyncDevbox(self._client, devbox_view.id)
[docs]
def from_id(self, devbox_id: str) -> AsyncDevbox:
"""Attach to an existing devbox by ID.
Returns immediately without waiting for the devbox to reach ``running``
state. Call ``await_running()`` on the returned :class:`AsyncDevbox` if
you need to wait for readiness (contrast with the synchronous SDK, which blocks).
:param devbox_id: Existing devbox ID
:type devbox_id: str
:return: Wrapper bound to the requested devbox
:rtype: AsyncDevbox
"""
return AsyncDevbox(self._client, devbox_id)
[docs]
async def list(
self,
**params: Unpack[SDKDevboxListParams],
) -> list[AsyncDevbox]:
"""List devboxes accessible to the caller.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKDevboxListParams` for available parameters
:return: Collection of devbox wrappers
:rtype: list[AsyncDevbox]
"""
page = await self._client.devboxes.list(
**params,
)
return [AsyncDevbox(self._client, item.id) for item in page.devboxes]
[docs]
class AsyncSnapshotOps:
"""High-level async manager for working with disk snapshots.
Accessed via ``runloop.snapshot`` from :class:`AsyncRunloopSDK`, provides
coroutines to list snapshots and access snapshot details.
Example:
>>> runloop = AsyncRunloopSDK()
>>> snapshots = await runloop.snapshot.list(devbox_id="dev-123")
>>> snapshot = await runloop.snapshot.from_id("snap-123")
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize the manager.
:param client: Generated AsyncRunloop client to wrap
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def list(
self,
**params: Unpack[SDKDiskSnapshotListParams],
) -> list[AsyncSnapshot]:
"""List snapshots created from devboxes.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKDiskSnapshotListParams` for available parameters
:return: Snapshot wrappers for each record
:rtype: list[AsyncSnapshot]
"""
page = await self._client.devboxes.disk_snapshots.list(
**params,
)
return [AsyncSnapshot(self._client, item.id) for item in page.snapshots]
[docs]
def from_id(self, snapshot_id: str) -> AsyncSnapshot:
"""Return a snapshot wrapper for the given ID.
:param snapshot_id: Snapshot ID to wrap
:type snapshot_id: str
:return: Wrapper for the snapshot resource
:rtype: AsyncSnapshot
"""
return AsyncSnapshot(self._client, snapshot_id)
[docs]
class AsyncBlueprintOps:
"""High-level async manager for creating and managing blueprints.
Accessed via ``runloop.blueprint`` from :class:`AsyncRunloopSDK`, provides
coroutines to create Dockerfile-based blueprints, inspect build logs,
and list existing blueprints.
Example:
>>> runloop = AsyncRunloopSDK()
>>> blueprint = await runloop.blueprint.create(
... name="my-blueprint",
... dockerfile="FROM ubuntu:22.04\\nRUN apt-get update",
... )
>>> blueprints = await runloop.blueprint.list()
To use a local directory as a build context, use an object.
Example:
>>> from datetime import timedelta
>>> from runloop_api_client.types.blueprint_build_parameters import BuildContext
>>> runloop = AsyncRunloopSDK()
>>> obj = await runloop.object_storage.upload_from_dir(
... "./",
... ttl=timedelta(hours=1),
... )
>>> blueprint = await runloop.blueprint.create(
... name="my-blueprint",
... dockerfile="FROM ubuntu:22.04\\nCOPY . .\\n",
... build_context=BuildContext(type="object", object_id=obj.id),
... )
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize the manager.
:param client: Generated AsyncRunloop client to wrap
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def create(
self,
**params: Unpack[SDKBlueprintCreateParams],
) -> AsyncBlueprint:
"""Create a blueprint and wait for the build to finish.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKBlueprintCreateParams` for available parameters
:return: Wrapper bound to the finished blueprint
:rtype: AsyncBlueprint
"""
blueprint = await self._client.blueprints.create_and_await_build_complete(
**params,
)
return AsyncBlueprint(self._client, blueprint.id)
[docs]
def from_id(self, blueprint_id: str) -> AsyncBlueprint:
"""Return a blueprint wrapper for the given ID.
:param blueprint_id: Blueprint ID to wrap
:type blueprint_id: str
:return: Wrapper for the blueprint resource
:rtype: AsyncBlueprint
"""
return AsyncBlueprint(self._client, blueprint_id)
[docs]
async def list(
self,
**params: Unpack[SDKBlueprintListParams],
) -> list[AsyncBlueprint]:
"""List available blueprints.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKBlueprintListParams` for available parameters
:return: Blueprint wrappers for each record
:rtype: list[AsyncBlueprint]
"""
page = await self._client.blueprints.list(
**params,
)
return [AsyncBlueprint(self._client, item.id) for item in page.blueprints]
[docs]
class AsyncStorageObjectOps:
"""High-level async manager for creating and managing storage objects.
Accessed via ``runloop.storage_object`` from :class:`AsyncRunloopSDK`, provides
coroutines to create, upload, download, and list storage objects with convenient
helpers for file and text uploads.
Example:
>>> runloop = AsyncRunloopSDK()
>>> obj = await runloop.storage_object.upload_from_text("Hello!", "greeting.txt")
>>> content = await obj.download_as_text()
>>> objects = await runloop.storage_object.list()
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize the manager.
:param client: Generated AsyncRunloop client to wrap
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def create(
self,
**params: Unpack[SDKObjectCreateParams],
) -> AsyncStorageObject:
"""Create a storage object and obtain an upload URL.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKObjectCreateParams` for available parameters
:return: Wrapper with upload URL set for immediate uploads
:rtype: AsyncStorageObject
"""
obj = await self._client.objects.create(**params)
return AsyncStorageObject(self._client, obj.id, upload_url=obj.upload_url, upload_headers=obj.upload_headers)
[docs]
def from_id(self, object_id: str) -> AsyncStorageObject:
"""Return a storage object wrapper by identifier.
:param object_id: Storage object identifier to wrap
:type object_id: str
:return: Wrapper for the storage object resource
:rtype: AsyncStorageObject
"""
return AsyncStorageObject(self._client, object_id, upload_url=None)
[docs]
async def list(
self,
**params: Unpack[SDKObjectListParams],
) -> list[AsyncStorageObject]:
"""List storage objects owned by the caller.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKObjectListParams` for available parameters
:return: Storage object wrappers for each record
:rtype: list[AsyncStorageObject]
"""
page = await self._client.objects.list(
**params,
)
return [AsyncStorageObject(self._client, item.id, upload_url=item.upload_url) for item in page.objects]
[docs]
async def upload_from_file(
self,
file_path: str | Path,
*,
name: Optional[str] = None,
content_type: Optional[ContentType] = None,
metadata: Optional[Dict[str, str]] = None,
ttl: Optional[timedelta] = None,
**options: Unpack[LongRequestOptions],
) -> AsyncStorageObject:
"""Create and upload an object from a local file path.
:param file_path: Local filesystem path to read
:type file_path: str | Path
:param name: Optional object name; defaults to the file name
:type name: Optional[str]
:param content_type: Optional MIME type to apply to the object
:type content_type: Optional[ContentType]
:param metadata: Optional key-value metadata
:type metadata: Optional[Dict[str, str]]
:param ttl: Optional Time-To-Live, after which the object is automatically deleted
:type ttl: Optional[timedelta]
:param options: See :typeddict:`~runloop_api_client.sdk._types.LongRequestOptions` for available options
:return: Wrapper for the uploaded object
:rtype: AsyncStorageObject
:raises OSError: If the local file cannot be read
"""
path = Path(file_path)
try:
content = await asyncio.to_thread(lambda: path.read_bytes())
except OSError as error:
raise OSError(f"Failed to read file {path}: {error}") from error
name = name or path.name
content_type = content_type or detect_content_type(str(file_path))
ttl_ms = int(ttl.total_seconds()) * 1000 if ttl else None
obj = await self.create(name=name, content_type=content_type, metadata=metadata, ttl_ms=ttl_ms, **options)
await obj.upload_content(content)
await obj.complete()
return obj
[docs]
async def upload_from_dir(
self,
dir_path: str | Path,
*,
name: Optional[str] = None,
metadata: Optional[Dict[str, str]] = None,
ttl: Optional[timedelta] = None,
ignore: TarFilter | None = None,
**options: Unpack[LongRequestOptions],
) -> AsyncStorageObject:
"""Create and upload an object from a local directory.
The resulting object will be uploaded as a compressed tarball.
:param dir_path: Local filesystem directory path to tar
:type dir_path: str | Path
:param name: Optional object name; defaults to the directory name + '.tar.gz'
:type name: Optional[str]
:param metadata: Optional key-value metadata
:type metadata: Optional[Dict[str, str]]
:param ttl: Optional Time-To-Live, after which the object is automatically deleted
:type ttl: Optional[timedelta]
:param ignore: Optional tar filter function compatible with
:meth:`tarfile.TarFile.add`. If provided, it will be called for each
member to allow modification or exclusion (by returning ``None``).
:type ignore: Optional[TarFilter]
:param options: See :typeddict:`~runloop_api_client.sdk._types.LongRequestOptions`
for available options
:return: Wrapper for the uploaded object
:rtype: AsyncStorageObject
:raises OSError: If the local directory cannot be read
:raises ValueError: If ``dir_path`` does not point to a directory
"""
path = Path(dir_path)
if not path.is_dir():
raise ValueError(f"dir_path must be a directory, got: {path}")
name = name or f"{path.name}.tar.gz"
ttl_ms = int(ttl.total_seconds()) * 1000 if ttl else None
def synchronous_io() -> bytes:
return build_directory_tar(path, tar_filter=ignore)
tar_bytes = await asyncio.to_thread(synchronous_io)
obj = await self.create(name=name, content_type="tgz", metadata=metadata, ttl_ms=ttl_ms, **options)
await obj.upload_content(tar_bytes)
await obj.complete()
return obj
[docs]
async def upload_from_text(
self,
text: str,
*,
name: str,
metadata: Optional[Dict[str, str]] = None,
ttl: Optional[timedelta] = None,
**options: Unpack[LongRequestOptions],
) -> AsyncStorageObject:
"""Create and upload an object from a text payload.
:param text: Text content to upload
:type text: str
:param name: Object display name
:type name: str
:param metadata: Optional key-value metadata
:type metadata: Optional[Dict[str, str]]
:param ttl: Optional Time-To-Live, after which the object is automatically deleted
:type ttl: Optional[timedelta]
:param options: See :typeddict:`~runloop_api_client.sdk._types.LongRequestOptions` for available options
:return: Wrapper for the uploaded object
:rtype: AsyncStorageObject
"""
ttl_ms = int(ttl.total_seconds()) * 1000 if ttl else None
obj = await self.create(name=name, content_type="text", metadata=metadata, ttl_ms=ttl_ms, **options)
await obj.upload_content(text)
await obj.complete()
return obj
[docs]
async def upload_from_bytes(
self,
data: bytes,
*,
name: str,
content_type: ContentType,
metadata: Optional[Dict[str, str]] = None,
ttl: Optional[timedelta] = None,
**options: Unpack[LongRequestOptions],
) -> AsyncStorageObject:
"""Create and upload an object from a bytes payload.
:param data: Binary payload to upload
:type data: bytes
:param name: Object display name
:type name: str
:param content_type: MIME type describing the payload
:type content_type: ContentType
:param metadata: Optional key-value metadata
:type metadata: Optional[Dict[str, str]]
:param ttl: Optional Time-To-Live, after which the object is automatically deleted
:type ttl: Optional[timedelta]
:param options: See :typeddict:`~runloop_api_client.sdk._types.LongRequestOptions` for available options
:return: Wrapper for the uploaded object
:rtype: AsyncStorageObject
"""
ttl_ms = int(ttl.total_seconds()) * 1000 if ttl else None
obj = await self.create(name=name, content_type=content_type, metadata=metadata, ttl_ms=ttl_ms, **options)
await obj.upload_content(data)
await obj.complete()
return obj
[docs]
class AsyncAxonOps:
"""[Beta] Create and manage axons (async). Access via ``runloop.axon``.
Example:
>>> runloop = AsyncRunloopSDK()
>>> axon = await runloop.axon.create()
>>> await axon.publish(event_type="test", origin="USER_EVENT", payload="{}", source="sdk")
"""
def __init__(self, client: AsyncRunloop) -> None:
self._client = client
[docs]
async def create(self, **params: Unpack[SDKAxonCreateParams]) -> AsyncAxon:
"""[Beta] Create a new axon."""
response = await self._client.axons.create(**params)
return AsyncAxon(self._client, response.id)
[docs]
def from_id(self, axon_id: str) -> AsyncAxon:
"""Get an AsyncAxon instance for an existing axon ID."""
return AsyncAxon(self._client, axon_id)
[docs]
async def list(self, **params: Unpack[SDKAxonListParams]) -> list[AsyncAxon]:
"""[Beta] List all axons, including deleted axons.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKAxonListParams` for available parameters
:return: Collection of axon wrappers
:rtype: list[AsyncAxon]
"""
result = await self._client.axons.list(**params)
return [AsyncAxon(self._client, axon.id) for axon in result.axons]
[docs]
class AsyncAgentOps:
"""High-level async manager for creating and managing agents.
Accessed via ``runloop.agent`` from :class:`AsyncRunloopSDK`, provides
coroutines to create, retrieve, and list agents from various sources (npm, pip, git, object storage).
Example:
>>> runloop = AsyncRunloopSDK()
>>> # Create agent from NPM package
>>> agent = await runloop.agent.create_from_npm(name="my-agent", package_name="@runloop/example-agent")
>>> # Create agent from Git repository
>>> agent = await runloop.agent.create_from_git(
... name="git-agent", repository="https://github.com/user/agent-repo", ref="main"
... )
>>> # List all agents
>>> agents = await runloop.agent.list(limit=10)
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize the manager.
:param client: Generated AsyncRunloop client to wrap
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def create(
self,
**params: Unpack[SDKAgentCreateParams],
) -> AsyncAgent:
"""Create a new agent.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKAgentCreateParams` for available parameters
:return: Wrapper bound to the newly created agent
:rtype: AsyncAgent
"""
agent_view = await self._client.agents.create(
**params,
)
return AsyncAgent(self._client, agent_view.id, agent_view)
[docs]
async def create_from_npm(
self,
*,
package_name: str,
registry_url: Optional[str] = None,
agent_setup: Optional[list[str]] = None,
**params: Unpack[SDKAgentCreateParams],
) -> AsyncAgent:
"""Create an agent from an NPM package.
:param package_name: NPM package name
:type package_name: str
:param registry_url: NPM registry URL, defaults to None
:type registry_url: Optional[str], optional
:param agent_setup: Setup commands to run after installation, defaults to None
:type agent_setup: Optional[list[str]], optional
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKAgentCreateParams` for additional parameters (excluding 'source')
:return: Wrapper bound to the newly created agent
:rtype: AsyncAgent
:raises ValueError: If 'source' is provided in params
"""
if "source" in params:
raise ValueError(
"Cannot specify 'source' when using create_from_npm(); source is automatically set to npm configuration"
)
npm_config: Npm = {"package_name": package_name}
if registry_url is not None:
npm_config["registry_url"] = registry_url
if agent_setup is not None:
npm_config["agent_setup"] = agent_setup
params["source"] = {"type": "npm", "npm": npm_config}
return await self.create(**params)
[docs]
async def create_from_pip(
self,
*,
package_name: str,
registry_url: Optional[str] = None,
agent_setup: Optional[list[str]] = None,
**params: Unpack[SDKAgentCreateParams],
) -> AsyncAgent:
"""Create an agent from a Pip package.
:param package_name: Pip package name
:type package_name: str
:param registry_url: Pip registry URL, defaults to None
:type registry_url: Optional[str], optional
:param agent_setup: Setup commands to run after installation, defaults to None
:type agent_setup: Optional[list[str]], optional
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKAgentCreateParams` for additional parameters (excluding 'source')
:return: Wrapper bound to the newly created agent
:rtype: AsyncAgent
:raises ValueError: If 'source' is provided in params
"""
if "source" in params:
raise ValueError(
"Cannot specify 'source' when using create_from_pip(); source is automatically set to pip configuration"
)
pip_config: Pip = {"package_name": package_name}
if registry_url is not None:
pip_config["registry_url"] = registry_url
if agent_setup is not None:
pip_config["agent_setup"] = agent_setup
params["source"] = {"type": "pip", "pip": pip_config}
return await self.create(**params)
[docs]
async def create_from_git(
self,
*,
repository: str,
ref: Optional[str] = None,
agent_setup: Optional[list[str]] = None,
**params: Unpack[SDKAgentCreateParams],
) -> AsyncAgent:
"""Create an agent from a Git repository.
:param repository: Git repository URL
:type repository: str
:param ref: Optional Git ref (branch/tag/commit), defaults to main/HEAD
:type ref: Optional[str], optional
:param agent_setup: Setup commands to run after cloning, defaults to None
:type agent_setup: Optional[list[str]], optional
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKAgentCreateParams` for additional parameters (excluding 'source')
:return: Wrapper bound to the newly created agent
:rtype: AsyncAgent
:raises ValueError: If 'source' is provided in params
"""
if "source" in params:
raise ValueError(
"Cannot specify 'source' when using create_from_git(); source is automatically set to git configuration"
)
git_config: Git = {"repository": repository}
if ref is not None:
git_config["ref"] = ref
if agent_setup is not None:
git_config["agent_setup"] = agent_setup
params["source"] = {"type": "git", "git": git_config}
return await self.create(**params)
[docs]
async def create_from_object(
self,
*,
object_id: str,
agent_setup: Optional[list[str]] = None,
**params: Unpack[SDKAgentCreateParams],
) -> AsyncAgent:
"""Create an agent from a storage object.
:param object_id: Storage object ID
:type object_id: str
:param agent_setup: Setup commands to run after unpacking, defaults to None
:type agent_setup: Optional[list[str]], optional
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKAgentCreateParams` for additional parameters (excluding 'source')
:return: Wrapper bound to the newly created agent
:rtype: AsyncAgent
:raises ValueError: If 'source' is provided in params
"""
if "source" in params:
raise ValueError(
"Cannot specify 'source' when using create_from_object(); source is automatically set to object configuration"
)
object_config: Object = {"object_id": object_id}
if agent_setup is not None:
object_config["agent_setup"] = agent_setup
params["source"] = {"type": "object", "object": object_config}
return await self.create(**params)
[docs]
def from_id(self, agent_id: str) -> AsyncAgent:
"""Attach to an existing agent by ID.
:param agent_id: Existing agent ID
:type agent_id: str
:return: Wrapper bound to the requested agent
:rtype: AsyncAgent
"""
return AsyncAgent(self._client, agent_id)
[docs]
async def list(
self,
**params: Unpack[SDKAgentListParams],
) -> list[AsyncAgent]:
"""List agents accessible to the caller.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKAgentListParams` for available parameters
:return: Collection of agent wrappers
:rtype: list[AsyncAgent]
"""
page = await self._client.agents.list(
**params,
)
return [AsyncAgent(self._client, item.id, item) for item in page.agents]
[docs]
async def list_public(
self,
**params: Unpack[SDKAgentListPublicParams],
) -> List[AsyncAgent]:
"""List public agents.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKAgentListPublicParams` for available parameters
:return: Collection of public agent wrappers
:rtype: list[AsyncAgent]
"""
page = await self._client.agents.list_public(
**params,
)
return [AsyncAgent(self._client, item.id, item) for item in page.agents]
[docs]
async def devbox_counts(
self,
**options: Unpack[BaseRequestOptions],
) -> AgentDevboxCountsView:
"""Get devbox counts grouped by agent name.
:param options: Optional request configuration
:return: Devbox counts per agent name and total
:rtype: AgentDevboxCountsView
"""
return await self._client.agents.devbox_counts(
**options,
)
[docs]
class AsyncNetworkPolicyOps:
"""High-level async manager for creating and managing network policies.
Accessed via ``runloop.network_policy`` from :class:`AsyncRunloopSDK`, provides
coroutines to create, retrieve, update, delete, and list network policies.
Example:
>>> runloop = AsyncRunloopSDK()
>>> policy = await runloop.network_policy.create(
... name="my-policy",
... allowed_hostnames=["github.com", "*.npmjs.org"],
... )
>>> policies = await runloop.network_policy.list()
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize AsyncNetworkPolicyOps.
:param client: AsyncRunloop client instance
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def create(self, **params: Unpack[SDKNetworkPolicyCreateParams]) -> AsyncNetworkPolicy:
"""Create a new network policy.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKNetworkPolicyCreateParams` for available parameters
:return: The newly created network policy
:rtype: AsyncNetworkPolicy
"""
response = await self._client.network_policies.create(**params)
return AsyncNetworkPolicy(self._client, response.id)
[docs]
def from_id(self, network_policy_id: str) -> AsyncNetworkPolicy:
"""Get an AsyncNetworkPolicy instance for an existing network policy ID.
:param network_policy_id: ID of the network policy
:type network_policy_id: str
:return: AsyncNetworkPolicy instance for the given ID
:rtype: AsyncNetworkPolicy
"""
return AsyncNetworkPolicy(self._client, network_policy_id)
[docs]
async def list(self, **params: Unpack[SDKNetworkPolicyListParams]) -> list[AsyncNetworkPolicy]:
"""List all network policies, optionally filtered by parameters.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKNetworkPolicyListParams` for available parameters
:return: List of network policies
:rtype: list[AsyncNetworkPolicy]
"""
page = await self._client.network_policies.list(**params)
return [AsyncNetworkPolicy(self._client, item.id) for item in page.network_policies]
[docs]
class AsyncGatewayConfigOps:
"""High-level async manager for creating and managing gateway configurations.
Accessed via ``runloop.gateway_config`` from :class:`AsyncRunloopSDK`, provides
coroutines to create, retrieve, update, delete, and list gateway configs. Gateway configs
define how to proxy API requests through the credential gateway, enabling secure API
proxying without exposing API keys.
Example:
>>> runloop = AsyncRunloopSDK()
>>> gateway_config = await runloop.gateway_config.create(
... name="my-api-gateway",
... endpoint="https://api.example.com",
... auth_mechanism={"type": "bearer"},
... )
>>> # Use with a devbox
>>> devbox = await runloop.devbox.create(
... name="my-devbox",
... gateways={
... "MY_API": {
... "gateway": gateway_config.id,
... "secret": "my-api-key-secret",
... },
... },
... )
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize AsyncGatewayConfigOps.
:param client: AsyncRunloop client instance
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def create(self, **params: Unpack[SDKGatewayConfigCreateParams]) -> AsyncGatewayConfig:
"""Create a new gateway config.
Example:
>>> gateway_config = await runloop.gateway_config.create(
... name="my-gateway",
... endpoint="https://api.example.com",
... auth_mechanism={"type": "header", "key": "x-api-key"},
... description="Gateway for My API",
... )
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKGatewayConfigCreateParams` for available parameters
:return: The newly created gateway config
:rtype: AsyncGatewayConfig
"""
response = await self._client.gateway_configs.create(**params)
return AsyncGatewayConfig(self._client, response.id)
[docs]
def from_id(self, gateway_config_id: str) -> AsyncGatewayConfig:
"""Get an AsyncGatewayConfig instance for an existing gateway config ID.
Example:
>>> gateway_config = runloop.gateway_config.from_id("gwc_1234567890")
>>> info = await gateway_config.get_info()
>>> print(f"Gateway Config name: {info.name}")
:param gateway_config_id: ID of the gateway config
:type gateway_config_id: str
:return: AsyncGatewayConfig instance for the given ID
:rtype: AsyncGatewayConfig
"""
return AsyncGatewayConfig(self._client, gateway_config_id)
[docs]
async def list(self, **params: Unpack[SDKGatewayConfigListParams]) -> list[AsyncGatewayConfig]:
"""List all gateway configs, optionally filtered by parameters.
Example:
>>> configs = await runloop.gateway_config.list(limit=10)
>>> for config in configs:
... print(config.id)
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKGatewayConfigListParams` for available parameters
:return: List of gateway configs
:rtype: list[AsyncGatewayConfig]
"""
page = await self._client.gateway_configs.list(**params)
return [AsyncGatewayConfig(self._client, item.id) for item in page.gateway_configs]
[docs]
class AsyncMcpConfigOps:
"""High-level async manager for creating and managing MCP configurations.
Accessed via ``runloop.mcp_config`` from :class:`AsyncRunloopSDK`, provides methods
to create, retrieve, update, delete, and list MCP configs. MCP configs define
how to connect to upstream MCP (Model Context Protocol) servers, specifying the
target endpoint and which tools are allowed.
Example:
>>> runloop = AsyncRunloopSDK()
>>> mcp_config = await runloop.mcp_config.create(
... name="my-mcp-server",
... endpoint="https://mcp.example.com",
... allowed_tools=["*"],
... )
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize AsyncMcpConfigOps.
:param client: AsyncRunloop client instance
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def create(self, **params: Unpack[SDKMcpConfigCreateParams]) -> AsyncMcpConfig:
"""Create a new MCP config.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKMcpConfigCreateParams` for available parameters
:return: The newly created MCP config
:rtype: AsyncMcpConfig
"""
response = await self._client.mcp_configs.create(**params)
return AsyncMcpConfig(self._client, response.id)
[docs]
def from_id(self, mcp_config_id: str) -> AsyncMcpConfig:
"""Get an AsyncMcpConfig instance for an existing MCP config ID.
:param mcp_config_id: ID of the MCP config
:type mcp_config_id: str
:return: AsyncMcpConfig instance for the given ID
:rtype: AsyncMcpConfig
"""
return AsyncMcpConfig(self._client, mcp_config_id)
[docs]
async def list(self, **params: Unpack[SDKMcpConfigListParams]) -> list[AsyncMcpConfig]:
"""List all MCP configs, optionally filtered by parameters.
:param params: See :typeddict:`~runloop_api_client.sdk._types.SDKMcpConfigListParams` for available parameters
:return: List of MCP configs
:rtype: list[AsyncMcpConfig]
"""
page = await self._client.mcp_configs.list(**params)
return [AsyncMcpConfig(self._client, item.id) for item in page.mcp_configs]
[docs]
class AsyncSecretOps:
"""High-level async manager for creating and managing Secrets.
Accessed via ``runloop.secret`` from :class:`AsyncRunloopSDK`, provides methods to
create, retrieve, update, list, and delete secrets. Secrets are encrypted
key-value pairs that can be injected into Devboxes as environment variables.
Example:
>>> runloop = AsyncRunloopSDK()
>>> secret = await runloop.secret.create(name="MY_API_KEY", value="secret-value")
>>> secrets = await runloop.secret.list()
"""
def __init__(self, client: AsyncRunloop) -> None:
"""Initialize AsyncSecretOps.
:param client: AsyncRunloop client instance
:type client: AsyncRunloop
"""
self._client = client
[docs]
async def create(self, name: str, value: str, **options: Unpack[LongRequestOptions]) -> AsyncSecret:
"""Create a new secret.
Example:
>>> secret = await runloop.secret.create(
... name="DATABASE_PASSWORD",
... value="my-secure-password",
... )
>>> print(f"Created secret: {secret.name}")
:param name: Account-scoped secret name (must be a valid env var name)
:type name: str
:param value: Secret value to store (encrypted at rest)
:type value: str
:param options: Optional request configuration
:return: The created AsyncSecret instance
:rtype: AsyncSecret
"""
view = await self._client.secrets.create(name=name, value=value, **options)
return AsyncSecret(self._client, view.name, view.id)
[docs]
def from_name(self, name: str) -> AsyncSecret:
"""Get an AsyncSecret instance by name without making an API call.
Use ``get_info()`` on the returned AsyncSecret to fetch the actual data.
Example:
>>> secret = runloop.secret.from_name("MY_API_KEY")
>>> info = await secret.get_info()
>>> print(f"Secret ID: {info.id}")
:param name: The literal account-scoped name of the secret
:type name: str
:return: An AsyncSecret instance (no API call made)
:rtype: AsyncSecret
"""
return AsyncSecret(self._client, name)
[docs]
def from_id(self, id: str) -> AsyncSecretById:
"""Get a lazy exact-ID handle; its operations never fall back to a name."""
return AsyncSecretById(self._client, id)
[docs]
async def update(
self,
secret: "AsyncSecret | str",
value: str,
**options: Unpack[LongRequestOptions],
) -> AsyncSecret:
"""Update an existing secret's value.
Example:
>>> updated = await runloop.secret.update("DATABASE_PASSWORD", "new-password")
>>> # Or using an AsyncSecret object
>>> updated = await runloop.secret.update(secret, "new-password")
:param secret: The secret to update (AsyncSecret object or name string)
:type secret: AsyncSecret | str
:param value: The new secret value
:type value: str
:param options: Optional request configuration
:return: The updated AsyncSecret instance
:rtype: AsyncSecret
"""
name = secret if isinstance(secret, str) else secret.name
await self._client.secrets.update(name, value=value, **options)
return AsyncSecret(self._client, name)
[docs]
async def list(self, **options: Unpack[BaseRequestOptions]) -> list[AsyncSecret]:
"""List all secrets.
Example:
>>> secrets = await runloop.secret.list()
>>> for s in secrets:
... print(s.name)
:param options: Optional request configuration
:return: List of AsyncSecret instances
:rtype: list[AsyncSecret]
"""
result = await self._client.secrets.list(**options)
return [AsyncSecret(self._client, view.name, view.id) for view in result.secrets]
[docs]
async def delete(
self,
secret: "AsyncSecret | str",
**options: Unpack[LongRequestOptions],
) -> SecretView:
"""Delete a secret. This action is irreversible.
Example:
>>> await runloop.secret.delete("DATABASE_PASSWORD")
>>> # Or using an AsyncSecret object
>>> await runloop.secret.delete(secret)
:param secret: The secret to delete (AsyncSecret object or name string)
:type secret: AsyncSecret | str
:param options: Optional request configuration
:return: The deleted secret metadata
:rtype: SecretView
"""
name = secret if isinstance(secret, str) else secret.name
return await self._client.secrets.delete(name, **options)
class AsyncRunloopSDK:
"""High-level asynchronous entry point for the Runloop SDK.
Provides a Pythonic, object-oriented interface for managing devboxes,
blueprints, snapshots, and storage objects. Exposes the generated async REST
client via the ``api`` attribute for advanced use cases.
:ivar api: Direct access to the generated async REST API client
:vartype api: AsyncRunloop
:ivar agent: High-level async interface for agent management.
:vartype agent: AsyncAgentOps
:ivar axon: [Beta] High-level async interface for axon management
:vartype axon: AsyncAxonOps
:ivar devbox: High-level async interface for devbox management
:vartype devbox: AsyncDevboxOps
:ivar blueprint: High-level async interface for blueprint management
:vartype blueprint: AsyncBlueprintOps
:ivar snapshot: High-level async interface for snapshot management
:vartype snapshot: AsyncSnapshotOps
:ivar storage_object: High-level async interface for storage object management
:vartype storage_object: AsyncStorageObjectOps
:ivar network_policy: High-level async interface for network policy management
:vartype network_policy: AsyncNetworkPolicyOps
:ivar gateway_config: High-level async interface for gateway config management
:vartype gateway_config: AsyncGatewayConfigOps
:ivar mcp_config: High-level async interface for MCP config management
:vartype mcp_config: AsyncMcpConfigOps
:ivar secret: High-level async interface for secret management
:vartype secret: AsyncSecretOps
Example:
>>> runloop = AsyncRunloopSDK() # Uses RUNLOOP_API_KEY env var
>>> devbox = await runloop.devbox.create(name="my-devbox")
>>> result = await devbox.cmd.exec("echo 'hello'")
>>> print(await result.stdout())
>>> await devbox.shutdown()
"""
api: AsyncRunloop
agent: AsyncAgentOps
axon: AsyncAxonOps
devbox: AsyncDevboxOps
blueprint: AsyncBlueprintOps
gateway_config: AsyncGatewayConfigOps
mcp_config: AsyncMcpConfigOps
network_policy: AsyncNetworkPolicyOps
secret: AsyncSecretOps
snapshot: AsyncSnapshotOps
storage_object: AsyncStorageObjectOps
def __init__(
self,
*,
bearer_token: str | None = None,
base_url: str | httpx.URL | None = None,
timeout: float | Timeout | None | NotGiven = not_given,
max_retries: int = DEFAULT_MAX_RETRIES,
default_headers: Mapping[str, str] | None = None,
default_query: Mapping[str, object] | None = None,
http_client: httpx.AsyncClient | None = None,
api_pool_shards: int = DEFAULT_API_POOL_SHARDS,
background_pool_shards: int = DEFAULT_BACKGROUND_POOL_SHARDS,
transfer_pool_shards: int = DEFAULT_TRANSFER_POOL_SHARDS,
) -> None:
"""Configure the asynchronous SDK wrapper.
:param bearer_token: API token; falls back to ``RUNLOOP_API_KEY`` env var, defaults to None
:type bearer_token: str | None, optional
:param base_url: Override the API base URL, defaults to None
:type base_url: str | httpx.URL | None, optional
:param timeout: Request timeout (seconds) or ``Timeout`` object, defaults to not_given
:type timeout: float | Timeout | None | NotGiven, optional
:param max_retries: Maximum automatic retry attempts, defaults to DEFAULT_MAX_RETRIES
:type max_retries: int, optional
:param default_headers: Headers merged into every request, defaults to None
:type default_headers: Mapping[str, str] | None, optional
:param default_query: Default query parameters merged into every request, defaults to None
:type default_query: Mapping[str, object] | None, optional
:param http_client: Custom ``httpx.AsyncClient`` instance to reuse, defaults to None
:type http_client: httpx.AsyncClient | None, optional
:param api_pool_shards: H2 shards for short RPCs (round-robin), defaults to 8
:type api_pool_shards: int, optional
:param background_pool_shards: H2 shards for long-polls (round-robin), defaults to 16
:type background_pool_shards: int, optional
:param transfer_pool_shards: H2 shards for upload/download (round-robin), defaults to 2
:type transfer_pool_shards: int, optional
"""
self.api = AsyncRunloop(
bearer_token=bearer_token,
base_url=base_url,
timeout=timeout,
max_retries=max_retries,
default_headers=default_headers,
default_query=default_query,
http_client=http_client,
api_pool_shards=api_pool_shards,
background_pool_shards=background_pool_shards,
transfer_pool_shards=transfer_pool_shards,
)
self.agent = AsyncAgentOps(self.api)
self.axon = AsyncAxonOps(self.api)
self.devbox = AsyncDevboxOps(self.api)
self.blueprint = AsyncBlueprintOps(self.api)
self.gateway_config = AsyncGatewayConfigOps(self.api)
self.mcp_config = AsyncMcpConfigOps(self.api)
self.network_policy = AsyncNetworkPolicyOps(self.api)
self.secret = AsyncSecretOps(self.api)
self.snapshot = AsyncSnapshotOps(self.api)
self.storage_object = AsyncStorageObjectOps(self.api)
[docs]
async def aclose(self) -> None:
"""Close the underlying HTTP client and release resources."""
await self.api.close()
async def __aenter__(self) -> "AsyncRunloopSDK":
"""Allow ``async with AsyncRunloopSDK() as runloop`` usage.
:return: The active SDK instance
:rtype: AsyncRunloopSDK
"""
return self
async def __aexit__(self, *_exc_info: object) -> None:
"""Ensure the API client closes when leaving the context manager."""
await self.aclose()