Skip to content

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 = "docker",
    run_args: list[str] = list(),
)

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 = 'docker'

Path to the docker/container executable.

run_args class-attribute instance-attribute

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

Additional arguments to pass to the docker/container executable.

DockerEnvironment

DockerEnvironment(
    *,
    config_class: type = DockerEnvironmentConfig,
    **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
30
31
32
33
34
35
36
def __init__(self, *, config_class: type = DockerEnvironmentConfig, **kwargs):
    """This class executes bash commands in a Docker container using direct docker commands.
    See `DockerEnvironmentConfig` for keyword arguments.
    """
    self.container_id: str | None = None
    self.config = config_class(**kwargs)
    self._start_container()

container_id instance-attribute

container_id: str | None = None

config instance-attribute

config = config_class(**kwargs)

execute

execute(command: str, cwd: str = '') -> 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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def execute(self, command: str, cwd: str = "") -> 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=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
89
90
91
92
93
94
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
        print(f"Stopping container {self.container_id}")
        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)