Code Execution Backends

The stirrup.tools.code_backends module provides code execution backends.

CodeExecToolProvider (Base Class)

stirrup.tools.code_backends.CodeExecToolProvider

CodeExecToolProvider(
    *,
    allowed_commands: list[str] | None = None,
    shell_timeout: int = SHELL_TIMEOUT,
)

Bases: ToolProvider, ABC

Abstract base class for code execution tool providers.

CodeExecToolProvider is a ToolProvider that manages code execution environments (sandboxes, containers, local temp directories) and returns a code_exec Tool.

Subclasses must implement: - aenter(): Initialize environment and return the code_exec tool - aexit(): Cleanup the execution environment - run_command(): Execute a command and return raw result - read_file_bytes(): Read file content as bytes from the environment - write_file_bytes(): Write bytes to a file in the environment

Default implementations are provided for: - save_output_files(): Save files to local dir or another exec env (uses primitives) - upload_files(): Upload files from local or another exec env (uses primitives)

Overridable output-path hooks (see the extending docs for the contract): - output_source_roots(): Declare absolute roots for resolving output paths - resolve_output_source(): Resolve/reject a source before it is read - output_destination_identity(): Reject escapes and identify cross-env destinations

All providers accept an optional allowlist of command patterns. With an allowlist, each command is parsed with shlex, matched against the patterns, and executed without a shell: arguments are passed literally, and shell syntax such as pipes, redirects, and expansions is rejected. Without an allowlist, all commands are allowed and run through a shell.

Usage with Agent

from stirrup.clients.chat_completions_client import ChatCompletionsClient

client = ChatCompletionsClient(model="gpt-5") agent = Agent( client=client, name="assistant", tools=[LocalCodeExecToolProvider(), CALCULATOR_TOOL], )

Initialize execution environment with optional command allowlist.

Parameters:

Name	Type	Description	Default
allowed_commands	list[str] | None	Optional list of regex patterns, matched from the start of the parsed command. When set, commands run without a shell: arguments are literal and shell syntax is rejected (see _prepare_command). When None, all commands are allowed and run through a shell.	None
shell_timeout	int	Per-command wall-clock timeout (seconds) applied to every code_exec invocation from the LLM. Defaults to SHELL_TIMEOUT. Callers should set this to match their application's expected long-running-command budget rather than rely on the stirrup default.	SHELL_TIMEOUT

Source code in src/stirrup/tools/code_backends/base.py

def __init__(
    self,
    *,
    allowed_commands: list[str] | None = None,
    shell_timeout: int = SHELL_TIMEOUT,
) -> None:
    """Initialize execution environment with optional command allowlist.

    Args:
        allowed_commands: Optional list of regex patterns, matched from the
                         start of the parsed command. When set, commands run
                         without a shell: arguments are literal and shell
                         syntax is rejected (see ``_prepare_command``). When
                         None, all commands are allowed and run through a
                         shell.
        shell_timeout: Per-command wall-clock timeout (seconds) applied to
                       every ``code_exec`` invocation from the LLM. Defaults
                       to ``SHELL_TIMEOUT``. Callers should set this to match
                       their application's expected long-running-command
                       budget rather than rely on the stirrup default.

    """
    self._allowed_commands = allowed_commands
    self._shell_timeout = shell_timeout
    self._compiled_allowed: list[re.Pattern[str]] | None = None
    if allowed_commands is not None:
        self._compiled_allowed = [re.compile(p) for p in allowed_commands]

run_command abstractmethod async

run_command(
    cmd: str, *, timeout: int | None = None
) -> CommandResult

Execute a shell command and return raw CommandResult.

Parameters:

Name	Type	Description	Default
cmd	str	Shell command to execute (bash syntax).	required
timeout	int | None	Per-call wall-clock timeout (seconds). If None, falls back to self._shell_timeout configured on the provider, so direct callers and the LLM code_exec tool share the same default.	None

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def run_command(self, cmd: str, *, timeout: int | None = None) -> CommandResult:
    """Execute a shell command and return raw CommandResult.

    Args:
        cmd: Shell command to execute (bash syntax).
        timeout: Per-call wall-clock timeout (seconds). If None, falls back
            to ``self._shell_timeout`` configured on the provider, so direct
            callers and the LLM ``code_exec`` tool share the same default.
    """
    ...

read_file_bytes abstractmethod async

read_file_bytes(path: str) -> bytes

Read file content as bytes from this execution environment.

Parameters:

