Overview

This guide walks through setting up a custom ETL connector for AWS Step Functions using a pre-built example implementation.

Step Functions is AWS's serverless workflow orchestrator. This example connector talks to the Step Functions API with boto3 to collect state-machine metadata and execution history, so Monte Carlo can track runs, alert on failures, and surface a performance dashboard. The agent framework handles scheduling, pagination, and delivery — you only supply credentials.

This implementation is available as a reference in the mcd-public-resources repository. You can use it as-is or as a starting point for your own customizations.

Feature support

Prerequisites

• A running custom-connector-setup repository — see Custom ETL Connectors for initial setup.
• AWS credentials that can reach the Step Functions API in the target account and region (see Configure credentials for the options).
• An IAM policy granting the read-only Step Functions actions listed below.

Step Functions–specific notes

Understanding how Step Functions' concepts map to Monte Carlo's ETL data model is helpful if you plan to customize the connector:

• Jobs are discovered by listing state machines — fetch_metadata lists every state machine in the account/region and parses each one's Amazon States Language (ASL) definition into tasks, recursing into Parallel branches and Map processors so nested states appear too. Task edges (Next/Choices/Catch) become the job's DAG.
• Job-to-job lineage — a state that starts another state machine via the states:startExecution integration records the child's ARN as a triggered job, so Monte Carlo links parent → child state machines automatically.
• Runs of any trigger are reported — polling lists all executions in the collection window regardless of what started them. (Step Functions doesn't expose how an execution was triggered, so no trigger type is reported.)
• No schedule — state machines are triggered externally (e.g. EventBridge Scheduler), so the connector emits no schedule.
• EXPRESS workflows are skipped — ListExecutions doesn't support EXPRESS state machines (their history lives in CloudWatch Logs). STANDARD workflows are fully supported.
• Distributed Map — child iterations run as separate executions with their own ARNs, so per-item detail isn't nested under the parent run's task_runs. Inline Map/Parallel iterations are captured.

Setup

1. Scaffold the connector

Use the create_connector script with the --etl flag to scaffold the connector directory. This generates a unique identifier for your connector in manifest.json — this ID is how Monte Carlo distinguishes connector types, so it must be generated per-installation.

python scripts/create_connector.py aws_step_functions --etl

You will be asked what the Step Functions name for "Group", "Job" and "Task" are. This will change how these assets are named in Monte Carlo. Step Functions has no group concept, so leave that prompt blank. We recommend:

What does this tool call a group of jobs? (optional, blank to skip): 
What does this tool call a job? (default: Job): State Machine
What does this tool call a task? (default: Task): State
Icon URL (leave blank to skip): 

The script also prompts for an optional icon URL to display the Step Functions logo alongside the connection in the Monte Carlo UI.

You will now have a manifest.json file created. Replace run_status_mapping and credentials_schema with the following:

  "run_status_mapping": {
    "SUCCEEDED": "success",
    "FAILED": "failed",
    "RUNNING": "in_progress",
    "TIMED_OUT": "timed_out",
    "ABORTED": "cancelled",
    "PENDING_REDRIVE": "queued"
  },
  "credentials_schema": {
    "connect_args": {
      "type": "dict",
      "required": true,
      "schema": {
        "region_name":           { "type": "string", "required": true,  "empty": false },
        "role_arn":              { "type": "string", "required": false },
        "external_id":           { "type": "string", "required": false },
        "role_session_name":     { "type": "string", "required": false },
        "aws_access_key_id":     { "type": "string", "required": false },
        "aws_secret_access_key": { "type": "string", "required": false },
        "aws_session_token":     { "type": "string", "required": false },
        "endpoint_url":          { "type": "string", "required": false }
      }
    }
  }

The full manifest file should look something like this with a unique connection_type:

{
  "connection_type": "custom-etl-connector-XXXXXXX",
  "connection_name": "aws_step_functions",
  "asset_class": "etl",
  "terminology": {
    "job": "State Machine",
    "task": "State"
  },
  "run_status_mapping": {
    "SUCCEEDED": "success",
    "FAILED": "failed",
    "RUNNING": "in_progress",
    "TIMED_OUT": "timed_out",
    "ABORTED": "cancelled",
    "PENDING_REDRIVE": "queued"
  },
  "credentials_schema": {
    "connect_args": {
      "type": "dict",
      "required": true,
      "schema": {
        "region_name":           { "type": "string", "required": true,  "empty": false },
        "role_arn":              { "type": "string", "required": false },
        "external_id":           { "type": "string", "required": false },
        "role_session_name":     { "type": "string", "required": false },
        "aws_access_key_id":     { "type": "string", "required": false },
        "aws_secret_access_key": { "type": "string", "required": false },
        "aws_session_token":     { "type": "string", "required": false },
        "endpoint_url":          { "type": "string", "required": false }
      }
    }
  }
}

2. Replace with the example implementation

Clone or download the example from mcd-public-resources and replace the stub connector.py with the pre-built implementation:

cp <path-to-mcd-public-resources>/custom_connectors/aws_step_functions/connector.py etl_connectors/aws_step_functions/

The connector uses boto3, so add it to the connector's requirements.txt:

boto3>=1.34.0

No Dockerfile.extra is needed — boto3 is a pure-Python package with no system dependencies.

3. Configure credentials

Step Functions is an AWS service, so the connector authenticates with standard AWS credentials. region_name is always required; how it authenticates depends on where the agent runs. Two things are independent here:

• Where the agent runs — any cloud or on-prem. Flexible.
• What it talks to — the Step Functions API, which is always AWS.

So the connector always needs AWS credentials; the only deployment-dependent choice is how they're supplied. You provide them through self-hosted credentials — the values you put in connect_args become the connection's credentials. Create etl_connectors/aws_step_functions/credentials.json using one of the models below.

Option A — Static IAM user access key (works from any agent host). A long-lived access key (AKIA…) works everywhere, including local testing. aws_session_token is only needed for temporary (ASIA…) credentials, which expire:

{
  "connect_args": {
    "region_name": "us-east-1",
    "aws_access_key_id": "AKIA...",
    "aws_secret_access_key": "..."
  }
}

Option B — Assume an IAM role (auto-refreshing; works from any agent host). The connector calls sts:AssumeRole and refreshes the temporary credentials automatically, so nothing expires. This avoids a long-lived key with broad access — you hand out only a scoped, assumable role. The base identity used to assume the role comes from the keys in Option A, or from the pod/instance role when the agent runs in AWS:

{
  "connect_args": {
    "region_name": "us-east-1",
    "aws_access_key_id": "AKIA...",
    "aws_secret_access_key": "...",
    "role_arn": "arn:aws:iam::<account-id>:role/mc-stepfunctions-readonly",
    "external_id": "<external-id-if-required>"
  }
}

Option C — Pod / instance role (AWS-hosted agents only). When the agent runs in AWS — EKS with IRSA / EKS Pod Identity, EC2 with an instance profile, or ECS with a task role — boto3's default credential chain resolves the attached role automatically, so no secrets are needed:

{
  "connect_args": {
    "region_name": "us-east-1"
  }
}

IAM permissions

The connector makes only these read-only calls. Attach this policy to the identity (or role) it uses:

{
  "Version": "2012-10-17",
  "Statement": [{
    "Sid": "MonteCarloStepFunctionsRead",
    "Effect": "Allow",
    "Action": [
      "states:ListStateMachines",
      "states:DescribeStateMachine",
      "states:ListExecutions",
      "states:DescribeExecution",
      "states:GetExecutionHistory"
    ],
    "Resource": "*"
  }]
}

ListStateMachines is account-scoped and needs Resource: "*"; the others can be narrowed to specific ARNs. The AWS managed policy AWSStepFunctionsReadOnlyAccess also covers all of these.

4. Build and run the test suite

Build the test Docker image, then run each test section to validate the connector against the Step Functions API:

docker compose build

# Verify connection (builds the client and lists state machines)
CONNECTOR=aws_step_functions docker compose run --rm test -m etl_connection

# Metadata collection (state machines + parsed task DAG)
CONNECTOR=aws_step_functions docker compose run --rm test -m etl_metadata

# Run details (polling + webhook modes)
CONNECTOR=aws_step_functions docker compose run --rm test -m etl_run_details

The test suite validates that the connector returns correctly structured dicts matching the EtlAsset and EtlRunEvent schemas. The collection window defaults to 7 days (configurable via the ETL_TEST_WINDOW_HOURS env var) — make sure the account has recent executions so the run-detail tests have data.

5. Export and build the agent image

Once all tests pass, build the deployable image:

python scripts/generate_agent_image.py aws_step_functions

6. Deploy, register, and connect

From here, follow the standard custom ETL connector workflow:

1. Push the image to your container registry and deploy the agent, providing your production credentials via self-hosted credentials. If the agent is deployed in AWS, you can use the pod/instance-role model (Option C) otherwise supply a static key or assumed role (Options A/B).
2. Register the connector by running agent validations in the Monte Carlo UI (Settings → Deployments → your agent → Validate).
3. Add the connection through the Monte Carlo UI (Settings → Integrations → Add → Orchestration → Custom ETL integration). Select Collect to have Monte Carlo pull on a schedule, and configure the webhook for near-real-time failure detection.

After setup, Monte Carlo collects state-machine metadata on the agent's schedule and polls run details every 60 minutes.

Near-real-time failure alerts (webhook)

Polling already catches failures with no extra setup — the agent collects run details every 60 minutes, so a failure surfaces within ~60 minutes. For near-real-time alerting, Monte Carlo exposes a webhook you POST to when an execution fails.

Step Functions publishes execution state changes to Amazon EventBridge automatically, and EventBridge can call the Monte Carlo webhook directly through an API Destination — no forwarding proxy required, because the Connection sets the required headers for you:

You get both values when you add the Custom ETL connection in Monte Carlo (step 6) — after choosing Collect, Monte Carlo lands you on a page with the webhook URL and a button to generate the integration key (its ID and secret).

flowchart LR
    SF["Step Functions<br>execution fails"] --> EB["EventBridge rule<br>(FAILED / TIMED_OUT / ABORTED)"]
    EB --> AD["API Destination<br>Connection adds x-mcd-id / x-mcd-token"]
    AD --> MC["Monte Carlo webhook"]
    MC --> A["Run collection<br>(failure alert)"]

Set it up with four resources. Fill in <INTEGRATION_KEY_ID>, <INTEGRATION_TOKEN>, and <MC_WEBHOOK_URL> from step 6.

1. Connection — holds the auth headers (stored encrypted):

aws events create-connection \
  --name monte-carlo-webhook \
  --authorization-type API_KEY \
  --auth-parameters '{
    "ApiKeyAuthParameters": { "ApiKeyName": "x-mcd-token", "ApiKeyValue": "<INTEGRATION_TOKEN>" },
    "InvocationHttpParameters": {
      "HeaderParameters": [
        { "Key": "x-mcd-id", "Value": "<INTEGRATION_KEY_ID>", "IsValueSecret": true }
      ]
    }
  }'

