Docker API

DockerSandbox

The container DockerWorkspace runs on. Use it directly to manage containers yourself; it runs commands through the CommandRunner protocol and has no file operations of its own — reach files through a workspace.

pydantic_ai_backends.backends.docker.sandbox.DockerSandbox

One Docker container: its lifecycle, its limits, and commands run in it.

The container starts lazily on the first command or an explicit :meth:start. Commands go through :meth:run_command, which keeps the workspace contract; files are reached through those commands by whoever uses the container — :class:~pydantic_ai_backends.workspaces.DockerWorkspace, or sandboxd, which also mounts the work directory from the host.

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:
    """One Docker container: its lifecycle, its limits, and commands run in it.

    The container starts lazily on the first command or an explicit
    :meth:`start`. Commands go through :meth:`run_command`, which keeps the
    workspace contract; files are reached through those commands by whoever uses
    the container — :class:`~pydantic_ai_backends.workspaces.DockerWorkspace`, or
    `sandboxd`, which also mounts the work directory from the host.

    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,
        oci_runtime: str | None = None,
        reattach_only: bool = False,
    ):
        """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.
            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.
            reattach_only: Start the container named `container_name` and never
                create one. For a caller whose files live only in that container:
                a replacement would be empty, so a missing or dead container
                raises `SandboxUnavailableError` from `start()` instead.

        Raises:
            ValueError: If `reattach_only` is set without `container_name`.
        """
        if reattach_only and container_name is None:
            raise ValueError("reattach_only needs a container_name to reattach to")
        self._id = session_id or sandbox_id or str(uuid.uuid4())

        self._container_name = container_name
        self._reattach_only = reattach_only
        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._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 id(self) -> str:
        """Unique identifier for this sandbox."""
        return self._id

    @property
    def last_activity(self) -> float:
        """Wall clock of the last operation, which idle cleanup reaps against."""
        return self._last_activity

    def touch(self) -> None:
        """Record activity, so idle cleanup does not reap a sandbox in use."""
        self._last_activity = time.time()

    @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

    @property
    def work_dir(self) -> str:
        """Directory commands start in and relative paths resolve against."""
        return self._work_dir

    # ── 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
        if self._reattach_only:
            raise SandboxUnavailableError(f"container {self._container_name} is gone")

        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 container_user(self) -> str | None:
        """The user the container runs as, as Docker reports it, or `None` before it exists.

        An empty string is the image's default, which for every image this
        library builds without `run_as_uid` is root. A name rather than a number
        is the image's own user, whose uid only the container knows.
        """
        if self._container is None:
            return None
        config = self._container.attrs.get("Config") or {}
        user = config.get("User")
        return user if isinstance(user, str) else None

    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 - the Kubernetes pod spells 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 of an anonymous container on garbage collection.

        A named container is left running: it exists to outlive this object, and
        a `DockerWorkspace` drops its sandbox after every run while the next run,
        or another backend on the same ref, is still using the container.
        Stopping it here killed their commands and every background process.

        `__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 and not self._container_name:
                self.stop()

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

    async def run_command(
        self,
        argv: Sequence[str],
        *,
        run_id: str,
        env: Mapping[str, str] | None = None,
        timeout: float | None = None,
        output_limit: int | None = None,
    ) -> CommandOutcome:
        """Run `argv` under the workspace failure contract.

        Raises when the container is unreachable, keeps stdout and stderr apart,
        and stops the command's process group when the caller times out or is
        cancelled. It is what a Pydantic AI workspace and `sandboxd`'s `/run` are
        built on.

        Args:
            argv: The program and its arguments, passed through literally.
            run_id: Names the run so :meth:`stop_command` can stop it.
            env: Variables layered over the container's environment.
            timeout: Seconds before the command is stopped; none when `None`.
            output_limit: Combined output bytes before the command is stopped;
                the module default when `None`.

        Raises:
            SandboxUnavailableError: The container is gone or not running.
        """
        # Imported here: `anyio` arrives with the `workspaces` and `server` extras,
        # the two callers of this method, and not with `docker`.
        import anyio.to_thread

        from pydantic_ai_backends.backends.docker import _exec

        await anyio.to_thread.run_sync(self._ensure_container, limiter=_exec.control_limiter())
        self._last_activity = time.time()
        assert self._container is not None
        try:
            return await _exec.run_in_container(
                self._container,
                argv,
                run_id=run_id,
                env=env,
                workdir=self._work_dir,
                timeout=timeout,
                output_limit=output_limit if output_limit is not None else MAX_RUN_OUTPUT_BYTES,
            )
        except SandboxUnavailableError:
            self._alive_checked_at = None
            raise

    async def stop_command(self, run_id: str) -> None:
        """Stop a command :meth:`run_command` started, and everything it started.

        Does nothing when the run has already finished or never started here.
        """
        from pydantic_ai_backends.backends.docker import _exec

        if self._container is not None:
            await _exec.stop_in_container(self._container, run_id)

runtime property

The runtime configuration for this sandbox.

id property

Unique identifier for this sandbox.

session_id property

Alias for the sandbox id, used for session management.

work_dir property

Directory commands start in and relative paths resolve against.

__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, oci_runtime=None, reattach_only=False)

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
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
reattach_only	bool	Start the container named container_name and never create one. For a caller whose files live only in that container: a replacement would be empty, so a missing or dead container raises SandboxUnavailableError from start() instead.	False

Raises:

Type	Description
ValueError	If reattach_only is set without container_name.

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,
    oci_runtime: str | None = None,
    reattach_only: bool = False,
):
    """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.
        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.
        reattach_only: Start the container named `container_name` and never
            create one. For a caller whose files live only in that container:
            a replacement would be empty, so a missing or dead container
            raises `SandboxUnavailableError` from `start()` instead.

    Raises:
        ValueError: If `reattach_only` is set without `container_name`.
    """
    if reattach_only and container_name is None:
        raise ValueError("reattach_only needs a container_name to reattach to")
    self._id = session_id or sandbox_id or str(uuid.uuid4())

    self._container_name = container_name
    self._reattach_only = reattach_only
    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._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

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 - the Kubernetes pod spells 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 - the Kubernetes pod spells 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

resource_usage()

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.

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

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

run_command(argv, *, run_id, env=None, timeout=None, output_limit=None) async

Run argv under the workspace failure contract.

Raises when the container is unreachable, keeps stdout and stderr apart, and stops the command's process group when the caller times out or is cancelled. It is what a Pydantic AI workspace and sandboxd's /run are built on.

Parameters:

Name	Type	Description	Default
argv	Sequence[str]	The program and its arguments, passed through literally.	required
run_id	str	Names the run so :meth:stop_command can stop it.	required
env	Mapping[str, str] | None	Variables layered over the container's environment.	None
timeout	float | None	Seconds before the command is stopped; none when None.	None
output_limit	int | None	Combined output bytes before the command is stopped; the module default when None.	None

Raises:

Type	Description
SandboxUnavailableError	The container is gone or not running.

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

async def run_command(
    self,
    argv: Sequence[str],
    *,
    run_id: str,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
    output_limit: int | None = None,
) -> CommandOutcome:
    """Run `argv` under the workspace failure contract.

    Raises when the container is unreachable, keeps stdout and stderr apart,
    and stops the command's process group when the caller times out or is
    cancelled. It is what a Pydantic AI workspace and `sandboxd`'s `/run` are
    built on.

    Args:
        argv: The program and its arguments, passed through literally.
        run_id: Names the run so :meth:`stop_command` can stop it.
        env: Variables layered over the container's environment.
        timeout: Seconds before the command is stopped; none when `None`.
        output_limit: Combined output bytes before the command is stopped;
            the module default when `None`.

    Raises:
        SandboxUnavailableError: The container is gone or not running.
    """
    # Imported here: `anyio` arrives with the `workspaces` and `server` extras,
    # the two callers of this method, and not with `docker`.
    import anyio.to_thread

    from pydantic_ai_backends.backends.docker import _exec

    await anyio.to_thread.run_sync(self._ensure_container, limiter=_exec.control_limiter())
    self._last_activity = time.time()
    assert self._container is not None
    try:
        return await _exec.run_in_container(
            self._container,
            argv,
            run_id=run_id,
            env=env,
            workdir=self._work_dir,
            timeout=timeout,
            output_limit=output_limit if output_limit is not None else MAX_RUN_OUTPUT_BYTES,
        )
    except SandboxUnavailableError:
        self._alive_checked_at = None
        raise

stop_command(run_id) async

Stop a command :meth:run_command started, and everything it started.

Does nothing when the run has already finished or never started here.

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

async def stop_command(self, run_id: str) -> None:
    """Stop a command :meth:`run_command` started, and everything it started.

    Does nothing when the run has already finished or never started here.
    """
    from pydantic_ai_backends.backends.docker import _exec

    if self._container is not None:
        await _exec.stop_in_container(self._container, run_id)

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,
        in_use: Callable[[str], bool] | 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.
            in_use: Whether a session has an operation under way, for a caller
                that tracks them. Idle cleanup never stops such a session: its
                last activity is when the operation began, and a command longer
                than the idle timeout is not idle.
        """
        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
        self._in_use = in_use
        # 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] = {}
        # Sessions being started, counted against `max_sessions` with the open
        # ones: a start suspends, and checking open ones alone let every start
        # under way see room for itself (AUD-1019-03).
        self._starting: set[str] = set()
        # Every `get_or_create` runs as one of these, shielded from its caller:
        # a start runs on a thread that cancelling the caller does not stop, so
        # the task finishes and registers what it started instead (AUD-1019-05).
        self._opening: set[asyncio.Task[Any]] = set()

    @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.

        Cancelling the caller does not cancel a start under way. A sandbox's
        `start` is a call on a worker thread that keeps going regardless, so a
        start abandoned half-way used to leave a running container no session
        named. The start finishes and the sandbox is registered like any other,
        for `release` or idle cleanup to stop; `shutdown` waits for it.

        Raises:
            SessionLimitExceeded: If `max_sessions` is reached and `session_id`
                is not an existing live session.
        """
        opening = asyncio.ensure_future(self._get_or_create(session_id, runtime))
        self._opening.add(opening)
        opening.add_done_callback(self._opened)
        return await asyncio.shield(opening)

    def _opened(self, task: asyncio.Task[Any]) -> None:
        self._opening.discard(task)
        # Retrieved here so an open whose caller went away, and which then
        # failed, is not reported as an exception nobody retrieved.
        if not task.cancelled() and task.exception() is not None:
            _logger.debug("Session open failed: %s", task.exception())

    async def _get_or_create(self, session_id: str, runtime: RuntimeConfig | str | None) -> Any:
        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)

            # Checked and claimed with no await between them, so it is atomic.
            if (
                self._max_sessions is not None
                and len(self._sessions) + len(self._starting) >= self._max_sessions
            ):
                raise SessionLimitExceeded(self._max_sessions)
            self._starting.add(session_id)
            try:
                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
            finally:
                self._starting.discard(session_id)

    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()
        idle = [
            session_id
            for session_id, sandbox in self._sessions.items()
            if self._idle(session_id, sandbox, now, max_idle)
        ]

        released = 0
        for session_id in idle:
            released += await self._release_if_idle(session_id, max_idle)

        self._prune_locks()
        return released

    def _idle(self, session_id: str, sandbox: Any, now: float, max_idle: int | None) -> bool:
        if self._in_use is not None and self._in_use(session_id):
            return False
        # 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.
        limit = max_idle if max_idle is not None else self._idle_limit_for(sandbox)
        return now - last_activity_of(sandbox, now) > limit

    async def _release_if_idle(self, session_id: str, max_idle: int | None) -> bool:
        """`release`, if the session is still idle once its lock is held.

        Checked again under the lock: between the scan and here, a request may
        have used the session or started an operation in it (AUD-1019-06).
        """
        async with self._session_lock(session_id):
            sandbox = self._sessions.get(session_id)
            if sandbox is None or not self._idle(session_id, sandbox, time.time(), max_idle):
                return False
            self._sessions.pop(session_id)
            await self._lifecycle(sandbox.stop)
        if self._on_release is not None:
            self._on_release(session_id)
        return True

    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()
        # A start under way registers its sandbox when it finishes; stopped
        # before then, that sandbox would outlive the shutdown.
        if self._opening:
            await asyncio.gather(*self._opening, return_exceptions=True)

        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, in_use=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
in_use	Callable[[str], bool] | None	Whether a session has an operation under way, for a caller that tracks them. Idle cleanup never stops such a session: its last activity is when the operation began, and a command longer than the idle timeout is not idle.	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,
    in_use: Callable[[str], bool] | 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.
        in_use: Whether a session has an operation under way, for a caller
            that tracks them. Idle cleanup never stops such a session: its
            last activity is when the operation began, and a command longer
            than the idle timeout is not idle.
    """
    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
    self._in_use = in_use
    # 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] = {}
    # Sessions being started, counted against `max_sessions` with the open
    # ones: a start suspends, and checking open ones alone let every start
    # under way see room for itself (AUD-1019-03).
    self._starting: set[str] = set()
    # Every `get_or_create` runs as one of these, shielded from its caller:
    # a start runs on a thread that cancelling the caller does not stop, so
    # the task finishes and registers what it started instead (AUD-1019-05).
    self._opening: set[asyncio.Task[Any]] = set()

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

