如何通过终端脚本将Step Functions等待中的活动任务标记为成功

你配置了带Activity类型任务的AWS Step Functions状态机，要求Activity任务保持等待状态，直到你在本地终端运行指定脚本传入输入，脚本执行完成后将等待中的任务标记为执行成功。

附状态机配置参考截图：

现有配置与代码 #

• 脚本执行命令：

node report-choice-step-success.js --stepfunction-arn <SFN-EXEC> --step-name ManualTask

• 现有report-choice-step-success.js脚本内容：

const main = () => {
    let sfnClient;
    const rolename =  `StepFunctionExecuter-LOCAL`;

    return getcreds({ accountId: '123456789012', region: 'us-east-1', rolename })
        .then(params => {
            sfnClient = new AWS.StepFunctions(params)
        })
        
        .then(() => startmystepfunction(sfnClient));
};

const startmystepfunction = (sfnClient) => {
    const stateMachineArn = `arn:aws:states:us-east-1:123456789012:stateMachine:MYSTEPFUNCTION`;
    const name = `Manual step`;

    const executionParams = { name, stateMachineArn };
    return sfnClient.startExecution(executionParams).promise()
           .then(response => {
                if (response && response.executionArn) { 
                    print(`Started SFN execution for arn: ${response.executionArn}`);
                }
           })
};

• 现有状态机定义配置：

{
  "Comment": "My state machine",
  "StartAt": "Manual step",
  "States": {
    "ManualStep": {
      "Type": "Task",
      "Resource": "arn:aws:states:::activity:manualtask",
      "End": true
    }
  }
}

1. 实现逻辑说明与示例 #

Step Functions Activity任务的标准工作流如下：

• 提前在Step Functions控制台创建好Activity，获取对应的Activity ARN
• 状态机运行到Activity节点时，会将任务放入待轮询队列，不会主动触发回调
• 工作端轮询队列拿到待执行任务（获取唯一的taskToken），完成自定义逻辑后，携带taskToken调用SendTaskSuccess/SendTaskFailure接口标记任务结果，状态机才会继续向下执行

2. 是否必须部署常驻Activity Worker？ #

不是必须。不需要部署一直运行的常驻服务，只要本地运行的脚本可以完成「轮询拿到匹配的待执行任务taskToken → 调用结果接口回传状态」两个步骤即可，脚本执行完成后直接退出完全符合要求。

3. 本地脚本能不能直接调用接口标记等待中的任务为成功？ #

可以，但必须先拿到对应任务的taskToken。无法直接通过执行ARN、步骤名标记任务成功，Step Functions的任务结果回传接口强制要求传入taskToken，这个token只有轮询GetActivityTask接口时才会返回，没有其他获取渠道。 #

1. 修正状态机配置 #

现有状态机的StartAt值和定义的状态名不匹配，先修正这个语法错误：

{
  "Comment": "My state machine",
  "StartAt": "ManualStep",
  "States": {
    "ManualStep": {
      "Type": "Task",
      "Resource": "arn:aws:states:us-east-1:123456789012:activity:manualtask",
      "End": true
    }
  }
}

注意：Resource字段需要填写你在Step Functions控制台创建Activity后拿到的完整ARN，不要直接复制示例值。

2. 修改本地脚本逻辑 #

现有脚本只有启动状态机执行的逻辑，缺少轮询任务、回传成功结果的部分，修改后的完整脚本如下：

const AWS = require('aws-sdk');
const yargs = require('yargs/yargs');
const { hideBin } = require('yargs/helpers');

// 解析命令行传入参数
const argv = yargs(hideBin(process.argv)).argv;
const ACTIVITY_ARN = `arn:aws:states:us-east-1:123456789012:activity:manualtask`; // 替换为自己的Activity ARN
const TARGET_EXECUTION_ARN = argv.stepfunctionArn;
const MANUAL_INPUT = argv.input || 'manual task completed'; // 支持命令行传入要回传给状态机的内容

const main = async () => {
    let sfnClient;
    const rolename =  `StepFunctionExecuter-LOCAL`;
    // 初始化SFN客户端
    const creds = await getcreds({ accountId: '123456789012', region: 'us-east-1', rolename });
    sfnClient = new AWS.StepFunctions(creds);

    // 未传入执行ARN时先启动执行，传入则直接进入轮询逻辑
    if(!TARGET_EXECUTION_ARN) {
        const stateMachineArn = `arn:aws:states:us-east-1:123456789012:stateMachine:MYSTEPFUNCTION`;
        const startRes = await sfnClient.startExecution({
            stateMachineArn,
            name: `manual-exec-${Date.now()}`
        }).promise();
        console.log(`Started SFN execution, arn: ${startRes.executionArn}`);
    }

    console.log('Polling for pending activity task...');
    // 轮询待处理任务，该接口最长阻塞60秒等待任务到来
    const activityTask = await sfnClient.getActivityTask({
        activityArn: ACTIVITY_ARN
    }).promise();

    if(!activityTask.taskToken) {
        console.log('No pending task found, please run the script after state machine reaches the activity step');
        return;
    }

    // 校验任务归属（可选，避免误处理其他执行的任务）
    const taskInput = JSON.parse(activityTask.input);
    console.log(`Got pending task, input from state machine: ${JSON.stringify(taskInput)}`);

    // 调用接口标记任务成功，传入返回给状态机的结果
    await sfnClient.sendTaskSuccess({
        taskToken: activityTask.taskToken,
        output: JSON.stringify({
            result: MANUAL_INPUT,
            completeTime: new Date().toISOString()
        })
    }).promise();

    console.log('Task marked as success successfully');
}

main().catch(err => {
    console.error('Run failed:', err);
    process.exit(1);
});

3. 运行方式 #

1. 先安装本地依赖：npm install aws-sdk yargs
2. 启动状态机，等待执行到ManualStep节点时，运行脚本即可：

node report-choice-step-success.js --stepfunction-arn <你的执行ARN> --input "手动审批通过"

脚本运行后会自动获取等待中的任务token，标记任务成功，状态机随即执行结束。

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