2. API Destination — the Monte Carlo webhook endpoint:

aws events create-api-destination \
  --name monte-carlo-webhook \
  --connection-arn <CONNECTION_ARN> \
  --invocation-endpoint "<MC_WEBHOOK_URL>" \
  --http-method POST

3. Rule — match failed executions (add stateMachineArn under detail to scope to specific jobs):

aws events put-rule \
  --name step-functions-failures \
  --event-pattern '{
    "source": ["aws.states"],
    "detail-type": ["Step Functions Execution Status Change"],
    "detail": { "status": ["FAILED", "TIMED_OUT", "ABORTED"] }
  }'

4. Target — pass the failed execution's ARN as the job_run_id query parameter so Monte Carlo collects just that run:

aws events put-targets \
  --rule step-functions-failures \
  --targets '[{
    "Id": "mc-webhook",
    "Arn": "<API_DESTINATION_ARN>",
    "RoleArn": "<EVENTBRIDGE_INVOKE_ROLE_ARN>",
    "HttpParameters": {
      "QueryStringParameters": { "job_run_id": "$.detail.executionArn" }
    }
  }]'

$.detail.executionArn is the failed execution's run_source_id; the RoleArn is an IAM role EventBridge assumes to call the destination (it needs events:InvokeApiDestination on the destination ARN).

