工作台/工作流节点

代码运行

FastGPT 代码运行节点介绍(适用于 4.14.8 及以上版本)

本文档适用于 FastGPT 4.14.8 及以上版本。4.14.7 及以下版本请参考 代码运行(弃)

功能

代码运行节点支持在安全沙盒中执行 JavaScript 和 Python 代码,用于数据处理、格式转换、逻辑计算等场景。

支持的语言

  • JavaScript(基于 Bun 运行时)
  • Python 3

注意事项

  • 私有化用户需要部署 fastgpt-sandbox 镜像,并配置 SANDBOX_URL 环境变量。
  • 沙盒默认最大运行 60s,可通过配置调整。
  • 代码运行在隔离的进程池中,无法访问文件系统和内网。

变量输入

可在自定义输入中添加代码运行需要的变量。

JavaScript — 在 main 函数参数中解构:

async function main({data1, data2}){
    return {
        result: data1 + data2
    }
}

Python — 在 main 函数参数中按变量名接收:

def main(data1, data2):
    return {"result": data1 + data2}

结果输出

务必返回一个 object 对象(JS)或 dict 字典(Python)。

自定义输出中,可以添加变量名来获取对应 key 下的值。例如返回了:

{
  "result": "hello",
  "count": 42
}

自定义输出中添加 resultcount 两个变量即可获取对应的值。

内置函数

httpRequest 发起 HTTP 请求

在沙盒内发起外部 HTTP 请求。自动拦截内网地址(SSRF 防护)。

JavaScript 示例:

async function main({url}){
    const res = await SystemHelper.httpRequest(url, {
        method: 'GET',       // 请求方法,默认 GET
        headers: {},         // 自定义请求头
        body: null,          // 请求体(对象会自动 JSON 序列化)
        timeout: 60          // 超时秒数,最大 60s
    })
    return {
        status: res.status,
        data: res.data
    }
}

Python 示例:

def main(url):
    res = SystemHelper.httpRequest(url, method="GET", headers={}, timeout=10)
    return {"status": res["status"], "data": res["data"]}

限制:

  • 每次执行最多 30 个请求
  • 单次请求超时 60s
  • 响应体最大 2MB
  • 仅允许 http/https 协议
  • 自动拦截内网 IP(127.0.0.0/8, 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16 等)

可用模块

JavaScript 白名单模块

以下 npm 模块可通过 require() 使用:

模块说明示例
lodash工具函数库const _ = require('lodash')
moment日期处理const moment = require('moment')
dayjs轻量日期库const dayjs = require('dayjs')
crypto-js加密库const CryptoJS = require('crypto-js')
uuidUUID 生成const { v4 } = require('uuid')
qs查询字符串解析const qs = require('qs')

其他模块(如 fs, child_process, net 等)被禁止使用。

Python 白名单模块

以下 Python 标准库和第三方库可直接 import:

数学和数值计算

模块说明
math数学函数
cmath复数数学
decimal十进制浮点运算
fractions分数运算
random随机数生成
statistics统计函数

数据结构和算法

模块说明
collections容器数据类型
array数组
heapq堆队列
bisect数组二分查找
queue队列
copy浅拷贝和深拷贝

函数式编程

模块说明
itertools迭代器工具
functools高阶函数
operator标准运算符

字符串和文本处理

模块说明
string字符串常量
re正则表达式
difflib差异计算
textwrap文本包装
unicodedataUnicode 数据库
codecs编解码器

日期和时间

模块说明
datetime日期时间
time时间访问
calendar日历

数据序列化

模块说明
jsonJSON 编解码
csvCSV 文件读写
base64Base64 编解码
binascii二进制和 ASCII 转换
struct字节串解析

加密和哈希

模块说明
hashlib哈希算法
hmacHMAC 消息认证
secrets安全随机数
uuidUUID 生成

类型和抽象

模块说明
typing类型提示
abc抽象基类
enum枚举类型
dataclasses数据类
contextlib上下文管理器

其他实用工具

模块说明
pprint美化打印
weakref弱引用

第三方库

模块说明
numpy数值计算
pandas数据分析
matplotlib数据可视化

禁止使用的模块: os, sys, subprocess, socket, urllib, http, requests 等涉及系统调用、网络访问、文件系统的模块。

安全限制

沙盒提供多层安全防护:

  • 模块拦截:JS 和 Python 均只允许使用白名单模块
  • 网络隔离:自动拦截内网 IP 请求(SSRF 防护)
  • 文件隔离:无法读写容器文件系统
  • 超时保护:默认 60s 超时,防止死循环
  • 进程隔离:每次执行在独立的沙盒进程中运行

使用示例

JavaScript 示例

数据格式转换
// 将逗号分隔的字符串转为数组
function main({input}){
    const items = input.split(',').map(s => s.trim()).filter(Boolean)
    return { items, count: items.length }
}
日期计算
const dayjs = require('dayjs')

function main(){
    const now = dayjs()
    return {
        today: now.format('YYYY-MM-DD'),
        nextWeek: now.add(7, 'day').format('YYYY-MM-DD'),
        timestamp: now.valueOf()
    }
}
HTTP 请求 - 获取天气
async function main({city}){
    const res = await SystemHelper.httpRequest(
        `https://api.example.com/weather?city=${city}`,
        { method: 'GET', timeout: 10 }
    )
    
    return {
        temperature: res.data.temp,
        weather: res.data.condition
    }
}
数据加密
const CryptoJS = require('crypto-js')

function main({text, key}){
    const encrypted = CryptoJS.AES.encrypt(text, key).toString()
    return { encrypted }
}

Python 示例

数据统计
import math

def main(numbers):
    if not numbers:
        return {"error": "no data"}
    
    mean = sum(numbers) / len(numbers)
    variance = sum((x - mean)**2 for x in numbers) / len(numbers)
    
    return {
        "mean": mean,
        "max": max(numbers),
        "min": min(numbers),
        "std": math.sqrt(variance)
    }
日期处理
from datetime import datetime, timedelta

def main(date_str):
    dt = datetime.strptime(date_str, "%Y-%m-%d")
    next_week = dt + timedelta(days=7)
    
    return {
        "input": date_str,
        "next_week": next_week.strftime("%Y-%m-%d"),
        "weekday": dt.strftime("%A")
    }
HTTP 请求 - API 调用
def main(api_url, api_key):
    res = SystemHelper.httpRequest(
        api_url,
        method="GET",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=10
    )
    
    return {
        "status": res["status"],
        "data": res["data"]
    }
JSON 数据处理
import json

def main(json_str):
    data = json.loads(json_str)
    
    # 提取特定字段
    result = {
        "names": [item["name"] for item in data if "name" in item],
        "count": len(data)
    }
    
    return result
正则表达式匹配
import re

def main(text):
    # 提取所有邮箱地址
    emails = re.findall(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', text)
    
    return {
        "emails": emails,
        "count": len(emails)
    }