Skip to content

Sandbox

The Sandbox class represents a running cloud sandbox environment. You get a Sandbox instance from LunetteClient.create_sandbox().

Sandbox

Sandbox

Represents a running sandbox environment.

Provides async operations for interacting with remote sandbox instances managed by the Lunette backend service.

Example:

async with LunetteClient() as client:
    sandbox = await client.create_sandbox({"image": "python:3.11-slim"})

    result = await sandbox.aexec("echo 'hello'")
    print(result.stdout)

    await sandbox.aupload("./script.py", "/workspace/script.py")
    result = await sandbox.aexec("python /workspace/script.py")

    await sandbox.destroy()
Source code in lunette/sandbox.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
class Sandbox:
    """Represents a running sandbox environment.

    Provides async operations for interacting with remote sandbox instances
    managed by the Lunette backend service.

    Example:

    ```python
    async with LunetteClient() as client:
        sandbox = await client.create_sandbox({"image": "python:3.11-slim"})

        result = await sandbox.aexec("echo 'hello'")
        print(result.stdout)

        await sandbox.aupload("./script.py", "/workspace/script.py")
        result = await sandbox.aexec("python /workspace/script.py")

        await sandbox.destroy()
    ```
    """

    def __init__(
        self,
        client: "LunetteClient",
        sandbox_id: str,
        service: ComposeService,
    ):
        """Initialize sandbox instance.

        Args:
            client: LunetteClient instance for API communication
            sandbox_id: Stable sandbox ID (persists across restores)
            service: Docker Compose service specification
        """
        self.client = client
        self.sandbox_id = sandbox_id
        self.service = service
        self._destroyed = False

    async def aexec(self, cmd: str) -> ExecResult:
        """Execute a command in the sandbox asynchronously.

        Args:
            cmd: Command to execute

        Returns:
            ExecResult with stdout, stderr, and exit code

        Raises:
            SandboxDestroyedError: If sandbox has been destroyed
            httpx.HTTPError: For HTTP-related errors
        """
        if self._destroyed:
            raise SandboxDestroyedError(f"Sandbox {self.sandbox_id} has been destroyed")

        logger.debug(f"[{self.sandbox_id}] Executing command: {cmd}")

        # Make HTTP POST to /sandboxes/{sandbox_id}/exec
        response = await self.client._client.post(
            f"/sandboxes/{self.sandbox_id}/exec", json={"command": cmd}
        )

        response.raise_for_status()
        result = response.json()

        exec_result = ExecResult(
            stdout=result["stdout"],
            stderr=result["stderr"],
            exit_code=result["exit_code"],
            success=(result["exit_code"] == 0),
        )

        # Log the result
        if exec_result.success:
            logger.debug(
                f"[{self.sandbox_id}] Command succeeded (exit_code={exec_result.exit_code})"
            )
        else:
            logger.warning(
                f"[{self.sandbox_id}] Command failed (exit_code={exec_result.exit_code})"
            )
            if exec_result.stderr:
                logger.warning(
                    f"[{self.sandbox_id}] stderr: {exec_result.stderr[:500]}"
                )

        return exec_result

    async def aupload(
        self,
        local_path: str,
        remote_path: str,
    ) -> None:
        """Upload a file to the sandbox asynchronously.

        Args:
            local_path: Path to local file
            remote_path: Destination path in sandbox

        Raises:
            FileNotFoundError: If local file doesn't exist
            SandboxDestroyedError: If sandbox has been destroyed
            httpx.HTTPError: For HTTP-related errors
        """
        if self._destroyed:
            raise SandboxDestroyedError(f"Sandbox {self.sandbox_id} has been destroyed")

        logger.debug(
            f"[{self.sandbox_id}] Uploading file: {local_path} -> {remote_path}"
        )

        # Read local file as bytes
        try:
            with open(local_path, "rb") as f:
                content_bytes = f.read()
        except FileNotFoundError:
            raise FileNotFoundError(f"Local file not found: {local_path}")

        # Try UTF-8 first, fall back to base64 for binary data
        try:
            content_str = content_bytes.decode("utf-8")
            encoding = "utf-8"
        except UnicodeDecodeError:
            # Binary data - use base64
            content_str = base64.b64encode(content_bytes).decode("ascii")
            encoding = "base64"

        # POST to write endpoint
        response = await self.client._client.post(
            f"/sandboxes/{self.sandbox_id}/write",
            json={"path": remote_path, "content": content_str, "encoding": encoding},
        )

        response.raise_for_status()
        logger.info(f"[{self.sandbox_id}] Successfully uploaded file to {remote_path}")

    async def adownload(
        self,
        remote_path: str,
        local_path: str,
    ) -> None:
        """Download a file from the sandbox asynchronously.

        Args:
            remote_path: Path to file in sandbox
            local_path: Destination path on local filesystem

        Raises:
            FileNotFoundError: If remote file doesn't exist
            SandboxDestroyedError: If sandbox has been destroyed
            httpx.HTTPError: For HTTP-related errors
        """
        if self._destroyed:
            raise SandboxDestroyedError(f"Sandbox {self.sandbox_id} has been destroyed")

        logger.debug(
            f"[{self.sandbox_id}] Downloading file: {remote_path} -> {local_path}"
        )

        # GET from read endpoint
        try:
            response = await self.client._client.get(
                f"/sandboxes/{self.sandbox_id}/read", params={"path": remote_path}
            )
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                raise FileNotFoundError(f"Remote file not found: {remote_path}")
            raise

        result = response.json()
        content = result["content"]
        encoding = result.get("encoding", "utf-8")

        # Decode based on encoding
        if encoding == "base64":
            content_bytes = base64.b64decode(content)
        else:
            content_bytes = content.encode("utf-8")

        # Write to local file as bytes
        with open(local_path, "wb") as f:
            f.write(content_bytes)

        logger.info(f"[{self.sandbox_id}] Successfully downloaded file to {local_path}")

    async def destroy(self) -> None:
        """Destroy the sandbox and clean up resources.

        This is idempotent - calling multiple times is safe.
        """
        if self._destroyed:
            return

        # Backend doesn't have DELETE endpoint yet, so just mark as destroyed locally
        # TODO: When backend adds DELETE /sandboxes/{container_id}, call it here
        self._destroyed = True

    def __repr__(self) -> str:
        status = "destroyed" if self._destroyed else "active"
        return f"Sandbox(sandbox_id={self.sandbox_id}, status={status})"

