一、背景：Agent时代AI代码执行的双刃剑：

随着人工智能的不断发展，大语言模型技术日益成熟。AI应用已经从最初专注于语音识别等简单任务的应用，演进到能够进行自主推理决策，完成复杂任务的Agent。基于大语言模型的AI Agent，具备自主性、目标导向性和交互性，能够在复杂多变的环境中感知信息、推理决策并执行任务。在各个行业场景中，AI Agent都展示出巨大的应用潜力，广泛应用于客户支持、销售拓展、医疗诊断和金融分析等场景。

openJiuwen作为开源Agent平台，致力于提供灵活、强大且易用的AI Agent开发与运行能力。基于该平台，开发者可快速构建处理各类简单或复杂任务的AI Agent，实现多Agent协同交互，高效开发生产级可靠AI Agent；并助力企业与个人快速搭建AI Agent系统或平台，推动商用级Agentic AI技术广泛应用与落地，它们不再是简单的聊天机器人，而是软件开发生命周期 中的积极参与者，能够编写、调试并执行代码以解决复杂问题，这种自主性带来了前所未有的生产力提升。

1.1 界线的模糊：数据与代码的界限

在默认情况下，应用程序拥有访问机器上所有资源的权限，这无疑带来了安全风险。不恰当的操作可能导致其他应用程序的资源被破坏，或者数据发生泄漏。例如，一个恶意用户可能会通过编写一段代码来破坏其他用户的文件系统或数据库。在这种情况下，传统的安全措施（如访问控制列表、防火墙等）就显得力不从心。

这种数据与代码的内在模糊性，为“提示注入”（Prompt Injection）等新型攻击打开了大门，攻击者可以通过精心构造的提示，让LLM将恶意用户输入误认为是高优先级的系统指令，从而劫持其控制流。因此，沙箱（Sandbox）技术发挥关键作用的地方。

openJiuwen的沙箱环境，核心原因是为了在‌安全隔离的前提下实现自主行为的测试与执行‌，防止其在运行过程中对真实系统、数据或物理世界造成不可控的风险。尤其是在智能体具备代码执行、工具调用、网页交互甚至控制机器人等能力时，沙箱成为必不可少的“安全护栏”。

---

1.2 沙箱的经典应用场景：

为了更好地理解沙箱的价值，我们可以看看它在现实世界中的一些经典应用：

• 恶意软件分析：安全研究人员会在沙箱中“引爆”可疑文件，观察其行为，例如是否修改注册表、尝试网络连接或下载其他恶意软件，而这一切都不会感染研究人员自己的系统。
• 浏览器安全：现代浏览器（如Chrome和Firefox）会将每个标签页运行在独立的沙箱进程中。这样，即使一个恶意网站试图利用浏览器漏洞，其影响也会被限制在该标签页内，无法危及整个计算机系统。
• 软件开发与测试：开发人员利用沙箱环境来测试新代码、新功能或API集成，确保这些变更不会对稳定的生产环境造成破坏或引入不可预见的错误。

从这些应用中可以看出，沙箱本质上是一种赋能技术。它不仅仅是阻止坏事发生，更是通过创造必要的安全条件，使得那些功能强大、富有创新性但本身具有风险的新技术得以存在和发展。如果没有浏览器沙箱，动态Web应用将寸步难行；如果没有移动操作系统的应用沙箱，整个应用商店生态系统也无法建立。同样，对于AI而言，如果没有沙箱，让一个AI智能体自主下载依赖、写入文件并执行代码的想法，在安全上是完全不可行的。

---

二、openJiuwen IDE运行时沙箱环境：

注意：下面是以docker环境中的沙箱环境为例分析。

沙箱看似是技术防护，实则是给AI行动权划边界。就像公司给员工开通系统权限，得按需分配、留痕可查。没有约束的智能体，再强大也难进企业大门，旨在解决AI智能体在执行任务时的安全性和合规性问题，Sandbox提供了一个隔离和受控的环境，确保代码执行安全且可追踪审计。

2.1 IDE运行时沙箱环境：

在智能体沙箱环境中，JavaScript 和 Python 是两种最常被支持的编程语言，主要用于安全地执行动态代码、进行数据分析、网页交互等任务。这两种语言因其广泛的应用场景和强大的功能，成为了智能体沙箱中不可或缺的组成部分。

• Python 3：适合数据处理，机器学习，科学计算，自动化脚本等场景
• Node.js：适合Web API开发，实时通信，前端构建工具，微服务等场景

• ‌Python‌：作为AI智能体开发的主流语言，Python在沙箱中被广泛支持，用于执行数值计算、数据分析、文件处理、调用机器学习库等任务。许多沙箱平台（如AgentRun、Dify）都提供专门的Python代码解释器沙箱，允许智能体安全地运行Python脚本。
  ‌
• JavaScript‌：JavaScript同样被主流沙箱平台支持，尤其在需要与网页交互或执行前端逻辑时。例如，AgentRun的Code Interpreter沙箱明确支持JavaScript执行，可用于网页抓取后的数据处理或自动化脚本编写。

