feat(plugin): Support sandbox for DB-GPT (#2901)

Co-authored-by: Fangyin Cheng <staneyffer@gmail.com>

packages/dbgpt-sandbox/README.md

@@ -0,0 +1,18 @@
+# DB-GPT Sandbox
+
+背景：AI Agent 逐步成为使用 AI 解决真实环境中各类问题的有力工具，然而真实环境的任务隔离性和安全性是企业落地中必然要考虑的问题。 DB-GPT Agent 目前不支持统一、可扩展的安全沙箱环境。 
+
+#### 预期目标：
+
+ 为 DB-GPT Agent 实现一个安全的沙箱执行环境（支持 Agent、工具的运行和多语言代码的执行）。 分三个部分：
+
+1. 基于 DB-GPT Agent + Docker 容器实现安全的代码执行环境，支持 Python、Shell、Node.js 等代码的执行，改造 DB-GPT 现有的代码执行智能体。 
+2. 支持有状态的沙箱环境，多次代码执行可以在相同的环境中，并且上次环境的变更能影响下次的执行（例如第一次执行安装 pypi 依赖，第二次执行安装后的依赖能正常使用） 
+3. 插件化的安全沙箱环境实现，设计统一的沙箱环境接口，支持 Docker、Podman、本地进程（基Cgroup/Namespace/WebAssembly等）等沙箱环境的实现。
+
+#### 产出要求：
+
+1. 项目设计文档（含架构图、原理图、实现细节等）
+2. 实现安全沙箱环境的核心模块（统一沙箱环境接口，Docker 实现和本地进程的实现）
+3. 提供完整的使用教程文档说明
+4. 基于沙箱环境，开发一个支持 Python 等代码执行的 Agent 案例

packages/dbgpt-sandbox/pyproject.toml

@@ -0,0 +1,57 @@
+[project]
+name = "dbgpt-sandbox"
+version = "0.7.3"
+description = "A secure sandbox execution environment for DB-GPT Agent"
+authors = [
+    { name = "csunny", email = "cfqcsunny@gmail.com" }
+]
+license = "MIT"
+readme = "README.md"
+requires-python = ">= 3.10"
+dependencies = [
+    "psutil>=5.9.0",
+    "colorama>=0.4.4",
+    "docker>=6.0.0",
+    "fastapi>=0.68.0",
+    "uvicorn>=0.15.0",
+    "pydantic>=1.8.0",
+    "python-multipart>=0.0.5",
+    "selenium>=4.0.0",
+    "typing-extensions>=4.0.0",
+]
+
+[project.urls]
+Homepage = "https://github.com/eosphoros-ai/DB-GPT"
+Documentation = "http://docs.dbgpt.cn/docs/overview"
+Repository = "https://github.com/eosphoros-ai/DB-GPT.git"
+Issues = "https://github.com/eosphoros-ai/DB-GPT/issues"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=6.0",
+    "pytest-asyncio>=0.18.0",
+    "black>=22.0",
+    "flake8>=4.0",
+]
+
+[tool.uv]
+managed = true
+dev-dependencies = []
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/dbgpt_sandbox"]
+exclude = [
+    "src/dbgpt_sandbox/**/tests",
+    "src/dbgpt_sandbox/**/tests/*",
+    "src/dbgpt_sandbox/tests",
+    "src/dbgpt_sandbox/tests/*",
+    "src/dbgpt_sandbox/**/examples",
+    "src/dbgpt_sandbox/**/examples/*"
+]
+
+[project.scripts]
+dbgpt-sandbox = "dbgpt_sandbox.sandbox.main:main"

packages/dbgpt-sandbox/src/__init__.py

@@ -0,0 +1,3 @@
+"""
+DB-GPT Core Package
+"""

packages/dbgpt-sandbox/src/dbgpt_sandbox/__init__.py

@@ -0,0 +1,5 @@
+"""
+DB-GPT Sandbox Package
+"""
+
+__version__ = "0.7.3"

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/config.py

@@ -0,0 +1,40 @@
+import os
+
+LANGUAGE_IMAGES = {
+    "python": "python:3.11-slim",
+    "python-vnc": "vnc-gui-browser:latest",
+    "javascript": "node:18-slim",
+    "java": "openjdk:11-jre-slim",
+    "cpp": "gcc:latest",
+    "go": "golang:1.21-alpine",
+    "rust": "rust:1.75-slim",
+}
+
+WORKING_DIR = "/workspace"
+
+
+def get_command_by_language(language: str, filename: str) -> str:
+    commands = {
+        "python-vnc": f"python3 {filename}",
+        "python": f"python {filename}",
+        "javascript": f"node {filename}",
+        "java": f"javac {filename} && java {filename[:-5]}",
+        "cpp": f"g++ -o program {filename} && ./program",
+        "go": f"go run {filename}",
+        "rust": f"rustc {filename} -o program && ./program",
+    }
+    return commands.get(language, f"cat {filename}")
+
+
+MAX_MEMORY = 256 * 1024 * 1024  # 256MB
+MAX_CPU_PERCENT = 50.0
+MAX_EXECUTION_TIME = 30  # seconds
+MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB
+MAX_DEPENDENCY_INSTALL_TIME = 300  # seconds
+MAX_DEPENDENCY_INSTALL_SIZE = 200 * 1024 * 1024  # 200MB
+MAX_PROCESSES = 10
+
+
+SANDBOX_RUNTIME = os.getenv(
+    "SANDBOX_RUNTIME", "local"
+)  # Optional values: docker, podman, nerdctl, local

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/control_layer/control_layer.py

@@ -0,0 +1,182 @@
+"""
+控制层：管理任务生命周期和执行，支持 TaskObject 的多类型任务
+"""
+
+import asyncio
+import uuid
+from typing import Any, Dict
+
+from ..config import WORKING_DIR
+from ..execution_layer.base import ExecutionResult, ExecutionStatus, SessionConfig
+from ..execution_layer.runtime_factory import RuntimeFactory
+from ..user_layer.schemas import TASK_TYPES, TaskObject
+
+
+class ControlLayer:
+    """控制层：管理任务生命周期和执行"""
+
+    def __init__(self):
+        self.runtime = RuntimeFactory.create()
+        self.tasks: Dict[str, Dict[str, Any]] = {}
+        self.task_locks: Dict[str, asyncio.Lock] = {}
+
+    async def handle_task(self, task: TaskObject) -> ExecutionResult:
+        """根据 TaskObject 的 task_type 分发处理"""
+        if task.task_type not in TASK_TYPES:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error=f"未知任务类型: {task.task_type}"
+            )
+
+        handler_map = {
+            "connect": self._handle_connect,
+            "configure": self._handle_configure,
+            "execute": self._handle_execute,
+            "manual": self._handle_manual,
+            "disconnect": self._handle_disconnect,
+            "status": self._handle_status,
+            "list": self._handle_list,
+            "get_file": self._handle_get_file,
+        }
+
+        handler = handler_map[task.task_type]
+
+        lock = self.task_locks.setdefault(task.task_id, asyncio.Lock())
+        async with lock:
+            return await handler(task)
+
+    async def _handle_connect(self, task: TaskObject) -> ExecutionResult:
+        """创建新的沙箱会话"""
+        session_id = task.session_id or str(uuid.uuid4())
+        config = SessionConfig(
+            language=task.language,
+            working_dir=WORKING_DIR,
+            max_memory=512 * 1024 * 1024,  # 512MB in bytes
+            max_cpus=task.config.get("max_cpus", 1),
+            environment_vars=task.config.get("env", {}),
+            network_disabled=task.config.get("network_disabled", False),
+        )
+
+        try:
+            session = await self.runtime.create_session(session_id, config)
+            self.tasks[task.task_id] = {
+                "task": task,
+                "session_id": session.session_id,
+                "status": "connected",
+            }
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS,
+                output=f"session {session.session_id} connected",
+            )
+        except Exception as e:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error=f"连接失败: {e}")
+
+    async def _handle_configure(self, task: TaskObject) -> ExecutionResult:
+        """配置沙箱环境，例如安装依赖"""
+        if task.task_id not in self.tasks:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="任务不存在")
+
+        session_id = self.tasks[task.task_id]["session_id"]
+        session = await self.runtime.get_session(session_id)
+        if not session:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="会话不存在")
+
+        deps = task.config.get("dependencies", [])
+        try:
+            if not deps:
+                self.tasks[task.task_id]["status"] = "configured"
+                return ExecutionResult(
+                    status=ExecutionStatus.SUCCESS, output="无依赖需要安装"
+                )
+            result = await session.install_dependencies(deps)
+            self.tasks[task.task_id]["status"] = (
+                "configured" if result.status == ExecutionStatus.SUCCESS else "failed"
+            )
+            return result
+        except Exception as e:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error=f"配置失败: {e}")
+
+    async def _handle_execute(self, task: TaskObject) -> ExecutionResult:
+        """在沙箱中执行代码"""
+        if task.task_id not in self.tasks:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="任务不存在")
+
+        session_id = self.tasks[task.task_id]["session_id"]
+        session = await self.runtime.get_session(session_id)
+        if not session:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="会话不存在")
+
+        try:
+            if task.language == "shell":
+                result = await session.execute(task.code_content or "", shell=True)
+            else:
+                result = await session.execute(task.code_content or "")
+            self.tasks[task.task_id]["status"] = (
+                "finished" if result.status == ExecutionStatus.SUCCESS else "failed"
+            )
+            self.tasks[task.task_id]["result"] = result
+            return result
+        except Exception as e:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error=f"执行失败: {e}")
+
+    async def _handle_manual(self, task: TaskObject) -> ExecutionResult:
+        """进入手动操作模式（返回可连接的 URL 或 token）"""
+        if task.task_id not in self.tasks:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="任务不存在")
+        session_id = self.tasks[task.task_id]["session_id"]
+        manual_url = f"http://sandbox-gui/{session_id}"
+        self.tasks[task.task_id]["status"] = "manual"
+        return ExecutionResult(status=ExecutionStatus.SUCCESS, output=manual_url)
+
+    async def _handle_disconnect(self, task: TaskObject) -> ExecutionResult:
+        """停止并销毁沙箱会话"""
+        if task.task_id not in self.tasks:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="任务不存在")
+
+        session_id = self.tasks[task.task_id]["session_id"]
+        success = await self.runtime.destroy_session(session_id)
+        self.tasks[task.task_id]["status"] = "stopped" if success else "error"
+        return ExecutionResult(
+            status=ExecutionStatus.SUCCESS if success else ExecutionStatus.ERROR,
+            output="会话已销毁" if success else "会话销毁失败",
+        )
+
+    async def _handle_status(self, task: TaskObject) -> ExecutionResult:
+        """获取任务/会话状态"""
+        if task.task_id not in self.tasks:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="任务不存在")
+
+        session_id = self.tasks[task.task_id]["session_id"]
+        session = await self.runtime.get_session(session_id)
+        if not session:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="会话不存在")
+
+        status = await session.get_status()
+        return ExecutionResult(status=ExecutionStatus.SUCCESS, output=str(status))
+
+    async def _handle_list(self, task: TaskObject) -> ExecutionResult:
+        """列出所有活跃会话"""
+        sessions = await self.runtime.list_sessions()
+        return ExecutionResult(status=ExecutionStatus.SUCCESS, output=str(sessions))
+
+    async def _handle_get_file(self, task: TaskObject) -> ExecutionResult:
+        """获取沙箱内指定文件内容"""
+        if task.task_id not in self.tasks:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="任务不存在")
+
+        session_id = self.tasks[task.task_id]["session_id"]
+        session = await self.runtime.get_session(session_id)
+        if not session:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="会话不存在")
+
+        filename = task.file_name
+
+        if not filename:
+            return ExecutionResult(status=ExecutionStatus.ERROR, error="未指定文件名")
+
+        try:
+            content = await session.get_file_content(filename)
+            return ExecutionResult(status=ExecutionStatus.SUCCESS, output=content)
+        except Exception as e:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error=f"获取文件失败: {e}"
+            )

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/display_layer/display_layer.py

@@ -0,0 +1,39 @@
+"""
+显示层，封装并管理 Docker 执行结果
+"""
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class DisplayResult:
+    status: str  # "success" / "error"
+    output: str
+    error: str
+    execution_time: float
+    exit_code: int
+    files: List[str] = field(default_factory=list)
+    logs: List[str] = field(default_factory=list)
+    gui_frame: Optional[Any] = None  # 可选 GUI 内容
+    gui_url: Optional[str] = None
+    screenshots: List[str] = field(default_factory=list)
+
+
+class DisplayLayer:
+    """显示层，封装并管理 Docker 执行结果"""
+
+    def __init__(self):
+        self.history: Dict[str, DisplayResult] = {}  # session_id -> last result
+
+    def add_result(self, session_id: str, result: DisplayResult):
+        """保存执行结果"""
+        self.history[session_id] = result
+
+    def get_result(self, session_id: str) -> Optional[DisplayResult]:
+        """获取某会话最新执行结果"""
+        return self.history.get(session_id)
+
+    def list_history(self) -> Dict[str, DisplayResult]:
+        """返回所有会话的最新执行结果"""
+        return self.history

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/execution_layer/__init__.py

@@ -0,0 +1,27 @@
+"""
+DB-GPT Sandbox Agent - 核心沙箱模块
+
+提供统一的代码执行沙箱接口，支持 Docker、本地运行时与 Podman。
+"""
+
+from .base import ExecutionResult, SandboxRuntime, SandboxSession
+from .docker_runtime import DockerRuntime
+from .local_runtime import LocalRuntime
+from .nerdctl_runtime import NerdctlRuntime
+from .podman_runtime import PodmanRuntime
+from .utils import EnvironmentDetector, ResourceLimits
+
+__version__ = "0.1.0"
+__author__ = "DB-GPT Team"
+
+__all__ = [
+    "SandboxRuntime",
+    "SandboxSession",
+    "ExecutionResult",
+    "DockerRuntime",
+    "LocalRuntime",
+    "PodmanRuntime",
+    "NerdctlRuntime",
+    "ResourceLimits",
+    "EnvironmentDetector",
+]

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/execution_layer/base.py

@@ -0,0 +1,158 @@
+"""
+沙箱运行时基础抽象类
+
+定义了统一的沙箱接口，包括会话管理、代码执行等核心功能。
+"""
+
+import time
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+
+class ExecutionStatus(Enum):
+    """执行状态枚举"""
+
+    SUCCESS = "success"
+    ERROR = "error"
+    TIMEOUT = "timeout"
+    RESOURCE_LIMIT = "resource_limit"
+
+
+@dataclass
+class ExecutionResult:
+    """代码执行结果"""
+
+    status: ExecutionStatus
+    output: str = ""
+    error: str = ""
+    execution_time: float = 0.0
+    memory_usage: int = 0  # bytes
+    exit_code: int = 0
+
+    def to_dict(self) -> Dict[str, Any]:
+        """转换为字典格式"""
+        return {
+            "status": self.status.value,
+            "output": self.output,
+            "error": self.error,
+            "execution_time": self.execution_time,
+            "memory_usage": self.memory_usage,
+            "exit_code": self.exit_code,
+        }
+
+
+@dataclass
+class SessionConfig:
+    """会话配置"""
+
+    language: str = "python"
+    timeout: int = 30  # seconds
+    max_memory: int = 256 * 1024 * 1024  # 256MB
+    # max_cpu_percent: float = 50.0
+    max_cpus: int = 1
+    working_dir: str = "/workspace"
+    environment_vars: Dict[str, str] = None
+    network_disabled: bool = False  # 是否禁用网络
+
+    def __post_init__(self):
+        if self.environment_vars is None:
+            self.environment_vars = {}
+
+
+class SandboxSession(ABC):
+    """沙箱会话抽象类"""
+
+    def __init__(self, session_id: str, config: SessionConfig):
+        self.session_id = session_id
+        self.config = config
+        self.created_at = time.time()
+        self.last_accessed = time.time()
+        self._is_active = False
+
+    @property
+    def is_active(self) -> bool:
+        """检查会话是否活跃"""
+        return self._is_active
+
+    @abstractmethod
+    async def start(self) -> bool:
+        """启动会话"""
+        pass
+
+    @abstractmethod
+    async def stop(self) -> bool:
+        """停止会话"""
+        pass
+
+    @abstractmethod
+    async def execute(self, code: str) -> ExecutionResult:
+        """执行代码"""
+        pass
+
+    @abstractmethod
+    async def get_status(self) -> Dict[str, Any]:
+        """获取会话状态"""
+        pass
+
+    async def install_dependencies(self, dependencies: List[str]) -> ExecutionResult:
+        """安装依赖（可选）。默认未实现，由具体运行时覆盖。
+        返回 ExecutionResult，status 为 ERROR 表示未实现或失败。
+        """
+        if not dependencies:
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS, output="无依赖需要安装", exit_code=0
+            )
+        return ExecutionResult(
+            status=ExecutionStatus.ERROR, error="依赖安装未实现", exit_code=1
+        )
+
+    def update_last_accessed(self):
+        """更新最后访问时间"""
+        self.last_accessed = time.time()
+
+
+class SandboxRuntime(ABC):
+    """沙箱运行时抽象类"""
+
+    def __init__(self, runtime_id: str):
+        self.runtime_id = runtime_id
+        self.sessions: Dict[str, SandboxSession] = {}
+
+    @abstractmethod
+    async def create_session(
+        self, session_id: str, config: SessionConfig
+    ) -> SandboxSession:
+        """创建新的沙箱会话"""
+        pass
+
+    @abstractmethod
+    async def destroy_session(self, session_id: str) -> bool:
+        """销毁沙箱会话"""
+        pass
+
+    @abstractmethod
+    async def list_sessions(self) -> List[str]:
+        """列出所有活跃会话"""
+        pass
+
+    @abstractmethod
+    async def get_session(self, session_id: str) -> Optional[SandboxSession]:
+        """获取指定会话"""
+        pass
+
+    @abstractmethod
+    async def cleanup_expired_sessions(self, max_idle_time: int = 3600) -> int:
+        """清理过期会话，返回清理的会话数量"""
+        pass
+
+    @abstractmethod
+    async def health_check(self) -> Dict[str, Any]:
+        """健康检查"""
+        pass
+
+    @abstractmethod
+    def supports_language(self, language: str) -> bool:
+        """检查是否支持指定编程语言"""
+        pass

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/execution_layer/docker_runtime.py

