Docker
Docker Environment class
Full source code
import logging
import os
import shlex
import subprocess
import uuid
from dataclasses import asdict, dataclass, field
from typing import Any
@dataclass
class DockerEnvironmentConfig:
image: str
cwd: str = "/"
"""Working directory in which to execute commands."""
env: dict[str, str] = field(default_factory=dict)
"""Environment variables to set in the container."""
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: int = 30
"""Timeout for executing commands in the container."""
executable: str = os.getenv("MSWEA_DOCKER_EXECUTABLE", "docker")
"""Path to the docker/container executable."""
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: str = "2h"
"""Max duration to keep container running. Uses the same format as the sleep command."""
pull_timeout: int = 120
"""Timeout in seconds for pulling images."""
class DockerEnvironment:
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()
def get_template_vars(self) -> dict[str, Any]:
return asdict(self.config)
def _start_container(self):
"""Start the Docker container and return the container ID."""
container_name = f"minisweagent-{uuid.uuid4().hex[:8]}"
cmd = [
self.config.executable,
"run",
"-d",
"--name",
container_name,
"-w",
self.config.cwd,
*self.config.run_args,
self.config.image,
"sleep",
self.config.container_timeout,
]
self.logger.debug(f"Starting container with command: {shlex.join(cmd)}")
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=self.config.pull_timeout, # docker pull might take a while
check=True,
)
self.logger.info(f"Started container {container_name} with ID {result.stdout.strip()}")
self.container_id = result.stdout.strip()
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}
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)
def __del__(self):
"""Cleanup container when object is destroyed."""
self.cleanup()
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,
)
instance-attribute
image: str
class-attribute
instance-attribute
cwd: str = '/'
Working directory in which to execute commands.
class-attribute
instance-attribute
env: dict[str, str] = field(default_factory=dict)
Environment variables to set in the container.
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.
class-attribute
instance-attribute
timeout: int = 30
Timeout for executing commands in the container.
class-attribute
instance-attribute
executable: str = getenv(
"MSWEA_DOCKER_EXECUTABLE", "docker"
)
Path to the docker/container executable.
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.
class-attribute
instance-attribute
container_timeout: str = '2h'
Max duration to keep container running. Uses the same format as the sleep command.
class-attribute
instance-attribute
pull_timeout: int = 120
Timeout in seconds for pulling images.
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
37 38 39 40 41 42 43 44 | |
instance-attribute
logger = logger or getLogger('minisweagent.environment')
instance-attribute
container_id: str | None = None
instance-attribute
config = config_class(**kwargs)
get_template_vars() -> dict[str, Any]
Source code in src/minisweagent/environments/docker.py
46 47 | |
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
76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | |
cleanup()
Stop and remove the Docker container.
Source code in src/minisweagent/environments/docker.py
100 101 102 103 104 | |
