Docker API

DockerSandbox

pydantic_ai_backends.backends.docker.sandbox.DockerSandbox

Bases: BaseSandbox

Docker-based sandbox for isolated command execution.

The container starts lazily on the first operation. File transfers use Docker's archive API rather than shell heredocs, so content with quotes, newlines or arbitrary bytes survives a round trip intact.

Example

Python

from pydantic_ai_backends import DockerSandbox, RuntimeConfig

sandbox = DockerSandbox(image="python:3.12-slim")

ml_runtime = RuntimeConfig(
    name="ml-env",
    base_image="python:3.12-slim",
    packages=["torch", "transformers"],
)
sandbox = DockerSandbox(runtime=ml_runtime)

Source code in src/pydantic_ai_backends/backends/docker/sandbox.py

class DockerSandbox(BaseSandbox):
    """Docker-based sandbox for isolated command execution.

    The container starts lazily on the first operation. File transfers use
    Docker's archive API rather than shell heredocs, so content with quotes,
    newlines or arbitrary bytes survives a round trip intact.

    Example:
        ```python
        from pydantic_ai_backends import DockerSandbox, RuntimeConfig

        sandbox = DockerSandbox(image="python:3.12-slim")

        ml_runtime = RuntimeConfig(
            name="ml-env",
            base_image="python:3.12-slim",
            packages=["torch", "transformers"],
        )
        sandbox = DockerSandbox(runtime=ml_runtime)
        ```
    """

    def __init__(
        self,
        image: str = "python:3.12-slim",
        sandbox_id: str | None = None,
        work_dir: str = "/workspace",
        auto_remove: bool = True,
        runtime: RuntimeConfig | str | None = None,
        session_id: str | None = None,
        idle_timeout: int = 3600,
        volumes: dict[str, str] | None = None,
        network_mode: str | None = None,
        container_name: str | None = None,
        mem_limit: str | None = None,
        memswap_limit: str | None = None,
        cpus: float | None = None,
        cpu_shares: int | None = None,
        pids_limit: int | None = DEFAULT_PIDS_LIMIT,
        tmpfs: dict[str, str] | None = None,
        max_read_bytes: int = DEFAULT_MAX_READ_BYTES,
        oci_runtime: str | None = None,
    ):
        """Initialize the sandbox without starting its container.

        Args:
            image: Docker image to use. Ignored when `runtime` is given.
            sandbox_id: Unique identifier for this sandbox.
            work_dir: Working directory inside the container. Ignored when
                `runtime` is given.
            auto_remove: Remove the container when it stops. Forced to `False`
                when `container_name` is set, since a named container exists to
                be reused.
            runtime: `RuntimeConfig`, or the name of a built-in runtime.
            session_id: Alias for `sandbox_id`, for session management.
            idle_timeout: Idle seconds after which `SessionManager` may reap it.
            volumes: Host-to-container mounts, as `{"/host": "/container"}`.
            network_mode: Docker network mode (`"bridge"`, `"none"`, `"host"`,
                `"container:<name|id>"`). Pass `"none"` for sandboxes that must
                not reach the network; it also skips per-container veth and
                firewall setup, so containers start measurably faster.
            container_name: Stable name to reattach to across restarts, which
                preserves installed packages and other filesystem state.
                Implies `auto_remove=False`.
            mem_limit: Memory ceiling in Docker syntax (`"512m"`, `"2g"`). Swap
                is pinned to the same value unless `memswap_limit` says
                otherwise, so a container over its ceiling is stopped rather
                than left swapping against the host.
            memswap_limit: Ceiling on memory *and* swap combined, in the same
                syntax. `None` pins it to `mem_limit`, which denies the container
                swap entirely — the right default, because a container swapping
                past its limit against a disk starves every other sandbox on the
                host.

                It is the wrong default on a host backed by `zram`, where swap
                is compressed RAM: the pages never leave memory, idle Python
                heaps compress to roughly a third, and the alternative to a
                little swapping is an OOM kill. Set this above `mem_limit` there
                and nowhere else. Ignored without `mem_limit`, since Docker
                rejects a swap ceiling with no memory ceiling under it.
            cpus: Hard CPU ceiling in cores, e.g. `1.5`. A container never
                exceeds it, which also means it cannot use cores that are sitting
                idle — on a small host that is often the wrong trade.
            cpu_shares: Relative CPU weight (Docker's default is 1024). Unlike
                `cpus` this only applies under contention, so one active sandbox
                may use the whole machine and several are still divided fairly.
                Composes with `cpus` when both are set.
            pids_limit: Maximum number of processes. `None` disables the limit.
            tmpfs: In-memory mounts, as `{"/tmp": "size=64m"}`. Writes to a
                tmpfs never reach the container's write layer, so scratch files
                are both faster and free of disk growth. `exec` is added to the
                options because Docker mounts a tmpfs `noexec`, which breaks
                installing any package that builds from source.

                Its pages count against `mem_limit`, not on top of it: a sandbox
                that fills a 64m `/tmp` has that much less left for its own
                processes, and one that tries to exceed the limit through `/tmp`
                is killed by its own cgroup rather than troubling the host.
            max_read_bytes: Largest file `read`/`read_bytes`/`edit` will pull
                out of the container. Oversized files are refused instead of
                being buffered into the host's memory.
            oci_runtime: Low-level runtime the daemon starts this container
                with — Docker's `--runtime`. `None` takes the daemon's default,
                normally `runc`.

                This is the one knob that changes the *isolation boundary*
                rather than a resource ceiling, which is why it is per sandbox:
                `"runsc"` (gVisor) moves syscall handling into userspace and
                `"kata"` gives the container its own kernel in a microVM, while
                a container under plain `runc` shares the host's. Untrusted
                model-written code is exactly the workload that argues for one
                of them.

                The runtime must already be registered with the daemon in
                `/etc/docker/daemon.json`; naming an unregistered one makes the
                daemon refuse to start the container. See the installation docs
                for the host side, including `crun` as a faster drop-in default.
        """
        super().__init__(session_id or sandbox_id)

        self._container_name = container_name
        self._auto_remove = False if container_name else auto_remove
        self._container: Container | None = None
        self._idle_timeout = idle_timeout
        self._last_activity = time.time()
        self._volumes = volumes or {}
        self._network_mode = network_mode
        self._mem_limit = mem_limit
        self._memswap_limit = memswap_limit
        self._cpus = cpus
        self._cpu_shares = cpu_shares
        self._pids_limit = pids_limit
        self._tmpfs = tmpfs or {}
        self._max_read_bytes = max_read_bytes
        self._oci_runtime = oci_runtime
        self._alive = False
        self._alive_checked_at: float | None = None

        if isinstance(runtime, str):
            from pydantic_ai_backends.backends.docker.runtimes import get_runtime

            runtime = get_runtime(runtime)
        self._runtime = runtime
        self._image = image
        self._work_dir = runtime.work_dir if runtime is not None else work_dir

    @property
    def runtime(self) -> RuntimeConfig | None:
        """The runtime configuration for this sandbox."""
        return self._runtime

    @property
    def session_id(self) -> str:
        """Alias for the sandbox id, used for session management."""
        return self._id

    @property
    def idle_timeout(self) -> int:
        """Idle seconds after which `SessionManager` may reap this sandbox."""
        return self._idle_timeout

    def _resolve_path(self, path: str) -> str:
        """Resolve a relative path against the container's working directory."""
        if not PurePosixPath(path).is_absolute():
            return str(PurePosixPath(self._work_dir) / path)
        return path

    # ── Container lifecycle ────────────────────────────────────────────

    def start(self) -> None:
        """Start the container now instead of on the first operation."""
        self._ensure_container()

    def _ensure_container(self) -> None:
        """Attach to or create the container backing this sandbox."""
        if self._container is not None:
            return

        # Everything below attaches or creates a container, so any cached
        # liveness answer belongs to a container that is no longer ours.
        self._alive_checked_at = None

        # Resolved before the submodule import so a missing optional dependency
        # surfaces the install hint instead of a bare ImportError.
        client = docker_client()

        existing = self._reattach(client)
        if existing is not None:
            self._container = existing
            return

        image = resolve_image(client, self._runtime, self._image)
        self._container = client.containers.run(image, **self._run_kwargs())

    def _reattach(self, client: DockerClient) -> Container | None:
        """Return the running named container for this sandbox, if there is one.

        A stopped container is started rather than replaced, so installed
        packages, caches and other filesystem state survive a restart.
        """
        import docker.errors

        if not self._container_name:
            return None

        try:
            existing = client.containers.get(self._container_name)
        except docker.errors.NotFound:
            return None

        if existing.status == "running":
            return existing
        if existing.status in REATTACHABLE_STATUSES:
            existing.start()
            return existing
        # Dead or being removed: a fresh container is the only way forward.
        return None

    def _environment(self) -> dict[str, str]:
        """What the container starts with: the sandbox defaults, then the runtime's.

        `UV_SYSTEM_PYTHON` is dropped for a runtime that runs unprivileged. A
        container's environment overrides its image's, so leaving it set would
        clobber the `0` the image asks for and send uv at the interpreter the
        sandbox user cannot write to — which fails with `Permission denied` and
        no way forward, the virtualenv built for exactly this being ignored.
        """
        env = dict(SANDBOX_ENV)
        if self._runtime is None:
            return env
        if self._runtime.run_as_uid is not None:
            del env["UV_SYSTEM_PYTHON"]
        env.update(self._runtime.env_vars)
        return env

    def _run_kwargs(self) -> dict[str, Any]:
        """Arguments for `containers.run`, including limits and hardening."""
        kwargs: dict[str, Any] = {
            "command": "sleep infinity",
            "detach": True,
            # `sleep` as PID 1 never calls `wait()`, so every process an agent
            # orphans — a backgrounded server, anything the command timeout
            # kills — is reparented to it and stays a zombie for the life of the
            # container. Measured: ten orphans, ten permanent zombies. They
            # accumulate against `pids_limit` until the session cannot fork at
            # all. `init` puts a real reaper in front, for 488 kB.
            "init": True,
            "working_dir": self._work_dir,
            "auto_remove": self._auto_remove,
            "environment": self._environment(),
            "volumes": {
                host: {"bind": container, "mode": "rw"} for host, container in self._volumes.items()
            }
            or None,
            # Sandboxed code is untrusted by definition, so deny it the one
            # cheap escalation route a container still leaves open: gaining
            # privileges by exec'ing a setuid binary.
            "security_opt": ["no-new-privileges:true"],
        }
        if self._container_name is not None:
            kwargs["name"] = self._container_name
        if self._runtime is not None and self._runtime.run_as_uid is not None:
            # Both halves of the pair, because a process writing into a
            # bind-mounted workspace is checked on its gid as well.
            kwargs["user"] = f"{self._runtime.run_as_uid}:{self._runtime.run_as_uid}"
        if self._network_mode is not None:
            kwargs["network_mode"] = self._network_mode
        if self._pids_limit is not None:
            kwargs["pids_limit"] = self._pids_limit
        if self._mem_limit is not None:
            # Without a matching swap ceiling the kernel lets a container over
            # its memory limit swap instead, which starves the whole host. A
            # host whose swap is `zram` can afford a wider one, and says so.
            kwargs["mem_limit"] = self._mem_limit
            kwargs["memswap_limit"] = self._memswap_limit or self._mem_limit
        if self._cpus is not None:
            kwargs["nano_cpus"] = int(self._cpus * 1_000_000_000)
        if self._cpu_shares is not None:
            kwargs["cpu_shares"] = self._cpu_shares
        if self._tmpfs:
            kwargs["tmpfs"] = {path: _with_exec(options) for path, options in self._tmpfs.items()}
        if self._oci_runtime is not None:
            kwargs["runtime"] = self._oci_runtime
        return kwargs

    def is_alive(self) -> bool:
        """Whether the container is running.

        The answer is cached for `ALIVE_CACHE_SECONDS`, since `reload()` is a
        daemon round trip and session managers call this on every request.
        """
        if self._container is None:
            return False

        now = time.monotonic()
        checked_at = self._alive_checked_at
        if checked_at is not None and now - checked_at < ALIVE_CACHE_SECONDS:
            return self._alive

        try:
            self._container.reload()
            status: str = self._container.status
        except Exception:
            self._alive = False
        else:
            self._alive = status == "running"

        self._alive_checked_at = now
        return self._alive

    def resource_usage(self) -> SandboxUsage | None:
        """Sample the container's current resource usage.

        One non-streaming `stats()` call, which costs a daemon round trip and
        should be polled sparingly rather than per request.
        """
        if self._container is None:
            return None
        try:
            return parse_usage(self._container.stats(stream=False))
        except Exception:
            return None

    def stop(self, purge: bool = False, *, remove: bool | None = None) -> None:
        """Stop the container.

        A container created without `container_name` runs with
        `auto_remove=True` and is discarded by the daemon on exit. A *named*
        container deliberately survives, since reuse across restarts is the
        whole point of naming it.

        Args:
            purge: Also remove the container, discarding its filesystem state.
                Named `purge` so that one call site can end any sandbox this
                library offers - `RemoteSandbox`, `DaytonaSandbox` and the
                Kubernetes pod all spell the same idea this way, and this one
                used to spell it `remove`. A caller holding "a sandbox" could
                not call `stop` without knowing which it had.
            remove: The old name for `purge`, still honoured so nothing that
                passes it breaks. Deprecated; pass `purge` instead.
        """
        if remove is not None:
            warnings.warn(
                "DockerSandbox.stop(remove=...) is deprecated; pass purge=... instead, "
                "which is what every other sandbox calls the same argument.",
                DeprecationWarning,
                stacklevel=2,
            )
            purge = remove

        container = getattr(self, "_container", None)
        if container is None:
            return

        with contextlib.suppress(Exception):
            container.stop()
        if purge:
            with contextlib.suppress(Exception):
                container.remove(force=True)
        self._container = None
        self._alive_checked_at = None

    def __del__(self) -> None:
        """Best-effort cleanup on garbage collection.

        `__del__` is unreliable for this — it may run during interpreter
        shutdown when modules are already torn down, or never run at all. Prefer
        the explicit :meth:`stop` lifecycle.
        """
        with contextlib.suppress(Exception):
            if getattr(self, "_container", None) is not None:
                self.stop()

    # ── Commands ───────────────────────────────────────────────────────

    def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Run a command in the container.

        Output beyond `MAX_EXECUTE_OUTPUT_BYTES` is discarded before decoding,
        so the cap is measured in bytes rather than characters.
        """
        self._ensure_container()
        self._last_activity = time.time()
        assert self._container is not None

        # The Docker SDK's exec_run takes no timeout, so the command is wrapped
        # in the `timeout` utility instead.
        argv = ["sh", "-c", command]
        if timeout is not None:
            argv = ["timeout", str(timeout), *argv]

        try:
            exit_code, output = self._container.exec_run(argv, workdir=self._work_dir)
            if not isinstance(output, bytes):
                output = b"".join(output)
        except Exception as e:
            return ExecuteResponse(output=f"Error: {e}", exit_code=1, truncated=False)

        # Sliced before decoding: decoding the whole payload only to throw most
        # of it away doubled peak memory on commands like `cat big.log`.
        return ExecuteResponse(
            output=output[:MAX_EXECUTE_OUTPUT_BYTES].decode("utf-8", errors="replace"),
            exit_code=exit_code,
            truncated=len(output) > MAX_EXECUTE_OUTPUT_BYTES,
        )

    # ── Files ──────────────────────────────────────────────────────────

    def read_bytes(self, path: str) -> bytes:
        """Read a whole file as bytes.

        Returns:
            The content, or `b""` when the file is missing, unreadable, or over
            `max_read_bytes`. Use `read` when the reason matters — it reports
            the limit explicitly.
        """
        try:
            return self._fetch_file_bytes(self._resolve_path(path))
        except ReadLimitExceeded:
            return b""

    def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read a slice of a text file, decoding or extracting it as needed."""
        resolved = self._resolve_path(path)
        try:
            data = self._fetch_file_bytes(resolved)
            if not data:
                return f"Error: File '{path}' not found"

            extension = Path(resolved).suffix.lower().lstrip(".")
            try:
                lines = bytes_to_text(extension, data).splitlines()
            except ValueError as e:
                return f"[Error: {e}]"

            if offset >= len(lines):
                return "[End of file]"

            end = offset + limit
            chunk = "\n".join(lines[offset:end])
            if end >= len(lines):
                return chunk
            remaining = len(lines) - end
            return f"{chunk}\n\n[... {remaining} more lines. Use offset={end} to read more.]"

        except ReadLimitExceeded as e:
            return f"[Error: {e}]"
        except Exception as e:
            return f"[Error reading file: {e}]"

    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit a file by replacing a string.

        The file is fetched, edited in Python and written back, so multiline
        strings need no shell escaping.
        """
        resolved = self._resolve_path(path)
        try:
            data = self._fetch_file_bytes(resolved)
            if not data:
                return EditResult(error=f"File '{path}' not found")

            extension = Path(resolved).suffix.lower().lstrip(".")
            try:
                content = bytes_to_text(extension, data)
            except ValueError as e:
                return EditResult(error=str(e))

            outcome = replace_in_content(content, old_string, new_string, replace_all)
            if not isinstance(outcome, Replacement):
                return EditResult(error=outcome)

            written = self.write(resolved, outcome.content)
            if written.error:
                return EditResult(error=written.error)
            return EditResult(path=resolved, occurrences=outcome.occurrences)

        except ReadLimitExceeded as e:
            return EditResult(error=str(e))
        except Exception as e:
            return EditResult(error=f"Failed to edit file: {e}")

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write a file, creating parent directories as needed."""
        path = self._resolve_path(path)
        self._ensure_container()
        assert self._container is not None

        try:
            parent = str(PurePosixPath(path).parent)
            mkdir = self.execute(f"mkdir -p {shlex.quote(parent)}")
            if mkdir.exit_code != 0:
                return WriteResult(error=f"Failed to create directory: {mkdir.output}")

            raw = content if isinstance(content, bytes) else content.encode()
            archive = _single_file_archive(PurePosixPath(path).name, raw)

            # put_archive returns False when the target is not a directory or
            # the upload otherwise fails.
            if not self._container.put_archive(parent, archive):
                return WriteResult(error=f"Failed to write file: put_archive to {parent}")
            return WriteResult(path=path)
        except Exception as e:
            return WriteResult(error=f"Failed to write file: {e}")

    def _fetch_file_bytes(self, path: str) -> bytes:
        """Fetch a file's raw bytes out of the container.

        Args:
            path: Absolute path inside the container.

        Returns:
            The content, or `b""` when the path is missing or holds no regular
            file.

        Raises:
            ReadLimitExceeded: If the file is over `max_read_bytes`.
        """
        self._ensure_container()
        assert self._container is not None

        try:
            raw_stream, stat = self._container.get_archive(path)
        except Exception:
            return b""

        # docker-py streams the archive from a generator, so it can be closed to
        # release the socket as soon as the file turns out to be too large. The
        # stub only promises an Iterator, which has no `close`.
        stream = cast("Generator[bytes, None, None]", raw_stream)

        # get_archive reports the size in a response header, so an oversized
        # file is refused before any of its content crosses the socket.
        reported_size = stat.get("size") if stat else None
        if reported_size is not None and reported_size > self._max_read_bytes:
            stream.close()
            raise ReadLimitExceeded(
                f"File is {reported_size} bytes, over the "
                f"{self._max_read_bytes}-byte read limit. {READ_LIMIT_HINT}"
            )

        # Accumulated straight into the buffer tarfile reads from; a
        # `b"".join(stream)` -> `BytesIO(...)` chain held several copies at once.
        buffer = io.BytesIO()
        try:
            for chunk in stream:
                buffer.write(chunk)
                # Re-checked while streaming to stay bounded even when the
                # daemon omits the size header.
                if buffer.tell() > self._max_read_bytes:
                    stream.close()
                    raise ReadLimitExceeded(
                        f"File exceeds the {self._max_read_bytes}-byte read limit. "
                        f"{READ_LIMIT_HINT}"
                    )
        except ReadLimitExceeded:
            raise
        except Exception:
            return b""

        return _extract_single_file(buffer)

