feat: integrate OpenSandbox for isolated code execution

Replace direct exec()/subprocess calls with OpenSandbox's containerized
execution when sandbox is enabled in config. All 4 execution paths
(RunCode, ExecuteNbCode, Terminal, shell_execute) now route through
SandboxExecutor which wraps the opensandbox + code-interpreter SDKs.
Falls back to local execution when sandbox is disabled or on failure.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: 安辰

config/config2.yaml

@@ -5,4 +5,19 @@ llm:
   api_type: "openai"  # or azure / ollama / groq etc.
   model: "gpt-4-turbo"  # or gpt-3.5-turbo
   base_url: "https://api.openai.com/v1"  # or forward url / other llm url
-  api_key: "YOUR_API_KEY"
+  api_key: "YOUR_API_KEY"
+
+## Sandbox Configuration (Optional)
+## Enables isolated code execution via OpenSandbox.
+## Requires: pip install opensandbox opensandbox-code-interpreter
+# sandbox:
+#   enabled: true
+#   domain: "localhost:8080"
+#   api_key: "YOUR_SANDBOX_API_KEY"
+#   image: "opensandbox/code-interpreter:v1.0.2"
+#   timeout: 600
+#   resource:
+#     cpu: "1"
+#     memory: "2Gi"
+#   env:
+#     PYTHON_VERSION: "3.11"

metagpt/actions/di/execute_nb_code.py

@@ -26,6 +26,7 @@
 
 from metagpt.actions import Action
 from metagpt.logs import logger
+from metagpt.tools.libs.sandbox_executor import get_sandbox_executor
 from metagpt.utils.report import NotebookReporter
 
 INSTALL_KEEPLEN = 500
