Skip to content

API Reference

Comprehensive API documentation for LLM Sandbox.

Core Classes

SandboxSession

SandboxSession module-attribute

SandboxSession = create_session

ArtifactSandboxSession

ArtifactSandboxSession

ArtifactSandboxSession(
    backend: SandboxBackend = SandboxBackend.DOCKER,
    image: str | None = None,
    dockerfile: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    *,
    keep_template: bool = False,
    commit_container: bool = False,
    verbose: bool = False,
    runtime_configs: dict | None = None,
    workdir: str | None = "/sandbox",
    enable_plotting: bool = True,
    security_policy: SecurityPolicy | None = None,
    container_id: str | None = None,
    **kwargs: Any,
)

Sandbox session with artifact extraction capabilities.

Create a new artifact sandbox session.

The ArtifactSandboxSession provides a secure environment for running code that generates artifacts like plots, images, or other files. It supports multiple container backends (Docker, Kubernetes, Podman) and can capture and extract artifacts from the execution.

PARAMETER DESCRIPTION
backend

Container backend to use (Docker, Kubernetes or Podman)

TYPE: SandboxBackend DEFAULT: DOCKER

image

Container image to use (e.g., "vndee/sandbox-python-311-bullseye")

TYPE: str DEFAULT: None

dockerfile

Path to Dockerfile

TYPE: str DEFAULT: None

lang

Programming language (e.g., "python")

TYPE: str DEFAULT: PYTHON

keep_template

Whether to keep the container template

TYPE: bool DEFAULT: False

commit_container

Whether to commit container changes

TYPE: bool DEFAULT: False

verbose

Enable verbose logging

TYPE: bool DEFAULT: False

runtime_configs

Additional runtime configurations

TYPE: dict DEFAULT: None

workdir

Working directory inside the container

TYPE: str DEFAULT: '/sandbox'

enable_plotting

Whether to enable plot extraction

TYPE: bool DEFAULT: True

security_policy

Security policy to enforce

TYPE: SecurityPolicy DEFAULT: None

container_id

ID of existing container/pod to connect to

TYPE: str DEFAULT: None

**kwargs

Additional keyword arguments for specific backends (e.g., client for Podman)

TYPE: Any DEFAULT: {}

RAISES DESCRIPTION
MissingDependencyError

If the required dependency for the chosen backend is not installed

UnsupportedBackendError

If the chosen backend is not supported

Examples:

Connect to existing container for artifact generation:

from llm_sandbox import ArtifactSandboxSession, SandboxBackend
from pathlib import Path
import base64

# Connect to existing container
with ArtifactSandboxSession(
    container_id='existing-container-id',
    lang="python",
    verbose=True,
    backend=SandboxBackend.DOCKER
) as session:
    # Code that generates plots in existing environment
    code = '''
    import matplotlib.pyplot as plt
    import numpy as np

    # Generate and plot data
    x = np.linspace(0, 10, 100)
    y = np.sin(x)

    plt.figure()
    plt.plot(x, y)
    plt.title('Plot from Existing Container')
    plt.show()
    '''

    result = session.run(code)
    print(f"Captured {len(result.plots)} plots")

    # Save captured plots
    for i, plot in enumerate(result.plots):
        plot_path = Path("plots") / f"existing_{i + 1:06d}.{plot.format.value}"
        with plot_path.open("wb") as f:
            f.write(base64.b64decode(plot.content_base64))

Basic usage with Docker backend:

from llm_sandbox import ArtifactSandboxSession, SandboxBackend
from pathlib import Path
import base64

# Create plots directory
Path("plots/docker").mkdir(parents=True, exist_ok=True)

# Run code that generates plots
with ArtifactSandboxSession(
    lang="python",
    verbose=True,
    image="ghcr.io/vndee/sandbox-python-311-bullseye",
    backend=SandboxBackend.DOCKER
) as session:
    # Example code that generates matplotlib, seaborn, and plotly plots
    code = '''
    import matplotlib.pyplot as plt
    import numpy as np

    # Generate and plot data
    x = np.linspace(0, 10, 100)
    y = np.sin(x)

    plt.figure()
    plt.plot(x, y)
    plt.title('Simple Sine Wave')
    plt.show()
    '''

    result = session.run(code)
    print(f"Captured {len(result.plots)} plots")

    # Save captured plots
    for i, plot in enumerate(result.plots):
        plot_path = Path("plots/docker") / f"{i + 1:06d}.{plot.format.value}"
        with plot_path.open("wb") as f:
            f.write(base64.b64decode(plot.content_base64))
Using Podman backend:
from podman import PodmanClient

# Initialize Podman client
podman_client = PodmanClient(base_url="unix:///path/to/podman.sock")

with ArtifactSandboxSession(
    client=podman_client,  # Podman specific
    lang="python",
    verbose=True,
    image="ghcr.io/vndee/sandbox-python-311-bullseye",
    backend=SandboxBackend.PODMAN
) as session:
    result = session.run(code)

Using Kubernetes backend:

with ArtifactSandboxSession(
    lang="python",
    verbose=True,
    image="ghcr.io/vndee/sandbox-python-311-bullseye",
    backend=SandboxBackend.KUBERNETES
) as session:
    result = session.run(code)

Source code in llm_sandbox/session.py
def __init__(
    self,
    backend: SandboxBackend = SandboxBackend.DOCKER,
    image: str | None = None,
    dockerfile: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    *,
    keep_template: bool = False,
    commit_container: bool = False,
    verbose: bool = False,
    runtime_configs: dict | None = None,
    workdir: str | None = "/sandbox",
    enable_plotting: bool = True,
    security_policy: SecurityPolicy | None = None,
    container_id: str | None = None,
    **kwargs: Any,
) -> None:
    """Create a new artifact sandbox session.

    The ArtifactSandboxSession provides a secure environment for running code that generates artifacts
    like plots, images, or other files. It supports multiple container backends (Docker, Kubernetes, Podman)
    and can capture and extract artifacts from the execution.

    Args:
        backend (SandboxBackend): Container backend to use (Docker, Kubernetes or Podman)
        image (str): Container image to use (e.g., "vndee/sandbox-python-311-bullseye")
        dockerfile (str, optional): Path to Dockerfile
        lang (str): Programming language (e.g., "python")
        keep_template (bool, optional): Whether to keep the container template
        commit_container (bool, optional): Whether to commit container changes
        verbose (bool, optional): Enable verbose logging
        runtime_configs (dict, optional): Additional runtime configurations
        workdir (str, optional): Working directory inside the container
        enable_plotting (bool, optional): Whether to enable plot extraction
        security_policy (SecurityPolicy, optional): Security policy to enforce
        container_id (str, optional): ID of existing container/pod to connect to
        **kwargs: Additional keyword arguments for specific backends (e.g., client for Podman)

    Raises:
        MissingDependencyError: If the required dependency for the chosen backend is not installed
        UnsupportedBackendError: If the chosen backend is not supported

    Examples:
        Connect to existing container for artifact generation:
        ```python
        from llm_sandbox import ArtifactSandboxSession, SandboxBackend
        from pathlib import Path
        import base64

        # Connect to existing container
        with ArtifactSandboxSession(
            container_id='existing-container-id',
            lang="python",
            verbose=True,
            backend=SandboxBackend.DOCKER
        ) as session:
            # Code that generates plots in existing environment
            code = '''
            import matplotlib.pyplot as plt
            import numpy as np

            # Generate and plot data
            x = np.linspace(0, 10, 100)
            y = np.sin(x)

            plt.figure()
            plt.plot(x, y)
            plt.title('Plot from Existing Container')
            plt.show()
            '''

            result = session.run(code)
            print(f"Captured {len(result.plots)} plots")

            # Save captured plots
            for i, plot in enumerate(result.plots):
                plot_path = Path("plots") / f"existing_{i + 1:06d}.{plot.format.value}"
                with plot_path.open("wb") as f:
                    f.write(base64.b64decode(plot.content_base64))
        ```

        Basic usage with Docker backend:
        ```python
        from llm_sandbox import ArtifactSandboxSession, SandboxBackend
        from pathlib import Path
        import base64

        # Create plots directory
        Path("plots/docker").mkdir(parents=True, exist_ok=True)

        # Run code that generates plots
        with ArtifactSandboxSession(
            lang="python",
            verbose=True,
            image="ghcr.io/vndee/sandbox-python-311-bullseye",
            backend=SandboxBackend.DOCKER
        ) as session:
            # Example code that generates matplotlib, seaborn, and plotly plots
            code = '''
            import matplotlib.pyplot as plt
            import numpy as np

            # Generate and plot data
            x = np.linspace(0, 10, 100)
            y = np.sin(x)

            plt.figure()
            plt.plot(x, y)
            plt.title('Simple Sine Wave')
            plt.show()
            '''

            result = session.run(code)
            print(f"Captured {len(result.plots)} plots")

            # Save captured plots
            for i, plot in enumerate(result.plots):
                plot_path = Path("plots/docker") / f"{i + 1:06d}.{plot.format.value}"
                with plot_path.open("wb") as f:
                    f.write(base64.b64decode(plot.content_base64))
        ```
        Using Podman backend:
        ```python
        from podman import PodmanClient

        # Initialize Podman client
        podman_client = PodmanClient(base_url="unix:///path/to/podman.sock")

        with ArtifactSandboxSession(
            client=podman_client,  # Podman specific
            lang="python",
            verbose=True,
            image="ghcr.io/vndee/sandbox-python-311-bullseye",
            backend=SandboxBackend.PODMAN
        ) as session:
            result = session.run(code)
        ```

        Using Kubernetes backend:
        ```python
        with ArtifactSandboxSession(
            lang="python",
            verbose=True,
            image="ghcr.io/vndee/sandbox-python-311-bullseye",
            backend=SandboxBackend.KUBERNETES
        ) as session:
            result = session.run(code)
        ```

    """
    # Create the base session
    self._session: BaseSession = create_session(
        backend=backend,
        image=image,
        dockerfile=dockerfile,
        lang=lang,
        keep_template=keep_template,
        commit_container=commit_container,
        verbose=verbose,
        runtime_configs=runtime_configs,
        workdir=workdir,
        security_policy=security_policy,
        container_id=container_id,
        **kwargs,
    )

    self.enable_plotting = enable_plotting