@@ -0,0 +1,459 @@
+"""
+基于 Docker 容器的代码执行环境，支持多语言和状态保持，所有 Docker 操作均异步化。
+"""
+
+import asyncio
+import base64
+import io
+import os
+import tarfile
+import tempfile
+import time
+from typing import Any, Dict, List, Optional
+
+try:
+    import docker
+except ImportError:
+    docker = None
+
+from ..config import LANGUAGE_IMAGES, get_command_by_language
+from ..display_layer.display_layer import DisplayResult
+from ..utils_function.logger import print_log
+from .base import (
+    ExecutionResult,
+    ExecutionStatus,
+    SandboxRuntime,
+    SandboxSession,
+    SessionConfig,
+)
+
+
+class DockerSandboxSession(SandboxSession):
+    """异步 Docker 沙箱会话实现"""
+
+    def __init__(self, session_id: str, config: SessionConfig, docker_client):
+        super().__init__(session_id, config)
+        self.docker_client = docker_client
+        self.container = None
+        self.image_name = self._get_image_name(config.language)
+
+    def _get_image_name(self, language: str) -> str:
+        return LANGUAGE_IMAGES.get(language, "python:3.11-slim")
+
+    async def start(self) -> bool:
+        """启动 Docker 容器"""
+        try:
+            container_config = {
+                "image": self.image_name,
+                "command": "tail -f /dev/null",
+                "detach": True,
+                "mem_limit": self.config.max_memory,
+                "cpuset_cpus": str(self.config.max_cpus),
+                "working_dir": self.config.working_dir,
+                "environment": self.config.environment_vars,
+                "network_disabled": self.config.network_disabled,
+                "volumes": {tempfile.gettempdir(): {"bind": "/tmp", "mode": "rw"}},
+                "name": f"sandbox_{self.session_id}",
+            }
+
+            if self.config.language.endswith("-vnc"):
+                container_config["ports"] = {"5900/tcp": None, "6080/tcp": None}
+                container_config["command"] = "/startup.sh"
+                print_log("INFO", f"使用 VNC/noVNC 容器: {self.image_name}")
+
+            self.container = await asyncio.to_thread(
+                self.docker_client.containers.run, **container_config
+            )
+            self._is_active = True
+
+            # ✅ 确认 startup.sh 存在并有执行权限
+            check = self.container.exec_run("ls -l /startup.sh")
+            print_log("DEBUG", f"startup.sh 状态: {check.output}")
+
+            # ✅ 查看容器启动日志
+            logs = self.container.logs(stdout=True, stderr=True, tail=50)
+            print_log("DEBUG", f"容器日志: {logs.decode('utf-8', errors='ignore')}")
+
+            await self._setup_environment()
+            return True
+        except Exception as e:
+            print(f"启动 Docker 容器失败: {e}")
+            return False
+
+    async def _setup_environment(self):
+        """设置执行环境"""
+        if not self.container:
+            return
+
+        await asyncio.to_thread(
+            self.container.exec_run, f"mkdir -p {self.config.working_dir}"
+        )
+
+        if self.config.language.startswith("python"):
+            await asyncio.to_thread(
+                self.container.exec_run,
+                "pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple",
+            )
+        elif self.config.language.startswith("javascript"):
+            await asyncio.to_thread(
+                self.container.exec_run, "npm init -y", workdir=self.config.working_dir
+            )
+
+    async def stop(self) -> bool:
+        """停止并删除容器"""
+        try:
+            if self.container:
+                await asyncio.to_thread(self.container.stop)
+                await asyncio.to_thread(self.container.remove)
+                self.container = None
+            self._is_active = False
+            return True
+        except Exception as e:
+            print(f"停止 Docker 容器失败: {e}")
+            return False
+
+    async def install_dependencies(self, dependencies: List[str]) -> ExecutionResult:
+        """在容器内安装依赖，支持 python/npm。"""
+        if not self.container or not self._is_active:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error="容器未启动或已停止", exit_code=-1
+            )
+        if not dependencies:
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS, output="无依赖需要安装", exit_code=0
+            )
+
+        try:
+            outputs = []
+            errors = []
+            exit_code = 0
+
+            if self.config.language.startswith("python"):
+                for dep in dependencies:
+                    res = await asyncio.to_thread(
+                        self.container.exec_run,
+                        f"pip install --no-input --disable-pip-version-check {dep}",
+                    )
+                    exit_code = res.exit_code
+                    if res.exit_code != 0:
+                        errors.append(f"pip install {dep} 失败")
+                        break
+                    else:
+                        outputs.append(f"installed: {dep}")
+            elif self.config.language.startswith("javascript"):
+                await asyncio.to_thread(
+                    self.container.exec_run,
+                    "npm init -y",
+                    workdir=self.config.working_dir,
+                )
+                dep_str = " ".join(dependencies)
+                res = await asyncio.to_thread(
+                    self.container.exec_run,
+                    f"npm install {dep_str}",
+                    workdir=self.config.working_dir,
+                )
+                exit_code = res.exit_code
+                if res.exit_code != 0:
+                    errors.append("npm install 失败")
+                else:
+                    outputs.append(f"installed: {dep_str}")
+            else:
+                return ExecutionResult(
+                    status=ExecutionStatus.ERROR,
+                    error=f"不支持的依赖安装语言: {self.config.language}",
+                    exit_code=1,
+                )
+
+            if errors:
+                return ExecutionResult(
+                    status=ExecutionStatus.ERROR,
+                    output="\n".join(outputs),
+                    error="\n".join(errors),
+                    exit_code=exit_code,
+                )
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS,
+                output="\n".join(outputs) or "安装完成",
+                exit_code=0,
+            )
+        except Exception as e:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error=f"依赖安装异常: {e}", exit_code=1
+            )
+
+    async def execute(self, code: str, shell=False) -> DisplayResult:
+        """在容器中执行代码，并封装 DisplayResult"""
+        if not self.container or not self._is_active:
+            return DisplayResult(
+                status="error",
+                output="",
+                error="容器未启动或已停止",
+                execution_time=0,
+                exit_code=-1,
+            )
+
+        if shell:
+            try:
+                self.update_last_accessed()
+                start_time = time.time()
+                result = self.container.exec_run(
+                    code, workdir=self.config.working_dir, demux=True
+                )
+                execution_time = time.time() - start_time
+
+                stdout, stderr = result.output
+                output_text = stdout.decode("utf-8") if stdout else ""
+                error_text = stderr.decode("utf-8") if stderr else ""
+
+                return DisplayResult(
+                    status="success" if result.exit_code == 0 else "error",
+                    output=output_text,
+                    error=error_text,
+                    execution_time=execution_time,
+                    exit_code=result.exit_code,
+                    files=[],
+                )
+            except Exception as e:
+                return DisplayResult(
+                    status="error",
+                    output="",
+                    error=f"执行失败: {str(e)}",
+                    execution_time=0,
+                    exit_code=-1,
+                )
+
+        self.update_last_accessed()
+        code_file = self._create_code_file(code)
+        tar_data = self._create_tar_from_file(code_file)
+
+        self.container.put_archive(self.config.working_dir, tar_data)
+
+        try:
+            exec_command = self._get_exec_command(os.path.basename(code_file))
+            start_time = time.time()
+            result = self.container.exec_run(
+                exec_command, workdir=self.config.working_dir, demux=True
+            )
+
+            execution_time = time.time() - start_time
+
+            stdout, stderr = result.output
+            output_text = stdout.decode("utf-8") if stdout else ""
+            error_text = stderr.decode("utf-8") if stderr else ""
+
+            return DisplayResult(
+                status="success" if result.exit_code == 0 else "error",
+                output=output_text,
+                error=error_text,
+                execution_time=execution_time,
+                exit_code=result.exit_code,
+                files=[os.path.basename(code_file)],
+            )
+        except Exception as e:
+            return DisplayResult(
+                status="error",
+                output="",
+                error=f"执行失败: {str(e)}",
+                execution_time=0,
+                exit_code=-1,
+            )
+        finally:
+            if "code_file" in locals():
+                os.unlink(code_file)
+
+    async def get_file_content(self, filename: str) -> Optional[DisplayResult]:
+        """从容器内获取文件内容"""
+        if not self.container or not self._is_active:
+            return None
+
+        file_path = os.path.join(self.config.working_dir, filename)
+
+        check = self.container.exec_run(f"test -f {file_path}")
+        if check.exit_code != 0:
+            return DisplayResult(
+                status="error",
+                output="",
+                error=f"文件不存在: {filename}",
+                execution_time=0,
+                exit_code=-1,
+            )
+
+        try:
+            bits, stat = self.container.get_archive(file_path)
+            file_data = io.BytesIO()
+            for chunk in bits:
+                file_data.write(chunk)
+            file_data.seek(0)
+
+            with tarfile.open(fileobj=file_data) as tar:
+                basename = os.path.basename(filename)
+
+                member = next((m for m in tar.getmembers() if m.name == basename), None)
+
+                if not member:
+                    member = next(
+                        (m for m in tar.getmembers() if m.name.endswith(basename)), None
+                    )
+
+                if not member:
+                    raise FileNotFoundError(
+                        f"{filename} 不在中: {[m.name for m in tar.getmembers()]}"
+                    )
+
+                extracted = tar.extractfile(member)
+                content_bytes = extracted.read()
+                # 统一转成 base64 字符串
+                file_content = base64.b64encode(content_bytes).decode("utf-8")
+
+            return DisplayResult(
+                status="success",
+                output=file_content,
+                error="",
+                execution_time=0,
+                exit_code=0,
+                files=[filename],
+            )
+
+        except Exception as e:
+            return DisplayResult(
+                status="error",
+                output="",
+                error=f"获取文件失败: {str(e)}",
+                execution_time=0,
+                exit_code=-1,
+            )
+
+    def _create_tar_from_file(self, filepath: str) -> bytes:
+        tar_stream = io.BytesIO()
+        filename = os.path.basename(filepath)
+
+        with tarfile.open(fileobj=tar_stream, mode="w") as tar:
+            with open(filepath, "rb") as f:
+                file_data = f.read()
+                tarinfo = tarfile.TarInfo(name=filename)
+                tarinfo.size = len(file_data)
+                tar.addfile(tarinfo=tarinfo, fileobj=io.BytesIO(file_data))
+
+        tar_stream.seek(0)
+        return tar_stream.read()
+
+    def _create_code_file(self, code: str) -> str:
+        extensions = {
+            "python": ".py",
+            "javascript": ".js",
+            "java": ".java",
+            "cpp": ".cpp",
+            "go": ".go",
+            "rust": ".rs",
+            "python-vnc": ".py",
+        }
+        ext = extensions.get(self.config.language, ".txt")
+        timestamp = int(time.time() * 1000)
+        filename = f"{self.session_id}_{timestamp}{ext}"
+        file_path = os.path.join(tempfile.gettempdir(), filename)
+        with open(file_path, "w", encoding="utf-8") as f:
+            f.write(code)
+        return file_path
+
+    def _get_exec_command(self, filename: str) -> str:
+        return get_command_by_language(self.config.language, filename)
+
+    async def get_status(self) -> Dict[str, Any]:
+        """获取容器状态"""
+        if not self.container:
+            return {"status": "stopped"}
+
+        try:
+            await asyncio.to_thread(self.container.reload)
+            stats = await asyncio.to_thread(self.container.stats, stream=False)
+            return {
+                "status": self.container.status,
+                "created_at": self.created_at,
+                "last_accessed": self.last_accessed,
+                "memory_usage": stats.get("memory", {}).get("usage", 0),
+                "cpu_usage": stats.get("cpu_stats", {})
+                .get("cpu_usage", {})
+                .get("total_usage", 0),
+            }
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
+
+
+class DockerRuntime(SandboxRuntime):
+    """异步 Docker 沙箱运行时管理"""
+
+    def __init__(self, runtime_id: str = "docker"):
+        super().__init__(runtime_id)
+        self.docker_client = docker.from_env()
+        self.supported_languages = list(LANGUAGE_IMAGES.keys())
+
+    async def create_session(
+        self, session_id: str, config: SessionConfig
+    ) -> SandboxSession:
+        if session_id in self.sessions:
+            raise ValueError(f"会话 {session_id} 已存在")
+
+        session = DockerSandboxSession(session_id, config, self.docker_client)
+
+        if await session.start():
+            self.sessions[session_id] = session
+            return session
+        else:
+            raise RuntimeError(f"启动会话 {session_id} 失败")
+
+    async def destroy_session(self, session_id: str) -> bool:
+        if session_id not in self.sessions:
+            return False
+
+        session = self.sessions[session_id]
+        asyncio.create_task(session.stop())
+        del self.sessions[session_id]
+        success = True
+
+        return success
+
+    async def list_sessions(self) -> List[str]:
+        return list(self.sessions.keys())
+
+    async def get_session(self, session_id: str) -> Optional[SandboxSession]:
+        return self.sessions.get(session_id)
+
+    async def cleanup_expired_sessions(self, max_idle_time: int = 3600) -> int:
+        current_time = time.time()
+        expired_sessions = [
+            sid
+            for sid, sess in self.sessions.items()
+            if current_time - sess.last_accessed > max_idle_time
+        ]
+
+        cleanup_count = 0
+        for sid in expired_sessions:
+            if await self.destroy_session(sid):
+                cleanup_count += 1
+        return cleanup_count
+
+    async def health_check(self) -> Dict[str, Any]:
+        try:
+            info = await asyncio.to_thread(self.docker_client.info)
+            return {
+                "status": "healthy",
+                "docker_version": info.get("ServerVersion", "unknown"),
+                "containers_running": info.get("ContainersRunning", 0),
+                "active_sessions": len(self.sessions),
+                "supported_languages": self.supported_languages,
+            }
+        except Exception as e:
+            return {"status": "unhealthy", "error": str(e)}
+
+    async def get_vnc_info(self) -> Dict[str, Any]:
+        if not self.container:
+            return {}
+        await asyncio.to_thread(self.container.reload)
+        ports = self.container.attrs["NetworkSettings"]["Ports"]
+        return {
+            "vnc_port": ports.get("5900/tcp", [{"HostPort": "5900"}])[0]["HostPort"],
+            "novnc_port": ports.get("6080/tcp", [{"HostPort": "6080"}])[0]["HostPort"],
+        }
+
+    def supports_language(self, language: str) -> bool:
+        return language.lower() in self.supported_languages

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/execution_layer/local_runtime.py

