mcpField
S
step-functions

step-functions

github.com/itsmostafa云与 DevOps1.2k119 次安装2 天前更新
代码仓库

关于 step-functions

面向使用 AWS Step Functions 构建工作流的开发者的参考技能，介绍状态机设计、CLI 与 boto3 操作、错误处理和执行故障排查。

主要功能

  1. 01Standard 与 Express 工作流
  2. 02ASL 状态机模式
  3. 03分支、并行与 Map 状态
  4. 04AWS CLI 和 boto3 示例
  5. 05重试与错误处理
  6. 06执行故障排查

适用场景

  1. 01编排 AWS 服务任务
  2. 02添加条件或并行工作流步骤
  3. 03使用 Map 状态处理数组
  4. 04调查失败或停滞的执行

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/itsmostafa/aws-agent-skills --skill step-functions

安装后在 Claude Code 中输入/step-functions确认已加载。

附带文件workflow-patterns.md

详细介绍

AWS Step Functions

AWS Step Functions 是一项无服务器编排服务，可让你使用状态机构建和运行工作流。将多个 AWS 服务协调起来，构建业务关键型应用程序。

目录

核心概念

工作流类型
类型描述定价
Standard长时间运行、持久化、恰好执行一次按状态转换次数计费
Express高吞吐量、短时运行按执行次数计费（时间 + 内存）
状态类型
状态描述
Task执行任务（Lambda、API 调用）
Choice条件分支
Parallel并发执行分支
Map遍历数组
Wait延迟执行
Pass将输入传递为输出
Succeed成功结束
Fail以失败结束
Amazon States Language (ASL)

用于定义状态机的 JSON 语言。

常见模式

简单的 Lambda 工作流
{
  "Comment": "Process order workflow",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:ValidateOrder",
      "Next": "ProcessPayment"
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:ProcessPayment",
      "Next": "FulfillOrder"
    },
    "FulfillOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:FulfillOrder",
      "End": true
    }
  }
}
创建状态机

AWS CLI：

aws stepfunctions create-state-machine \
  --name OrderWorkflow \
  --definition file://workflow.json \
  --role-arn arn:aws:iam::123456789012:role/StepFunctionsRole \
  --type STANDARD

boto3：

import boto3
import json

sfn = boto3.client('stepfunctions')

definition = {
    "Comment": "Order workflow",
    "StartAt": "ProcessOrder",
    "States": {
        "ProcessOrder": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:...",
            "End": True
        }
    }
}

response = sfn.create_state_machine(
    name='OrderWorkflow',
    definition=json.dumps(definition),
    roleArn='arn:aws:iam::123456789012:role/StepFunctionsRole',
    type='STANDARD'
)
开始执行
import boto3
import json

sfn = boto3.client('stepfunctions')

response = sfn.start_execution(
    stateMachineArn='arn:aws:states:us-east-1:123456789012:stateMachine:OrderWorkflow',
    name='order-12345',
    input=json.dumps({
        'order_id': '12345',
        'customer_id': 'cust-789',
        'items': [{'product_id': 'prod-1', 'quantity': 2}]
    })
)

