DockerExecutionPolicy Class — langchain Architecture

Architecture documentation for the DockerExecutionPolicy class in _execution.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  3ead6362_15e4_31b5_852b_83ad2c63fa17["DockerExecutionPolicy"]
  54947cfa_44ad_df6e_ebe6_8908dfdafc92["BaseExecutionPolicy"]
  3ead6362_15e4_31b5_852b_83ad2c63fa17 -->|extends| 54947cfa_44ad_df6e_ebe6_8908dfdafc92
  a7063f14_dbac_d9f7_9cc6_a20aa09f5461["_execution.py"]
  3ead6362_15e4_31b5_852b_83ad2c63fa17 -->|defined in| a7063f14_dbac_d9f7_9cc6_a20aa09f5461
  09da5721_9419_a889_ea0d_6bba39e39a53["__post_init__()"]
  3ead6362_15e4_31b5_852b_83ad2c63fa17 -->|method| 09da5721_9419_a889_ea0d_6bba39e39a53
  ba1b6b39_fb63_7014_9fdb_7866618f263b["spawn()"]
  3ead6362_15e4_31b5_852b_83ad2c63fa17 -->|method| ba1b6b39_fb63_7014_9fdb_7866618f263b
  0ca39ab6_f475_3199_9fda_998b3047cb82["_build_command()"]
  3ead6362_15e4_31b5_852b_83ad2c63fa17 -->|method| 0ca39ab6_f475_3199_9fda_998b3047cb82
  73ea7e95_02e6_bd8d_5b08_873ffdee5093["_should_mount_workspace()"]
  3ead6362_15e4_31b5_852b_83ad2c63fa17 -->|method| 73ea7e95_02e6_bd8d_5b08_873ffdee5093
  a45f9aa3_f5f0_035d_1dae_62dc52165b5e["_resolve_binary()"]
  3ead6362_15e4_31b5_852b_83ad2c63fa17 -->|method| a45f9aa3_f5f0_035d_1dae_62dc52165b5e

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/_execution.py lines 267–377

class DockerExecutionPolicy(BaseExecutionPolicy):
    """Run the shell inside a dedicated Docker container.

    Choose this policy when commands originate from untrusted users or you require
    strong isolation between sessions. By default the workspace is bind-mounted only
    when it refers to an existing non-temporary directory; ephemeral sessions run
    without a mount to minimise host exposure. The container's network namespace is
    disabled by default (`--network none`) and you can enable further hardening via
    `read_only_rootfs` and `user`.

    The security guarantees depend on your Docker daemon configuration. Run the agent on
    a host where Docker is locked down (rootless mode, AppArmor/SELinux, etc.) and
    review any additional volumes or capabilities passed through ``extra_run_args``. The
    default image is `python:3.12-alpine3.19`; supply a custom image if you need
    preinstalled tooling.
    """

    binary: str = "docker"
    image: str = "python:3.12-alpine3.19"
    remove_container_on_exit: bool = True
    network_enabled: bool = False
    extra_run_args: Sequence[str] | None = None
    memory_bytes: int | None = None
    cpu_time_seconds: typing.Any | None = None
    cpus: str | None = None
    read_only_rootfs: bool = False
    user: str | None = None

    def __post_init__(self) -> None:
        super().__post_init__()
        if self.memory_bytes is not None and self.memory_bytes <= 0:
            msg = "memory_bytes must be positive if provided."
            raise ValueError(msg)
        if self.cpu_time_seconds is not None:
            msg = (
                "DockerExecutionPolicy does not support cpu_time_seconds; configure CPU limits "
                "using Docker run options such as '--cpus'."
            )
            raise RuntimeError(msg)
        if self.cpus is not None and not self.cpus.strip():
            msg = "cpus must be a non-empty string when provided."
            raise ValueError(msg)
        if self.user is not None and not self.user.strip():
            msg = "user must be a non-empty string when provided."
            raise ValueError(msg)
        self.extra_run_args = tuple(self.extra_run_args or ())

    def spawn(
        self,
        *,
        workspace: Path,
        env: Mapping[str, str],
        command: Sequence[str],
    ) -> subprocess.Popen[str]:
        full_command = self._build_command(workspace, env, command)
        host_env = os.environ.copy()
        return _launch_subprocess(
            full_command,
            env=host_env,
            cwd=workspace,
            preexec_fn=None,
            start_new_session=False,
        )

    def _build_command(
        self,
        workspace: Path,
        env: Mapping[str, str],
        command: Sequence[str],
    ) -> list[str]:
        binary = self._resolve_binary()
        full_command: list[str] = [binary, "run", "-i"]
        if self.remove_container_on_exit:
            full_command.append("--rm")
        if not self.network_enabled:
            full_command.extend(["--network", "none"])
        if self.memory_bytes is not None:
            full_command.extend(["--memory", str(self.memory_bytes)])
        if self._should_mount_workspace(workspace):
            host_path = str(workspace)
            full_command.extend(["-v", f"{host_path}:{host_path}"])