root@gpc-VMware-Virtual-Platform:~# docker ps
CONTAINER ID   IMAGE                                                                                        COMMAND                   CREATED       STATUS                   PORTS                                                                                        NAMES
a259aba935fe   swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-web-amd64:0.1.4                           "/usr/local/bin/star…"   5 hours ago   Up 3 minutes (healthy)   80/tcp, 0.0.0.0:3000->3001/tcp, [::]:3000->3001/tcp                                          jiuwen-frontend-lmf80
35ba430e49fc   swr.cn-north-4.myhuaweicloud.com/openjiuwen/mysql-amd64:8.4.5                                "docker-entrypoint.s…"   5 hours ago   Up 3 minutes (healthy)   33060/tcp, 0.0.0.0:3001->3306/tcp, [::]:3001->3306/tcp                                       jiuwen-mysql-lmf80
048115662dcc   swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-server-amd64:0.1.4                        "/entrypoint.sh pyth…"   8 days ago    Up 3 minutes (healthy)   0.0.0.0:3007->8000/tcp, [::]:3007->8000/tcp                                                  jiuwen-backend-lmf80
0754aa952402   swr.cn-north-4.myhuaweicloud.com/openjiuwen/deepsearch-studio-server-amd64:0.1.0             "python start_backen…"   8 days ago    Up 3 minutes (healthy)   0.0.0.0:3012->8000/tcp, [::]:3012->8000/tcp                                                  jiuwen-deepsearch-lmf80
3b13df8d6640   swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-sandbox-gateway-amd64:0.1.4               "python -m openjiuwe…"   8 days ago    Up 3 minutes (healthy)   0.0.0.0:3009->8188/tcp, [::]:3009->8188/tcp                                                  jiuwen-sandbox-gateway-lmf80
42f8141eadf3   swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-sandbox-js-amd64:0.1.4                    "docker-entrypoint.s…"   8 days ago    Up 3 minutes (healthy)   0.0.0.0:3011->5002/tcp, [::]:3011->5002/tcp                                                  jiuwen-js-server-lmf80
813f4d48454f   swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-sandbox-py-amd64:0.1.4                    "python -m openjiuwe…"   8 days ago    Up 3 minutes (healthy)   0.0.0.0:3010->5001/tcp, [::]:3010->5001/tcp                                                  jiuwen-python-server-lmf80
0997cf0b4f48   swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-plugin-server-amd64:0.1.4                 "python -m openjiuwe…"   8 days ago    Up 3 minutes (healthy)   0.0.0.0:3008->8185/tcp, [::]:3008->8185/tcp                                                  jiuwen-plugin-server-lmf80
f54f9937c7ec   swr.cn-north-4.myhuaweicloud.com/openjiuwen/milvusdb/milvus-amd64:v2.6.2                     "/tini -- milvus run…"   8 days ago    Up 3 minutes (healthy)   0.0.0.0:3006->9091/tcp, [::]:3006->9091/tcp, 0.0.0.0:3005->19530/tcp, [::]:3005->19530/tcp   jiuwen-milvus-standalone-lmf80
af62ea67ef9e   swr.cn-north-4.myhuaweicloud.com/openjiuwen/minio/minio-amd64:RELEASE.2024-12-18T13-15-44Z   "/usr/bin/docker-ent…"   8 days ago    Up 3 minutes (healthy)   0.0.0.0:3003->9000/tcp, [::]:3003->9000/tcp, 0.0.0.0:3004->9001/tcp, [::]:3004->9001/tcp     jiuwen-minio-lmf80
009e437f5204   swr.cn-north-4.myhuaweicloud.com/openjiuwen/bitnami/etcd-amd64:v3.5.18                       "/opt/bitnami/script…"   8 days ago    Up 3 minutes (healthy)   2379-2380/tcp                                                                                jiuwen-etcd-lmf80

通过查看docker ps命令，可以看到openJiuwen的沙箱环境主要包括以下几个部分：

名称	镜像名称	镜像说明	端口
studio-web	swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-web-amd64:0.1.4	Web前端，提供用户界面	80
studio-server	swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-server-amd64:0.1.4	后端服务，提供API接口	8000
studio-plugin-server	swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-plugin-server-amd64:0.1.4	插件服务，提供插件管理功能	8185
studio-sandbox-gateway	swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-sandbox-gateway-amd64:0.1.4	沙箱网关，提供代码执行接口	8188
studio-sandbox-py	swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-sandbox-py-amd64:0.1.4	Python 沙箱，用于执行Python代码	5001
studio-sandbox-js	swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-sandbox-js-amd64:0.1.4	JavaScript 沙箱，用于执行JavaScript代码	5002
studio-mysql	swr.cn-north-4.myhuaweicloud.com/openjiuwen/mysql-amd64:8.4.5	数据库，用于存储数据	3306
studio-milvus	swr.cn-north-4.myhuaweicloud.com/openjiuwen/milvusdb/milvus-amd64:v2.6.2	分布式向量数据库，用于存储和查询向量数据	90
studio-minio	swr.cn-north-4.myhuaweicloud.com/openjiuwen/bitnami/minio-amd64:RELEASE.2024-12-18T13-15-44Z	存储服务，用于存储文件	9000
studio-etcd	swr.cn-north-4.myhuaweicloud.com/openjiuwen/bitnami/etcd-amd64:v3.5.18	分布式键值存储，用于配置管理	23

