--- updatedAt: 2026-09-17T20:21:17.000Z agentTools: projectIndex: https://docs.getmontecarlo.com/llms.txt --- # Example: AWS Step Functions (Public Preview) # Overview This guide walks through setting up a [custom ETL connector](https://docs.getmontecarlo.com/docs/custom-connectors) for [AWS Step Functions](https://aws.amazon.com/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](https://github.com/monte-carlo-data/mcd-public-resources/tree/main/custom_connectors/aws_step_functions) repository. You can use it as-is or as a starting point for your own customizations. **One account + region per connection** The connector collects every state machine the supplied credentials can reach in a single AWS **account and region**. To monitor multiple regions or accounts, add one integration per account/region. # Feature support | Category | Capability | Support | | -------------- | -------------------------------------- | -------------------- | | Step Functions | Execution Failure Alerts | ✅ | | Step Functions | Webhook triggered collection | ✅ | | Step Functions | Run History | ✅ | | Lineage | Table-to-Table Lineage | ✅ | | Lineage | Step Function-to-Step Function Lineage | :white\_check\_mark: | | Lineage | Column-level Lineage | ❌ | Table-to-table lineage is produced via **SQL query tagging**, not declared `inputs`/`outputs` — the Step Functions API exposes which services a state invokes, but not the warehouse tables read or written. See [Lineage via SQL query tagging](#lineage-via-sql-query-tagging) below. Job-to-job lineage (one state machine starting another) is derived automatically from the definition. # Prerequisites * A running [custom-connector-setup](https://github.com/monte-carlo-data/custom-connector-setup) repository — see [Custom ETL Connectors](https://docs.getmontecarlo.com/docs/custom-connectors) for initial setup. * AWS credentials that can reach the Step Functions API in the target account and region (see [Configure credentials](#3-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: | Monte Carlo concept | Step Functions concept | | :------------------ | :------------------------------------------------------------------------------------------ | | **Connection** | An AWS **account + region** — all state machines the credentials can reach. | | **Group** | *Not used* — Step Functions has no notion of the same job in multiple named environments.;; | | **Job** | A **state machine**, keyed by its `stateMachineArn`. | | **Task** | A **state** in the state machine's definition, emitted at runtime as `task_runs`. | | **Run** | An **execution**, keyed by its `executionArn`. | * **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. ```bash 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: ```text bash 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: ```text "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`: ```json { "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](https://github.com/monte-carlo-data/mcd-public-resources/tree/main/custom_connectors/aws_step_functions) and replace the stub `connector.py` with the pre-built implementation: ```bash cp /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`: ```text 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](https://docs.getmontecarlo.com/docs/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. **The generic agent runs anywhere** (any Kubernetes cluster, Docker, on-prem, or a non-AWS cloud), so it usually has **no ambient AWS identity**. In that case you must supply explicit credentials — a static key (Option A) or an assumed role (Option B). Option C (no keys, relying on the pod/instance role) works **only when the agent is deployed in AWS**. **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: ```json { "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: ```json { "connect_args": { "region_name": "us-east-1", "aws_access_key_id": "AKIA...", "aws_secret_access_key": "...", "role_arn": "arn:aws:iam:::role/mc-stepfunctions-readonly", "external_id": "" } } ``` **Option C — Pod / instance role (AWS-hosted agents only).** When the agent runs in AWS — EKS with [IRSA / EKS Pod Identity](https://docs.aws.amazon.com/eks/latest/userguide/pod-identities.html), 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: ```json { "connect_args": { "region_name": "us-east-1" } } ``` For **Monte Carlo's Kubernetes agent on EKS**, the service account is bound to an IAM role via IRSA/Pod Identity — but that role is provisioned for the agent's own needs and does **not** include Step Functions permissions by default. To use Option C you must attach the read-only policy below to **that** role. If you'd rather not modify the agent's role, use Option A or B instead. | Field | Description | Default | | :--------------------------- | :------------------------------------------------------ | :---------------- | | **region\_name** | AWS region of the Step Functions service (required) | | | **role\_arn** | IAM role to assume (auto-refreshing STS credentials) | | | **external\_id** | External ID required by the assumed role's trust policy | | | **role\_session\_name** | STS session name for the role assumption | `monte-carlo-etl` | | **aws\_access\_key\_id** | Static access key — omit to use the default chain | | | **aws\_secret\_access\_key** | Static secret key | | | **aws\_session\_token** | Session token for temporary base credentials | | | **endpoint\_url** | Override the service endpoint (testing/localstack) | | ### IAM permissions The connector makes only these read-only calls. Attach this policy to the identity (or role) it uses: ```json { "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. **Assume-role trust policy (Option B).** The target role's trust policy must allow the base identity to assume it, and the base identity needs `sts:AssumeRole` on the role: ```json { "Version": "2012-10-17", "Statement": [{ "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam:::role/" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "sts:ExternalId": "" } } }] } ``` Drop the `Condition` block if you don't set `external_id`. ## 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: ```bash 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](https://docs.getmontecarlo.com/docs/push-ingest-api) and [EtlRunEvent](https://docs.getmontecarlo.com/docs/push-ingest-api) 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: ```bash 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](https://docs.getmontecarlo.com/docs/custom-connectors), providing your production credentials via [self-hosted credentials](https://docs.getmontecarlo.com/docs/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](https://docs.getmontecarlo.com/docs/custom-connectors) in the Monte Carlo UI (Settings → Deployments → your agent → Validate). 3. **Add the connection** through the [Monte Carlo UI](https://docs.getmontecarlo.com/docs/custom-connectors) (Settings → Integrations → Add → Orchestration → Custom ETL integration). Select **Collect** to have Monte Carlo pull on a schedule, and configure the [webhook](https://docs.getmontecarlo.com/docs/custom-connectors#webhook-collection) 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](https://docs.getmontecarlo.com/docs/custom-connectors#webhook-collection) you `POST` to when an execution fails. Step Functions publishes execution state changes to [Amazon EventBridge](https://docs.aws.amazon.com/step-functions/latest/dg/cw-events.html) 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: | Header | Value | | ----------- | ---------------------- | | x-mcd-id | Integration key ID | | x-mcd-token | Integration key secret | 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). ```mermaid flowchart LR SF["Step Functions
execution fails"] --> EB["EventBridge rule
(FAILED / TIMED_OUT / ABORTED)"] EB --> AD["API Destination
Connection adds x-mcd-id / x-mcd-token"] AD --> MC["Monte Carlo webhook"] MC --> A["Run collection
(failure alert)"] ``` Set it up with four resources. Fill in ``, ``, and `` from step 6. **1. Connection** — holds the auth headers (stored encrypted): ```bash aws events create-connection \ --name monte-carlo-webhook \ --authorization-type API_KEY \ --auth-parameters '{ "ApiKeyAuthParameters": { "ApiKeyName": "x-mcd-token", "ApiKeyValue": "" }, "InvocationHttpParameters": { "HeaderParameters": [ { "Key": "x-mcd-id", "Value": "", "IsValueSecret": true } ] } }' ``` **2. API Destination** — the Monte Carlo webhook endpoint: ```bash aws events create-api-destination \ --name monte-carlo-webhook \ --connection-arn \ --invocation-endpoint "" \ --http-method POST ``` **3. Rule** — match failed executions (add `stateMachineArn` under `detail` to scope to specific jobs): ```bash 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: ```bash aws events put-targets \ --rule step-functions-failures \ --targets '[{ "Id": "mc-webhook", "Arn": "", "RoleArn": "", "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). You can also forward to the **bare** webhook URL (omit the query parameter) — Monte Carlo then collects all runs from the last hour and picks up the failed one. Scoping with `job_run_id` just makes collection precise. # 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`): ```sql -- {"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](https://docs.getmontecarlo.com/docs/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](https://docs.aws.amazon.com/step-functions/latest/dg/input-output-contextobject.html) 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: ```json { "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: ```json { "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" } } } ``` ```python # 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"], }),)) ``` **ARN must match.** `$$.StateMachine.Id` is the base state-machine ARN, which matches the `job_source_id` the connector reports for unversioned machines. If you invoke via a **version** or **alias** ARN, the execution's `stateMachineArn` can differ from the base ARN, and lineage won't line up. ### 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.