Skip to content

sandbox.md

Latest commit

 

History

History
2592 lines (1598 loc) · 59.3 KB

sandbox.md

File metadata and controls

2592 lines (1598 loc) · 59.3 KB

koyeb/sandbox

Koyeb Sandbox - Interactive execution environment for running arbitrary code on Koyeb

koyeb/sandbox.exec

Command execution utilities for Koyeb Sandbox instances Using SandboxClient HTTP API

CommandStatus Objects

class CommandStatus(str, Enum)

Command execution status

CommandResult Objects

@dataclass
class CommandResult()

Result of a command execution using Koyeb API models

success

@property
def success() -> bool

Check if command executed successfully

output

@property
def output() -> str

Get combined stdout and stderr output

SandboxCommandError Objects

class SandboxCommandError(SandboxError)

Raised when command execution fails

SandboxExecutor Objects

class SandboxExecutor()

Synchronous command execution interface for Koyeb Sandbox instances. Bound to a specific sandbox instance.

For async usage, use AsyncSandboxExecutor instead.

__call__

def __call__(
        command: str,
        cwd: Optional[str] = None,
        env: Optional[Dict[str, str]] = None,
        timeout: int = 30,
        on_stdout: Optional[Callable[[str], None]] = None,
        on_stderr: Optional[Callable[[str], None]] = None) -> CommandResult

Execute a command in a shell synchronously. Supports streaming output via callbacks.

Arguments:

  • command - Command to execute as a string (e.g., "python -c 'print(2+2)'")
  • cwd - Working directory for the command
  • env - Environment variables for the command
  • timeout - Command timeout in seconds (enforced for HTTP requests)
  • on_stdout - Optional callback for streaming stdout chunks
  • on_stderr - Optional callback for streaming stderr chunks

Returns:

  • CommandResult - Result of the command execution

Example:

```python
# Synchronous execution
result = sandbox.exec("echo hello")

# With streaming callbacks
result = sandbox.exec(
    "echo hello; sleep 1; echo world",
    on_stdout=lambda data: print(f"OUT: {data}"),
    on_stderr=lambda data: print(f"ERR: {data}"),
)
```

AsyncSandboxExecutor Objects

class AsyncSandboxExecutor(SandboxExecutor)

Async command execution interface for Koyeb Sandbox instances. Bound to a specific sandbox instance.

Inherits from SandboxExecutor and provides async command execution.

__call__

async def __call__(
        command: str,
        cwd: Optional[str] = None,
        env: Optional[Dict[str, str]] = None,
        timeout: int = 30,
        on_stdout: Optional[Callable[[str], None]] = None,
        on_stderr: Optional[Callable[[str], None]] = None) -> CommandResult

Execute a command in a shell asynchronously. Supports streaming output via callbacks.

Arguments:

  • command - Command to execute as a string (e.g., "python -c 'print(2+2)'")
  • cwd - Working directory for the command
  • env - Environment variables for the command
  • timeout - Command timeout in seconds (enforced for HTTP requests)
  • on_stdout - Optional callback for streaming stdout chunks
  • on_stderr - Optional callback for streaming stderr chunks

Returns:

  • CommandResult - Result of the command execution

Example:

```python
# Async execution
result = await sandbox.exec("echo hello")

# With streaming callbacks
result = await sandbox.exec(
    "echo hello; sleep 1; echo world",
    on_stdout=lambda data: print(f"OUT: {data}"),
    on_stderr=lambda data: print(f"ERR: {data}"),
)
```

koyeb/sandbox.filesystem

Filesystem operations for Koyeb Sandbox instances Using SandboxClient HTTP API

SandboxFilesystemError Objects

class SandboxFilesystemError(SandboxError)

Base exception for filesystem operations

SandboxFileNotFoundError Objects

class SandboxFileNotFoundError(SandboxFilesystemError)

Raised when file or directory not found

SandboxFileExistsError Objects

class SandboxFileExistsError(SandboxFilesystemError)

Raised when file already exists

FileInfo Objects

@dataclass
class FileInfo()

File information

SandboxFilesystem Objects

class SandboxFilesystem()

Synchronous filesystem operations for Koyeb Sandbox instances. Using SandboxClient HTTP API.

For async usage, use AsyncSandboxFilesystem instead.

write_file

def write_file(path: str,
               content: Union[str, bytes],
               encoding: str = "utf-8") -> None

Write content to a file synchronously.

Arguments:

  • path - Absolute path to the file
  • content - Content to write (string or bytes)
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary data.

read_file

def read_file(path: str, encoding: str = "utf-8") -> FileInfo

Read a file from the sandbox synchronously.

Arguments:

  • path - Absolute path to the file
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary data, which will decode the base64 content and return bytes.

Returns:

  • FileInfo - Object with content (str or bytes if base64) and encoding

mkdir

def mkdir(path: str) -> None

Create a directory synchronously.

Note: Parent directories are always created automatically by the API.

Arguments:

  • path - Absolute path to the directory

list_dir

def list_dir(path: str = ".") -> List[str]

List contents of a directory synchronously.

Arguments:

  • path - Path to the directory (default: current directory)

Returns:

  • List[str] - Names of files and directories within the specified path.

delete_file

def delete_file(path: str) -> None

Delete a file synchronously.

Arguments:

  • path - Absolute path to the file

delete_dir

def delete_dir(path: str) -> None

Delete a directory synchronously.

Arguments:

  • path - Absolute path to the directory

rename_file

def rename_file(old_path: str, new_path: str) -> None

Rename a file synchronously.

