CDK定义StepFunctions时如何传入执行阶段才确定的ECS任务ARN

完全可以实现，推荐两种实现方式，你可以根据自己的需求选择：

方案1：使用CustomState直接复用原有JSON配置（最稳妥） #

CDK的EcsRunTask是强类型封装，默认要求传入已经实例化的TaskDefinition对象，不支持直接传入动态JSON路径。你可以直接用CustomState完全照搬你原有的状态机JSON配置，没有任何限制：

import aws_cdk, constructs
from aws_cdk import aws_stepfunctions, aws_ecs, Aws

class StepsStack(aws_cdk.Stack):
    def __init__(self, scope: constructs.Construct, construct_id: str, *, cluster: aws_ecs.ICluster, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # 直接定义自定义任务状态，完全匹配原有JSON配置
        run_fargate_task = aws_stepfunctions.CustomState(
            self,
            "RunFargateTask",
            state_json={
                "Type": "Task",
                "Resource": f"arn:{Aws.PARTITION}:states:::ecs:runTask.sync",
                "Parameters": {
                    "LaunchType": "FARGATE",
                    "Cluster": cluster.cluster_arn,
                    "TaskDefinition.$": "$.ecs_task_arn"
                },
                "ResultPath": None
            }
        )
        
        state_machine = aws_stepfunctions.StateMachine(
            self,
            "MyStateMachine",
            definition=run_fargate_task
        )

        # 手动给状态机执行角色添加ECS跑任务的权限
        state_machine.add_to_role_policy(
            aws_cdk.aws_iam.PolicyStatement(
                actions=["ecs:RunTask"],
                resources=["*"] # 建议缩小范围到你实际会用到的任务ARN前缀，避免权限过大
            )
        )
        # 添加PassRole权限，允许ECS使用任务定义对应的执行角色
        state_machine.add_to_role_policy(
            aws_cdk.aws_iam.PolicyStatement(
                actions=["iam:PassRole"],
                resources=["*"] # 同样建议缩小范围到任务执行角色的ARN范围
            )
        )

这个方案的优势是完全和你原有JSON定义的状态机行为一致，不需要适配CDK封装类的参数限制。

方案2：使用EcsRunTask + 参数覆盖 #

如果你更习惯使用CDK封装的任务类，可以先传入一个占位的任务定义对象，再通过add_override方法覆盖参数为动态路径：

import aws_cdk, constructs
from aws_cdk import aws_stepfunctions, aws_stepfunctions_tasks, aws_ecs

class StepsStack(aws_cdk.Stack):
    def __init__(self, scope: constructs.Construct, construct_id: str, *, cluster: aws_ecs.ICluster, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # 先创建一个占位的Fargate任务定义（只用来满足类型要求，不会实际使用）
        placeholder_task_def = aws_ecs.FargateTaskDefinition(self, "PlaceholderTask")
        
        run_task = aws_stepfunctions_tasks.EcsRunTask(
            self,
            "RunFargateTask",
            cluster=cluster,
            launch_target=aws_stepfunctions_tasks.EcsFargateLaunchTarget(platform_version=aws_ecs.FargatePlatformVersion.LATEST),
            task_definition=placeholder_task_def,
            result_path=aws_stepfunctions.ResultPath.DISCARD # 对应原有配置的ResultPath: null
        )
        
        # 覆盖任务定义参数为动态路径
        run_task.add_override("Parameters.TaskDefinition.$", "$.ecs_task_arn")
        # 删除默认生成的静态TaskDefinition参数
        run_task.add_override("Parameters.TaskDefinition", aws_cdk.RemovalPolicy.DESTROY)
        
        state_machine = aws_stepfunctions.StateMachine(
            self,
            "MyStateMachine",
            definition=run_task
        )
        # 同样需要添加对应的ECS和PassRole权限，和方案1一致

注意事项 #

两种方案都需要手动给状态机执行角色配置对应的ecs:RunTask和iam:PassRole权限，因为动态任务ARN的场景下CDK无法自动生成最小权限，你可以根据实际使用的任务ARN范围缩小权限资源配置，避免权限过大。

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