@@ -0,0 +1,407 @@
+"""
+基于本地进程的代码执行环境，使用 subprocess 和资源限制。
+"""
+
+import asyncio
+import os
+import subprocess
+import tempfile
+import time
+from typing import Any, Dict, List, Optional
+
+import psutil
+
+from .base import (
+    ExecutionResult,
+    ExecutionStatus,
+    SandboxRuntime,
+    SandboxSession,
+    SessionConfig,
+)
+from .utils import PathUtils, ProcessManager, SecurityUtils
+
+
+class LocalSandboxSession(SandboxSession):
+    """本地沙箱会话实现"""
+
+    def __init__(self, session_id: str, config: SessionConfig):
+        super().__init__(session_id, config)
+        self.work_dir = None
+        self.process_pool = []
+        self.path_utils = PathUtils()
+        self.process_manager = ProcessManager()
+        self.security_utils = SecurityUtils()
+
+    async def start(self) -> bool:
+        """启动本地沙箱会话"""
+        try:
+            # 创建工作目录
+            self.work_dir = self.path_utils.create_temp_dir(
+                f"sandbox_{self.session_id}_"
+            )
+
+            # 设置环境变量
+            self._setup_environment()
+
+            self._is_active = True
+            return True
+
+        except Exception as e:
+            print(f"启动本地沙箱失败: {e}")
+            return False
+
+    def _setup_environment(self):
+        """设置执行环境"""
+        # 设置基本环境变量
+        os.environ.update(self.config.environment_vars)
+
+        # 创建必要的子目录
+        os.makedirs(os.path.join(self.work_dir, "input"), exist_ok=True)
+        os.makedirs(os.path.join(self.work_dir, "output"), exist_ok=True)
+
+    async def stop(self) -> bool:
+        """停止本地沙箱会话"""
+        try:
+            # 清理所有子进程
+            for pid in self.process_pool:
+                self.process_manager.kill_process_tree(pid)
+
+            # 清理工作目录
+            if self.work_dir and os.path.exists(self.work_dir):
+                self.path_utils.cleanup_directory(self.work_dir)
+
+            self._is_active = False
+            return True
+
+        except Exception as e:
+            print(f"停止本地沙箱失败: {e}")
+            return False
+
+    async def execute(self, code: str) -> ExecutionResult:
+        """在本地进程中执行代码"""
+        if not self._is_active or not self.work_dir:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error="会话未启动或工作目录不存在"
+            )
+
+        self.update_last_accessed()
+
+        # 安全检查
+        warnings = self.security_utils.validate_code(code, self.config.language)
+        if warnings and any("危险操作" in w for w in warnings):
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR,
+                error=f"代码安全检查失败: {'; '.join(warnings)}",
+            )
+
+        try:
+            # 创建代码文件
+            code_file = self._create_code_file(code)
+
+            # 获取执行命令
+            command = self._get_exec_command(code_file)
+
+            # 执行代码
+            start_time = time.time()
+            result = await self._run_with_limits(command)
+            execution_time = time.time() - start_time
+
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS
+                if result["returncode"] == 0
+                else ExecutionStatus.ERROR,
+                output=result["stdout"],
+                error=result["stderr"],
+                execution_time=execution_time,
+                memory_usage=result.get("memory_usage", 0),
+                exit_code=result["returncode"],
+            )
+
+        except asyncio.TimeoutError:
+            return ExecutionResult(
+                status=ExecutionStatus.TIMEOUT,
+                error=f"执行超时 ({self.config.timeout}秒)",
+            )
+        except Exception as e:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error=f"执行失败: {str(e)}"
+            )
+        finally:
+            if "code_file" in locals():
+                os.unlink(code_file)
+
+    def _create_code_file(self, code: str) -> str:
+        """创建临时代码文件"""
+        extensions = {
+            "python": ".py",
+            "javascript": ".js",
+            "java": ".java",
+            "cpp": ".cpp",
+            "c": ".c",
+            "go": ".go",
+            "rust": ".rs",
+            "bash": ".sh",
+        }
+
+        ext = extensions.get(self.config.language, ".txt")
+
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=ext, dir=self.work_dir, delete=False, encoding="utf-8"
+        ) as f:
+            f.write(code)
+            return f.name
+
+    def _get_exec_command(self, code_file: str) -> List[str]:
+        """根据语言获取执行命令"""
+        filename = os.path.basename(code_file)
+
+        commands = {
+            "python": ["python", code_file],
+            "javascript": ["node", code_file],
+            "java": [
+                "sh",
+                "-c",
+                f"cd {os.path.dirname(code_file)} && \
+                    javac {filename} && java {filename[:-5]}",
+            ],
+            "cpp": [
+                "sh",
+                "-c",
+                f"cd {os.path.dirname(code_file)} && g++ -o program {filename} \
+                      && ./program",
+            ],
+            "c": [
+                "sh",
+                "-c",
+                f"cd {os.path.dirname(code_file)} && gcc -o program {filename} \
+                      && ./program",
+            ],
+            "go": ["go", "run", code_file],
+            "rust": [
+                "sh",
+                "-c",
+                f"cd {os.path.dirname(code_file)} && rustc {filename} \
+                  -o program && ./program",
+            ],
+            "bash": ["bash", code_file],
+        }
+
+        return commands.get(self.config.language, ["cat", code_file])
+
+    async def _run_with_limits(self, command: List[str]) -> Dict[str, Any]:
+        """在资源限制下运行命令"""
+        process = None
+        try:
+            # 启动进程
+            process = await asyncio.create_subprocess_exec(
+                *command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                cwd=self.work_dir,
+                env=dict(os.environ, **self.config.environment_vars),
+            )
+
+            # 记录进程 PID
+            if process.pid:
+                self.process_pool.append(process.pid)
+
+            # 设置资源监控
+            memory_usage = 0
+            if process.pid:
+                try:
+                    proc_info = psutil.Process(process.pid)
+                    memory_usage = proc_info.memory_info().rss
+
+                    # 检查内存限制
+                    if memory_usage > self.config.max_memory:
+                        process.terminate()
+                        return {
+                            "returncode": -1,
+                            "stdout": "",
+                            "stderr": "内存使用超出限制",
+                            "memory_usage": memory_usage,
+                        }
+                except psutil.NoSuchProcess:
+                    pass
+
+            # 等待进程完成（带超时）
+            stdout, stderr = await asyncio.wait_for(
+                process.communicate(), timeout=self.config.timeout
+            )
+
+            # 清理进程记录
+            if process.pid in self.process_pool:
+                self.process_pool.remove(process.pid)
+
+            return {
+                "returncode": process.returncode,
+                "stdout": stdout.decode("utf-8", errors="replace"),
+                "stderr": stderr.decode("utf-8", errors="replace"),
+                "memory_usage": memory_usage,
+            }
+
+        except asyncio.TimeoutError:
+            # 超时处理
+            if process and process.pid:
+                self.process_manager.kill_process_tree(process.pid)
+                if process.pid in self.process_pool:
+                    self.process_pool.remove(process.pid)
+            raise
+        except Exception as e:
+            # 其他异常处理
+            if process and process.pid:
+                self.process_manager.kill_process_tree(process.pid)
+                if process.pid in self.process_pool:
+                    self.process_pool.remove(process.pid)
+            raise e
+
+    async def get_status(self) -> Dict[str, Any]:
+        """获取会话状态"""
+        if not self._is_active:
+            return {"status": "stopped"}
+
+        active_processes = []
+        for pid in self.process_pool[:]:  # 复制列表避免修改时的问题
+            try:
+                proc = psutil.Process(pid)
+                active_processes.append(
+                    {
+                        "pid": pid,
+                        "status": proc.status(),
+                        "memory": proc.memory_info().rss,
+                        "cpu_percent": proc.cpu_percent(),
+                    }
+                )
+            except psutil.NoSuchProcess:
+                # 进程已结束，从池中移除
+                self.process_pool.remove(pid)
+
+        return {
+            "status": "running",
+            "work_dir": self.work_dir,
+            "created_at": self.created_at,
+            "last_accessed": self.last_accessed,
+            "active_processes": active_processes,
+            "process_count": len(active_processes),
+        }
+
+
+class LocalRuntime(SandboxRuntime):
+    """本地沙箱运行时"""
+
+    def __init__(self, runtime_id: str = "local"):
+        super().__init__(runtime_id)
+        self.supported_languages = self._detect_supported_languages()
+
+    def _detect_supported_languages(self) -> List[str]:
+        """检测系统支持的编程语言"""
+        languages = []
+
+        # 检查常见编程语言的可用性
+        language_commands = {
+            "python": ["python", "--version"],
+            "javascript": ["node", "--version"],
+            "java": ["java", "-version"],
+            "cpp": ["g++", "--version"],
+            "c": ["gcc", "--version"],
+            "go": ["go", "version"],
+            "rust": ["rustc", "--version"],
+            "bash": ["bash", "--version"],
+        }
+
+        for lang, cmd in language_commands.items():
+            try:
+                result = subprocess.run(
+                    cmd,
+                    capture_output=True,
+                    timeout=2,  # 减少超时时间
+                    shell=True,  # 在Windows上使用shell
+                )
+                if result.returncode == 0:
+                    languages.append(lang)
+            except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
+                # 如果命令不可用，跳过
+                pass
+
+        # 确保至少支持Python（假设在Python环境中运行）
+        if "python" not in languages:
+            languages.append("python")
+
+        return languages
+
+    async def create_session(
+        self, session_id: str, config: SessionConfig
+    ) -> SandboxSession:
+        """创建新的本地沙箱会话"""
+        if session_id in self.sessions:
+            raise ValueError(f"会话 {session_id} 已存在")
+
+        session = LocalSandboxSession(session_id, config)
+
+        if await session.start():
+            self.sessions[session_id] = session
+            return session
+        else:
+            raise RuntimeError(f"启动会话 {session_id} 失败")
+
+    async def destroy_session(self, session_id: str) -> bool:
+        """销毁本地沙箱会话"""
+        if session_id not in self.sessions:
+            return False
+
+        session = self.sessions[session_id]
+        success = await session.stop()
+
+        if success:
+            del self.sessions[session_id]
+
+        return success
+
+    async def list_sessions(self) -> List[str]:
+        """列出所有活跃会话"""
+        return list(self.sessions.keys())
+
+    async def get_session(self, session_id: str) -> Optional[SandboxSession]:
+        """获取指定会话"""
+        return self.sessions.get(session_id)
+
+    async def cleanup_expired_sessions(self, max_idle_time: int = 3600) -> int:
+        """清理过期会话"""
+        current_time = time.time()
+        expired_sessions = []
+
+        for session_id, session in self.sessions.items():
+            if current_time - session.last_accessed > max_idle_time:
+                expired_sessions.append(session_id)
+
+        cleanup_count = 0
+        for session_id in expired_sessions:
+            if await self.destroy_session(session_id):
+                cleanup_count += 1
+
+        return cleanup_count
+
+    async def health_check(self) -> Dict[str, Any]:
+        """本地运行时健康检查"""
+        try:
+            import psutil
+
+            return {
+                "status": "healthy",
+                "system_info": {
+                    "cpu_count": psutil.cpu_count(),
+                    "memory_total": psutil.virtual_memory().total,
+                    "memory_available": psutil.virtual_memory().available,
+                    "disk_usage": psutil.disk_usage("/").percent
+                    if os.name != "nt"
+                    else psutil.disk_usage("C:").percent,
+                },
+                "active_sessions": len(self.sessions),
+                "supported_languages": self.supported_languages,
+            }
+        except Exception as e:
+            return {"status": "unhealthy", "error": str(e)}
+
+    def supports_language(self, language: str) -> bool:
+        """检查是否支持指定编程语言"""
+        return language.lower() in self.supported_languages

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/execution_layer/nerdctl_runtime.py

