API reference

Every public symbol is importable directly from the top-level aws_stepfunctions_toolkit package — there are no deep import paths. The reference below is generated from the source docstrings and tracks __all__, so it always matches what the package actually exports.

from aws_stepfunctions_toolkit import WorkflowRunner, DockerBatchStrategy, DockerfileImage

AWS Step Functions Toolkit.

Run Step Functions state machines end-to-end on your machine: the toolkit walks the state machine, uses the AWS test_state API for the engine-side logic (Path/Parameters/ResultSelector/next-state), and lets you plug in strategies for the states the test API can’t run remotely (.sync Batch jobs, .waitForTaskToken) — typically by building and running their container locally via Docker.

Everything you need is importable from this top-level package.

class aws_stepfunctions_toolkit.StateExecutionStrategy

Bases: ABC

class aws_stepfunctions_toolkit.DockerBatchStrategy(s3_bucket, image_source, execution_id=None, volumes=None, variables=None, user=None, gpus=None, extra_run_envs=None, region=None, s3_client=None)

Bases: StateExecutionStrategy

Runs a Batch (or any container) task locally via Docker.

The image is produced by a pluggable ImageSource (DockerfileImage for a plain Dockerfile build, PrebuiltImage for an existing/ECR image, BakeImage for docker buildx bake, or your own). This strategy only resolves the container’s Command/Environment from the state, runs the container, and reads its output.

The container is expected to write its result JSON to /tmp/output.json (the toolkit mounts a writable temp dir at /tmp and injects OUTPUT_PATH/S3_OUTPUT_PATH).

Parameters:

• s3_bucket (str)

• image_source (ImageSource)

• execution_id (str | None)

• volumes (list | None)

• variables (dict | None)

• user (str | None)

• gpus (str | None)

• extra_run_envs (dict | None)

• region (str | None)

• s3_client (S3Client | None)

aws_stepfunctions_toolkit.BatchImageStrategy

alias of DockerBatchStrategy

aws_stepfunctions_toolkit.LocalBatchImageStrategy

alias of DockerBatchStrategy

class aws_stepfunctions_toolkit.LocalExecutionStrategy(entrypoint=None, s3_bucket=None, execution_id=None, cwd=None, extra_env=None, inherit_env=True, variables=None)

Bases: StateExecutionStrategy

Run a step locally as a subprocess — directly in your terminal, no Docker.

This is the no-Docker counterpart to DockerBatchStrategy: it resolves the step’s Command + Environment from the ASL (via the test API), prepends entrypoint (the program to run — the local equivalent of a container’s ENTRYPOINT), injects OUTPUT_PATH (a temp file) and S3_OUTPUT_PATH, removes TaskToken, runs the process, and reads the result JSON the process writes to OUTPUT_PATH.

The same job code (e.g. one using BatchJobInterface) runs unchanged here or in a container, since both honor the OUTPUT_PATH contract.

LocalExecutionStrategy(entrypoint=[“python”, “jobs/process_data/main.py”])

Parameters:

• entrypoint (list[str] | None)

• s3_bucket (str | None)

• execution_id (str | None)

• cwd (str | None)

• extra_env (dict | None)

• inherit_env (bool)

• variables (dict | None)

class aws_stepfunctions_toolkit.CallableStrategy(handler)

Bases: StateExecutionStrategy

Wrap a plain function as a strategy — the simplest way to define your own handler.

handler receives the state’s input dict and returns the state’s result either as a dict/list (json-encoded for you) or as a pre-serialized JSON string.

CallableStrategy(lambda input_data: {“ok”: True, “echo”: input_data})

Parameters:

handler (Callable[[dict], dict | list | str])

class aws_stepfunctions_toolkit.StaticMockResponseStrategy(result)

Bases: StateExecutionStrategy

Returns a fixed, caller-supplied JSON string as the state’s result.

Parameters:

result (str)

class aws_stepfunctions_toolkit.BatchJobResponseStrategy(job_queue, job_definition, job_name=None, batch_client=None, variables=None, region=None)

Bases: StateExecutionStrategy

Submits a real AWS Batch job (rather than running the container locally).

Parameters:

• job_queue (str)

• job_definition (str)

• job_name (str | None)

• batch_client (BatchClient | None)

• variables (dict | None)

• region (str | None)

class aws_stepfunctions_toolkit.GetLatestConfigurationStrategy(application, environment, configuration_profile, appconfigdata_client=None, region=None)

Bases: StateExecutionStrategy