• 一、Python 沙箱，用于执行Python代码：

  • 镜像名称：swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-sandbox-py-amd64:0.1.4
  • 镜像说明：Python 沙箱，用于执行Python代码。
  • 端口：3010->5001

• 二、JavaScript 沙箱，用于执行JavaScript代码：

  • 镜像名称：swr.cn-north-4.myhuaweicloud.com/openjiuwen/studio-sandbox-js-amd64:0.1.4
  • 镜像说明：JavaScript 沙箱，用于执行JavaScript代码。
  • 端口：3011->5002

---

2.2 js沙箱环境：

接下来我们，可以使用docker exec命令，进入js沙箱环境，来查看js沙箱环境是如何工作的，在项目目录下存在一个kernel.js文件，内容如下：

// kernel.js - 支持自定义 timeout 的 JS 沙箱
const http = require("http");
const { spawn } = require("child_process");

const server = http.createServer(async (req, res) => {
  res.setHeader("Access-Control-Allow-Origin", "*");
  res.setHeader("Access-Control-Allow-Methods", "POST, GET, OPTIONS");
  res.setHeader("Access-Control-Allow-Headers", "Content-Type");

  if (req.method === "OPTIONS") {
    res.writeHead(200);
    res.end();
    return;
  }

  if (req.method === "GET" && req.url === "/health") {
    res.writeHead(200, { "Content-Type": "application/json" });
    res.end(JSON.stringify({ status: "ok" }));
    return;
  }

  if (req.method === "POST" && req.url === "/run") {
    let body = "";
    req.on("data", (chunk) => {
      body += chunk;
    });

    req.on("end", () => {
      runUserCode(body, res);
    });
  } else {
    res.writeHead(404, { "Content-Type": "application/json" });
    res.end(
      JSON.stringify({
        return: null,
        error: "Not Found",
      })
    );
  }
});

/**
 * 执行用户代码
 */
async function runUserCode(body, res) {
  try {
    const {
      code = "",
      inputs = {},
      timeout: timeoutSec = 10,
    } = JSON.parse(body);

    // 验证 timeout：必须是数字，且在合理范围内
    if (typeof timeoutSec !== "number" || timeoutSec <= 0) {
      res.writeHead(400, { "Content-Type": "application/json" });
      res.end(
        JSON.stringify({
          return: null,
          error: "'timeout' must be a number and > 0 seconds.",
        })
      );
      return;
    }

    // 转换为毫秒（spawn 的 timeout 单位是毫秒）
    const timeoutMs = Math.floor(timeoutSec * 1000);

    const fullCode = `
// 用户代码
${code}

// 沙箱包装代码
(async () => {
  const safeStringify = (obj) => {
    return JSON.stringify(obj, (key, value) =>
      typeof value === 'bigint' ? value.toString() + 'n' : value
    );
  };

  try {
    const args = new Args(${JSON.stringify(inputs)});
    let result = main(args);
    if (result && typeof result.then === 'function') {
      result = await result;
    }
    process.stdout.write(safeStringify({ 
      return: result === undefined ? null : result, 
      error: null 
    }) + '\\n');
    process.exit(0);
  } catch (e) {
    process.stdout.write(safeStringify({ 
      return: null, 
      error: e.message || String(e) 
    }) + '\\n');
    process.exit(0);
  }
})();
`;

    // 创建子进程执行代码
    const child = spawn("node", ["--eval", fullCode], {
      timeout: timeoutMs,
      stdio: ["ignore", "pipe", "pipe"],
    });

    let output = "";
    child.stdout.on("data", (data) => {
      output += data;
    });

    child.stderr.on("data", (data) => {
      output += data;
    });

    child.on("close", (code, signal) => {
      // 如果是超时 kill（signal === 'SIGTERM'），返回超时错误
      if (signal === "SIGTERM") {
        res.writeHead(200, { "Content-Type": "application/json" });
        res.end(
          JSON.stringify({
            return: null,
            error: `Execution timed out (> ${timeoutSec} seconds).`,
          })
        );
        return;
      }

      try {
        const trimmed = output.trim();
        if (!trimmed) {
          throw new Error("Empty output from sandbox");
        }
        const result = JSON.parse(trimmed);
        res.writeHead(200, { "Content-Type": "application/json" });
        res.end(JSON.stringify(result));
      } catch (parseErr) {
        res.writeHead(500, { "Content-Type": "application/json" });
        res.end(
          JSON.stringify({
            return: null,
            error:
              "Sandbox internal error: " +
              (parseErr.message || "unknown") +
              " Output: " +
              output,
          })
        );
      }
    });

    child.on("error", (err) => {
      res.writeHead(500, { "Content-Type": "application/json" });
      res.end(
        JSON.stringify({
          return: null,
          error: "Failed to spawn sandbox process: " + err.message,
        })
      );
    });
  } catch (e) {
    res.writeHead(400, { "Content-Type": "application/json" });
    res.end(
      JSON.stringify({
        return: null,
        error: "Invalid request: " + (e.message || "malformed JSON"),
      })
    );
  }
}

// 启动服务器
const PORT = process.env.PORT || 5002;
server.listen(PORT, "0.0.0.0", () => {
  console.log(`✅ JS sandbox listening on http://0.0.0.0:${PORT}`);
});

