Build a stateful, sandboxed code interpreter

This example demonstrates how to build a stateful code interpreter using a Lightning Sandbox and the lightning-sdk.

We'll run a long-lived Python "driver program" inside a sandbox that listens for code to execute, runs it, and returns the results. Because we're running in a fully isolated sandbox, we can safely use the "unsafe" exec() to execute the code. And because the driver keeps a single globals dict alive between calls, the interpreter is stateful: variables you define in one snippet are still there in the next.

Copy-paste friendly: every Python block below is part of one script. If you paste them in order into a file (e.g. interpreter_sandbox.py) and drop in your API key, you get a complete, runnable program -- the same one shipped in the SDK repo. The last section shows how to grab and run it straight from GitHub.

Imports and configuration

The SDK reads your credentials from the environment once at import time. Rather than rely on import order, we configure the key in-process with Sandbox.configure(...) (you can also set LIGHTNING_SANDBOX_API_KEY in your environment instead). See Authentication for how to create a teamspace-scoped key.

We also define a few module-level constants up front. The *_MARKER sentinels are explained in the next section -- both the driver and the client need them, so they live at the top.

import base64
import inspect
import json
import sys
import time

from lightning_sdk.sandbox import PtyCreateOpts, Sandbox

API_KEY = "sk-lit-..."  # or set LIGHTNING_SANDBOX_API_KEY in your environment

# Sentinel markers the driver wraps each JSON reply in. A PTY echoes our input
# and mixes stdout+stderr into one stream, so we can't assume "one JSON object
# per line". Instead the driver brackets each reply with these unlikely tokens
# and we slice them out of the accumulated terminal output. They are printable
# tokens, not ASCII control bytes (which a terminal would interpret).
RESULT_START = "<<<<<INTERP_RESULT_8f3a2b1c>>>>>"
RESULT_END = "<<<<<INTERP_END_8f3a2b1c>>>>>"
READY_MARKER = "<<<<<INTERP_READY_8f3a2b1c>>>>>"  # printed once when ready

The driver program

Our code interpreter uses a Python "driver program" that runs inside the sandbox. It reads one command at a time from its standard input (stdin), executes the code, and writes the captured stdout / stderr back out on its standard output (stdout).

The function re-declares its own imports and sentinel constants so that it is fully self-contained when we ship it into the sandbox as source text (see the next section).

def driver_program():
    import base64
    import json
    import sys
    from contextlib import redirect_stderr, redirect_stdout
    from io import StringIO

    RESULT_START = "<<<<<INTERP_RESULT_8f3a2b1c>>>>>"
    RESULT_END = "<<<<<INTERP_END_8f3a2b1c>>>>>"
    READY_MARKER = "<<<<<INTERP_READY_8f3a2b1c>>>>>"

    # When you `exec` code in Python, you can pass in a dictionary that defines
    # the global variables the code has access to. We reuse one dict for the
    # whole session, which is what makes the interpreter stateful.
    session_globals: dict = {}

    sys.stderr.write(READY_MARKER + "\n")
    sys.stderr.flush()

    while True:
        line = sys.stdin.readline()
        if not line:  # stdin closed -> shut the driver down
            break
        line = line.strip()
        if not line:
            continue

        # Each command line is URL-safe base64 of the code to run. Raw JSON on
        # stdin is unreliable through the PTY, which drops double quotes.
        try:
            code = base64.urlsafe_b64decode(line.encode()).decode()
        except Exception:
            continue  # ignore terminal echo / malformed noise

        # Capture the executed code's outputs.
        stdout_io, stderr_io = StringIO(), StringIO()
        with redirect_stdout(stdout_io), redirect_stderr(stderr_io):
            try:
                exec(code, session_globals)
            except Exception as e:
                print(f"Execution Error: {e}", file=sys.stderr)
        result = {
            "stdout": stdout_io.getvalue(),
            "stderr": stderr_io.getvalue(),
        }

        sys.stdout.write(RESULT_START + json.dumps(result) + RESULT_END)
        sys.stdout.flush()

Two details worth calling out, both driven by how we talk to the running process (explained below):

