WebSocket跨语言调用：Python客户端如何调用JS服务器函数？

我已经通过WebSocket搭建了基于JSON的客户端-服务器通信链路，服务器运行在JavaScript环境中，包含以下两个函数：

function FirstFunction(arg1, arg2) {
    // 示例逻辑：返回两数之和
    return arg1 + arg2;
}

function SecondFunction(argOther) {
    // 示例逻辑：返回参数的平方
    return argOther ** 2;
}

我希望Python客户端发送包含目标函数名及对应参数的JSON对象，让服务器自动调用匹配的函数，类似如下伪逻辑：

# Python客户端发送的JSON结构
data = {"method": "FirstFunction", "params": {"arg1": 10, "arg2": 20}}

// 服务器侧期望的调用逻辑（伪代码）
result = data["method"](data.params.arg1, data.params.arg2)

请问具体该如何实现这一需求？

服务器端（JavaScript）实现 #

核心思路是通过函数映射表安全管理可调用函数，避免代码注入风险，同时解析客户端请求的JSON数据，匹配函数并执行。

1. 创建可调用函数映射 #

将允许客户端调用的函数存入一个对象，作为白名单：

const availableMethods = {
    FirstFunction,
    SecondFunction
};

2. 处理WebSocket消息（以Node.js的ws库为例） #

在WebSocket连接的消息监听中，解析请求、验证方法、执行函数并返回结果：

const WebSocket = require('ws');
const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', (ws) => {
    ws.on('message', (message) => {
        try {
            // 解析客户端发来的JSON数据
            const request = JSON.parse(message.toString());
            
            // 验证方法是否存在
            if (!request.method || !availableMethods.hasOwnProperty(request.method)) {
                ws.send(JSON.stringify({ error: '无效的方法名' }));
                return;
            }

            const targetFunc = availableMethods[request.method];
            const params = request.params || {};

            // 提取函数的参数名列表，按顺序匹配传入参数
            const paramNames = targetFunc.toString()
                .match(/\((.*?)\)/)[1]
                .split(',')
                .map(name => name.trim())
                .filter(Boolean);
            
            const args = paramNames.map(name => params[name]);
            const result = targetFunc(...args);

            // 返回执行结果
            ws.send(JSON.stringify({ result }));
        } catch (err) {
            // 处理解析或执行错误
            ws.send(JSON.stringify({ error: err.message }));
        }
    });
});

客户端（Python）实现 #

使用websockets库建立连接，构造符合格式的JSON请求并接收返回结果。

1. 安装依赖 #

pip install websockets

2. 客户端调用代码 #

import asyncio
import websockets
import json

async def invoke_server_method(method_name, params):
    async with websockets.connect('ws://localhost:8080') as ws:
        # 构造请求JSON
        request_json = json.dumps({
            "method": method_name,
            "params": params
        })
        await ws.send(request_json)
        # 接收并解析返回结果
        response = await ws.recv()
        return json.loads(response)

# 调用示例
if __name__ == "__main__":
    # 调用FirstFunction
    res1 = asyncio.run(invoke_server_method("FirstFunction", {"arg1": 10, "arg2": 20}))
    print("FirstFunction结果:", res1)

    # 调用SecondFunction
    res2 = asyncio.run(invoke_server_method("SecondFunction", {"argOther": 5}))
    print("SecondFunction结果:", res2)

关键注意事项 #

• 安全优先：绝对禁止使用eval()或直接从全局对象中查找函数，必须用预定义的映射表限制可调用范围，避免代码注入攻击。
• 参数适配：通过解析函数的参数名来匹配传入的参数，保证参数传递的正确性，无需依赖固定的参数顺序。
• 错误处理：两端都要处理JSON解析失败、方法不存在、函数执行异常等情况，避免程序崩溃。

内容的提问来源于stack exchange，提问作者Hiperfly