process.on("SIGTERM", () => {
  server.close(() => {
    console.log("Process terminated");
  });
});

这段代码实现了一个基于Node.js的HTTP服务器，核心功能是提供一个安全的JS代码沙箱执行环境：

• 1. 接收客户端POST请求提交的JS代码、输入参数和超时时间
• 2. 在独立的子进程中执行用户代码，防止阻塞主进程
• 3. 支持自定义执行超时时间，超时后自动终止代码执行
• 4. 捕获代码执行中的错误并返回标准化的JSON结果
• 5. 提供跨域支持和健康检查接口

---

1. 模块导入层（基础依赖）

const http = require("http");
const { spawn } = require("child_process");

• http：Node.js内置的HTTP模块，用于创建HTTP服务器
• spawn：从child_process模块导入，用于创建独立的子进程执行用户代码（核心沙箱机制）

2. HTTP服务器初始化层

const server = http.createServer(async (req, res) => {
  // 服务器核心逻辑
});

创建HTTP服务器实例，所有请求都会进入这个异步处理函数。

2.1 跨域（CORS）配置（请求预处理）

res.setHeader("Access-Control-Allow-Origin", "*");
res.setHeader("Access-Control-Allow-Methods", "POST, GET, OPTIONS");
res.setHeader("Access-Control-Allow-Headers", "Content-Type");

if (req.method === "OPTIONS") {
  res.writeHead(200);
  res.end();
  return;
}

• 配置跨域响应头，允许所有域名访问（*），支持POST/GET/OPTIONS方法
• 处理预检请求（OPTIONS）：直接返回200，避免跨域请求被浏览器拦截

2.2 健康检查接口（/health）

if (req.method === "GET" && req.url === "/health") {
  res.writeHead(200, { "Content-Type": "application/json" });
  res.end(JSON.stringify({ status: "ok" }));
  return;
}

• 提供GET类型的健康检查接口，用于监控服务状态，返回{status: "ok"}

2.3 核心业务接口（/run）

if (req.method === "POST" && req.url === "/run") {
  let body = "";
  req.on("data", (chunk) => {
    body += chunk;
  });

  req.on("end", () => {
    runUserCode(body, res);
  });
}

• 处理POST类型的/run接口（核心执行入口）
• 流式接收请求体（Node.js HTTP请求体默认是流），拼接完整请求数据
• 请求体接收完成后，调用runUserCode函数处理代码执行逻辑

2.4 404处理

} else {
  res.writeHead(404, { "Content-Type": "application/json" });
  res.end(
    JSON.stringify({
      return: null,
      error: "Not Found",
    })
  );
}

• 非预期的请求方法/路径，返回404错误，格式为标准化的JSON

3. 核心业务逻辑层（runUserCode函数）

这是整个沙箱的核心函数，负责解析请求、构建执行环境、启动子进程、处理执行结果。

3.1 异常捕获与请求参数解析

try {
  const {
    code = "",
    inputs = {},
    timeout: timeoutSec = 10,
  } = JSON.parse(body);

  // 验证 timeout：必须是数字，且在合理范围内
  if (typeof timeoutSec !== "number" || timeoutSec <= 0) {
    // 返回400错误
    return;
  }

  const timeoutMs = Math.floor(timeoutSec * 1000); // 转换为毫秒
} catch (e) {
  // 解析请求体失败（如JSON格式错误），返回400错误
}

• 解析POST请求体的JSON数据，提取核心参数：
  • code：用户提交的待执行JS代码
  • inputs：用户代码的输入参数
  • timeout：自定义超时时间（秒），默认10秒
• 验证超时参数的合法性，非法则返回400错误
• 将秒转换为毫秒（子进程timeout参数单位是毫秒）

3.2 沙箱代码构建（核心安全机制）

const fullCode = `
// 用户代码
${code}

// 沙箱包装代码
(async () => {
  const safeStringify = (obj) => {
    return JSON.stringify(obj, (key, value) =>
      typeof value === 'bigint' ? value.toString() + 'n' : value
    );
  };

  try {
    const args = new Args(${JSON.stringify(inputs)});
    let result = main(args);
    if (result && typeof result.then === 'function') {
      result = await result;
    }
    process.stdout.write(safeStringify({ 
      return: result === undefined ? null : result, 
      error: null 
    }) + '\\n');
    process.exit(0);
  } catch (e) {
    process.stdout.write(safeStringify({ 
      return: null, 
      error: e.message || String(e) 
    }) + '\\n');
    process.exit(0);
  }
})();
`;

• 代码拼接：将用户代码与沙箱包装代码合并，形成完整的执行代码
• 安全序列化：safeStringify处理BigInt等JSON不支持的类型，避免序列化失败
• 执行逻辑：
  1. 创建Args实例（用户代码预期的入参载体）
  2. 调用用户代码的main函数（核心执行入口）
  3. 处理异步结果：如果main返回Promise，等待其执行完成
  4. 标准化输出：将执行结果/错误通过stdout输出（子进程与主进程的通信方式）
  5. 强制退出：执行完成后调用process.exit(0)，避免子进程残留

3.3 子进程创建与执行

const child = spawn("node", ["--eval", fullCode], {
  timeout: timeoutMs,
  stdio: ["ignore", "pipe", "pipe"],
});

