Trigger an AWS Step Function with an API Gateway REST API using CDK

AWS documentation can be rough. Have you ever looked for an example of something you're trying to set up but only finding bits and pieces of what you need across several different sites? That was my experience recently when trying to set up a REST API with API Gateway that would trigger a Step Function.

There are some good tutorials and examples for doing this, just in the AWS console. What about infrastructure-as-code geeks? There isn't quite so much. And so, this tutorial. In it, you will learn how to use CDK to set up the following:

an API Gateway REST API that takes a single parameter

an IAM role that allows the API to connect to your Step Function

an API Gateway integration to connect your API to your Step Function, passing along a parameter

This tutorial is for current CDK users looking for examples of connecting AWS services like Step Functions to APIs set up in CDK. While it uses Python CDK, translating to Typescript or other languages should be trivial.

🚨 WARNING: 🚨 Deploying to AWS may incur charges. To ensure this doesn't happen, tear down any deployed resources with cdk destroy.

Define the Step Function

Define a step function as you usually would. For this article, let's assume you created a step function called item_step_function.

Create the API

Use aws_cdk.aws_apigateway's RestApi constructor to create the base API object. You will use this for all further API setup:

from aws_cdk import (
    core,
    aws_apigateway as apigateway,
    aws_iam as iam,
    aws_stepfunctions as sfn,
)
import json

item_step_function = sfn.StateMachine([...])
item_api = apigateway.RestApi(self, "item-api")

🚨 WARNING 🚨 This example code does not do any additional authorization beyond what is done by AWS by default. You may wish to add additional security measures for a production workload.

Set up Role

The earlier you set up the IAM permissions, the better. The proper IAM permissions will allow your API to trigger your step function. You will use the Role construct in the aws_iam package for this.

First, you will instantiate the Role, give it a name, and pick the service that will assume it. Since you want API Gateway to have access to Step Functions, you will use "apigateway.amazonaws.com":

item_api_role = iam.Role(
    self,
    f"item-api-role",
    role_name=f"item-api-role",
    assumed_by=iam.ServicePrincipal("apigateway.amazonaws.com"),
)

Once that is set up, you will need to add a policy to the Role, which defines the permissions that the Role grants to the service that assumes it. AWS IAM provides several managed policies that cover most use cases, so it is unlikely that you will need to craft your own. For this guide, you will use the AWSStepFunctionsFullAccess managed policy, but in most cases, you will want to use a more restrictive managed or custom-built policy.

item_api_role.add_managed_policy(
    iam.ManagedPolicy.from_aws_managed_policy_name("AWSStepFunctionsFullAccess")
)

With these two calls, you've created a role with a policy that will allow your API to interact with your Step Function. You will still need to link with this Role with the API itself, but that will come later.

Set up resources

Resources are any path-based pieces of your request URI. While you can nest resources using the .add_resource() method, you will add a single resource level for this tutorial. To use a resource as a parameter, surround your parameter name with curly braces. Note that you have to add it to the root of the API object:

step_function_trigger_resource = item_api.root.add_resource("{item_id}")