Cancelling the caller does not cancel a start under way. A sandbox's start is a call on a worker thread that keeps going regardless, so a start abandoned half-way used to leave a running container no session named. The start finishes and the sandbox is registered like any other, for release or idle cleanup to stop; shutdown waits for it.

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.

    Cancelling the caller does not cancel a start under way. A sandbox's
    `start` is a call on a worker thread that keeps going regardless, so a
    start abandoned half-way used to leave a running container no session
    named. The start finishes and the sandbox is registered like any other,
    for `release` or idle cleanup to stop; `shutdown` waits for it.

    Raises:
        SessionLimitExceeded: If `max_sessions` is reached and `session_id`
            is not an existing live session.
    """
    opening = asyncio.ensure_future(self._get_or_create(session_id, runtime))
    self._opening.add(opening)
    opening.add_done_callback(self._opened)
    return await asyncio.shield(opening)

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()
    idle = [
        session_id
        for session_id, sandbox in self._sessions.items()
        if self._idle(session_id, sandbox, now, max_idle)
    ]

    released = 0
    for session_id in idle:
        released += await self._release_if_idle(session_id, max_idle)

    self._prune_locks()
    return released

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())

stop_cleanup_loop()

Stop the background cleanup loop.

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

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

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()
    # A start under way registers its sandbox when it finishes; stopped
    # before then, that sandbox would outlive the shutdown.
    if self._opening:
        await asyncio.gather(*self._opening, return_exceptions=True)

    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, DockerWorkspace 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 DockerWorkspace

workspace = DockerWorkspace(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:24-slim	nothing
node-typescript	built on node:24-slim	typescript, tsx, vitest
node-react	built on node:24-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.