• spawn：启动独立的Node.js子进程
• 参数说明：
  • "node"：执行器（Node.js可执行文件）
  • "--eval"：执行字符串形式的代码
  • fullCode：待执行的完整代码
• 配置项：
  • timeout：子进程超时时间（毫秒）
  • stdio：标准输入忽略，标准输出/错误通过管道传递（主进程可捕获）

3.4 子进程输出捕获

let output = "";
child.stdout.on("data", (data) => {
  output += data;
});

child.stderr.on("data", (data) => {
  output += data;
});

• 捕获子进程的标准输出（stdout）和标准错误（stderr）
• 拼接所有输出数据，用于后续解析执行结果

3.5 子进程退出处理（核心结果处理）

child.on("close", (code, signal) => {
  // 1. 超时处理（signal === 'SIGTERM'）
  if (signal === "SIGTERM") {
    // 返回超时错误
    return;
  }

  // 2. 正常退出处理
  try {
    const trimmed = output.trim();
    if (!trimmed) {
      throw new Error("Empty output from sandbox");
    }
    const result = JSON.parse(trimmed);
    // 返回标准化结果
  } catch (parseErr) {
    // 解析输出失败，返回500错误
  }
});

• 超时判断：子进程因超时而被终止时，signal为SIGTERM，返回超时错误
• 结果解析：
  1. 去除输出首尾空格，避免空输出导致解析失败
  2. 解析JSON格式的输出，转换为标准化结果
  3. 解析失败则返回内部错误（包含原始输出，便于调试）

3.6 子进程启动失败处理

child.on("error", (err) => {
  // 返回500错误，提示子进程启动失败
});

• 捕获子进程创建失败的错误（如Node.js可执行文件不存在）
• 返回标准化的500错误，包含具体错误信息

4. 服务器启动与进程管理

// 启动服务器
const PORT = process.env.PORT || 5002;
server.listen(PORT, "0.0.0.0", () => {
  console.log(`✅ JS sandbox listening on http://0.0.0.0:${PORT}`);
});

// 优雅退出
process.on("SIGTERM", () => {
  server.close(() => {
    console.log("Process terminated");
  });
});

• 启动服务器：监听指定端口（优先环境变量PORT，默认5002），绑定所有网卡（0.0.0.0）
• 优雅退出：捕获SIGTERM信号（如容器停止指令），关闭服务器后退出进程，避免端口占用

---

1. 隔离性：通过子进程执行用户代码，避免用户代码阻塞主进程或污染主进程环境
2. 超时控制：基于子进程的timeout配置，实现精准的代码执行超时终止
3. 标准化输出：统一的JSON结果格式（{return: ..., error: ...}），便于客户端解析
4. 错误全覆盖：捕获请求解析、子进程启动、代码执行、结果解析等全链路错误
5. 跨域支持：完整的CORS配置，支持前端跨域调用

---

2.3 python沙箱环境：

接下来我们，可以使用docker exec命令，进入python沙箱环境，来查看python沙箱环境是如何工作的，内容如下：

可以看到里面包含FastAPI 是一个现代、高性能的 Python Web 框架，专门用于构建 API，它基于 Python 类型提示，兼具易用性和高性能，还能自动生成交互式 API 文档，是目前构建后端 API 的热门选择。

import asyncio
import inspect
import json
import os
import traceback
from multiprocessing import Process, Queue
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from dotenv import load_dotenv


load_dotenv(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), '.env'))


# ========== 子进程执行函数 ==========
def _run_code_in_process(code: str, inputs: dict, result_queue: Queue, timeout: float):
    """在独立子进程中执行用户代码，使用完整 Python 环境（无 RestrictedPython）"""
    try:
        # 使用标准 Python 全局命名空间（无限制）
        user_globals = {
            '__builtins__': __builtins__,  # 完整内置函数
            '__name__': '__main__',
            '__file__': '<user_code>',
        }

        # 执行用户代码
        exec(code, user_globals)

        # 检查必要组件
        if 'Args' not in user_globals:
            raise RuntimeError("User code must define a class named 'Args'.")
        if 'main' not in user_globals:
            raise RuntimeError("User code must define a function named 'main'.")

        Args = user_globals['Args']
        main_func = user_globals['main']
        args = Args(inputs)

        # —————— 统一异步/同步执行 + 精确超时 ——————
        async def _run():
            if inspect.iscoroutinefunction(main_func):
                return await main_func(args)
            else:
                loop = asyncio.get_event_loop()
                return await loop.run_in_executor(None, main_func, args)

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            result = loop.run_until_complete(asyncio.wait_for(_run(), timeout=timeout))
        finally:
            loop.close()

        # 验证结果可 JSON 序列化
        json.dumps(result)

        result_queue.put({"return": result, "error": None})

    except Exception as e:
        err_trace = traceback.format_exc()
        result_queue.put({"return": None, "error": err_trace})


# ========== 主执行入口 ==========
def exec_code(code: str, inputs: dict, timeout: float = 10.0):
    if timeout <= 0 or timeout > 3000:
        return {"return": None, "error": "Timeout must be between 0 and 3000 seconds."}

    result_queue = Queue()
    process = Process(
        target=_run_code_in_process,
        args=(code, inputs, result_queue, timeout)
    )
    process.start()
    process.join(timeout=timeout + 0.5)

    if process.is_alive():
        process.terminate()
        process.join()
        return {"return": None, "error": f"Execution timed out (> {timeout} seconds)."}

    if not result_queue.empty():
        return result_queue.get()
    else:
        return {"return": None, "error": "Execution failed silently."}


