Spaces:
Running
Running
File size: 2,088 Bytes
fb1a676 | 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 | """
NeuraPrompt Agent — Code Execution Tools (v7.5)
"""
import subprocess
import tempfile
import os
import re
from pathlib import Path
import logging
log = logging.getLogger("agent.tools.code")
def run_python(code: str) -> str:
"""Execute Python code safely"""
try:
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write(code)
temp_file = f.name
result = subprocess.run(
["python3", temp_file],
capture_output=True,
text=True,
timeout=20
)
Path(temp_file).unlink(missing_ok=True)
output = result.stdout.strip()
error = result.stderr.strip()
if error and "ModuleNotFoundError" in error:
# Auto-install missing package
match = re.search(r"No module named '(\w+)'", error)
if match:
package = match.group(1)
subprocess.run(["pip", "install", package], capture_output=True)
return run_python(code) # Retry after install
return output + (f"\n{error}" if error else "") or "(No output)"
except subprocess.TimeoutExpired:
return "Error: Code execution timed out (max 20 seconds)"
except Exception as e:
return f"Python execution error: {str(e)}"
def run_shell(command: str) -> str:
"""Execute shell commands safely"""
BLOCKED_COMMANDS = ["rm -rf", "mkfs", "dd if=", "shutdown", "reboot", "poweroff", "format"]
if any(blocked in command.lower() for blocked in BLOCKED_COMMANDS):
return "Error: Command blocked for safety reasons."
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=20
)
return (result.stdout + "\n" + result.stderr).strip() or "(No output)"
except subprocess.TimeoutExpired:
return "Error: Command timed out"
except Exception as e:
return f"Shell error: {str(e)}"
|