Sandbox clients

BaseSandboxClient

Bases: ABC, Generic[ClientOptionsT]

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

class BaseSandboxClient(abc.ABC, Generic[ClientOptionsT]):
    backend_id: str
    supports_default_options: bool = False
    _dependencies: Dependencies | None = None

    def _resolve_dependencies(self) -> Dependencies | None:
        if self._dependencies is None:
            return None
        # Sessions get clones instead of the shared template so per-session factory caches and
        # owned resources do not leak across unrelated sandboxes.
        return self._dependencies.clone()

    def _wrap_session(
        self,
        inner: BaseSandboxSession,
        *,
        instrumentation: Instrumentation | None = None,
    ) -> SandboxSession:
        # Always return the instrumented wrapper so callers get consistent events and dependency
        # lifecycle handling regardless of which backend created the inner session.
        return SandboxSession(
            inner,
            instrumentation=instrumentation,
            dependencies=self._resolve_dependencies(),
        )

    def _validate_manifest_for_create(
        self,
        manifest: Manifest,
    ) -> Manifest:
        from .._mount_security import validate_manifest_mount_credential_boundaries

        validate_manifest_mount_credential_boundaries(
            manifest,
            provider_backend_id=self.backend_id,
        )
        return manifest

    @abc.abstractmethod
    async def create(
        self,
        *,
        snapshot: SnapshotSpec | SnapshotBase | None = None,
        manifest: Manifest | None = None,
        options: ClientOptionsT,
    ) -> SandboxSession:
        """Create a new session.

        Args:
            snapshot: Snapshot or spec used to create a snapshot instance for
                the session. If omitted, the session uses a no-op snapshot.
            manifest: Optional manifest to materialize into the workspace when
                the session starts.
            options: Sandbox-specific settings. For example, Docker expects
                ``DockerSandboxClientOptions(image="...")``.
        Returns:
            A `SandboxSession` that can be entered with `async with` or closed explicitly with
            `await session.aclose()`.
        """

    @abc.abstractmethod
    async def delete(self, session: SandboxSession) -> SandboxSession:
        """Delete a session and release sandbox resources."""

    @abc.abstractmethod
    async def resume(
        self,
        state: SandboxSessionState,
    ) -> SandboxSession:
        """Resume an owning session from a previously persisted `SandboxSessionState`.

        Providers should first try to reattach to the backend sandbox identified
        by `state`. If that resource still exists, including after unclean
        process/client shutdown where `delete()` was never called, the returned
        session should target the same backend sandbox and be able to clean it
        up later.

        If the original backend sandbox is unavailable, providers may create a
        replacement and should hydrate its workspace from `state.snapshot`
        during `SandboxSession.start()`.

        The returned session owns its provider lifecycle; pass a live
        `session=` when you want to reuse an already-running sandbox session.
        """

    @redact_mount_error_data_sync
    def serialize_session_state(self, state: SandboxSessionState) -> dict[str, object]:
        """Serialize backend-specific sandbox state into a JSON-compatible payload."""
        from ...exceptions import _raise_data_redacted_error
        from .._mount_security import (
            _manifest_has_configured_mount_authority,
            _manifest_mount_provenance_error,
            _mark_mount_validation_error,
            _redact_mount_serialization_error,
        )

        provenance_error = _manifest_mount_provenance_error(state.manifest)
        if provenance_error is not None:
            _mark_mount_validation_error(provenance_error)
            state = cast(Any, None)
            _raise_data_redacted_error(provenance_error)

        try:
            return self._serialize_session_state(state)
        except MountConfigError:
            raise
        except Exception as error:
            if not _manifest_has_configured_mount_authority(state.manifest):
                raise
            raise _redact_mount_serialization_error(error) from None

    def _serialize_session_state(self, state: SandboxSessionState) -> dict[str, object]:
        redacted_paths = set(state.path_grants_require_rebind)
        persistent_grants = []
        for grant in state.manifest.extra_path_grants:
            if grant.host_path is not None:
                redacted_paths.add(grant.path)
                continue
            persistent_grants.append(grant)

        persistent_manifest = state.manifest.model_copy(
            update={"extra_path_grants": tuple(persistent_grants)},
        )
        persistent_state = state.model_copy(update={"manifest": persistent_manifest})
        payload = cast(dict[str, object], persistent_state.model_dump(mode="json"))
        if redacted_paths:
            payload[REDACTED_HOST_PATH_GRANT_PATHS_KEY] = sorted(redacted_paths)
        return payload

    @staticmethod
    def _deserialize_session_state_payload(
        payload: Mapping[str, object],
        state_class: type[SandboxSessionState],
    ) -> SandboxSessionState:
        from ...exceptions import _replace_data_redacted_process_control_error
        from .._mount_security import (
            _redact_mount_state_validation_error,
            sanitize_raw_session_state_mount_authority,
        )

        safe_error: BaseException | None = None
        persisted_payload: dict[str, object] | None = None
        try:
            sanitized, _redacted = sanitize_raw_session_state_mount_authority(payload)
            if not isinstance(sanitized, dict):
                raise TypeError("sandbox session state payload must be a mapping")
            persisted_payload = sanitized
            if isinstance(payload, dict):
                payload.clear()
                payload.update(sanitized)
                persisted_payload = payload
            state = state_class.model_validate(persisted_payload)
        except BaseException as error:
            safe_error = _replace_data_redacted_process_control_error(error)
            if safe_error is None:
                safe_error = _redact_mount_state_validation_error(
                    error,
                    message="sandbox session state payload is invalid",
                )

        if safe_error is not None:
            if isinstance(payload, dict):
                payload.clear()
            sanitized = cast(Any, None)
            persisted_payload = None
            state_class = cast(Any, None)
            _raise_data_redacted_error(safe_error)
        assert persisted_payload is not None
        return SandboxSessionState._mark_persisted_path_grants(state, payload=persisted_payload)

    @abc.abstractmethod
    def deserialize_session_state(self, payload: dict[str, object]) -> SandboxSessionState:
        """Deserialize backend-specific sandbox state from a JSON-compatible payload."""

