Sandbox

SandboxScriptError

Bases: Exception

Base exception for an invalid or incompletely consumed sandbox script.

Source code in src/agents/testing/sandbox.py

class SandboxScriptError(Exception):
    """Base exception for an invalid or incompletely consumed sandbox script."""

InvalidSandboxStep

Bases: SandboxScriptError

Raised when a sandbox step is invalid at factory construction time.

Source code in src/agents/testing/sandbox.py

class InvalidSandboxStep(SandboxScriptError):
    """Raised when a sandbox step is invalid at factory construction time."""

    def __init__(
        self,
        message: str,
        *,
        reason: SandboxStepReason,
        input_index: int,
        method: str | None,
    ) -> None:
        super().__init__(message)
        self.reason = reason
        self.input_index = input_index
        self.method = method

UnexpectedSandboxCall

Bases: SandboxScriptError

Raised when a call does not match the next configured sandbox step.

Source code in src/agents/testing/sandbox.py

class UnexpectedSandboxCall(SandboxScriptError):
    """Raised when a call does not match the next configured sandbox step."""

    def __init__(
        self,
        message: str,
        *,
        call: SandboxCall,
        call_index: int,
        expected_method: str | None,
        remaining_steps: int,
    ) -> None:
        super().__init__(message)
        self.call = call
        self.call_index = call_index
        self.actual_method: str = call.method
        self.expected_method = expected_method
        self.remaining_steps = remaining_steps

SandboxCallMatcherError

Bases: SandboxScriptError

Raised when a sandbox step matcher rejects its call.

Source code in src/agents/testing/sandbox.py

class SandboxCallMatcherError(SandboxScriptError):
    """Raised when a sandbox step matcher rejects its call."""

    def __init__(self, message: str, *, call: SandboxCall, call_index: int) -> None:
        super().__init__(message)
        self.call = call
        self.call_index = call_index
        self.method: str = call.method

UnconsumedSandboxSteps

Bases: SandboxScriptError

Raised when configured sandbox steps remain unconsumed.

Source code in src/agents/testing/sandbox.py

class UnconsumedSandboxSteps(SandboxScriptError):
    """Raised when configured sandbox steps remain unconsumed."""

    def __init__(
        self,
        message: str,
        *,
        remaining_steps: int,
        pending_methods: tuple[str, ...],
    ) -> None:
        super().__init__(message)
        self.remaining_steps = remaining_steps
        self.pending_methods = pending_methods

SandboxCall dataclass

A detached invocation-time sandbox call snapshot.

Source code in src/agents/testing/sandbox.py

@dataclass(frozen=True)
class SandboxCall:
    """A detached invocation-time sandbox call snapshot."""

    call_index: int
    method: str
    args: tuple[Any, ...]
    kwargs: Mapping[str, Any]

SandboxStepSpec

Bases: TypedDict

Dictionary form of one FIFO scripted sandbox call.

Source code in src/agents/testing/sandbox.py

class SandboxStepSpec(TypedDict, total=False):
    """Dictionary form of one FIFO scripted sandbox call."""

    method: SandboxMethod
    match: SandboxMatcher
    result: Any
    responder: SandboxResponder
    error: Exception

ScriptedSandboxSession

Bases: BaseSandboxSession, ABC

The typed result interface for scripted_sandbox_session.

Source code in src/agents/testing/sandbox.py

class ScriptedSandboxSession(BaseSandboxSession, abc.ABC):
    """The typed result interface for ``scripted_sandbox_session``."""

    @property
    @abc.abstractmethod
    def calls(self) -> tuple[SandboxCall, ...]:
        """Return detached call-history snapshots in invocation order."""

    @property
    @abc.abstractmethod
    def remaining_steps(self) -> int:
        """Return the number of configured calls that remain."""

    @abc.abstractmethod
    def assert_complete(self) -> None:
        """Raise when configured sandbox calls remain unconsumed."""

calls abstractmethod property

calls: tuple[SandboxCall, ...]

Return detached call-history snapshots in invocation order.

remaining_steps abstractmethod property

remaining_steps: int

Return the number of configured calls that remain.

assert_complete abstractmethod

assert_complete() -> None

Raise when configured sandbox calls remain unconsumed.

Source code in src/agents/testing/sandbox.py

@abc.abstractmethod
def assert_complete(self) -> None:
    """Raise when configured sandbox calls remain unconsumed."""

stop async

stop() -> None

Persist/snapshot the workspace.

