From 4930ffbf2c9f7c1b101d94fae3cff72348f2131e Mon Sep 17 00:00:00 2001 From: embersax Date: Thu, 4 Sep 2025 22:00:10 -0700 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat:=20add=20E2B=20sandbox=20found?= =?UTF-8?q?ation=20for=20secure=20code=20execution?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add e2b-code-interpreter dependency (>=2.0.0) to pyproject.toml - Create sandbox_execution.py with SandboxExecutor class - Provide basic sandbox execution functionality - Maintain compatibility with existing execute_command interface This is the foundation for integrating E2B's secure sandbox environment. --- mle/function/sandbox_execution.py | 175 ++++++++++++++++++++++++++++++ pyproject.toml | 1 + 2 files changed, 176 insertions(+) create mode 100644 mle/function/sandbox_execution.py diff --git a/mle/function/sandbox_execution.py b/mle/function/sandbox_execution.py new file mode 100644 index 0000000..5d69af8 --- /dev/null +++ b/mle/function/sandbox_execution.py @@ -0,0 +1,175 @@ +""" +Tools for secure code execution using E2B sandbox. +""" + +import os +from typing import Optional, Callable, Any +from collections import deque + +try: + from e2b_code_interpreter import Sandbox + E2B_AVAILABLE = True +except ImportError: + E2B_AVAILABLE = False + Sandbox = None + + +class SandboxExecutor: + """ + A secure sandbox executor using E2B for isolated code execution. + """ + + def __init__(self, api_key: Optional[str] = None): + """ + Initialize the sandbox executor. + + Args: + api_key: E2B API key. If not provided, will try to read from E2B_API_KEY env var. + """ + if not E2B_AVAILABLE: + raise ImportError( + "e2b-code-interpreter is not installed. " + "Please install it with: pip install e2b-code-interpreter>=2.0.0" + ) + + self.api_key = api_key or os.getenv("E2B_API_KEY") + if not self.api_key: + raise ValueError( + "E2B API key is required. Please set E2B_API_KEY environment variable " + "or pass api_key parameter." + ) + + self.sandbox = None + self._stdout_buffer = [] + self._stderr_buffer = [] + + def __enter__(self): + """Context manager entry.""" + self.start() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit.""" + self.close() + + def start(self): + """Start the sandbox if not already started.""" + if not self.sandbox: + self.sandbox = Sandbox(api_key=self.api_key) + + def close(self): + """Close the sandbox if open.""" + if self.sandbox: + self.sandbox.close() + self.sandbox = None + + def _on_stdout(self, stdout: str): + """Callback for stdout output.""" + self._stdout_buffer.append(stdout) + print(stdout, end='') + + def _on_stderr(self, stderr: str): + """Callback for stderr output.""" + self._stderr_buffer.append(stderr) + print(stderr, end='') + + def execute_command(self, command: str, max_lines: int = 30) -> str: + """ + Execute a command in the sandbox. + + This method is designed to be compatible with the existing execute_command function + in mle/function/execution.py, maintaining the same interface and return format. + + Args: + command: The command to execute + max_lines: Maximum number of output lines to return + + Returns: + A string with exit code and output in the same format as the original function + """ + if not self.sandbox: + self.start() + + # Clear buffers + self._stdout_buffer.clear() + self._stderr_buffer.clear() + + try: + # Execute the command in the sandbox with output callbacks + execution = self.sandbox.run_code( + command, + on_stdout=self._on_stdout, + on_stderr=self._on_stderr + ) + + # Determine exit code based on error status + exit_code = 1 if execution.error else 0 + + # Combine stdout and stderr + output_lines = [] + + # Add stdout + if self._stdout_buffer: + output_lines.extend(self._stdout_buffer) + + # Add stderr + if self._stderr_buffer: + output_lines.extend(self._stderr_buffer) + + # Add execution results if available + if hasattr(execution, 'results') and execution.results: + output_lines.extend(str(execution.results).split('\n')) + + # Add error message if present + if execution.error: + output_lines.append(f"Error: {execution.error}") + + # Join all output + full_output = ''.join(output_lines) + + # Limit output lines similar to the original function + lines = full_output.split('\n') + if len(lines) > max_lines: + # Use deque to efficiently get the last N lines + output_buffer = deque(lines, maxlen=max_lines) + limited_output = '\n'.join(output_buffer) + return f"Exit code: {exit_code}\nOutput (last {max_lines} lines):\n{limited_output}" + else: + return f"Exit code: {exit_code}\nOutput:\n{full_output}" + + except Exception as e: + return f"Error running command in sandbox: {str(e)}" + + def is_available(self) -> bool: + """Check if the sandbox is available and properly configured.""" + return E2B_AVAILABLE and bool(self.api_key) + + +def execute_in_sandbox(command: str, max_lines: int = 30, api_key: Optional[str] = None) -> str: + """ + Convenience function to execute a single command in a sandbox. + + Args: + command: The command to execute + max_lines: Maximum number of output lines to return + api_key: Optional E2B API key + + Returns: + Command output in the same format as mle/function/execution.py + """ + with SandboxExecutor(api_key=api_key) as executor: + return executor.execute_command(command, max_lines) + + +def check_sandbox_availability() -> bool: + """ + Check if E2B sandbox is available and configured. + + Returns: + True if e2b-code-interpreter is installed and API key is set + """ + if not E2B_AVAILABLE: + return False + + api_key = os.getenv("E2B_API_KEY") + return bool(api_key) \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index e08ebbf..6fc48d4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ dependencies = [ "mem0ai~=0.1.114", "pip>=23.3.1", "setuptools>=65.5.0", + "e2b-code-interpreter>=2.0.0", ] [project.optional-dependencies]