Arguments:

  • old_path - Current file path
  • new_path - New file path

move_file

def move_file(source_path: str, destination_path: str) -> None

Move a file to a different directory synchronously.

Arguments:

  • source_path - Current file path
  • destination_path - Destination path

write_files

def write_files(files: List[Dict[str, str]]) -> None

Write multiple files in a single operation synchronously.

Arguments:

  • files - List of dictionaries, each with 'path', 'content', and optional 'encoding'.

exists

def exists(path: str) -> bool

Check if file/directory exists synchronously

is_file

def is_file(path: str) -> bool

Check if path is a file synchronously

is_dir

def is_dir(path: str) -> bool

Check if path is a directory synchronously

upload_file

def upload_file(local_path: str,
                remote_path: str,
                encoding: str = "utf-8") -> None

Upload a local file to the sandbox synchronously.

Arguments:

  • local_path - Path to the local file
  • remote_path - Destination path in the sandbox
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary files.

Raises:

  • SandboxFileNotFoundError - If local file doesn't exist
  • UnicodeDecodeError - If file cannot be decoded with specified encoding

download_file

def download_file(remote_path: str,
                  local_path: str,
                  encoding: str = "utf-8") -> None

Download a file from the sandbox to a local path synchronously.

Arguments:

  • remote_path - Path to the file in the sandbox
  • local_path - Destination path on the local filesystem
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary files.

Raises:

  • SandboxFileNotFoundError - If remote file doesn't exist

ls

def ls(path: str = ".") -> List[str]

List directory contents synchronously.

Arguments:

  • path - Path to list

Returns:

List of file/directory names

rm

def rm(path: str, recursive: bool = False) -> None

Remove file or directory synchronously.

Arguments:

  • path - Path to remove
  • recursive - Remove recursively

open

def open(path: str, mode: str = "r", encoding: str = "utf-8") -> SandboxFileIO

Open a file in the sandbox synchronously.

Arguments:

  • path - Path to the file
  • mode - Open mode ('r', 'w', 'a', etc.)
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary data.

Returns:

  • SandboxFileIO - File handle

AsyncSandboxFilesystem Objects

class AsyncSandboxFilesystem(SandboxFilesystem)

Async filesystem operations for Koyeb Sandbox instances. Inherits from SandboxFilesystem and provides async methods.

write_file

@async_wrapper("write_file")
async def write_file(path: str,
                     content: Union[str, bytes],
                     encoding: str = "utf-8") -> None

Write content to a file asynchronously.

Arguments:

  • path - Absolute path to the file
  • content - Content to write (string or bytes)
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary data.

read_file

@async_wrapper("read_file")
async def read_file(path: str, encoding: str = "utf-8") -> FileInfo

Read a file from the sandbox asynchronously.

Arguments:

  • path - Absolute path to the file
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary data, which will decode the base64 content and return bytes.

Returns:

  • FileInfo - Object with content (str or bytes if base64) and encoding

mkdir

@async_wrapper("mkdir")
async def mkdir(path: str) -> None

Create a directory asynchronously.

Note: Parent directories are always created automatically by the API.

Arguments:

  • path - Absolute path to the directory

list_dir

@async_wrapper("list_dir")
async def list_dir(path: str = ".") -> List[str]

List contents of a directory asynchronously.

Arguments:

  • path - Path to the directory (default: current directory)

Returns:

  • List[str] - Names of files and directories within the specified path.

delete_file

@async_wrapper("delete_file")
async def delete_file(path: str) -> None

Delete a file asynchronously.

Arguments:

  • path - Absolute path to the file

delete_dir

@async_wrapper("delete_dir")
async def delete_dir(path: str) -> None

Delete a directory asynchronously.

Arguments:

  • path - Absolute path to the directory

rename_file

@async_wrapper("rename_file")
async def rename_file(old_path: str, new_path: str) -> None

Rename a file asynchronously.

Arguments:

  • old_path - Current file path
  • new_path - New file path

move_file

@async_wrapper("move_file")
async def move_file(source_path: str, destination_path: str) -> None

Move a file to a different directory asynchronously.

Arguments:

  • source_path - Current file path
  • destination_path - Destination path

write_files

async def write_files(files: List[Dict[str, str]]) -> None

Write multiple files in a single operation asynchronously.

Arguments:

  • files - List of dictionaries, each with 'path', 'content', and optional 'encoding'.

exists

@async_wrapper("exists")
async def exists(path: str) -> bool

Check if file/directory exists asynchronously

is_file

@async_wrapper("is_file")
async def is_file(path: str) -> bool

Check if path is a file asynchronously

is_dir

@async_wrapper("is_dir")
async def is_dir(path: str) -> bool

Check if path is a directory asynchronously

upload_file

@async_wrapper("upload_file")
async def upload_file(local_path: str,
                      remote_path: str,
                      encoding: str = "utf-8") -> None

Upload a local file to the sandbox asynchronously.

Arguments:

  • local_path - Path to the local file
  • remote_path - Destination path in the sandbox
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary files.

download_file

@async_wrapper("download_file")
async def download_file(remote_path: str,
                        local_path: str,
                        encoding: str = "utf-8") -> None

Download a file from the sandbox to a local path asynchronously.

Arguments:

  • remote_path - Path to the file in the sandbox
  • local_path - Destination path on the local filesystem
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary files.

ls

async def ls(path: str = ".") -> List[str]

List directory contents asynchronously.

Arguments:

  • path - Path to list

Returns:

List of file/directory names

rm

@async_wrapper("rm")
async def rm(path: str, recursive: bool = False) -> None