Note: stop() is intentionally persistence-only. Sandboxes that need to tear down sandbox resources (Docker containers, remote sessions, etc.) should implement shutdown() instead.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def stop(self) -> None:
    """
    Persist/snapshot the workspace.

    Note: `stop()` is intentionally persistence-only. Sandboxes that need to tear down
    sandbox resources (Docker containers, remote sessions, etc.) should implement
    `shutdown()` instead.
    """
    validate_manifest_mount_credential_boundaries(
        self.state.manifest,
        provider_backend_id=self.state.type,
    )
    try:
        try:
            await self._before_stop()
            await self._persist_snapshot()
        except Exception as e:
            wrapped = self._wrap_stop_error(e)
            if wrapped is e:
                raise
            raise wrapped from e
    finally:
        await self._after_stop()

supports_docker_volume_mounts

supports_docker_volume_mounts() -> bool

Return whether this backend attaches Docker volume mounts before manifest apply.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def supports_docker_volume_mounts(self) -> bool:
    """Return whether this backend attaches Docker volume mounts before manifest apply."""

    return False

shutdown async

shutdown() -> None

Tear down sandbox resources (best-effort).

Default is a no-op. Sandbox-specific sessions (e.g. Docker) should override.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def shutdown(self) -> None:
    """
    Tear down sandbox resources (best-effort).

    Default is a no-op. Sandbox-specific sessions (e.g. Docker) should override.
    """
    await self._before_shutdown()
    await self._shutdown_backend()
    await self._after_shutdown()

aclose async

aclose() -> None

Run the session cleanup lifecycle outside of async with.

This performs the same session-owned cleanup as __aexit__(): persist/snapshot the workspace via stop(), tear down session resources via shutdown(), and close session-scoped dependencies. If the session came from a sandbox client, call the client's delete() separately for backend-specific deletion such as removing a Docker container or deleting a temporary host workspace.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def aclose(self) -> None:
    """Run the session cleanup lifecycle outside of ``async with``.

    This performs the same session-owned cleanup as ``__aexit__()``: persist/snapshot the
    workspace via ``stop()``, tear down session resources via ``shutdown()``, and close
    session-scoped dependencies. If the session came from a sandbox client, call the client's
    ``delete()`` separately for backend-specific deletion such as removing a Docker container
    or deleting a temporary host workspace.
    """

    lock = self._aclose_lock
    if lock is None:
        lock = asyncio.Lock()
        self._aclose_lock = lock
    async with lock:
        await self._aclose_impl()

register_pre_stop_hook

register_pre_stop_hook(
    hook: Callable[[], Awaitable[None]],
) -> None

Register an async hook to run once before the session workspace is persisted.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def register_pre_stop_hook(self, hook: Callable[[], Awaitable[None]]) -> None:
    """Register an async hook to run once before the session workspace is persisted."""

    hooks = self._pre_stop_hooks
    if hooks is None:
        hooks = []
        self._pre_stop_hooks = hooks
    hooks.append(hook)
    self._pre_stop_hooks_ran = False

run_pre_stop_hooks async

run_pre_stop_hooks() -> None

Run registered pre-stop hooks once before workspace persistence.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def run_pre_stop_hooks(self) -> None:
    """Run registered pre-stop hooks once before workspace persistence."""

    lock = self._pre_stop_hooks_lock
    if lock is None:
        lock = asyncio.Lock()
        self._pre_stop_hooks_lock = lock
    async with lock:
        hooks = self._pre_stop_hooks
        if hooks is None or self._pre_stop_hooks_ran:
            return
        self._pre_stop_hooks_ran = True
        cleanup_error: BaseException | None = None
        for hook in hooks:
            try:
                await hook()
            except BaseException as exc:
                if cleanup_error is None:
                    cleanup_error = exc
        if cleanup_error is not None:
            self._pre_stop_hooks_failed = True
            raise cleanup_error

register_persist_workspace_skip_path

register_persist_workspace_skip_path(
    path: Path | str,
) -> Path

Exclude a runtime-created workspace path from future workspace snapshots.

Use this for session side effects that are not part of durable workspace state, such as generated mount config or ephemeral sink output.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def register_persist_workspace_skip_path(self, path: Path | str) -> Path:
    """Exclude a runtime-created workspace path from future workspace snapshots.

    Use this for session side effects that are not part of durable workspace state, such as
    generated mount config or ephemeral sink output.
    """

    rel_path = Manifest._coerce_rel_path(path)
    Manifest._validate_rel_path(rel_path)
    if rel_path in (Path(""), Path(".")):
        raise ValueError("Persist workspace skip paths must target a concrete relative path.")
    overlapping_mounts = self._overlapping_mount_relpaths(rel_path)
    if overlapping_mounts:
        overlapping_mount = min(overlapping_mounts, key=lambda p: (len(p.parts), p.as_posix()))
        raise MountConfigError(
            message="persist workspace skip path must not overlap mount path",
            context={
                "skip_path": rel_path.as_posix(),
                "mount_path": overlapping_mount.as_posix(),
            },
        )

    if self._runtime_persist_workspace_skip_relpaths is None:
        self._runtime_persist_workspace_skip_relpaths = set()
    self._runtime_persist_workspace_skip_relpaths.add(rel_path)
    return rel_path