execution_arn = response['executionArn']
Choice 状态（条件逻辑）
{
  "StartAt": "CheckOrderValue",
  "States": {
    "CheckOrderValue": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.total",
          "NumericGreaterThan": 1000,
          "Next": "HighValueOrder"
        },
        {
          "Variable": "$.priority",
          "StringEquals": "rush",
          "Next": "RushOrder"
        }
      ],
      "Default": "StandardOrder"
    },
    "HighValueOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:ProcessHighValue",
      "End": true
    },
    "RushOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:ProcessRush",
      "End": true
    },
    "StandardOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:ProcessStandard",
      "End": true
    }
  }
}
并行执行
{
  "StartAt": "ProcessInParallel",
  "States": {
    "ProcessInParallel": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "UpdateInventory",
          "States": {
            "UpdateInventory": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:...:function:UpdateInventory",
              "End": true
            }
          }
        },
        {
          "StartAt": "SendNotification",
          "States": {
            "SendNotification": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:...:function:SendNotification",
              "End": true
            }
          }
        },
        {
          "StartAt": "UpdateAnalytics",
          "States": {
            "UpdateAnalytics": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:...:function:UpdateAnalytics",
              "End": true
            }
          }
        }
      ],
      "Next": "Complete"
    },
    "Complete": {
      "Type": "Succeed"
    }
  }
}
Map 状态（迭代）
{
  "StartAt": "ProcessItems",
  "States": {
    "ProcessItems": {
      "Type": "Map",
      "ItemsPath": "$.items",
      "MaxConcurrency": 10,
      "Iterator": {
        "StartAt": "ProcessItem",
        "States": {
          "ProcessItem": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:...:function:ProcessItem",
            "End": true
          }
        }
      },
      "ResultPath": "$.processedItems",
      "End": true
    }
  }
}
错误处理
{
  "StartAt": "ProcessWithRetry",
  "States": {
    "ProcessWithRetry": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:Process",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException", "Lambda.TooManyRequestsException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 6,
          "BackoffRate": 2
        },
        {
          "ErrorEquals": ["States.Timeout"],
          "IntervalSeconds": 5,
          "MaxAttempts": 3,
          "BackoffRate": 1.5
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["CustomError"],
          "ResultPath": "$.error",
          "Next": "HandleCustomError"
        },
        {
          "ErrorEquals": ["States.ALL"],
          "ResultPath": "$.error",
          "Next": "HandleAllErrors"
        }
      ],
      "End": true
    },
    "HandleCustomError": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:HandleCustom",
      "End": true
    },
    "HandleAllErrors": {
      "Type": "Fail",
      "Error": "ProcessingFailed",
      "Cause": "An error occurred during processing"
    }
  }
}

CLI 参考

状态机管理
命令描述
aws stepfunctions create-state-machine创建状态机
aws stepfunctions update-state-machine更新定义
aws stepfunctions delete-state-machine删除状态机
aws stepfunctions list-state-machines列出状态机
aws stepfunctions describe-state-machine获取详细信息
执行
命令描述
aws stepfunctions start-execution开始执行
aws stepfunctions stop-execution停止执行
aws stepfunctions describe-execution获取执行详细信息
aws stepfunctions list-executions列出执行
aws stepfunctions get-execution-history获取执行历史记录

最佳实践

设计
  • 让状态各司其职 — 每个状态只承担一个目的
  • 使用有意义的状态名称
  • 实施全面的错误处理
  • 使用 Parallel 执行独立任务
  • 使用 Map 进行批处理
性能
  • 对于高吞吐量、短时任务，使用 Express 工作流
  • 设置适当的超时时间
  • 限制 Map 并发数，避免触发限流
  • 尽可能使用 SDK 集成（避免使用 Lambda 包装器）
可靠性
  • 重试瞬时错误
  • 捕获并处理特定错误
  • 使用幂等操作
  • 启用 X-Ray 跟踪
成本优化
  • 短工作流（< 5 分钟）使用 Express
  • 合并相关操作，减少状态转换次数
  • 使用 Wait 状态代替 Lambda 延迟

故障排除

执行失败
# Get execution history
aws stepfunctions get-execution-history \
  --execution-arn arn:aws:states:us-east-1:123456789012:execution:MyWorkflow:exec-123 \
  --query 'events[?type==`TaskFailed` || type==`ExecutionFailed`]'
Lambda 超时

原因：

  • Lambda 运行时间过长
  • 任务超时时间过短

解决方法：

{
  "Type": "Task",
  "Resource": "arn:aws:lambda:...",
  "TimeoutSeconds": 300,
  "HeartbeatSeconds": 60
}
状态卡住

检查：

  • Task 状态正在等待回调
  • Wait 状态的等待时间尚未结束
  • 活动工作器无响应
状态机无效
# Validate definition
aws stepfunctions validate-state-machine-definition \
  --definition file://workflow.json

参考资料

评论

支持 Markdown