星辰RPA 帮助文档

Appearance

本地健康检查 ​

🏠 本地客户端状态检查

此接口用于检测本地RPA客户端的运行状态和账号匹配性。通过本地端口直接调用，无需网络连接。

接口描述 ​

通过本地端口直接检测RPA客户端的运行状态，并验证API KEY对应的账号是否与本地启动的账号匹配。此接口用于本地环境下的状态监控。

请求方式 ​

GET http://localhost:13159/api/rpa-openapi/health/local-check

认证要求 ​

需要带 BEARER KEY

请在请求头中添加：

Authorization: Bearer {YOUR_API_KEY}

请求参数 ​

此接口不需要额外的请求参数，认证信息通过请求头提供。

响应格式 ​

成功响应 - 本地RPA正常启动且账号匹配 (HTTP 200) ​

json
{
  "code": "0000",
  "msg": "本地RPA正常启动",
  "data": {
    "equality": true
  }
}

失败响应 - 账号不匹配 (HTTP 200) ​

json
{
  "code": "5001",
  "msg": "本地RPA启动账号与Agent账号不匹配",
  "data": {
    "equality": false
  }
}

连接错误 - 本地RPA未启动 ​

如果本地RPA客户端未启动，将无法连接到 localhost:13159 端口，会收到连接错误：

Connection refused (Connection refused)

或类似的网络连接错误信息。

响应字段说明 ​

字段名类型说明
codeString响应代码，"0000" 表示成功，"5001" 表示账号不匹配
msgString响应消息，描述本地客户端状态
dataObject返回数据对象
data.equalityBoolean账号匹配状态，true 表示匹配，false 表示不匹配

状态说明 ​

本地RPA正常运行 ​

  • code: "0000"
  • msg: "本地RPA正常启动"
  • data.equality: true
  • 表示本地RPA客户端正在运行，且API KEY对应的账号与本地启动账号匹配

账号不匹配 ​

  • code: "5001"
  • msg: "本地RPA启动账号与Agent账号不匹配"
  • data.equality: false
  • 表示本地RPA客户端正在运行，但API KEY对应的账号与本地启动账号不一致

连接失败 ​

  • 无响应（连接错误）
  • 表示本地RPA客户端未启动或端口13159不可访问

说明 ​

  • 此接口通过本地端口直接调用，无需网络连接
  • 用于验证本地RPA客户端的运行状态和账号匹配性
  • 正常情况下，本地账号应该与API KEY对应的账号一致
  • 如果遇到账号不匹配的情况，可能需要重新登录或检查账号配置
  • 连接失败通常表示RPA客户端未启动，需要先启动客户端

代码示例 ​

cURL ​

bash
curl -X GET "http://localhost:13159/api/rpa-openapi/health/local-check" \
  -H "Authorization: Bearer {YOUR_API_KEY}"

Python ​

python
import requests

def check_local_health(api_key):
    """检查本地RPA健康状态"""
    url = "http://localhost:13159/api/rpa-openapi/health/local-check"
    headers = {
        "Authorization": f"Bearer {api_key}"
    }

    try:
        response = requests.get(url, headers=headers, timeout=5)
        result = response.json()

        if result['code'] == "0000":
            print("✅ 本地RPA正常运行，账号匹配")
            return True, "本地RPA正常启动"
        else:
            print("⚠️ 本地RPA运行中，但账号不匹配")
            return False, result['msg']

    except requests.exceptions.ConnectionError:
        print("❌ 本地RPA客户端未启动")
        return False, "本地RPA客户端未启动"
    except requests.exceptions.Timeout:
        print("⏰ 连接超时")
        return False, "连接超时"
    except Exception as e:
        print(f"❌ 检查失败: {str(e)}")
        return False, f"检查失败: {str(e)}"

# 使用示例
api_key = "{YOUR_API_KEY}"
is_healthy, message = check_local_health(api_key)

JavaScript (Node.js) ​

javascript
const https = require('http'); // 注意：使用http而不是https，因为是localhost