# ========== FastAPI 接口 ==========
app = FastAPI()


@app.post("/run")
async def run(request: Request):
    data = await request.json()
    if not data:
        return JSONResponse({"return": None, "error": "Invalid JSON payload."}, status_code=400)

    code = data.get("code", "")
    inputs = data.get("inputs", {})
    timeout = data.get("timeout", 10)

    if not isinstance(code, str) or not isinstance(inputs, dict):
        return JSONResponse(
            {
                "return": None,
                "error": "'code' must be string and 'inputs' must be dict.",
            },
            status_code=400,
        )
    if not isinstance(timeout, (int, float)):
        return JSONResponse({"return": None, "error": "'timeout' must be a number."}, status_code=400)

    if not code.strip():
        return JSONResponse({"return": None, "error": "No code provided."})

    try:
        result = exec_code(code, inputs, timeout=float(timeout))
    except Exception as e:
        result = {"return": None, "error": traceback.format_exc()}

    return JSONResponse(result)


@app.get("/health")
def health_check():
    return JSONResponse({"status": "ok"})

def main():
    host = os.getenv("HOST", "0.0.0.0")
    port = int(os.getenv("PORT", 5001))
    uvicorn.run(app, host=host, port=port)

if __name__ == "__main__":

这是一个基于 FastAPI 构建的代码执行服务，核心功能是接收用户提交的 Python 代码和输入参数，在独立子进程中安全执行并返回结果，还包含了超时控制和错误处理。

这段代码实现了一个 HTTP API 服务，核心能力是：

1. 接收用户上传的 Python 代码片段、输入参数和超时时间
2. 在独立的子进程中执行这段代码（避免影响主进程，且能强制终止）
3. 统一处理同步/异步的 main 函数执行
4. 严格控制执行超时，捕获所有执行错误
5. 通过 FastAPI 提供 HTTP 接口，返回执行结果或错误信息

---

1. 导入依赖模块

import asyncio          # 处理异步函数执行
import inspect          # 检查函数类型（同步/异步）
import json             # 序列化/验证执行结果
import os               # 处理环境变量和路径
import traceback        # 捕获详细的异常堆栈信息
from multiprocessing import Process, Queue  # 子进程和进程间通信
import uvicorn          # FastAPI 的 ASGI 服务器
from fastapi import FastAPI, Request       # FastAPI 核心组件
from fastapi.responses import JSONResponse # JSON 格式响应
from dotenv import load_dotenv             # 加载 .env 环境变量

这些是代码运行所需的核心依赖：

• multiprocessing：核心是用 Process 创建子进程隔离执行环境，Queue 实现子进程和主进程的结果传递
• fastapi/uvicorn：构建和运行 HTTP API 服务
• asyncio：处理异步函数执行
• traceback：捕获详细的错误信息，方便调试

2. 加载环境变量

load_dotenv(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), '.env'))

• 从当前文件的上上级目录加载 .env 文件
• 作用：读取配置（如后续的服务端口、主机地址），避免硬编码

3. 子进程执行核心函数 _run_code_in_process

这是最核心的函数，在独立子进程中执行用户代码，参数说明：

• code: 用户提交的 Python 代码字符串
• inputs: 用户传入的参数字典
• result_queue: 进程间通信的队列，用于传递执行结果/错误
• timeout: 执行超时时间（秒）

函数内部逻辑拆解：

def _run_code_in_process(code: str, inputs: dict, result_queue: Queue, timeout: float):
    try:
        # 1. 初始化用户代码的全局命名空间
        user_globals = {
            '__builtins__': __builtins__,  # 完整的 Python 内置函数（无限制）
            '__name__': '__main__',
            '__file__': '<user_code>',     # 标记代码来源（虚拟路径）
        }

        # 2. 执行用户代码（将代码加载到自定义的全局命名空间中）
        exec(code, user_globals)

        # 3. 检查用户代码必须包含的组件
        if 'Args' not in user_globals:
            raise RuntimeError("User code must define a class named 'Args'.")
        if 'main' not in user_globals:
            raise RuntimeError("User code must define a function named 'main'.")

        # 4. 初始化参数和主函数
        Args = user_globals['Args']    # 获取用户定义的 Args 类
        main_func = user_globals['main']  # 获取用户定义的 main 函数
        args = Args(inputs)           # 用用户输入初始化 Args 实例

        # 5. 统一处理同步/异步函数执行
        async def _run():
            if inspect.iscoroutinefunction(main_func):
                # 如果 main 是异步函数，直接 await 执行
                return await main_func(args)
            else:
                # 如果是同步函数，用事件循环的线程池执行（不阻塞异步循环）
                loop = asyncio.get_event_loop()
                return await loop.run_in_executor(None, main_func, args)

        # 6. 创建新的事件循环并执行（带超时）
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            # 执行 _run 并严格控制超时
            result = loop.run_until_complete(asyncio.wait_for(_run(), timeout=timeout))
        finally:
            loop.close()  # 确保事件循环关闭，避免资源泄漏

        # 7. 验证结果可 JSON 序列化（避免返回无法序列化的对象）
        json.dumps(result)

        # 8. 将成功结果放入队列
        result_queue.put({"return": result, "error": None})

    except Exception as e:
        # 捕获所有异常，将错误堆栈放入队列
        err_trace = traceback.format_exc()
        result_queue.put({"return": None, "error": err_trace})

