AWS Step Functions Toolkit¶

Run a Step Functions state machine end-to-end on your laptop — and choose, per step, how each one runs.

This package was created out of the need to test a Step Functions pipeline built almost entirely from batch:submitJob.sync steps. AWS’s test_state API can’t run .sync integrations (or .waitForTaskToken), and deploying the real state machine to AWS to check each change was far too slow.

So instead, this toolkit walks your real ASL definition state-by-state, uses test_state for the engine logic (Path/Parameters/ResultSelector/Output + next state), and lets you plug in a strategy for the steps it can’t run — usually by building and running that step’s container locally with Docker. Run a whole pipeline locally and swap any step between a mock, your own function, a local container, or a real AWS call.

Install¶

pip install aws-stepfunctions-toolkit

Quick start¶

The snippet below shows the shape of a run — it is not runnable on its own (you supply your own state machine). The runnable starter lives in examples/quickstart/ (set ROLE_ARN and python run.py; no Docker — its task steps are mocked). See the Setup guide for AWS credentials and the test_state IAM role.

import json
from aws_stepfunctions_toolkit import WorkflowRunner, StaticMockResponseStrategy, CallableStrategy

# Bring your own: your Step Functions state machine, exported as Amazon States Language (ASL) JSON.
definition = json.loads(open("state_machine.asl.json").read())

# For each step you don't want to run for real, say how to produce its result.
mock_mapping = {
    "Enrich": CallableStrategy(lambda data: {"Payload": json.dumps({"enriched": True})}),
    "Notify": StaticMockResponseStrategy(json.dumps({"Payload": json.dumps({"status": "sent"})})),
}

runner = WorkflowRunner(
    role_arn="arn:aws:iam::<account>:role/<role-with-test-state-perms>",
    asl_registry={"main": definition},
    mock_mapping=mock_mapping,
)

final_output = runner.start(initial_input={"order_id": 123})
print(final_output)

States without an entry in mock_mapping are handled automatically (test_state for ordinary states; built-in recursion for Map / Parallel / nested state machines). To run a step in a real local container instead of mocking it, use DockerBatchStrategy.

Note

JSONata throughout. The examples (and the toolkit’s mock-result handling) use the JSONata query language. Set "QueryLanguage": "JSONata" on each state (not just at the top level): the runner tests one state at a time, so a state must declare its own query language.

Documentation¶

Guides

Requirements¶

Python 3.13+
An AWS role/credentials allowed to call test_state (the toolkit calls the real API for the engine logic).
Docker, only if you use DockerBatchStrategy to run steps in containers.

New to this? The Setup guide covers AWS credentials, creating the test_state IAM role, and Docker, step by step.

Examples¶

Each folder under examples/ is self-contained (its own ASL, runnable script, and README). Start with quickstart/ (copy-and-run, no Docker), local-subprocess/ (run a step’s code as a local subprocess), or docker-batch/ (real local containers plus a nested machine).

License¶

See LICENSE.