Functions

__enter__

__enter__() -> ArtifactSandboxSession

Enter the context manager.

Source code in llm_sandbox/session.py
def __enter__(self) -> "ArtifactSandboxSession":
    """Enter the context manager."""
    self._session.__enter__()
    return self

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Exit the context manager.

Source code in llm_sandbox/session.py
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit the context manager."""
    return self._session.__exit__(exc_type, exc_val, exc_tb)

__getattr__

__getattr__(name: str) -> Any

Delegate any other attributes/methods to the underlying session.

Source code in llm_sandbox/session.py
def __getattr__(self, name: str) -> Any:
    """Delegate any other attributes/methods to the underlying session."""
    return getattr(self._session, name)

run

run(code: str, libraries: list | None = None) -> ExecutionResult

Run code in the sandbox session and extract any generated artifacts.

This method executes the provided code in an isolated environment and captures any generated artifacts (e.g., plots, figures). When plotting is enabled, it delegates to the language handler's run_with_artifacts method for language-specific artifact extraction.

PARAMETER DESCRIPTION
code

The code to execute. Can include plotting commands from matplotlib, seaborn, plotly, or other visualization libraries.

TYPE: str

libraries

Additional libraries to install before running the code. Defaults to None.

TYPE: list | None DEFAULT: None

RETURNS DESCRIPTION
ExecutionResult

An object containing: - exit_code (int): The exit code of the execution - stdout (str): Standard output from the code execution - stderr (str): Standard error from the code execution - plots (list[Plot]): List of captured plots, each containing: - content_base64 (str): Base64 encoded plot data - format (PlotFormat): Format of the plot (e.g., 'png', 'svg')

TYPE: ExecutionResult

RAISES DESCRIPTION
LanguageNotSupportPlotError

If the language does not support plot detection

Examples:

Basic plotting example:

with ArtifactSandboxSession(
    lang="python",
    verbose=True,
    image="ghcr.io/vndee/sandbox-python-311-bullseye"
) as session:
    code = '''
    import matplotlib.pyplot as plt
    import numpy as np

    x = np.linspace(0, 10, 100)
    y = np.sin(x)
    plt.plot(x, y)
    plt.title('Sine Wave')
    plt.show()
    '''
    result = session.run(code)
    print(f"Generated {len(result.plots)} plots")

Multiple plot types and libraries:

code = '''
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.express as px
import pandas as pd
import numpy as np

# Matplotlib plot
plt.figure(figsize=(10, 6))
x = np.linspace(0, 10, 100)
plt.plot(x, np.sin(x))
plt.title('Matplotlib: Sine Wave')
plt.show()

# Seaborn plot
data = pd.DataFrame({
    'x': np.random.randn(100),
    'y': np.random.randn(100)
})
sns.scatterplot(data=data, x='x', y='y')
plt.title('Seaborn: Scatter Plot')
plt.show()

# Plotly plot
fig = px.line(data, x='x', y='y', title='Plotly: Line Plot')
fig.show()
'''

result = session.run(code, libraries=['plotly'])

# Save the generated plots
for i, plot in enumerate(result.plots):
    with open(f'plot_{i}.{plot.format.value}', 'wb') as f:
        f.write(base64.b64decode(plot.content_base64))

Installing additional libraries:

code = '''
import torch
import torch.nn as nn
print(f"PyTorch version: {torch.__version__}")
'''

result = session.run(code, libraries=['torch'])
print(result.stdout)

Source code in llm_sandbox/session.py
def run(self, code: str, libraries: list | None = None) -> ExecutionResult:
    """Run code in the sandbox session and extract any generated artifacts.

    This method executes the provided code in an isolated environment and captures any
    generated artifacts (e.g., plots, figures). When plotting is enabled, it delegates
    to the language handler's run_with_artifacts method for language-specific artifact
    extraction.

    Args:
        code (str): The code to execute. Can include plotting commands from matplotlib,
                    seaborn, plotly, or other visualization libraries.
        libraries (list | None, optional): Additional libraries to install before running
                                            the code. Defaults to None.

    Returns:
        ExecutionResult: An object containing:
            - exit_code (int): The exit code of the execution
            - stdout (str): Standard output from the code execution
            - stderr (str): Standard error from the code execution
            - plots (list[Plot]): List of captured plots, each containing:
                - content_base64 (str): Base64 encoded plot data
                - format (PlotFormat): Format of the plot (e.g., 'png', 'svg')

    Raises:
        LanguageNotSupportPlotError: If the language does not support plot detection

    Examples:
        Basic plotting example:
        ```python
        with ArtifactSandboxSession(
            lang="python",
            verbose=True,
            image="ghcr.io/vndee/sandbox-python-311-bullseye"
        ) as session:
            code = '''
            import matplotlib.pyplot as plt
            import numpy as np

            x = np.linspace(0, 10, 100)
            y = np.sin(x)
            plt.plot(x, y)
            plt.title('Sine Wave')
            plt.show()
            '''
            result = session.run(code)
            print(f"Generated {len(result.plots)} plots")
        ```

        Multiple plot types and libraries:
        ```python
        code = '''
        import matplotlib.pyplot as plt
        import seaborn as sns
        import plotly.express as px
        import pandas as pd
        import numpy as np

        # Matplotlib plot
        plt.figure(figsize=(10, 6))
        x = np.linspace(0, 10, 100)
        plt.plot(x, np.sin(x))
        plt.title('Matplotlib: Sine Wave')
        plt.show()

        # Seaborn plot
        data = pd.DataFrame({
            'x': np.random.randn(100),
            'y': np.random.randn(100)
        })
        sns.scatterplot(data=data, x='x', y='y')
        plt.title('Seaborn: Scatter Plot')
        plt.show()

        # Plotly plot
        fig = px.line(data, x='x', y='y', title='Plotly: Line Plot')
        fig.show()
        '''

        result = session.run(code, libraries=['plotly'])

        # Save the generated plots
        for i, plot in enumerate(result.plots):
            with open(f'plot_{i}.{plot.format.value}', 'wb') as f:
                f.write(base64.b64decode(plot.content_base64))
        ```

        Installing additional libraries:
        ```python
        code = '''
        import torch
        import torch.nn as nn
        print(f"PyTorch version: {torch.__version__}")
        '''

        result = session.run(code, libraries=['torch'])
        print(result.stdout)
        ```

    """
    # Check if plotting is enabled and language supports it
    if self.enable_plotting and not self._session.language_handler.is_support_plot_detection:
        raise LanguageNotSupportPlotError(self._session.language_handler.name)

    # Delegate to language handler for language-specific artifact extraction
    result, plots = self._session.language_handler.run_with_artifacts(
        container=self._session,  # type: ignore[arg-type]
        code=code,
        libraries=libraries,
        enable_plotting=self.enable_plotting,
        output_dir="/tmp/sandbox_plots",
    )

    return ExecutionResult(
        exit_code=result.exit_code,
        stdout=result.stdout,
        stderr=result.stderr,
        plots=plots,
    )

Data Classes

ConsoleOutput

ConsoleOutput dataclass

ConsoleOutput(exit_code: int = 0, stderr: str = '', stdout: str = '')

Represents the standard output and standard error from code execution or a command.

ATTRIBUTE DESCRIPTION
exit_code

The exit code of the executed code or command. 0 typically indicates success.

TYPE: int

stderr

The content written to the standard error stream.

TYPE: str

stdout

The content written to the standard output stream.

TYPE: str

Functions

success

success() -> bool

Check if the execution was successful (exit code is 0).

RETURNS DESCRIPTION
bool

True if exit_code is 0, False otherwise.

TYPE: bool

Source code in llm_sandbox/data.py
def success(self) -> bool:
    r"""Check if the execution was successful (exit code is 0).

    Returns:
        bool: True if `exit_code` is 0, False otherwise.

    """
    return not self.exit_code

text

text() -> str

Get the text representation of the console output (stdout).

.. deprecated:: 0.1.0 The text property is deprecated and will be removed in a future version. Use the stdout attribute directly instead.

RETURNS DESCRIPTION
str

The content of the standard output stream.

TYPE: str

Source code in llm_sandbox/data.py
def text(self) -> str:
    r"""Get the text representation of the console output (stdout).

    .. deprecated:: 0.1.0
        The `text` property is deprecated and will be removed in a future version.
        Use the `stdout` attribute directly instead.

    Returns:
        str: The content of the standard output stream.

    """
    warnings.warn(
        "The 'text' property is deprecated and will be removed in a future version. "
        "Use 'stdout' attribute directly instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.stdout

ExecutionResult

ExecutionResult dataclass

ExecutionResult(
    exit_code: int = 0, stderr: str = "", stdout: str = "", plots: list[PlotOutput] = list()
)

Bases: ConsoleOutput

Represents the comprehensive result of code execution within a sandbox session.

This class extends ConsoleOutput to include any plots or other file artifacts that were generated and captured during the execution.

ATTRIBUTE DESCRIPTION
plots

A list of PlotOutput objects, each representing a captured plot or visual artifact. Defaults to an empty list.

TYPE: list[PlotOutput]

PlotOutput

PlotOutput dataclass

PlotOutput(
    format: FileType,
    content_base64: str,
    width: int | None = None,
    height: int | None = None,
    dpi: int | None = None,
)

Represents a plot, chart, or other visual artifact output from code execution.

ATTRIBUTE DESCRIPTION
format

The format of the plot (e.g., PNG, SVG, PDF).

TYPE: FileType

content_base64

The raw content of the plot, base64 encoded.

TYPE: str

width

The width of the plot in pixels. Defaults to None.

TYPE: int | None

height

The height of the plot in pixels. Defaults to None.

TYPE: int | None

dpi

The dots per inch (resolution) of the plot. Defaults to None.

TYPE: int | None


Security Classes

SecurityPolicy

SecurityPolicy

Bases: BaseModel

A security policy.

Functions

add_pattern

add_pattern(pattern: SecurityPattern) -> None

Add a security pattern to the policy.

Source code in llm_sandbox/security.py
def add_pattern(self, pattern: SecurityPattern) -> None:
    """Add a security pattern to the policy."""
    if self.patterns is None:
        self.patterns = []
    self.patterns.append(pattern)

add_restricted_module

add_restricted_module(module: RestrictedModule) -> None

Add a restricted module to the policy.

Source code in llm_sandbox/security.py
def add_restricted_module(self, module: RestrictedModule) -> None:
    """Add a restricted module to the policy."""
    if self.restricted_modules is None:
        self.restricted_modules = []
    self.restricted_modules.append(module)

SecurityPattern

SecurityPattern

Bases: BaseModel

A security pattern.

Functions

validate_pattern classmethod

validate_pattern(v: str) -> str

Validate that the pattern is a valid regex pattern.

Source code in llm_sandbox/security.py
@field_validator("pattern")
@classmethod
def validate_pattern(cls, v: str) -> str:
    """Validate that the pattern is a valid regex pattern."""
    try:
        re.compile(v)
    except re.error as e:
        raise InvalidRegexPatternError(v) from e
    return v

RestrictedModule

RestrictedModule

Bases: BaseModel

A dangerous module.

SecurityIssueSeverity

SecurityIssueSeverity

Bases: IntEnum

Severity of a security issue.


Enumerations

SandboxBackend

SandboxBackend

Bases: StrEnum

Enumeration of supported sandbox backend technologies.

Each value represents a different containerization or virtualization technology that can be used to isolate code execution.

SupportedLanguage

SupportedLanguage

Bases: StrEnum

Dataclass defining constants for supported programming languages.

Each attribute represents a language identifier string used by the sandbox to select appropriate language handlers and container images.

FileType

FileType

Bases: Enum

Enumeration of file types supported by artifact extractors.

This enum lists common file formats that can be generated by code running in the sandbox and subsequently extracted, such as images, data files, and documents.


Functions

create_session

create_session

create_session(
    backend: SandboxBackend = SandboxBackend.DOCKER, *args: Any, **kwargs: Any
) -> BaseSession

Create a new sandbox session for executing code in an isolated environment.

This function creates a sandbox session that supports multiple programming languages and provides features like package installation, file operations, and secure code execution. For backward compatibility, we also keep a SandboxSession alias for this function.

PARAMETER DESCRIPTION
backend

Container backend to use. Options: - SandboxBackend.DOCKER (default) - SandboxBackend.KUBERNETES - SandboxBackend.PODMAN - SandboxBackend.MICROMAMBA

TYPE: SandboxBackend DEFAULT: DOCKER

*args

Additional positional arguments passed to the session constructor

TYPE: Any DEFAULT: ()

**kwargs

Additional keyword arguments passed to the session constructor. Common options include: - lang (str): Programming language ("python", "java", "javascript", "cpp", "go") - verbose (bool): Enable verbose logging - keep_template (bool): Keep the container template - image (str): Custom container image to use - container_id (str): ID of existing container/pod to connect to

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
Session

A sandbox session instance for the specified backend

TYPE: BaseSession

RAISES DESCRIPTION
MissingDependencyError

If the required dependency for the chosen backend is not installed

UnsupportedBackendError

If the chosen backend is not supported

Examples:

Connect to existing Docker container:

# Assumes you have a running container with ID 'abc123...'
with SandboxSession(container_id='abc123def456', lang="python") as session:
    result = session.run("print('Hello from existing container!')")
    print(result.stdout)

    # Install libraries in existing container
    session.install(["numpy"])
    result = session.run("import numpy as np; print(np.random.rand())")

    # Execute commands
    result = session.execute_command("ls -la")

    # Copy files
    session.copy_to_runtime("local_file.py", "/container/path/file.py")

Connect to existing Kubernetes pod:

# Assumes you have a running pod with name 'my-pod-abc123'
with SandboxSession(
    backend=SandboxBackend.KUBERNETES,
    container_id='my-pod-abc123',  # pod name
    lang="python"
) as session:
    result = session.run("print('Hello from existing pod!')")

Connect to existing Podman container:

from podman import PodmanClient

client = PodmanClient()
with SandboxSession(
    backend=SandboxBackend.PODMAN,
    client=client,
    container_id='podman-container-id',
    lang="python"
) as session:
    result = session.run("print('Hello from existing Podman container!')")

Python session with package installation:

with SandboxSession(lang="python", keep_template=True, verbose=True) as session:
    # Basic code execution
    result = session.run("print('Hello, World!')")
    print(result.stdout)  # Output: Hello, World!

    # Install and use packages
    result = session.run(
        "import numpy as np\nprint(np.random.rand())",
        libraries=["numpy"]
    )

    # Install additional packages during session
    session.install(["pandas"])
    result = session.run("import pandas as pd\nprint(pd.__version__)")

    # Copy files to runtime
    session.copy_to_runtime("README.md", "/sandbox/data.csv")

Java session:

with SandboxSession(lang="java", keep_template=True, verbose=True) as session:
    result = session.run(\"\"\"
        public class Main {
            public static void main(String[] args) {
                System.out.println("Hello, World!");
            }
        }
    \"\"\")

JavaScript session with npm packages:

with SandboxSession(lang="javascript", keep_template=True, verbose=True) as session:
    # Basic code execution
    result = session.run("console.log('Hello, World!')")

    # Using npm packages
    result = session.run(\"\"\"
        const axios = require('axios');
        axios.get('https://jsonplaceholder.typicode.com/posts/1')
            .then(response => console.log(response.data));
    \"\"\", libraries=["axios"])

C++ session:

with SandboxSession(lang="cpp", keep_template=True, verbose=True) as session:
    result = session.run(\"\"\"
        #include <iostream>
        #include <vector>
        #include <algorithm>
        int main() {
            std::vector<int> v = {1, 2, 3, 4, 5};
            std::reverse(v.begin(), v.end());
            for (int i : v) {
                std::cout << i << " ";
            }
            std::cout << std::endl;
            return 0;
        }
    \"\"\", libraries=["libstdc++"])

Go session with external packages:

with SandboxSession(lang="go", keep_template=True, verbose=True) as session:
    result = session.run(\"\"\"
        package main
        import (
            "fmt"
            "github.com/spyzhov/ajson"
        )
        func main() {
            json := []byte(`{"price": 100}`)
            root, _ := ajson.Unmarshal(json)
            nodes, _ := root.JSONPath("$..price")
            for _, node := range nodes {
                node.SetNumeric(node.MustNumeric() * 1.25)
            }
            result, _ := ajson.Marshal(root)
            fmt.Printf("%s", result)
        }
    \"\"\", libraries=["github.com/spyzhov/ajson"])

Source code in llm_sandbox/session.py
def create_session(
    backend: SandboxBackend = SandboxBackend.DOCKER,
    *args: Any,
    **kwargs: Any,
) -> BaseSession:
    r"""Create a new sandbox session for executing code in an isolated environment.

    This function creates a sandbox session that supports multiple programming languages
    and provides features like package installation, file operations, and secure code execution.
    For backward compatibility, we also keep a `SandboxSession` alias for this function.

    Args:
        backend (SandboxBackend): Container backend to use. Options:
            - SandboxBackend.DOCKER (default)
            - SandboxBackend.KUBERNETES
            - SandboxBackend.PODMAN
            - SandboxBackend.MICROMAMBA
        *args: Additional positional arguments passed to the session constructor
        **kwargs: Additional keyword arguments passed to the session constructor.
                Common options include:
                    - lang (str): Programming language ("python", "java", "javascript", "cpp", "go")
                    - verbose (bool): Enable verbose logging
                    - keep_template (bool): Keep the container template
                    - image (str): Custom container image to use
                    - container_id (str): ID of existing container/pod to connect to

    Returns:
        Session: A sandbox session instance for the specified backend

    Raises:
        MissingDependencyError: If the required dependency for the chosen backend is not installed
        UnsupportedBackendError: If the chosen backend is not supported

    Examples:
        Connect to existing Docker container:
        ```python
        # Assumes you have a running container with ID 'abc123...'
        with SandboxSession(container_id='abc123def456', lang="python") as session:
            result = session.run("print('Hello from existing container!')")
            print(result.stdout)

            # Install libraries in existing container
            session.install(["numpy"])
            result = session.run("import numpy as np; print(np.random.rand())")

            # Execute commands
            result = session.execute_command("ls -la")

            # Copy files
            session.copy_to_runtime("local_file.py", "/container/path/file.py")
        ```

        Connect to existing Kubernetes pod:
        ```python
        # Assumes you have a running pod with name 'my-pod-abc123'
        with SandboxSession(
            backend=SandboxBackend.KUBERNETES,
            container_id='my-pod-abc123',  # pod name
            lang="python"
        ) as session:
            result = session.run("print('Hello from existing pod!')")
        ```

        Connect to existing Podman container:
        ```python
        from podman import PodmanClient

        client = PodmanClient()
        with SandboxSession(
            backend=SandboxBackend.PODMAN,
            client=client,
            container_id='podman-container-id',
            lang="python"
        ) as session:
            result = session.run("print('Hello from existing Podman container!')")
        ```

        Python session with package installation:
        ```python
        with SandboxSession(lang="python", keep_template=True, verbose=True) as session:
            # Basic code execution
            result = session.run("print('Hello, World!')")
            print(result.stdout)  # Output: Hello, World!

            # Install and use packages
            result = session.run(
                "import numpy as np\nprint(np.random.rand())",
                libraries=["numpy"]
            )

            # Install additional packages during session
            session.install(["pandas"])
            result = session.run("import pandas as pd\nprint(pd.__version__)")

            # Copy files to runtime
            session.copy_to_runtime("README.md", "/sandbox/data.csv")
        ```

        Java session:
        ```python
        with SandboxSession(lang="java", keep_template=True, verbose=True) as session:
            result = session.run(\"\"\"
                public class Main {
                    public static void main(String[] args) {
                        System.out.println("Hello, World!");
                    }
                }
            \"\"\")
        ```

        JavaScript session with npm packages:
        ```python
        with SandboxSession(lang="javascript", keep_template=True, verbose=True) as session:
            # Basic code execution
            result = session.run("console.log('Hello, World!')")

            # Using npm packages
            result = session.run(\"\"\"
                const axios = require('axios');
                axios.get('https://jsonplaceholder.typicode.com/posts/1')
                    .then(response => console.log(response.data));
            \"\"\", libraries=["axios"])
        ```

        C++ session:
        ```python
        with SandboxSession(lang="cpp", keep_template=True, verbose=True) as session:
            result = session.run(\"\"\"
                #include <iostream>
                #include <vector>
                #include <algorithm>
                int main() {
                    std::vector<int> v = {1, 2, 3, 4, 5};
                    std::reverse(v.begin(), v.end());
                    for (int i : v) {
                        std::cout << i << " ";
                    }
                    std::cout << std::endl;
                    return 0;
                }
            \"\"\", libraries=["libstdc++"])
        ```

        Go session with external packages:
        ```python
        with SandboxSession(lang="go", keep_template=True, verbose=True) as session:
            result = session.run(\"\"\"
                package main
                import (
                    "fmt"
                    "github.com/spyzhov/ajson"
                )
                func main() {
                    json := []byte(`{"price": 100}`)
                    root, _ := ajson.Unmarshal(json)
                    nodes, _ := root.JSONPath("$..price")
                    for _, node := range nodes {
                        node.SetNumeric(node.MustNumeric() * 1.25)
                    }
                    result, _ := ajson.Marshal(root)
                    fmt.Printf("%s", result)
                }
            \"\"\", libraries=["github.com/spyzhov/ajson"])
        ```

    """
    # Check if required dependency is installed
    _check_dependency(backend)

    # Create the appropriate session based on backend
    match backend:
        case SandboxBackend.DOCKER:
            from .docker import SandboxDockerSession

            return SandboxDockerSession(*args, **kwargs)
        case SandboxBackend.KUBERNETES:
            from .kubernetes import SandboxKubernetesSession

            return SandboxKubernetesSession(*args, **kwargs)
        case SandboxBackend.PODMAN:
            from .podman import SandboxPodmanSession

            return SandboxPodmanSession(*args, **kwargs)
        case SandboxBackend.MICROMAMBA:
            from .micromamba import MicromambaSession

            return MicromambaSession(*args, **kwargs)
        case _:
            raise UnsupportedBackendError(backend=backend)

Exceptions

The library defines a base exception llm_sandbox.exceptions.SandboxError and various specific exceptions that inherit from it. Please refer to the llm_sandbox.exceptions module for a complete list.

SandboxTimeoutError

SandboxTimeoutError

SandboxTimeoutError(message: str, timeout_duration: float | None = None)

Bases: SandboxError

Raised when an operation times out.

Initialize the TimeoutError.

Source code in llm_sandbox/exceptions.py
def __init__(self, message: str, timeout_duration: float | None = None) -> None:
    """Initialize the TimeoutError."""
    super().__init__(message)
    self.timeout_duration = timeout_duration

Functions

Common exceptions include: - ContainerError - SecurityError - ResourceError - ValidationError - LanguageNotSupportedError - ImageNotFoundError - SandboxTimeoutError - Raised when operations exceed configured timeout limits


Language Handlers

AbstractLanguageHandler

AbstractLanguageHandler

AbstractLanguageHandler(logger: Logger | None = None)

Bases: ABC

Abstract base class for language-specific handlers.

Initialize the language handler.

Source code in llm_sandbox/language_handlers/base.py
def __init__(self, logger: logging.Logger | None = None) -> None:
    """Initialize the language handler."""
    self.config: LanguageConfig
    self.logger: logging.Logger = logger or logging.getLogger(__name__)

Attributes

file_extension property

file_extension: str

Get file extension for language.

is_support_library_installation property

is_support_library_installation: bool

Get if the language supports library installation.

is_support_plot_detection property

is_support_plot_detection: bool

Get if the language supports plot detection.

name property

name: str

Get name of the language.

supported_plot_libraries property

supported_plot_libraries: list[PlotLibrary]

Get supported plotting libraries.

Functions

extract_plots

extract_plots(container: ContainerProtocol, output_dir: str) -> list[PlotOutput]

Extract plots from the code.

Subclasses should override this method to provide custom plot extraction logic.

PARAMETER DESCRIPTION
container

The container protocol instance to run code in

TYPE: ContainerProtocol

output_dir

Directory where plots should be saved

TYPE: str

RETURNS DESCRIPTION
list[PlotOutput]

list[PlotOutput]: List of plot outputs

Source code in llm_sandbox/language_handlers/base.py
def extract_plots(self, container: "ContainerProtocol", output_dir: str) -> list[PlotOutput]:  # noqa: ARG002
    """Extract plots from the code.

    Subclasses should override this method to provide custom plot extraction logic.

    Args:
        container: The container protocol instance to run code in
        output_dir: Directory where plots should be saved

    Returns:
        list[PlotOutput]: List of plot outputs

    """
    return []

filter_comments

filter_comments(code: str) -> str

Filter out comments from code in a language-specific way.

PARAMETER DESCRIPTION
code

The code to filter comments from.

TYPE: str

RETURNS DESCRIPTION
str

The code with comments removed.

TYPE: str

Source code in llm_sandbox/language_handlers/base.py
def filter_comments(self, code: str) -> str:
    """Filter out comments from code in a language-specific way.

    Args:
        code (str): The code to filter comments from.

    Returns:
        str: The code with comments removed.

    """
    # First remove multi-line comments
    code = re.sub(self.get_multiline_comment_patterns(), "", code)

    # Then handle single-line comments
    filtered_lines = []
    for line in code.split("\n"):
        # Remove inline comments
        clean_line = re.sub(self.get_inline_comment_patterns(), "", line)
        # Keep the line if it has non-whitespace content
        if clean_line.strip():
            filtered_lines.append(clean_line)
        else:
            # Preserve empty lines for readability
            filtered_lines.append("")
    return "\n".join(filtered_lines)

get_execution_commands

get_execution_commands(code_file: str) -> list[str]

Get commands to execute code file.

Source code in llm_sandbox/language_handlers/base.py
def get_execution_commands(self, code_file: str) -> list[str]:
    """Get commands to execute code file."""
    if not self.config.execution_commands:
        raise CommandFailedError(self.config.name, 1, "No execution commands found")
    return [command.format(file=code_file) for command in self.config.execution_commands]

get_import_patterns abstractmethod

get_import_patterns(module: str) -> str

Get the regex patterns for import statements.

Source code in llm_sandbox/language_handlers/base.py
@abstractmethod
def get_import_patterns(self, module: str) -> str:
    """Get the regex patterns for import statements."""

get_inline_comment_patterns abstractmethod staticmethod

get_inline_comment_patterns() -> str

Get the regex for inline comment patterns.

Source code in llm_sandbox/language_handlers/base.py
@staticmethod
@abstractmethod
def get_inline_comment_patterns() -> str:
    """Get the regex for inline comment patterns."""

get_library_installation_command

get_library_installation_command(library: str) -> str

Get command to install library.

Source code in llm_sandbox/language_handlers/base.py
def get_library_installation_command(self, library: str) -> str:
    """Get command to install library."""
    if not self.config.package_manager:
        raise PackageManagerError(self.config.name)
    return f"{self.config.package_manager} {library}"

get_multiline_comment_patterns abstractmethod staticmethod

get_multiline_comment_patterns() -> str

Get the regex patterns for multiline comment.

Source code in llm_sandbox/language_handlers/base.py
@staticmethod
@abstractmethod
def get_multiline_comment_patterns() -> str:
    """Get the regex patterns for multiline comment."""

inject_plot_detection_code

inject_plot_detection_code(code: str) -> str

Inject code to detect and capture plots.

Subclasses should override this method to provide custom plot detection code.

PARAMETER DESCRIPTION
code

The code to inject plot detection code into.

TYPE: str

RETURNS DESCRIPTION
str

The code with plot detection code injected.

Source code in llm_sandbox/language_handlers/base.py
def inject_plot_detection_code(self, code: str) -> str:
    """Inject code to detect and capture plots.

    Subclasses should override this method to provide custom plot detection code.

    Args:
        code: The code to inject plot detection code into.

    Returns:
        The code with plot detection code injected.

    """
    return code

run_with_artifacts

run_with_artifacts(
    container: ContainerProtocol,
    code: str,
    libraries: list | None = None,
    enable_plotting: bool = True,
    output_dir: str = "/tmp/sandbox_plots",
) -> tuple[Any, list[PlotOutput]]

Run code and extract artifacts (plots) in a language-specific manner.

This method provides a language-specific implementation for running code with artifact extraction. Languages that support plot detection can override this method to provide custom artifact extraction logic.

PARAMETER DESCRIPTION
container

The container protocol instance to run code in

TYPE: ContainerProtocol

code

The code to execute

TYPE: str

libraries

Optional list of libraries to install before running

TYPE: list | None DEFAULT: None

enable_plotting

Whether to enable plot detection and extraction

TYPE: bool DEFAULT: True

output_dir

Directory where plots should be saved

TYPE: str DEFAULT: '/tmp/sandbox_plots'

RETURNS DESCRIPTION
tuple

(execution_result, list_of_plots)

TYPE: tuple[Any, list[PlotOutput]]

Source code in llm_sandbox/language_handlers/base.py
def run_with_artifacts(
    self,
    container: "ContainerProtocol",
    code: str,
    libraries: list | None = None,
    enable_plotting: bool = True,
    output_dir: str = "/tmp/sandbox_plots",
) -> tuple[Any, list[PlotOutput]]:
    """Run code and extract artifacts (plots) in a language-specific manner.

    This method provides a language-specific implementation for running code
    with artifact extraction. Languages that support plot detection can override
    this method to provide custom artifact extraction logic.

    Args:
        container: The container protocol instance to run code in
        code: The code to execute
        libraries: Optional list of libraries to install before running
        enable_plotting: Whether to enable plot detection and extraction
        output_dir: Directory where plots should be saved

    Returns:
        tuple: (execution_result, list_of_plots)

    """
    # Default implementation for languages without plot support
    if enable_plotting and self.is_support_plot_detection:
        # Inject plot detection code
        injected_code = self.inject_plot_detection_code(code)

        # Run the code with plot detection
        result = container.run(injected_code, libraries)

        # Extract plots
        plots = self.extract_plots(container, output_dir)

        return result, plots
    # Run code without plot detection
    result = container.run(code, libraries)
    return result, []

LanguageConfig

LanguageConfig dataclass

LanguageConfig(
    name: str,
    file_extension: str,
    execution_commands: list[str],
    package_manager: str | None,
    is_support_library_installation: bool = True,
    plot_detection: PlotDetectionConfig | None = None,
)

Language-specific configuration.


Backend-Specific APIs

Docker Backend

SandboxDockerSession

SandboxDockerSession(
    client: DockerClient | None = None,
    image: str | None = None,
    dockerfile: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    keep_template: bool = False,
    commit_container: bool = False,
    verbose: bool = False,
    stream: bool = False,
    runtime_configs: dict | None = None,
    workdir: str = "/sandbox",
    security_policy: SecurityPolicy | None = None,
    default_timeout: float | None = None,
    execution_timeout: float | None = None,
    session_timeout: float | None = None,
    container_id: str | None = None,
    **kwargs: Any,
)

Bases: BaseSession

Sandbox session implemented using Docker containers.

This class provides a sandboxed environment for code execution by leveraging Docker. It handles Docker image management (pulling, building from Dockerfile), container creation and lifecycle, code execution, library installation, and file operations within the Docker container.

Initialize Docker session.

PARAMETER DESCRIPTION
client

The Docker client to use.

TYPE: DockerClient | None DEFAULT: None

image

The image to use.

TYPE: str | None DEFAULT: None

dockerfile

The Dockerfile to use.

TYPE: str | None DEFAULT: None

lang

The language to use.

TYPE: str DEFAULT: PYTHON

keep_template

Whether to keep the template image.

TYPE: bool DEFAULT: False

commit_container

Whether to commit the container to a new image.

TYPE: bool DEFAULT: False

verbose

Whether to enable verbose output.

TYPE: bool DEFAULT: False

stream

Whether to stream the output.

TYPE: bool DEFAULT: False

runtime_configs

The runtime configurations to use.

TYPE: dict | None DEFAULT: None

workdir

The working directory to use.

TYPE: str DEFAULT: '/sandbox'

security_policy

The security policy to use.

TYPE: SecurityPolicy | None DEFAULT: None

default_timeout

The default timeout to use.

TYPE: float | None DEFAULT: None

execution_timeout

The execution timeout to use.

TYPE: float | None DEFAULT: None

session_timeout

The session timeout to use.

TYPE: float | None DEFAULT: None

container_id

ID of existing container to connect to.

TYPE: str | None DEFAULT: None

**kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
None

None

Source code in llm_sandbox/docker.py
def __init__(
    self,  # NOSONAR
    client: docker.DockerClient | None = None,
    image: str | None = None,
    dockerfile: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    keep_template: bool = False,
    commit_container: bool = False,
    verbose: bool = False,
    stream: bool = False,
    runtime_configs: dict | None = None,
    workdir: str = "/sandbox",
    security_policy: SecurityPolicy | None = None,
    default_timeout: float | None = None,
    execution_timeout: float | None = None,
    session_timeout: float | None = None,
    container_id: str | None = None,
    **kwargs: Any,
) -> None:
    r"""Initialize Docker session.

    Args:
        client (docker.DockerClient | None): The Docker client to use.
        image (str | None): The image to use.
        dockerfile (str | None): The Dockerfile to use.
        lang (str): The language to use.
        keep_template (bool): Whether to keep the template image.
        commit_container (bool): Whether to commit the container to a new image.
        verbose (bool): Whether to enable verbose output.
        stream (bool): Whether to stream the output.
        runtime_configs (dict | None): The runtime configurations to use.
        workdir (str): The working directory to use.
        security_policy (SecurityPolicy | None): The security policy to use.
        default_timeout (float | None): The default timeout to use.
        execution_timeout (float | None): The execution timeout to use.
        session_timeout (float | None): The session timeout to use.
        container_id (str | None): ID of existing container to connect to.
        **kwargs: Additional keyword arguments.

    Returns:
        None

    """
    config = SessionConfig(
        image=image,
        dockerfile=dockerfile,
        lang=SupportedLanguage(lang.upper()),
        verbose=verbose,
        workdir=workdir,
        runtime_configs=runtime_configs or {},
        security_policy=security_policy,
        default_timeout=default_timeout,
        execution_timeout=execution_timeout,
        session_timeout=session_timeout,
        container_id=container_id,
    )

    super().__init__(config=config, **kwargs)

    self.client: docker.DockerClient

    if not client:
        self._log("Using local Docker context since client is not provided.")
        self.client = docker.from_env()
    else:
        self.client = client

    self.container_api = DockerContainerAPI(self.client, stream)

    self.docker_image: Image
    self.keep_template: bool = keep_template
    self.commit_container: bool = commit_container
    self.is_create_template: bool = False
    self.stream: bool = stream

    if mounts := kwargs.get("mounts"):
        warnings.warn(
            "The 'mounts' parameter is deprecated and will be removed in a future version. "
            "Put the mounts in 'runtime_configs' instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        existing_mounts = self.config.runtime_configs.setdefault("mounts", [])
        if isinstance(mounts, list):
            existing_mounts.extend(mounts)
        else:
            existing_mounts.append(mounts)

Functions

close

close() -> None

Close the Docker sandbox session.

This method cleans up Docker resources by: 1. Committing the container to a new image if commit_container is True. 2. Stopping and removing the running Docker container (only if we created it). 3. Removing the Docker image if is_create_template is True (image was built or pulled during this session), keep_template is False, and the image is not in use by other containers.

Note: When using existing containers, we only disconnect but don't stop/remove the container.

RAISES DESCRIPTION
ImageNotFoundError

If the image to be removed is not found (should not typically occur).

Source code in llm_sandbox/docker.py
def close(self) -> None:
    r"""Close the Docker sandbox session.

    This method cleans up Docker resources by:
    1. Committing the container to a new image if `commit_container` is True.
    2. Stopping and removing the running Docker container (only if we created it).
    3. Removing the Docker image if `is_create_template` is True (image was built or pulled
        during this session), `keep_template` is False, and the image is not in use by
        other containers.

    Note: When using existing containers, we only disconnect but don't stop/remove the container.

    Raises:
        ImageNotFoundError: If the image to be removed is not found (should not typically occur).

    """
    super().close()

    if self.container:
        if self.keep_template and self.docker_image:
            self._commit_container()

        # Only stop/remove container if we created it (not existing container)
        if not self.using_existing_container:
            try:
                self.container.stop()
                self.container.wait()
                self.container.remove(force=True)
                self._log("Stopped and removed container")
            except Exception:  # noqa: BLE001
                self._log("Error cleaning up container")
        else:
            self._log("Disconnected from existing container")

        self.container = None

    if self.is_create_template and not self.keep_template and self.docker_image:
        self._cleanup_image()

get_archive

get_archive(path: str) -> tuple[bytes, dict]

Get archive from container.

Source code in llm_sandbox/docker.py
def get_archive(self, path: str) -> tuple[bytes, dict]:
    """Get archive from container."""
    if not self.container:
        raise NotOpenSessionError

    data, stat = self.container.get_archive(path)
    return b"".join(data), stat

open

open() -> None

Open Docker session.

This method prepares the Docker environment for code execution by: - Building or pulling the Docker image (if not using existing container) - Creating a container or connecting to existing one - Setting up the environment (if not using existing container)

RAISES DESCRIPTION
ImagePullError

If the image cannot be pulled.

ImageNotFoundError

If the image cannot be found.

ContainerError

If existing container cannot be found or accessed.

Source code in llm_sandbox/docker.py
def open(self) -> None:
    r"""Open Docker session.

    This method prepares the Docker environment for code execution by:
    - Building or pulling the Docker image (if not using existing container)
    - Creating a container or connecting to existing one
    - Setting up the environment (if not using existing container)

    Raises:
        ImagePullError: If the image cannot be pulled.
        ImageNotFoundError: If the image cannot be found.
        ContainerError: If existing container cannot be found or accessed.

    """
    super().open()

    if self.using_existing_container and self.config.container_id:
        # Connect to existing container
        self._connect_to_existing_container(self.config.container_id)
    else:
        # Create new container
        self._prepare_image()

        container_config = {"image": self.docker_image, "detach": True, "tty": True, "user": "root"}
        container_config.update(self.config.runtime_configs)

        self.container = self.container_api.create_container(container_config)
        self.container_api.start_container(self.container)

    # Setup environment (skipped for existing containers)
    self.environment_setup()

Kubernetes Backend

SandboxKubernetesSession

SandboxKubernetesSession(
    client: CoreV1Api | None = None,
    image: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    verbose: bool = False,
    kube_namespace: str = "default",
    env_vars: dict[str, str] | None = None,
    pod_manifest: dict | None = None,
    workdir: str = "/sandbox",
    security_policy: SecurityPolicy | None = None,
    default_timeout: float | None = None,
    execution_timeout: float | None = None,
    session_timeout: float | None = None,
    container_id: str | None = None,
    **kwargs: Any,
)

Bases: BaseSession

Sandbox session implemented using Kubernetes Pods.

This class provides a sandboxed environment for code execution by leveraging Kubernetes. It handles Pod creation and lifecycle based on a provided or default manifest, code execution, library installation, and file operations within the Kubernetes Pod.

Initialize Kubernetes session.

PARAMETER DESCRIPTION
client

The Kubernetes client to use.

TYPE: CoreV1Api | None DEFAULT: None

image

The image to use.

TYPE: str | None DEFAULT: None

lang

The language to use.

TYPE: str DEFAULT: PYTHON

verbose

Whether to enable verbose output.

TYPE: bool DEFAULT: False

kube_namespace

The Kubernetes namespace to use.

TYPE: str DEFAULT: 'default'

env_vars

The environment variables to use.

TYPE: dict[str, str] | None DEFAULT: None

pod_manifest

The Kubernetes pod manifest to use.

TYPE: dict | None DEFAULT: None

workdir

The working directory to use.

TYPE: str DEFAULT: '/sandbox'

security_policy

The security policy to use.

TYPE: SecurityPolicy | None DEFAULT: None

default_timeout

The default timeout to use.

TYPE: float | None DEFAULT: None

execution_timeout

The execution timeout to use.

TYPE: float | None DEFAULT: None

session_timeout

The session timeout to use.

TYPE: float | None DEFAULT: None

container_id

ID of existing pod to connect to.

TYPE: str | None DEFAULT: None

**kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
None

None

Source code in llm_sandbox/kubernetes.py
def __init__(
    self,  # NOSONAR (too many arguments)
    client: CoreV1Api | None = None,
    image: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    verbose: bool = False,
    kube_namespace: str = "default",
    env_vars: dict[str, str] | None = None,
    pod_manifest: dict | None = None,
    workdir: str = "/sandbox",
    security_policy: SecurityPolicy | None = None,
    default_timeout: float | None = None,
    execution_timeout: float | None = None,
    session_timeout: float | None = None,
    container_id: str | None = None,  # This will be pod_id for Kubernetes
    **kwargs: Any,
) -> None:
    r"""Initialize Kubernetes session.

    Args:
        client (CoreV1Api | None): The Kubernetes client to use.
        image (str | None): The image to use.
        lang (str): The language to use.
        verbose (bool): Whether to enable verbose output.
        kube_namespace (str): The Kubernetes namespace to use.
        env_vars (dict[str, str] | None): The environment variables to use.
        pod_manifest (dict | None): The Kubernetes pod manifest to use.
        workdir (str): The working directory to use.
        security_policy (SecurityPolicy | None): The security policy to use.
        default_timeout (float | None): The default timeout to use.
        execution_timeout (float | None): The execution timeout to use.
        session_timeout (float | None): The session timeout to use.
        container_id (str | None): ID of existing pod to connect to.
        **kwargs: Additional keyword arguments.

    Returns:
        None

    """
    config = SessionConfig(
        image=image,
        lang=SupportedLanguage(lang.upper()),
        verbose=verbose,
        workdir=workdir,
        security_policy=security_policy,
        default_timeout=default_timeout,
        execution_timeout=execution_timeout,
        session_timeout=session_timeout,
        container_id=container_id,
    )

    super().__init__(config=config, **kwargs)

    if not client:
        self._log("Using local Kubernetes context since client is not provided.")
        from kubernetes import config as k8s_config

        k8s_config.load_kube_config()
        self.client = CoreV1Api()
    else:
        self.client = client

    self.kube_namespace = kube_namespace
    self.container_api = KubernetesContainerAPI(self.client, kube_namespace)

    # Generate unique pod name (only if not using existing pod)
    if not self.using_existing_container:
        short_uuid = uuid.uuid4().hex[:8]
        self.pod_name = f"sandbox-{lang.lower()}-{short_uuid}"
        self.env_vars = env_vars
        self.pod_manifest = pod_manifest or self._default_pod_manifest()
        self._reconfigure_with_pod_manifest()
    elif container_id:
        self.pod_name = container_id

    # For compatibility with base class
    self.stream = False

Functions

close

close() -> None

Close Kubernetes session.

Source code in llm_sandbox/kubernetes.py
def close(self) -> None:
    """Close Kubernetes session."""
    super().close()

    if self.container:
        # Only delete pod if we created it (not existing pod)
        if not self.using_existing_container:
            try:
                self.container_api.stop_container(self.container)
                self._log("Deleted pod")
            except Exception as e:  # noqa: BLE001
                self._log(f"Error cleaning up pod: {e}", "error")
        else:
            self._log("Disconnected from existing pod")

        self.container = None

get_archive

get_archive(path: str) -> tuple[bytes, dict]

Get archive from Kubernetes pod.

Source code in llm_sandbox/kubernetes.py
def get_archive(self, path: str) -> tuple[bytes, dict]:
    """Get archive from Kubernetes pod."""
    if not self.container:
        raise NotOpenSessionError

    return self.container_api.copy_from_container(self.container, path)

open

open() -> None

Open Kubernetes session.

Source code in llm_sandbox/kubernetes.py
def open(self) -> None:
    """Open Kubernetes session."""
    super().open()

    if self.using_existing_container and self.config.container_id:
        # Connect to existing pod
        self._connect_to_existing_container(self.config.container_id)
    else:
        # Create new pod
        container_config = {"pod_manifest": self.pod_manifest}
        self.container = self.container_api.create_container(container_config)

    # Setup environment only for newly-created pods
    if not self.using_existing_container:
        self.environment_setup()

Podman Backend

SandboxPodmanSession

SandboxPodmanSession(
    client: PodmanClient | None = None,
    image: str | None = None,
    dockerfile: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    keep_template: bool = False,
    commit_container: bool = False,
    verbose: bool = False,
    mounts: list | None = None,
    stream: bool = False,
    runtime_configs: dict | None = None,
    workdir: str | None = "/sandbox",
    security_policy: SecurityPolicy | None = None,
    default_timeout: float | None = None,
    execution_timeout: float | None = None,
    session_timeout: float | None = None,
    container_id: str | None = None,
    **kwargs: dict[str, Any],
)

Bases: SandboxDockerSession

Sandbox session implemented using Podman containers.

This class provides a sandboxed environment for code execution by leveraging Podman. It inherits from SandboxDockerSession since Podman is designed to be Docker-compatible, only overriding the differences in client initialization and API behavior.

Initialize Podman session.

PARAMETER DESCRIPTION
client

The Podman client to use.

TYPE: PodmanClient | None DEFAULT: None

image

The image to use.

TYPE: str | None DEFAULT: None

dockerfile

The Dockerfile to use.

TYPE: str | None DEFAULT: None

lang

The language to use.

TYPE: str DEFAULT: PYTHON

keep_template

Whether to keep the template image.

TYPE: bool DEFAULT: False

commit_container

Whether to commit the container to a new image.

TYPE: bool DEFAULT: False

verbose

Whether to enable verbose output.

TYPE: bool DEFAULT: False

mounts

The mounts to use.

TYPE: list | None DEFAULT: None

stream

Whether to stream the output.

TYPE: bool DEFAULT: False

runtime_configs

The runtime configurations to use.

TYPE: dict | None DEFAULT: None

workdir

The working directory to use.

TYPE: str | None DEFAULT: '/sandbox'

security_policy

The security policy to use.

TYPE: SecurityPolicy | None DEFAULT: None

default_timeout

The default timeout to use.

TYPE: float | None DEFAULT: None

execution_timeout

The execution timeout to use.

TYPE: float | None DEFAULT: None

session_timeout

The session timeout to use.

TYPE: float | None DEFAULT: None

container_id

ID of existing container to connect to.

TYPE: str | None DEFAULT: None

**kwargs

Additional keyword arguments.

TYPE: dict[str, Any] DEFAULT: {}

RETURNS DESCRIPTION
None

None

Source code in llm_sandbox/podman.py
def __init__(
    self,  # NOSONAR
    client: PodmanClient | None = None,
    image: str | None = None,
    dockerfile: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    keep_template: bool = False,
    commit_container: bool = False,
    verbose: bool = False,
    mounts: list | None = None,
    stream: bool = False,
    runtime_configs: dict | None = None,
    workdir: str | None = "/sandbox",
    security_policy: SecurityPolicy | None = None,
    default_timeout: float | None = None,
    execution_timeout: float | None = None,
    session_timeout: float | None = None,
    container_id: str | None = None,
    **kwargs: dict[str, Any],
) -> None:
    r"""Initialize Podman session.

    Args:
        client (PodmanClient | None): The Podman client to use.
        image (str | None): The image to use.
        dockerfile (str | None): The Dockerfile to use.
        lang (str): The language to use.
        keep_template (bool): Whether to keep the template image.
        commit_container (bool): Whether to commit the container to a new image.
        verbose (bool): Whether to enable verbose output.
        mounts (list | None): The mounts to use.
        stream (bool): Whether to stream the output.
        runtime_configs (dict | None): The runtime configurations to use.
        workdir (str | None): The working directory to use.
        security_policy (SecurityPolicy | None): The security policy to use.
        default_timeout (float | None): The default timeout to use.
        execution_timeout (float | None): The execution timeout to use.
        session_timeout (float | None): The session timeout to use.
        container_id (str | None): ID of existing container to connect to.
        **kwargs: Additional keyword arguments.

    Returns:
        None

    """
    config = SessionConfig(
        image=image,
        dockerfile=dockerfile,
        lang=SupportedLanguage(lang.upper()),
        verbose=verbose,
        workdir=workdir or "/sandbox",
        runtime_configs=runtime_configs or {},
        security_policy=security_policy,
        default_timeout=default_timeout,
        execution_timeout=execution_timeout,
        session_timeout=session_timeout,
        container_id=container_id,
    )

    # Initialize BaseSession (skip Docker's __init__)
    from llm_sandbox.core.session_base import BaseSession

    BaseSession.__init__(self, config=config, **kwargs)

    if not client:
        self._log("Using local Podman context since client is not provided.")
        self.client = PodmanClient.from_env()
    else:
        self.client = client

    self.container_api = PodmanContainerAPI(self.client, stream)

    # Set other attributes
    self.docker_image: Image
    self.keep_template: bool = keep_template
    self.commit_container: bool = commit_container
    self.is_create_template: bool = False
    self.stream: bool = stream

    if mounts:
        import warnings

        warnings.warn(
            "The 'mounts' parameter is deprecated and will be removed in a future version. "
            "Put the mounts in 'runtime_configs' instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        self.config.runtime_configs.setdefault("mounts", []).append(mounts)

Functions

Micromamba Backend

MicromambaSession

MicromambaSession(
    client: DockerClient | None = None,
    image: str = "mambaorg/micromamba:latest",
    dockerfile: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    keep_template: bool = False,
    verbose: bool = False,
    mounts: list[Mount] | None = None,
    environment: str = "base",
    commit_container: bool = False,
    stream: bool = True,
    runtime_configs: dict | None = None,
    workdir: str = "/sandbox",
    security_policy: SecurityPolicy | None = None,
    default_timeout: float | None = None,
    execution_timeout: float | None = None,
    session_timeout: float | None = None,
    container_id: str | None = None,
    **kwargs: Any,
)

Bases: SandboxDockerSession

Extends BaseSession to execute commands within a Micromamba environment.

This session leverages a Docker container (typically one with Micromamba pre-installed, like "mambaorg/micromamba:latest") and wraps executed commands with micromamba run to ensure they operate within a specified Micromamba environment.

Reference: https://github.com/vndee/llm-sandbox/pull/3

Initialize a new Micromamba-enabled sandbox session.

PARAMETER DESCRIPTION
client

An existing Docker client instance. If None, a new client is created from the local Docker environment. Defaults to None.

TYPE: DockerClient | None DEFAULT: None

image

The Docker image to use, which should have Micromamba installed. Defaults to "mambaorg/micromamba:latest".

TYPE: str DEFAULT: 'mambaorg/micromamba:latest'

dockerfile

Path to a Dockerfile to build a custom image. The resulting image should have Micromamba. Defaults to None.

TYPE: str | None DEFAULT: None

lang

The primary programming language. This mainly influences default file extensions for code execution. Defaults to SupportedLanguage.PYTHON.

TYPE: str DEFAULT: PYTHON

keep_template

If True, the Docker image will not be removed after the session ends. Defaults to False.

TYPE: bool DEFAULT: False

verbose

If True, print detailed log messages. Defaults to False.

TYPE: bool DEFAULT: False

mounts

A list of Docker Mount objects to be mounted into the container. Defaults to None.

TYPE: list[Mount] | None DEFAULT: None

environment

The name of the Micromamba environment to activate and run commands within (e.g., "base", "my_env"). Defaults to "base".

TYPE: str DEFAULT: 'base'

commit_container

If True, the Docker container's state will be committed to a new image after the session ends. Defaults to False.

TYPE: bool DEFAULT: False

stream

If True, the output from execute_command will be streamed. Defaults to True.

TYPE: bool DEFAULT: True

runtime_configs

Additional configurations for the container runtime. Defaults to None.

TYPE: dict | None DEFAULT: None

workdir

The working directory inside the container. Defaults to "/sandbox".

TYPE: str DEFAULT: '/sandbox'

security_policy

The security policy to use for the session. Defaults to None.

TYPE: SecurityPolicy | None DEFAULT: None

default_timeout

The default timeout for the session. Defaults to None.

TYPE: float | None DEFAULT: None

execution_timeout

The execution timeout for the session. Defaults to None.

TYPE: float | None DEFAULT: None

session_timeout

The session timeout for the session. Defaults to None.

TYPE: float | None DEFAULT: None

container_id

ID of existing container to connect to. Defaults to None.

TYPE: str | None DEFAULT: None

**kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

Source code in llm_sandbox/micromamba.py
def __init__(
    self,  # NOSONAR
    client: docker.DockerClient | None = None,
    image: str = "mambaorg/micromamba:latest",
    dockerfile: str | None = None,
    lang: str = SupportedLanguage.PYTHON,
    keep_template: bool = False,
    verbose: bool = False,
    mounts: list[Mount] | None = None,
    environment: str = "base",
    commit_container: bool = False,
    stream: bool = True,
    runtime_configs: dict | None = None,
    workdir: str = "/sandbox",
    security_policy: SecurityPolicy | None = None,
    default_timeout: float | None = None,
    execution_timeout: float | None = None,
    session_timeout: float | None = None,
    container_id: str | None = None,
    **kwargs: Any,
) -> None:
    r"""Initialize a new Micromamba-enabled sandbox session.

    Args:
        client (docker.DockerClient | None, optional): An existing Docker client instance.
            If None, a new client is created from the local Docker environment. Defaults to None.
        image (str, optional): The Docker image to use, which should have Micromamba installed.
            Defaults to "mambaorg/micromamba:latest".
        dockerfile (str | None, optional): Path to a Dockerfile to build a custom image.
            The resulting image should have Micromamba. Defaults to None.
        lang (str, optional): The primary programming language. This mainly influences default file
            extensions for code execution. Defaults to SupportedLanguage.PYTHON.
        keep_template (bool, optional): If True, the Docker image will not be removed after the
            session ends. Defaults to False.
        verbose (bool, optional): If True, print detailed log messages. Defaults to False.
        mounts (list[Mount] | None, optional): A list of Docker `Mount` objects to be mounted
            into the container. Defaults to None.
        environment (str, optional): The name of the Micromamba environment to activate and run
            commands within (e.g., "base", "my_env"). Defaults to "base".
        commit_container (bool, optional): If True, the Docker container's state will be committed
            to a new image after the session ends. Defaults to False.
        stream (bool, optional): If True, the output from `execute_command` will be streamed.
            Defaults to True.
        runtime_configs (dict | None, optional): Additional configurations for the container runtime.
            Defaults to None.
        workdir (str, optional): The working directory inside the container.
            Defaults to "/sandbox".
        security_policy (SecurityPolicy | None, optional): The security policy to use for the session.
            Defaults to None.
        default_timeout (float | None, optional): The default timeout for the session.
            Defaults to None.
        execution_timeout (float | None, optional): The execution timeout for the session.
            Defaults to None.
        session_timeout (float | None, optional): The session timeout for the session.
            Defaults to None.
        container_id (str | None, optional): ID of existing container to connect to.
            Defaults to None.
        **kwargs: Additional keyword arguments.

    """
    super().__init__(
        client=client,
        image=image,
        dockerfile=dockerfile,
        lang=lang,
        keep_template=keep_template,
        verbose=verbose,
        mounts=mounts,
        commit_container=commit_container,
        stream=stream,
        runtime_configs=runtime_configs or {},
        workdir=workdir,
        security_policy=security_policy,
        default_timeout=default_timeout,
        execution_timeout=execution_timeout,
        session_timeout=session_timeout,
        container_id=container_id,
        **kwargs,
    )

    self.environment = environment

    self.container_api = MicromambaContainerAPI(self.client, environment, stream)

Functions


Type Hints

Protocol Types

class ContainerProtocol(Protocol):
    """Protocol for container objects"""

    def execute_command(self, command: str, workdir: str | None = None) -> Any:
        ...

    def get_archive(self, path: str) -> tuple:
        ...

    def run(self, code: str, libraries: list | None = None) -> Any:
        ...

Complete Example

from llm_sandbox import (
    SandboxSession,
    SandboxBackend,
    ArtifactSandboxSession,
    get_security_policy,
    SecurityPolicy,
    SecurityPattern,
    SecurityIssueSeverity
)
from llm_sandbox.exceptions import SandboxTimeoutError
import base64

# Basic usage
with SandboxSession(lang="python") as session:
    result = session.run("print('Hello, World!')")
    print(result.stdout)

# With timeout configuration
with SandboxSession(
    lang="python",
    execution_timeout=30.0,  # 30 seconds for code execution
    session_timeout=300.0,   # 5 minutes session lifetime
    default_timeout=10.0     # Default timeout for operations
) as session:
    try:
        # This will use the execution_timeout (30s)
        result = session.run("print('Normal execution')")

        # Override timeout for specific execution
        result = session.run("""
