Bases: PythonEnvironment
A Python environment using the system's Python installation.
Initialize a system Python environment.
PARAMETER | DESCRIPTION |
executable | Optional path to a specific Python executable. If None, uses the current Python executable. TYPE: Optional[str] DEFAULT: None |
Source code in autogen/environments/system_python_environment.py
| def __init__(
self,
executable: Optional[str] = None,
):
"""
Initialize a system Python environment.
Args:
executable: Optional path to a specific Python executable. If None, uses the current Python executable.
"""
self._executable = executable or sys.executable
super().__init__()
|
get_executable
Get the path to the Python executable.
Source code in autogen/environments/system_python_environment.py
| def get_executable(self) -> str:
"""Get the path to the Python executable."""
return self._executable
|
execute_code async
execute_code(code, script_path, timeout=30)
Execute code using the system Python.
Source code in autogen/environments/system_python_environment.py
| async def execute_code(self, code: str, script_path: str, timeout: int = 30) -> dict[str, Any]:
"""Execute code using the system Python."""
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)}"}
|
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 |
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
|