modal.sandbox

modal.sandbox.Sandbox

class Sandbox(modal.object.Object)

A Sandbox object lets you interact with a running sandbox. This API is similar to Python’s asyncio.subprocess.Process.

Refer to the guide on how to spawn and use sandboxes.

def __init__(self, *args, **kwargs):

create

@staticmethod
def create(
    *entrypoint_args: str,
    app: Optional["modal.app._App"] = None,  # Optionally associate the sandbox with an app
    environment_name: Optional[str] = None,  # Optionally override the default environment
    image: Optional[_Image] = None,  # The image to run as the container for the sandbox.
    mounts: Sequence[_Mount] = (),  # Mounts to attach to the sandbox.
    secrets: Sequence[_Secret] = (),  # Environment variables to inject into the sandbox.
    network_file_systems: Dict[Union[str, os.PathLike], _NetworkFileSystem] = {},
    timeout: Optional[int] = None,  # Maximum execution time of the sandbox in seconds.
    workdir: Optional[str] = None,  # Working directory of the sandbox.
    gpu: GPU_T = None,
    cloud: Optional[str] = None,
    region: Optional[Union[str, Sequence[str]]] = None,  # Region or regions to run the sandbox on.
    cpu: Optional[float] = None,  # How many CPU cores to request. This is a soft limit.
    # Specify, in MiB, a memory request which is the minimum memory required.
    # Or, pass (request, limit) to additionally specify a hard limit in MiB.
    memory: Optional[Union[int, Tuple[int, int]]] = None,
    block_network: bool = False,  # Whether to block network access
    # List of CIDRs the sandbox is allowed to access. If None, all CIDRs are allowed.
    cidr_allowlist: Optional[Sequence[str]] = None,
    volumes: Dict[
        Union[str, os.PathLike], Union[_Volume, _CloudBucketMount]
    ] = {},  # Mount points for Modal Volumes and CloudBucketMounts
    pty_info: Optional[api_pb2.PTYInfo] = None,
    # List of ports to tunnel into the sandbox. Encrypted ports are tunneled with TLS.
    encrypted_ports: Sequence[int] = [],
    # List of ports to tunnel into the sandbox without encryption.
    unencrypted_ports: Sequence[int] = [],
    _experimental_scheduler_placement: Optional[
        SchedulerPlacement
    ] = None,  # Experimental controls over fine-grained scheduling (alpha).
    client: Optional[_Client] = None,
) -> "_Sandbox":

from_id

@staticmethod
def from_id(sandbox_id: str, client: Optional[_Client] = None) -> "_Sandbox":

Construct a Sandbox from an id and look up the Sandbox result.

The ID of a Sandbox object can be accessed using .object_id.

set_tags

def set_tags(self, tags: Dict[str, str], *, client: Optional[_Client] = None):

Set tags (key-value pairs) on the Sandbox. Tags can be used to filter results in Sandbox.list.

wait

def wait(self, raise_on_termination: bool = True):

Wait for the Sandbox to finish running.

tunnels

def tunnels(self, timeout: int = 50) -> Dict[int, Tunnel]:

Get tunnel metadata for the sandbox.

Raises SandboxTimeoutError if the tunnels are not available after the timeout.

Returns a dictionary of Tunnel objects which are keyed by the container port.

NOTE: Previous to client v0.64.152, this returned a list of TunnelData objects.

terminate

def terminate(self):

Terminate Sandbox execution.

This is a no-op if the Sandbox has already finished running.

poll

def poll(self) -> Optional[int]:

Check if the Sandbox has finished running.

Returns None if the Sandbox is still running, else returns the exit code.

exec

def exec(
    self,
    *cmds: str,
    pty_info: Optional[api_pb2.PTYInfo] = None,  # Deprecated: internal use only
    stdout: StreamType = StreamType.PIPE,
    stderr: StreamType = StreamType.PIPE,
    timeout: Optional[int] = None,
    workdir: Optional[str] = None,
    secrets: Sequence[_Secret] = (),
    # Encode output as text.
    text: bool = True,
    # Control line-buffered output.
    # -1 means unbuffered, 1 means line-buffered (only available if `text=True`).
    bufsize: Literal[-1, 1] = -1,
    # Internal option to set terminal size and metadata
    _pty_info: Optional[api_pb2.PTYInfo] = None,
):

Execute a command in the Sandbox and return a ContainerProcess handle.

Usage

app = modal.App.lookup("my-app", create_if_missing=True)

sandbox = modal.Sandbox.create("sleep", "infinity", app=app)

process = sandbox.exec("bash", "-c", "for i in $(seq 1 10); do echo foo $i; sleep 0.5; done")

for line in process.stdout:
    print(line)

stdout

@property
def stdout(self) -> _StreamReader[str]:

StreamReader for the sandbox’s stdout stream.

stderr

@property
def stderr(self) -> _StreamReader[str]:

StreamReader for the sandbox’s stderr stream.

stdin

@property
def stdin(self) -> _StreamWriter:

StreamWriter for the sandbox’s stdin stream.

returncode

@property
def returncode(self) -> Optional[int]:

Return code of the sandbox process if it has finished running, else None.

list

@staticmethod
def list(
    *, app_id: Optional[str] = None, tags: Optional[Dict[str, str]] = None, client: Optional[_Client] = None
) -> AsyncGenerator["_Sandbox", None]:

List all sandboxes for the current environment or app ID (if specified). If tags are specified, only sandboxes that have at least those tags are returned. Returns an iterator over Sandbox objects.