关键要点：

• 用 exec 执行用户代码，且放在自定义的 user_globals 命名空间中
• 强制要求用户代码定义 Args 类（接收输入参数）和 main 函数（执行逻辑）
• 自动识别 main 是同步/异步函数，统一用异步方式执行
• 执行结果必须能 JSON 序列化（否则会触发异常）
• 所有结果/错误通过 Queue 传递回主进程

4. 主执行入口 exec_code

封装子进程的创建、运行、超时控制逻辑，是对外提供的核心函数：

def exec_code(code: str, inputs: dict, timeout: float = 10.0):
    # 1. 校验超时时间范围（0-3000秒）
    if timeout <= 0 or timeout > 3000:
        return {"return": None, "error": "Timeout must be between 0 and 3000 seconds."}

    # 2. 创建进程间通信队列
    result_queue = Queue()
    # 3. 创建子进程，指定要执行的函数和参数
    process = Process(
        target=_run_code_in_process,
        args=(code, inputs, result_queue, timeout)
    )
    # 4. 启动子进程
    process.start()
    # 5. 等待子进程结束（超时时间 = 执行超时 + 0.5秒缓冲）
    process.join(timeout=timeout + 0.5)

    # 6. 如果子进程还在运行（超时），强制终止
    if process.is_alive():
        process.terminate()  # 强制杀死子进程
        process.join()       # 等待进程彻底退出
        return {"return": None, "error": f"Execution timed out (> {timeout} seconds)."}

    # 7. 从队列获取执行结果
    if not result_queue.empty():
        return result_queue.get()
    else:
        return {"return": None, "error": "Execution failed silently."}

核心作用：

• 隔离用户代码的执行环境（子进程），即使代码卡死/崩溃也不影响主服务
• 严格控制执行超时，超时后强制终止子进程
• 处理子进程的异常退出（如静默失败）

5. FastAPI 接口定义

构建 HTTP API 服务，对外提供代码执行能力：

5.1 初始化 FastAPI 应用

app = FastAPI()

5.2 /run 接口（核心 POST 接口）

接收用户提交的代码、参数、超时时间，调用 exec_code 执行并返回结果：

@app.post("/run")
async def run(request: Request):
    # 1. 解析请求体的 JSON 数据
    data = await request.json()
    if not data:
        return JSONResponse({"return": None, "error": "Invalid JSON payload."}, status_code=400)

    # 2. 提取请求参数
    code = data.get("code", "")
    inputs = data.get("inputs", {})
    timeout = data.get("timeout", 10)

    # 3. 参数类型校验
    if not isinstance(code, str) or not isinstance(inputs, dict):
        return JSONResponse(
            {
                "return": None,
                "error": "'code' must be string and 'inputs' must be dict.",
            },
            status_code=400,
        )
    if not isinstance(timeout, (int, float)):
        return JSONResponse({"return": None, "error": "'timeout' must be a number."}, status_code=400)

    # 4. 检查代码是否为空
    if not code.strip():
        return JSONResponse({"return": None, "error": "No code provided."})

    # 5. 执行代码并捕获执行过程中的异常
    try:
        result = exec_code(code, inputs, timeout=float(timeout))
    except Exception as e:
        result = {"return": None, "error": traceback.format_exc()}

    # 6. 返回执行结果
    return JSONResponse(result)

5.3 /health 接口（健康检查）

用于监控服务状态，通常被容器/网关用来检查服务是否存活：

@app.get("/health")
def health_check():
    return JSONResponse({"status": "ok"})

6. 服务启动逻辑

def main():
    # 从环境变量读取配置（默认值：0.0.0.0:5001）
    host = os.getenv("HOST", "0.0.0.0")
    port = int(os.getenv("PORT", 5001))
    # 启动 uvicorn 服务器，运行 FastAPI 应用
    uvicorn.run(app, host=host, port=port)

if __name__ == "__main__":
    main()  # 注意：原代码这里少了 main() 调用，补充后才能启动服务

• 0.0.0.0：允许外部网络访问（而非仅本地）
• 端口默认 5001，可通过环境变量 PORT 覆盖
• 用 uvicorn 作为 ASGI 服务器运行 FastAPI 应用（FastAPI 推荐的生产级服务器）

---

示例：如何调用这个服务

假设服务运行在 http://localhost:3010，可以用 curl 调用：

curl -X POST http://localhost:3010/run \
-H "Content-Type: application/json" \
-d '{
    "code": "class Args:\n    def __init__(self, inputs):\n        self.num = inputs.get(\"num\", 0)\ndef main(args):\n    return args.num * 2",
    "inputs": {"num": 10},
    "timeout": 5
}'

返回结果：

{"return": 20, "error": null}

---

