This assumes you are familar with the basic configurations needed to connect to Amazon Web Services (AWS) using Python boto3. If not, check out my article Python (Scripting) - Getting Started with Amazon Web Services (AWS) boto3.

This also assumes you have created one or more Step Function State Machines.

Here is the minimal boilerplate code without any error handling to list your EC2 instances. This will use the default profile in your .aws/credentials profile. If you want to used a named profile, check out my article Python (Scripting) - Getting Started with Amazon Web Services (AWS) boto3.

import boto3
import os

client = boto3.client('ec2')
response = client.describe_instances(
    Filters = [
        { 'Name': 'tag:environment', 'Values': ['staging'] },
        { 'Name': 'tag:purpose',     'Values': ['postgres'] }                         
    ]
)
print(f"response = {response}")

Which should return something like this.

{'stateMachines': [
  {'stateMachineArn': 'arn:aws:states:us-east-1:123456789012:stateMachine:myStateMachine', 
   'name': 'myStateMachine',
   'type': 'STANDARD',
   'creationDate': datetime.datetime(2023, 12, 27, 2, 39, 32, 497000, tzinfo=tzlocal())}
  ],
  'ResponseMetadata': {
    'RequestId': 'dcac2e33-effa-4f60-95c3-d7bbba7f1351',
    'HTTPStatusCode': 200,
    'HTTPHeaders': {
      'x-amzn-requestid': 'dcac2e33-effa-4f60-95c3-d7bbba7f1351',
      'date': 'Thu, 28 Dec 2023 12:07:48 GMT',
      'content-type': 'application/x-amz-json-1.0',
      'content-length': '211',
      'connection': 'keep-alive'},
      'RetryAttempts': 0
  }
}

The "default" profile in your .aws/credentials file will be used. Session can be used to use some other profile.

#!/usr/bin/python3
import boto3

session = boto3.Session(profile_name='johndoe')
client = session.client('ec2')

response = client.describe_instances(
    Filters = [
        { 'Name': 'tag:environment', 'Values': ['staging'] },
        { 'Name': 'tag:purpose',     'Values': ['postgres'] }                         
    ]
)
print(f"response = {response}")

Or, os.environ['AWS_PROFILE'] can be used to specify the profile in /home/john.doe/.aws/config and /home/john.doe/.aws/credentials to use.

import boto3
import os

os.environ['AWS_PROFILE'] = 'johndoe'

client = boto3.client('ec2')
response = client.describe_instances(
    Filters = [
        { 'Name': 'tag:environment', 'Values': ['staging'] },
        { 'Name': 'tag:purpose',     'Values': ['postgres'] }                         
    ]
)
print(f"response = {response}")

And here is how you can access certain key/values from the dictionary.

#!/usr/bin/python3
import boto3

client         = boto3.client('stepfunctions')
state_machines = client.list_state_machines()

for state_machine in state_machines['stateMachines']:
  print(f"state machine ARN = {state_machine['stateMachineArn']}")

Which should return something like this.

state machine ARN = arn:aws:states:us-east-1:123456789012:stateMachine:myStateMachine

Now with the ARN, you can use start_execution to execute the Step Function State Machine.

#!/usr/bin/python3
import boto3
import json
import sys

client         = boto3.client('stepfunctions')
state_machines = client.list_state_machines()

for state_machine in state_machines['stateMachines']:
  print(f"state machine ARN = {state_machine['stateMachineArn']}")

  payload = json.dumps({'email':'john.doe@example.com'})

  try:
    response = client.start_execution(stateMachineArn=state_machine['stateMachineArn'], input=payload)
  except Exception as exception:
    print(f"exception = {exception}")
  else:
    print(f"response = {response}")

And something like this should be returned.

{'executionArn': 'arn:aws:states:us-east-1:123456789012:execution:myStateMachine:60d888c1-a52b-461a-b5ae-b3eecba77fcf', 
 'startDate': datetime.datetime(2023, 12, 28, 12, 21, 14, 517000, tzinfo=tzlocal()), 
 'ResponseMetadata': {
   'RequestId': '5bcecf56-27f6-497a-83ad-377c5b625dae',
   'HTTPStatusCode': 200,
   'HTTPHeaders': {
      'x-amzn-requestid': '5bcecf56-27f6-497a-83ad-377c5b625dae',
      'date': 'Thu, 28 Dec 2023 12:21:14 GMT',
      'content-type': 'application/x-amz-json-1.0',
      'content-length': '162',
      'connection': 'keep-alive'
   },
   'RetryAttempts': 0
  }
}

You will probably also want to check the execution result and any response data, perhaps like this.

#!/usr/bin/python3
import boto3
import json
import sys

client         = boto3.client('stepfunctions')
state_machines = client.list_state_machines()

for state_machine in state_machines['stateMachines']:
  print(f"state machine ARN = {state_machine['stateMachineArn']}")

  payload = json.dumps({'email':'john.doe@example.com'})

  try:
    response = client.start_execution(stateMachineArn=state_machine['stateMachineArn'], input=payload)
  except Exception as exception:
    print(f"exception = {exception}")
  else:
    print(f"response = {response}")

  executionArn = start_execution_response['executionArn']

  attempt = 0

  while attempt < 30:
    try:
      describe_execution = client.describe_execution(executionArn=executionArn)
    except Exception as exception:
      print(f"exception = {exception}")
    else:
      print(f"status = {describe_execution['status']}")

      if describe_execution['status'] == "FAILED":
        print(f"error = {describe_execution['error']}")
        print(f"cause = {describe_execution['cause']}")
        break
      if describe_execution['status'] == "SUCCEEDED":
        json_loads = json.loads(describe_execution['output'])
        print(f"json_loads['ResponseBody']['result'] = {json_loads['ResponseBody']['result']}")
        print(f"json_loads['ResponseBody']['message'] = {json_loads['ResponseBody']['message']}")
        break

    attempt += 1
    time.sleep(1)