Sandbox & Execution¶

The sandbox is an isolated Python execution environment that the agent uses to run code. All data manipulation, model training, and chart generation happen inside the sandbox — your Python environment is never modified.

The default sandbox is backed by SRT (Sandbox Runtime), a lightweight subprocess-based executor. Each agent session gets its own sandbox instance; state (variables, DataFrames, trained models) persists across turns within a session and is cleared on /reset.

SandboxExecResult¶

Every code execution returns a SandboxExecResult:

from opendatasci import SandboxExecResult

result: SandboxExecResult = await sandbox.execute("print(1 + 1)")

result.success    # bool — True if no exception was raised
result.stdout     # str  — captured stdout
result.output     # Any  — the return value of the last expression (if any)
result.error      # str | None — exception traceback on failure
result.code       # str  — the code that was executed

TUI command allowlist¶

The agent can also run read-only shell commands via sandbox.execute_cli(). Only the commands in the allowlist below are permitted. Shell operators ||, ;, backticks, $(, >, and < are always rejected; | (pipe) and && are allowed.

Category	Commands
Directory	`ls`, `dir`, `pwd`, `tree`
File viewing	`cat`, `head`, `tail`, `file`, `stat`
Search	`grep`, `find`, `which`
Text processing	`cut`, `sort`, `uniq`, `wc`, `awk`, `sed`, `tr`, `strings`
Comparison / checksums	`diff`, `cmp`, `md5sum`, `sha256sum`, `shasum`
Structured / binary	`jq`, `xxd`, `od`
Output / info	`echo`, `printf`, `date`, `uname`, `printenv`, `env`
Archive inspection	`unzip`, `tar`, `zip`

Custom sandbox¶

Subclass BaseSandbox and BaseSandboxFactory to implement a remote or containerised sandbox:

from opendatasci.sandbox.base import BaseSandbox, BaseSandboxFactory, SandboxExecResult
from contextlib import asynccontextmanager
from pathlib import Path
from typing import AsyncIterator

class MyRemoteSandbox(BaseSandbox):
    async def execute(self, code: str) -> SandboxExecResult:
        # send code to a remote executor …
        ...

    async def execute_cli(self, command: str) -> SandboxExecResult: ...
    def reset(self) -> None: ...
    # close() is optional — override it to release external resources
    # (e.g. a remote connection); the default implementation is a no-op.

class MyRemoteSandboxFactory(BaseSandboxFactory):
    @asynccontextmanager
    async def create(self, workspace_path: Path | None = None) -> AsyncIterator[MyRemoteSandbox]:
        sandbox = MyRemoteSandbox()
        try:
            yield sandbox
        finally:
            await sandbox.close()

Pass the factory to Agent:

from opendatasci.agents.agents import Agent
from opendatasci import LocalWorkspace, OpenDataSciConfig

async with Agent(
    workspace=LocalWorkspace("data.csv"),
    sandbox_factory=MyRemoteSandboxFactory(),
    config=OpenDataSciConfig(),
) as agent:
    ...

Reference¶

SandboxExecResult `dataclass` ¶

SandboxExecResult(success: bool, output: Any = None, stdout: str = '', error: str | None = None, code: str = '')

Result of a single Python or TUI execution in the sandbox.

BaseSandbox ¶

Bases: ABC

Stateful code execution sandbox scoped to a single agent session.

Responsible for running Python code and TUI commands, capturing output, and preserving state (variables, results) across turns within the same conversation.

execute `abstractmethod` `async` ¶

execute(code: str) -> SandboxExecResult

Execute Python code and return the result.

execute_cli `abstractmethod` `async` ¶

execute_cli(command: str) -> SandboxExecResult

Execute a TUI command and return the result.

Implementations must validate command against validate_cli_command before running it and return a failed SandboxExecResult on violation.

reset `abstractmethod` ¶

reset() -> None

Clear all session state (history, variables, results).

close `async` ¶

close() -> None

Release external resources held by this sandbox.

The default implementation is a no-op. Override in sandboxes that manage external processes or connections (e.g. Docker containers).

BaseSandboxFactory ¶

Bases: ABC

Abstract factory that creates :class:BaseSandbox instances via async context managers.

Callers must always acquire a sandbox through :meth:create so that teardown is guaranteed regardless of how the calling code exits::

async with factory.create(workspace_path=path) as sandbox:
    result = await sandbox.execute(code)
# sandbox.close() has been called here

create `abstractmethod` ¶

create(workspace_path: Path | None = None) -> AbstractAsyncContextManager[BaseSandbox]

Return an async context manager that yields a fresh :class:BaseSandbox.

The sandbox is closed automatically when the context manager exits, whether by normal return, exception, or cancellation.

Parameters:

Name	Type	Description	Default
`workspace_path`	`Path \| None`	Optional filesystem root that the sandbox should treat as its working directory.	`None`

validate_cli_command ¶

validate_cli_command(command: str) -> str | None

Validate command against the TUI allowlist.

Returns an error string describing the first violation found, or None if the command is valid. This function is pure (no side effects) and can be called from tests directly.

SRTSandbox ¶

SRTSandbox(workspace_path: Path | None = None, *, command_timeout: int | None = None)

Bases: BaseSandbox

Session-scoped sandbox powered by Anthropic's Sandbox Runtime (SRT).

Executes Python snippets and allowlisted TUI commands in an OS-level sandbox — no Docker or remote container is required. Python state (variables, results) is preserved across calls within the same instance.

Each :meth:execute/:meth:execute_cli/:meth:reset call is serialized by a per-instance lock so overlapping calls cannot interleave their read-modify-write of the on-disk state.pkl.

Note: every :meth:execute spins up a fresh interpreter that unpickles the entire namespace, runs, and re-pickles it. This keeps executions hermetic but makes cost O(state) per call; a persistent-kernel runner would remove that overhead and is the natural next step for long interactive sessions.

SRTSandboxFactory ¶

SRTSandboxFactory(*, command_timeout: int | None = None)

Bases: BaseSandboxFactory

Factory that creates :class:SRTSandbox instances as async context managers.

Usage::

factory = SRTSandboxFactory()
async with factory.create(workspace_path=path) as sandbox:
    result = await sandbox.execute(code)
# sandbox is closed here