Remove file or directory asynchronously.

Arguments:

  • path - Path to remove
  • recursive - Remove recursively

open

def open(path: str,
         mode: str = "r",
         encoding: str = "utf-8") -> AsyncSandboxFileIO

Open a file in the sandbox asynchronously.

Arguments:

  • path - Path to the file
  • mode - Open mode ('r', 'w', 'a', etc.)
  • encoding - File encoding (default: "utf-8"). Use "base64" for binary data.

Returns:

  • AsyncSandboxFileIO - Async file handle

SandboxFileIO Objects

class SandboxFileIO()

Synchronous file I/O handle for sandbox files

read

def read() -> Union[str, bytes]

Read file content synchronously

write

def write(content: Union[str, bytes]) -> None

Write content to file synchronously

close

def close() -> None

Close the file

AsyncSandboxFileIO Objects

class AsyncSandboxFileIO()

Async file I/O handle for sandbox files

read

async def read() -> Union[str, bytes]

Read file content asynchronously

write

async def write(content: Union[str, bytes]) -> None

Write content to file asynchronously

close

def close() -> None

Close the file

koyeb/sandbox.sandbox

Koyeb Sandbox - Python SDK for creating and managing Koyeb sandboxes

ProcessInfo Objects

@dataclass
class ProcessInfo()

Type definition for process information returned by list_processes.

id

Process ID (UUID string)

command

The command that was executed

status

Process status (e.g., "running", "completed")

pid

OS process ID (if running)

exit_code

Exit code (if completed)

started_at

ISO 8601 timestamp when process started

ExposedPort Objects

@dataclass
class ExposedPort()

Result of exposing a port via TCP proxy.

Sandbox Objects

class Sandbox()

Synchronous sandbox for running code on Koyeb infrastructure. Provides creation and deletion functionality with proper health polling.

id

@property
def id() -> str

Get the service ID of the sandbox.

create

@classmethod
def create(cls,
           image: str = "koyeb/sandbox",
           name: str = "quick-sandbox",
           wait_ready: bool = True,
           instance_type: str = "micro",
           exposed_port_protocol: Optional[str] = None,
           env: Optional[Dict[str, str]] = None,
           region: Optional[str] = None,
           api_token: Optional[str] = None,
           timeout: int = 300,
           idle_timeout: int = 300,
           enable_tcp_proxy: bool = False,
           privileged: bool = False,
           registry_secret: Optional[str] = None,
           _experimental_enable_light_sleep: bool = False,
           _experimental_deep_sleep_value: int = 3900,
           delete_after_delay: int = 0,
           delete_after_inactivity_delay: int = 0,
           app_id: Optional[str] = None,
           enable_mesh: bool = None,
           poll_interval: float = DEFAULT_POLL_INTERVAL) -> Sandbox

Create a new sandbox instance.

Arguments:

  • image - Docker image to use (default: koyeb/sandbox)
  • name - Name of the sandbox
  • wait_ready - Wait for sandbox to be ready (default: True)
  • instance_type - Instance type (default: micro)
  • exposed_port_protocol - Protocol to expose ports with ("http" or "http2"). If None, defaults to "http". If provided, must be one of "http" or "http2".
  • env - Environment variables
  • region - Region to deploy to. Defaults to KOYEB_REGION env var, or "na" if not set.
  • api_token - Koyeb API token (if None, will try to get from KOYEB_API_TOKEN env var)
  • timeout - Timeout for sandbox creation in seconds
  • idle_timeout - Sleep timeout in seconds. Behavior depends on _experimental_enable_light_sleep:
    • If _experimental_enable_light_sleep is True: sets light_sleep value (deep_sleep=3900)
    • If _experimental_enable_light_sleep is False: sets deep_sleep value
    • If 0: disables scale-to-zero (keep always-on)
    • If None: uses default values
  • enable_tcp_proxy - If True, enables TCP proxy for direct TCP access to port 3031
  • privileged - If True, run the container in privileged mode (default: False)
  • registry_secret - Name of a Koyeb secret containing registry credentials for pulling private images. Create the secret via Koyeb dashboard or CLI first.
  • _experimental_enable_light_sleep - If True, uses idle_timeout for light_sleep and sets deep_sleep=3900. If False, uses idle_timeout for deep_sleep (default: False)
  • delete_after_create - If >0, automatically delete the sandbox if there was no activity after this many seconds since creation.
  • delete_after_sleep - If >0, automatically delete the sandbox if service sleeps due to inactivity after this many seconds.
  • app_id - If provided, create the sandbox service in an existing app instead of creating a new one.
  • enable_mesh - Enable or disable mesh for this sandbox. Disabled by default
  • poll_interval - Time between health checks in seconds when wait_ready is True (default: 0.5)

Returns:

  • Sandbox - A new Sandbox instance

Raises:

  • ValueError - If API token is not provided
  • SandboxTimeoutError - If wait_ready is True and sandbox does not become ready within timeout

Example:

Public image (default)

sandbox = Sandbox.create()

Private image with registry secret

sandbox = Sandbox.create( ... image="ghcr.io/myorg/myimage:latest", ... registry_secret="my-ghcr-secret" ... )

get_from_id

@classmethod
def get_from_id(cls, id: str, api_token: Optional[str] = None) -> "Sandbox"

Get a sandbox by service ID.

Arguments:

  • id - Service ID of the sandbox
  • api_token - Koyeb API token (if None, will try to get from KOYEB_API_TOKEN env var)

Returns:

  • Sandbox - The Sandbox instance

