Add E2B SandBox Support.

Add E2B sandbox for code execution, and basic Sandbox for computer use.

app/config.py

@@ -167,6 +167,7 @@ class AppConfig(BaseModel):
     run_flow_config: Optional[RunflowSettings] = Field(
         None, description="Run flow configuration"
     )
+    cloud_sandbox: Optional[dict] = Field(None, description="Cloud sandbox configuration")
 
     class Config:
         arbitrary_types_allowed = True
@@ -283,6 +284,7 @@ class Config:
             run_flow_settings = RunflowSettings(**run_flow_config)
         else:
             run_flow_settings = RunflowSettings()
+        cloud_sandbox_config = raw_config.get("cloud_sandbox", {})
         config_dict = {
             "llm": {
                 "default": default_settings,
@@ -296,6 +298,7 @@ class Config:
             "search_config": search_settings,
             "mcp_config": mcp_settings,
             "run_flow_config": run_flow_settings,
+            "cloud_sandbox": cloud_sandbox_config,
         }
 
         self._config = AppConfig(**config_dict)
@@ -326,6 +329,10 @@ class Config:
         """Get the Run Flow configuration"""
         return self._config.run_flow_config
 
+    @property
+    def cloud_sandbox(self) -> dict:
+        return getattr(self._config, "cloud_sandbox", {})
+
     @property
     def workspace_root(self) -> Path:
         """Get the workspace root directory"""

app/tool/__init__.py

@@ -7,6 +7,7 @@ from app.tool.str_replace_editor import StrReplaceEditor
 from app.tool.terminate import Terminate
 from app.tool.tool_collection import ToolCollection
 from app.tool.web_search import WebSearch
+from app.tool.linux_sandbox import LinuxSandboxTool
 
 
 __all__ = [
@@ -19,4 +20,5 @@ __all__ = [
     "ToolCollection",
     "CreateChatCompletion",
     "PlanningTool",
+    "LinuxSandboxTool",
 ]

app/tool/linux_sandbox.py

@@ -0,0 +1,36 @@
+from typing import Optional
+from app.tool.base import BaseTool, ToolResult
+from app.config import config
+
+class LinuxSandboxTool(BaseTool):
+    """A tool to start an e2b DesktopSandbox and return the VNC URL, using config.cloud_sandbox for credentials."""
+
+    name: str = "linux_sandbox"
+    description: str = (
+        "Starts an e2b DesktopSandbox and returns the VNC URL. "
+        "API key and domain are read from config.cloud_sandbox."
+    )
+    parameters: dict = {"type": "object", "properties": {}, "required": []}
+
+    async def execute(self, **kwargs) -> ToolResult:
+        """
+        Start an e2b DesktopSandbox and return the VNC URL, using config.cloud_sandbox.
+        """
+        try:
+            from e2b_desktop import Sandbox as DesktopSandbox
+        except ImportError:
+            return ToolResult(error="e2b_desktop is not installed. Please install it via pip.")
+        sandbox_config = getattr(config, "cloud_sandbox", None)
+        if not sandbox_config or not sandbox_config.get("api_key") or not sandbox_config.get("domain"):
+            return ToolResult(error="cloud_sandbox config missing or incomplete. Please set api_key and domain in config.cloud_sandbox.")
+        try:
+            desktop = DesktopSandbox(api_key=sandbox_config["api_key"], domain=sandbox_config["domain"])
+            desktop.stream.start()
+            url = desktop.stream.get_url()
+            return ToolResult(output={
+                "status": "started",
+                "vnc_url": url,
+                "domain": sandbox_config["domain"],
+            })
+        except Exception as e:
+            return ToolResult(error=f"Failed to start e2b DesktopSandbox: {e}")

app/tool/sandbox_python_execute.py

@@ -0,0 +1,118 @@
+from dotenv import load_dotenv
+load_dotenv()
+from typing import Dict
+from app.tool.base import BaseTool
+import asyncio
+import multiprocessing
+import sys
+from io import StringIO
+from app.config import config
+
+class SandboxPythonExecute(BaseTool):
+    """A tool for executing Python code either locally (with timeout) or in a sandboxed environment using e2b_code_interpreter."""
+
+    name: str = "python_execute"
+    description: str = (
+        "Executes Python code string. Note: Only print outputs are visible, function return values are not captured. Use print statements to see results. "
+        "Set mode='sandbox' to run in a secure sandbox (e2b), otherwise runs locally."
+    )
+    parameters: dict = {
+        "type": "object",
+        "properties": {
+            "code": {
+                "type": "string",
+                "description": "The Python code to execute.",
+            },
+            "mode": {
+                "type": "string",
+                "enum": ["local", "sandbox"],
+                "description": "Execution mode: 'local' (default) or 'sandbox' (e2b sandbox)",
+            },
+            "timeout": {
+                "type": "integer",
+                "description": "Execution timeout in seconds (default: 5 for local, 10 for sandbox)",
+            },
+        },
+        "required": ["code"],
+    }
+
+    def _run_code(self, code: str, result_dict: dict, safe_globals: dict) -> None:
+        original_stdout = sys.stdout
+        try:
+            output_buffer = StringIO()
+            sys.stdout = output_buffer
+            exec(code, safe_globals, safe_globals)
+            result_dict["observation"] = output_buffer.getvalue()
+            result_dict["success"] = True
+        except Exception as e:
+            result_dict["observation"] = str(e)
+            result_dict["success"] = False
+        finally:
+            sys.stdout = original_stdout
+
+    async def execute(
+        self,
+        code: str,
+        timeout: int = None,
+        mode: str = "local",
+    ) -> Dict:
+        """
+        Executes the provided Python code in the selected environment.
+        Args:
+            code (str): The Python code to execute.
+            timeout (int): Execution timeout in seconds.
+            mode (str): 'local' or 'sandbox'.
+        Returns:
+            Dict: Contains 'observation' with execution output or error message and 'success' status.
+        """
+        if mode == "sandbox":
+            # Use e2b_code_interpreter Sandbox
+            try:
+                from e2b_code_interpreter import Sandbox
+            except ImportError:
+                return {"observation": "e2b_code_interpreter not installed.", "success": False}
+            # Get sandbox config from app.config.config
+            sandbox_config = getattr(config, "cloud_sandbox", {})
+            # You can pass config to Sandbox() if needed, e.g. Sandbox(api_key=sandbox_config.get("api_key"))
+            sbx = Sandbox(**sandbox_config) if sandbox_config else Sandbox()
+            try:
+                # Default timeout for sandbox is 10s if not set
+                effective_timeout = timeout if timeout is not None else 10
+                async def run():
+                    execution = sbx.run_code(code)
+                    logs = execution.logs if hasattr(execution, 'logs') else str(execution)
+                    success = execution.error is None if hasattr(execution, 'error') else True
+                    observation = logs
+                    if not success:
+                        observation = f"Error: {getattr(execution, 'error', 'Unknown error')}\n{logs}"
+                    return {"observation": observation, "success": success}
+                result = await asyncio.wait_for(run(), timeout=effective_timeout)
+                return result
+            except asyncio.TimeoutError:
+                return {"observation": f"Execution timeout after {effective_timeout} seconds", "success": False}
+            except Exception as e:
+                return {"observation": str(e), "success": False}
+            finally:
+                sbx.kill()
+        else:
+            # Local execution (default)
+            effective_timeout = timeout if timeout is not None else 5
+            with multiprocessing.Manager() as manager:
+                result = manager.dict({"observation": "", "success": False})
+                if isinstance(__builtins__, dict):
+                    safe_globals = {"__builtins__": __builtins__}
+                else:
+                    safe_globals = {"__builtins__": __builtins__.__dict__.copy()}
+                proc = multiprocessing.Process(
+                    target=self._run_code, args=(code, result, safe_globals)
+                )
+                proc.start()
+                proc.join(effective_timeout)
+                if proc.is_alive():
+                    proc.terminate()
+                    proc.join(1)
+                    return {
+                        "observation": f"Execution timeout after {effective_timeout} seconds",
+                        "success": False,
+                    }
+                return dict(result)

config/config.example.toml

@@ -95,6 +95,11 @@ temperature = 0.0                          # Controls randomness for vision mode
 #timeout = 300
 #network_enabled = true
 
+# [cloud_sandbox]
+# e2b_api_key = "YOUR_E2B_API_KEY"
+# domain = "YOUR_E2B_DOMAIN"
+
+
 # MCP (Model Context Protocol) configuration
 [mcp]
 server_reference = "app.mcp.server" # default server module reference
@@ -103,3 +108,5 @@ server_reference = "app.mcp.server" # default server module reference
 # Your can add additional agents into run-flow workflow to solve different-type tasks.
 [runflow]
 use_data_analysis_agent = false     # The Data Analysi Agent to solve various data analysis tasks
+
+