Docker

minisweagent.environments.docker

DockerEnvironmentConfig `dataclass`

DockerEnvironmentConfig(
    image: str,
    cwd: str = "/",
    env: dict[str, str] = dict(),
    forward_env: list[str] = list(),
    timeout: int = 30,
    executable: str = getenv(
        "MSWEA_DOCKER_EXECUTABLE", "docker"
    ),
    run_args: list[str] = (lambda: ["--rm"])(),
    container_timeout: str = "2h",
    pull_timeout: int = 120,
)

image `instance-attribute`

image: str

cwd `class-attribute` `instance-attribute`

cwd: str = '/'

Working directory in which to execute commands.

env `class-attribute` `instance-attribute`

env: dict[str, str] = field(default_factory=dict)

Environment variables to set in the container.

forward_env `class-attribute` `instance-attribute`

forward_env: list[str] = field(default_factory=list)

Environment variables to forward to the container. Variables are only forwarded if they are set in the host environment. In case of conflict with env, the env variables take precedence.

timeout `class-attribute` `instance-attribute`

timeout: int = 30

Timeout for executing commands in the container.

executable `class-attribute` `instance-attribute`

executable: str = getenv(
    "MSWEA_DOCKER_EXECUTABLE", "docker"
)

Path to the docker/container executable.

run_args `class-attribute` `instance-attribute`

run_args: list[str] = field(
    default_factory=lambda: ["--rm"]
)

Additional arguments to pass to the docker/container executable. Default is ["--rm"], which removes the container after it exits.

container_timeout `class-attribute` `instance-attribute`

container_timeout: str = '2h'

Max duration to keep container running. Uses the same format as the sleep command.

pull_timeout `class-attribute` `instance-attribute`

pull_timeout: int = 120

Timeout in seconds for pulling images.

DockerEnvironment

DockerEnvironment(
    *,
    config_class: type = DockerEnvironmentConfig,
    logger: Logger | None = None,
    **kwargs,
)

This class executes bash commands in a Docker container using direct docker commands. See DockerEnvironmentConfig for keyword arguments.

Source code in src/minisweagent/environments/docker.py

def __init__(self, *, config_class: type = DockerEnvironmentConfig, logger: logging.Logger | None = None, **kwargs):
    """This class executes bash commands in a Docker container using direct docker commands.
    See `DockerEnvironmentConfig` for keyword arguments.
    """
    self.logger = logger or logging.getLogger("minisweagent.environment")
    self.container_id: str | None = None
    self.config = config_class(**kwargs)
    self._start_container()

logger `instance-attribute`

logger = logger or getLogger('minisweagent.environment')

container_id `instance-attribute`

container_id: str | None = None

config `instance-attribute`

config = config_class(**kwargs)

get_template_vars

get_template_vars() -> dict[str, Any]

Source code in src/minisweagent/environments/docker.py

def get_template_vars(self) -> dict[str, Any]:
    return asdict(self.config)

execute

execute(
    command: str,
    cwd: str = "",
    *,
    timeout: int | None = None,
) -> dict[str, Any]

Execute a command in the Docker container and return the result as a dict.

Source code in src/minisweagent/environments/docker.py

def execute(self, command: str, cwd: str = "", *, timeout: int | None = None) -> dict[str, Any]:
    """Execute a command in the Docker container and return the result as a dict."""
    cwd = cwd or self.config.cwd
    assert self.container_id, "Container not started"

    cmd = [self.config.executable, "exec", "-w", cwd]
    for key in self.config.forward_env:
        if (value := os.getenv(key)) is not None:
            cmd.extend(["-e", f"{key}={value}"])
    for key, value in self.config.env.items():
        cmd.extend(["-e", f"{key}={value}"])
    cmd.extend([self.container_id, "bash", "-lc", command])

    result = subprocess.run(
        cmd,
        text=True,
        timeout=timeout or self.config.timeout,
        encoding="utf-8",
        errors="replace",
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
    )
    return {"output": result.stdout, "returncode": result.returncode}

cleanup

cleanup()

Stop and remove the Docker container.

Source code in src/minisweagent/environments/docker.py

def cleanup(self):
    """Stop and remove the Docker container."""
    if getattr(self, "container_id", None) is not None:  # if init fails early, container_id might not be set
        cmd = f"(timeout 60 {self.config.executable} stop {self.container_id} || {self.config.executable} rm -f {self.container_id}) >/dev/null 2>&1 &"
        subprocess.Popen(cmd, shell=True)