""" 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)}"