@@ -0,0 +1,430 @@
+"""
+通过 nerdctl CLI 与 containerd 交互，提供与 Docker/Podman 一致的沙箱能力。
+"""
+
+import asyncio
+import contextlib
+import os
+import tempfile
+import time
+from typing import Any, Dict, List, Optional, Tuple
+
+from ..display_layer.display_layer import DisplayResult
+from ..utils_function.logger import print_log
+from .base import (
+    ExecutionResult,
+    ExecutionStatus,
+    SandboxRuntime,
+    SandboxSession,
+    SessionConfig,
+)
+
+
+async def _run_cmd(
+    cmd: List[str], timeout: Optional[int] = None, cwd: Optional[str] = None
+) -> Tuple[int, str, str]:
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        cwd=cwd,
+    )
+    try:
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+    except asyncio.TimeoutError:
+        with contextlib.suppress(ProcessLookupError):
+            proc.kill()
+        raise
+    return (
+        proc.returncode,
+        stdout.decode(errors="replace"),
+        stderr.decode(errors="replace"),
+    )
+
+
+class NerdctlSandboxSession(SandboxSession):
+    """异步 Nerdctl 沙箱会话实现"""
+
+    def __init__(self, session_id: str, config: SessionConfig):
+        super().__init__(session_id, config)
+        self.container_name = f"sandbox_{self.session_id}"
+        self.image_name = self._get_image_name(config.language)
+
+    def _get_image_name(self, language: str) -> str:
+        language_images = {
+            "python": "docker.io/library/python:3.11-slim",
+            "python-vnc": "vnc-gui-browser:latest",
+            "javascript": "docker.io/library/node:18-slim",
+            "java": "docker.io/library/openjdk:11-jre-slim",
+            "cpp": "docker.io/library/gcc:latest",
+            "go": "docker.io/library/golang:1.21-alpine",
+            "rust": "docker.io/library/rust:1.75-slim",
+        }
+        return language_images.get(language, "docker.io/library/python:3.11-slim")
+
+    async def start(self) -> bool:
+        try:
+            args = [
+                "nerdctl",
+                "run",
+                "-d",
+                "--name",
+                self.container_name,
+                "--memory",
+                str(self.config.max_memory),
+            ]
+
+            if getattr(self.config, "max_cpus", None):
+                args += ["--cpus", str(self.config.max_cpus)]
+
+            # nerdctl 的网络隔离先忽略
+
+            if getattr(self.config, "working_dir", None):
+                args += ["-w", self.config.working_dir]
+
+            for k, v in (self.config.environment_vars or {}).items():
+                args += ["-e", f"{k}={v}"]
+
+            args += ["-v", f"{tempfile.gettempdir()}:/tmp:rw"]
+
+            is_vnc = str(self.config.language).endswith("-vnc")
+            if is_vnc:
+                args += ["-p", "5900:5900", "-p", "6080:6080"]
+
+            if is_vnc:
+                args += [self.image_name, "sh", "-lc", "/startup.sh"]
+            else:
+                args += [self.image_name, "sh", "-lc", "tail -f /dev/null"]
+
+            code, out, err = await _run_cmd(args, timeout=60)
+            if code != 0:
+                print_log("ERROR", f"nerdctl 启动失败: {err.strip()}")
+                return False
+
+            self._is_active = True
+            if getattr(self.config, "working_dir", None):
+                await _run_cmd(
+                    [
+                        "nerdctl",
+                        "exec",
+                        "-w",
+                        "/",
+                        self.container_name,
+                        "sh",
+                        "-lc",
+                        f"mkdir -p {self.config.working_dir}",
+                    ],
+                    timeout=30,
+                )
+            return True
+        except Exception as e:
+            print_log("ERROR", f"启动 nerdctl 容器失败: {e}")
+            return False
+
+    async def stop(self) -> bool:
+        try:
+            await _run_cmd(["nerdctl", "stop", self.container_name], timeout=30)
+            await _run_cmd(["nerdctl", "rm", self.container_name], timeout=30)
+            self._is_active = False
+            return True
+        except Exception as e:
+            print_log("ERROR", f"停止 nerdctl 容器失败: {e}")
+            return False
+
+    def _create_code_file(self, code: str) -> str:
+        extensions = {
+            "python": ".py",
+            "python-vnc": ".py",
+            "javascript": ".js",
+            "java": ".java",
+            "cpp": ".cpp",
+            "go": ".go",
+            "rust": ".rs",
+            "bash": ".sh",
+        }
+        ext = extensions.get(self.config.language, ".txt")
+        filename = f"{self.session_id}_{int(time.time() * 1000)}{ext}"
+        path = os.path.join(tempfile.gettempdir(), filename)
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(code)
+        return path
+
+    def _get_exec_command(self, filename: str) -> str:
+        cmds = {
+            "python-vnc": f"python3 {filename}",
+            "python": f"python {filename}",
+            "javascript": f"node {filename}",
+            "java": f"javac {filename} && java {filename[:-5]}",
+            "cpp": f"g++ -o program {filename} && ./program",
+            "go": f"go run {filename}",
+            "rust": f"rustc {filename} -o program && ./program",
+            "bash": f"sh {filename}",
+        }
+        return cmds.get(self.config.language, f"cat {filename}")
+
+    async def install_dependencies(self, dependencies: List[str]) -> ExecutionResult:
+        """通过 nerdctl exec 安装 pip/npm 依赖。"""
+        if not self._is_active:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error="容器未启动", exit_code=-1
+            )
+        if not dependencies:
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS, output="无依赖需要安装", exit_code=0
+            )
+
+        try:
+            outputs: List[str] = []
+            errors: List[str] = []
+            exit_code = 0
+            workdir = self.config.working_dir or "/workspace"
+
+            if self.config.language.startswith("python"):
+                for dep in dependencies:
+                    code, out, err = await _run_cmd(
+                        [
+                            "nerdctl",
+                            "exec",
+                            self.container_name,
+                            "sh",
+                            "-lc",
+                            f"pip install --no-input --disable-pip-version-check {dep}",
+                        ],
+                        timeout=300,
+                    )
+                    exit_code = code
+                    if code != 0:
+                        errors.append(f"pip install {dep} 失败: {err.strip()}")
+                        break
+                    outputs.append(f"installed: {dep}")
+            elif self.config.language == "javascript":
+                await _run_cmd(
+                    [
+                        "nerdctl",
+                        "exec",
+                        "-w",
+                        workdir,
+                        self.container_name,
+                        "sh",
+                        "-lc",
+                        "npm init -y",
+                    ],
+                    timeout=120,
+                )
+                dep_str = " ".join(dependencies)
+                code, out, err = await _run_cmd(
+                    [
+                        "nerdctl",
+                        "exec",
+                        "-w",
+                        workdir,
+                        self.container_name,
+                        "sh",
+                        "-lc",
+                        f"npm install {dep_str}",
+                    ],
+                    timeout=600,
+                )
+                exit_code = code
+                if code != 0:
+                    errors.append(f"npm install 失败: {err.strip()}")
+                else:
+                    outputs.append(f"installed: {dep_str}")
+            else:
+                return ExecutionResult(
+                    status=ExecutionStatus.ERROR,
+                    error=f"不支持的依赖安装语言: {self.config.language}",
+                    exit_code=1,
+                )
+
+            if errors:
+                return ExecutionResult(
+                    status=ExecutionStatus.ERROR,
+                    output="\n".join(outputs),
+                    error="\n".join(errors),
+                    exit_code=exit_code,
+                )
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS,
+                output="\n".join(outputs) or "安装完成",
+                exit_code=0,
+            )
+        except Exception as e:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error=f"依赖安装异常: {e}", exit_code=1
+            )
+
+    async def execute(self, code: str) -> DisplayResult:
+        if not self._is_active:
+            return DisplayResult(
+                status="error",
+                output="",
+                error="容器未启动",
+                execution_time=0,
+                exit_code=-1,
+            )
+
+        # GUI 容器：直接提示已启动并返回访问 URL
+        if str(self.config.language).endswith("-vnc"):
+            return DisplayResult(
+                status="success",
+                output="GUI容器已启动",
+                error="",
+                execution_time=0,
+                exit_code=0,
+                files=[],
+                gui_url="http://localhost:6080/vnc.html",
+            )
+
+        self.update_last_accessed()
+        code_file = self._create_code_file(code)
+        try:
+            workdir = self.config.working_dir or "/workspace"
+            code_basename = os.path.basename(code_file)
+            code_in_container = f"{workdir}/{code_basename}"
+
+            # nerdctl cp 支持
+            cp_code, _, cp_err = await _run_cmd(
+                [
+                    "nerdctl",
+                    "cp",
+                    code_file,
+                    f"{self.container_name}:{code_in_container}",
+                ],
+                timeout=30,
+            )
+            if cp_code != 0:
+                return DisplayResult(
+                    status="error",
+                    output="",
+                    error=f"拷贝文件失败: {cp_err}",
+                    execution_time=0,
+                    exit_code=-1,
+                )
+
+            exec_cmd = self._get_exec_command(code_basename)
+            start = time.time()
+            code, out, err = await _run_cmd(
+                [
+                    "nerdctl",
+                    "exec",
+                    "-w",
+                    workdir,
+                    self.container_name,
+                    "sh",
+                    "-lc",
+                    exec_cmd,
+                ],
+                timeout=self.config.timeout,
+            )
+            duration = time.time() - start
+
+            return DisplayResult(
+                status="success" if code == 0 else "error",
+                output=out,
+                error=err,
+                execution_time=duration,
+                exit_code=code,
+                files=[code_basename],
+            )
+        except asyncio.TimeoutError:
+            return DisplayResult(
+                status="error",
+                output="",
+                error=f"执行超时({self.config.timeout}s)",
+                execution_time=self.config.timeout,
+                exit_code=124,
+            )
+        except Exception as e:
+            return DisplayResult(
+                status="error",
+                output="",
+                error=f"执行失败: {e}",
+                execution_time=0,
+                exit_code=-1,
+            )
+        finally:
+            with contextlib.suppress(Exception):
+                os.unlink(code_file)
+
+    async def get_status(self) -> Dict[str, Any]:
+        if not self._is_active:
+            return {"status": "stopped"}
+        try:
+            code, out, err = await _run_cmd(
+                ["nerdctl", "ps", "-a", "--format", "{{json .}}"], timeout=15
+            )
+            if code != 0:
+                return {"status": "error", "error": err.strip()}
+            return {
+                "status": "running",
+                "raw": out,
+                "created_at": self.created_at,
+                "last_accessed": self.last_accessed,
+            }
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
+
+
+class NerdctlRuntime(SandboxRuntime):
+    def __init__(self, runtime_id: str = "nerdctl"):
+        super().__init__(runtime_id)
+        self.supported_languages = [
+            "python",
+            "python-vnc",
+            "javascript",
+            "java",
+            "cpp",
+            "go",
+            "rust",
+            "bash",
+        ]
+
+    async def create_session(
+        self, session_id: str, config: SessionConfig
+    ) -> SandboxSession:
+        if session_id in self.sessions:
+            raise ValueError(f"会话 {session_id} 已存在")
+        sess = NerdctlSandboxSession(session_id, config)
+        ok = await sess.start()
+        if not ok:
+            raise RuntimeError(f"启动会话 {session_id} 失败")
+        self.sessions[session_id] = sess
+        return sess
+
+    async def destroy_session(self, session_id: str) -> bool:
+        sess = self.sessions.get(session_id)
+        if not sess:
+            return False
+        asyncio.create_task(sess.stop())
+        del self.sessions[session_id]
+        return True
+
+    async def list_sessions(self) -> List[str]:
+        return list(self.sessions.keys())
+
+    async def get_session(self, session_id: str) -> Optional[SandboxSession]:
+        return self.sessions.get(session_id)
+
+    async def cleanup_expired_sessions(self, max_idle_time: int = 3600) -> int:
+        now = time.time()
+        expired = [
+            sid
+            for sid, s in self.sessions.items()
+            if now - s.last_accessed > max_idle_time
+        ]
+        for sid in expired:
+            await self.destroy_session(sid)
+        return len(expired)
+
+    async def health_check(self) -> Dict[str, Any]:
+        try:
+            code, out, err = await _run_cmd(["nerdctl", "version"], timeout=10)
+            if code == 0:
+                return {"status": "healthy", "version": out}
+            return {"status": "unhealthy", "error": err}
+        except Exception as e:
+            return {"status": "unhealthy", "error": str(e)}
+
+    def supports_language(self, language: str) -> bool:
+        return language.lower() in self.supported_languages

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/execution_layer/podman_runtime.py

@@ -0,0 +1,448 @@
+"""
+基于 Podman 容器的代码执行环境，支持多语言和状态保持
+"""
+
+import asyncio
+import contextlib
+import os
+import tempfile
+import time
+from typing import Any, Dict, List, Optional, Tuple
+
+from ..display_layer.display_layer import DisplayResult
+from ..utils_function.logger import print_log
+from .base import (
+    ExecutionResult,
+    ExecutionStatus,
+    SandboxRuntime,
+    SandboxSession,
+    SessionConfig,
+)
+
+
+async def _run_cmd(
+    cmd: List[str], timeout: Optional[int] = None, cwd: Optional[str] = None
+) -> Tuple[int, str, str]:
+    """以异步方式运行命令并返回 (code, stdout, stderr)"""
+    proc = await asyncio.create_subprocess_exec(
+        *cmd,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+        cwd=cwd,
+    )
+    try:
+        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+    except asyncio.TimeoutError:
+        with contextlib.suppress(ProcessLookupError):
+            proc.kill()
+        raise
+    return (
+        proc.returncode,
+        stdout.decode(errors="replace"),
+        stderr.decode(errors="replace"),
+    )
+
+
+class PodmanSandboxSession(SandboxSession):
+    """异步 Podman 沙箱会话实现"""
+
+    def __init__(self, session_id: str, config: SessionConfig):
+        super().__init__(session_id, config)
+        self.container_name = f"sandbox_{self.session_id}"
+        self.image_name = self._get_image_name(config.language)
+
+    def _get_image_name(self, language: str) -> str:
+        language_images = {
+            "python": "docker.io/library/python:3.11-slim",
+            "python-vnc": "vnc-gui-browser:latest",
+            "javascript": "docker.io/library/node:18-slim",
+            "java": "docker.io/library/openjdk:11-jre-slim",
+            "cpp": "docker.io/library/gcc:latest",
+            "go": "docker.io/library/golang:1.21-alpine",
+            "rust": "docker.io/library/rust:1.75-slim",
+        }
+        return language_images.get(language, "docker.io/library/python:3.11-slim")
+
+    async def start(self) -> bool:
+        """启动 Podman 容器"""
+        try:
+            args = [
+                "podman",
+                "run",
+                "-d",
+                "--name",
+                self.container_name,
+                "--memory",
+                str(self.config.max_memory),
+            ]
+
+            # CPU 限制
+            if getattr(self.config, "max_cpus", None):
+                args += ["--cpus", str(self.config.max_cpus)]
+
+            # 网络
+            if getattr(self.config, "network_disabled", False):
+                args += ["--network", "none"]
+
+            # 工作目录
+            if getattr(self.config, "working_dir", None):
+                args += ["-w", self.config.working_dir]
+
+            # 环境变量
+            for k, v in (self.config.environment_vars or {}).items():
+                args += ["-e", f"{k}={v}"]
+
+            # 挂载临时目录
+            args += ["-v", f"{tempfile.gettempdir()}:/tmp:rw"]
+
+            # VNC GUI 容器端口映射
+            is_vnc = str(self.config.language).endswith("-vnc")
+            if is_vnc:
+                # 简化处理：固定映射 5900, 6080 端口。若端口占用将启动失败。
+                args += ["-p", "5900:5900", "-p", "6080:6080"]
+
+            # 镜像和命令
+            if is_vnc:
+                args += [self.image_name, "sh", "-lc", "/startup.sh"]
+            else:
+                args += [self.image_name, "sh", "-lc", "tail -f /dev/null"]
+
+            code, out, err = await _run_cmd(args, timeout=60)
+            if code != 0:
+                print_log("ERROR", f"Podman 启动失败: {err.strip()}")
+                return False
+
+            self._is_active = True
+            # 拉起后创建工作目录，以防镜像中不存在
+            if getattr(self.config, "working_dir", None):
+                await _run_cmd(
+                    [
+                        "podman",
+                        "exec",
+                        "-w",
+                        "/",
+                        self.container_name,
+                        "sh",
+                        "-lc",
+                        f"mkdir -p {self.config.working_dir}",
+                    ],
+                    timeout=30,
+                )
+            return True
+        except Exception as e:
+            print_log("ERROR", f"启动 Podman 容器失败: {e}")
+            return False
+
+    async def stop(self) -> bool:
+        try:
+            await _run_cmd(["podman", "stop", self.container_name], timeout=30)
+            await _run_cmd(["podman", "rm", self.container_name], timeout=30)
+            self._is_active = False
+            return True
+        except Exception as e:
+            print_log("ERROR", f"停止 Podman 容器失败: {e}")
+            return False
+
+    def _create_code_file(self, code: str) -> str:
+        extensions = {
+            "python": ".py",
+            "python-vnc": ".py",
+            "javascript": ".js",
+            "java": ".java",
+            "cpp": ".cpp",
+            "go": ".go",
+            "rust": ".rs",
+            "bash": ".sh",
+        }
+        ext = extensions.get(self.config.language, ".txt")
+        filename = f"{self.session_id}_{int(time.time() * 1000)}{ext}"
+        path = os.path.join(tempfile.gettempdir(), filename)
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(code)
+        return path
+
+    def _get_exec_command(self, filename: str) -> str:
+        cmds = {
+            "python-vnc": f"python3 {filename}",
+            "python": f"python {filename}",
+            "javascript": f"node {filename}",
+            "java": f"javac {filename} && java {filename[:-5]}",
+            "cpp": f"g++ -o program {filename} && ./program",
+            "go": f"go run {filename}",
+            "rust": f"rustc {filename} -o program && ./program",
+            "bash": f"sh {filename}",
+        }
+        return cmds.get(self.config.language, f"cat {filename}")
+
+    async def install_dependencies(self, dependencies: List[str]) -> ExecutionResult:
+        """通过 podman exec 安装 pip/npm 依赖。"""
+        if not self._is_active:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error="容器未启动", exit_code=-1
+            )
+        if not dependencies:
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS, output="无依赖需要安装", exit_code=0
+            )
+
+        try:
+            outputs: List[str] = []
+            errors: List[str] = []
+            exit_code = 0
+            workdir = self.config.working_dir or "/workspace"
+
+            if self.config.language.startswith("python"):
+                for dep in dependencies:
+                    code, out, err = await _run_cmd(
+                        [
+                            "podman",
+                            "exec",
+                            self.container_name,
+                            "sh",
+                            "-lc",
+                            f"pip install --no-input --disable-pip-version-check {dep}",
+                        ],
+                        timeout=300,
+                    )
+                    exit_code = code
+                    if code != 0:
+                        errors.append(f"pip install {dep} 失败: {err.strip()}")
+                        break
+                    outputs.append(f"installed: {dep}")
+            elif self.config.language == "javascript":
+                # 确保初始化
+                await _run_cmd(
+                    [
+                        "podman",
+                        "exec",
+                        "-w",
+                        workdir,
+                        self.container_name,
+                        "sh",
+                        "-lc",
+                        "npm init -y",
+                    ],
+                    timeout=120,
+                )
+                dep_str = " ".join(dependencies)
+                code, out, err = await _run_cmd(
+                    [
+                        "podman",
+                        "exec",
+                        "-w",
+                        workdir,
+                        self.container_name,
+                        "sh",
+                        "-lc",
+                        f"npm install {dep_str}",
+                    ],
+                    timeout=600,
+                )
+                exit_code = code
+                if code != 0:
+                    errors.append(f"npm install 失败: {err.strip()}")
+                else:
+                    outputs.append(f"installed: {dep_str}")
+            else:
+                return ExecutionResult(
+                    status=ExecutionStatus.ERROR,
+                    error=f"不支持的依赖安装语言: {self.config.language}",
+                    exit_code=1,
+                )
+
+            if errors:
+                return ExecutionResult(
+                    status=ExecutionStatus.ERROR,
+                    output="\n".join(outputs),
+                    error="\n".join(errors),
+                    exit_code=exit_code,
+                )
+            return ExecutionResult(
+                status=ExecutionStatus.SUCCESS,
+                output="\n".join(outputs) or "安装完成",
+                exit_code=0,
+            )
+        except Exception as e:
+            return ExecutionResult(
+                status=ExecutionStatus.ERROR, error=f"依赖安装异常: {e}", exit_code=1
+            )
+
+    async def execute(self, code: str) -> DisplayResult:
+        if not self._is_active:
+            return DisplayResult(
+                status="error",
+                output="",
+                error="容器未启动",
+                execution_time=0,
+                exit_code=-1,
+            )
+
+        # GUI 容器：直接提示已启动并返回访问 URL
+        if str(self.config.language).endswith("-vnc"):
+            return DisplayResult(
+                status="success",
+                output="GUI容器已启动",
+                error="",
+                execution_time=0,
+                exit_code=0,
+                files=[],
+                gui_url="http://localhost:6080/vnc.html",
+            )
+
+        self.update_last_accessed()
+        code_file = self._create_code_file(code)
+        try:
+            # 拷贝代码到容器
+            workdir = self.config.working_dir or "/workspace"
+            code_basename = os.path.basename(code_file)
+            code_in_container = f"{workdir}/{code_basename}"
+
+            cp_code, _, cp_err = await _run_cmd(
+                [
+                    "podman",
+                    "cp",
+                    code_file,
+                    f"{self.container_name}:{code_in_container}",
+                ],
+                timeout=30,
+            )
+            if cp_code != 0:
+                return DisplayResult(
+                    status="error",
+                    output="",
+                    error=f"拷贝文件失败: {cp_err}",
+                    execution_time=0,
+                    exit_code=-1,
+                )
+
+            # 执行命令
+            exec_cmd = self._get_exec_command(code_basename)
+            start = time.time()
+            code, out, err = await _run_cmd(
+                [
+                    "podman",
+                    "exec",
+                    "-w",
+                    workdir,
+                    self.container_name,
+                    "sh",
+                    "-lc",
+                    exec_cmd,
+                ],
+                timeout=self.config.timeout,
+            )
+            duration = time.time() - start
+
+            return DisplayResult(
+                status="success" if code == 0 else "error",
+                output=out,
+                error=err,
+                execution_time=duration,
+                exit_code=code,
+                files=[code_basename],
+            )
+        except asyncio.TimeoutError:
+            return DisplayResult(
+                status="error",
+                output="",
+                error=f"执行超时({self.config.timeout}s)",
+                execution_time=self.config.timeout,
+                exit_code=124,
+            )
+        except Exception as e:
+            return DisplayResult(
+                status="error",
+                output="",
+                error=f"执行失败: {e}",
+                execution_time=0,
+                exit_code=-1,
+            )
+        finally:
+            with contextlib.suppress(Exception):
+                os.unlink(code_file)
+
+    async def get_status(self) -> Dict[str, Any]:
+        if not self._is_active:
+            return {"status": "stopped"}
+        try:
+            # podman inspect 读取状态
+            code, out, err = await _run_cmd(
+                ["podman", "inspect", self.container_name, "--format", "json"],
+                timeout=15,
+            )
+            if code != 0:
+                return {"status": "error", "error": err.strip()}
+            return {
+                "status": "running",
+                "raw": out,
+                "created_at": self.created_at,
+                "last_accessed": self.last_accessed,
+            }
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
+
+
+class PodmanRuntime(SandboxRuntime):
+    """异步 Podman 沙箱运行时管理"""
+
+    def __init__(self, runtime_id: str = "podman"):
+        super().__init__(runtime_id)
+        self.supported_languages = [
+            "python",
+            "python-vnc",
+            "javascript",
+            "java",
+            "cpp",
+            "go",
+            "rust",
+            "bash",
+        ]
+
+    async def create_session(
+        self, session_id: str, config: SessionConfig
+    ) -> SandboxSession:
+        if session_id in self.sessions:
+            raise ValueError(f"会话 {session_id} 已存在")
+        sess = PodmanSandboxSession(session_id, config)
+        ok = await sess.start()
+        if not ok:
+            raise RuntimeError(f"启动会话 {session_id} 失败")
+        self.sessions[session_id] = sess
+        return sess
+
+    async def destroy_session(self, session_id: str) -> bool:
+        sess = self.sessions.get(session_id)
+        if not sess:
+            return False
+        asyncio.create_task(sess.stop())
+        del self.sessions[session_id]
+        return True
+
+    async def list_sessions(self) -> List[str]:
+        return list(self.sessions.keys())
+
+    async def get_session(self, session_id: str) -> Optional[SandboxSession]:
+        return self.sessions.get(session_id)
+
+    async def cleanup_expired_sessions(self, max_idle_time: int = 3600) -> int:
+        now = time.time()
+        expired = [
+            sid
+            for sid, s in self.sessions.items()
+            if now - s.last_accessed > max_idle_time
+        ]
+        for sid in expired:
+            await self.destroy_session(sid)
+        return len(expired)
+
+    async def health_check(self) -> Dict[str, Any]:
+        try:
+            code, out, err = await _run_cmd(["podman", "version"], timeout=10)
+            if code == 0:
+                return {"status": "healthy", "version": out}
+            return {"status": "unhealthy", "error": err}
+        except Exception as e:
+            return {"status": "unhealthy", "error": str(e)}
+
+    def supports_language(self, language: str) -> bool:
+        return language.lower() in self.supported_languages

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/execution_layer/runtime_factory.py