1. 核心设计：用子进程隔离用户代码执行环境，避免代码崩溃/卡死影响主服务，同时支持同步/异步函数执行。
2. 安全控制：严格的超时限制（超时强制终止子进程）、参数校验、结果 JSON 序列化验证。
3. 接口能力：通过 FastAPI 提供 /run（执行代码）和 /health（健康检查）两个接口，输入输出均为 JSON 格式，易于集成。

这段代码的典型应用场景是在线代码执行平台、自动化测试工具、动态脚本运行服务等。需要注意的是，由于允许执行任意 Python 代码，生产环境中需额外增加安全限制（如资源限制、代码白名单等）。

---

三、插件IDE运行时实践：

在openJiuwen智能体开发中，插件是扩展其功能、弥补大模型短板的关键工具。以下将介绍如何使用Python开发一个智能体插件，并提供一个可直接运行的小工具示例。‌

开发一个智能体插件通常遵循“定义工具-编写代码-调试发布”的流程，其核心在于利用Python代码实现确定性逻辑，以处理大模型不擅长的计算、数据格式化或隐私敏感任务。‌

• ‌定义工具接口‌：明确插件的功能、输入参数与输出格式。例如，一个密码生成器插件需定义“密码长度”和“是否包含符号”作为输入，将“生成的密码字符串”作为输出。‌
• ‌编写核心逻辑‌：在云端IDE中使用Python实现功能。代码需结构清晰，妥善处理输入参数，并严格按照定义的输出格式返回结果。‌
• ‌测试与集成‌：在发布前必须进行充分调试，确保功能正确。发布后，可在智能体平台中将插件装配至智能体，并通过提示词指导AI调用该工具。‌

import random
import string

def main(args: Args):
  # 1. 获取输入参数，并设置默认值
  length = args.params['len'] if args.params['len'] else 12
  use_symbols = args.params['symbol']

  # 2. 定义字符集
  # 基础字符集：大小写字母 + 数字
  chars = string.ascii_letters + string.digits
  # 若需要特殊符号，则追加
  if use_symbols:
      chars += "!@#$%^&*"

  # 3. 核心生成逻辑
  # 使用random.choice从字符集中随机选取，循环length次
  password = "".join(random.choice(chars) for _ in range(length))

  return {'res': password}

工具说明‌：

• ‌功能‌：生成包含大小写字母、数字及可选特殊符号的随机密码。
• ‌输入‌：length（密码长度，整数类型），symbols（是否使用符号，布尔类型）。
• ‌输出‌：返回一个字典，包含生成的password字符串和状态信息message。‌

通过为智能体开发自定义Python插件，可以精准解决特定业务需求，提升智能体处理复杂、逻辑性任务的可靠性与能力。‌

在调试的过程中，发现遇到错误，在执行过程中，发现输入参数的类型不匹配。

Execution failed after 1/1 attempts: Sandbox run error: Traceback (most recent call last): File "/sandbox/site-packages/openjiuwen_sandbox_pyserver/kernel.py", line 51, in _run_code_in_process result = loop.run_until_complete(asyncio.wait_for(_run(), timeout=timeout)) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete return future.result() ^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/asyncio/tasks.py", line 479, in wait_for return fut.result() ^^^^^^^^^^^^ File "/sandbox/site-packages/openjiuwen_sandbox_pyserver/kernel.py", line 46, in _run return await loop.run_in_executor(None, main_func, args) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/concurrent/futures/thread.py", line 58, in run result = self.fn(*self.args, **self.kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "<string>", line 26, in main TypeError: 'str' object cannot be interpreted as an integer

经发现，输入参数的类型不匹配，导致代码运行出错，将输入参数类型改为整数，问题解决。

---

四、总结：

当智能体被赋予执行代码或访问外部资源的能力时，一旦出现逻辑错误或被恶意诱导（如“越狱”攻击），可能触发危险操作，例如：

• 删除关键文件
• 访问敏感数据库
• 连接非法服务器

沙箱通过‌资源隔离技术‌（如虚拟机、容器）限制智能体的权限，使其无法触及主机系统的敏感资源。这就像给孩子一个玩具厨房做饭——可以模拟全过程，但不会点燃真正的燃气。

✅ 典型应用：AI代码解释器在沙箱中运行用户提交的Python脚本，即使代码包含rm -rf /也不会影响宿主系统。

在复杂任务中，多个智能体可能协同完成目标。沙箱提供了一个可监控、可回滚的实验场：

• 记录每个决策步骤
• 实时观察资源占用
• 一键重置至初始状态

这种“数字训练场”机制，极大提升了开发效率和调试精度。对于从事AI算法开发的用户而言，这尤其有助于强化学习中的策略迭代与异常分析。

满足企业级部署的安全合规要求，在金融、医疗等高敏感行业，AI智能体的部署必须符合严格的安全标准。沙箱环境帮助企业实现：

• 最小权限原则（仅开放必要API）
• 网络访问控制（禁止未授权外联）
• 完整日志审计（便于追溯责任）

---

openJiuwen 项目核心组件：

• Agent Studio（智能体工作室）：
  https://atomgit.com/openJiuwen/agent-studio
  可视化智能体开发平台，提供零码、低码可视化开发和工作流编排能力，以及模型、知识库、插件等各资源管理能力

• Agent Core（智能体核心）：
  https://atomgit.com/openJiuwen/agent-core
  智能体核心引擎，提供Agent开发、运行、调优与演进相关的全套SDK能力