Bases: PythonEnvironment
A Python environment using a virtual environment (venv).
Initialize a virtual environment for Python execution.
If you pass in a venv_path the path will be checked for a valid venv. If the venv doesn't exist it will be created using the python_version or python_path provided.
If the python_version or python_path is provided and the venv_path is not, a temporary directory will be created for venv and it will be setup with the provided python version.
If python_path is provided, it will take precedence over python_version.
The python version will not be installed if it doesn't exist and a RuntimeError will be raised.
| PARAMETER | DESCRIPTION |
python_version | The Python version to use (e.g., "3.11"), otherwise defaults to the current executing Python version. Ignored if venv_path is provided and has a valid environment already. TYPE: Optional[str] DEFAULT: None |
python_path | Optional direct path to a Python executable to use (must include the executable). Takes precedence over python_version if both are provided. TYPE: Optional[str] DEFAULT: None |
venv_path | Optional path for the virtual environment, will create it if it doesn't exist. If None, creates a temp directory. TYPE: Optional[str] DEFAULT: None |
Source code in autogen/environments/venv_python_environment.py
| def __init__(
self,
python_version: Optional[str] = None,
python_path: Optional[str] = None,
venv_path: Optional[str] = None,
):
"""
Initialize a virtual environment for Python execution.
If you pass in a venv_path the path will be checked for a valid venv. If the venv doesn't exist it will be created using the python_version or python_path provided.
If the python_version or python_path is provided and the venv_path is not, a temporary directory will be created for venv and it will be setup with the provided python version.
If python_path is provided, it will take precedence over python_version.
The python version will not be installed if it doesn't exist and a RuntimeError will be raised.
Args:
python_version: The Python version to use (e.g., "3.11"), otherwise defaults to the current executing Python version. Ignored if venv_path is provided and has a valid environment already.
python_path: Optional direct path to a Python executable to use (must include the executable). Takes precedence over python_version if both are provided.
venv_path: Optional path for the virtual environment, will create it if it doesn't exist. If None, creates a temp directory.
"""
self.python_version = python_version
self.python_path = python_path
self.venv_path = venv_path
self.created_venv = False
self._executable = None
super().__init__()
|
instance-attribute
python_version = python_version
instance-attribute
python_path = python_path
instance-attribute
instance-attribute
Get the path to the Python executable in the virtual environment.
Source code in autogen/environments/venv_python_environment.py
| def get_executable(self) -> str:
"""Get the path to the Python executable in the virtual environment."""
if not self._executable or not os.path.exists(self._executable):
raise RuntimeError("Virtual environment Python executable not found")
return self._executable
|
async
execute_code(code, script_path, timeout=30)
Execute code in the virtual environment.
Source code in autogen/environments/venv_python_environment.py
| async def execute_code(self, code: str, script_path: str, timeout: int = 30) -> dict[str, Any]:
"""Execute code in the virtual environment."""
try:
# Get the Python executable
python_executable = self.get_executable()
# Verify the executable exists
if not os.path.exists(python_executable):
return {"success": False, "error": f"Python executable not found at {python_executable}"}
# Ensure the directory for the script exists
script_dir = os.path.dirname(script_path)
if script_dir:
os.makedirs(script_dir, exist_ok=True)
# Write the code to the script file using asyncify (from base class)
await asyncify(self._write_to_file)(script_path, code)
logging.info(f"Wrote code to {script_path}")
try:
# Execute directly with subprocess using asyncify for better reliability
result = await asyncify(self._run_subprocess)([python_executable, script_path], timeout)
# Main execution result
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr,
"returncode": result.returncode,
}
except subprocess.TimeoutExpired:
return {"success": False, "error": f"Execution timed out after {timeout} seconds"}
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
|