Name	Type	Description	Default
path	str	File path within this execution environment (relative or absolute within the env's working directory).	required

Returns:

Type	Description
bytes	File contents as bytes.

Raises:

Type	Description
FileNotFoundError	If file does not exist.
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def read_file_bytes(self, path: str) -> bytes:
    """Read file content as bytes from this execution environment.

    Args:
        path: File path within this execution environment (relative or absolute
              within the env's working directory).

    Returns:
        File contents as bytes.

    Raises:
        FileNotFoundError: If file does not exist.
        RuntimeError: If execution environment not started.

    """
    ...

write_file_bytes abstractmethod async

write_file_bytes(path: str, content: bytes) -> None

Write bytes to a file in this execution environment.

Parameters:

Name	Type	Description	Default
path	str	Destination path within this execution environment.	required
content	bytes	File contents to write.	required

Raises:

Type	Description
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

@abstractmethod
async def write_file_bytes(self, path: str, content: bytes) -> None:
    """Write bytes to a file in this execution environment.

    Args:
        path: Destination path within this execution environment.
        content: File contents to write.

    Raises:
        RuntimeError: If execution environment not started.

    """
    ...

save_output_files async

save_output_files(
    paths: list[str],
    output_dir: Path | str,
    dest_env: CodeExecToolProvider | None = None,
) -> SaveOutputFilesResult

Save files from this execution environment to a destination.

Each source path is preserved relative to output_dir: relative sources keep their directory structure, and absolute sources are made relative to a matching root from output_source_roots(). Sources that are empty, contain traversal, fall outside every declared root, or fail resolve_output_source() are rejected, and a second source that resolves to an already-claimed destination is rejected rather than overwriting it (first requested output wins). Every rejection is recorded in SaveOutputFilesResult.failed as source_path -> reason. All implementations copy files and leave sources in place. Host-backed providers preserve file modes.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of file paths in this execution environment to save.	required
output_dir	Path | str	Directory path to save files to.	required
dest_env	CodeExecToolProvider | None	If provided, output_dir is interpreted as a path within dest_env (cross-environment transfer). If None, output_dir is a local filesystem path.	None

Returns:

Type	Description
SaveOutputFilesResult	SaveOutputFilesResult containing lists of saved files and any failures.

Source code in src/stirrup/tools/code_backends/base.py

async def save_output_files(
    self,
    paths: list[str],
    output_dir: Path | str,
    dest_env: "CodeExecToolProvider | None" = None,
) -> SaveOutputFilesResult:
    """Save files from this execution environment to a destination.

    Each source path is preserved relative to ``output_dir``: relative
    sources keep their directory structure, and absolute sources are made
    relative to a matching root from ``output_source_roots()``. Sources
    that are empty, contain traversal, fall outside every declared root,
    or fail ``resolve_output_source()`` are rejected, and a second source
    that resolves to an already-claimed destination is rejected rather
    than overwriting it (first requested output wins). Every rejection is
    recorded in ``SaveOutputFilesResult.failed`` as ``source_path ->
    reason``. All implementations copy files and leave sources in place.
    Host-backed providers preserve file modes.

    Args:
        paths: List of file paths in this execution environment to save.
        output_dir: Directory path to save files to.
        dest_env: If provided, output_dir is interpreted as a path within dest_env
                  (cross-environment transfer). If None, output_dir is a local
                  filesystem path.

    Returns:
        SaveOutputFilesResult containing lists of saved files and any failures.

    """
    result = SaveOutputFilesResult()
    output_dir_str = str(output_dir)
    reserved_destinations: dict[str, str] = {}
    local_output_root = Path(output_dir)
    if dest_env is None:
        local_output_root.mkdir(parents=True, exist_ok=True)

    for source_path in dict.fromkeys(paths):
        try:
            relative_path = _safe_output_relative_path(
                source_path,
                source_roots=self.output_source_roots(),
            )
            resolved_source_path = await self.resolve_output_source(source_path)
            content = await self.read_file_bytes(resolved_source_path)

            if dest_env is not None:
                # Transfer to another exec env (cross-environment)
                destination = PurePosixPath(output_dir_str) / relative_path
                identity = await dest_env.output_destination_identity(destination.as_posix(), output_dir)
                if identity in reserved_destinations:
                    raise ValueError(
                        f"Output destination collision: {destination} "
                        f"(already claimed by {reserved_destinations[identity]})"
                    )
                reserved_destinations[identity] = source_path
                logger.debug(
                    "CROSS-ENV TRANSFER: %s (%d bytes) -> %s (dest_env: %s)",
                    source_path,
                    len(content),
                    destination,
                    type(dest_env).__name__,
                )
                await dest_env.write_file_bytes(destination.as_posix(), content)
                result.saved.append(SavedFile(source_path, destination, len(content)))
            else:
                # Save to local filesystem
                destination = local_output_root.joinpath(*relative_path.parts)
                _reserve_output_destination(
                    destination,
                    reserved_destinations,
                    claimant=source_path,
                    output_root=local_output_root,
                )
                logger.debug(
                    "SAVE TO LOCAL: %s (%d bytes) -> %s",
                    source_path,
                    len(content),
                    destination,
                )
                _atomic_write_bytes(content, destination)
                result.saved.append(SavedFile(source_path, destination, len(content)))
        except (ValueError, FileNotFoundError) as e:
            logger.warning("TRANSFER FAILED: %s -> %s: %s", source_path, output_dir_str, e)
            result.failed[source_path] = str(e)
        except Exception as e:
            logger.exception("TRANSFER FAILED: %s -> %s", source_path, output_dir_str)
            result.failed[source_path] = str(e)

    return result

upload_files async

upload_files(
    *paths: Path | str,
    source_env: CodeExecToolProvider | None = None,
    dest_dir: str | None = None,
) -> UploadFilesResult

Upload files to this execution environment.

Parameters:

Name	Type	Description	Default
*paths	Path | str	File or directory paths to upload. If source_env is None, these are local filesystem paths. If source_env is provided, these are paths within source_env (cross-environment transfer).	()
source_env	CodeExecToolProvider | None	If provided, paths are within source_env. If None, paths are local filesystem paths.	None
dest_dir	str | None	Destination directory in this environment. If None, uses the environment's working directory.	None

Returns:

Type	Description
UploadFilesResult	UploadFilesResult containing lists of uploaded files and any failures.

Raises:

Type	Description
RuntimeError	If execution environment not started.

Source code in src/stirrup/tools/code_backends/base.py

async def upload_files(
    self,
    *paths: Path | str,
    source_env: "CodeExecToolProvider | None" = None,
    dest_dir: str | None = None,
) -> UploadFilesResult:
    """Upload files to this execution environment.

    Args:
        *paths: File or directory paths to upload. If source_env is None, these
                are local filesystem paths. If source_env is provided, these are
                paths within source_env (cross-environment transfer).
        source_env: If provided, paths are within source_env. If None, paths are
                    local filesystem paths.
        dest_dir: Destination directory in this environment.
                  If None, uses the environment's working directory.

    Returns:
        UploadFilesResult containing lists of uploaded files and any failures.

    Raises:
        RuntimeError: If execution environment not started.

    """
    result = UploadFilesResult()
    dest_dir_str = dest_dir or ""

    for path in paths:
        path_str = str(path)
        try:
            if source_env:
                # Cross-environment transfer: read from source_env
                # Check if it's a directory first
                if await source_env.is_directory(path_str):
                    # Handle directory recursively
                    # Preserve directory name when dest_dir not specified
                    dir_name = Path(path_str).name
                    files = await source_env.list_files(path_str)
                    for rel_file_path in files:
                        src_file_path = f"{path_str}/{rel_file_path}"
                        # If dest_dir specified, put files directly there
                        # Otherwise, preserve the source directory name
                        if dest_dir_str:
                            dest_path = f"{dest_dir_str}/{rel_file_path}"
                        else:
                            dest_path = f"{dir_name}/{rel_file_path}"
                        content = await source_env.read_file_bytes(src_file_path)
                        logger.debug(
                            "UPLOAD CROSS-ENV (dir): %s (%d bytes) from %s -> %s",
                            src_file_path,
                            len(content),
                            type(source_env).__name__,
                            dest_path,
                        )
                        await self.write_file_bytes(dest_path, content)
                        result.uploaded.append(UploadedFile(Path(src_file_path), dest_path, len(content)))
                else:
                    # Single file transfer
                    content = await source_env.read_file_bytes(path_str)
                    filename = Path(path_str).name
                    dest_path = f"{dest_dir_str}/{filename}" if dest_dir_str else filename
                    logger.debug(
                        "UPLOAD CROSS-ENV: %s (%d bytes) from %s -> %s",
                        path_str,
                        len(content),
                        type(source_env).__name__,
                        dest_path,
                    )
                    await self.write_file_bytes(dest_path, content)
                    result.uploaded.append(UploadedFile(Path(path_str), dest_path, len(content)))
            else:
                # Local filesystem upload - must be handled by subclass
                # This is a fallback that reads from local fs and writes to env
                local_path = Path(path)
                if local_path.is_dir():
                    # Handle directory recursively
                    for file_path in local_path.rglob("*"):
                        if file_path.is_file():
                            rel_path = file_path.relative_to(local_path)
                            dest_path = f"{dest_dir_str}/{rel_path}" if dest_dir_str else str(rel_path)
                            content = file_path.read_bytes()
                            logger.debug(
                                "UPLOAD FROM LOCAL: %s (%d bytes) -> %s",
                                file_path,
                                len(content),
                                dest_path,
                            )
                            await self.write_file_bytes(dest_path, content)
                            result.uploaded.append(UploadedFile(file_path, dest_path, len(content)))
                else:
                    filename = local_path.name
                    dest_path = f"{dest_dir_str}/{filename}" if dest_dir_str else filename
                    content = local_path.read_bytes()
                    logger.debug(
                        "UPLOAD FROM LOCAL: %s (%d bytes) -> %s",
                        local_path,
                        len(content),
                        dest_path,
                    )
                    await self.write_file_bytes(dest_path, content)
                    result.uploaded.append(UploadedFile(local_path, dest_path, len(content)))
        except Exception as e:
            logger.debug("UPLOAD FAILED: %s -> %s: %s", path_str, dest_dir_str, e)
            result.failed[path_str] = str(e)

    return result

get_code_exec_tool

get_code_exec_tool(
    *,
    name: str = "code_exec",
    description: str | None = None,
) -> Tool[CodeExecutionParams, ToolUseCountMetadata]

Create a code execution tool for this environment.

Parameters:

Name	Type	Description	Default
name	str	Tool name	'code_exec'
description	str | None	Tool description	None

Returns:

Type	Description
Tool[CodeExecutionParams, ToolUseCountMetadata]	Tool[CodeExecutionParams] that executes commands in this environment

Source code in src/stirrup/tools/code_backends/base.py

def get_code_exec_tool(
    self,
    *,
    name: str = "code_exec",
    description: str | None = None,
) -> Tool[CodeExecutionParams, ToolUseCountMetadata]:
    """Create a code execution tool for this environment.

    Args:
        name: Tool name
        description: Tool description

    Returns:
        Tool[CodeExecutionParams] that executes commands in this environment

    """
    env = self

    async def executor(params: CodeExecutionParams) -> ToolResult[ToolUseCountMetadata]:
        # timeout=None defers to env._shell_timeout, set in CodeExecToolProvider.__init__.
        result = await env.run_command(params.cmd)
        return format_result(result)

    return Tool[CodeExecutionParams, ToolUseCountMetadata](
        name=name,
        description=description
        or "Execute a shell command in the execution environment. Returns exit code, stdout, and stderr as XML.",
        parameters=CodeExecutionParams,
        executor=executor,  # ty: ignore[invalid-argument-type]
    )

get_view_image_tool

get_view_image_tool(
    *,
    name: str = "view_image",
    description: str | None = None,
) -> Tool[ViewImageParams, ToolUseCountMetadata]

Create a view_image tool for this environment.

Parameters:

Name	Type	Description	Default
name	str	Tool name	'view_image'
description	str | None	Tool description	None

Returns:

Type	Description
Tool[ViewImageParams, ToolUseCountMetadata]	Tool[ViewImageParams, ToolUseCountMetadata] that views images in this environment

Source code in src/stirrup/tools/code_backends/base.py

def get_view_image_tool(
    self,
    *,
    name: str = "view_image",
    description: str | None = None,
) -> Tool[ViewImageParams, ToolUseCountMetadata]:
    """Create a view_image tool for this environment.

    Args:
        name: Tool name
        description: Tool description

    Returns:
        Tool[ViewImageParams, ToolUseCountMetadata] that views images in this environment

    """
    env = self

    async def executor(params: ViewImageParams) -> ToolResult[ToolUseCountMetadata]:
        try:
            image = await env.view_image(params.path)
            return ToolResult(
                content=["Viewing image at path: " + params.path, image],
                metadata=ToolUseCountMetadata(),
            )
        except FileNotFoundError:
            return ToolResult(
                content=f"Image `{params.path}` not found.",
                success=False,
                metadata=ToolUseCountMetadata(),
            )
        except ValueError as e:
            return ToolResult(
                content=str(e),
                success=False,
                metadata=ToolUseCountMetadata(),
            )

    return Tool[ViewImageParams, ToolUseCountMetadata](
        name=name,
        description=description or "View an image file from the execution environment's filesystem.",
        parameters=ViewImageParams,
        executor=executor,  # ty: ignore[invalid-argument-type]
    )

LocalCodeExecToolProvider

Executes code in an isolated temporary directory on the host machine.

stirrup.tools.code_backends.LocalCodeExecToolProvider

LocalCodeExecToolProvider(
    *,
    allowed_commands: list[str] | None = None,
    temp_base_dir: Path | str | None = None,
    description: str | None = None,
    shell_timeout: int = SHELL_TIMEOUT,
)

Bases: CodeExecToolProvider

Local code execution tool provider using an isolated temp directory.

Commands are executed with the temp directory as the working directory. An optional allowlist can restrict which commands are permitted.

Usage with Agent

from stirrup.clients.chat_completions_client import ChatCompletionsClient

client = ChatCompletionsClient(model="gpt-5") agent = Agent( client=client, name="assistant", tools=[LocalCodeExecToolProvider(), CALCULATOR_TOOL], )

async with agent.session(output_dir="./output") as session: await session.run("Run some Python code")

Standalone usage

provider = LocalCodeExecToolProvider()

async with provider as tool: # tool is a Tool instance for code execution result = await provider.run_command("python script.py") await provider.save_output_files(["output.txt"], "/path/to/output")

Initialize LocalCodeExecToolProvider configuration.

Parameters:

Name	Type	Description	Default
allowed_commands	list[str] | None	Optional list of regex patterns. When set, commands are matched against the patterns and run without a shell, with literal arguments. When None, all commands are allowed and run through bash.	None
temp_base_dir	Path | str | None	Optional base directory for creating the execution environment temp directory. If None, uses the system default temp directory.	None
description	str | None	Optional description of the tool. If None, uses the default description.	None
shell_timeout	int	Per-command wall-clock timeout (seconds) for every code_exec invocation, and the default for direct run_command calls that pass timeout=None. Defaults to SHELL_TIMEOUT.	SHELL_TIMEOUT

Source code in src/stirrup/tools/code_backends/local.py

def __init__(
    self,
    *,
    allowed_commands: list[str] | None = None,
    temp_base_dir: Path | str | None = None,
    description: str | None = None,
    shell_timeout: int = SHELL_TIMEOUT,
) -> None:
    """Initialize LocalCodeExecToolProvider configuration.

    Args:
        allowed_commands: Optional list of regex patterns. When set,
                         commands are matched against the patterns and run
                         without a shell, with literal arguments. When
                         None, all commands are allowed and run through
                         bash.
        temp_base_dir: Optional base directory for creating the execution environment
                      temp directory. If None, uses the system default temp directory.
        description: Optional description of the tool. If None, uses the default description.
        shell_timeout: Per-command wall-clock timeout (seconds) for every
            ``code_exec`` invocation, and the default for direct
            ``run_command`` calls that pass ``timeout=None``. Defaults to
            ``SHELL_TIMEOUT``.

    """
    super().__init__(allowed_commands=allowed_commands, shell_timeout=shell_timeout)
    self._temp_dir: Path | None = None
    self._temp_base_dir: Path | None = Path(temp_base_dir) if temp_base_dir else None
    self._description = (
        description
        or "Execute a shell command in the execution environment. Returns exit code, stdout, and stderr as XML. Use `uv` to manage packages."
    )

_temp_dir instance-attribute

_temp_dir: Path | None = None

_temp_base_dir instance-attribute

_temp_base_dir: Path | None = (
    Path(temp_base_dir) if temp_base_dir else None
)

_description instance-attribute

_description = (
    description
    or "Execute a shell command in the execution environment. Returns exit code, stdout, and stderr as XML. Use `uv` to manage packages."
)

temp_dir property

temp_dir: Path | None

Return the temp directory path, or None if not started.

_check_absolute_paths

_check_absolute_paths(cmd: str) -> CommandResult | None

Check if command contains absolute paths that could escape the temp directory.

Returns:

Type	Description
CommandResult | None	CommandResult with error if absolute paths detected, None otherwise.

Note

This check is specific to LocalCodeExecToolProvider since Docker and E2B providers are already sandboxed and absolute paths are safe within them.

Source code in src/stirrup/tools/code_backends/local.py

def _check_absolute_paths(self, cmd: str) -> CommandResult | None:
    """Check if command contains absolute paths that could escape the temp directory.

    Returns:
        CommandResult with error if absolute paths detected, None otherwise.

    Note:
        This check is specific to LocalCodeExecToolProvider since Docker and E2B
        providers are already sandboxed and absolute paths are safe within them.
    """
    absolute_patterns = [
        r"~/",  # ~/path - home directory shortcut
        r"/(?:home|Users|tmp|var|etc)/",  # /home/, /Users/, /tmp/, etc.
        r"\$HOME/",  # $HOME/path
        r"\$\{HOME\}/",  # ${HOME}/path
    ]
    for pattern in absolute_patterns:
        if re.search(pattern, cmd):
            return CommandResult(
                exit_code=1,
                stdout="",
                stderr=(
                    "Command appears to use absolute paths which could write outside "
                    "the execution environment. Use relative paths instead."
                ),
                error_kind="absolute_path_detected",
                advice=(
                    "Use relative paths (e.g., './output.txt' instead of '~/output.txt'). "
                    "For full filesystem access, use DockerCodeExecToolProvider or E2BCodeExecToolProvider."
                ),
            )
    return None

__aenter__ async

__aenter__() -> Tool[
    CodeExecutionParams, ToolUseCountMetadata
]

Create temp directory and return the code_exec tool.

Source code in src/stirrup/tools/code_backends/local.py

async def __aenter__(self) -> "Tool[CodeExecutionParams, ToolUseCountMetadata]":
    """Create temp directory and return the code_exec tool."""
    if self._temp_base_dir:
        self._temp_base_dir.mkdir(parents=True, exist_ok=True)
    self._temp_dir = Path(tempfile.mkdtemp(prefix="local_exec_env_", dir=self._temp_base_dir))
    logger.debug("Created local execution environment temp directory: %s", self._temp_dir)
    return self.get_code_exec_tool(description=self._description)

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None

Cleanup the local execution environment.

Source code in src/stirrup/tools/code_backends/local.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Cleanup the local execution environment."""
    if self._temp_dir and self._temp_dir.exists():
        try:
            shutil.rmtree(self._temp_dir)
        except Exception as exc:
            logger.warning("Failed to cleanup temp directory %s: %s", self._temp_dir, exc)
    self._temp_dir = None

_resolve_and_validate_path

_resolve_and_validate_path(path: str) -> Path

Resolve a path and validate it's within the temp directory.

Parameters:

Name	Type	Description	Default
path	str	File path (relative or absolute within the temp dir).	required

Returns:

Type	Description
Path	Resolved absolute Path.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.
FileNotFoundError	If path does not exist (for reads).

Source code in src/stirrup/tools/code_backends/local.py

def _resolve_and_validate_path(self, path: str) -> Path:
    """Resolve a path and validate it's within the temp directory.

    Args:
        path: File path (relative or absolute within the temp dir).

    Returns:
        Resolved absolute Path.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.
        FileNotFoundError: If path does not exist (for reads).

    """
    if self._temp_dir is None:
        raise RuntimeError("ExecutionEnvironment not started.")

    resolved = Path(path)
    if not resolved.is_absolute():
        resolved = self._temp_dir / resolved

    # Security: ensure path is within temp directory
    try:
        resolved.resolve().relative_to(self._temp_dir.resolve())
    except ValueError as e:
        raise ValueError(f"Path is outside execution environment: {path}") from e

    return resolved

read_file_bytes async

read_file_bytes(path: str) -> bytes

Read file content as bytes from the temp directory.

Parameters:

Name	Type	Description	Default
path	str	File path (relative or absolute within the temp dir).	required

Returns:

Type	Description
bytes	File contents as bytes.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.
FileNotFoundError	If file does not exist.

Source code in src/stirrup/tools/code_backends/local.py

async def read_file_bytes(self, path: str) -> bytes:
    """Read file content as bytes from the temp directory.

    Args:
        path: File path (relative or absolute within the temp dir).

    Returns:
        File contents as bytes.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.
        FileNotFoundError: If file does not exist.

    """
    resolved = self._resolve_and_validate_path(path)
    if not resolved.exists():
        raise FileNotFoundError(f"File not found: {path}")
    return resolved.read_bytes()

write_file_bytes async

write_file_bytes(path: str, content: bytes) -> None

Write bytes to a file in the temp directory.

Parameters:

Name	Type	Description	Default
path	str	Destination path (relative or absolute within the temp dir).	required
content	bytes	File contents to write.	required

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.

Source code in src/stirrup/tools/code_backends/local.py

async def write_file_bytes(self, path: str, content: bytes) -> None:
    """Write bytes to a file in the temp directory.

    Args:
        path: Destination path (relative or absolute within the temp dir).
        content: File contents to write.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.

    """
    resolved = self._resolve_and_validate_path(path)
    _atomic_write_bytes(content, resolved)

file_exists async

file_exists(path: str) -> bool

Check if a file exists in the temp directory.

Parameters:

Name	Type	Description	Default
path	str	File path (relative or absolute within the temp dir).	required

Returns:

Type	Description
bool	True if the file exists, False otherwise.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.

Source code in src/stirrup/tools/code_backends/local.py

async def file_exists(self, path: str) -> bool:
    """Check if a file exists in the temp directory.

    Args:
        path: File path (relative or absolute within the temp dir).

    Returns:
        True if the file exists, False otherwise.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.

    """
    resolved = self._resolve_and_validate_path(path)
    return resolved.exists() and resolved.is_file()

is_directory async

is_directory(path: str) -> bool

Check if a path is a directory in the temp directory.

Parameters:

Name	Type	Description	Default
path	str	Path (relative or absolute within the temp dir).	required

Returns:

Type	Description
bool	True if the path exists and is a directory, False otherwise.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.

Source code in src/stirrup/tools/code_backends/local.py

async def is_directory(self, path: str) -> bool:
    """Check if a path is a directory in the temp directory.

    Args:
        path: Path (relative or absolute within the temp dir).

    Returns:
        True if the path exists and is a directory, False otherwise.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.

    """
    resolved = self._resolve_and_validate_path(path)
    return resolved.exists() and resolved.is_dir()

list_files async

list_files(path: str) -> list[str]

List all files recursively in a directory within the temp directory.

Parameters:

Name	Type	Description	Default
path	str	Directory path (relative or absolute within the temp dir).	required

Returns:

Type	Description
list[str]	List of file paths (relative to the given path) for all files in the directory.
list[str]	Returns an empty list if the path is a file or doesn't exist.

Raises:

Type	Description
RuntimeError	If environment not started.
ValueError	If path is outside temp directory.

Source code in src/stirrup/tools/code_backends/local.py

async def list_files(self, path: str) -> list[str]:
    """List all files recursively in a directory within the temp directory.

    Args:
        path: Directory path (relative or absolute within the temp dir).

    Returns:
        List of file paths (relative to the given path) for all files in the directory.
        Returns an empty list if the path is a file or doesn't exist.

    Raises:
        RuntimeError: If environment not started.
        ValueError: If path is outside temp directory.

    """
    resolved = self._resolve_and_validate_path(path)
    if not resolved.exists() or not resolved.is_dir():
        return []

    files = []
    for file_path in resolved.rglob("*"):
        if file_path.is_file():
            rel_path = file_path.relative_to(resolved)
            files.append(str(rel_path))
    return files

run_command async

run_command(
    cmd: str, *, timeout: int | None = None
) -> CommandResult

Execute command in the temp directory.

Parameters:

Name	Type	Description	Default
cmd	str	Shell command to execute (bash syntax).	required
timeout	int | None	Maximum time in seconds to wait for command completion. If None, falls back to self._shell_timeout (set in __init__), so direct callers share the same budget as the LLM code_exec tool.	None

Returns:

Type	Description
CommandResult	CommandResult with exit_code, stdout, stderr, and optional error info.

Source code in src/stirrup/tools/code_backends/local.py

async def run_command(self, cmd: str, *, timeout: int | None = None) -> CommandResult:
    """Execute command in the temp directory.

    Args:
        cmd: Shell command to execute (bash syntax).
        timeout: Maximum time in seconds to wait for command completion.
            If None, falls back to ``self._shell_timeout`` (set in
            ``__init__``), so direct callers share the same budget as the
            LLM ``code_exec`` tool.

    Returns:
        CommandResult with exit_code, stdout, stderr, and optional error info.

    """
    if self._temp_dir is None:
        raise RuntimeError(
            "ExecutionEnvironment not started. Ensure current Agent is equipped with a CodeExecToolProvider."
        )
    if timeout is None:
        timeout = self._shell_timeout

    # With an allowlist, the parsed argv must run directly (no shell).
    argv, rejection = self._prepare_command(cmd)
    if rejection is not None:
        return rejection

    # Check for absolute paths (local environment is not sandboxed)
    absolute_path_error = self._check_absolute_paths(cmd)
    if absolute_path_error:
        return absolute_path_error

    process = None
    try:
        with anyio.fail_after(timeout):
            # start_new_session=True puts the child at the head of its own
            # session, so pgid == child pid and we can killpg the whole tree
            # on timeout. Without it, process.kill() only terminates the
            # root process, leaving grandchildren reparented to init (same
            # bug shape as the docker orphan leak that motivated this PR).
            process = await anyio.open_process(
                argv if argv is not None else ["bash", "-c", cmd],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                cwd=self._temp_dir,
                start_new_session=True,
            )

            # Read all output from streams concurrently
            stdout_chunks: list[bytes] = []
            stderr_chunks: list[bytes] = []

            async def read_stdout() -> None:
                if process.stdout:
                    stdout_chunks.extend([chunk async for chunk in process.stdout])

            async def read_stderr() -> None:
                if process.stderr:
                    stderr_chunks.extend([chunk async for chunk in process.stderr])

            async with anyio.create_task_group() as tg:
                tg.start_soon(read_stdout)
                tg.start_soon(read_stderr)

            await process.wait()

            return CommandResult(
                exit_code=process.returncode or 0,
                stdout=b"".join(stdout_chunks).decode("utf-8", errors="replace"),
                stderr=b"".join(stderr_chunks).decode("utf-8", errors="replace"),
            )

    except TimeoutError:
        if process:
            with contextlib.suppress(ProcessLookupError):
                os.killpg(process.pid, signal.SIGKILL)
            with anyio.move_on_after(5):
                await process.wait()
        return CommandResult(
            exit_code=1,
            stdout="",
            stderr=f"Command timed out after {timeout} seconds",
            error_kind="timeout",
        )
    except Exception as exc:
        return CommandResult(
            exit_code=1,
            stdout="",
            stderr=str(exc),
            error_kind="execution_error",
        )

save_output_files async

save_output_files(
    paths: list[str],
    output_dir: Path | str,
    dest_env: CodeExecToolProvider | None = None,
) -> SaveOutputFilesResult

Copy files from the temp directory to a destination.

Source files remain in the execution environment. Existing destination files are atomically replaced, while duplicate destinations within one call are rejected and reported in failed.

When dest_env is provided (cross-environment transfer), files are copied using the base class implementation via read/write primitives.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of file paths in the execution environment (relative or absolute). Relative paths are resolved against the execution environment temp directory.	required
output_dir	Path | str	Directory path to save files to.	required
dest_env	CodeExecToolProvider | None	If provided, output_dir is interpreted as a path within dest_env (cross-environment transfer). If None, output_dir is a local filesystem path.	None

Returns:

Type	Description
SaveOutputFilesResult	SaveOutputFilesResult containing lists of saved files and any failures.

Source code in src/stirrup/tools/code_backends/local.py

async def save_output_files(
    self,
    paths: list[str],
    output_dir: Path | str,
    dest_env: "CodeExecToolProvider | None" = None,
) -> SaveOutputFilesResult:
    """Copy files from the temp directory to a destination.

    Source files remain in the execution environment. Existing destination
    files are atomically replaced, while duplicate destinations within one
    call are rejected and reported in ``failed``.

    When dest_env is provided (cross-environment transfer), files are copied
    using the base class implementation via read/write primitives.

    Args:
        paths: List of file paths in the execution environment (relative or absolute).
               Relative paths are resolved against the execution environment temp directory.
        output_dir: Directory path to save files to.
        dest_env: If provided, output_dir is interpreted as a path within dest_env
                  (cross-environment transfer). If None, output_dir is a local
                  filesystem path.

    Returns:
        SaveOutputFilesResult containing lists of saved files and any failures.

    """
    if self._temp_dir is None:
        raise RuntimeError(
            "ExecutionEnvironment not started. Ensure current Agent is equipped with a CodeExecToolProvider."
        )

    # If dest_env is provided, use the base class implementation (cross-env transfer)
    if dest_env is not None:
        return await super().save_output_files(paths, output_dir, dest_env)

    return _save_host_output_files(
        paths,
        output_dir,
        source_roots=self.output_source_roots(),
        resolve_source=self._resolve_and_validate_path,
    )

output_source_roots

output_source_roots() -> tuple[OutputSourceRoot, ...]

Return absolute execution roots accepted when resolving output paths.

Source code in src/stirrup/tools/code_backends/local.py

def output_source_roots(self) -> tuple[OutputSourceRoot, ...]:
    """Return absolute execution roots accepted when resolving output paths."""
    return (OutputSourceRoot.for_host(self._temp_dir),) if self._temp_dir is not None else ()

output_destination_identity async

output_destination_identity(
    destination: str, output_root: Path | str
) -> str

Validate a cross-environment destination and return its host identity.

Source code in src/stirrup/tools/code_backends/local.py

async def output_destination_identity(self, destination: str, output_root: Path | str) -> str:
    """Validate a cross-environment destination and return its host identity."""
    resolved_destination = self._resolve_and_validate_path(destination)
    resolved_root = self._resolve_and_validate_path(str(output_root))
    return _host_output_destination_identity(resolved_destination, resolved_root)

upload_files async

upload_files(
    *paths: Path | str,
    source_env: CodeExecToolProvider | None = None,
    dest_dir: str | None = None,
) -> UploadFilesResult

Upload files to the execution environment.

When source_env is None (local filesystem), files are COPIED (not moved) - originals remain on the local filesystem. Directories are uploaded recursively, preserving their structure.

When source_env is provided (cross-environment transfer), files are copied using the base class implementation via read/write primitives.

Parameters:

Name	Type	Description	Default
*paths	Path | str	File or directory paths to upload. If source_env is None, these are local filesystem paths. If source_env is provided, these are paths within source_env.	()
source_env	CodeExecToolProvider | None	If provided, paths are within source_env. If None, paths are local filesystem paths.	None
dest_dir	str | None	Destination subdirectory within the temp directory. If None, files are placed directly in the temp directory.	None

Returns:

Type	Description
UploadFilesResult	UploadFilesResult containing lists of uploaded files and any failures.

Source code in src/stirrup/tools/code_backends/local.py

async def upload_files(
    self,
    *paths: Path | str,
    source_env: "CodeExecToolProvider | None" = None,
    dest_dir: str | None = None,
) -> UploadFilesResult:
    """Upload files to the execution environment.

    When source_env is None (local filesystem), files are COPIED (not moved) -
    originals remain on the local filesystem.
    Directories are uploaded recursively, preserving their structure.

    When source_env is provided (cross-environment transfer), files are copied
    using the base class implementation via read/write primitives.

    Args:
        *paths: File or directory paths to upload. If source_env is None, these
                are local filesystem paths. If source_env is provided, these are
                paths within source_env.
        source_env: If provided, paths are within source_env. If None, paths are
                    local filesystem paths.
        dest_dir: Destination subdirectory within the temp directory.
                  If None, files are placed directly in the temp directory.

    Returns:
        UploadFilesResult containing lists of uploaded files and any failures.

    """
    if self._temp_dir is None:
        raise RuntimeError(
            "ExecutionEnvironment not started. Ensure current Agent is equipped with a CodeExecToolProvider."
        )

    # If source_env is provided, use the base class implementation (cross-env transfer)
    if source_env is not None:
        return await super().upload_files(*paths, source_env=source_env, dest_dir=dest_dir)

    # Local filesystem - use optimized copy operation
    dest_base = self._temp_dir / dest_dir if dest_dir else self._temp_dir
    dest_base.mkdir(parents=True, exist_ok=True)

    result = UploadFilesResult()

    for source in paths:
        source = Path(source).resolve()

        if not source.exists():
            result.failed[str(source)] = "File or directory does not exist"
            logger.warning("Upload source does not exist: %s", source)
            continue

        try:
            if source.is_file():
                dest = dest_base / source.name
                shutil.copy2(source, dest)
                result.uploaded.append(
                    UploadedFile(
                        source_path=source,
                        dest_path=str(dest.relative_to(self._temp_dir)),
                        size=source.stat().st_size,
                    ),
                )
                logger.debug("Uploaded file: %s -> %s", source, dest)

            elif source.is_dir():
                # If dest_dir was explicitly provided, copy contents directly to dest_base
                # Otherwise, create a subdirectory with the source's name
                if dest_dir:
                    dest = dest_base
                    # Copy contents of source directory into dest_base
                    for item in source.iterdir():
                        item_dest = dest / item.name
                        if item.is_file():
                            shutil.copy2(item, item_dest)
                        else:
                            shutil.copytree(item, item_dest, dirs_exist_ok=True)
                else:
                    dest = dest_base / source.name
                    shutil.copytree(source, dest, dirs_exist_ok=True)
                # Track all individual files uploaded
                for file_path in source.rglob("*"):
                    if file_path.is_file():
                        relative = file_path.relative_to(source)
                        dest_file = dest / relative
                        result.uploaded.append(
                            UploadedFile(
                                source_path=file_path,
                                dest_path=str(dest_file.relative_to(self._temp_dir)),
                                size=file_path.stat().st_size,
                            ),
                        )
                logger.debug("Uploaded directory: %s -> %s", source, dest)

        except Exception as exc:
            result.failed[str(source)] = str(exc)
            logger.exception("Failed to upload: %s", source)

    return result

view_image async

view_image(path: str) -> ImageContentBlock

Read and return an image file from the local execution environment.

Parameters:

Name	Type	Description	Default
path	str	Path to image file (relative to temp directory, or absolute within it).	required

Returns:

Type	Description
ImageContentBlock	ImageContentBlock containing the image data.

Raises:

Type	Description
RuntimeError	If execution environment not started.
FileNotFoundError	If file does not exist.
ValueError	If path is outside temp directory, is a directory, or not a valid image.

Source code in src/stirrup/tools/code_backends/local.py

async def view_image(self, path: str) -> ImageContentBlock:
    """Read and return an image file from the local execution environment.

    Args:
        path: Path to image file (relative to temp directory, or absolute within it).

    Returns:
        ImageContentBlock containing the image data.

    Raises:
        RuntimeError: If execution environment not started.
        FileNotFoundError: If file does not exist.
        ValueError: If path is outside temp directory, is a directory, or not a valid image.

    """
    file_bytes = await self.read_file_bytes(path)
    return ImageContentBlock(data=file_bytes)

DockerCodeExecToolProvider

Executes code in a Docker container.

Note

Requires pip install stirrup[docker] (or: uv add stirrup[docker])

from stirrup.tools.code_backends.docker import DockerCodeExecToolProvider

provider = DockerCodeExecToolProvider.from_image("python:3.12-slim")

E2BCodeExecToolProvider

Executes code in an E2B cloud sandbox.

Note

Requires pip install stirrup[e2b] (or: uv add stirrup[e2b]) and E2B_API_KEY environment variable.

from stirrup.tools.code_backends.e2b import E2BCodeExecToolProvider

provider = E2BCodeExecToolProvider()

Data Types

stirrup.tools.code_backends.CodeExecutionParams

Bases: BaseModel

Shell command to execute in the execution environment.

cmd instance-attribute

cmd: Annotated[
    str,
    Field(
        description="Shell command to execute (bash syntax). IMPORTANT: Use only relative paths. Do not use absolute paths (starting with / or ~) or reference directories outside the working directory."
    ),
]

stirrup.tools.code_backends.CommandResult dataclass

CommandResult(
    exit_code: int,
    stdout: str,
    stderr: str,
    error_kind: str | None = None,
    advice: str | None = None,
)

Raw result from command execution (before formatting).

exit_code instance-attribute

exit_code: int

stdout instance-attribute

stdout: str

stderr instance-attribute

stderr: str

error_kind class-attribute instance-attribute

error_kind: str | None = None

advice class-attribute instance-attribute

advice: str | None = None

stirrup.tools.code_backends.SavedFile dataclass

SavedFile(
    source_path: str, output_path: PurePath, size: int
)

Information about a file saved from the execution environment.

source_path instance-attribute

source_path: str

output_path instance-attribute

output_path: PurePath

size instance-attribute

size: int

stirrup.tools.code_backends.SaveOutputFilesResult dataclass

SaveOutputFilesResult(
    saved: list[SavedFile] = list(),
    failed: dict[str, str] = dict(),
)

Result of saving output files from the execution environment.

saved class-attribute instance-attribute

saved: list[SavedFile] = field(default_factory=list)

failed class-attribute instance-attribute

failed: dict[str, str] = field(default_factory=dict)

stirrup.tools.code_backends.UploadedFile dataclass

UploadedFile(source_path: Path, dest_path: str, size: int)

Information about a file uploaded to the execution environment.

source_path instance-attribute

source_path: Path

dest_path instance-attribute

dest_path: str

size instance-attribute

size: int

stirrup.tools.code_backends.UploadFilesResult dataclass

UploadFilesResult(
    uploaded: list[UploadedFile] = list(),
    failed: dict[str, str] = dict(),
)

Result of uploading files to the execution environment.

uploaded class-attribute instance-attribute

uploaded: list[UploadedFile] = field(default_factory=list)

failed class-attribute instance-attribute

failed: dict[str, str] = field(default_factory=dict)