Spaces:
Sleeping
Sleeping
File size: 3,589 Bytes
ba5110e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 |
"""
Code execution tool with sandbox isolation.
Provides CodeTool class for safe Python code execution.
"""
import subprocess
import sys
import tempfile
import os
from typing import Dict, Any
class CodeTool:
"""
Safe Python code executor using subprocess isolation.
"""
def __init__(self, timeout: int = 30):
self.timeout = timeout
def execute(self, code: str) -> Dict[str, Any]:
"""
Execute Python code in isolated subprocess.
Args:
code: Python code to execute
Returns:
Dict with keys: success, output, error
"""
# Create temporary file
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(code)
temp_path = f.name
try:
# Execute in subprocess
result = subprocess.run(
[sys.executable, temp_path],
capture_output=True,
text=True,
timeout=self.timeout,
cwd=tempfile.gettempdir(),
env={**os.environ, "PYTHONPATH": ""}
)
if result.returncode == 0:
return {
"success": True,
"output": result.stdout.strip(),
"error": None
}
else:
return {
"success": False,
"output": result.stdout.strip() if result.stdout else None,
"error": result.stderr.strip() if result.stderr else "Unknown error"
}
except subprocess.TimeoutExpired:
return {
"success": False,
"output": None,
"error": f"Code execution timed out after {self.timeout} seconds"
}
except Exception as e:
return {
"success": False,
"output": None,
"error": str(e)
}
finally:
# Cleanup
try:
os.unlink(temp_path)
except:
pass
# Legacy function for backwards compatibility
def execute_python_code(code: str, timeout: int = 30) -> Dict[str, Any]:
"""Execute Python code (legacy wrapper)."""
tool = CodeTool(timeout=timeout)
return tool.execute(code)
async def execute_with_correction(
code: str,
correction_fn,
max_corrections: int = 2,
timeout: int = 30
) -> tuple:
"""
Execute code with automatic correction on error.
Args:
code: Initial Python code
correction_fn: Async function(code, error) -> corrected_code
max_corrections: Maximum correction attempts
timeout: Execution timeout
Returns:
Tuple of (success: bool, result: str, attempts: int)
"""
tool = CodeTool(timeout=timeout)
current_code = code
attempts = 0
while attempts <= max_corrections:
result = tool.execute(current_code)
if result["success"]:
return True, result["output"], attempts
if attempts >= max_corrections:
break
# Try to correct the code
try:
current_code = await correction_fn(current_code, result["error"])
attempts += 1
except Exception as e:
return False, f"Correction failed: {str(e)}", attempts
return False, result.get("error", "Max corrections reached"), attempts
|