Resolves a state’s result from AWS AppConfig (start session + get latest configuration).

Parameters:

• application (str)

• environment (str)

• configuration_profile (str)

• appconfigdata_client (AppConfigDataClient | None)

• region (str | None)

class aws_stepfunctions_toolkit.AbstractMockMapResponseStrategy

Bases: StateExecutionStrategy, ABC

Base for Map states: implement get_items to supply the items to iterate.

Each item is run through the Map’s ItemProcessor via run_sub_machine.

class aws_stepfunctions_toolkit.StandardFlowStrategy

Bases: StateExecutionStrategy

Handles Map, Parallel, and Nested SMs via recursion.

aws_stepfunctions_toolkit.get_container_overrides(sfn_client, state_def, role_arn, variables, input_data, context)

Use the test API to resolve a Batch task’s ContainerOverrides (Command + Environment).

Runs test_state in TRACE mode with an empty mock so AWS evaluates the state’s Arguments/Parameters, then returns the resolved ContainerOverrides.

Parameters:

• state_def (dict)

• role_arn (str)

• variables (dict)

• input_data (dict)

• context (str | None)

Return type:

dict

class aws_stepfunctions_toolkit.ImageSource

Bases: ABC

Produces a locally-runnable Docker image reference.

abstractmethod ensure_image()

Build or pull as needed; return an image ref runnable by docker run.

Return type:

str

class aws_stepfunctions_toolkit.PrebuiltImage(image, login_ecr=False, region=None)

Bases: ImageSource

Run an image that already exists (local tag or a registry/ECR reference).

Set login_ecr=True to authenticate to ECR before the (implicit) pull.

Parameters:

• image (str)

• login_ecr (bool)

• region (str | None)

ensure_image()

Build or pull as needed; return an image ref runnable by docker run.

Return type:

str

class aws_stepfunctions_toolkit.DockerfileImage(context, dockerfile=None, build_args=None, tag=None, target=None, region=None, login_ecr=False)

Bases: ImageSource

Build an image from a plain Dockerfile + build context (the common dev path).

context is the build-context directory. dockerfile defaults to <context>/Dockerfile. tag is auto-generated if not given.

Parameters:

• context (str)

• dockerfile (str | None)

• build_args (dict | None)

• tag (str | None)

• target (str | None)

• region (str | None)

• login_ecr (bool)

ensure_image()

Build or pull as needed; return an image ref runnable by docker run.

Return type:

str

class aws_stepfunctions_toolkit.BakeImage(bake_file, target, tag=None, base_dir=None, bake_variables=None, codeartifact_domain=None, codeartifact_domain_owner=None, region=None)

Bases: ImageSource

Build an image via docker buildx bake (advanced / monorepo setups).

Optionally injects a BASE_DIR bake variable (base_dir) and a CODEARTIFACT_AUTH_TOKEN bake variable (when both codeartifact_* are set, for private package installs during the build).

Parameters:

• bake_file (str)

• target (str)

• tag (str | None)

• base_dir (str | None)

• bake_variables (dict | None)

• codeartifact_domain (str | None)

• codeartifact_domain_owner (str | None)

• region (str | None)

ensure_image()

Build or pull as needed; return an image ref runnable by docker run.

Return type:

str

aws_stepfunctions_toolkit.login_to_ecr(region=None)

Authenticate the local Docker daemon against the caller’s ECR registry.

Parameters:

region (str | None)

Return type:

str

aws_stepfunctions_toolkit.get_codeartifact_token(domain, domain_owner, region=None)

Fetch an AWS CodeArtifact authorization token (e.g. to inject into a Docker build).

Parameters:

• domain (str)

• domain_owner (str)

• region (str | None)

Return type:

str

class aws_stepfunctions_toolkit.ExecutionContext(*, Execution=_Execution(Id='String', Input='{}', Name='String', RoleArn='String', StartTime='2025-12-14T18:00:00Z', RedriveCount=12, RedriveTime='2025-12-14T18:00:00Z'), State=_State(EnteredTime='2025-12-14T18:00:00Z', Name='String', RetryCount=12), StateMachine=_StateMachine(Id='String', Name='String'), Task=None)

Bases: BaseModel

Parameters:

• Execution (_Execution)

• State (_State)

• StateMachine (_StateMachine)