@@ -0,0 +1,62 @@
+"""
+自动沙箱运行时工厂
+
+根据本机环境优先级自动选择 Docker/Podman/Nerdctl/Local 运行时。
+"""
+
+from ..config import SANDBOX_RUNTIME
+from .docker_runtime import DockerRuntime
+from .local_runtime import LocalRuntime
+from .nerdctl_runtime import NerdctlRuntime
+from .podman_runtime import PodmanRuntime
+from .utils import EnvironmentDetector
+
+
+class RuntimeFactory:
+    """自动选择最佳沙箱运行时"""
+
+    @staticmethod
+    def create(runtime_preference: str = None):
+        """
+        创建最佳可用运行时。
+        """
+        env_choice = SANDBOX_RUNTIME
+        if env_choice:
+            runtime_preference = env_choice.lower()
+
+        if runtime_preference:
+            runtime_preference = runtime_preference.lower()
+            if (
+                runtime_preference == "docker"
+                and EnvironmentDetector.is_docker_sdk_available()
+            ):
+                return DockerRuntime()
+            if (
+                runtime_preference == "podman"
+                and EnvironmentDetector.is_podman_available()
+            ):
+                return PodmanRuntime()
+            if (
+                runtime_preference == "nerdctl"
+                and EnvironmentDetector.is_nerdctl_available()
+            ):
+                return NerdctlRuntime()
+            if runtime_preference == "local":
+                return LocalRuntime()
+            raise RuntimeError(f"指定的运行时不可用: {runtime_preference}")
+
+        if EnvironmentDetector.is_docker_sdk_available():
+            try:
+                print("检测到 Docker SDK 可用")
+                import docker
+
+                client = docker.from_env()
+                client.info()
+                return DockerRuntime()
+            except Exception:
+                pass
+        if EnvironmentDetector.is_podman_available():
+            return PodmanRuntime()
+        if EnvironmentDetector.is_nerdctl_available():
+            return NerdctlRuntime()
+        return LocalRuntime()

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/execution_layer/utils.py

@@ -0,0 +1,215 @@
+"""
+提供资源限制、路径处理、环境检测等工具功能。
+"""
+
+import os
+import platform
+import shutil
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+import psutil
+
+from ..config import (
+    MAX_CPU_PERCENT,
+    MAX_EXECUTION_TIME,
+    MAX_FILE_SIZE,
+    MAX_MEMORY,
+    MAX_PROCESSES,
+)
+
+
+@dataclass
+class ResourceLimits:
+    """资源限制配置"""
+
+    max_memory: int = MAX_MEMORY
+    max_cpu_percent: float = MAX_CPU_PERCENT
+    max_execution_time: int = MAX_EXECUTION_TIME  # seconds
+    max_file_size: int = MAX_FILE_SIZE  # 10MB
+    max_processes: int = MAX_PROCESSES
+
+
+class EnvironmentDetector:
+    """环境检测工具"""
+
+    @staticmethod
+    def is_docker_available() -> bool:
+        """检查 Docker CLI 是否可用"""
+        return shutil.which("docker") is not None
+
+    @staticmethod
+    def is_docker_sdk_available() -> bool:
+        """检查 Docker Python SDK 是否可用"""
+        try:
+            return True
+        except Exception:
+            return False
+
+    @staticmethod
+    def is_podman_available() -> bool:
+        """检查 Podman 是否可用"""
+        return shutil.which("podman") is not None
+
+    @staticmethod
+    def is_nerdctl_available() -> bool:
+        """检查 Nerdctl 是否可用"""
+        return shutil.which("nerdctl") is not None
+
+    @staticmethod
+    def get_system_info() -> Dict[str, Any]:
+        """获取系统信息"""
+        return {
+            "platform": platform.platform(),
+            "python_version": platform.python_version(),
+            "cpu_count": psutil.cpu_count(),
+            "memory_total": psutil.virtual_memory().total,
+            "memory_available": psutil.virtual_memory().available,
+            "disk_usage": psutil.disk_usage("/").percent
+            if os.name != "nt"
+            else psutil.disk_usage("C:").percent,
+        }
+
+    @staticmethod
+    def check_resource_availability(limits: ResourceLimits) -> Dict[str, bool]:
+        """检查资源是否满足限制要求"""
+        memory = psutil.virtual_memory()
+
+        return {
+            "memory_ok": memory.available >= limits.max_memory,
+            "cpu_ok": psutil.cpu_count() >= 1,
+            "disk_ok": True,
+        }
+
+
+class PathUtils:
+    """路径处理工具"""
+
+    @staticmethod
+    def ensure_safe_path(path: str, base_dir: str) -> str:
+        """确保路径安全，防止路径遍历攻击"""
+        normalized_path = os.path.normpath(path)
+        normalized_base = os.path.normpath(base_dir)
+
+        if not normalized_path.startswith(normalized_base):
+            raise ValueError(f"不安全的路径: {path}")
+
+        return normalized_path
+
+    @staticmethod
+    def create_temp_dir(prefix: str = "sandbox_") -> str:
+        """创建临时目录"""
+        import tempfile
+
+        return tempfile.mkdtemp(prefix=prefix)
+
+    @staticmethod
+    def cleanup_directory(directory: str) -> bool:
+        """清理目录"""
+        try:
+            if os.path.exists(directory):
+                shutil.rmtree(directory)
+            return True
+        except Exception as e:
+            print(f"清理目录失败: {e}")
+            return False
+
+
+class ProcessManager:
+    """进程管理工具"""
+
+    @staticmethod
+    def kill_process_tree(pid: int) -> bool:
+        """终止进程树"""
+        try:
+            parent = psutil.Process(pid)
+            children = parent.children(recursive=True)
+
+            for child in children:
+                try:
+                    child.terminate()
+                except psutil.NoSuchProcess:
+                    pass
+
+            gone, alive = psutil.wait_procs(children, timeout=3)
+
+            for p in alive:
+                try:
+                    p.kill()
+                except psutil.NoSuchProcess:
+                    pass
+
+            try:
+                parent.terminate()
+                parent.wait(timeout=3)
+            except (psutil.NoSuchProcess, psutil.TimeoutExpired):
+                try:
+                    parent.kill()
+                except psutil.NoSuchProcess:
+                    pass
+
+            return True
+        except Exception as e:
+            print(f"终止进程树失败: {e}")
+            return False
+
+    @staticmethod
+    def get_process_stats(pid: int) -> Optional[Dict[str, Any]]:
+        """获取进程统计信息"""
+        try:
+            process = psutil.Process(pid)
+            return {
+                "cpu_percent": process.cpu_percent(),
+                "memory_info": process.memory_info()._asdict(),
+                "status": process.status(),
+                "create_time": process.create_time(),
+            }
+        except psutil.NoSuchProcess:
+            return None
+
+
+class SecurityUtils:
+    """安全工具"""
+
+    @staticmethod
+    def validate_code(code: str, language: str) -> List[str]:
+        """验证代码安全性，返回警告列表"""
+        warnings = []
+
+        dangerous_patterns = [
+            "import os",
+            "import subprocess",
+            "import sys",
+            "__import__",
+            "eval(",
+            "exec(",
+            "open(",
+            "file(",
+            "input(",
+            "raw_input(",
+            "socket",
+            "urllib",
+            "requests",
+            "rmdir",
+            "remove",
+            "unlink",
+            "delete",
+        ]
+
+        code_lower = code.lower()
+        for pattern in dangerous_patterns:
+            if pattern in code_lower:
+                warnings.append(f"检测到潜在危险操作: {pattern}")
+
+        if language == "python":
+            if "pickle" in code_lower:
+                warnings.append("检测到 pickle 模块使用，可能存在安全风险")
+
+        return warnings
+
+
+path_utils = PathUtils()
+resource_limits = ResourceLimits()
+environment_detector = EnvironmentDetector()
+process_manager = ProcessManager()
+security_utils = SecurityUtils()

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/main.py

@@ -0,0 +1,61 @@
+"""
+DB-GPT Sandbox Main Entry Point
+"""
+
+import argparse
+import logging
+import sys
+
+from .user_layer.service import initialize_sandbox
+
+logger = logging.getLogger(__name__)
+
+
+def setup_logging():
+    """设置日志"""
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    )
+
+
+def parse_args():
+    """解析命令行参数"""
+    parser = argparse.ArgumentParser(description="DB-GPT Sandbox Server")
+    parser.add_argument("--host", type=str, default="0.0.0.0", help="Host to bind to")
+    parser.add_argument("--port", type=int, default=8000, help="Port to bind to")
+    parser.add_argument(
+        "--log-level",
+        type=str,
+        default="info",
+        choices=["debug", "info", "warning", "error"],
+        help="Log level",
+    )
+    return parser.parse_args()
+
+
+def run_sandbox_server(
+    host: str = "0.0.0.0", port: int = 8000, log_level: str = "info"
+):
+    """运行沙箱服务器"""
+    setup_logging()
+
+    logger.info(f"Starting DB-GPT Sandbox server on {host}:{port}")
+
+    try:
+        initialize_sandbox(host=host, port=port, log_level=log_level)
+    except KeyboardInterrupt:
+        logger.info("Shutting down DB-GPT Sandbox server...")
+    except Exception as e:
+        logger.error(f"Failed to start server: {e}")
+        sys.exit(1)
+
+
+def main():
+    """默认入口函数"""
+    args = parse_args()
+    run_sandbox_server(args.host, args.port, args.log_level)
+
+
+if __name__ == "__main__":
+    main()

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/user_layer/__init__.py

@@ -0,0 +1 @@
+# uvicorn user_layer.service:app --reload --host 0.0.0.0 --port 8000

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/user_layer/schemas.py

@@ -0,0 +1,58 @@
+from typing import Any, Dict, Optional
+
+TASK_TYPES = [
+    "connect",
+    "configure",
+    "execute",
+    "manual",
+    "disconnect",
+    "status",
+    "list",
+    "get_file",
+]
+
+
+# -------------------- TaskObject 封装 --------------------
+class TaskObject:
+    """封装用户任务信息"""
+
+    def __init__(
+        self,
+        task_type: str,
+        user_id: str,
+        task_id: str,
+        session_id: str,
+        language: str,
+        code_content: Optional[str] = None,
+        config: Optional[Dict[str, Any]] = None,
+        manual_action: Optional[str] = None,
+        file_name: Optional[str] = None,
+    ):
+        self.task_type = task_type
+        if task_type not in TASK_TYPES:
+            raise ValueError(f"Invalid task_type: {task_type}")
+        self.user_id = user_id
+        self.task_id = task_id
+        self.session_id = session_id
+        self.language = language
+        self.code_content = code_content
+        self.config = config or {}
+        self.manual_action = manual_action
+        self.file_name = file_name
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "user_id": self.user_id,
+            "task_id": self.task_id,
+            "session_id": self.session_id,
+            "language": self.language,
+            "code_content": self.code_content,
+            "config": self.config,
+            "manual_action": self.manual_action,
+            "file_name": self.file_name,
+        }
+
+    def __repr__(self):
+        return f"TaskObject(session_id={self.session_id}, \
+            language={self.language}, action={self.manual_action}, \
+            file_name={self.file_name})"

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/user_layer/service.py

