如何在Python中获取运行中Step Functions启动的ECS Fargate任务ARN及详情

我通过Python文件启动一个AWS Step Functions状态机，该状态机会启动ECS Fargate任务。请问在状态机运行期间，能否在Python代码中获取这个运行中的ECS Fargate任务的ARN及详情？

更新：以下是我使用的Python代码片段：

import boto3
import time
import sys
import json

def main():
    client = boto3.client('stepfunctions')
    
    # 为了演示强制进入分支
    num = 1
    if num == 1:
        # 启动状态机执行
        response = client.start_execution(
            stateMachineArn='arn:aws:states:us-west-2:1234567890:stateMachine:stateMachine',
            input='{"in_put": "{totalPartitionCount:3}"}'
        )

    while True:
        time.sleep(2) 
 
        # 查询状态机执行状态
        sf_response = client.describe_execution(executionArn=response['executionArn'])
        status = sf_response['status'] 

        # 我现在每2秒获取一次状态机状态，想问能不能同时从状态机执行信息里拿到ECS任务的ARN/详情？
        # 我还想用ECS API在这个循环里打印ECS任务的相关信息

        if status == 'RUNNING':
            print("%s: %s" % ("> Status...", status))
            continue
        elif status == 'FAILED':
            sys.exit()
            raise Exception("%s: %s" % ("! ERROR ! Execution FAILED: ", sf_response))
        else:
            # SUCCEEDED, ABORTED 等其他状态
            print(json.dumps(sf_response, indent=4, sort_keys=True, default=str))
            break

    # 状态机执行完成后打印输出
    print("--------------------------------------------------")
    print(json.dumps(sf_response['output'], indent=4, sort_keys=True, default=str))
    print('--------------------------------------------------')

可以实现，以下是几种可行的方法：

方法1：从状态机执行历史中提取ECS任务ARN #

通过调用Step Functions的get_execution_history API，在状态机运行期间获取执行事件，从中筛选出ECS任务启动成功的事件，提取任务ARN。

修改Python循环中的RUNNING分支代码：

if status == 'RUNNING':
    print("%s: %s" % ("> Status...", status))
    
    # 获取状态机执行历史
    history_response = client.get_execution_history(
        executionArn=response['executionArn'],
        reverseOrder=True,
        maxResults=10
    )
    
    # 遍历事件找ECS任务执行结果
    ecs_client = boto3.client('ecs')
    for event in history_response['events']:
        if event['type'] == 'TaskSucceeded' and 'ecs:runTask' in event.get('taskSucceededEventDetails', {}).get('resource', ''):
            ecs_task_output = json.loads(event['taskSucceededEventDetails']['output'])
            ecs_task_arn = ecs_task_output['tasks'][0]['taskArn']
            print(f"ECS任务ARN: {ecs_task_arn}")
            
            # 调用ECS API获取任务详情
            task_details = ecs_client.describe_tasks(
                cluster='你的ECS集群名称/ARN',
                tasks=[ecs_task_arn]
            )
            print(json.dumps(task_details, indent=4, default=str))
            break
    continue

方法2：通过ECS API直接筛选匹配任务 #

如果不想依赖状态机执行历史，可以根据ECS任务的特征（比如任务定义ARN、集群、启动时间），直接调用ECS API查询运行中的任务并匹配目标任务。

修改Python循环中的RUNNING分支代码：

if status == 'RUNNING':
    print("%s: %s" % ("> Status...", status))
    
    ecs_client = boto3.client('ecs')
    # 列出集群中运行的目标任务定义对应的任务
    tasks_response = ecs_client.list_tasks(
        cluster='你的ECS集群名称/ARN',
        taskDefinition='你的ECS任务定义ARN',
        desiredStatus='RUNNING'
    )
    
    if tasks_response['taskArns']:
        task_details = ecs_client.describe_tasks(
            cluster='你的ECS集群名称/ARN',
            tasks=tasks_response['taskArns']
        )
        # 匹配状态机启动时间附近的任务（避免匹配旧任务）
        execution_start_time = sf_response['startDate']
        for task in task_details['tasks']:
            task_start_time = task['startedAt']
            if abs((task_start_time - execution_start_time).total_seconds()) < 60:
                print(f"匹配到ECS任务ARN: {task['taskArn']}")
                print(json.dumps(task, indent=4, default=str))
                break
    continue

方法3：修改状态机配置，输出ECS任务详情 #

如果有权限修改状态机定义，可以在调用ECS任务的步骤中，通过ResultPath保留ECS返回的任务详情，后续直接从状态机执行输出中获取。

状态机中ECS任务步骤示例：

"RunECSTask": {
  "Type": "Task",
  "Resource": "arn:aws:states:::ecs:runTask.sync",
  "Parameters": {
    "Cluster": "你的ECS集群ARN",
    "TaskDefinition": "你的ECS任务定义ARN",
    "LaunchType": "FARGATE"
  },
  "ResultPath": "$.ecsTaskInfo",
  "Next": "后续状态"
}

这种情况下，状态机执行完成后，sf_response['output']会包含ecsTaskInfo字段，里面有任务ARN和详情。如果要在运行期间获取，还是需要结合方法1的执行历史查询。

• 确保Python程序使用的IAM角色拥有以下权限：states:GetExecutionHistory、ecs:DescribeTasks、ecs:ListTasks。
• 方法1的准确性最高，直接从状态机执行事件中获取，不会出现匹配错误；方法2适合无法修改状态机的场景，但需要确保任务筛选条件足够精准。

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