__init__(client, sandbox_id, service)

Initialize sandbox instance.

Parameters:

Name Type Description Default
client LunetteClient

LunetteClient instance for API communication

 required
sandbox_id str

Stable sandbox ID (persists across restores)

 required
service ComposeService

Docker Compose service specification

 required
Source code in lunette/sandbox.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(
    self,
    client: "LunetteClient",
    sandbox_id: str,
    service: ComposeService,
):
    """Initialize sandbox instance.

    Args:
        client: LunetteClient instance for API communication
        sandbox_id: Stable sandbox ID (persists across restores)
        service: Docker Compose service specification
    """
    self.client = client
    self.sandbox_id = sandbox_id
    self.service = service
    self._destroyed = False

aexec(cmd) async

Execute a command in the sandbox asynchronously.

Parameters:

Name Type Description Default
cmd str

Command to execute

 required

Returns:

Type Description
ExecResult

ExecResult with stdout, stderr, and exit code

Raises:

Type Description
SandboxDestroyedError

If sandbox has been destroyed
HTTPError

For HTTP-related errors
Source code in lunette/sandbox.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
async def aexec(self, cmd: str) -> ExecResult:
    """Execute a command in the sandbox asynchronously.

    Args:
        cmd: Command to execute

    Returns:
        ExecResult with stdout, stderr, and exit code

    Raises:
        SandboxDestroyedError: If sandbox has been destroyed
        httpx.HTTPError: For HTTP-related errors
    """
    if self._destroyed:
        raise SandboxDestroyedError(f"Sandbox {self.sandbox_id} has been destroyed")

    logger.debug(f"[{self.sandbox_id}] Executing command: {cmd}")

    # Make HTTP POST to /sandboxes/{sandbox_id}/exec
    response = await self.client._client.post(
        f"/sandboxes/{self.sandbox_id}/exec", json={"command": cmd}
    )

    response.raise_for_status()
    result = response.json()

    exec_result = ExecResult(
        stdout=result["stdout"],
        stderr=result["stderr"],
        exit_code=result["exit_code"],
        success=(result["exit_code"] == 0),
    )

    # Log the result
    if exec_result.success:
        logger.debug(
            f"[{self.sandbox_id}] Command succeeded (exit_code={exec_result.exit_code})"
        )
    else:
        logger.warning(
            f"[{self.sandbox_id}] Command failed (exit_code={exec_result.exit_code})"
        )
        if exec_result.stderr:
            logger.warning(
                f"[{self.sandbox_id}] stderr: {exec_result.stderr[:500]}"
            )

    return exec_result

aupload(local_path, remote_path) async

Upload a file to the sandbox asynchronously.

Parameters:

Name Type Description Default
local_path str

Path to local file

 required
remote_path str

Destination path in sandbox

 required

Raises:

Type Description
FileNotFoundError

If local file doesn't exist
SandboxDestroyedError

If sandbox has been destroyed
HTTPError