@@ -0,0 +1,350 @@
+import logging
+from typing import Any, Dict, List
+
+from fastapi import APIRouter, FastAPI
+from pydantic import BaseModel
+
+from ..control_layer.control_layer import ControlLayer
+from .schemas import TaskObject
+
+logger = logging.getLogger(__name__)
+
+# 创建全局路由器
+router = APIRouter()
+
+# 全局用户层实例
+user_layer_instance = None
+
+
+def get_user_layer() -> "UserLayer":
+    """获取用户层实例"""
+    global user_layer_instance
+    if user_layer_instance is None:
+        user_layer_instance = UserLayer()
+    return user_layer_instance
+
+
+class UserLayer:
+    """用户层：处理用户请求和会话管理"""
+
+    def __init__(self):
+        self.control = ControlLayer()
+        self.active_sessions: Dict[str, str] = {}  # session_id -> task_id
+
+    async def connect(
+        self, user_id: str, task_id: str, image_type: str
+    ) -> Dict[str, Any]:
+        session_id = f"{user_id}_{task_id}"
+        task = TaskObject(
+            task_type="connect",
+            user_id=user_id,
+            task_id=task_id,
+            session_id=session_id,
+            language=image_type,
+        )
+        result = await self.control.handle_task(task)
+        self.active_sessions[session_id] = task_id
+        return {
+            "status": result.status,
+            "output": getattr(result, "output", None),
+            "error": getattr(result, "error", None),
+        }
+
+    async def configure_environment(
+        self, user_id: str, task_id: str, config_info: Dict
+    ) -> Dict[str, Any]:
+        session_id = f"{user_id}_{task_id}"
+        task = TaskObject(
+            task_type="configure",
+            user_id=user_id,
+            task_id=task_id,
+            session_id=session_id,
+            language=config_info.get("language", "python"),
+            config=config_info,
+        )
+        result = await self.control.handle_task(task)
+        return {
+            "status": result.status,
+            "output": getattr(result, "output", None),
+            "error": getattr(result, "error", None),
+        }
+
+    async def disconnect(self, user_id: str, task_id: str) -> Dict[str, Any]:
+        session_id = f"{user_id}_{task_id}"
+        task = TaskObject(
+            task_type="disconnect",
+            user_id=user_id,
+            task_id=task_id,
+            session_id=session_id,
+            language="unknown",
+        )
+        result = await self.control.handle_task(task)
+        self.active_sessions.pop(session_id, None)
+        return {
+            "status": result.status,
+            "output": getattr(result, "output", None),
+            "error": getattr(result, "error", None),
+        }
+
+    async def execute_code(
+        self, session_id: str, code_type: str, code_content: str
+    ) -> Dict[str, Any]:
+        task_id = self.active_sessions.get(session_id, f"{session_id}_auto")
+        task = TaskObject(
+            task_type="execute",
+            user_id="user",
+            task_id=task_id,
+            session_id=session_id,
+            language=code_type,
+            code_content=code_content,
+        )
+        result = await self.control.handle_task(task)
+        return {
+            "status": result.status,
+            "output": getattr(result, "output", None),
+            "error": getattr(result, "error", None),
+        }
+
+    async def get_file(self, session_id: str, filename: str) -> Dict[str, Any]:
+        task_id = self.active_sessions.get(session_id, f"{session_id}_auto")
+        task = TaskObject(
+            task_type="get_file",
+            user_id="user",
+            task_id=task_id,
+            session_id=session_id,
+            language="python",
+            file_name=filename,
+        )
+        result = await self.control.handle_task(task)
+        return {
+            "status": result.status,
+            "output": getattr(result, "output", None),
+            "error": getattr(result, "error", None),
+        }
+
+    async def manual_operation(self, session_id: str, action: str) -> Dict[str, Any]:
+        task_id = self.active_sessions.get(session_id, f"{session_id}_auto")
+        task = TaskObject(
+            task_type="manual",
+            user_id="user",
+            task_id=task_id,
+            session_id=session_id,
+            language="python",
+            manual_action=action,
+        )
+        result = await self.control.handle_task(task)
+        return {
+            "status": result.status,
+            "output": getattr(result, "output", None),
+            "error": getattr(result, "error", None),
+        }
+
+    async def get_execution_status(self, session_id: str) -> Dict[str, Any]:
+        task_id = self.active_sessions.get(session_id, f"{session_id}_auto")
+        task = TaskObject(
+            task_type="status",
+            user_id="user",
+            task_id=task_id,
+            session_id=session_id,
+            language="python",
+        )
+        result = await self.control.handle_task(task)
+        return {
+            "status": result.status,
+            "output": getattr(result, "output", None),
+            "error": getattr(result, "error", None),
+        }
+
+    async def list_sessions(self) -> List[str]:
+        return await self.control.runtime.list_sessions()
+
+    def get_available_methods(self) -> List[Dict[str, str]]:
+        return [
+            {"path": "/api/connect", "method": "POST", "description": "建立沙箱会话"},
+            {"path": "/api/configure", "method": "POST", "description": "配置沙箱环境"},
+            {
+                "path": "/api/disconnect",
+                "method": "POST",
+                "description": "断开并销毁沙箱会话",
+            },
+            {"path": "/api/execute", "method": "POST", "description": "执行代码"},
+            {
+                "path": "/api/manual",
+                "method": "POST",
+                "description": "进入手动操作模式",
+            },
+            {
+                "path": "/api/status",
+                "method": "POST",
+                "description": "获取任务/会话状态",
+            },
+            {
+                "path": "/api/sessions",
+                "method": "GET",
+                "description": "列出所有活跃会话",
+            },
+            {
+                "path": "/api/get_file",
+                "method": "POST",
+                "description": "获取沙箱内指定文件内容",
+            },
+            {
+                "path": "/api/methods",
+                "method": "GET",
+                "description": "获取所有可用接口和方法",
+            },
+        ]
+
+
+# -------------------- API 请求模型 --------------------
+class ConnectRequest(BaseModel):
+    user_id: str
+    task_id: str
+    image_type: str
+
+
+class ConfigureRequest(BaseModel):
+    user_id: str
+    task_id: str
+    config_info: Dict
+
+
+class DisconnectRequest(BaseModel):
+    user_id: str
+    task_id: str
+
+
+class ExecuteRequest(BaseModel):
+    session_id: str
+    code_type: str
+    code_content: str
+
+
+class ManualOperationRequest(BaseModel):
+    session_id: str
+    action: str
+
+
+class StatusRequest(BaseModel):
+    session_id: str
+
+
+class FileRequest(BaseModel):
+    session_id: str
+    file_name: str
+
+
+# -------------------- API 路由 --------------------
+@router.get("/health")
+async def api_health_check():
+    """健康检查 API"""
+    return {"status": "ok"}
+
+
+@router.post("/connect")
+async def api_connect(req: ConnectRequest):
+    """建立沙箱会话"""
+    user_layer = get_user_layer()
+    return await user_layer.connect(req.user_id, req.task_id, req.image_type)
+
+
+@router.post("/configure")
+async def api_configure(req: ConfigureRequest):
+    """配置沙箱环境"""
+    user_layer = get_user_layer()
+    return await user_layer.configure_environment(
+        req.user_id, req.task_id, req.config_info
+    )
+
+
+@router.post("/disconnect")
+async def api_disconnect(req: DisconnectRequest):
+    """断开并销毁沙箱会话"""
+    user_layer = get_user_layer()
+    return await user_layer.disconnect(req.user_id, req.task_id)
+
+
+@router.post("/execute")
+async def api_execute(req: ExecuteRequest):
+    """执行代码"""
+    user_layer = get_user_layer()
+    return await user_layer.execute_code(
+        req.session_id, req.code_type, req.code_content
+    )
+
+
+@router.post("/manual")
+async def api_manual(req: ManualOperationRequest):
+    """进入手动操作模式"""
+    user_layer = get_user_layer()
+    return await user_layer.manual_operation(req.session_id, req.action)
+
+
+@router.post("/status")
+async def api_status(req: StatusRequest):
+    """获取任务/会话状态"""
+    user_layer = get_user_layer()
+    return await user_layer.get_execution_status(req.session_id)
+
+
+@router.get("/sessions")
+async def api_list_sessions():
+    """列出所有活跃会话"""
+    user_layer = get_user_layer()
+    sessions = await user_layer.list_sessions()
+    return {"sessions": sessions}
+
+
+@router.post("/get_file")
+async def api_get_file(req: FileRequest):
+    """获取沙箱内指定文件内容"""
+    user_layer = get_user_layer()
+    return await user_layer.get_file(req.session_id, req.file_name)
+
+
+@router.get("/methods")
+async def api_methods():
+    """获取所有可用接口和方法"""
+    user_layer = get_user_layer()
+    return {"methods": user_layer.get_available_methods()}
+
+
+def initialize_sandbox(
+    app: FastAPI = None,
+    host: str = "0.0.0.0",
+    port: int = 8000,
+    log_level: str = "info",
+):
+    """初始化沙箱服务
+
+    Args:
+        app: FastAPI 应用实例，如果为 None 则创建新实例并运行服务器
+        host: 主机地址
+        port: 端口号
+        log_level: 日志级别
+    """
+    if app:
+        # 将路由注册到现有应用
+        app.include_router(router, tags=["Sandbox"])
+        logger.info("Sandbox routes registered to existing FastAPI app")
+    else:
+        # 创建新应用并运行服务器
+        import uvicorn
+        from fastapi import FastAPI
+
+        app = FastAPI(
+            title="DB-GPT Sandbox API",
+            description="Secure sandbox execution environment for DB-GPT Agent",
+            version="0.7.3",
+        )
+
+        # 包含路由
+        app.include_router(router, prefix="/api", tags=["Sandbox"])
+
+        # 添加根路径
+        @app.get("/")
+        async def root():
+            return {"message": "DB-GPT Sandbox API is running"}
+
+        logger.info(f"Starting DB-GPT Sandbox server on {host}:{port}")
+        uvicorn.run(app, host=host, port=port, log_level=log_level)

packages/dbgpt-sandbox/src/dbgpt_sandbox/sandbox/utils_function/logger.py

@@ -0,0 +1,15 @@
+from colorama import Fore, Style, init
+
+# 初始化 colorama（Windows 下必须）
+init(autoreset=True)
+
+
+def print_log(level: str, msg: str):
+    COLORS = {
+        "INFO": Fore.GREEN,
+        "WARNING": Fore.YELLOW,
+        "ERROR": Fore.RED,
+        "DEBUG": Fore.BLUE,
+    }
+    color = COLORS.get(level.upper(), "")
+    print(f"{color}{level.upper()}:{Style.RESET_ALL} {msg}")

packages/dbgpt-sandbox/src/docker_images/create_docker.py

@@ -0,0 +1,107 @@
+import docker
+
+# 镜像与容器名配置
+IMAGE_NAME = "ubuntu-vnc-gui-v2"
+CONTAINER_NAME = "vncdemov2"
+
+# Dockerfile 内容
+DOCKERFILE_CONTENT = """
+FROM ubuntu:20.04
+
+ENV DEBIAN_FRONTEND=noninteractive
+
+RUN apt-get update && apt-get install -y \
+    python3 \
+    python3-pip \
+    xvfb \
+    x11vnc \
+    novnc \
+    websockify \
+    net-tools \
+    xterm \
+    wget \
+    fluxbox \
+    dos2unix \
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+RUN mkdir -p /root/.vnc && \
+    x11vnc -storepasswd 123456 /root/.vnc/passwd
+
+COPY startup.sh /startup.sh
+RUN chmod +x /startup.sh && dos2unix /startup.sh
+
+EXPOSE 80 5900
+
+CMD ["/startup.sh"]
+"""
+
+# 启动脚本内容
+STARTUP_SCRIPT = """#!/bin/bash
+
+Xvfb :0 -screen 0 1280x960x24 -listen tcp -ac +extension GLX +extension RENDER &
+
+export DISPLAY=:0
+
+fluxbox &
+
+x11vnc -display :0 -forever -shared -rfbauth /root/.vnc/passwd -rfbport 5900 &
+
+websockify --web=/usr/share/novnc 80 localhost:5900
+"""
+
+# 写入 Dockerfile 和 startup.sh
+with open("Dockerfile", "w") as f:
+    f.write(DOCKERFILE_CONTENT)
+
+with open("startup.sh", "w") as f:
+    f.write(STARTUP_SCRIPT)
+
+# 获取 Docker 客户端
+client = docker.from_env()
+
+# 检查镜像是否存在
+try:
+    client.images.get(IMAGE_NAME)
+    print(f"✅ Image '{IMAGE_NAME}' already exists, skipping build.")
+except docker.errors.ImageNotFound:
+    print("🔧 Building Docker image...")
+    image, logs = client.images.build(path=".", tag=IMAGE_NAME)
+    for line in logs:
+        if "stream" in line:
+            print(line["stream"], end="")
+    print(f"\n✅ Image '{IMAGE_NAME}' built successfully!")
+
+# 检查容器是否存在
+try:
+    container = client.containers.get(CONTAINER_NAME)
+    if container.status != "running":
+        print(f"🔄 Container '{CONTAINER_NAME}' exists but not running. Restarting...")
+        container.start()
+    else:
+        print(f"✅ Container '{CONTAINER_NAME}' is already running.")
+except docker.errors.NotFound:
+    print(f"🚀 Starting new container '{CONTAINER_NAME}'...")
+    try:
+        # # 可选：设置挂载目录用于持久化
+        # volumes = {
+        #     # os.path.abspath("./vncdata"): {'bind': '/root/.vnc', 'mode': 'rw'}
+        # }
+
+        container = client.containers.run(
+            IMAGE_NAME,
+            name=CONTAINER_NAME,
+            ports={
+                "80/tcp": 6080,
+                "5900/tcp": 5900,
+            },
+            detach=True,
+            tty=True,
+            device_requests=[
+                docker.types.DeviceRequest(count=-1, capabilities=[["gpu"]])
+            ],
+        )
+        print(f"✅ Container '{CONTAINER_NAME}' is now running.")
+    except docker.errors.APIError as e:
+        print(f"❌ Error starting container: {e}")
+
+print("👉 Access VNC via browser: http://localhost:6080/vnc.html (Password: 123456)")

packages/dbgpt-sandbox/src/docker_images/create_docker_brower.py

