星辰RPA 帮助文档

Appearance

停止当前工作流 ​

⏹️ 紧急停止运行中的任务

此接口用于紧急停止当前正在运行的任何机器人任务。适用于需要立即中断执行的情况。

接口描述 ​

根据 API KEY 停止对应用户当前正在运行的所有机器人任务。此接口提供紧急停止功能，可以立即中断正在执行的工作流。

请求方式 ​

POST https://newapi.iflyrpa.com/api/rpa-openapi/workflows/stop-current

认证要求 ​

需要带 BEARER KEY

请在请求头中添加：

Authorization: Bearer {YOUR_API_KEY}

请求参数 ​

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

响应格式 ​

成功响应 - 停止成功 (HTTP 200) ​

json
{
  "code": "0000",
  "msg": "停止成功",
  "data": {}
}

失败响应 - 停止失败 (HTTP 200) ​

json
{
  "code": "5001",
  "msg": "停止失败或没有正在运行的机器人",
  "data": null
}

响应字段说明 ​

字段名类型说明
codeString响应代码，"0000" 表示成功，"5001" 表示失败
msgString响应消息，描述停止操作结果
dataObject返回数据，成功时为空对象，失败时为 null

状态说明 ​

停止成功 ​

  • code: "0000"
  • msg: "停止成功"
  • data: {}
  • 表示成功停止了正在运行的机器人任务

停止失败 ​

  • code: "5001"
  • msg: "停止失败或没有正在运行的机器人"
  • data: null
  • 表示没有正在运行的任务，或停止操作失败

说明 ​

  • 此接口会停止当前用户所有正在运行的机器人任务
  • 适用于紧急情况下的任务中断
  • 停止操作是立即生效的，不等待任务完成
  • 如果没有正在运行的任务，接口会返回失败响应
  • 建议谨慎使用此接口，避免中断重要任务

代码示例 ​

cURL ​

bash
curl -X POST "https://newapi.iflyrpa.com/api/rpa-openapi/workflows/stop-current" \
  -H "Authorization: Bearer {YOUR_API_KEY}"

Python ​

python
import requests

url = "https://newapi.iflyrpa.com/api/rpa-openapi/workflows/stop-current"
headers = {
    "Authorization": "Bearer {YOUR_API_KEY}"
}

response = requests.post(url, headers=headers)
result = response.json()

print(f"响应码: {result['code']}")
print(f"消息: {result['msg']}")

if result['code'] == "0000":
    print("✅ 成功停止所有运行中的任务")
else:
    print("❌ 停止失败或没有正在运行的任务")

JavaScript ​

javascript
const apiKey = "{YOUR_API_KEY}";
const url = "https://newapi.iflyrpa.com/api/rpa-openapi/workflows/stop-current";

fetch(url, {
  method: "POST",
  headers: {
    "Authorization": `Bearer ${apiKey}`
  }
})
  .then(response => response.json())
  .then(result => {
    console.log(`响应码: ${result.code}`);
    console.log(`消息: ${result.msg}`);

    if (result.code === "0000") {
      console.log("✅ 成功停止所有运行中的任务");
    } else {
      console.log("❌ 停止失败或没有正在运行的任务");
    }
  })
  .catch(error => console.error("Error:", error));

带错误处理的完整示例（Python） ​

python
import requests
import time

def stop_current_workflows():
    """停止当前所有运行中的工作流"""
    url = "https://newapi.iflyrpa.com/api/rpa-openapi/workflows/stop-current"
    headers = {
        "Authorization": "Bearer {YOUR_API_KEY}"
    }

    try:
        response = requests.post(url, headers=headers, timeout=30)
        result = response.json()

        if result['code'] == "0000":
            print("✅ 紧急停止成功：所有运行中的机器人任务已停止")
            return True
        else:
            print(f"⚠️ 停止操作响应：{result['msg']}")
            return False

    except requests.exceptions.Timeout:
        print("❌ 请求超时，请检查网络连接")
        return False
    except requests.exceptions.RequestException as e:
        print(f"❌ 请求失败: {str(e)}")
        return False
    except Exception as e:
        print(f"❌ 未知错误: {str(e)}")
        return False

def emergency_stop():
    """紧急停止流程"""
    print("🚨 正在执行紧急停止操作...")
    success = stop_current_workflows()

    if success:
        print("🎯 所有机器人任务已停止")
        print("💡 提示：已停止所有正在运行的工作流")
    else:
        print("ℹ️ 没有找到正在运行的机器人任务或停止失败")

    return success

# 使用示例
if __name__ == "__main__":
    emergency_stop()

定时监控和自动停止（Python） ​

python
import requests
import time
import threading

def check_and_stop_if_needed(max_runtime_seconds=300):
    """
    检查工作流运行时间，如果超过指定时间则自动停止

    Args:
        max_runtime_seconds: 最大允许运行时间（秒）
    """
    start_time = time.time()

    while True:
        elapsed = time.time() - start_time

        if elapsed > max_runtime_seconds:
            print(f"⚠️ 工作流运行时间已超过 {max_runtime_seconds} 秒，执行自动停止")
            stop_current_workflows()
            break

        time.sleep(30)  # 每30秒检查一次

def monitor_workflow_execution(execution_id, max_runtime=300):
    """监控工作流执行，超时自动停止"""
    print(f"开始监控执行ID: {execution_id}")

    # 启动监控线程
    monitor_thread = threading.Thread(
        target=check_and_stop_if_needed,
        args=(max_runtime,)
    )
    monitor_thread.daemon = True
    monitor_thread.start()

    # 这里可以添加其他监控逻辑
    print("监控线程已启动，按 Ctrl+C 手动停止")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n🛑 收到停止信号，正在停止所有任务...")
        stop_current_workflows()
        print("✅ 手动停止完成")

# 使用示例
# monitor_workflow_execution("your-execution-id", max_runtime=600)  # 10分钟超时