Sandbox Sync
CommandHandle
class CommandHandle()Command execution handle.
It provides methods for waiting for the command to finish, retrieving stdout/stderr, and killing the command.
pid
@property
def pid()Command process ID.
__iter__
def __iter__()Iterate over the command output.
Returns:
Generator of command outputs
disconnect
def disconnect() -> NoneDisconnect from the command.
The command is not killed, but SDK stops receiving events from the command.
You can reconnect to the command using sandbox.commands.connect method.
wait
def wait(on_pty: Optional[Callable[[PtyOutput], None]] = None,
on_stdout: Optional[Callable[[str], None]] = None,
on_stderr: Optional[Callable[[str], None]] = None) -> CommandResultWait for the command to finish and returns the result.
If the command exits with a non-zero exit code, it throws a CommandExitException.
Arguments:
on_pty: Callback for pty outputon_stdout: Callback for stdout outputon_stderr: Callback for stderr output
Returns:
CommandResult result of command execution
kill
def kill() -> boolKills the command.
It uses SIGKILL signal to kill the command.
Returns:
Whether the command was killed successfully
Pty
class Pty()Module for interacting with PTYs (pseudo-terminals) in the sandbox.
kill
def kill(pid: int, request_timeout: Optional[float] = None) -> boolKill PTY.
Arguments:
pid: Process ID of the PTYrequest_timeout: Timeout for the request in seconds
Returns:
true if the PTY was killed, false if the PTY was not found
send_stdin
def send_stdin(pid: int,
data: bytes,
request_timeout: Optional[float] = None) -> NoneSend input to a PTY.
Arguments:
pid: Process ID of the PTYdata: Input data to sendrequest_timeout: Timeout for the request in seconds
create
def create(size: PtySize,
user: Optional[Username] = None,
cwd: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
timeout: Optional[float] = 60,
request_timeout: Optional[float] = None) -> CommandHandleStart a new PTY (pseudo-terminal).
Arguments:
size: Size of the PTYuser: User to use for the PTYcwd: Working directory for the PTYenvs: Environment variables for the PTYtimeout: Timeout for the PTY in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Handle to interact with the PTY
connect
def connect(pid: int,
timeout: Optional[float] = 60,
request_timeout: Optional[float] = None) -> CommandHandleConnect to a running PTY.
Arguments:
pid: Process ID of the PTY to connect to. You can get the list of running PTYs usingsandbox.pty.list().timeout: Timeout for the PTY connection in seconds. Using0will not limit the connection timerequest_timeout: Timeout for the request in seconds
Returns:
Handle to interact with the PTY
resize
def resize(pid: int,
size: PtySize,
request_timeout: Optional[float] = None) -> NoneResize PTY.
Call this when the terminal window is resized and the number of columns and rows has changed.
Arguments:
pid: Process ID of the PTYsize: New size of the PTYrequest_timeout: Timeout for the request in secondss
Commands
class Commands()Module for executing commands in the sandbox.
list
def list(request_timeout: Optional[float] = None) -> List[ProcessInfo]Lists all running commands and PTY sessions.
Arguments:
request_timeout: Timeout for the request in seconds
Returns:
List of running commands and PTY sessions
kill
def kill(pid: int, request_timeout: Optional[float] = None) -> boolKills a running command specified by its process ID.
It uses SIGKILL signal to kill the command.
Arguments:
pid: Process ID of the command. You can get the list of processes usingsandbox.commands.list()request_timeout: Timeout for the request in seconds
Returns:
True if the command was killed, False if the command was not found
send_stdin
def send_stdin(pid: int, data: str, request_timeout: Optional[float] = None)Send data to command stdin.
:param pid Process ID of the command. You can get the list of processes using sandbox.commands.list().
:param data: Data to send to the command
:param request_timeout: Timeout for the request in seconds
run
@overload
def run(cmd: str,
background: Union[Literal[False], None] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[Username] = None,
cwd: Optional[str] = None,
on_stdout: Optional[Callable[[str], None]] = None,
on_stderr: Optional[Callable[[str], None]] = None,
stdin: Optional[bool] = None,
timeout: Optional[float] = 60,
request_timeout: Optional[float] = None) -> CommandResultStart a new command and wait until it finishes executing.
Arguments:
cmd: Command to executebackground:Falseif the command should be executed in the foreground,Trueif the command should be executed in the backgroundenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandon_stdout: Callback for command stdout outputon_stderr: Callback for command stderr outputstdin: IfTrue, the command will have a stdin stream that you can send data to usingsandbox.commands.send_stdin()timeout: Timeout for the command connection in seconds. Using0will not limit the command connection timerequest_timeout: Timeout for the request in seconds
Returns:
CommandResult result of the command execution
run
@overload
def run(cmd: str,
background: Literal[True],
envs: Optional[Dict[str, str]] = None,
user: Optional[Username] = None,
cwd: Optional[str] = None,
on_stdout: None = None,
on_stderr: None = None,
stdin: Optional[bool] = None,
timeout: Optional[float] = 60,
request_timeout: Optional[float] = None) -> CommandHandleStart a new command and return a handle to interact with it.
Arguments:
cmd: Command to executebackground:Falseif the command should be executed in the foreground,Trueif the command should be executed in the backgroundenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandstdin: IfTrue, the command will have a stdin stream that you can send data to usingsandbox.commands.send_stdin()timeout: Timeout for the command connection in seconds. Using0will not limit the command connection timerequest_timeout: Timeout for the request in seconds
Returns:
CommandHandle handle to interact with the running command
connect
def connect(pid: int,
timeout: Optional[float] = 60,
request_timeout: Optional[float] = None)Connects to a running command.
You can use CommandHandle.wait() to wait for the command to finish and get execution results.
Arguments:
pid: Process ID of the command to connect to. You can get the list of processes usingsandbox.commands.list()timeout: Timeout for the connection in seconds. Using0will not limit the connection timerequest_timeout: Timeout for the request in seconds
Returns:
CommandHandle handle to interact with the running command
Git
class Git()Module for running git operations in the sandbox.
__init__
def __init__(commands: Commands) -> NoneCreate a Git helper bound to the sandbox command runner.
Arguments:
commands: Command runner used to execute git commands
clone
def clone(url: str,
path: Optional[str] = None,
branch: Optional[str] = None,
depth: Optional[int] = None,
username: Optional[str] = None,
password: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None,
dangerously_store_credentials: bool = False)Clone a git repository into the sandbox.
Arguments:
url: Git repository URLpath: Destination path for the clonebranch: Branch to check outdepth: If set, perform a shallow clone with this depthusername: Username for HTTP(S) authenticationpassword: Password or token for HTTP(S) authenticationenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in secondsdangerously_store_credentials: Store credentials in the cloned repository when True
Returns:
Command result from the command runner
init
def init(path: str,
bare: bool = False,
initial_branch: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Initialize a new git repository.
Arguments:
path: Destination path for the repositorybare: Create a bare repository when Trueinitial_branch: Initial branch name (for example, “main”)envs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
remote_add
def remote_add(path: str,
name: str,
url: str,
fetch: bool = False,
overwrite: bool = False,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Add (or update) a remote for a repository.
Arguments:
path: Repository pathname: Remote name (for example, “origin”)url: Remote URLfetch: Fetch the remote after adding it when Trueoverwrite: Overwrite the remote URL if it already exists when Trueenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
remote_get
def remote_get(path: str,
name: str,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None) -> Optional[str]Get the URL for a git remote.
Returns None when the remote does not exist.
Arguments:
path: Repository pathname: Remote name (for example, “origin”)envs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Remote URL if present, otherwise None
status
def status(path: str,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None) -> GitStatusGet repository status information.
Arguments:
path: Repository pathenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Parsed git status
branches
def branches(path: str,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None) -> GitBranchesList branches in a repository.
Arguments:
path: Repository pathenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Parsed branch list
create_branch
def create_branch(path: str,
branch: str,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Create and check out a new branch.
Arguments:
path: Repository pathbranch: Branch name to createenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
checkout_branch
def checkout_branch(path: str,
branch: str,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Check out an existing branch.
Arguments:
path: Repository pathbranch: Branch name to check outenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
delete_branch
def delete_branch(path: str,
branch: str,
force: bool = False,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Delete a branch.
Arguments:
path: Repository pathbranch: Branch name to deleteforce: Force deletion with-DwhenTrueenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
add
def add(path: str,
files: Optional[List[str]] = None,
all: bool = True,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Stage files for commit.
Arguments:
path: Repository pathfiles: Files to add; when omitted, adds the current directoryall: WhenTrueandfilesis omitted, stage all changesenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
commit
def commit(path: str,
message: str,
author_name: Optional[str] = None,
author_email: Optional[str] = None,
allow_empty: bool = False,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Create a commit in the repository.
Arguments:
path: Repository pathmessage: Commit messageauthor_name: Commit author nameauthor_email: Commit author emailallow_empty: Allow empty commits whenTrueenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
reset
def reset(path: str,
mode: Optional[str] = None,
target: Optional[str] = None,
paths: Optional[List[str]] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Reset the current HEAD to a specified state.
Arguments:
path: Repository pathmode: Reset mode (soft, mixed, hard, merge, keep)target: Commit, branch, or ref to reset to (defaults to HEAD)paths: Paths to resetenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
restore
def restore(path: str,
paths: List[str],
staged: Optional[bool] = None,
worktree: Optional[bool] = None,
source: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Restore working tree files or unstage changes.
Arguments:
path: Repository pathpaths: Paths to restore (use [”.”] for all)staged: When True, restore the index (unstage)worktree: When True, restore working tree filessource: Restore from the given source (commit, branch, or ref)envs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
push
def push(path: str,
remote: Optional[str] = None,
branch: Optional[str] = None,
set_upstream: bool = True,
username: Optional[str] = None,
password: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Push commits to a remote.
Arguments:
path: Repository pathremote: Remote name, e.g.originbranch: Branch name to pushset_upstream: Set upstream tracking whenTrueusername: Username for HTTP(S) authenticationpassword: Password or token for HTTP(S) authenticationenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
pull
def pull(path: str,
remote: Optional[str] = None,
branch: Optional[str] = None,
username: Optional[str] = None,
password: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Pull changes from a remote.
Arguments:
path: Repository pathremote: Remote name, e.g.originbranch: Branch name to pullusername: Username for HTTP(S) authenticationpassword: Password or token for HTTP(S) authenticationenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
set_config
def set_config(key: str,
value: str,
scope: str = "global",
path: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Set a git config value.
Use scope="local" together with path to configure a specific repository.
Arguments:
key: Git config key (e.g.pull.rebase)value: Git config valuescope: Config scope:global,local, orsystempath: Repository path required whenscopeislocalenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
get_config
def get_config(key: str,
scope: str = "global",
path: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None) -> Optional[str]Get a git config value.
Returns None when the key is not set in the requested scope.
Arguments:
key: Git config key (e.g.pull.rebase)scope: Config scope:global,local, orsystempath: Repository path required whenscopeislocalenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Config value if present, otherwise None
dangerously_authenticate
def dangerously_authenticate(username: str,
password: str,
host: str = "github.com",
protocol: str = "https",
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Dangerously authenticate git globally via the credential helper.
This persists credentials in the credential store and may be accessible to agents running on the sandbox. Prefer short-lived credentials when possible.
Arguments:
username: Username for HTTP(S) authenticationpassword: Password or token for HTTP(S) authenticationhost: Host to authenticate for, defaults togithub.comprotocol: Protocol to authenticate for, defaults tohttpsenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
configure_user
def configure_user(name: str,
email: str,
scope: str = "global",
path: Optional[str] = None,
envs: Optional[Dict[str, str]] = None,
user: Optional[str] = None,
cwd: Optional[str] = None,
timeout: Optional[float] = None,
request_timeout: Optional[float] = None)Configure git user name and email.
Arguments:
name: Git user nameemail: Git user emailscope: Config scope:global,local, orsystempath: Repository path required whenscopeislocalenvs: Environment variables used for the commanduser: User to run the command ascwd: Working directory to run the commandtimeout: Timeout for the command connection in secondsrequest_timeout: Timeout for the request in seconds
Returns:
Command result from the command runner
WatchHandle
class WatchHandle()Handle for watching filesystem events. It is used to get the latest events that have occurred in the watched directory.
Use .stop() to stop watching the directory.
stop
def stop()Stop watching the directory. After you stop the watcher you won’t be able to get the events anymore.
get_new_events
def get_new_events() -> List[FilesystemEvent]Get the latest events that have occurred in the watched directory since the last call, or from the beginning of the watching, up until now.
Returns:
List of filesystem events
Filesystem
class Filesystem()Module for interacting with the filesystem in the sandbox.
read
@overload
def read(path: str,
format: Literal["text"] = "text",
user: Optional[Username] = None,
request_timeout: Optional[float] = None,
gzip: bool = False) -> strRead file content as a str.
Arguments:
path: Path to the fileuser: Run the operation as this userformat: Format of the file content—textby defaultrequest_timeout: Timeout for the request in secondsgzip: Use gzip compression for the request
Returns:
File content as a str
read
@overload
def read(path: str,
format: Literal["bytes"],
user: Optional[Username] = None,
request_timeout: Optional[float] = None,
gzip: bool = False) -> bytearrayRead file content as a bytearray.
Arguments:
path: Path to the fileuser: Run the operation as this userformat: Format of the file content—bytesrequest_timeout: Timeout for the request in secondsgzip: Use gzip compression for the request
Returns:
File content as a bytearray
read
@overload
def read(path: str,
format: Literal["stream"],
user: Optional[Username] = None,
request_timeout: Optional[float] = None,
gzip: bool = False) -> Iterator[bytes]Read file content as a Iterator[bytes].
Arguments:
path: Path to the fileuser: Run the operation as this userformat: Format of the file content—streamrequest_timeout: Timeout for the request in secondsgzip: Use gzip compression for the request
Returns:
File content as an Iterator[bytes]
write
def write(path: str,
data: Union[str, bytes, IO],
user: Optional[Username] = None,
request_timeout: Optional[float] = None,
gzip: bool = False,
use_octet_stream: bool = False) -> WriteInfoWrite content to a file on the path.
Writing to a file that doesn’t exist creates the file. Writing to a file that already exists overwrites the file. Writing to a file at path that doesn’t exist creates the necessary directories.
Arguments:
path: Path to the filedata: Data to write to the file, can be astr,bytes, orIO.user: Run the operation as this userrequest_timeout: Timeout for the request in secondsgzip: Use gzip compression for the requestuse_octet_stream: Upload usingapplication/octet-streaminstead ofmultipart/form-data. Defaults toFalse. Requires envd 0.5.7 or later — when not supported, the upload falls back tomultipart/form-data.
Returns:
Information about the written file
write_files
def write_files(files: List[WriteEntry],
user: Optional[Username] = None,
request_timeout: Optional[float] = None,
gzip: bool = False,
use_octet_stream: bool = False) -> List[WriteInfo]Writes a list of files to the filesystem.
When writing to a file that doesn’t exist, the file will get created. When writing to a file that already exists, the file will get overwritten. When writing to a file at path that doesn’t exist, the necessary directories will be created.
Arguments:
files: list of files to write asWriteEntryobjects, each containingpathanddatauser: Run the operation as this userrequest_timeout: Timeout for the requestgzip: Use gzip compression for the requestuse_octet_stream: Upload usingapplication/octet-streaminstead ofmultipart/form-data. Defaults toFalse. Requires envd 0.5.7 or later — when not supported, the upload falls back tomultipart/form-data.
Returns:
Information about the written files
list
def list(path: str,
depth: Optional[int] = 1,
user: Optional[Username] = None,
request_timeout: Optional[float] = None) -> List[EntryInfo]List entries in a directory.
Arguments:
path: Path to the directorydepth: Depth of the directory to listuser: Run the operation as this userrequest_timeout: Timeout for the request in seconds
Returns:
List of entries in the directory
exists
def exists(path: str,
user: Optional[Username] = None,
request_timeout: Optional[float] = None) -> boolCheck if a file or a directory exists.
Arguments:
path: Path to a file or a directoryuser: Run the operation as this userrequest_timeout: Timeout for the request in seconds
Returns:
True if the file or directory exists, False otherwise
get_info
def get_info(path: str,
user: Optional[Username] = None,
request_timeout: Optional[float] = None) -> EntryInfoGet information about a file or directory.
Arguments:
path: Path to a file or a directoryuser: Run the operation as this userrequest_timeout: Timeout for the request in seconds
Returns:
Information about the file or directory like name, type, and path
remove
def remove(path: str,
user: Optional[Username] = None,
request_timeout: Optional[float] = None) -> NoneRemove a file or a directory.
Arguments:
path: Path to a file or a directoryuser: Run the operation as this userrequest_timeout: Timeout for the request in seconds
rename
def rename(old_path: str,
new_path: str,
user: Optional[Username] = None,
request_timeout: Optional[float] = None) -> EntryInfoRename a file or directory.
Arguments:
old_path: Path to the file or directory to renamenew_path: New path to the file or directoryuser: Run the operation as this userrequest_timeout: Timeout for the request in seconds
Returns:
Information about the renamed file or directory
make_dir
def make_dir(path: str,
user: Optional[Username] = None,
request_timeout: Optional[float] = None) -> boolCreate a new directory and all directories along the way if needed on the specified path.
Arguments:
path: Path to a new directory. For example ‘/dirA/dirB’ when creating ‘dirB’.user: Run the operation as this userrequest_timeout: Timeout for the request in seconds
Returns:
True if the directory was created, False if the directory already exists
watch_dir
def watch_dir(path: str,
user: Optional[Username] = None,
request_timeout: Optional[float] = None,
recursive: bool = False) -> WatchHandleWatch directory for filesystem events.
Arguments:
path: Path to a directory to watchuser: Run the operation as this userrequest_timeout: Timeout for the request in secondsrecursive: Watch directory recursively
Returns:
WatchHandle object for stopping watching directory
SandboxPaginator
class SandboxPaginator(SandboxPaginatorBase)Paginator for listing sandboxes.
Example:
paginator = Sandbox.list()
while paginator.has_next:
sandboxes = paginator.next_items()
print(sandboxes)next_items
def next_items() -> List[SandboxInfo]Returns the next page of sandboxes.
Call this method only if has_next is True, otherwise it will raise an exception.
Returns:
List of sandboxes
SnapshotPaginator
class SnapshotPaginator(SnapshotPaginatorBase)Paginator for listing snapshots.
Example:
paginator = Sandbox.list_snapshots()
while paginator.has_next:
snapshots = paginator.next_items()
print(snapshots)next_items
def next_items() -> List[SnapshotInfo]Returns the next page of snapshots.
Call this method only if has_next is True, otherwise it will raise an exception.
Returns:
List of snapshots
Sandbox
class Sandbox(SandboxApi)E2B cloud sandbox is a secure and isolated cloud environment.
The sandbox allows you to:
- Access Linux OS
- Create, list, and delete files and directories
- Run commands
- Run isolated code
- Access the internet
Check docs here.
Use the Sandbox.create() to create a new sandbox.
Example:
from e2b import Sandbox
sandbox = Sandbox.create()files
@property
def files() -> FilesystemModule for interacting with the sandbox filesystem.
commands
@property
def commands() -> CommandsModule for running commands in the sandbox.
pty
@property
def pty() -> PtyModule for interacting with the sandbox pseudo-terminal.
git
@property
def git() -> GitModule for running git operations in the sandbox.
__init__
def __init__(**opts: Unpack[SandboxOpts]):deprecated: This constructor is deprecated
Use Sandbox.create() to create a new sandbox instead.
is_running
def is_running(request_timeout: Optional[float] = None) -> boolCheck if the sandbox is running.
Arguments:
request_timeout: Timeout for the request in seconds
Returns:
True if the sandbox is running, False otherwise
Example
sandbox = Sandbox.create()
sandbox.is_running() # Returns True
sandbox.kill()
sandbox.is_running() # Returns Falsecreate
@classmethod
def create(cls,
template: Optional[str] = None,
timeout: Optional[int] = None,
metadata: Optional[Dict[str, str]] = None,
envs: Optional[Dict[str, str]] = None,
secure: bool = True,
allow_internet_access: bool = True,
mcp: Optional[McpServer] = None,
network: Optional[SandboxNetworkOpts] = None,
lifecycle: Optional[SandboxLifecycle] = None,
volume_mounts: Optional[SandboxVolumeMount] = None,
**opts: Unpack[ApiParams]) -> SelfCreate a new sandbox.
By default, the sandbox is created from the default base sandbox template.
Arguments:
template: Sandbox template name or IDtimeout: Timeout for the sandbox in seconds, default to 300 seconds. The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.metadata: Custom metadata for the sandboxenvs: Custom environment variables for the sandboxsecure: Envd is secured with access token and cannot be used without it, defaults toTrue.allow_internet_access: Allow sandbox to access the internet, defaults toTrue. If set toFalse, it works the same as setting networkdeny_outto[0.0.0.0/0].mcp: MCP server to enable in the sandboxnetwork: Sandbox network configurationlifecycle: Sandbox lifecycle configuration —on_timeout:"kill"(default) or"pause";auto_resume:False(default) orTrue(only whenon_timeout="pause"). Example:{"on_timeout": "pause", "auto_resume": True}volume_mounts: Dictionary mapping mount paths to Volume instances or volume names
Returns:
A Sandbox instance for the new sandbox Use this method instead of using the constructor to create a new sandbox.
connect
@overload
def connect(timeout: Optional[int] = None, **opts: Unpack[ApiParams]) -> SelfConnect to a sandbox. If the sandbox is paused, it will be automatically resumed.
Sandbox must be either running or be paused.
With sandbox ID you can connect to the same sandbox from different places or environments (serverless functions, etc).
Arguments:
timeout: Timeout for the sandbox in seconds For running sandboxes, the timeout will update only if the new timeout is longer than the existing one.
Returns:
A running sandbox instance @example
sandbox = Sandbox.create()
sandbox.pause()
same_sandbox = sandbox.connect()
### connect
```python
@overload
@staticmethod
def connect(sandbox_id: str,
timeout: Optional[int] = None,
**opts: Unpack[ApiParams]) -> "Sandbox"Connect to a sandbox. If the sandbox is paused, it will be automatically resumed.
Sandbox must be either running or be paused.
With sandbox ID you can connect to the same sandbox from different places or environments (serverless functions, etc).
Arguments:
sandbox_id: Sandbox IDtimeout: Timeout for the sandbox in seconds. For running sandboxes, the timeout will update only if the new timeout is longer than the existing one.
Returns:
A running sandbox instance @example
sandbox = Sandbox.create()
Sandbox.pause(sandbox.sandbox_id)
same_sandbox = Sandbox.connect(sandbox.sandbox_id)connect
@class_method_variant("_cls_connect_sandbox")
def connect(timeout: Optional[int] = None, **opts: Unpack[ApiParams]) -> SelfConnect to a sandbox. If the sandbox is paused, it will be automatically resumed.
Sandbox must be either running or be paused.
With sandbox ID you can connect to the same sandbox from different places or environments (serverless functions, etc).
Arguments:
timeout: Timeout for the sandbox in seconds. For running sandboxes, the timeout will update only if the new timeout is longer than the existing one.
Returns:
A running sandbox instance @example
sandbox = Sandbox.create()
sandbox.pause()
same_sandbox = sandbox.connect()kill
@overload
def kill(**opts: Unpack[ApiParams]) -> boolKill the sandbox.
Returns:
True if the sandbox was killed, False if the sandbox was not found
kill
@overload
@staticmethod
def kill(sandbox_id: str, **opts: Unpack[ApiParams]) -> boolKill the sandbox specified by sandbox ID.
Arguments:
sandbox_id: Sandbox ID
Returns:
True if the sandbox was killed, False if the sandbox was not found
kill
@class_method_variant("_cls_kill")
def kill(**opts: Unpack[ApiParams]) -> boolKill the sandbox specified by sandbox ID.
Returns:
True if the sandbox was killed, False if the sandbox was not found
set_timeout
@overload
def set_timeout(timeout: int, **opts: Unpack[ApiParams]) -> NoneSet the timeout of the sandbox.
This method can extend or reduce the sandbox timeout set when creating the sandbox or from the last call to .set_timeout.
The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.
Arguments:
timeout: Timeout for the sandbox in seconds
set_timeout
@overload
@staticmethod
def set_timeout(sandbox_id: str, timeout: int,
**opts: Unpack[ApiParams]) -> NoneSet the timeout of the sandbox specified by sandbox ID.
This method can extend or reduce the sandbox timeout set when creating the sandbox or from the last call to .set_timeout.
The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.
Arguments:
sandbox_id: Sandbox IDtimeout: Timeout for the sandbox in seconds
set_timeout
@class_method_variant("_cls_set_timeout")
def set_timeout(timeout: int, **opts: Unpack[ApiParams]) -> NoneSet the timeout of the sandbox.
This method can extend or reduce the sandbox timeout set when creating the sandbox or from the last call to .set_timeout.
The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.
Arguments:
timeout: Timeout for the sandbox in seconds
get_info
@overload
def get_info(**opts: Unpack[ApiParams]) -> SandboxInfoGet sandbox information like sandbox ID, template, metadata, started at/end at date.
Returns:
Sandbox info
get_info
@overload
@staticmethod
def get_info(sandbox_id: str, **opts: Unpack[ApiParams]) -> SandboxInfoGet sandbox information like sandbox ID, template, metadata, started at/end at date.
Arguments:
sandbox_id: Sandbox ID
Returns:
Sandbox info
get_info
@class_method_variant("_cls_get_info")
def get_info(**opts: Unpack[ApiParams]) -> SandboxInfoGet sandbox information like sandbox ID, template, metadata, started at/end at date.
Returns:
Sandbox info
get_metrics
@overload
def get_metrics(start: Optional[datetime.datetime] = None,
end: Optional[datetime.datetime] = None,
**opts: Unpack[ApiParams]) -> List[SandboxMetrics]Get the metrics of the current sandbox.
Arguments:
start: Start time for the metrics, defaults to the start of the sandboxend: End time for the metrics, defaults to the current time
Returns:
List of sandbox metrics containing CPU, memory and disk usage information
get_metrics
@overload
@staticmethod
def get_metrics(sandbox_id: str,
start: Optional[datetime.datetime] = None,
end: Optional[datetime.datetime] = None,
**opts: Unpack[ApiParams]) -> List[SandboxMetrics]Get the metrics of the sandbox specified by sandbox ID.
Arguments:
sandbox_id: Sandbox IDstart: Start time for the metrics, defaults to the start of the sandboxend: End time for the metrics, defaults to the current time
Returns:
List of sandbox metrics containing CPU, memory and disk usage information
get_metrics
@class_method_variant("_cls_get_metrics")
def get_metrics(start: Optional[datetime.datetime] = None,
end: Optional[datetime.datetime] = None,
**opts: Unpack[ApiParams]) -> List[SandboxMetrics]Get the metrics of the sandbox specified by sandbox ID.
Arguments:
start: Start time for the metrics, defaults to the start of the sandboxend: End time for the metrics, defaults to the current time
Returns:
List of sandbox metrics containing CPU, memory and disk usage information
beta_create
@classmethod
def beta_create(cls,
template: Optional[str] = None,
timeout: Optional[int] = None,
auto_pause: bool = False,
metadata: Optional[Dict[str, str]] = None,
envs: Optional[Dict[str, str]] = None,
secure: bool = True,
allow_internet_access: bool = True,
mcp: Optional[McpServer] = None,
volume_mounts: Optional[SandboxVolumeMount] = None,
**opts: Unpack[ApiParams]) -> Self[BETA] This feature is in beta and may change in the future.
Create a new sandbox.
By default, the sandbox is created from the default base sandbox template.
Arguments:
template: Sandbox template name or IDtimeout: Timeout for the sandbox in seconds, default to 300 seconds. The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.auto_pause: Automatically pause the sandbox after the timeout expires. Defaults toFalse.metadata: Custom metadata for the sandboxenvs: Custom environment variables for the sandboxsecure: Envd is secured with access token and cannot be used without it, defaults toTrue.allow_internet_access: Allow sandbox to access the internet, defaults toTrue.mcp: MCP server to enable in the sandboxvolume_mounts: Dictionary mapping mount paths to Volume instances or volume names
Returns:
A Sandbox instance for the new sandbox Use this method instead of using the constructor to create a new sandbox.
pause
@overload
def pause(**opts: Unpack[ApiParams]) -> NonePause the sandbox.
pause
@overload
@staticmethod
def pause(sandbox_id: str, **opts: Unpack[ApiParams]) -> NonePause the sandbox specified by sandbox ID.
Arguments:
sandbox_id: Sandbox ID
pause
@class_method_variant("_cls_pause")
def pause(**opts: Unpack[ApiParams]) -> NonePause the sandbox.
Returns:
Sandbox ID that can be used to resume the sandbox
beta_pause
@class_method_variant("_cls_pause")
def beta_pause(**opts: Unpack[ApiParams]) -> None:deprecated: Use pause() instead.
create_snapshot
@overload
def create_snapshot(**opts: Unpack[ApiParams]) -> SnapshotInfoCreate a snapshot of the sandbox’s current state.
The sandbox will be paused while the snapshot is being created. The snapshot can be used to create new sandboxes with the same filesystem and state. Snapshots are persistent and survive sandbox deletion.
Use the returned snapshot_id with Sandbox.create(snapshot_id) to create a new sandbox from the snapshot.
Returns:
Snapshot information including the snapshot ID
create_snapshot
@overload
@staticmethod
def create_snapshot(sandbox_id: str,
**opts: Unpack[ApiParams]) -> SnapshotInfoCreate a snapshot from the sandbox specified by sandbox ID.
The sandbox will be paused while the snapshot is being created.
Arguments:
sandbox_id: Sandbox ID
Returns:
Snapshot information including the snapshot ID
create_snapshot
@class_method_variant("_cls_create_snapshot")
def create_snapshot(**opts: Unpack[ApiParams]) -> SnapshotInfoCreate a snapshot of the sandbox’s current state.
The sandbox will be paused while the snapshot is being created. The snapshot can be used to create new sandboxes with the same filesystem and state. Snapshots are persistent and survive sandbox deletion.
Use the returned snapshot_id with Sandbox.create(snapshot_id) to create a new sandbox from the snapshot.
Returns:
Snapshot information including the snapshot ID
list_snapshots
@overload
def list_snapshots(limit: Optional[int] = None,
next_token: Optional[str] = None,
**opts: Unpack[ApiParams]) -> SnapshotPaginatorList snapshots for this sandbox.
Arguments:
limit: Maximum number of snapshots to return per pagenext_token: Token for pagination
Returns:
Paginator for listing snapshots
list_snapshots
@overload
@staticmethod
def list_snapshots(sandbox_id: Optional[str] = None,
limit: Optional[int] = None,
next_token: Optional[str] = None,
**opts: Unpack[ApiParams]) -> SnapshotPaginatorList all snapshots.
Arguments:
sandbox_id: Filter snapshots by source sandbox IDlimit: Maximum number of snapshots to return per pagenext_token: Token for pagination
Returns:
Paginator for listing snapshots
list_snapshots
@class_method_variant("_cls_list_snapshots")
def list_snapshots(limit: Optional[int] = None,
next_token: Optional[str] = None,
**opts: Unpack[ApiParams]) -> SnapshotPaginatorList snapshots for this sandbox.
Arguments:
limit: Maximum number of snapshots to return per pagenext_token: Token for pagination
Returns:
Paginator for listing snapshots
delete_snapshot
@staticmethod
def delete_snapshot(snapshot_id: str, **opts: Unpack[ApiParams]) -> boolDelete a snapshot.
Arguments:
snapshot_id: Snapshot ID
Returns:
True if the snapshot was deleted, False if it was not found
get_mcp_token
def get_mcp_token() -> Optional[str]Get the MCP token for the sandbox.
Returns:
MCP token for the sandbox, or None if MCP is not enabled.
SandboxApi
class SandboxApi(SandboxBase)list
@staticmethod
def list(query: Optional[SandboxQuery] = None,
limit: Optional[int] = None,
next_token: Optional[str] = None,
**opts: Unpack[ApiParams]) -> SandboxPaginatorList all running sandboxes.
Arguments:
query: Filter the list of sandboxes by metadata or state, e.g.SandboxListQuery(metadata={"key": "value"})orSandboxListQuery(state=[SandboxState.RUNNING])limit: Maximum number of sandboxes to return per pagenext_token: Token for pagination
Returns:
List of running sandboxes