Sentinel markers. We bracket every JSON reply with RESULT_START / RESULT_END and announce readiness with READY_MARKER. Because we read the driver's output from a terminal stream that echoes our input and merges stdout with stderr, we can't assume "one JSON object per line" -- so we slice each reply out of the accumulated output using these unlikely tokens.
Base64 on the way in. The interactive transport drops lines that contain double quotes ("), so we can't feed raw JSON in. Instead the client base64-encodes the code (URL-safe alphabet, no quotes) and the driver decodes it. The reply on the way out is plain JSON, which is unaffected.

The interpreter client

We wrap the connection to the driver in a small SandboxInterpreter class. It ships the driver into the sandbox, opens a live PTY session to talk to it, and exposes a run_code(...) method.

Why a PTY? The driver needs a live, bidirectional connection: we write code to its stdin and read results from its stdout repeatedly, for the lifetime of the session. Lightning AI's run_command is request/response (no persistent stdin), so the right tool for a long-lived interactive process is a PTY.

class SandboxInterpreter:
    """A stateful code interpreter backed by a Lightning Sandbox PTY session."""

    def __init__(self, sandbox: Sandbox):
        self.sandbox = sandbox
        self._buf = bytearray()
        self._consumed = 0  # how many complete result blocks we've returned
        self.pty = None

    def start(self) -> None:
        # Ship the driver into the sandbox. We turn the function into source
        # text with inspect.getsource and append a call so running the file
        # boots the interpreter loop.
        driver_text = inspect.getsource(driver_program)
        driver_command = f"{driver_text}\n\ndriver_program()\n"
        self.sandbox.write_file("/root/driver.py", driver_command)

        # Open a live PTY and accumulate everything it emits into a buffer.
        self.pty = self.sandbox.process.create_pty(
            PtyCreateOpts(
                session_name="interpreter",
                cwd="/root",
                cols=200,
                rows=50,
                on_data=lambda chunk: self._buf.extend(chunk),
            )
        )
        self.pty.wait_for_connection(timeout=30)

        # `-u` keeps stdout/stderr unbuffered so replies arrive live.
        self.pty.send_input("python3 -u /root/driver.py\n")
        self._wait_for(READY_MARKER, timeout=30)

    def _wait_for(self, marker: str, timeout: float) -> None:
        deadline = time.time() + timeout
        while time.time() < deadline:
            if marker in self._buf.decode(errors="replace"):
                return
            time.sleep(0.1)
        raise TimeoutError("Timed out waiting for driver.")

    def _next_result(self, timeout: float) -> dict:
        """Block until the next complete result block is available, then parse it."""
        deadline = time.time() + timeout
        while time.time() < deadline:
            text = self._buf.decode(errors="replace")
            blocks, search_from = [], 0
            while True:
                start = text.find(RESULT_START, search_from)
                if start == -1:
                    break
                end = text.find(RESULT_END, start)
                if end == -1:
                    break  # reply still streaming in
                blocks.append(text[start + len(RESULT_START):end])
                search_from = end + len(RESULT_END)

            if len(blocks) > self._consumed:
                payload = blocks[self._consumed]
                self._consumed += 1
                return json.loads(payload)
            time.sleep(0.05)
        raise TimeoutError("Timed out waiting for code execution result.")

    def run_code(self, code: str, timeout: float = 30.0) -> dict:
        """Send a snippet to the interpreter, print its output, return the result."""
        payload = base64.urlsafe_b64encode(code.encode()).decode()
        self.pty.send_input(payload + "\n")
        result = self._next_result(timeout)

        print(result.get("stdout", ""), end="")
        if result.get("stderr"):
            # stderr in red
            print(
                "\033[91m" + result["stderr"] + "\033[0m",
                end="",
                file=sys.stderr,
            )
        return result

    def close(self) -> None:
        if self.pty is not None:
            self.pty.kill()

Creating the sandbox

Lightning AI's curated runtimes ship one language each, and python313 gives us exactly what we need. Once created, the sandbox stays alive (until its timeout) waiting for us to connect.

Sandbox.create(...) blocks until the sandbox is running and returns a handle we can use:

def create_sandbox(name: str, instance_type: str, timeout_ms: int) -> Sandbox:
    """Create a python313 sandbox."""
    return Sandbox.create(
        name=name,
        instance_type=instance_type,
        runtime="python313",
        timeout=timeout_ms,
    )

Putting it together

Finally, main() wires everything up: configure the SDK, create the sandbox, start the interpreter, run some code, and delete the sandbox at the end.

Now we can execute some code in the sandbox! The interpreter is stateful, so we can define variables and reuse them across calls, and we'll also see what happens when code raises an error.

def main() -> None:
    Sandbox.configure(api_key=API_KEY)

    sandbox = create_sandbox("code-interpreter", "cpu-1", 30 * 60 * 1000)
    print(f"Sandbox {sandbox.sandbox_id} is running; starting interpreter")

    interpreter = SandboxInterpreter(sandbox)
    interpreter.start()
    print("Interpreter ready.\n")

    # A first snippet.
    print("--- hello, world ---")
    interpreter.run_code("print('hello, world!')")  # hello, world!

    # State persists across calls.
    print("\n--- stateful variables ---")
    interpreter.run_code("x = 10")
    interpreter.run_code("y = 5")
    interpreter.run_code("result = x + y")
    interpreter.run_code("print(f'The result is: {result}')")  # The result is: 15

    # Errors are captured and reported.
    print("\n--- error handling ---")
    interpreter.run_code("print('Attempting to divide by zero...')")
    interpreter.run_code("1 / 0")  # Execution Error: division by zero

    print("\nDone.")

    # Remote sandboxes are not garbage-collected, so tear it down when done.
    interpreter.close()
    print(f"\nTerminating sandbox {sandbox.sandbox_id}")
    sandbox.delete()


if __name__ == "__main__":
    main()

Run it with your API key in the environment:

export LIGHTNING_SANDBOX_API_KEY=sk-lit-...
python interpreter_sandbox.py

The output should look something like

Creating sandbox 'code-interpreter' (cpu-1, python313)
Sandbox <sandbox-id> is running; starting interpreter
Interpreter ready.

--- hello, world ---
hello, world!

--- stateful variables ---
The result is: 15

--- error handling ---
Attempting to divide by zero...
Execution Error: division by zero      # (printed in red on stderr)

Done.

Terminating sandbox <sandbox-id>

Run the complete example

Paste all of the Python blocks above (in order) into a file, set your API key, and run it -- or skip the assembly and grab the ready-made script straight from the SDK repo:

git clone https://github.com/Lightning-AI/sdk.git
cd sdk/python/examples

pip install lightning-sdk
export LIGHTNING_SANDBOX_API_KEY=sk-lit-...   # or pass --api-key sk-lit-...

python interpreter_sandbox.py

Pass --help to see all available options:

python interpreter_sandbox.py --help

Was this page helpful?