create abstractmethod async

create(
    *,
    snapshot: SnapshotSpec | SnapshotBase | None = None,
    manifest: Manifest | None = None,
    options: ClientOptionsT,
) -> SandboxSession

Create a new session.

Parameters:

Name	Type	Description	Default
snapshot	SnapshotSpec | SnapshotBase | None	Snapshot or spec used to create a snapshot instance for the session. If omitted, the session uses a no-op snapshot.	None
manifest	Manifest | None	Optional manifest to materialize into the workspace when the session starts.	None
options	ClientOptionsT	Sandbox-specific settings. For example, Docker expects DockerSandboxClientOptions(image="...").	required

Returns: A SandboxSession that can be entered with async with or closed explicitly with await session.aclose().

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

@abc.abstractmethod
async def create(
    self,
    *,
    snapshot: SnapshotSpec | SnapshotBase | None = None,
    manifest: Manifest | None = None,
    options: ClientOptionsT,
) -> SandboxSession:
    """Create a new session.

    Args:
        snapshot: Snapshot or spec used to create a snapshot instance for
            the session. If omitted, the session uses a no-op snapshot.
        manifest: Optional manifest to materialize into the workspace when
            the session starts.
        options: Sandbox-specific settings. For example, Docker expects
            ``DockerSandboxClientOptions(image="...")``.
    Returns:
        A `SandboxSession` that can be entered with `async with` or closed explicitly with
        `await session.aclose()`.
    """

delete abstractmethod async

delete(session: SandboxSession) -> SandboxSession

Delete a session and release sandbox resources.

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

@abc.abstractmethod
async def delete(self, session: SandboxSession) -> SandboxSession:
    """Delete a session and release sandbox resources."""

resume abstractmethod async

resume(state: SandboxSessionState) -> SandboxSession

Resume an owning session from a previously persisted SandboxSessionState.

Providers should first try to reattach to the backend sandbox identified by state. If that resource still exists, including after unclean process/client shutdown where delete() was never called, the returned session should target the same backend sandbox and be able to clean it up later.

If the original backend sandbox is unavailable, providers may create a replacement and should hydrate its workspace from state.snapshot during SandboxSession.start().

The returned session owns its provider lifecycle; pass a live session= when you want to reuse an already-running sandbox session.

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

@abc.abstractmethod
async def resume(
    self,
    state: SandboxSessionState,
) -> SandboxSession:
    """Resume an owning session from a previously persisted `SandboxSessionState`.

    Providers should first try to reattach to the backend sandbox identified
    by `state`. If that resource still exists, including after unclean
    process/client shutdown where `delete()` was never called, the returned
    session should target the same backend sandbox and be able to clean it
    up later.

    If the original backend sandbox is unavailable, providers may create a
    replacement and should hydrate its workspace from `state.snapshot`
    during `SandboxSession.start()`.

    The returned session owns its provider lifecycle; pass a live
    `session=` when you want to reuse an already-running sandbox session.
    """