Raises:

  • ValueError - If API token is not provided or id is invalid
  • SandboxError - If sandbox is not found or retrieval fails

wait_ready

def wait_ready(timeout: int = DEFAULT_INSTANCE_WAIT_TIMEOUT,
               poll_interval: Optional[float] = None) -> bool

Wait for sandbox to become ready with exponential backoff polling.

Starts polling at 0.1s intervals, doubling each time up to poll_interval.

Arguments:

  • timeout - Maximum time to wait in seconds
  • poll_interval - Maximum time between health checks in seconds (defaults to instance poll_interval)

Returns:

  • bool - True if sandbox became ready, False if timeout

wait_tcp_proxy_ready

def wait_tcp_proxy_ready(timeout: int = DEFAULT_INSTANCE_WAIT_TIMEOUT,
                         poll_interval: Optional[float] = None) -> bool

Wait for TCP proxy to become ready and available.

Polls the deployment metadata with exponential backoff until the TCP proxy information is available. Starts at 0.1s intervals, doubling up to poll_interval.

Arguments:

  • timeout - Maximum time to wait in seconds
  • poll_interval - Maximum time between checks in seconds (defaults to instance poll_interval)

Returns:

  • bool - True if TCP proxy became ready, False if timeout

delete

def delete() -> None

Delete the sandbox instance.

get_domain

def get_domain() -> Optional[str]

Get the public domain of the sandbox.

Returns the domain (e.g., "app-name-org.koyeb.app/r/routing_key/" or "app-name-org.koyeb.app") without protocol. To get the full URL with protocol, use sandbox._get_url()

Returns:

  • Optional[str] - The domain or None if unavailable

get_tcp_proxy_info

def get_tcp_proxy_info() -> Optional[tuple[str, int]]

Get the TCP proxy host and port for the sandbox.

Returns the TCP proxy host and port as a tuple (host, port) for direct TCP access to port 3031. This is only available if enable_tcp_proxy=True was set when creating the sandbox.

Returns:

Optional[tuple[str, int]]: A tuple of (host, port) or None if unavailable

is_healthy

def is_healthy() -> bool

Check if sandbox is healthy and ready for operations

filesystem

@property
def filesystem() -> "SandboxFilesystem"

Get filesystem operations interface

exec

@property
def exec() -> "SandboxExecutor"

Get command execution interface

expose_port

def expose_port(port: int) -> ExposedPort

Expose a port to external connections via TCP proxy.

Binds the specified internal port to the TCP proxy, allowing external connections to reach services running on that port inside the sandbox. Automatically unbinds any existing port before binding the new one.

Arguments:

  • port - The internal port number to expose (must be a valid port number between 1 and 65535)

Returns:

  • ExposedPort - An object with port and exposed_at attributes:

Raises:

  • ValueError - If port is not in valid range [1, 65535]
  • SandboxError - If the port binding operation fails

Notes:

  • Only one port can be exposed at a time
  • Any existing port binding is automatically unbound before binding the new port
  • The port must be available and accessible within the sandbox environment
  • The TCP proxy is accessed via get_tcp_proxy_info() which returns (host, port)

Example:

result = sandbox.expose_port(8080) result.port 8080 result.exposed_at 'https://app-name-org.koyeb.app'

unexpose_port

def unexpose_port() -> None

Unexpose a port from external connections.

Removes the TCP proxy port binding, stopping traffic forwarding to the previously bound port.

Raises:

  • SandboxError - If the port unbinding operation fails

Notes:

  • After unexposing, the TCP proxy will no longer forward traffic
  • Safe to call even if no port is currently bound

launch_process

def launch_process(cmd: str,
                   cwd: Optional[str] = None,
                   env: Optional[Dict[str, str]] = None) -> str

Launch a background process in the sandbox.

Starts a long-running background process that continues executing even after the method returns. Use this for servers, workers, or other long-running tasks.

Arguments:

  • cmd - The shell command to execute as a background process
  • cwd - Optional working directory for the process
  • env - Optional environment variables to set/override for the process

Returns:

  • str - The unique process ID (UUID string) that can be used to manage the process

Raises:

  • SandboxError - If the process launch fails

Example:

process_id = sandbox.launch_process("python -u server.py") print(f"Started process: {process_id}")

kill_process

def kill_process(process_id: str) -> None

Kill a background process by its ID.

Terminates a running background process. This sends a SIGTERM signal to the process, allowing it to clean up gracefully. If the process doesn't terminate within a timeout, it will be forcefully killed with SIGKILL.

Arguments:

  • process_id - The unique process ID (UUID string) to kill

Raises:

  • SandboxError - If the process kill operation fails

Example:

sandbox.kill_process("550e8400-e29b-41d4-a716-446655440000")

list_processes

def list_processes() -> List[ProcessInfo]

List all background processes.

Returns information about all currently running and recently completed background processes. This includes both active processes and processes that have completed (which remain in memory until server restart).

Returns:

  • List[ProcessInfo] - List of process objects, each containing:
    • id: Process ID (UUID string)
    • command: The command that was executed
    • status: Process status (e.g., "running", "completed")
    • pid: OS process ID (if running)
    • exit_code: Exit code (if completed)
    • started_at: ISO 8601 timestamp when process started
    • completed_at: ISO 8601 timestamp when process completed (if applicable)

Raises:

  • SandboxError - If listing processes fails

Example:

processes = sandbox.list_processes() for process in processes: ... print(f"{process.id}: {process.command} - {process.status}")

kill_all_processes

def kill_all_processes() -> int

Kill all running background processes.

Convenience method that lists all processes and kills them all. This is useful for cleanup operations.