import time
time.sleep(5)
print('Long operation completed')
        """, timeout=15.0)  # Override with 15 seconds

    except SandboxTimeoutError as e:
        print(f"Operation timed out: {e}")

# With security policy
policy = get_security_policy("production")
policy.add_pattern(SecurityPattern(
    pattern=r"requests\.get\(.*internal\.company",
    description="Internal network access",
    severity=SecurityIssueSeverity.HIGH
))

with SandboxSession(
    lang="python",
    security_policy=policy,
    runtime_configs={
        "cpu_count": 2,
        "mem_limit": "512m",
        "timeout": 30
    }
) as session:
    # Check code safety
    code = "import requests; requests.get('https://api.example.com')"
    is_safe, violations = session.is_safe(code)

    if is_safe:
        result = session.run(code, libraries=["requests"])
        print(result.stdout)
    else:
        print("Code failed security check")

# With artifact extraction
with ArtifactSandboxSession(
    lang="python",
    backend=SandboxBackend.DOCKER
) as session:
    result = session.run("""
import matplotlib.pyplot as plt
import numpy as np

x = np.linspace(0, 10, 100)
y = np.sin(x)
plt.plot(x, y)
plt.title('Sine Wave')
plt.show()
    """, libraries=["matplotlib", "numpy"])

    # Save plots
    for i, plot in enumerate(result.plots):
        with open(f"plot_{i}.{plot.format.value}", "wb") as f:
            f.write(base64.b64decode(plot.content_base64))

# Kubernetes backend
with SandboxSession(
    backend=SandboxBackend.KUBERNETES,
    lang="python",
    kube_namespace="default",
    pod_manifest={
        "spec": {
            "containers": [{
                "resources": {
                    "limits": {
                        "memory": "512Mi",
                        "cpu": "1"
                    }
                }
            }]
        }
    }
) as session:
    result = session.run("print('Running in Kubernetes!')")
    print(result.stdout)