""" Code execution tools for the GAIA Agent. Includes Python, JavaScript, and Bash executors with sandboxed execution. """ import os import subprocess import tempfile import shutil from langchain_core.tools import tool @tool def python_executor(code: str) -> str: """Execute Python code and return the output. Use this for calculations, data processing, string manipulation, or any computation. The code should print() any results you want to see. Args: code: Python code to execute """ try: with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, encoding='utf-8') as f: f.write(code) temp_path = f.name result = subprocess.run( ['python', temp_path], capture_output=True, text=True, timeout=60 ) os.unlink(temp_path) output = result.stdout.strip() if result.stderr: output += f"\nStderr: {result.stderr}" return output or "Code executed with no output." except subprocess.TimeoutExpired: if 'temp_path' in locals(): os.unlink(temp_path) return "Execution timed out (60s limit)" except Exception as e: if 'temp_path' in locals() and os.path.exists(temp_path): os.unlink(temp_path) return f"Execution error: {str(e)}" @tool def javascript_executor(code: str) -> str: """Execute JavaScript code using Node.js and return the output. Use this for JSON processing, string manipulation, or JavaScript-specific operations. Use console.log() to output results. Args: code: JavaScript code to execute """ # Check if Node.js is available if shutil.which("node") is None: return "Error: Node.js is not installed or not in PATH. Please install Node.js to use this tool." try: with tempfile.NamedTemporaryFile(mode='w', suffix='.js', delete=False, encoding='utf-8') as f: f.write(code) temp_path = f.name result = subprocess.run( ['node', temp_path], capture_output=True, text=True, timeout=60 ) os.unlink(temp_path) output = result.stdout.strip() if result.stderr: output += f"\nStderr: {result.stderr}" return output or "Code executed with no output." except subprocess.TimeoutExpired: if 'temp_path' in locals(): os.unlink(temp_path) return "Execution timed out (60s limit)" except Exception as e: if 'temp_path' in locals() and os.path.exists(temp_path): os.unlink(temp_path) return f"Execution error: {str(e)}" @tool def bash_executor(command: str) -> str: """Execute a Bash/Shell command and return the output. Use this for file operations, text processing with sed/awk/grep, or system commands. On Windows, this uses PowerShell. On Unix/Linux/Mac, this uses bash. Args: command: Shell command to execute """ try: # Determine the shell to use based on the OS import platform system = platform.system().lower() if system == "windows": # Use PowerShell on Windows shell_cmd = ["powershell", "-NoProfile", "-NonInteractive", "-Command", command] else: # Use bash on Unix-like systems shell_cmd = ["bash", "-c", command] result = subprocess.run( shell_cmd, capture_output=True, text=True, timeout=60, cwd=tempfile.gettempdir() # Run in temp directory for safety ) output = result.stdout.strip() if result.stderr: output += f"\nStderr: {result.stderr}" if result.returncode != 0: output += f"\nExit code: {result.returncode}" return output or "Command executed with no output." except subprocess.TimeoutExpired: return "Execution timed out (60s limit)" except Exception as e: return f"Execution error: {str(e)}"