Lineage via SQL query tagging

Table ↔ state-machine lineage is not available from the Step Functions API — states invoke services (Lambda, Glue, ECS, …), not warehouse tables — so the connector intentionally omits inputs/outputs. To get warehouse lineage, tag the SQL that runs inside your states with a JSON comment carrying mcd_job_id set to the connector's job_source_id (the stateMachineArn):

-- {"mcd_job_id": "arn:aws:states:us-east-1:123456789012:stateMachine:MyJob"}
CREATE TABLE ... AS SELECT ...;

Optionally add mcd_task_id (a state name, i.e. a task's task_source_id) to attribute lineage to a specific state, or mcd_resource_id (the ETL connection's resource UUID) to disambiguate when multiple connections share an ARN. Monte Carlo ingests these tags through its standard warehouse query-log collection and resolves them back to the jobs and tasks this connector reports. See Lineage Between Jobs and Tables for details.

Injecting the tag from the state machine definition

You don't have to hardcode the ARN. The Step Functions Context Object exposes the state machine ARN as $$.StateMachine.Id (the same value the connector uses as job_source_id) and the current state name as $$.State.Name.

Direct SQL integrations (Athena, Redshift Data API) — the query text is a Parameters field, so prepend the tag comment right in the definition with intrinsic functions:

