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
)

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)

All code execution providers support an optional allowlist of command patterns. If provided, only commands matching at least one pattern are allowed. If None, all commands are allowed.

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. If provided, only commands matching at least one pattern are allowed. If None, all commands are allowed.	None

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

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

    Args:
        allowed_commands: Optional list of regex patterns. If provided, only
                         commands matching at least one pattern are allowed.
                         If None, all commands are allowed.

    """
    self._allowed_commands = allowed_commands
    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 = SHELL_TIMEOUT
) -> CommandResult

Execute a shell command and return raw CommandResult.

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

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

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.

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.

    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)

    for source_path in paths:
        try:
            content = await self.read_file_bytes(source_path)
            filename = Path(source_path).name
            dest_path = f"{output_dir_str}/{filename}"

            if dest_env:
                # Transfer to another exec env (cross-environment)
                logger.debug(
                    "CROSS-ENV TRANSFER: %s (%d bytes) -> %s (dest_env: %s)",
                    source_path,
                    len(content),
                    dest_path,
                    type(dest_env).__name__,
                )
                await dest_env.write_file_bytes(dest_path, content)
                result.saved.append(SavedFile(source_path, Path(dest_path), len(content)))
            else:
                # Save to local filesystem
                output_path = Path(output_dir) / filename
                output_path.parent.mkdir(parents=True, exist_ok=True)
                logger.debug(
                    "SAVE TO LOCAL: %s (%d bytes) -> %s",
                    source_path,
                    len(content),
                    output_path,
                )
                output_path.write_bytes(content)
                result.saved.append(SavedFile(source_path, output_path, len(content)))
        except Exception as e:
            logger.debug("TRANSFER FAILED: %s -> %s: %s", source_path, output_dir_str, e)
            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
                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]:
        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.",
                metadata=ToolUseCountMetadata(),
            )
        except ValueError as e:
            return ToolResult(
                content=str(e),
                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,
)

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. If provided, only commands matching at least one pattern are allowed. If None, all commands are allowed.	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

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,
) -> None:
    """Initialize LocalCodeExecToolProvider configuration.

    Args:
        allowed_commands: Optional list of regex patterns. If provided, only
                         commands matching at least one pattern are allowed.
                         If None, all commands are allowed.
        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.

    """
    super().__init__(allowed_commands=allowed_commands)
    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.info("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)
    resolved.parent.mkdir(parents=True, exist_ok=True)
    resolved.write_bytes(content)

run_command async

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

Execute command in the temp directory.

Parameters:

Name	Type	Description	Default
cmd	str	Shell command to execute (bash syntax).	required
timeout	int	Maximum time in seconds to wait for command completion.	SHELL_TIMEOUT

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 = SHELL_TIMEOUT) -> 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.

    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."
        )

    # Check allowlist
    if not self._check_allowed(cmd):
        return CommandResult(
            exit_code=1,
            stdout="",
            stderr=f"Command not allowed: '{cmd}' does not match any allowed patterns",
            error_kind="command_not_allowed",
            advice="Only commands matching the allowlist patterns are permitted.",
        )

    # 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):
            # Use shell=True by wrapping in a shell command
            process = await anyio.open_process(
                ["bash", "-c", cmd],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                cwd=self._temp_dir,
            )

            # 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:
            process.kill()
        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

Move files from the temp directory to a destination.

When dest_env is None (local filesystem), files are MOVED (not copied) - originals are deleted from the execution environment. Existing files in output_dir are silently overwritten.

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:
    """Move files from the temp directory to a destination.

    When dest_env is None (local filesystem), files are MOVED (not copied) -
    originals are deleted from the execution environment.
    Existing files in output_dir are silently overwritten.

    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)

    # Local filesystem - use optimized move operation
    output_dir_path = Path(output_dir)
    output_dir_path.mkdir(parents=True, exist_ok=True)

    result = SaveOutputFilesResult()

    for source_path_str in paths:
        try:
            source_path = Path(source_path_str)
            if not source_path.is_absolute():
                source_path = self._temp_dir / source_path

            # Security: ensure path is within temp directory
            try:
                source_path.resolve().relative_to(self._temp_dir.resolve())
            except ValueError:
                result.failed[source_path_str] = "Path is outside execution environment directory"
                logger.warning("Attempted to access path outside execution environment: %s", source_path_str)
                continue

            if not source_path.exists():
                result.failed[source_path_str] = "File does not exist"
                logger.warning("Execution environment file does not exist: %s", source_path_str)
                continue

            if not source_path.is_file():
                result.failed[source_path_str] = "Path is not a file"
                logger.warning("Execution environment path is not a file: %s", source_path_str)
                continue

            file_size = source_path.stat().st_size
            dest_path = output_dir_path / source_path.name

            # Move file (overwrites if exists)
            shutil.move(str(source_path), str(dest_path))
            logger.info("Moved file: %s -> %s", source_path, dest_path)

            result.saved.append(
                SavedFile(
                    source_path=source_path_str,
                    output_path=dest_path,
                    size=file_size,
                ),
            )

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

    return result

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():
                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: Path, size: int)

Information about a file saved from the execution environment.

source_path instance-attribute

source_path: str

output_path instance-attribute

output_path: Path

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)