exec async

exec(
    *command: str | Path,
    timeout: float | None = None,
    shell: bool | list[str] = True,
    user: str | User | None = None,
) -> ExecResult

Execute a command inside the session.

:param command: Command and args (will be stringified). :param timeout: Optional wall-clock timeout in seconds. :param shell: Whether to run this command in a shell. If True is provided, the command will be run prefixed by sh -lc. A custom shell prefix may be used by providing a list.

:returns: An ExecResult containing stdout/stderr and exit code.

:raises TimeoutError: If the sandbox cannot complete within timeout.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def exec(
    self,
    *command: str | Path,
    timeout: float | None = None,
    shell: bool | list[str] = True,
    user: str | User | None = None,
) -> ExecResult:
    """Execute a command inside the session.

    :param command: Command and args (will be stringified).
    :param timeout: Optional wall-clock timeout in seconds.
    :param shell: Whether to run this command in a shell. If ``True`` is provided,
        the command will be run prefixed by ``sh -lc``. A custom shell prefix may be used
        by providing a list.

    :returns: An ``ExecResult`` containing stdout/stderr and exit code.

    :raises TimeoutError: If the sandbox cannot complete within `timeout`.
    """

    sanitized_command = self._prepare_exec_command(*command, shell=shell, user=user)
    return await self._exec_internal(*sanitized_command, timeout=timeout)

read abstractmethod async

read(
    path: Path, *, user: str | User | None = None
) -> IOBase

Read a file from the session's workspace.

:param path: Absolute path in the container or path relative to the workspace root. :param user: Optional sandbox user to perform the read as. :returns: A readable file-like object. :raises: FileNotFoundError: If the path does not exist.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
    """Read a file from the session's workspace.

    :param path: Absolute path in the container or path relative to the
            workspace root.
    :param user: Optional sandbox user to perform the read as.
    :returns: A readable file-like object.
    :raises: FileNotFoundError: If the path does not exist.
    """

write abstractmethod async

write(
    path: Path,
    data: IOBase,
    *,
    user: str | User | None = None,
) -> None

Write a file into the session's workspace.

:param path: Absolute path in the container or path relative to the workspace root. :param data: A file-like object positioned at the start of the payload. :param user: Optional sandbox user to perform the write as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def write(
    self,
    path: Path,
    data: io.IOBase,
    *,
    user: str | User | None = None,
) -> None:
    """Write a file into the session's workspace.

    :param path: Absolute path in the container or path relative to the
            workspace root.
    :param data: A file-like object positioned at the start of the payload.
    :param user: Optional sandbox user to perform the write as.
    """

running abstractmethod async

running() -> bool

:returns: whether the underlying sandbox is currently running.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def running(self) -> bool:
    """
    :returns: whether the underlying sandbox is currently running.
    """

persist_workspace abstractmethod async

persist_workspace() -> IOBase

Serialize the session's workspace into a byte stream.

:returns: A readable byte stream representing the workspace contents. Portable tar streams must use workspace-relative member paths rather than embedding the source backend's workspace root directory.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def persist_workspace(self) -> io.IOBase:
    """Serialize the session's workspace into a byte stream.

    :returns: A readable byte stream representing the workspace contents.
        Portable tar streams must use workspace-relative member paths rather than
        embedding the source backend's workspace root directory.
    """

hydrate_workspace abstractmethod async

hydrate_workspace(data: IOBase) -> None

Populate the session's workspace from a serialized byte stream.

:param data: A readable byte stream as produced by persist_workspace. Portable tar streams are extracted underneath this session's workspace root.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def hydrate_workspace(self, data: io.IOBase) -> None:
    """Populate the session's workspace from a serialized byte stream.

    :param data: A readable byte stream as produced by `persist_workspace`.
        Portable tar streams are extracted underneath this session's workspace root.
    """

ls async

ls(
    path: Path | str, *, user: str | User | None = None
) -> list[FileEntry]

List directory contents.

:param path: Path to list. :param user: Optional sandbox user to list as. :returns: A list of FileEntry objects.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def ls(
    self,
    path: Path | str,
    *,
    user: str | User | None = None,
) -> list[FileEntry]:
    """List directory contents.

    :param path: Path to list.
    :param user: Optional sandbox user to list as.
    :returns: A list of `FileEntry` objects.
    """
    path = await self._validate_path_access(path)

    path_arg = sandbox_path_str(path)
    cmd = ("ls", "-la", "--", path_arg)
    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

    return parse_ls_la(result.stdout.decode("utf-8", errors="replace"), base=path_arg)

rm async

rm(
    path: Path | str,
    *,
    recursive: bool = False,
    user: str | User | None = None,
) -> None

Remove a file or directory.