{
  "Type": "Task",
  "Resource": "arn:aws:states:::athena:startQueryExecution.sync",
  "Parameters": {
    "QueryString.$": "States.Format('-- \\{\"mcd_job_id\":\"{}\",\"mcd_task_id\":\"{}\"\\}\n{}', $$.StateMachine.Id, $$.State.Name, $.sql)",
    "WorkGroup": "primary",
    "ResultConfiguration": { "OutputLocation": "s3://.../results/" }
  }
}

The \\{ / \\} escapes are required because States.Format uses {} as its own placeholders; $.sql is your actual query.

Lambda / Glue / ECS that run SQL internally — the definition can't reach the SQL, so pass the ids in as input and apply them in code:

{
  "Type": "Task",
  "Resource": "arn:aws:states:::lambda:invoke",
  "Parameters": {
    "FunctionName": "run-warehouse-sql",
    "Payload": {
      "mcd_job_id.$": "$$.StateMachine.Id",
      "mcd_task_id.$": "$$.State.Name",
      "sql.$": "$.sql"
    }
  }
}

# in the Lambda (Snowflake example)
cur.execute("ALTER SESSION SET QUERY_TAG = %s", (json.dumps({
    "mcd_job_id": event["mcd_job_id"],
    "mcd_task_id": event["mcd_task_id"],
}),))

Ingest the warehouse's query logs

SQL query tagging only works if Monte Carlo is collecting query logs from the warehouse the states run against (Snowflake, BigQuery, etc.). That warehouse must be connected to Monte Carlo through its own separate integration with query log collection enabled — this is what makes the tables appear in the catalog and what lets Monte Carlo read the tags and resolve the state-machine → table edges. Without it, the tags are never ingested and no lineage is produced.