serialize_session_state

serialize_session_state(
    state: SandboxSessionState,
) -> dict[str, object]

Serialize backend-specific sandbox state into a JSON-compatible payload.

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

@redact_mount_error_data_sync
def serialize_session_state(self, state: SandboxSessionState) -> dict[str, object]:
    """Serialize backend-specific sandbox state into a JSON-compatible payload."""
    from ...exceptions import _raise_data_redacted_error
    from .._mount_security import (
        _manifest_has_configured_mount_authority,
        _manifest_mount_provenance_error,
        _mark_mount_validation_error,
        _redact_mount_serialization_error,
    )

    provenance_error = _manifest_mount_provenance_error(state.manifest)
    if provenance_error is not None:
        _mark_mount_validation_error(provenance_error)
        state = cast(Any, None)
        _raise_data_redacted_error(provenance_error)

    try:
        return self._serialize_session_state(state)
    except MountConfigError:
        raise
    except Exception as error:
        if not _manifest_has_configured_mount_authority(state.manifest):
            raise
        raise _redact_mount_serialization_error(error) from None

deserialize_session_state abstractmethod

deserialize_session_state(
    payload: dict[str, object],
) -> SandboxSessionState

Deserialize backend-specific sandbox state from a JSON-compatible payload.

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

@abc.abstractmethod
def deserialize_session_state(self, payload: dict[str, object]) -> SandboxSessionState:
    """Deserialize backend-specific sandbox state from a JSON-compatible payload."""

BaseSandboxClientOptions

Bases: BaseModel

Polymorphic base for sandbox client options that need JSON round-trips.

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

class BaseSandboxClientOptions(BaseModel):
    """Polymorphic base for sandbox client options that need JSON round-trips."""

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    type: str
    _subclass_registry: ClassVar[dict[str, SandboxClientOptionsClass]] = {}

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        if args:
            positional_fields = [name for name in type(self).model_fields if name != "type"]
            if len(args) > len(positional_fields):
                raise TypeError(
                    f"{type(self).__name__}() takes at most {len(positional_fields)} positional "
                    f"arguments but {len(args)} were given"
                )
            for field_name, value in zip(positional_fields, args, strict=False):
                if field_name in kwargs:
                    raise TypeError(
                        f"{type(self).__name__}() got multiple values for argument {field_name!r}"
                    )
                kwargs[field_name] = value
        super().__init__(**kwargs)

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: object) -> None:
        super().__pydantic_init_subclass__(**kwargs)

        type_field = cls.model_fields.get("type")
        type_default = type_field.default if type_field is not None else None
        if not isinstance(type_default, str) or type_default == "":
            raise TypeError(f"{cls.__name__} must define a non-empty string default for `type`")

        existing = BaseSandboxClientOptions._subclass_registry.get(type_default)
        if (
            existing is not None
            and existing is not cls
            and (existing.__module__, existing.__qualname__) != (cls.__module__, cls.__qualname__)
        ):
            raise TypeError(
                f"sandbox client options type `{type_default}` is already registered by "
                f"{existing.__name__}"
            )
        if existing is not None:
            return
        BaseSandboxClientOptions._subclass_registry[type_default] = cls

    @classmethod
    def parse(cls, payload: object) -> BaseSandboxClientOptions:
        if isinstance(payload, BaseSandboxClientOptions):
            return payload

        if isinstance(payload, dict):
            options_type = payload.get("type")
            if isinstance(options_type, str):
                options_class = cls._options_class_for_type(options_type)
                if options_class is not None:
                    return options_class.model_validate(payload)

            raise ValueError(f"unknown sandbox client options type `{options_type}`")

        raise TypeError(
            "sandbox client options payload must be a BaseSandboxClientOptions or object payload"
        )

    @model_serializer(mode="wrap")
    def _serialize_always_include_type(self, handler: Any) -> dict[str, Any]:
        data = handler(self)
        if isinstance(data, dict):
            data["type"] = self.type
        return cast(dict[str, Any], data)

    @classmethod
    def _options_class_for_type(
        cls,
        options_type: str,
    ) -> SandboxClientOptionsClass | None:
        return BaseSandboxClientOptions._subclass_registry.get(options_type)