@@ -0,0 +1,214 @@
+import docker
+
+# 镜像与容器名配置
+IMAGE_NAME = "vnc-gui-browser"
+CONTAINER_NAME = "gui-browser"
+
+# Dockerfile 内容
+DOCKERFILE_CONTENT = """
+FROM ubuntu:20.04
+
+ENV DEBIAN_FRONTEND=noninteractive
+ENV LANG=zh_CN.UTF-8
+ENV LANGUAGE=zh_CN:zh
+ENV LC_ALL=zh_CN.UTF-8
+ENV DISPLAY=:0
+
+# 安装基础依赖、桌面环境、VNC/noVNC、浏览器和中文字体
+RUN apt-get update && apt-get install -y \
+    python3 python3-pip xvfb x11vnc fluxbox xterm wget dos2unix \
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+RUN apt-get update && apt-get install -y \
+    novnc websockify net-tools curl unzip firefox \
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+RUN apt-get update && apt-get install -y \
+    fonts-noto-cjk fonts-wqy-zenhei fonts-wqy-microhei language-pack-zh-hans \
+    libgtk-3-0 libdbus-glib-1-2 \
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+# 安装 geckodriver
+RUN GECKODRIVER_VERSION=$(\
+    curl -s https://api.github.com/repos/mozilla/geckodriver/releases/latest \
+| grep "tag_name" | cut -d '"' -f 4) \
+    && wget -q https://github.com/mozilla/geckodriver/releases/download/\
+    $GECKODRIVER_VERSION/geckodriver-$GECKODRIVER_VERSION-linux64.tar.gz \
+    && tar -xzf geckodriver-*.tar.gz -C /usr/local/bin \
+    && rm geckodriver-*.tar.gz
+
+# 安装 Selenium
+RUN pip3 install selenium
+
+# 设置 VNC 密码
+RUN mkdir -p /root/.vnc && \
+    x11vnc -storepasswd 123456 /root/.vnc/passwd
+
+# 拷贝启动脚本和 Selenium 示例
+COPY startup.sh /startup.sh
+COPY demo_selenium.py /demo_selenium.py
+RUN chmod +x /startup.sh && dos2unix /startup.sh
+
+EXPOSE 80 5900
+
+CMD ["/startup.sh"]
+"""
+
+# 启动脚本内容
+STARTUP_SCRIPT = """#!/bin/bash
+set -e
+
+# -----------------------------
+# 函数：查找未被占用的 DISPLAY
+# -----------------------------
+find_free_display() {
+    for i in $(seq 0 99); do
+        if [ ! -e "/tmp/.X${i}-lock" ]; then
+            echo ":$i"
+            return
+        fi
+    done
+    echo ":1"
+}
+
+# -----------------------------
+# 清理遗留锁文件
+# -----------------------------
+cleanup_locks() {
+    echo "[INFO] Cleaning up old X lock files..."
+    rm -f /tmp/.X*-lock || true
+}
+
+# -----------------------------
+# 启动服务
+# -----------------------------
+start_services() {
+    DISPLAY_ID=$(find_free_display)
+    export DISPLAY=$DISPLAY_ID
+    echo "[INFO] Using DISPLAY=$DISPLAY"
+
+    echo "[INFO] Starting Xvfb..."
+    Xvfb $DISPLAY -screen 0 1280x960x24 &
+
+    sleep 1  # 等待 Xvfb 初始化
+
+    echo "[INFO] Starting window manager..."
+    fluxbox &
+
+    echo "[INFO] Starting x11vnc..."
+    x11vnc -display $DISPLAY -forever -shared -rfbauth \
+        /root/.vnc/passwd -rfbport 5900 -auth guess &
+
+    echo "[INFO] Starting noVNC..."
+    if [ -d /usr/share/novnc ]; then
+        exec websockify --web=/usr/share/novnc/ 80 localhost:5900
+    elif [ -d /usr/share/novnc/utils ]; then
+        exec websockify --web=/usr/share/novnc/utils/ 80 localhost:5900
+    else
+        echo "⚠️ noVNC not found, keeping container alive"
+        tail -f /dev/null
+    fi
+}
+
+# -----------------------------
+# 后台启动 Selenium 示例（可选）
+# -----------------------------
+start_selenium_demo() {
+    if [ -f /demo_selenium.py ]; then
+        python3 /demo_selenium.py &
+    fi
+}
+
+# -----------------------------
+# 主流程
+# -----------------------------
+cleanup_locks
+start_services
+start_selenium_demo
+"""
+
+# Selenium 示例脚本
+DEMO_SELENIUM = """from selenium import webdriver
+from selenium.webdriver.firefox.options import Options
+import time
+
+options = Options()
+options.headless = False   # 必须 False，这样才能在 VNC 桌面看到浏览器动作
+
+driver = webdriver.Firefox(options=options)
+
+try:
+    print("🌍 打开网页 https://www.python.org ...")
+    driver.get("https://www.python.org")
+
+    time.sleep(5)  # 等待页面加载
+
+    title = driver.title
+    print(f"✅ 网页标题: {title}")
+
+    screenshot_path = "/root/screenshot.png"
+    driver.save_screenshot(screenshot_path)
+    print(f"📸 截图已保存到 {screenshot_path}")
+
+finally:
+    driver.quit()
+"""
+
+# 写入文件
+with open("Dockerfile", "w") as f:
+    f.write(DOCKERFILE_CONTENT)
+
+with open("startup.sh", "w", encoding="utf-8") as f:
+    f.write(STARTUP_SCRIPT)
+
+with open("demo_selenium.py", "w", encoding="utf-8") as f:
+    f.write(DEMO_SELENIUM)
+
+# 获取 Docker 客户端
+client = docker.from_env()
+
+# 检查镜像是否存在
+try:
+    client.images.get(IMAGE_NAME)
+    print(f"✅ Image '{IMAGE_NAME}' already exists, skipping build.")
+except docker.errors.ImageNotFound:
+    print("🔧 Building Docker image...")
+    image, logs = client.images.build(path=".", tag=IMAGE_NAME)
+    for line in logs:
+        if "stream" in line:
+            print(line["stream"], end="")
+    print(f"\n✅ Image '{IMAGE_NAME}' built successfully!")
+
+# 检查容器是否存在
+try:
+    container = client.containers.get(CONTAINER_NAME)
+    if container.status != "running":
+        print(f"🔄 Container '{CONTAINER_NAME}' exists but not running. Restarting...")
+        container.start()
+    else:
+        print(f"✅ Container '{CONTAINER_NAME}' is already running.")
+except docker.errors.NotFound:
+    print(f"🚀 Starting new container '{CONTAINER_NAME}'...")
+    try:
+        container = client.containers.run(
+            IMAGE_NAME,
+            name=CONTAINER_NAME,
+            ports={
+                "80/tcp": 6080,
+                "5900/tcp": 5900,
+            },
+            detach=True,
+            tty=True,
+            device_requests=[
+                docker.types.DeviceRequest(count=-1, capabilities=[["gpu"]])
+            ],
+        )
+        print(f"✅ Container '{CONTAINER_NAME}' is now running.")
+    except docker.errors.APIError as e:
+        print(f"❌ Error starting container: {e}")
+
+print("👉 Access GUI via browser: http://localhost:6080/vnc.html (Password: 123456)")
+print(
+    "👉 Selenium will auto-open Firefox, go to python.org, and save screenshot \
+        at /root/screenshot.png"
+)

packages/dbgpt-sandbox/src/docker_images/create_docker_python.py

@@ -0,0 +1,100 @@
+import docker
+
+# 镜像与容器名配置
+IMAGE_NAME = "python-vnc"
+CONTAINER_NAME = "pythonvnc"
+
+# Dockerfile 内容（基于 python:3.11-slim）
+DOCKERFILE_CONTENT = """
+FROM python:3.11-slim
+
+ENV DEBIAN_FRONTEND=noninteractive
+
+# 安装 VNC、Xvfb、noVNC 和桌面环境
+RUN apt-get update && apt-get install -y --no-install-recommends \\
+    x11vnc \\
+    xvfb \\
+    fluxbox \\
+    novnc \\
+    websockify \\
+    wget \\
+    net-tools \\
+    xterm \\
+    dos2unix \\
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+# 设置 VNC 密码
+RUN mkdir -p /root/.vnc && \\
+    x11vnc -storepasswd 123456 /root/.vnc/passwd
+
+# 拷贝启动脚本
+COPY startup.sh /startup.sh
+RUN chmod +x /startup.sh && dos2unix /startup.sh
+
+EXPOSE 80 5900
+
+CMD ["/startup.sh"]
+"""
+
+# 启动脚本内容
+STARTUP_SCRIPT = """#!/bin/bash
+
+Xvfb :0 -screen 0 1280x960x24 -listen tcp -ac +extension GLX +extension RENDER &
+
+export DISPLAY=:0
+
+fluxbox &
+
+x11vnc -display :0 -forever -shared -rfbauth /root/.vnc/passwd -rfbport 5900 &
+
+websockify --web=/usr/share/novnc 80 localhost:5900
+"""
+
+# 写入 Dockerfile 和 startup.sh
+with open("Dockerfile", "w") as f:
+    f.write(DOCKERFILE_CONTENT)
+
+with open("startup.sh", "w") as f:
+    f.write(STARTUP_SCRIPT)
+
+# 获取 Docker 客户端
+client = docker.from_env()
+
+# 检查镜像是否存在
+try:
+    client.images.get(IMAGE_NAME)
+    print(f"✅ Image '{IMAGE_NAME}' already exists, skipping build.")
+except docker.errors.ImageNotFound:
+    print("🔧 Building Docker image...")
+    image, logs = client.images.build(path=".", tag=IMAGE_NAME)
+    for line in logs:
+        if "stream" in line:
+            print(line["stream"], end="")
+    print(f"\n✅ Image '{IMAGE_NAME}' built successfully!")
+
+# 检查容器是否存在
+try:
+    container = client.containers.get(CONTAINER_NAME)
+    if container.status != "running":
+        print(f"🔄 Container '{CONTAINER_NAME}' exists but not running. Restarting...")
+        container.start()
+    else:
+        print(f"✅ Container '{CONTAINER_NAME}' is already running.")
+except docker.errors.NotFound:
+    print(f"🚀 Starting new container '{CONTAINER_NAME}'...")
+    try:
+        container = client.containers.run(
+            IMAGE_NAME,
+            name=CONTAINER_NAME,
+            ports={
+                "80/tcp": 6080,
+                "5900/tcp": 5900,
+            },
+            detach=True,
+            tty=True,
+        )
+        print(f"✅ Container '{CONTAINER_NAME}' is now running.")
+    except docker.errors.APIError as e:
+        print(f"❌ Error starting container: {e}")
+
+print("👉 Access VNC via browser: http://localhost:6080/vnc.html (Password: 123456)")

packages/dbgpt-sandbox/src/docker_images/create_docker_python_gpu.py

@@ -0,0 +1,108 @@
+import docker
+
+# 镜像与容器名配置
+IMAGE_NAME = "python-vnc-gpu"
+CONTAINER_NAME = "vnc-gpu-demo"
+
+# Dockerfile 内容
+DOCKERFILE_CONTENT = """
+FROM python:3.11-slim
+
+ENV DEBIAN_FRONTEND=noninteractive
+
+# 安装系统依赖和 VNC/图形组件
+RUN apt-get update && apt-get install -y --no-install-recommends \\
+    x11vnc \\
+    xvfb \\
+    fluxbox \\
+    novnc \\
+    websockify \\
+    wget \\
+    net-tools \\
+    xterm \\
+    dos2unix \\
+    pciutils \\
+    gnupg2 \\
+    && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+# 安装 NVIDIA GPU 工具（如 nvidia-smi）
+RUN apt-get update && apt-get install -y nvidia-utils-525 || true
+
+# 设置 VNC 密码
+RUN mkdir -p /root/.vnc && \\
+    x11vnc -storepasswd 123456 /root/.vnc/passwd
+
+# 拷贝启动脚本
+COPY startup.sh /startup.sh
+RUN chmod +x /startup.sh && dos2unix /startup.sh
+
+EXPOSE 80 5900
+
+CMD ["/startup.sh"]
+"""
+
+# 启动脚本内容
+STARTUP_SCRIPT = """#!/bin/bash
+
+Xvfb :0 -screen 0 1280x960x24 -listen tcp -ac +extension GLX +extension RENDER &
+
+export DISPLAY=:0
+
+fluxbox &
+
+x11vnc -display :0 -forever -shared -rfbauth /root/.vnc/passwd -rfbport 5900 &
+
+websockify --web=/usr/share/novnc 80 localhost:5900
+"""
+
+# 写入 Dockerfile 和 startup.sh
+with open("Dockerfile", "w") as f:
+    f.write(DOCKERFILE_CONTENT)
+
+with open("startup.sh", "w") as f:
+    f.write(STARTUP_SCRIPT)
+
+# 获取 Docker 客户端
+client = docker.from_env()
+
+# 构建镜像
+try:
+    client.images.get(IMAGE_NAME)
+    print(f"✅ Image '{IMAGE_NAME}' already exists, skipping build.")
+except docker.errors.ImageNotFound:
+    print("🔧 Building Docker image...")
+    image, logs = client.images.build(path=".", tag=IMAGE_NAME)
+    for line in logs:
+        if "stream" in line:
+            print(line["stream"], end="")
+    print(f"\n✅ Image '{IMAGE_NAME}' built successfully!")
+
+# 启动容器
+try:
+    container = client.containers.get(CONTAINER_NAME)
+    if container.status != "running":
+        print(f"🔄 Container '{CONTAINER_NAME}' exists but not running. Restarting...")
+        container.start()
+    else:
+        print(f"✅ Container '{CONTAINER_NAME}' is already running.")
+except docker.errors.NotFound:
+    print(f"🚀 Starting new container '{CONTAINER_NAME}'...")
+    try:
+        container = client.containers.run(
+            IMAGE_NAME,
+            name=CONTAINER_NAME,
+            ports={
+                "80/tcp": 6080,
+                "5900/tcp": 5900,
+            },
+            detach=True,
+            tty=True,
+            device_requests=[
+                docker.types.DeviceRequest(count=-1, capabilities=[["gpu"]])
+            ],
+        )
+        print(f"✅ Container '{CONTAINER_NAME}' is now running.")
+    except docker.errors.APIError as e:
+        print(f"❌ Error starting container: {e}")
+
+print("👉 Access GUI via browser: http://localhost:6080/vnc.html (Password: 123456)")

packages/dbgpt-sandbox/src/docs/architecture.md

@@ -0,0 +1,44 @@
+# DB-GPT Sandbox Agent 架构
+
+本项目实现了一个可扩展的多容器/本地沙箱执行框架，支持 Python、Shell、Node.js 等多语言代码的有状态执行，统一接口、插件化扩展、依赖安装、环境变更等能力。
+
+- 多运行时：Docker、Podman、Nerdctl、本地进程（Local）。
+- 统一抽象：会话生命周期、代码执行、状态查询、依赖安装（可选）。
+- 有状态：同一会话内多次执行共享环境，安装的依赖后续可用。
+- 自动选择：通过 RuntimeFactory 自动按优先级选择最佳运行时，或通过环境变量/参数强制指定。
+- 插件化：新增语言/依赖管理/沙箱类型时，仅需最小改动，接口清晰。
+
+## 分层设计
+
+- sandbox/execution_layer（执行层）
+  - base.py：统一抽象（SandboxRuntime、SandboxSession、ExecutionResult、SessionConfig）
+  - docker_runtime.py / podman_runtime.py / nerdctl_runtime.py / local_runtime.py：具体运行时实现
+  - runtime_factory.py：自动选择 Docker → Podman → Nerdctl → Local
+  - utils.py：资源、路径、进程、安全、环境检测工具
+- sandbox/control_layer（控制层）
+  - control_layer.py：跨任务会话管理、依赖安装、执行调度、状态查询
+- sandbox/display_layer（显示层）
+  - display_layer.py：DisplayResult 用于容器型运行时的结果封装（包含 GUI / 文件 等）
+- sandbox/user_layer（用户层）
+  - service.py/schemas.py：对外 API 统一调度，面向产品接口
+
+注意：LocalRuntime.execute 返回 ExecutionResult；容器运行时返回 DisplayResult。若需在控制层统一结果，可在控制层进行适配（将 DisplayResult 映射为 ExecutionResult 或在 API 层做多态支持）。
+
+## 会话与有状态依赖
+
+- 会话（SandboxSession）在 start 后进入活跃状态（is_active=True），同一会话内：
+  - 多次 execute 共享同一环境/容器实例
+- 运行时负责具体的依赖安装策略：
+  - Python：pip install --no-input --disable-pip-version-check
+  - JavaScript：npm init -y + npm install 包
+
+## 支持语言
+
+- Docker/Podman/Nerdctl：python、python-vnc、javascript、java、cpp、go、rust
+- Local：运行时探测系统可用语言，至少保证 python
+
+## 安全与资源
+
+- 超时、内存限制（容器通过参数、Local 通过 psutil 监控）
+- 安全检查（SecurityUtils.validate_code）对常见危险操作进行告警
+- 网络可禁用（network_disabled），适配安全隔离场景

packages/dbgpt-sandbox/src/docs/interfaces.md

@@ -0,0 +1,44 @@
+# 接口说明
+
+本文档描述了 dbgpt-sandbox-agent 的主要接口设计与使用说明。
+
+## 会话管理接口
+- `create_session(user_id: str, task_id: str, image_type: str) -> str`
+  - 创建一个新的会话，返回 session_id。
+  - 参数：
+    - user_id: 用户标识
+    - task_id: 任务标识
+    - image_type: 运行时类型（如 python、javascript）
+
+- `configure_session(session_id: str, config_info: dict) -> bool`
+  - 配置会话参数，如依赖、资源限制等。
+  - 参数：
+    - session_id: 会话标识
+    - config_info: 配置字典，支持 keys:
+      - language: 语言（默认 python）
+      - dependencies: 依赖列表（仅容器运行时支持自动安装 pip/npm）
+      - max_memory: 如 "512m"（容器）
+      - max_cpus: 整数
+      - network_disabled: 是否禁网（默认 False）
+      - env: 环境变量字典
+
+- `execute_code(session_id: str, code_type: str, code_content: str) -> dict`
+  - 在指定会话中执行代码，返回执行结果。
+  - 参数：
+    - session_id: 会话标识
+    - code_type: 代码类型（如 python、javascript）
+    - code_content: 代码内容字符串
+  - 返回值：包含输出、错误、耗时、退出码等信息的字典
+
+- `get_session_status(session_id: str) -> dict`
+  - 查询会话当前状态与资源使用情况。
+  - 参数：
+    - session_id: 会话标识
+  - 返回值：包含状态、内存、CPU 使用等信息的字典
+
+- `disconnect_session(user_id: str, task_id: str) -> bool`  
+  - 停止并销毁指定用户任务的会话。
+  - 参数：
+    - user_id: 用户标识
+    - task_id: 任务标识
+  - 返回值：操作是否成功  

