EC2-based Step Functions Worker

step-function.json

{
  "Comment": "Simple example that gets status code and puts in SSM Parameter Store",
  "StartAt": "process-input",
  "States": {
    "process-input": {
      "Type": "Task",
      "Resource": "arn:aws:states:::activity:process-input",
      "ResultPath": "$",
      "End": true
    }
  }
}

step_function_worker.py

#!/usr/bin/env python

import boto3
import time
import json
from botocore.config import Config as BotoCoreConfig

region = 'us-west-2'
boto_config = BotoCoreConfig(read_timeout=70, region_name=region)
sf = boto3.client('stepfunctions', config=boto_config)
activity_arn = 'arn:aws:states:::activity:process-input'

def processor(task_input):
    print(json.loads(task_input))
    return json.dumps(task_input)

while True:
    response = sf.get_activity_task(
        activityArn=activity_arn,
        workerName='step-function-worker'
    )
    if 'taskToken' not in response:
        print('INFO: No task token')
    else:
        print(response['taskToken'])
        print("===================")
        activity_token = response['taskToken']
        task_input = response['input']
        result = processor(task_input)
        sf.send_task_success(taskToken=activity_token, output=result)