DockerPythonEnvironment(image='python:3.11-slim', container_name_prefix='ag2_docker_env_', volumes=None, environment=None, network=None, pip_packages=None, requirements_file=None, dockerfile=None, build_args=None, cleanup_container=True, keep_container_running=False, container_startup_timeout=30)
Bases: PythonEnvironment
A Python environment using Docker containers for isolated execution.
Initialize a Docker Python environment.
| PARAMETER | DESCRIPTION |
image | Docker image to use (ignored if dockerfile is provided) TYPE: str DEFAULT: 'python:3.11-slim' |
container_name_prefix | Prefix for container names TYPE: str DEFAULT: 'ag2_docker_env_' |
volumes | Dictionary mapping host paths to container paths for mounting TYPE: Optional[dict[str, str]] DEFAULT: None |
environment | Dictionary of environment variables to set in the container TYPE: Optional[dict[str, str]] DEFAULT: None |
network | Docker network to attach the container to TYPE: Optional[str] DEFAULT: None |
pip_packages | List of pip packages to install in the container TYPE: Optional[list[str]] DEFAULT: None |
requirements_file | Path to requirements.txt file to install in the container TYPE: Optional[str] DEFAULT: None |
dockerfile | Optional path to a Dockerfile to build and use instead of pulling an image TYPE: Optional[str] DEFAULT: None |
build_args | Optional build arguments for the Dockerfile TYPE: Optional[dict[str, str]] DEFAULT: None |
cleanup_container | Whether to remove the container after use TYPE: bool DEFAULT: True |
keep_container_running | Whether to keep the container running after execution TYPE: bool DEFAULT: False |
container_startup_timeout | Timeout in seconds for container startup TYPE: int DEFAULT: 30 |
Source code in autogen/environments/docker_python_environment.py
| def __init__(
self,
image: str = "python:3.11-slim",
container_name_prefix: str = "ag2_docker_env_",
volumes: Optional[dict[str, str]] = None,
environment: Optional[dict[str, str]] = None,
network: Optional[str] = None,
pip_packages: Optional[list[str]] = None,
requirements_file: Optional[str] = None,
dockerfile: Optional[str] = None,
build_args: Optional[dict[str, str]] = None,
cleanup_container: bool = True,
keep_container_running: bool = False,
container_startup_timeout: int = 30,
):
"""
Initialize a Docker Python environment.
Args:
image: Docker image to use (ignored if dockerfile is provided)
container_name_prefix: Prefix for container names
volumes: Dictionary mapping host paths to container paths for mounting
environment: Dictionary of environment variables to set in the container
network: Docker network to attach the container to
pip_packages: List of pip packages to install in the container
requirements_file: Path to requirements.txt file to install in the container
dockerfile: Optional path to a Dockerfile to build and use instead of pulling an image
build_args: Optional build arguments for the Dockerfile
cleanup_container: Whether to remove the container after use
keep_container_running: Whether to keep the container running after execution
container_startup_timeout: Timeout in seconds for container startup
"""
self.image = image
self.container_name_prefix = container_name_prefix
self.volumes = volumes or {}
self.environment = environment or {}
self.network = network
self.pip_packages = pip_packages or []
self.requirements_file = requirements_file
self.dockerfile = dockerfile
self.build_args = build_args or {}
self.cleanup_container = cleanup_container
self.keep_container_running = keep_container_running
self.container_startup_timeout = container_startup_timeout
# Internal state
self._container_id = None
self._container_name = None
self._custom_image_name = None
self._temp_dir = None
super().__init__()
|
instance-attribute
container_name_prefix = container_name_prefix
instance-attribute
instance-attribute
environment = environment or {}
instance-attribute
instance-attribute
pip_packages = pip_packages or []
instance-attribute
requirements_file = requirements_file
instance-attribute
instance-attribute
build_args = build_args or {}
instance-attribute
cleanup_container = cleanup_container
instance-attribute
keep_container_running = keep_container_running
instance-attribute
container_startup_timeout = container_startup_timeout
Get the path to the Python executable in the Docker container.
Source code in autogen/environments/docker_python_environment.py
| def get_executable(self) -> str:
"""Get the path to the Python executable in the Docker container."""
# This is a virtual path in the container
return "python"
|
async
execute_code(code, script_path, timeout=30)
Execute code in the Docker container.
Source code in autogen/environments/docker_python_environment.py
| async def execute_code(self, code: str, script_path: str, timeout: int = 30) -> dict[str, Any]:
"""Execute code in the Docker container."""
# Ensure the container is running
if not self._container_id:
return {"success": False, "error": "Docker container not started"}
try:
# Calculate the relative path within the temp directory
if os.path.isabs(script_path):
rel_path = os.path.basename(script_path)
host_script_path = os.path.join(self._temp_dir, rel_path)
else:
rel_path = script_path
host_script_path = os.path.join(self._temp_dir, rel_path)
# Ensure the directory for the script exists
script_dir = os.path.dirname(host_script_path)
if script_dir:
os.makedirs(script_dir, exist_ok=True)
# Write the code to the script file on the host
await asyncify(self._write_to_file)(host_script_path, code)
# Path to the script in the container
container_script_path = f"/workspace/{rel_path}"
# Execute the script in the container
exec_cmd = ["docker", "exec", self._container_name, "python", container_script_path]
# Run the command with a timeout
result = await asyncify(self._run_subprocess_with_timeout)(exec_cmd, timeout)
return {
"success": result[0],
"stdout": result[1],
"stderr": result[2],
"returncode": result[3] if result[0] else 1,
}
except Exception as e:
return {"success": False, "error": f"Execution error: {str(e)}"}
|
classmethod
get_current_python_environment(python_environment=None)
Get the current Python environment or the specified one if provided.
| PARAMETER | DESCRIPTION |
python_environment | Optional environment to return if specified. TYPE: Optional[PythonEnvironment] DEFAULT: None |
Source code in autogen/environments/python_environment.py
| @classmethod
def get_current_python_environment(
cls, python_environment: Optional["PythonEnvironment"] = None
) -> Optional["PythonEnvironment"]:
"""
Get the current Python environment or the specified one if provided.
Args:
python_environment: Optional environment to return if specified.
Returns:
The current Python environment or None if none is active.
"""
if python_environment is not None:
return python_environment
try:
return cls._current_python_environment.get()
except LookupError:
return None
|