runtime property

The runtime configuration for this sandbox.

session_id property

Alias for the sandbox id, used for session management.

__init__(image='python:3.12-slim', sandbox_id=None, work_dir='/workspace', auto_remove=True, runtime=None, session_id=None, idle_timeout=3600, volumes=None, network_mode=None, container_name=None, mem_limit=None, memswap_limit=None, cpus=None, cpu_shares=None, pids_limit=DEFAULT_PIDS_LIMIT, tmpfs=None, max_read_bytes=DEFAULT_MAX_READ_BYTES, oci_runtime=None)

Initialize the sandbox without starting its container.

Parameters:

Name	Type	Description	Default
image	str	Docker image to use. Ignored when runtime is given.	'python:3.12-slim'
sandbox_id	str | None	Unique identifier for this sandbox.	None
work_dir	str	Working directory inside the container. Ignored when runtime is given.	'/workspace'
auto_remove	bool	Remove the container when it stops. Forced to False when container_name is set, since a named container exists to be reused.	True
runtime	RuntimeConfig | str | None	RuntimeConfig, or the name of a built-in runtime.	None
session_id	str | None	Alias for sandbox_id, for session management.	None
idle_timeout	int	Idle seconds after which SessionManager may reap it.	3600
volumes	dict[str, str] | None	Host-to-container mounts, as {"/host": "/container"}.	None
network_mode	str | None	Docker network mode ("bridge", "none", "host", "container:<name|id>"). Pass "none" for sandboxes that must not reach the network; it also skips per-container veth and firewall setup, so containers start measurably faster.	None
container_name	str | None	Stable name to reattach to across restarts, which preserves installed packages and other filesystem state. Implies auto_remove=False.	None
mem_limit	str | None	Memory ceiling in Docker syntax ("512m", "2g"). Swap is pinned to the same value unless memswap_limit says otherwise, so a container over its ceiling is stopped rather than left swapping against the host.	None
memswap_limit	str | None	Ceiling on memory and swap combined, in the same syntax. None pins it to mem_limit, which denies the container swap entirely — the right default, because a container swapping past its limit against a disk starves every other sandbox on the host. It is the wrong default on a host backed by zram, where swap is compressed RAM: the pages never leave memory, idle Python heaps compress to roughly a third, and the alternative to a little swapping is an OOM kill. Set this above mem_limit there and nowhere else. Ignored without mem_limit, since Docker rejects a swap ceiling with no memory ceiling under it.	None
cpus	float | None	Hard CPU ceiling in cores, e.g. 1.5. A container never exceeds it, which also means it cannot use cores that are sitting idle — on a small host that is often the wrong trade.	None
cpu_shares	int | None	Relative CPU weight (Docker's default is 1024). Unlike cpus this only applies under contention, so one active sandbox may use the whole machine and several are still divided fairly. Composes with cpus when both are set.	None
pids_limit	int | None	Maximum number of processes. None disables the limit.	DEFAULT_PIDS_LIMIT
tmpfs	dict[str, str] | None	In-memory mounts, as {"/tmp": "size=64m"}. Writes to a tmpfs never reach the container's write layer, so scratch files are both faster and free of disk growth. exec is added to the options because Docker mounts a tmpfs noexec, which breaks installing any package that builds from source. Its pages count against mem_limit, not on top of it: a sandbox that fills a 64m /tmp has that much less left for its own processes, and one that tries to exceed the limit through /tmp is killed by its own cgroup rather than troubling the host.	None
max_read_bytes	int	Largest file read/read_bytes/edit will pull out of the container. Oversized files are refused instead of being buffered into the host's memory.	DEFAULT_MAX_READ_BYTES
oci_runtime	str | None	Low-level runtime the daemon starts this container with — Docker's --runtime. None takes the daemon's default, normally runc. This is the one knob that changes the isolation boundary rather than a resource ceiling, which is why it is per sandbox: "runsc" (gVisor) moves syscall handling into userspace and "kata" gives the container its own kernel in a microVM, while a container under plain runc shares the host's. Untrusted model-written code is exactly the workload that argues for one of them. The runtime must already be registered with the daemon in /etc/docker/daemon.json; naming an unregistered one makes the daemon refuse to start the container. See the installation docs for the host side, including crun as a faster drop-in default.	None

Source code in src/pydantic_ai_backends/backends/docker/sandbox.py

def __init__(
    self,
    image: str = "python:3.12-slim",
    sandbox_id: str | None = None,
    work_dir: str = "/workspace",
    auto_remove: bool = True,
    runtime: RuntimeConfig | str | None = None,
    session_id: str | None = None,
    idle_timeout: int = 3600,
    volumes: dict[str, str] | None = None,
    network_mode: str | None = None,
    container_name: str | None = None,
    mem_limit: str | None = None,
    memswap_limit: str | None = None,
    cpus: float | None = None,
    cpu_shares: int | None = None,
    pids_limit: int | None = DEFAULT_PIDS_LIMIT,
    tmpfs: dict[str, str] | None = None,
    max_read_bytes: int = DEFAULT_MAX_READ_BYTES,
    oci_runtime: str | None = None,
):
    """Initialize the sandbox without starting its container.

    Args:
        image: Docker image to use. Ignored when `runtime` is given.
        sandbox_id: Unique identifier for this sandbox.
        work_dir: Working directory inside the container. Ignored when
            `runtime` is given.
        auto_remove: Remove the container when it stops. Forced to `False`
            when `container_name` is set, since a named container exists to
            be reused.
        runtime: `RuntimeConfig`, or the name of a built-in runtime.
        session_id: Alias for `sandbox_id`, for session management.
        idle_timeout: Idle seconds after which `SessionManager` may reap it.
        volumes: Host-to-container mounts, as `{"/host": "/container"}`.
        network_mode: Docker network mode (`"bridge"`, `"none"`, `"host"`,
            `"container:<name|id>"`). Pass `"none"` for sandboxes that must
            not reach the network; it also skips per-container veth and
            firewall setup, so containers start measurably faster.
        container_name: Stable name to reattach to across restarts, which
            preserves installed packages and other filesystem state.
            Implies `auto_remove=False`.
        mem_limit: Memory ceiling in Docker syntax (`"512m"`, `"2g"`). Swap
            is pinned to the same value unless `memswap_limit` says
            otherwise, so a container over its ceiling is stopped rather
            than left swapping against the host.
        memswap_limit: Ceiling on memory *and* swap combined, in the same
            syntax. `None` pins it to `mem_limit`, which denies the container
            swap entirely — the right default, because a container swapping
            past its limit against a disk starves every other sandbox on the
            host.

            It is the wrong default on a host backed by `zram`, where swap
            is compressed RAM: the pages never leave memory, idle Python
            heaps compress to roughly a third, and the alternative to a
            little swapping is an OOM kill. Set this above `mem_limit` there
            and nowhere else. Ignored without `mem_limit`, since Docker
            rejects a swap ceiling with no memory ceiling under it.
        cpus: Hard CPU ceiling in cores, e.g. `1.5`. A container never
            exceeds it, which also means it cannot use cores that are sitting
            idle — on a small host that is often the wrong trade.
        cpu_shares: Relative CPU weight (Docker's default is 1024). Unlike
            `cpus` this only applies under contention, so one active sandbox
            may use the whole machine and several are still divided fairly.
            Composes with `cpus` when both are set.
        pids_limit: Maximum number of processes. `None` disables the limit.
        tmpfs: In-memory mounts, as `{"/tmp": "size=64m"}`. Writes to a
            tmpfs never reach the container's write layer, so scratch files
            are both faster and free of disk growth. `exec` is added to the
            options because Docker mounts a tmpfs `noexec`, which breaks
            installing any package that builds from source.

            Its pages count against `mem_limit`, not on top of it: a sandbox
            that fills a 64m `/tmp` has that much less left for its own
            processes, and one that tries to exceed the limit through `/tmp`
            is killed by its own cgroup rather than troubling the host.
        max_read_bytes: Largest file `read`/`read_bytes`/`edit` will pull
            out of the container. Oversized files are refused instead of
            being buffered into the host's memory.
        oci_runtime: Low-level runtime the daemon starts this container
            with — Docker's `--runtime`. `None` takes the daemon's default,
            normally `runc`.

            This is the one knob that changes the *isolation boundary*
            rather than a resource ceiling, which is why it is per sandbox:
            `"runsc"` (gVisor) moves syscall handling into userspace and
            `"kata"` gives the container its own kernel in a microVM, while
            a container under plain `runc` shares the host's. Untrusted
            model-written code is exactly the workload that argues for one
            of them.

            The runtime must already be registered with the daemon in
            `/etc/docker/daemon.json`; naming an unregistered one makes the
            daemon refuse to start the container. See the installation docs
            for the host side, including `crun` as a faster drop-in default.
    """
    super().__init__(session_id or sandbox_id)

    self._container_name = container_name
    self._auto_remove = False if container_name else auto_remove
    self._container: Container | None = None
    self._idle_timeout = idle_timeout
    self._last_activity = time.time()
    self._volumes = volumes or {}
    self._network_mode = network_mode
    self._mem_limit = mem_limit
    self._memswap_limit = memswap_limit
    self._cpus = cpus
    self._cpu_shares = cpu_shares
    self._pids_limit = pids_limit
    self._tmpfs = tmpfs or {}
    self._max_read_bytes = max_read_bytes
    self._oci_runtime = oci_runtime
    self._alive = False
    self._alive_checked_at: float | None = None

    if isinstance(runtime, str):
        from pydantic_ai_backends.backends.docker.runtimes import get_runtime

        runtime = get_runtime(runtime)
    self._runtime = runtime
    self._image = image
    self._work_dir = runtime.work_dir if runtime is not None else work_dir

execute(command, timeout=None)

Run a command in the container.

Output beyond MAX_EXECUTE_OUTPUT_BYTES is discarded before decoding, so the cap is measured in bytes rather than characters.

Source code in src/pydantic_ai_backends/backends/docker/sandbox.py

def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
    """Run a command in the container.

    Output beyond `MAX_EXECUTE_OUTPUT_BYTES` is discarded before decoding,
    so the cap is measured in bytes rather than characters.
    """
    self._ensure_container()
    self._last_activity = time.time()
    assert self._container is not None

    # The Docker SDK's exec_run takes no timeout, so the command is wrapped
    # in the `timeout` utility instead.
    argv = ["sh", "-c", command]
    if timeout is not None:
        argv = ["timeout", str(timeout), *argv]

    try:
        exit_code, output = self._container.exec_run(argv, workdir=self._work_dir)
        if not isinstance(output, bytes):
            output = b"".join(output)
    except Exception as e:
        return ExecuteResponse(output=f"Error: {e}", exit_code=1, truncated=False)

    # Sliced before decoding: decoding the whole payload only to throw most
    # of it away doubled peak memory on commands like `cat big.log`.
    return ExecuteResponse(
        output=output[:MAX_EXECUTE_OUTPUT_BYTES].decode("utf-8", errors="replace"),
        exit_code=exit_code,
        truncated=len(output) > MAX_EXECUTE_OUTPUT_BYTES,
    )

read(path, offset=0, limit=2000)

Read a slice of a text file, decoding or extracting it as needed.

Source code in src/pydantic_ai_backends/backends/docker/sandbox.py

def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
    """Read a slice of a text file, decoding or extracting it as needed."""
    resolved = self._resolve_path(path)
    try:
        data = self._fetch_file_bytes(resolved)
        if not data:
            return f"Error: File '{path}' not found"

        extension = Path(resolved).suffix.lower().lstrip(".")
        try:
            lines = bytes_to_text(extension, data).splitlines()
        except ValueError as e:
            return f"[Error: {e}]"

        if offset >= len(lines):
            return "[End of file]"

        end = offset + limit
        chunk = "\n".join(lines[offset:end])
        if end >= len(lines):
            return chunk
        remaining = len(lines) - end
        return f"{chunk}\n\n[... {remaining} more lines. Use offset={end} to read more.]"

    except ReadLimitExceeded as e:
        return f"[Error: {e}]"
    except Exception as e:
        return f"[Error reading file: {e}]"

write(path, content)

Write a file, creating parent directories as needed.

Source code in src/pydantic_ai_backends/backends/docker/sandbox.py

def write(self, path: str, content: str | bytes) -> WriteResult:
    """Write a file, creating parent directories as needed."""
    path = self._resolve_path(path)
    self._ensure_container()
    assert self._container is not None

    try:
        parent = str(PurePosixPath(path).parent)
        mkdir = self.execute(f"mkdir -p {shlex.quote(parent)}")
        if mkdir.exit_code != 0:
            return WriteResult(error=f"Failed to create directory: {mkdir.output}")

        raw = content if isinstance(content, bytes) else content.encode()
        archive = _single_file_archive(PurePosixPath(path).name, raw)

        # put_archive returns False when the target is not a directory or
        # the upload otherwise fails.
        if not self._container.put_archive(parent, archive):
            return WriteResult(error=f"Failed to write file: put_archive to {parent}")
        return WriteResult(path=path)
    except Exception as e:
        return WriteResult(error=f"Failed to write file: {e}")

start()

Start the container now instead of on the first operation.

Source code in src/pydantic_ai_backends/backends/docker/sandbox.py

def start(self) -> None:
    """Start the container now instead of on the first operation."""
    self._ensure_container()

stop(purge=False, *, remove=None)

Stop the container.

A container created without container_name runs with auto_remove=True and is discarded by the daemon on exit. A named container deliberately survives, since reuse across restarts is the whole point of naming it.

Parameters:

Name	Type	Description	Default
purge	bool	Also remove the container, discarding its filesystem state. Named purge so that one call site can end any sandbox this library offers - RemoteSandbox, DaytonaSandbox and the Kubernetes pod all spell the same idea this way, and this one used to spell it remove. A caller holding "a sandbox" could not call stop without knowing which it had.	False
remove	bool | None	The old name for purge, still honoured so nothing that passes it breaks. Deprecated; pass purge instead.	None

Source code in src/pydantic_ai_backends/backends/docker/sandbox.py

def stop(self, purge: bool = False, *, remove: bool | None = None) -> None:
    """Stop the container.

    A container created without `container_name` runs with
    `auto_remove=True` and is discarded by the daemon on exit. A *named*
    container deliberately survives, since reuse across restarts is the
    whole point of naming it.

    Args:
        purge: Also remove the container, discarding its filesystem state.
            Named `purge` so that one call site can end any sandbox this
            library offers - `RemoteSandbox`, `DaytonaSandbox` and the
            Kubernetes pod all spell the same idea this way, and this one
            used to spell it `remove`. A caller holding "a sandbox" could
            not call `stop` without knowing which it had.
        remove: The old name for `purge`, still honoured so nothing that
            passes it breaks. Deprecated; pass `purge` instead.
    """
    if remove is not None:
        warnings.warn(
            "DockerSandbox.stop(remove=...) is deprecated; pass purge=... instead, "
            "which is what every other sandbox calls the same argument.",
            DeprecationWarning,
            stacklevel=2,
        )
        purge = remove

    container = getattr(self, "_container", None)
    if container is None:
        return

    with contextlib.suppress(Exception):
        container.stop()
    if purge:
        with contextlib.suppress(Exception):
            container.remove(force=True)
    self._container = None
    self._alive_checked_at = None

is_alive()

Whether the container is running.

The answer is cached for ALIVE_CACHE_SECONDS, since reload() is a daemon round trip and session managers call this on every request.

Source code in src/pydantic_ai_backends/backends/docker/sandbox.py

def is_alive(self) -> bool:
    """Whether the container is running.

    The answer is cached for `ALIVE_CACHE_SECONDS`, since `reload()` is a
    daemon round trip and session managers call this on every request.
    """
    if self._container is None:
        return False

    now = time.monotonic()
    checked_at = self._alive_checked_at
    if checked_at is not None and now - checked_at < ALIVE_CACHE_SECONDS:
        return self._alive

    try:
        self._container.reload()
        status: str = self._container.status
    except Exception:
        self._alive = False
    else:
        self._alive = status == "running"

    self._alive_checked_at = now
    return self._alive

BaseSandbox

pydantic_ai_backends.backends.docker.sandbox.BaseSandbox

Bases: _SandboxIdentity, ABC

Base class for synchronous sandboxes that expose a shell.

Parameters:

Name	Type	Description	Default
sandbox_id	str | None	Unique identifier for this sandbox. Generated when omitted.	None

Source code in src/pydantic_ai_backends/backends/base.py

class BaseSandbox(_SandboxIdentity, ABC):
    """Base class for synchronous sandboxes that expose a shell.

    Args:
        sandbox_id: Unique identifier for this sandbox. Generated when omitted.
    """

    def start(self) -> None:
        """Start the sandbox eagerly.

        The default is a no-op, since sandboxes start on first use.
        """

    def is_alive(self) -> bool:
        """Whether the sandbox is running and responsive.

        **Override this.** The default answer is "no", and `SessionManager` reads
        it as "replace the sandbox" — so a subclass that never overrides it has a
        sandbox built, stopped and rebuilt on every single operation. It defaults
        to `False` rather than `True` because a sandbox wrongly believed alive is
        the harder failure to diagnose, but neither default is right for a real
        sandbox: answer it.
        """
        return False

    def stop(self, purge: bool = False) -> None:
        """Stop and clean up the sandbox.

        Args:
            purge: Also discard what the sandbox accumulated — its filesystem,
                and any workspace kept for it outside the container. Left off,
                the sandbox ends while its files survive for the next attach;
                turned on, the thing the sandbox belonged to is gone for good.

                Accepted by every sandbox so one call site can end any of them,
                and the subclasses that cannot tell the two apart say so: a
                Daytona sandbox and a Kubernetes pod are deleted either way,
                because neither keeps a filesystem this library can reattach to.
        """

    @abstractmethod
    def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Run a command in the sandbox.

        Args:
            command: Command to execute.
            timeout: Maximum execution time in seconds.
        """
        ...

    @abstractmethod
    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit a file by replacing a string.

        Args:
            path: File path to edit.
            old_string: String to find and replace.
            new_string: Replacement string.
            replace_all: Replace every occurrence instead of only the first.
        """
        ...

    def exists(self, path: str) -> bool:
        """Whether `path` is a regular file, via `test -f`."""
        return self.execute(_shell.exists_command(path), timeout=5).exit_code == 0

    def ls_info(self, path: str) -> list[FileInfo]:
        """List one directory using `ls -la`."""
        command = _shell.ls_command(path)
        return _shell.parse_ls(self.execute(command, timeout=FILE_OP_TIMEOUT), path)

    def read_bytes(self, path: str) -> bytes:
        """Read a whole file with `cat`, or `b""` on any failure."""
        command = _shell.read_bytes_command(path)
        return _shell.parse_read_bytes(self.execute(command, timeout=FILE_OP_TIMEOUT))

    def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read a slice of a file, numbered by its real line positions."""
        command = _shell.read_command(path, offset, limit)
        return _shell.parse_read(self.execute(command, timeout=FILE_OP_TIMEOUT))

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write a file, carrying the content base64-encoded."""
        command = _shell.write_command(path, content)
        return _shell.parse_write(self.execute(command, timeout=FILE_OP_TIMEOUT), path)

    def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
        """Match files with `find`."""
        command = _shell.glob_command(pattern, path)
        return _shell.parse_glob(self.execute(command, timeout=SEARCH_TIMEOUT))

    def grep_raw(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Search file contents with `grep`."""
        command = _shell.grep_command(pattern, path, glob, ignore_hidden)
        return _shell.parse_grep(self.execute(command, timeout=SEARCH_TIMEOUT))

execute(command, timeout=None) abstractmethod

Run a command in the sandbox.

Parameters:

Name	Type	Description	Default
command	str	Command to execute.	required
timeout	int | None	Maximum execution time in seconds.	None

Source code in src/pydantic_ai_backends/backends/base.py

@abstractmethod
def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
    """Run a command in the sandbox.

    Args:
        command: Command to execute.
        timeout: Maximum execution time in seconds.
    """
    ...

ls_info(path)

List one directory using ls -la.

Source code in src/pydantic_ai_backends/backends/base.py

def ls_info(self, path: str) -> list[FileInfo]:
    """List one directory using `ls -la`."""
    command = _shell.ls_command(path)
    return _shell.parse_ls(self.execute(command, timeout=FILE_OP_TIMEOUT), path)

read(path, offset=0, limit=2000)

Read a slice of a file, numbered by its real line positions.

Source code in src/pydantic_ai_backends/backends/base.py

def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
    """Read a slice of a file, numbered by its real line positions."""
    command = _shell.read_command(path, offset, limit)
    return _shell.parse_read(self.execute(command, timeout=FILE_OP_TIMEOUT))

write(path, content)

Write a file, carrying the content base64-encoded.

Source code in src/pydantic_ai_backends/backends/base.py

def write(self, path: str, content: str | bytes) -> WriteResult:
    """Write a file, carrying the content base64-encoded."""
    command = _shell.write_command(path, content)
    return _shell.parse_write(self.execute(command, timeout=FILE_OP_TIMEOUT), path)

edit(path, old_string, new_string, replace_all=False) abstractmethod

Edit a file by replacing a string.

Parameters:

Name	Type	Description	Default
path	str	File path to edit.	required
old_string	str	String to find and replace.	required
new_string	str	Replacement string.	required
replace_all	bool	Replace every occurrence instead of only the first.	False

Source code in src/pydantic_ai_backends/backends/base.py

@abstractmethod
def edit(
    self, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult:
    """Edit a file by replacing a string.

    Args:
        path: File path to edit.
        old_string: String to find and replace.
        new_string: Replacement string.
        replace_all: Replace every occurrence instead of only the first.
    """
    ...

glob_info(pattern, path='/')

Match files with find.

Source code in src/pydantic_ai_backends/backends/base.py

def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
    """Match files with `find`."""
    command = _shell.glob_command(pattern, path)
    return _shell.parse_glob(self.execute(command, timeout=SEARCH_TIMEOUT))

grep_raw(pattern, path=None, glob=None, ignore_hidden=True)

Search file contents with grep.

Source code in src/pydantic_ai_backends/backends/base.py

def grep_raw(
    self,
    pattern: str,
    path: str | None = None,
    glob: str | None = None,
    ignore_hidden: bool = True,
) -> list[GrepMatch] | str:
    """Search file contents with `grep`."""
    command = _shell.grep_command(pattern, path, glob, ignore_hidden)
    return _shell.parse_grep(self.execute(command, timeout=SEARCH_TIMEOUT))

SessionManager

pydantic_ai_backends.backends.docker.session.SessionManager

Creates, reuses and reaps one sandbox per session id.

Example

Python

from pydantic_ai_backends import SessionManager

manager = SessionManager(default_runtime="python-datascience")
sandbox = await manager.get_or_create("user-123")
cleaned = await manager.cleanup_idle(max_idle=1800)

Source code in src/pydantic_ai_backends/backends/docker/session.py

class SessionManager:
    """Creates, reuses and reaps one sandbox per session id.

    Example:
        ```python
        from pydantic_ai_backends import SessionManager

        manager = SessionManager(default_runtime="python-datascience")
        sandbox = await manager.get_or_create("user-123")
        cleaned = await manager.cleanup_idle(max_idle=1800)
        ```
    """

    def __init__(
        self,
        sandbox_factory: SandboxFactory | None = None,
        default_runtime: RuntimeConfig | str | None = None,
        default_idle_timeout: int = 3600,
        workspace_root: str | Path | None = None,
        max_sessions: int | None = None,
        on_release: Callable[[str], None] | None = None,
        executor: Executor | None = None,
    ):
        """Initialize the manager.

        Args:
            sandbox_factory: Builds a sandbox for a session id. It must offer
                `start()`, `stop()` and `is_alive()`; `last_activity` (or a
                legacy `_last_activity`) enables idle cleanup, and sandboxes
                without one are never reaped. Defaults to `DockerSandbox`.
            default_runtime: Runtime for new Docker sandboxes. Only used on the
                default path.
            default_idle_timeout: Idle seconds before a session may be reaped,
                for sandboxes that do not carry their own timeout.
            workspace_root: Root for persistent session storage. Only used on
                the default path, where `{workspace_root}/{session_id}/workspace`
                is created and mounted into the container.
            max_sessions: Ceiling on simultaneously open sessions. Once reached,
                :meth:`get_or_create` raises :class:`SessionLimitExceeded` for
                new session ids rather than starting unbounded containers.
            on_release: Called with a session id just after its sandbox is
                stopped, by :meth:`release` and therefore by idle cleanup too.
                For a caller keeping its own per-session state, this is the only
                notice that a reaping happened — polling `sessions` would mean
                discovering it late, or never. It runs inside a cleanup pass, so
                anything it raises aborts the rest of that pass.
            executor: Thread pool the blocking `start()` and `stop()` calls run
                on. `None` uses asyncio's default pool, which is shared with
                every other `to_thread` caller in the process — a service
                handing out sandboxes wants its own. Assignable afterwards, for
                a caller whose pool outlives fewer things than its manager does.
        """
        self._sessions: dict[str, Any] = {}
        self._sandbox_factory = sandbox_factory
        self._default_runtime = default_runtime
        self._default_idle_timeout = default_idle_timeout
        self._cleanup_task: asyncio.Task[None] | None = None
        self._workspace_root = Path(workspace_root) if workspace_root else None
        self._max_sessions = max_sessions
        self._on_release = on_release
        self.executor = executor
        # Per-session locks serialize every lifecycle change for one id —
        # `get_or_create` and `release` both — so two awaits cannot each create
        # and start a sandbox (one of which would be overwritten in the dict and
        # leaked), and a release cannot land in the middle of a creation.
        self._locks: dict[str, _SessionLock] = {}

    @property
    def sessions(self) -> dict[str, Any]:
        """Copy of the active sessions."""
        return dict(self._sessions)

    @property
    def session_count(self) -> int:
        """Number of active sessions."""
        return len(self._sessions)

    async def get_or_create(
        self,
        session_id: str,
        runtime: RuntimeConfig | str | None = None,
    ) -> Any:
        """Return the session's live sandbox, creating one when needed.

        Args:
            session_id: Unique identifier for the session.
            runtime: Runtime to use. Only applies on the default Docker path.

        Raises:
            SessionLimitExceeded: If `max_sessions` is reached and `session_id`
                is not an existing live session.
        """
        async with self._session_lock(session_id):
            existing = self._sessions.get(session_id)
            if existing is not None:
                if await alive_of(existing):
                    _record_activity(existing)
                    return existing
                # Dead, but very likely still holding client-side resources — an
                # SSH connection, an HTTP client with an open pool. Dropping the
                # reference alone leaks those until garbage collection, which for
                # an `httpx.Client` means a warning and an unclosed socket. It is
                # already dead, so a failing stop is expected and ignored.
                #
                # `pop`, not `del`: `release` holds the same lock now, but a
                # caller reaching into `_sessions` directly must not turn a
                # missing key into a KeyError out of a public coroutine.
                self._sessions.pop(session_id, None)
                with contextlib.suppress(Exception):
                    await self._lifecycle(existing.stop)

            if self._max_sessions is not None and len(self._sessions) >= self._max_sessions:
                raise SessionLimitExceeded(self._max_sessions)

            if self._sandbox_factory is not None:
                sandbox = self._sandbox_factory(session_id)
            else:
                sandbox = self._create_docker_sandbox(session_id, runtime)

            try:
                await self._lifecycle(sandbox.start)
            except Exception:
                # An unregistered sandbox is one nothing else will ever stop, so
                # a partial start would leak whatever it did manage to create.
                with contextlib.suppress(Exception):
                    await self._lifecycle(sandbox.stop)
                raise

            self._sessions[session_id] = sandbox
            return sandbox

    async def _lifecycle(self, call: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
        """Invoke a sandbox's `start` or `stop`, whether it is sync or async.

        A blocking call goes to a thread: starting a sandbox pulls or builds an
        image and stopping one waits for the process inside to die, and on the
        loop either stalls every other session for the duration.

        A natively async sandbox is awaited directly instead. Handing its
        coroutine *function* to a thread would merely create a coroutine nobody
        awaits — so the sandbox would silently never start, never stop, and its
        `is_alive` would return a truthy coroutine object for ever.
        """
        if inspect.iscoroutinefunction(call):
            return await call(*args, **kwargs)
        bound = functools.partial(call, *args, **kwargs)
        if self.executor is None:
            return await asyncio.to_thread(bound)
        return await asyncio.get_running_loop().run_in_executor(self.executor, bound)

    def _create_docker_sandbox(
        self,
        session_id: str,
        runtime: RuntimeConfig | str | None = None,
    ) -> Any:
        """Build a `DockerSandbox`, the default when no factory is given."""
        from pydantic_ai_backends.backends.docker.sandbox import DockerSandbox

        volumes: dict[str, str] | None = None
        if self._workspace_root:
            workspace = self._workspace_root / session_id / "workspace"
            workspace.mkdir(parents=True, exist_ok=True)
            volumes = {str(workspace.resolve()): "/workspace"}

        return DockerSandbox(
            runtime=runtime or self._default_runtime,
            session_id=session_id,
            idle_timeout=self._default_idle_timeout,
            volumes=volumes,
        )

    @asynccontextmanager
    async def _session_lock(self, session_id: str) -> AsyncIterator[None]:
        """Hold the lock serializing every lifecycle change for one session id.

        Interned rather than created per call, so `get_or_create` and `release`
        contend on the same object — and reference-counted while entered, so
        :meth:`_prune_locks` cannot drop one out from under a task. Counting
        rather than asking `Lock.locked()`: between a holder releasing and the
        woken waiter resuming, a lock reads as unlocked while a task is very much
        still queued on it, and interning a fresh one for the next caller is how
        two of them end up creating a sandbox for the same session at once.
        """
        entry = self._locks.setdefault(session_id, _SessionLock())
        entry.users += 1
        try:
            async with entry.lock:
                yield
        finally:
            entry.users -= 1

    async def release(self, session_id: str) -> bool:
        """Stop a session's sandbox. Returns whether the session existed.

        Takes the session's lock, so a release cannot land in the middle of a
        `get_or_create` for the same id. Without it the two interleaved: the
        release removed the entry `get_or_create` was about to delete (a
        `KeyError` out of a public coroutine) and removed the lock it was
        holding, after which a third caller interned a new lock and created a
        second sandbox for one session — the leak the lock is meant to prevent.
        """
        async with self._session_lock(session_id):
            sandbox = self._sessions.pop(session_id, None)
            if sandbox is None:
                return False
            await self._lifecycle(sandbox.stop)

        self._prune_locks()
        # Outside the lock: `on_release` belongs to the caller, and one that
        # opens or releases a session from it would otherwise deadlock.
        if self._on_release is not None:
            self._on_release(session_id)
        return True

    async def cleanup_idle(self, max_idle: int | None = None) -> int:
        """Stop the sandboxes that have been idle too long.

        Args:
            max_idle: Idle ceiling in seconds applied to every session,
                overriding each sandbox's own. When omitted, a sandbox's
                `idle_timeout` wins, falling back to `default_idle_timeout`.

        Returns:
            Number of sessions cleaned up.
        """
        now = time.time()
        # A custom factory may return a sandbox that never records activity.
        # Treating that as "just used" keeps it alive instead of raising and
        # taking the whole cleanup loop down with it.
        idle = [
            session_id
            for session_id, sandbox in self._sessions.items()
            if now - last_activity_of(sandbox, now)
            > (max_idle if max_idle is not None else self._idle_limit_for(sandbox))
        ]

        for session_id in idle:
            await self.release(session_id)

        self._prune_locks()
        return len(idle)

    def _idle_limit_for(self, sandbox: Any) -> int:
        """Idle ceiling for one sandbox, preferring its own configured timeout.

        `DockerSandbox` accepts an `idle_timeout` and documents it as the idle
        cleanup window, so a per-sandbox value has to win over the manager-wide
        default for the parameter to mean anything.
        """
        for attr in ("idle_timeout", LEGACY_IDLE_TIMEOUT_ATTR):
            configured = getattr(sandbox, attr, None)
            if isinstance(configured, int):
                return configured
        return self._default_idle_timeout

    def _prune_locks(self) -> None:
        """Drop locks interned for session ids that no longer have a sandbox.

        `get_or_create` interns a lock before it knows whether the sandbox can
        be created, so every rejected or failed creation would otherwise leave
        an entry behind for good. A lock any task is inside — holding it, or
        queued for it — is left alone: that entry is what mutual exclusion for
        the id currently is, and replacing it means two tasks proceed at once.
        """
        stale = [
            session_id
            for session_id, entry in self._locks.items()
            if session_id not in self._sessions and entry.users == 0
        ]
        for session_id in stale:
            del self._locks[session_id]

    def start_cleanup_loop(self, interval: int = DEFAULT_CLEANUP_INTERVAL) -> None:
        """Reap idle sessions periodically in the background.

        The loop survives a failing pass: an unreachable daemon or one
        uncooperative sandbox is logged and retried on the next tick, because a
        loop that exits leaves every future container to accumulate unnoticed.

        Args:
            interval: Seconds between passes.
        """
        if self._cleanup_task is not None:
            return

        async def loop() -> None:
            while True:
                await asyncio.sleep(interval)
                try:
                    await self.cleanup_idle()
                except asyncio.CancelledError:
                    raise
                except Exception:
                    _logger.exception("Idle sandbox cleanup failed; retrying next interval")

        self._cleanup_task = asyncio.create_task(loop())

    def stop_cleanup_loop(self) -> None:
        """Stop the background cleanup loop."""
        if self._cleanup_task is not None:
            self._cleanup_task.cancel()
            self._cleanup_task = None

    async def shutdown(self) -> int:
        """Stop every session and the cleanup loop.

        Sessions are stopped concurrently, because stopping one is seconds of
        waiting for the process inside to die and they do not wait on each
        other: sequentially, a full pool turned a shutdown into minutes, and
        an orchestrator that loses patience kills the process mid-teardown.

        Returns:
            Number of sessions that were stopped.
        """
        self.stop_cleanup_loop()

        session_ids = list(self._sessions)
        # `return_exceptions`, so one uncooperative sandbox cannot leave the
        # rest of the pool running — a shutdown has no later attempt.
        outcomes = await asyncio.gather(
            *(self.release(session_id) for session_id in session_ids),
            return_exceptions=True,
        )
        for session_id, outcome in zip(session_ids, outcomes, strict=True):
            if isinstance(outcome, BaseException):
                _logger.warning(
                    "Session %s did not stop cleanly during shutdown: %s", session_id, outcome
                )
        return len(session_ids)

    def __contains__(self, session_id: str) -> bool:
        return session_id in self._sessions

    def __len__(self) -> int:
        return len(self._sessions)

sessions property

Copy of the active sessions.

session_count property

Number of active sessions.

__init__(sandbox_factory=None, default_runtime=None, default_idle_timeout=3600, workspace_root=None, max_sessions=None, on_release=None, executor=None)

Initialize the manager.

Parameters:

Name	Type	Description	Default
sandbox_factory	SandboxFactory | None	Builds a sandbox for a session id. It must offer start(), stop() and is_alive(); last_activity (or a legacy _last_activity) enables idle cleanup, and sandboxes without one are never reaped. Defaults to DockerSandbox.	None
default_runtime	RuntimeConfig | str | None	Runtime for new Docker sandboxes. Only used on the default path.	None
default_idle_timeout	int	Idle seconds before a session may be reaped, for sandboxes that do not carry their own timeout.	3600
workspace_root	str | Path | None	Root for persistent session storage. Only used on the default path, where {workspace_root}/{session_id}/workspace is created and mounted into the container.	None
max_sessions	int | None	Ceiling on simultaneously open sessions. Once reached, :meth:get_or_create raises :class:SessionLimitExceeded for new session ids rather than starting unbounded containers.	None
on_release	Callable[[str], None] | None	Called with a session id just after its sandbox is stopped, by :meth:release and therefore by idle cleanup too. For a caller keeping its own per-session state, this is the only notice that a reaping happened — polling sessions would mean discovering it late, or never. It runs inside a cleanup pass, so anything it raises aborts the rest of that pass.	None
executor	Executor | None	Thread pool the blocking start() and stop() calls run on. None uses asyncio's default pool, which is shared with every other to_thread caller in the process — a service handing out sandboxes wants its own. Assignable afterwards, for a caller whose pool outlives fewer things than its manager does.	None

Source code in src/pydantic_ai_backends/backends/docker/session.py

def __init__(
    self,
    sandbox_factory: SandboxFactory | None = None,
    default_runtime: RuntimeConfig | str | None = None,
    default_idle_timeout: int = 3600,
    workspace_root: str | Path | None = None,
    max_sessions: int | None = None,
    on_release: Callable[[str], None] | None = None,
    executor: Executor | None = None,
):
    """Initialize the manager.

    Args:
        sandbox_factory: Builds a sandbox for a session id. It must offer
            `start()`, `stop()` and `is_alive()`; `last_activity` (or a
            legacy `_last_activity`) enables idle cleanup, and sandboxes
            without one are never reaped. Defaults to `DockerSandbox`.
        default_runtime: Runtime for new Docker sandboxes. Only used on the
            default path.
        default_idle_timeout: Idle seconds before a session may be reaped,
            for sandboxes that do not carry their own timeout.
        workspace_root: Root for persistent session storage. Only used on
            the default path, where `{workspace_root}/{session_id}/workspace`
            is created and mounted into the container.
        max_sessions: Ceiling on simultaneously open sessions. Once reached,
            :meth:`get_or_create` raises :class:`SessionLimitExceeded` for
            new session ids rather than starting unbounded containers.
        on_release: Called with a session id just after its sandbox is
            stopped, by :meth:`release` and therefore by idle cleanup too.
            For a caller keeping its own per-session state, this is the only
            notice that a reaping happened — polling `sessions` would mean
            discovering it late, or never. It runs inside a cleanup pass, so
            anything it raises aborts the rest of that pass.
        executor: Thread pool the blocking `start()` and `stop()` calls run
            on. `None` uses asyncio's default pool, which is shared with
            every other `to_thread` caller in the process — a service
            handing out sandboxes wants its own. Assignable afterwards, for
            a caller whose pool outlives fewer things than its manager does.
    """
    self._sessions: dict[str, Any] = {}
    self._sandbox_factory = sandbox_factory
    self._default_runtime = default_runtime
    self._default_idle_timeout = default_idle_timeout
    self._cleanup_task: asyncio.Task[None] | None = None
    self._workspace_root = Path(workspace_root) if workspace_root else None
    self._max_sessions = max_sessions
    self._on_release = on_release
    self.executor = executor
    # Per-session locks serialize every lifecycle change for one id —
    # `get_or_create` and `release` both — so two awaits cannot each create
    # and start a sandbox (one of which would be overwritten in the dict and
    # leaked), and a release cannot land in the middle of a creation.
    self._locks: dict[str, _SessionLock] = {}

get_or_create(session_id, runtime=None) async

Return the session's live sandbox, creating one when needed.

Parameters:

Name	Type	Description	Default
session_id	str	Unique identifier for the session.	required
runtime	RuntimeConfig | str | None	Runtime to use. Only applies on the default Docker path.	None

Raises:

Type	Description
SessionLimitExceeded	If max_sessions is reached and session_id is not an existing live session.

Source code in src/pydantic_ai_backends/backends/docker/session.py

async def get_or_create(
    self,
    session_id: str,
    runtime: RuntimeConfig | str | None = None,
) -> Any:
    """Return the session's live sandbox, creating one when needed.

    Args:
        session_id: Unique identifier for the session.
        runtime: Runtime to use. Only applies on the default Docker path.

    Raises:
        SessionLimitExceeded: If `max_sessions` is reached and `session_id`
            is not an existing live session.
    """
    async with self._session_lock(session_id):
        existing = self._sessions.get(session_id)
        if existing is not None:
            if await alive_of(existing):
                _record_activity(existing)
                return existing
            # Dead, but very likely still holding client-side resources — an
            # SSH connection, an HTTP client with an open pool. Dropping the
            # reference alone leaks those until garbage collection, which for
            # an `httpx.Client` means a warning and an unclosed socket. It is
            # already dead, so a failing stop is expected and ignored.
            #
            # `pop`, not `del`: `release` holds the same lock now, but a
            # caller reaching into `_sessions` directly must not turn a
            # missing key into a KeyError out of a public coroutine.
            self._sessions.pop(session_id, None)
            with contextlib.suppress(Exception):
                await self._lifecycle(existing.stop)

        if self._max_sessions is not None and len(self._sessions) >= self._max_sessions:
            raise SessionLimitExceeded(self._max_sessions)

        if self._sandbox_factory is not None:
            sandbox = self._sandbox_factory(session_id)
        else:
            sandbox = self._create_docker_sandbox(session_id, runtime)

        try:
            await self._lifecycle(sandbox.start)
        except Exception:
            # An unregistered sandbox is one nothing else will ever stop, so
            # a partial start would leak whatever it did manage to create.
            with contextlib.suppress(Exception):
                await self._lifecycle(sandbox.stop)
            raise

        self._sessions[session_id] = sandbox
        return sandbox

release(session_id) async

Stop a session's sandbox. Returns whether the session existed.

Takes the session's lock, so a release cannot land in the middle of a get_or_create for the same id. Without it the two interleaved: the release removed the entry get_or_create was about to delete (a KeyError out of a public coroutine) and removed the lock it was holding, after which a third caller interned a new lock and created a second sandbox for one session — the leak the lock is meant to prevent.

Source code in src/pydantic_ai_backends/backends/docker/session.py

async def release(self, session_id: str) -> bool:
    """Stop a session's sandbox. Returns whether the session existed.

    Takes the session's lock, so a release cannot land in the middle of a
    `get_or_create` for the same id. Without it the two interleaved: the
    release removed the entry `get_or_create` was about to delete (a
    `KeyError` out of a public coroutine) and removed the lock it was
    holding, after which a third caller interned a new lock and created a
    second sandbox for one session — the leak the lock is meant to prevent.
    """
    async with self._session_lock(session_id):
        sandbox = self._sessions.pop(session_id, None)
        if sandbox is None:
            return False
        await self._lifecycle(sandbox.stop)

    self._prune_locks()
    # Outside the lock: `on_release` belongs to the caller, and one that
    # opens or releases a session from it would otherwise deadlock.
    if self._on_release is not None:
        self._on_release(session_id)
    return True

cleanup_idle(max_idle=None) async

Stop the sandboxes that have been idle too long.

Parameters:

Name	Type	Description	Default
max_idle	int | None	Idle ceiling in seconds applied to every session, overriding each sandbox's own. When omitted, a sandbox's idle_timeout wins, falling back to default_idle_timeout.	None

Returns:

Type	Description
int	Number of sessions cleaned up.

Source code in src/pydantic_ai_backends/backends/docker/session.py

async def cleanup_idle(self, max_idle: int | None = None) -> int:
    """Stop the sandboxes that have been idle too long.

    Args:
        max_idle: Idle ceiling in seconds applied to every session,
            overriding each sandbox's own. When omitted, a sandbox's
            `idle_timeout` wins, falling back to `default_idle_timeout`.

    Returns:
        Number of sessions cleaned up.
    """
    now = time.time()
    # A custom factory may return a sandbox that never records activity.
    # Treating that as "just used" keeps it alive instead of raising and
    # taking the whole cleanup loop down with it.
    idle = [
        session_id
        for session_id, sandbox in self._sessions.items()
        if now - last_activity_of(sandbox, now)
        > (max_idle if max_idle is not None else self._idle_limit_for(sandbox))
    ]

    for session_id in idle:
        await self.release(session_id)

    self._prune_locks()
    return len(idle)

start_cleanup_loop(interval=DEFAULT_CLEANUP_INTERVAL)

Reap idle sessions periodically in the background.

The loop survives a failing pass: an unreachable daemon or one uncooperative sandbox is logged and retried on the next tick, because a loop that exits leaves every future container to accumulate unnoticed.

Parameters:

Name	Type	Description	Default
interval	int	Seconds between passes.	DEFAULT_CLEANUP_INTERVAL

Source code in src/pydantic_ai_backends/backends/docker/session.py

def start_cleanup_loop(self, interval: int = DEFAULT_CLEANUP_INTERVAL) -> None:
    """Reap idle sessions periodically in the background.

    The loop survives a failing pass: an unreachable daemon or one
    uncooperative sandbox is logged and retried on the next tick, because a
    loop that exits leaves every future container to accumulate unnoticed.

    Args:
        interval: Seconds between passes.
    """
    if self._cleanup_task is not None:
        return

    async def loop() -> None:
        while True:
            await asyncio.sleep(interval)
            try:
                await self.cleanup_idle()
            except asyncio.CancelledError:
                raise
            except Exception:
                _logger.exception("Idle sandbox cleanup failed; retrying next interval")

    self._cleanup_task = asyncio.create_task(loop())

shutdown() async

Stop every session and the cleanup loop.

Sessions are stopped concurrently, because stopping one is seconds of waiting for the process inside to die and they do not wait on each other: sequentially, a full pool turned a shutdown into minutes, and an orchestrator that loses patience kills the process mid-teardown.

Returns:

Type	Description
int	Number of sessions that were stopped.

Source code in src/pydantic_ai_backends/backends/docker/session.py

async def shutdown(self) -> int:
    """Stop every session and the cleanup loop.

    Sessions are stopped concurrently, because stopping one is seconds of
    waiting for the process inside to die and they do not wait on each
    other: sequentially, a full pool turned a shutdown into minutes, and
    an orchestrator that loses patience kills the process mid-teardown.

    Returns:
        Number of sessions that were stopped.
    """
    self.stop_cleanup_loop()

    session_ids = list(self._sessions)
    # `return_exceptions`, so one uncooperative sandbox cannot leave the
    # rest of the pool running — a shutdown has no later attempt.
    outcomes = await asyncio.gather(
        *(self.release(session_id) for session_id in session_ids),
        return_exceptions=True,
    )
    for session_id, outcome in zip(session_ids, outcomes, strict=True):
        if isinstance(outcome, BaseException):
            _logger.warning(
                "Session %s did not stop cleanly during shutdown: %s", session_id, outcome
            )
    return len(session_ids)

RuntimeConfig

The runtime descriptor (image, setup commands, environment) used by DockerSandbox and the session manager is documented in the type reference: RuntimeConfig.

Built-in Runtimes

Python

from pydantic_ai_backends import BUILTIN_RUNTIMES

# Available runtimes
print(sorted(BUILTIN_RUNTIMES))

# Use a runtime
from pydantic_ai_backends import DockerSandbox

sandbox = DockerSandbox(runtime="python-datascience")

Runtime	Image	What it adds
python-minimal	python:3.12-slim	standard library only
python-datascience	built on python:3.12-slim	pandas, numpy, matplotlib, scikit-learn, seaborn
python-analytics	built on python:3.12-slim	duckdb, polars, pyarrow
python-web	built on python:3.12-slim	fastapi, uvicorn, sqlalchemy, httpx
python-scraping	built on python:3.12-slim	httpx, beautifulsoup4, lxml, markdownify
python-documents	built on python:3.12-slim	pypdf, python-docx, openpyxl, pillow
node-minimal	node:20-slim	nothing
node-typescript	built on node:20-slim	typescript, tsx, vitest
node-react	built on node:20-slim	typescript, vite, react, react-dom, @types/react
bun	oven/bun:1-slim	Bun's own bundler, test runner and package manager
deno	denoland/deno:alpine	TypeScript with no install step
go	golang:1.23-alpine	Go toolchain
rust	rust:1-slim	Rust toolchain with cargo

A runtime naming an image starts as fast as a pull. One naming a base_image plus packages builds an image on first use and hits the cache afterwards, which is worth it when installing them per session would dominate.