:param path: Path to remove. :param recursive: If true, remove directories recursively. :param user: Optional sandbox user to remove as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def rm(
    self,
    path: Path | str,
    *,
    recursive: bool = False,
    user: str | User | None = None,
) -> None:
    """Remove a file or directory.

    :param path: Path to remove.
    :param recursive: If true, remove directories recursively.
    :param user: Optional sandbox user to remove as.
    """
    path = await self._validate_path_access(path, for_write=True)

    cmd: list[str] = ["rm"]
    if recursive:
        cmd.append("-rf")
    cmd.extend(["--", sandbox_path_str(path)])

    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

mkdir async

mkdir(
    path: Path | str,
    *,
    parents: bool = False,
    user: str | User | None = None,
) -> None

Create a directory.

:param path: Directory to create on the remote. :param parents: If true, create missing parents. :param user: Optional sandbox user to create the directory as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def mkdir(
    self,
    path: Path | str,
    *,
    parents: bool = False,
    user: str | User | None = None,
) -> None:
    """Create a directory.

    :param path: Directory to create on the remote.
    :param parents: If true, create missing parents.
    :param user: Optional sandbox user to create the directory as.
    """
    path = await self._validate_path_access(path, for_write=True)

    cmd: list[str] = ["mkdir"]
    if parents:
        cmd.append("-p")
    cmd.append(sandbox_path_str(path))

    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

extract async

extract(
    path: Path | str,
    data: IOBase,
    *,
    compression_scheme: Literal["tar", "zip"] | None = None,
    archive_limits: SandboxArchiveLimits | None = None,
) -> None

Write a compressed archive to a destination on the remote. Optionally extract the archive once written.

:param path: Path on the host machine to extract to :param data: a file-like io stream. :param compression_scheme: either "tar" or "zip". If not provided, it will try to infer from the path. :param archive_limits: optional per-call archive resource limits. If omitted, the session default is used.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def extract(
    self,
    path: Path | str,
    data: io.IOBase,
    *,
    compression_scheme: Literal["tar", "zip"] | None = None,
    archive_limits: SandboxArchiveLimits | None = None,
) -> None:
    """
    Write a compressed archive to a destination on the remote.
    Optionally extract the archive once written.

    :param path: Path on the host machine to extract to
    :param data: a file-like io stream.
    :param compression_scheme: either "tar" or "zip". If not provided,
        it will try to infer from the path.
    :param archive_limits: optional per-call archive resource limits. If omitted,
        the session default is used.
    """
    if archive_limits is not None:
        archive_limits.validate()
    effective_archive_limits = (
        archive_limits if archive_limits is not None else self._archive_limits
    )

    await archive_ops.extract_archive(
        self,
        path,
        data,
        compression_scheme=compression_scheme,
        archive_limits=effective_archive_limits,
    )

should_provision_manifest_accounts_on_resume

should_provision_manifest_accounts_on_resume() -> bool

Return whether resume should reprovision manifest-managed users and groups.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def should_provision_manifest_accounts_on_resume(self) -> bool:
    """Return whether resume should reprovision manifest-managed users and groups."""

    return not self._system_state_preserved_on_start()

scripted_sandbox_session

scripted_sandbox_session(
    steps: Iterable[
        SandboxStepSpec | Mapping[str, Any]
    ] = (),
    *,
    manifest: Manifest | None = None,
) -> ScriptedSandboxSession

Create a deterministic provider-free sandbox session for agent workflow tests.

Each FIFO step defines method plus exactly one of result, responder, or error. An optional match callable receives a detached SandboxCall. The returned object is the session itself, so pass it directly to SandboxRunConfig(session=session). Only configured model-facing methods are visible. The two PTY methods are exposed together when either one is configured because they form one advertised session capability. Use a custom BaseSandboxSession or a real provider for lifecycle, persistence, mount, or broader filesystem behavior.

Source code in src/agents/testing/sandbox.py

def scripted_sandbox_session(
    steps: Iterable[SandboxStepSpec | Mapping[str, Any]] = (),
    *,
    manifest: Manifest | None = None,
) -> ScriptedSandboxSession:
    """Create a deterministic provider-free sandbox session for agent workflow tests.

    Each FIFO step defines ``method`` plus exactly one of ``result``, ``responder``, or ``error``.
    An optional ``match`` callable receives a detached ``SandboxCall``. The returned object is the
    session itself, so pass it directly to ``SandboxRunConfig(session=session)``. Only configured
    model-facing methods are visible. The two PTY methods are exposed together when either one is
    configured because they form one advertised session capability. Use a custom
    ``BaseSandboxSession`` or a real provider for lifecycle, persistence, mount, or broader
    filesystem behavior.
    """
    normalized = [_normalize_step(step, index) for index, step in enumerate(steps)]
    return _ScriptedSandboxSession(normalized, manifest=manifest)