Returns:

  • int - The number of processes that were killed

Raises:

  • SandboxError - If listing or killing processes fails

Example:

count = sandbox.kill_all_processes() print(f"Killed {count} processes")

update_lifecycle

def update_lifecycle(delete_after_delay: Optional[int] = None,
                     delete_after_inactivity: Optional[int] = None) -> None

Update the sandbox's life cycle settings.

Arguments:

  • delete_after_delay - If >0, automatically delete the sandbox if there was no activity after this many seconds since creation.
  • delete_after_inactivity - If >0, automatically delete the sandbox if service sleeps due to inactivity after this many seconds.

Raises:

  • SandboxError - If updating life cycle fails

Example:

sandbox.update_life_cycle(delete_after_delay=600, delete_after_inactivity=300)

__enter__

def __enter__() -> "Sandbox"

Context manager entry - returns self.

__exit__

def __exit__(exc_type, exc_val, exc_tb) -> None

Context manager exit - automatically deletes the sandbox.

AsyncSandbox Objects

class AsyncSandbox(Sandbox)

Async sandbox for running code on Koyeb infrastructure. Inherits from Sandbox and provides async wrappers for all operations.

get_from_id

@classmethod
async def get_from_id(cls,
                      id: str,
                      api_token: Optional[str] = None) -> "AsyncSandbox"

Get a sandbox by service ID asynchronously.

Arguments:

  • id - Service ID of the sandbox
  • api_token - Koyeb API token (if None, will try to get from KOYEB_API_TOKEN env var)

Returns:

  • AsyncSandbox - The AsyncSandbox instance

Raises:

  • ValueError - If API token is not provided or id is invalid
  • SandboxError - If sandbox is not found or retrieval fails

create

@classmethod
async def create(cls,
                 image: str = "koyeb/sandbox",
                 name: str = "quick-sandbox",
                 wait_ready: bool = True,
                 instance_type: str = "micro",
                 exposed_port_protocol: Optional[str] = None,
                 env: Optional[Dict[str, str]] = None,
                 region: Optional[str] = None,
                 api_token: Optional[str] = None,
                 timeout: int = 300,
                 idle_timeout: int = 0,
                 enable_tcp_proxy: bool = False,
                 privileged: bool = False,
                 registry_secret: Optional[str] = None,
                 _experimental_enable_light_sleep: bool = False,
                 _experimental_deep_sleep_value: int = 3900,
                 delete_after_delay: int = 0,
                 delete_after_inactivity_delay: int = 0,
                 app_id: Optional[str] = None,
                 enable_mesh: bool = False,
                 poll_interval: float = DEFAULT_POLL_INTERVAL) -> AsyncSandbox

Create a new sandbox instance with async support.

Arguments:

  • image - Docker image to use (default: koyeb/sandbox)
  • name - Name of the sandbox
  • wait_ready - Wait for sandbox to be ready (default: True)
  • instance_type - Instance type (default: micro)
  • exposed_port_protocol - Protocol to expose ports with ("http" or "http2"). If None, defaults to "http". If provided, must be one of "http" or "http2".
  • env - Environment variables
  • region - Region to deploy to. Defaults to KOYEB_REGION env var, or "na" if not set.
  • api_token - Koyeb API token (if None, will try to get from KOYEB_API_TOKEN env var)
  • timeout - Timeout for sandbox creation in seconds
  • idle_timeout - Sleep timeout in seconds. Behavior depends on _experimental_enable_light_sleep:
    • If _experimental_enable_light_sleep is True: sets light_sleep value (deep_sleep uses _experimental_deep_sleep_value)
    • If _experimental_enable_light_sleep is False: sets deep_sleep value
    • If 0: disables scale-to-zero (keep always-on)
    • If None: uses default values
  • enable_tcp_proxy - If True, enables TCP proxy for direct TCP access to port 3031
  • privileged - If True, run the container in privileged mode (default: False)
  • registry_secret - Name of a Koyeb secret containing registry credentials for pulling private images. Create the secret via Koyeb dashboard or CLI first.
  • _experimental_enable_light_sleep - If True, uses idle_timeout for light_sleep and configurable deep_sleep (default: False)
  • _experimental_deep_sleep_value - Number of seconds for deep sleep when light sleep is enabled (default: 3900). Only used if _experimental_enable_light_sleep is True
  • delete_after_delay - If >0, automatically delete the sandbox if there was no activity after this many seconds since creation.
  • delete_after_inactivity_delay - If >0, automatically delete the sandbox if service sleeps due to inactivity after this many seconds.
  • app_id - If provided, create the sandbox service in an existing app instead of creating a new one.
  • enable_mesh - Enable or disable mesh for this sandbox. Disabled by default
  • poll_interval - Time between health checks in seconds when wait_ready is True (default: 0.5)

Returns:

  • AsyncSandbox - A new AsyncSandbox instance

Raises:

  • ValueError - If API token is not provided
  • SandboxTimeoutError - If wait_ready is True and sandbox does not become ready within timeout

wait_ready

async def wait_ready(timeout: int = DEFAULT_INSTANCE_WAIT_TIMEOUT,
                     poll_interval: Optional[float] = None) -> bool

Wait for sandbox to become ready with exponential backoff async polling.

Starts polling at 0.1s intervals, doubling each time up to poll_interval.

Arguments:

  • timeout - Maximum time to wait in seconds
  • poll_interval - Maximum time between health checks in seconds (defaults to instance poll_interval)

Returns:

  • bool - True if sandbox became ready, False if timeout

wait_tcp_proxy_ready