packages/dbgpt-sandbox/src/docs/usage.md

@@ -0,0 +1,20 @@
+# 使用说明
+
+## 环境准备
+- Python 3.8+
+- Docker / Podman / Nerdctl（可选，但推荐至少安装一个容器后端）
+- Windows 用户建议使用 WSL2 + Ubuntu 20.04+（WSL2 支持 Docker Desktop 或 Podman）
+- Linux 用户建议安装 Docker 或 Podman
+- Mac 用户建议安装 Docker Desktop 或 Podman
+- 本地运行时不做容器隔离，适合无容器环境的回退/开发调试
+
+## 自动选择运行时
+由 `sandbox/execution_layer/runtime_factory.py` 控制：
+- 优先级：Docker → Podman → Nerdctl → Local
+- 可通过config.py中的 SANDBOX_RUNTIME 常量强制指定运行时
+
+## 启动 API 服务
+- Linux / Mac
+```bash
+./scripts/start_api.sh             # 自动创建 .venv 并安装依赖后启动
+```

packages/dbgpt-sandbox/src/scripts/start_api.cmd

@@ -0,0 +1,61 @@
+@echo off
+setlocal enabledelayedexpansion
+
+REM One-click start for Sandbox User API (Windows cmd.exe)
+REM Usage: scripts\start_api.cmd [docker|podman|nerdctl|local]
+
+set "SCRIPT_DIR=%~dp0"
+set "ROOT=%SCRIPT_DIR%.."
+pushd "%ROOT%" >nul
+
+echo [INFO] Project root: %CD%
+
+REM Optional: runtime preference from arg1
+if "%~1"=="" (
+  set "SANDBOX_RUNTIME="
+) else (
+  set "SANDBOX_RUNTIME=%~1"
+  echo [INFO] Prefer runtime: %SANDBOX_RUNTIME%
+)
+
+where python >nul 2>nul
+if errorlevel 1 (
+  echo [ERROR] Python not found in PATH. Please install Python 3.10+.
+  popd >nul
+  exit /b 1
+)
+
+if not exist ".venv\Scripts\python.exe" (
+  echo [INFO] Creating virtual environment: .venv
+  python -m venv .venv
+  if errorlevel 1 (
+    echo [ERROR] Failed to create venv.
+    popd >nul
+    exit /b 1
+  )
+)
+
+echo [INFO] Upgrading pip and installing requirements...
+".venv\Scripts\python.exe" -m pip install --upgrade pip
+if errorlevel 1 (
+  echo [WARN] pip upgrade failed, continue...
+)
+".venv\Scripts\python.exe" -m pip install -r requirements.txt
+if errorlevel 1 (
+  echo [ERROR] pip install failed.
+  popd >nul
+  exit /b 1
+)
+
+echo [INFO] Starting API server at http://127.0.0.1:8000 ...
+if not "%SANDBOX_RUNTIME%"=="" (
+  set "SANDBOX_RUNTIME=%SANDBOX_RUNTIME%"
+)
+REM 先进入 sandbox 目录再运行
+pushd sandbox >nul
+".venv\Scripts\python.exe" -m uvicorn user_layer.service:app --host 127.0.0.1 --port 8000 --reload
+set "EXITCODE=%ERRORLEVEL%"
+popd >nul
+
+popd >nul
+exit /b %EXITCODE%

packages/dbgpt-sandbox/src/scripts/start_api.ps1

@@ -0,0 +1,45 @@
+# One-click start for Sandbox User API (PowerShell)
+# Usage: .\scripts\start_api.ps1 -Runtime docker|podman|nerdctl|local
+param(
+  [string]$Runtime = ""
+)
+
+$ErrorActionPreference = "Stop"
+$scriptDir = Split-Path -Parent $MyInvocation.MyCommand.Path
+$root = Join-Path $scriptDir ".."
+Push-Location $root
+
+Write-Host "[INFO] Project root: $PWD"
+
+if ($Runtime) {
+  $env:SANDBOX_RUNTIME = $Runtime
+  Write-Host "[INFO] Prefer runtime: $env:SANDBOX_RUNTIME"
+}
+
+# Ensure Python
+if (-not (Get-Command python -ErrorAction SilentlyContinue)) {
+  Write-Error "Python not found in PATH. Please install Python 3.10+."
+  Pop-Location
+  exit 1
+}
+
+# Create venv if missing
+if (-not (Test-Path ".venv/Scripts/python.exe")) {
+  Write-Host "[INFO] Creating virtual environment: .venv"
+  python -m venv .venv
+}
+
+# Install requirements
+Write-Host "[INFO] Installing requirements..."
+& .venv/Scripts/python.exe -m pip install --upgrade pip
+& .venv/Scripts/python.exe -m pip install -r requirements.txt
+
+# Start server
+Write-Host "[INFO] Starting API server at http://127.0.0.1:8000 ..."
+Push-Location sandbox
+& .venv/Scripts/python.exe -m uvicorn user_layer.service:app --host 127.0.0.1 --port 8000 --reload
+$code = $LASTEXITCODE
+Pop-Location
+
+Pop-Location
+exit $code

packages/dbgpt-sandbox/src/scripts/start_api.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+# One-click start for Sandbox User API (Linux/macOS)
+# Usage:
+#   ./scripts/start_api.sh                 # 自动创建 .venv 并安装依赖后启动
+#   ./scripts/start_api.sh docker          # 强制选择 Docker 作为运行时
+#   SANDBOX_RUNTIME=local ./scripts/start_api.sh  # 通过环境变量指定
+
+set -Eeuo pipefail
+
+RUNTIME_ARG="${1:-}"
+SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
+ROOT_DIR="$(cd -- "${SCRIPT_DIR}/.." && pwd)"
+cd "$ROOT_DIR"
+
+echo "[INFO] Project root: $PWD"
+
+# Prefer runtime via first argument (optional)
+if [[ -n "$RUNTIME_ARG" ]]; then
+  export SANDBOX_RUNTIME="$RUNTIME_ARG"
+  echo "[INFO] Prefer runtime: $SANDBOX_RUNTIME"
+fi
+
+# Detect Python 3
+PY_BIN="$(command -v python3 || true)"
+if [[ -z "$PY_BIN" ]]; then
+  PY_BIN="$(command -v python || true)"
+fi
+if [[ -z "$PY_BIN" ]]; then
+  echo "[ERROR] Python 3 not found in PATH. Please install Python 3.10+." >&2
+  exit 1
+fi
+
+# Ensure venv
+VENV_PY=".venv/bin/python"
+if [[ ! -x "$VENV_PY" ]]; then
+  echo "[INFO] Creating virtual environment: .venv"
+  "$PY_BIN" -m venv .venv
+fi
+
+# Install requirements
+echo "[INFO] Installing requirements..."
+"$VENV_PY" -m pip install --upgrade pip
+"$VENV_PY" -m pip install -r requirements.txt
+
+# Start server
+echo "[INFO] Starting API server at http://127.0.0.1:8000 ..."
+cd sandbox
+exec "$VENV_PY" -m uvicorn user_layer.service:app --host 127.0.0.1 --port 8000 --reload

packages/dbgpt-sandbox/test_sandbox_integration.py

@@ -0,0 +1,273 @@
+#!/usr/bin/env python3
+"""
+DB-GPT Sandbox Integration Test Script
+Test code execution functionality of sandbox service
+"""
+
+import asyncio
+import logging
+import sys
+from typing import Any, Dict
+
+import requests
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+class SandboxTester:
+    """Sandbox test class"""
+
+    def __init__(self, base_url: str = "http://localhost:8000"):
+        self.base_url = base_url
+        self.session = requests.Session()
+
+    def health_check(self) -> bool:
+        """Health check"""
+        try:
+            response = self.session.get(f"{self.base_url}/api/health", timeout=5)
+            return response.status_code == 200 and response.json().get("status") == "ok"
+        except Exception as e:
+            logger.error(f"Health check failed: {e}")
+            return False
+
+    def get_methods(self) -> Dict[str, Any]:
+        """Get available methods"""
+        try:
+            response = self.session.get(f"{self.base_url}/api/methods", timeout=5)
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Failed to get available methods: {e}")
+            return {}
+
+    def connect(
+        self, user_id: str, task_id: str, image_type: str = "python"
+    ) -> Dict[str, Any]:
+        """Establish sandbox session"""
+        try:
+            response = self.session.post(
+                f"{self.base_url}/api/connect",
+                json={
+                    "user_id": user_id,
+                    "task_id": task_id,
+                    "image_type": image_type,
+                },
+                timeout=60,
+            )
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Connection failed: {e}")
+            return {"status": "error", "error": str(e)}
+
+    def execute_code(
+        self, session_id: str, code_type: str, code_content: str
+    ) -> Dict[str, Any]:
+        """Execute code"""
+        try:
+            response = self.session.post(
+                f"{self.base_url}/api/execute",
+                json={
+                    "session_id": session_id,
+                    "code_type": code_type,
+                    "code_content": code_content,
+                },
+                timeout=60,
+            )
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Code execution failed: {e}")
+            return {"status": "error", "error": str(e)}
+
+    def get_status(self, session_id: str) -> Dict[str, Any]:
+        """Get execution status"""
+        try:
+            response = self.session.post(
+                f"{self.base_url}/api/status",
+                json={"session_id": session_id},
+                timeout=5,
+            )
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Failed to get status: {e}")
+            return {"status": "error", "error": str(e)}
+
+    def disconnect(self, user_id: str, task_id: str) -> Dict[str, Any]:
+        """Disconnect session"""
+        try:
+            response = self.session.post(
+                f"{self.base_url}/api/disconnect",
+                json={
+                    "user_id": user_id,
+                    "task_id": task_id,
+                },
+                timeout=10,
+            )
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Disconnection failed: {e}")
+            return {"status": "error", "error": str(e)}
+
+    def list_sessions(self) -> Dict[str, Any]:
+        """List all active sessions"""
+        try:
+            response = self.session.get(f"{self.base_url}/api/sessions", timeout=5)
+            response.raise_for_status()
+            return response.json()
+        except Exception as e:
+            logger.error(f"Failed to list sessions: {e}")
+            return {"sessions": []}
+
+
+async def run_integration_tests():
+    """Run integration tests"""
+    tester = SandboxTester()
+
+    try:
+        logger.info("=== Starting Integration Tests ===")
+
+        # Health check
+        logger.info("1. Health check")
+        if not tester.health_check():
+            logger.error(
+                "Health check failed, please ensure sandbox service is running"
+            )
+            logger.error(
+                "Run command: SANDBOX_RUNTIME=local uv run --no-sync dbgpt-sandbox"
+            )
+            return False
+        logger.info("✓ Health check passed")
+
+        # Get available methods
+        logger.info("2. Get available methods")
+        methods = tester.get_methods()
+        if not methods:
+            logger.error("Failed to get available methods")
+            return False
+        logger.info(f"✓ Available methods: {len(methods.get('methods', []))}")
+
+        # Test connection
+        logger.info("3. Test connection")
+        import time
+
+        user_id = "test_user"
+        task_id = f"test_task_{int(time.time())}"
+        connect_result = tester.connect(user_id, task_id, "python")
+
+        if connect_result.get("status") != "success":
+            logger.error(f"Connection failed: {connect_result}")
+            return False
+        logger.info("✓ Connection successful")
+
+        session_id = f"{user_id}_{task_id}"
+
+        # Test simple Python code execution
+        logger.info("4. Test simple Python code execution")
+        simple_code = "print('Hello from sandbox!')"
+        execute_result = tester.execute_code(session_id, "python", simple_code)
+
+        if execute_result.get("status") != "success":
+            logger.error(f"Code execution failed: {execute_result}")
+            return False
+        logger.info(
+            f"✓ Simple code execution successful: {execute_result.get('output')}"
+        )
+
+        # Test mathematical calculations
+        logger.info("5. Test mathematical calculations")
+        math_code = """
+import math
+result = math.sqrt(16)
+print(f"sqrt(16) = {result}")
+print(f"2 + 3 * 4 = {2 + 3 * 4}")
+"""
+        math_result = tester.execute_code(session_id, "python", math_code)
+
+        if math_result.get("status") != "success":
+            logger.error(f"Math calculation failed: {math_result}")
+            return False
+        logger.info(f"✓ Math calculation successful: {math_result.get('output')}")
+
+        # Test string operations
+        logger.info("6. Test string operations")
+        string_code = """
+text = "Hello from sandbox!"
+reversed_text = text[::-1]
+print(f"Original text: {text}")
+print(f"Reversed text: {reversed_text}")
+print(f"Text length: {len(text)}")
+"""
+        string_result = tester.execute_code(session_id, "python", string_code)
+
+        if string_result.get("status") != "success":
+            logger.error(f"String operation failed: {string_result}")
+            return False
+        logger.info(f"✓ String operation successful: {string_result.get('output')}")
+
+        # Test error handling
+        logger.info("7. Test error handling")
+        error_code = """
+# Intentionally trigger an error
+result = 1 / 0
+"""
+        error_result = tester.execute_code(session_id, "python", error_code)
+
+        # This should return error status, which is expected
+        if error_result.get("status") == "success":
+            logger.warning("Error code unexpectedly executed successfully")
+        else:
+            logger.info(
+                f"✓ Error handling working properly: {error_result.get('error')}"
+            )
+
+        # Get session status
+        logger.info("8. Get session status")
+        status_result = tester.get_status(session_id)
+        logger.info(f"✓ Session status: {status_result}")
+
+        # List all sessions
+        logger.info("9. List all sessions")
+        sessions_result = tester.list_sessions()
+        logger.info(f"✓ Active sessions: {sessions_result}")
+
+        # Disconnect
+        logger.info("10. Disconnect")
+        disconnect_result = tester.disconnect(user_id, task_id)
+
+        if disconnect_result.get("status") != "success":
+            logger.error(f"Disconnection failed: {disconnect_result}")
+            return False
+        logger.info("✓ Disconnection successful")
+
+        logger.info("=== All Integration Tests Passed! ===")
+        return True
+
+    except Exception as e:
+        logger.error(f"Error occurred during integration testing: {e}")
+        return False
+
+
+def main():
+    """Main function"""
+    try:
+        # Run async tests
+        success = asyncio.run(run_integration_tests())
+        sys.exit(0 if success else 1)
+
+    except KeyboardInterrupt:
+        logger.info("Test interrupted by user")
+        sys.exit(1)
+    except Exception as e:
+        logger.error(f"Test execution failed: {e}")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

pyproject.toml

@@ -20,6 +20,7 @@ dbgpt-serve = { workspace = true }
 dbgpt-app = { workspace = true }
 dbgpt-acc-auto = { workspace = true }
 dbgpt-acc-flash-attn = { workspace = true }
+dbgpt-sandbox = { workspace = true }
 
 [[tool.uv.index]]
 name = "testpypi"
@@ -34,6 +35,7 @@ members = [
   "packages/dbgpt-core",
   "packages/dbgpt-ext",
   "packages/dbgpt-serve",
+  "packages/dbgpt-sandbox",
   "packages/dbgpt-accelerator/dbgpt-acc*",
 ]
 
@@ -80,4 +82,4 @@ select = ["E", "F", "I"]
 
 [tool.ruff.lint.isort]
 # Specify the local modules (first-party)
-known-first-party = ["dbgpt", "dbgpt_acc_auto", "dbgpt_client", "dbgpt_ext", "dbgpt_serve", "dbgpt_app"]
+known-first-party = ["dbgpt", "dbgpt_acc_auto", "dbgpt_client", "dbgpt_ext", "dbgpt_serve", "dbgpt_app", "dbgpt_sandbox"]