DockerPythonEnvironment - AG2

DockerPythonEnvironment

autogen.environments.docker_python_environment.DockerPythonEnvironment

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:strDEFAULT:'python:3.11-slim'
container_name_prefix Prefix for container names
TYPE:strDEFAULT:'ag2_docker_env_'
volumes Dictionary mapping host paths to container paths for mounting
TYPE:`dict[str, str]
environment Dictionary of environment variables to set in the container
TYPE:`dict[str, str]
network Docker network to attach the container to
TYPE:`str
pip_packages List of pip packages to install in the container
TYPE:`list[str]
requirements_file Path to requirements.txt file to install in the container
TYPE:`str
dockerfile Optional path to a Dockerfile to build and use instead of pulling an image
TYPE:`str
build_args Optional build arguments for the Dockerfile
TYPE:`dict[str, str]
cleanup_container Whether to remove the container after use
TYPE:boolDEFAULT:True
keep_container_running Whether to keep the container running after execution
TYPE:boolDEFAULT:False
container_startup_timeout Timeout in seconds for container startup
TYPE:intDEFAULT:30

Source code in autogen/environments/docker_python_environment.py
<br>23<br>24<br>25<br>26<br>27<br>28<br>29<br>30<br>31<br>32<br>33<br>34<br>35<br>36<br>37<br>38<br>39<br>40<br>41<br>42<br>43<br>44<br>45<br>46<br>47<br>48<br>49<br>50<br>51<br>52<br>53<br>54<br>55<br>56<br>57<br>58<br>59<br>60<br>61<br>62<br>63<br>64<br>65<br>66<br>67<br>68<br>69<br>70<br>71<br>72<br>73<br> ```
def init(
self,
image: str = "python:3.11-slim",
container_name_prefix: str = "ag2_docker_env",
volumes: dict[str, str]

image instance-attribute

image = image

container_name_prefix instance-attribute

container_name_prefix = container_name_prefix

volumes instance-attribute

volumes = volumes or {}

environment instance-attribute

environment = environment or {}

network instance-attribute

network = network

pip_packages instance-attribute

pip_packages = pip_packages or []

requirements_file instance-attribute

requirements_file = requirements_file

dockerfile instance-attribute

dockerfile = dockerfile

build_args instance-attribute

build_args = build_args or {}

cleanup_container instance-attribute

cleanup_container = cleanup_container

keep_container_running instance-attribute

keep_container_running = keep_container_running

container_startup_timeout instance-attribute

container_startup_timeout = container_startup_timeout

get_executable

get_executable()

Get the path to the Python executable in the Docker container.

Source code in autogen/environments/docker_python_environment.py
<br>297<br>298<br>299<br>300<br> <br>def get_executable(self) -> str:<br> """Get the path to the Python executable in the Docker container."""<br> # This is a virtual path in the container<br> return "python"<br>

execute_code async

execute_code(code, script_path, timeout=30)

Execute code in the Docker container.

Source code in autogen/environments/docker_python_environment.py
<br>302<br>303<br>304<br>305<br>306<br>307<br>308<br>309<br>310<br>311<br>312<br>313<br>314<br>315<br>316<br>317<br>318<br>319<br>320<br>321<br>322<br>323<br>324<br>325<br>326<br>327<br>328<br>329<br>330<br>331<br>332<br>333<br>334<br>335<br>336<br>337<br>338<br>339<br>340<br>341<br>342<br> <br>async def execute_code(self, code: str, script_path: str, timeout: int = 30) -> dict[str, Any]:<br> """Execute code in the Docker container."""<br> # Ensure the container is running<br> if not self._container_id:<br> return {"success": False, "error": "Docker container not started"}<br> try:<br> # Calculate the relative path within the temp directory<br> if os.path.isabs(script_path):<br> rel_path = os.path.basename(script_path)<br> host_script_path = os.path.join(self._temp_dir, rel_path)<br> else:<br> rel_path = script_path<br> host_script_path = os.path.join(self._temp_dir, rel_path)<br> # Ensure the directory for the script exists<br> script_dir = os.path.dirname(host_script_path)<br> if script_dir:<br> os.makedirs(script_dir, exist_ok=True)<br> # Write the code to the script file on the host<br> await to_thread.run_sync(self._write_to_file, host_script_path, code)<br> # Path to the script in the container<br> container_script_path = f"/workspace/{rel_path}"<br> # Execute the script in the container<br> exec_cmd = ["docker", "exec", self._container_name, "python", container_script_path]<br> # Run the command with a timeout<br> result = await to_thread.run_sync(self._run_subprocess_with_timeout, exec_cmd, timeout)<br> return {<br> "success": result[0],<br> "stdout": result[1],<br> "stderr": result[2],<br> "returncode": result[3] if result[0] else 1,<br> }<br> except Exception as e:<br> return {"success": False, "error": f"Execution error: {str(e)}"}<br>

get_current_python_environment 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
RETURNS DESCRIPTION
Optional[PythonEnvironment] The current Python environment or None if none is active.

Source code in autogen/environments/python_environment.py
<br>108<br>109<br>110<br>111<br>112<br>113<br>114<br>115<br>116<br>117<br>118<br>119<br>120<br>121<br>122<br>123<br>124<br>125<br> <br>@classmethod<br>def get_current_python_environment(<br> cls, python_environment: Optional["PythonEnvironment"] = None<br>) -> Optional["PythonEnvironment"]:<br> """Get the current Python environment or the specified one if provided.<br> Args:<br> python_environment: Optional environment to return if specified.<br> Returns:<br> The current Python environment or None if none is active.<br> """<br> if python_environment is not None:<br> return python_environment<br> try:<br> return cls._current_python_environment.get()<br> except LookupError:<br> return None<br>