async def wait_tcp_proxy_ready(timeout: int = DEFAULT_INSTANCE_WAIT_TIMEOUT,
                               poll_interval: Optional[float] = None) -> bool

Wait for TCP proxy to become ready and available asynchronously.

Polls the deployment metadata with exponential backoff until the TCP proxy information is available. Starts at 0.1s intervals, doubling up to poll_interval.

Arguments:

  • timeout - Maximum time to wait in seconds
  • poll_interval - Maximum time between checks in seconds (defaults to instance poll_interval)

Returns:

  • bool - True if TCP proxy became ready, False if timeout

delete

@async_wrapper("delete")
async def delete() -> None

Delete the sandbox instance asynchronously.

is_healthy

@async_wrapper("is_healthy")
async def is_healthy() -> bool

Check if sandbox is healthy and ready for operations asynchronously

exec

@property
def exec() -> "AsyncSandboxExecutor"

Get async command execution interface

filesystem

@property
def filesystem() -> "AsyncSandboxFilesystem"

Get filesystem operations interface

expose_port

@async_wrapper("expose_port")
async def expose_port(port: int) -> ExposedPort

Expose a port to external connections via TCP proxy asynchronously.

unexpose_port

@async_wrapper("unexpose_port")
async def unexpose_port() -> None

Unexpose a port from external connections asynchronously.

launch_process

@async_wrapper("launch_process")
async def launch_process(cmd: str,
                         cwd: Optional[str] = None,
                         env: Optional[Dict[str, str]] = None) -> str

Launch a background process in the sandbox asynchronously.

kill_process

@async_wrapper("kill_process")
async def kill_process(process_id: str) -> None

Kill a background process by its ID asynchronously.

list_processes

@async_wrapper("list_processes")
async def list_processes() -> List[ProcessInfo]

List all background processes asynchronously.

kill_all_processes

async def kill_all_processes() -> int

Kill all running background processes asynchronously.

update_lifecycle

@async_wrapper("update_lifecycle")
async def update_lifecycle(
        delete_after_delay: Optional[int] = None,
        delete_after_inactivity: Optional[int] = None) -> None

Update the sandbox's life cycle settings asynchronously.

__aenter__

async def __aenter__() -> "AsyncSandbox"

Async context manager entry - returns self.

__aexit__

async def __aexit__(exc_type, exc_val, exc_tb) -> None

Async context manager exit - automatically deletes the sandbox.

koyeb/sandbox.utils

Utility functions for Koyeb Sandbox

DEFAULT_INSTANCE_WAIT_TIMEOUT

seconds

DEFAULT_POLL_INTERVAL

seconds

DEFAULT_COMMAND_TIMEOUT

seconds

DEFAULT_HTTP_TIMEOUT

seconds for HTTP requests

get_api_client

def get_api_client(
    api_token: Optional[str] = None,
    host: Optional[str] = None
) -> tuple[AppsApi, ServicesApi, InstancesApi, CatalogInstancesApi,
           DeploymentsApi]

Get configured API clients for Koyeb operations.