@@ -243,6 +244,30 @@ async def run_cell(self, cell: NotebookNode, cell_index: int) -> Tuple[bool, str
         except Exception:
             return self.parse_outputs(self.nb.cells[-1].outputs)
 
+    def _sandbox_enabled(self) -> bool:
+        """Check if sandbox execution is enabled in config."""
+        try:
+            return self.config.sandbox.enabled
+        except (AttributeError, Exception):
+            return False
+
+    async def _run_code_in_sandbox(self, code: str) -> Tuple[bool, str]:
+        """Execute code in the OpenSandbox via CodeInterpreter."""
+        executor = await get_sandbox_executor(self.config.sandbox)
+        stdout, stderr = await executor.run_code(code, language="python")
+
+        if stderr:
+            output = stderr
+            output = remove_escape_and_color_codes(output)
+            output = output[-5000:]
+            return False, output
+        else:
+            output = stdout
+            output = remove_escape_and_color_codes(output)
+            output = remove_log_and_warning_lines(output)
+            output = output[:5000]
+            return True, output
+
     async def run(self, code: str, language: Literal["python", "markdown"] = "python") -> Tuple[str, bool]:
         """
         return the output of code execution, and a success indicator (bool) of code execution.
@@ -254,12 +279,17 @@ async def run(self, code: str, language: Literal["python", "markdown"] = "python
                 # add code to the notebook
                 self.add_code_cell(code=code)
 
-                # build code executor
-                await self.build()
+                if self._sandbox_enabled():
+                    success, outputs = await self._run_code_in_sandbox(code)
+                    # Store output in notebook cell for consistency
+                    self.add_output_to_cell(self.nb.cells[-1], outputs)
+                else:
+                    # build code executor
+                    await self.build()
 
-                # run code
-                cell_index = len(self.nb.cells) - 1
-                success, outputs = await self.run_cell(self.nb.cells[-1], cell_index)
+                    # run code
+                    cell_index = len(self.nb.cells) - 1
+                    success, outputs = await self.run_cell(self.nb.cells[-1], cell_index)
 
                 if "!pip" in code:
                     success = False

metagpt/actions/run_code.py

@@ -24,6 +24,7 @@
 from metagpt.actions.action import Action
 from metagpt.logs import logger
 from metagpt.schema import RunCodeContext, RunCodeResult
+from metagpt.tools.libs.sandbox_executor import get_sandbox_executor
 from metagpt.utils.exceptions import handle_exception
 
 PROMPT_TEMPLATE = """
@@ -79,8 +80,21 @@ class RunCode(Action):
     name: str = "RunCode"
     i_context: RunCodeContext = Field(default_factory=RunCodeContext)
 
+    def _sandbox_enabled(self) -> bool:
+        """Check if sandbox execution is enabled in config."""
+        try:
+            return self.context.config.sandbox.enabled
+        except (AttributeError, Exception):
+            return False
+
     @classmethod
-    async def run_text(cls, code) -> Tuple[str, str]:
+    async def run_text(cls, code, sandbox_config=None) -> Tuple[str, str]:
+        if sandbox_config and sandbox_config.enabled:
+            try:
+                executor = await get_sandbox_executor(sandbox_config)
+                return await executor.run_code(code, language="python")
+            except Exception as e:
+                logger.warning(f"Sandbox execution failed, falling back to local: {e}")
         try:
             # We will document_store the result in this dictionary
             namespace = {}
@@ -93,6 +107,9 @@ async def run_script(self, working_directory, additional_python_paths=[], comman
         working_directory = str(working_directory)
         additional_python_paths = [str(path) for path in additional_python_paths]
 
+        if self._sandbox_enabled():
+            return await self._run_script_in_sandbox(working_directory, additional_python_paths, command)
+
         # Copy the current environment variables
         env = self.context.new_environ()
 
@@ -117,6 +134,32 @@ async def run_script(self, working_directory, additional_python_paths=[], comman
             stdout, stderr = process.communicate()
         return stdout.decode("utf-8"), stderr.decode("utf-8")
 
+    async def _run_script_in_sandbox(self, working_directory, additional_python_paths, command) -> Tuple[str, str]:
+        """Execute a script inside the OpenSandbox."""
+        executor = await get_sandbox_executor(self.context.config.sandbox)
+        sandbox_workdir = "/workspace"
+
+        # Upload source files and requirements to sandbox
+        work_path = Path(working_directory)
+        if work_path.exists():
+            for f in work_path.iterdir():
+                if f.is_file() and f.suffix in (".py", ".txt", ".json", ".yaml", ".yml"):
+                    await executor.upload_file(str(f), f"{sandbox_workdir}/{f.name}")
+
+        # Install dependencies inside sandbox
+        req_file = work_path / "requirements.txt"
+        if req_file.exists() and req_file.stat().st_size > 0:
+            await executor.run_command(
+                f"pip install -r {sandbox_workdir}/requirements.txt", working_directory=sandbox_workdir
+            )
+        await executor.run_command("pip install pytest", working_directory=sandbox_workdir)
+
+        # Build and run the command inside sandbox
+        cmd_str = " ".join(command)
+        logger.info(f"Running in sandbox: {cmd_str}")
+        stdout, stderr, _ = await executor.run_command(cmd_str, working_directory=sandbox_workdir)
+        return stdout, stderr
+
     async def run(self, *args, **kwargs) -> RunCodeResult:
         logger.info(f"Running {' '.join(self.i_context.command)}")
         if self.i_context.mode == "script":
@@ -126,7 +169,8 @@ async def run(self, *args, **kwargs) -> RunCodeResult:
                 additional_python_paths=self.i_context.additional_python_paths,
             )
         elif self.i_context.mode == "text":
-            outs, errs = await self.run_text(code=self.i_context.code)
+            sandbox_config = self.context.config.sandbox if self._sandbox_enabled() else None
+            outs, errs = await self.run_text(code=self.i_context.code, sandbox_config=sandbox_config)
 
         logger.info(f"{outs=}")
         logger.info(f"{errs=}")

metagpt/config2.py

@@ -21,6 +21,7 @@
 from metagpt.configs.role_custom_config import RoleCustomConfig
 from metagpt.configs.role_zero_config import RoleZeroConfig
 from metagpt.configs.s3_config import S3Config
+from metagpt.configs.sandbox_config import SandboxConfig
 from metagpt.configs.search_config import SearchConfig
 from metagpt.configs.workspace_config import WorkspaceConfig
 from metagpt.const import CONFIG_ROOT, METAGPT_ROOT
@@ -67,6 +68,9 @@ class Config(CLIParams, YamlModel):
     browser: BrowserConfig = BrowserConfig()
     mermaid: MermaidConfig = MermaidConfig()
 
+    # Sandbox Parameters
+    sandbox: SandboxConfig = Field(default_factory=SandboxConfig)
+
     # Storage Parameters
     s3: Optional[S3Config] = None
     redis: Optional[RedisConfig] = None

metagpt/configs/sandbox_config.py

@@ -0,0 +1,23 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+@Time    : 2024/3/18
+@File    : sandbox_config.py
+"""
+from typing import Dict, List, Optional
+
+from metagpt.utils.yaml_model import YamlModel
+
+
+class SandboxConfig(YamlModel):
+    """Config for OpenSandbox integration."""
+
+    enabled: bool = False
+    domain: str = ""
+    api_key: str = ""
+    image: str = "opensandbox/code-interpreter:v1.0.2"
+    entrypoint: List[str] = ["/opt/opensandbox/code-interpreter.sh"]
+    timeout: int = 600
+    resource: Dict[str, str] = {"cpu": "1", "memory": "2Gi"}
+    env: Dict[str, str] = {}
+    network_policy: Optional[Dict] = None