Your request URI will look something like [https://aws-generated-tld.com/1337](https://aws-generated-tld.com/1337) where 1337 is the item_id.

🚨 Note 🚨 You can also set up a query string parameter if you wish. For this tutorial, we will stick with path-based parameters.

Connect Your API to Your Step Function

CDK provides an AWSIntegration construct that is supposed to make it easier to integrate with other AWS services. It does not. At least, not by itself.

The AWSIntegration construct is difficult to use because implementations for different services aren't well-documented. You may not even know the internal service name for Step Functions or any other service you wish to integrate and have difficulty finding it. (if you do, the AWS CLI is here to help: aws list-services).

Request Templates

Before setting the integration itself, you need to set up a request template. A request template allows you to build the request you are making to your Step Function, including transmitting your API parameters to the Step Function.

The template is a dictionary and should have a single key of "application/json". Its value is a JSONified dictionary with your step function's ARN and input, which is part of the Step Function StartExecution request syntax. You will use some methods built into the request from Amazon, specifically $input.params(), which allows you to grab some or all of your request's parameters.

I strongly recommend you escape any Javascript by wrapping your $input.params() call with $util.escapeJavaScript():
"$util.escapeJavaScript($input.params('item_id'))".

Your template should look like this:

request_template = {
    "application/json": json.dumps(
        {
            "stateMachineArn": item_state_machine.state_machine_arn,
            "input": "{\"item_id\": \"$util.escapeJavaScript($input.params('item_id'))\"}",
        }
    )
}

Step Function Integration

Next, you will define the integration itself. The integration requires you to set a few parameters, but it's not always clear from the CDK documents which are the correct ones to set. This ambiguity is thanks to how general the AWSIntegration construct is: it allows you to use any service, but you have to know what parameters their requests need.

For all integrations, you need to provide the service name. For Step Functions, it's "states". Then you need to provide the action you want to do. To start a Step Function execution, you'll use "StartExecution". This is determined again by the service's API, and you can read more about "StartExecution" in the AWS Step Function documentation.

Then, you'll provide options. In CDK, these are IntegrationOptions. Here, you can define many options, but for this tutorial, the important ones are:

credentials_role: This will take the item_api_role you set up earlier, attaching the Role (and its attached policy) to the API itself via the integration.

integration_responses: This is a list of possible responses to API requests. At the very least, you'll want to return an IntegrationResponse object with a 200 status code, but you can define all sorts of situations that would trigger different status codes.

request_templates: This is where you'll attach the request template you made in the pre

Investigate these options and determine which options are right for your use case. To keep things simple, this example will pass credentials through the integration to the integrated service and only return a 200 response:

item_sfn_integration = apigateway.AwsIntegration(
    service="states",
    action="StartExecution",
    options=apigateway.IntegrationOptions(
        credentials_role=item_api_role,
        integration_responses=[
            apigateway.IntegrationResponse(status_code="200")
        ],
        request_templates=request_template,
    ),
)

Here, you created an integration between the Step Functions service and the API itself. You can think of the integration as the portal that your parameters pass through when traveling from the API to your integrated service.

Connect Integration to REST verbs

Do you remember the resource you set up earlier, defining the item_id parameter for the API? This object comes with an add_method() function, which you can use to connect a REST verb (such as GET, POST, PUT, etc.) to your integration. Doing this will allow a request using the correct verb to reach your integration.

Since you're sending data via the API, you'll use POST and wire it to item_sfn_integration like so:

step_function_trigger_resource.add_method(
    "POST",
    item_sfn_integration,
    method_responses=[apigateway.MethodResponse(status_code="200")],
)

Here, you connected your integration to your step_function_trigger_resource via the POST verb, and you set it to respond with a 200 response status. Like the integration, you can set multiple method response statuses.

Testing and Wrapping Up

To test this, deploy with cdk deploy <location>, open the API Gateway console, and navigate to your API and the REST verb you set up. Just click Test, add your parameter, and check the output. You can also navigate to the Step Functions console and check on your execution.

What's Next?

Now that you've learned how to wire an API up to a Step Function, you can do several things to dive deeper. Here are some suggestions:

Change your path parameter to a query string. How does this change how you set up the API and the service integration?

Make a more complex API. Multiple resources and multiple levels of resources. Can you mimic the structure of a public API, like Reddit's?

Full Example

from aws_cdk import (
    core,
    aws_apigateway as apigateway,
    aws_iam as iam,
    aws_stepfunctions as sfn,
)
import json

item_step_function = sfn.StateMachine([...])

# Initialize the API
item_api = apigateway.RestApi(self, "item-api")

# Set up IAM role and policy
item_api_role = iam.Role(
    self,
    f"item-api-role",
    role_name=f"item-api-role",
    assumed_by=iam.ServicePrincipal("apigateway.amazonaws.com"),
)
item_api_role.add_managed_policy(
    iam.ManagedPolicy.from_aws_managed_policy_name("AWSStepFunctionsFullAccess")
)

# Set up API resources
step_function_trigger_resource = item_api.root.add_resource("{item_id}")

# Set up request template and integration
request_template = {
    "application/json": json.dumps(
        {
            "stateMachineArn": item_state_machine.state_machine_arn,
            "input": "{\"item_id\": \"$util.escapeJavaScript($input.params('item_id'))\"}",
        }
    )
}
item_sfn_integration = apigateway.AwsIntegration(
    service="states",
    action="StartExecution",
    options=apigateway.IntegrationOptions(
        credentials_role=item_api_role,
        integration_responses=[
            apigateway.IntegrationResponse(status_code="200")
        ],
        request_templates=request_template,
    ),
)

# Connect integrations to REST verbs
step_function_trigger_resource.add_method(
    "POST",
    item_sfn_integration,
    method_responses=[apigateway.MethodResponse(status_code="200")],
)