Arguments:

  • api_token - Koyeb API token. If not provided, will try to get from KOYEB_API_TOKEN env var
  • host - Koyeb API host URL. If not provided, will try to get from KOYEB_API_HOST env var (defaults to https://app.koyeb.com)

Returns:

Tuple of (AppsApi, ServicesApi, InstancesApi, CatalogInstancesApi) instances

Raises:

  • ValueError - If API token is not provided

build_env_vars

def build_env_vars(env: Optional[Dict[str, str]]) -> List[DeploymentEnv]

Build environment variables list from dictionary.

Arguments:

  • env - Dictionary of environment variables

Returns:

List of DeploymentEnv objects

create_docker_source

def create_docker_source(
        image: str,
        command_args: List[str],
        privileged: Optional[bool] = None,
        image_registry_secret: Optional[str] = None) -> DockerSource

Create Docker source configuration.

Arguments:

  • image - Docker image name
  • command_args - Command and arguments to run (optional, empty list means use image default)
  • privileged - If True, run the container in privileged mode (default: None/False)
  • image_registry_secret - Name of the secret containing registry credentials for pulling private images

Returns:

DockerSource object

create_koyeb_sandbox_ports

def create_koyeb_sandbox_ports(protocol: str = "http") -> List[DeploymentPort]

Create port configuration for koyeb/sandbox image.

Creates two ports:

  • Port 3030 exposed on HTTP, mounted on /koyeb-sandbox/
  • Port 3031 exposed with the specified protocol, mounted on /

Arguments:

  • protocol - Protocol to use for port 3031 ("http" or "http2"), defaults to "http"

Returns:

List of DeploymentPort objects configured for koyeb/sandbox

create_koyeb_sandbox_proxy_ports

def create_koyeb_sandbox_proxy_ports() -> List[DeploymentProxyPort]

Create TCP proxy port configuration for koyeb/sandbox image.

Creates proxy port for direct TCP access:

  • Port 3031 exposed via TCP proxy

Returns:

List of DeploymentProxyPort objects configured for TCP proxy access

create_koyeb_sandbox_routes

def create_koyeb_sandbox_routes() -> List[DeploymentRoute]

Create route configuration for koyeb/sandbox image to make it publicly accessible.

Creates two routes:

  • Port 3030 accessible at /koyeb-sandbox/
  • Port 3031 accessible at /

Returns:

List of DeploymentRoute objects configured for koyeb/sandbox

create_deployment_definition

def create_deployment_definition(
        name: str,
        docker_source: DockerSource,
        env_vars: List[DeploymentEnv],
        instance_type: str,
        exposed_port_protocol: Optional[str] = None,
        region: Optional[str] = None,
        routes: Optional[List[DeploymentRoute]] = None,
        idle_timeout: int = 300,
        enable_tcp_proxy: bool = False,
        _experimental_enable_light_sleep: bool = False,
        _experimental_deep_sleep_value: int = 3900,
        enable_mesh: bool = None) -> DeploymentDefinition

Create deployment definition for a sandbox service.

Arguments:

  • name - Service name
  • docker_source - Docker configuration
  • env_vars - Environment variables
  • instance_type - Instance type
  • exposed_port_protocol - Protocol to expose ports with ("http" or "http2"). If None, defaults to "http". If provided, must be one of "http" or "http2".
  • region - Region to deploy to. Defaults to KOYEB_REGION env var, or "na" if not set.
  • routes - List of routes for public access
  • idle_timeout - Number of seconds to wait before sleeping the instance if it receives no traffic
  • enable_tcp_proxy - If True, enables TCP proxy for direct TCP access to port 3031
  • _experimental_enable_light_sleep - If True, uses light sleep when reaching idle_timeout. Light Sleep reduces cold starts to ~200ms. After scaling to zero, the service stays in Light Sleep for idle_timeout seconds before going into Deep Sleep.
  • _experimental_deep_sleep_value - Number of seconds for deep sleep when light sleep is enabled (default: 3900). Only used if _experimental_enable_light_sleep is True. Ignored otherwise.
  • enable_mesh - Enable or disable mesh for this sandbox. Disabled by default

Returns:

DeploymentDefinition object

escape_shell_arg

def escape_shell_arg(arg: str) -> str

Escape a shell argument for safe use in shell commands.

Arguments:

  • arg - The argument to escape

Returns:

Properly escaped shell argument

validate_port

def validate_port(port: int) -> None

Validate that a port number is in the valid range.

Arguments:

  • port - Port number to validate

Raises:

  • ValueError - If port is not in valid range [1, 65535]

check_error_message

def check_error_message(error_msg: str, error_type: str) -> bool

Check if an error message matches a specific error type. Uses case-insensitive matching against known error patterns.

Arguments:

  • error_msg - The error message to check
  • error_type - The type of error to check for (key in ERROR_MESSAGES)

Returns:

True if error message matches the error type

run_sync_in_executor

async def run_sync_in_executor(method: Callable[..., Any], *args: Any,
                               **kwargs: Any) -> Any

Run a synchronous method in an async executor.

Helper function to wrap synchronous methods for async execution. Used by AsyncSandbox and AsyncSandboxFilesystem to wrap sync parent methods.

Arguments:

  • method - The synchronous method to run
  • *args - Positional arguments for the method
  • **kwargs - Keyword arguments for the method

Returns:

Result of the synchronous method call

async_wrapper

def async_wrapper(method_name: str)

Decorator to automatically create async wrapper for sync methods.

This decorator creates an async method that wraps a sync method from the parent class. The sync method is called via super() and executed in an executor.

Arguments:

  • method_name - Name of the sync method to wrap (from parent class)

    Usage: @async_wrapper("delete") async def delete(self) -> None: """Delete the sandbox instance asynchronously.""" pass # Implementation is handled by decorator

create_sandbox_client

def create_sandbox_client(conn_info: Optional['ConnectionInfo'],
                          existing_client: Optional[Any] = None) -> Any

Create or return existing SandboxClient instance with validation.

Helper function to create SandboxClient instances with consistent validation. Used by Sandbox, SandboxExecutor, and SandboxFilesystem to avoid duplication.

Arguments:

  • conn_info - The information needed to connect to the sandbox executor API
  • existing_client - Existing client instance to return if not None

Returns:

  • SandboxClient - Configured client instance

Raises:

  • SandboxError - If sandbox URL or secret is not available

SandboxError Objects

class SandboxError(Exception)

Base exception for sandbox operations

SandboxTimeoutError Objects

class SandboxTimeoutError(SandboxError)

Raised when a sandbox operation times out

koyeb/sandbox.executor_client

Sandbox Executor API Client

A simple Python client for interacting with the Sandbox Executor API.

ConnectionInfo Objects

@dataclass
class ConnectionInfo()

Information needed to connect to a sandbox

SandboxClient Objects

class SandboxClient()

Client for the Sandbox Executor API.

__init__

def __init__(conn_info: ConnectionInfo, timeout: float = DEFAULT_HTTP_TIMEOUT)

Initialize the Sandbox Client.

Arguments:

  • conn_info - The parameters needed to connect to the sandbox
  • timeout - Request timeout in seconds (default: 30)

close

def close() -> None

Close the HTTP session and release resources.

__enter__

def __enter__()

Context manager entry - returns self.

__exit__

def __exit__(exc_type, exc_val, exc_tb) -> None

Context manager exit - automatically closes the session.

__del__

def __del__()

Clean up session on deletion (fallback, not guaranteed to run).

health

def health() -> Dict[str, str]

Check the health status of the server.

Uses a short timeout and no retries since callers (wait_ready) already handle polling with backoff.

Returns:

Dict with status information

Raises:

  • requests.HTTPError - If the health check fails
  • requests.Timeout - If the health check times out

run

def run(cmd: str,
        cwd: Optional[str] = None,
        env: Optional[Dict[str, str]] = None,
        timeout: Optional[float] = None) -> Dict[str, Any]

Execute a shell command in the sandbox.

Arguments:

  • cmd - The shell command to execute
  • cwd - Optional working directory for command execution
  • env - Optional environment variables to set/override
  • timeout - Optional timeout in seconds for the request

Returns:

Dict containing stdout, stderr, error (if any), and exit code

run_streaming

def run_streaming(cmd: str,
                  cwd: Optional[str] = None,
                  env: Optional[Dict[str, str]] = None,
                  timeout: Optional[float] = None) -> Iterator[Dict[str, Any]]

Execute a shell command in the sandbox and stream the output in real-time.

This method uses Server-Sent Events (SSE) to stream command output line-by-line as it's produced. Use this for long-running commands where you want real-time output. For simple commands where buffered output is acceptable, use run() instead.

Arguments:

  • cmd - The shell command to execute
  • cwd - Optional working directory for command execution
  • env - Optional environment variables to set/override
  • timeout - Optional timeout in seconds for the streaming request

Yields:

Dict events with the following types:

  • output events (as command produces output):

  • {"stream" - "stdout"|"stderr", "data": "line of output"}

    • complete event (when command finishes):

  • {"code" - <exit_code>, "error": false}

    • error event (if command fails to start):

  • {"error" - "error message"}

Example:

client = SandboxClient("http://localhost:8080", "secret") for event in client.run_streaming("echo 'Hello'; sleep 1; echo 'World'"): ... if "stream" in event: ... print(f"{event['stream']}: {event['data']}") ... elif "code" in event: ... print(f"Exit code: {event['code']}")

write_file

def write_file(path: str, content: str) -> Dict[str, Any]

Write content to a file.

Arguments:

  • path - The file path to write to
  • content - The content to write

Returns:

Dict with success status and error if any

read_file

def read_file(path: str) -> Dict[str, Any]

Read content from a file.

Arguments:

  • path - The file path to read from

Returns:

Dict with file content and error if any

delete_file

def delete_file(path: str) -> Dict[str, Any]

Delete a file.

Arguments:

  • path - The file path to delete

Returns:

Dict with success status and error if any

make_dir

def make_dir(path: str) -> Dict[str, Any]

Create a directory (including parent directories).

Arguments:

  • path - The directory path to create

Returns:

Dict with success status and error if any

delete_dir

def delete_dir(path: str) -> Dict[str, Any]

Recursively delete a directory and all its contents.

Arguments:

  • path - The directory path to delete

Returns:

Dict with success status and error if any

list_dir

def list_dir(path: str) -> Dict[str, Any]

List the contents of a directory.

Arguments:

  • path - The directory path to list

Returns:

Dict with entries list and error if any

bind_port

def bind_port(port: int) -> Dict[str, Any]

Bind a port to the TCP proxy for external access.

Configures the TCP proxy to forward traffic to the specified port inside the sandbox. This allows you to expose services running inside the sandbox to external connections.

Arguments:

  • port - The port number to bind to (must be a valid port number)

Returns:

Dict with success status, message, and port information

Notes:

  • Only one port can be bound at a time
  • Binding a new port will override the previous binding
  • The port must be available and accessible within the sandbox environment

unbind_port

def unbind_port(port: Optional[int] = None) -> Dict[str, Any]

Unbind a port from the TCP proxy.

Removes the TCP proxy port binding, stopping traffic forwarding to the previously bound port.

Arguments:

  • port - Optional port number to unbind. If provided, it must match the currently bound port. If not provided, any existing binding will be removed.

Returns:

Dict with success status and message

Notes:

  • If a port is specified and doesn't match the currently bound port, the request will fail
  • After unbinding, the TCP proxy will no longer forward traffic

start_process

def start_process(cmd: str,
                  cwd: Optional[str] = None,
                  env: Optional[Dict[str, str]] = None) -> Dict[str, Any]

Start a background process in the sandbox.

Starts a long-running background process that continues executing even after the API call completes. Use this for servers, workers, or other long-running tasks.

Arguments:

  • cmd - The shell command to execute as a background process
  • cwd - Optional working directory for the process
  • env - Optional environment variables to set/override for the process

Returns:

Dict with process id and success status:

  • id: The unique process ID (UUID string)
  • success: True if the process was started successfully

Example:

client = SandboxClient("http://localhost:8080", "secret") result = client.start_process("python -u server.py") process_id = result["id"] print(f"Started process: {process_id}")

kill_process

def kill_process(process_id: str) -> Dict[str, Any]

Kill a background process by its ID.

Terminates a running background process. This sends a SIGTERM signal to the process, allowing it to clean up gracefully. If the process doesn't terminate within a timeout, it will be forcefully killed with SIGKILL.

Arguments:

  • process_id - The unique process ID (UUID string) to kill

Returns:

Dict with success status and error message if any

Example:

client = SandboxClient("http://localhost:8080", "secret") result = client.kill_process("550e8400-e29b-41d4-a716-446655440000") if result.get("success"): ... print("Process killed successfully")

list_processes

def list_processes() -> Dict[str, Any]

List all background processes.

Returns information about all currently running and recently completed background processes. This includes both active processes and processes that have completed (which remain in memory until server restart).

Returns:

Dict with a list of processes:

  • processes: List of process objects, each containing:
  • id: Process ID (UUID string)
  • command: The command that was executed
  • status: Process status (e.g., "running", "completed")
  • pid: OS process ID (if running)
  • exit_code: Exit code (if completed)
  • started_at: ISO 8601 timestamp when process started
  • completed_at: ISO 8601 timestamp when process completed (if applicable)

Example:

client = SandboxClient("http://localhost:8080", "secret") result = client.list_processes() for process in result.get("processes", []): ... print(f"{process['id']}: {process['command']} - {process['status']}")