For HTTP-related errors
Source code in lunette/sandbox.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
async def aupload(
    self,
    local_path: str,
    remote_path: str,
) -> None:
    """Upload a file to the sandbox asynchronously.

    Args:
        local_path: Path to local file
        remote_path: Destination path in sandbox

    Raises:
        FileNotFoundError: If local file doesn't exist
        SandboxDestroyedError: If sandbox has been destroyed
        httpx.HTTPError: For HTTP-related errors
    """
    if self._destroyed:
        raise SandboxDestroyedError(f"Sandbox {self.sandbox_id} has been destroyed")

    logger.debug(
        f"[{self.sandbox_id}] Uploading file: {local_path} -> {remote_path}"
    )

    # Read local file as bytes
    try:
        with open(local_path, "rb") as f:
            content_bytes = f.read()
    except FileNotFoundError:
        raise FileNotFoundError(f"Local file not found: {local_path}")

    # Try UTF-8 first, fall back to base64 for binary data
    try:
        content_str = content_bytes.decode("utf-8")
        encoding = "utf-8"
    except UnicodeDecodeError:
        # Binary data - use base64
        content_str = base64.b64encode(content_bytes).decode("ascii")
        encoding = "base64"

    # POST to write endpoint
    response = await self.client._client.post(
        f"/sandboxes/{self.sandbox_id}/write",
        json={"path": remote_path, "content": content_str, "encoding": encoding},
    )

    response.raise_for_status()
    logger.info(f"[{self.sandbox_id}] Successfully uploaded file to {remote_path}")

adownload(remote_path, local_path) async

Download a file from the sandbox asynchronously.

Parameters:

Name Type Description Default
remote_path str

Path to file in sandbox

 required
local_path str

Destination path on local filesystem

 required

Raises:

Type Description
FileNotFoundError

If remote file doesn't exist
SandboxDestroyedError

If sandbox has been destroyed
HTTPError

For HTTP-related errors
Source code in lunette/sandbox.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
async def adownload(
    self,
    remote_path: str,
    local_path: str,
) -> None:
    """Download a file from the sandbox asynchronously.

    Args:
        remote_path: Path to file in sandbox
        local_path: Destination path on local filesystem

    Raises:
        FileNotFoundError: If remote file doesn't exist
        SandboxDestroyedError: If sandbox has been destroyed
        httpx.HTTPError: For HTTP-related errors
    """
    if self._destroyed:
        raise SandboxDestroyedError(f"Sandbox {self.sandbox_id} has been destroyed")

    logger.debug(
        f"[{self.sandbox_id}] Downloading file: {remote_path} -> {local_path}"
    )

    # GET from read endpoint
    try:
        response = await self.client._client.get(
            f"/sandboxes/{self.sandbox_id}/read", params={"path": remote_path}
        )
        response.raise_for_status()
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            raise FileNotFoundError(f"Remote file not found: {remote_path}")
        raise

    result = response.json()
    content = result["content"]
    encoding = result.get("encoding", "utf-8")

    # Decode based on encoding
    if encoding == "base64":
        content_bytes = base64.b64decode(content)
    else:
        content_bytes = content.encode("utf-8")

    # Write to local file as bytes
    with open(local_path, "wb") as f:
        f.write(content_bytes)

    logger.info(f"[{self.sandbox_id}] Successfully downloaded file to {local_path}")

destroy() async

Destroy the sandbox and clean up resources.

This is idempotent - calling multiple times is safe.

Source code in lunette/sandbox.py 
229
230
231
232
233
234
235
236
237
238
239
async def destroy(self) -> None:
    """Destroy the sandbox and clean up resources.

    This is idempotent - calling multiple times is safe.
    """
    if self._destroyed:
        return

    # Backend doesn't have DELETE endpoint yet, so just mark as destroyed locally
    # TODO: When backend adds DELETE /sandboxes/{container_id}, call it here
    self._destroyed = True

ExecResult

ExecResult

Result from executing a command in a sandbox.

Source code in lunette/sandbox.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class ExecResult:
    """Result from executing a command in a sandbox."""

    def __init__(
        self,
        stdout: str,
        stderr: str,
        exit_code: int,
        success: bool,
    ):
        self.stdout = stdout
        self.stderr = stderr
        self.exit_code = exit_code
        self.success = success

    def __repr__(self) -> str:
        return f"ExecResult(exit_code={self.exit_code}, success={self.success})"

Exceptions

SandboxDestroyedError

Bases: Exception

Raised when attempting to use a destroyed sandbox.

Source code in lunette/sandbox.py 
17
18
19
20
class SandboxDestroyedError(Exception):
    """Raised when attempting to use a destroyed sandbox."""

    pass