function checkLocalHealth(apiKey) {
    return new Promise((resolve, reject) => {
        const options = {
            hostname: 'localhost',
            port: 13159,
            path: '/api/rpa-openapi/health/local-check',
            method: 'GET',
            headers: {
                'Authorization': `Bearer ${apiKey}`
            }
        };

        const req = https.request(options, (res) => {
            let data = '';

            res.on('data', (chunk) => {
                data += chunk;
            });

            res.on('end', () => {
                try {
                    const result = JSON.parse(data);

                    if (result.code === "0000") {
                        console.log("✅ 本地RPA正常运行，账号匹配");
                        resolve({ success: true, message: "本地RPA正常启动" });
                    } else {
                        console.log("⚠️ 本地RPA运行中，但账号不匹配");
                        resolve({ success: false, message: result.msg });
                    }
                } catch (e) {
                    reject(new Error(`解析响应失败: ${e.message}`));
                }
            });
        });

        req.on('error', (e) => {
            if (e.code === 'ECONNREFUSED') {
                console.log("❌ 本地RPA客户端未启动");
                resolve({ success: false, message: "本地RPA客户端未启动" });
            } else {
                console.log(`❌ 连接错误: ${e.message}`);
                reject(e);
            }
        });

        req.setTimeout(5000, () => {
            req.destroy();
            reject(new Error("连接超时"));
        });

        req.end();
    });
}

// 使用示例
const apiKey = "{YOUR_API_KEY}";
checkLocalHealth(apiKey)
    .then(result => {
        console.log("检查结果:", result);
    })
    .catch(error => {
        console.error("检查失败:", error.message);
    });

完整的本地环境检查工具（Python） ​

python
import requests
import socket
import time

class LocalRPAHealthChecker:
    """本地RPA健康检查工具"""

    def __init__(self, api_key, port=13159):
        self.api_key = api_key
        self.port = port
        self.base_url = f"http://localhost:{port}"

    def check_port_open(self):
        """检查端口是否开放"""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2)
            result = sock.connect_ex(('localhost', self.port))
            sock.close()
            return result == 0
        except:
            return False

    def check_local_health(self):
        """检查本地RPA健康状态"""
        url = f"{self.base_url}/api/rpa-openapi/health/local-check"
        headers = {
            "Authorization": f"Bearer {self.api_key}"
        }

        try:
            response = requests.get(url, headers=headers, timeout=5)
            result = response.json()

            return {
                "status": "connected",
                "code": result.get("code"),
                "message": result.get("msg"),
                "equality": result.get("data", {}).get("equality"),
                "raw_result": result
            }

        except requests.exceptions.ConnectionError:
            return {
                "status": "disconnected",
                "message": "本地RPA客户端未启动",
                "code": None,
                "equality": None
            }
        except requests.exceptions.Timeout:
            return {
                "status": "timeout",
                "message": "连接超时",
                "code": None,
                "equality": None
            }
        except Exception as e:
            return {
                "status": "error",
                "message": f"检查失败: {str(e)}",
                "code": None,
                "equality": None
            }

    def get_health_report(self):
        """获取完整的健康报告"""
        port_open = self.check_port_open()
        health_result = self.check_local_health()

        report = {
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "port_check": {
                "port": self.port,
                "is_open": port_open
            },
            "health_check": health_result,
            "overall_status": self._determine_overall_status(port_open, health_result)
        }

        return report

    def _determine_overall_status(self, port_open, health_result):
        """确定整体状态"""
        if not port_open:
            return "CLIENT_NOT_RUNNING"
        elif health_result["status"] == "connected":
            if health_result.get("code") == "0000":
                return "HEALTHY"
            else:
                return "ACCOUNT_MISMATCH"
        else:
            return "UNKNOWN_ERROR"

    def print_report(self):
        """打印健康报告"""
        report = self.get_health_report()

        print("=== 本地RPA健康检查报告 ===")
        print(f"检查时间: {report['timestamp']}")
        print()

        print("端口检查:")
        print(f"  端口 {report['port_check']['port']}: {'开放' if report['port_check']['is_open'] else '关闭'}")
        print()

        print("健康检查:")
        if report['health_check']['status'] == 'connected':
            print(f"  连接状态: 已连接")
            print(f"  响应代码: {report['health_check']['code']}")
            print(f"  消息: {report['health_check']['message']}")
            print(f"  账号匹配: {'是' if report['health_check']['equality'] else '否'}")
        else:
            print(f"  连接状态: {report['health_check']['status']}")
            print(f"  消息: {report['health_check']['message']}")
        print()

        print("整体状态:")
        status = report['overall_status']
        if status == "HEALTHY":
            print("  ✅ 本地RPA运行正常，账号匹配")
        elif status == "ACCOUNT_MISMATCH":
            print("  ⚠️ 本地RPA运行中，但账号不匹配")
        elif status == "CLIENT_NOT_RUNNING":
            print("  ❌ 本地RPA客户端未启动")
        else:
            print(f"  ❓ 未知状态: {status}")

# 使用示例
if __name__ == "__main__":
    api_key = "{YOUR_API_KEY}"
    checker = LocalRPAHealthChecker(api_key)
    checker.print_report()