Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| 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)}" | |
| 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)}" | |
| 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)}" | |