• Task (_Task | None)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aws_stepfunctions_toolkit.StartExecutionResult(*, OutputDetails={'Included': False}, Input='', ExecutionArn='arn:aws:states:us-east-1:000000000000:execution:ExampleStateMachine:test', RedriveCount=0, InputDetails={'Included': False}, RedriveStatus='NOT_REDRIVABLE', RedriveStatusReason='Execution is SUCCEEDED and cannot be redriven', StartDate='1769015654832', StateMachineArn='arn:aws:states:us-east-1:000000000000:stateMachine:ExampleStateMachine', Status='SUCCEEDED', StopDate='1769015943200', Output='{}')

Bases: BaseModel

Parameters:

• OutputDetails (dict)

• Input (str)

• ExecutionArn (str)

• RedriveCount (int)

• InputDetails (dict)

• RedriveStatus (str)

• RedriveStatusReason (str)

• StartDate (str)

• StateMachineArn (str)

• Status (str)

• StopDate (str)

• Output (str)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aws_stepfunctions_toolkit.AslDefinition(*, StartAt, States, **extra_data)

Bases: BaseModel

Parameters:

• StartAt (str)

• States (dict)

• extra_data (Any)

model_config = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aws_stepfunctions_toolkit.AslDefinitionDict

Bases: TypedDict

class aws_stepfunctions_toolkit.DockerBatchConfig(*, s3_bucket, bake_file, volumes=[], variables={}, target_mapping, user=None)

Bases: BaseModel

Parameters:

• s3_bucket (str)

• bake_file (str)

• volumes (list)

• variables (dict)

• target_mapping (dict[str, str])

• user (str | None)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aws_stepfunctions_toolkit.ExecutionHistory(events)

Bases: object

Iterator for Step Functions execution history events.

Parameters:

events (list[HistoryEventTypeDef])

classmethod from_execution_arn(sfn_client, execution_arn)

Retrieve complete execution history with pagination.

Parameters:

• sfn_client (SFNClient)

• execution_arn (str)

Return type:

ExecutionHistory

iter(start=None)

Iterate over history events starting from given index.

Parameters:

start (int | HistoryEventTypeDef)

Return type:

Iterator[HistoryEventTypeDef]

class aws_stepfunctions_toolkit.EventFilter(events)

Bases: object

Filter methods for execution history events.

Parameters:

events (list[HistoryEventTypeDef])

by_type(event_type)

Return all events matching the given type.

Parameters:

event_type (str)

Return type:

list[HistoryEventTypeDef]

by_state_name(state_name)

Return all events from state entry until next state entry.

Parameters:

state_name (str)

Return type:

list[HistoryEventTypeDef]

by_resource_type(resource_type)

Return all task events matching the resource type.

Parameters:

resource_type (str)

Return type:

list[HistoryEventTypeDef]

by_resource(resource)

Return all task events matching the resource.

Parameters:

resource (str)

Return type:

list[HistoryEventTypeDef]

aws_stepfunctions_toolkit.generate_mock_data(execution_arn, output_dir=None, region=None, sfn_client=None)

Generate mock data files from Step Functions execution.

Parameters:

• execution_arn (str)

• output_dir (str | Path)

• region (str)

• sfn_client (SFNClient)

Return type:

dict[str, Any]

class aws_stepfunctions_toolkit.BatchJobInterface(logger=None, task_token_env_var='TaskToken', output_path_env_var='OUTPUT_PATH', test_mode_env_var='ENVIRONMENT', test_mode_values=('dev', 'test'), region=None)

Bases: ABC, Generic[InputT, OutputT]

Parameters:

• logger (logging.Logger | None)

• task_token_env_var (str)

• output_path_env_var (str)

• test_mode_env_var (str)

• test_mode_values (tuple[str, ...])

• region (str | None)

abstractmethod should_run(input_data)

Return True if the job should actually do its work for this input.

Parameters:

input_data (InputT)

Return type:

bool

abstractmethod run(input_data)

Do the work and return the typed output.

Parameters:

input_data (InputT)

Return type:

OutputT

abstractmethod create_skip_output(input_data)

Build the output to return when the job is skipped (test mode / should_run False).

Parameters:

input_data (InputT)

Return type:

OutputT

class aws_stepfunctions_toolkit.BasicJobInput(*, last_step_results, force=False)

Bases: BaseModel

Parameters:

• last_step_results (LastStepResults)

• force (bool)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aws_stepfunctions_toolkit.BasicJobOutput(*, filepath, did_run)

Bases: BaseModel

Parameters:

• filepath (str)

• did_run (bool)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class aws_stepfunctions_toolkit.LastStepResults(*, filepath)

Bases: BaseModel

Parameters:

filepath (str)

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].