I tried retrieving multiple SSM parameters in a single task using PythonOperator on Amazon MWAA Serverless

Hello. I'm Takeda from the Service Development Division.

Workflow definitions in Amazon MWAA Serverless are written in YAML. There are many situations where you want to store configuration values like bucket names and output destination prefixes in AWS Systems Manager Parameter Store and retrieve them at runtime. However, until now, MWAA Serverless had no easy way to retrieve multiple parameters at once. Even when retrieving them via Lambda, the LambdaInvokeFunctionOperator stores the response as a string in XCom, making it impossible to extract values by key from templates. Workarounds such as splitting tasks per parameter were necessary. Details of those workarounds are summarized in the following article.

https://dev.classmethod.jp/articles/amazon-mwaa-serverless-lambda-multi-value-workarounds/

In August 2026, PythonOperator and BashOperator became available in MWAA Serverless.

https://aws.amazon.com/about-aws/whats-new/2026/08/mwaa-serverless-pythonoperator-bashoperator/

If you can call boto3 from your own Python function, you should be able to use get_parameters to retrieve multiple parameters at once and store them as a dictionary in XCom. I verified this by actually running it.

For basic usage of PythonOperator, please refer to the following article.

https://dev.classmethod.jp/articles/update-amazon-mwaa-serverless-support-pythonoperator-bashoperator/

Returning a dictionary makes XCom a dictionary type

The key point is that when python_callable returns a dictionary, it is stored in XCom as a dictionary type. While MWAA Serverless templates have restrictions on dot access (.key), bracket access (['key']) works. With a dictionary-type XCom, individual values can be extracted from templates in downstream tasks.

First, I isolated and verified this behavior alone. At the same time, I also checked whether the task could reach SSM. The code is as follows.

# probe.py
import boto3
from botocore.config import Config

CFG = Config(connect_timeout=3, read_timeout=5, retries={"max_attempts": 0})


def get_config():
    try:
        boto3.client("ssm", config=CFG).get_parameter(Name="/myapp/dummy")
        ssm = "ok"
    except Exception as e:
        ssm = f"{e.__class__.__name__}"
    result = {"ssm": ssm, "data_bucket": "my-bucket", "output_prefix": "data/"}
    print(f"result = {result}")
    return result

The workflow definition is as follows. In the downstream BashOperator, three keys are extracted using bracket access.

pybash_ssmprobe:
  dag_id: pybash_ssmprobe
  schedule: null
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  tasks:
    get_config:
      operator: airflow.providers.standard.operators.python.PythonOperator
      task_id: get_config
      python_callable: probe.get_config
    bash_show:
      operator: airflow.providers.standard.operators.bash.BashOperator
      task_id: bash_show
      bash_command: "echo \"bucket={{ ti.xcom_pull(task_ids='get_config')['data_bucket'] }} prefix={{ ti.xcom_pull(task_ids='get_config')['output_prefix'] }} ssm={{ ti.xcom_pull(task_ids='get_config')['ssm'] }}\""
      dependencies:
        - get_config

When run without a VPC (NetworkConfiguration not specified), the workflow completed with SUCCESS, and the rendered output of bash_show was as follows.

bucket=my-bucket prefix=data/ ssm=ConnectTimeoutError

Two things became clear.

• Bracket access to a dictionary-type XCom works even with a PythonOperator return value (bucket= and prefix= are expanded)
• SSM cannot be reached from a task without a VPC (ConnectTimeoutError)

A custom VPC is required to reach SSM. The range of AWS services reachable from tasks without a VPC is verified in the following article.

https://dev.classmethod.jp/articles/amazon-mwaa-serverless-vpc-network-reachability/

Setting up a VPC and endpoints

I used a private routing configuration without a NAT gateway. The three VPC endpoints I prepared are as follows.

• S3 gateway endpoint (required for retrieving definitions and code)
• SSM interface endpoint (the target of this call)
• CloudWatch Logs interface endpoint (required to output task logs)

The VPC itself uses the 10.1.0.0/16 range with two private subnets in separate AZs, and a security group that allows inbound self-referencing. The SSM interface endpoint was created with the following command.

# SSM (interface endpoint, private DNS enabled)
aws ec2 create-vpc-endpoint --vpc-id vpc-xxxxxxxx \
  --service-name com.amazonaws.ap-northeast-1.ssm --vpc-endpoint-type Interface \
  --subnet-ids subnet-private-a subnet-private-c \
  --security-group-ids sg-xxxxxxxx --private-dns-enabled

A VPC created with create-vpc has DNS hostnames (enableDnsHostnames) disabled by default. If you try to create an endpoint with --private-dns-enabled in this state, it will fail with InvalidParameter. Please enable the DNS attributes first using modify-vpc-attribute.

aws ec2 modify-vpc-attribute --vpc-id vpc-xxxxxxxx --enable-dns-hostnames
aws ec2 modify-vpc-attribute --vpc-id vpc-xxxxxxxx --enable-dns-support

In addition to S3 and CloudWatch Logs permissions, add read permissions for the target parameters to the execution role.

{
    "Sid": "SsmGetParameters",
    "Effect": "Allow",
    "Action": ["ssm:GetParameters"],
    "Resource": "arn:aws:ssm:ap-northeast-1:123456789012:parameter/myapp/*"
}

I created three test parameters of type String.

Retrieving multiple parameters in a single task

The retrieval code is as follows. It uses get_parameters to retrieve three parameters at once and returns them as a dictionary with the parameter name prefix removed.

# ssmcfg.py
import boto3

PREFIX = "/myapp/"
NAMES = ["data_bucket", "output_prefix", "table_name"]


def get_config():
    ssm = boto3.client("ssm")
    resp = ssm.get_parameters(Names=[PREFIX + n for n in NAMES])
    result = {p["Name"].removeprefix(PREFIX): p["Value"] for p in resp["Parameters"]}
    print(f"result = {result}")
    return result

The workflow definition has three tasks. list_files (S3ListOperator) receives the bucket name and prefix, and bash_show receives the table name, each using bracket access.

pybash_ssmcfg:
  dag_id: pybash_ssmcfg
  schedule: null
  default_args:
    owner: airflow
    start_date: "2024-01-01"
  tasks:
    get_config:
      operator: airflow.providers.standard.operators.python.PythonOperator
      task_id: get_config
      python_callable: ssmcfg.get_config
    list_files:
      operator: airflow.providers.amazon.aws.operators.s3.S3ListOperator
      task_id: list_files
      bucket: "{{ ti.xcom_pull(task_ids='get_config')['data_bucket'] }}"
      prefix: "{{ ti.xcom_pull(task_ids='get_config')['output_prefix'] }}"
      dependencies:
        - get_config
    bash_show:
      operator: airflow.providers.standard.operators.bash.BashOperator
      task_id: bash_show
      bash_command: "echo \"table={{ ti.xcom_pull(task_ids='get_config')['table_name'] }} first_file={{ ti.xcom_pull(task_ids='list_files')[0] }}\""
      dependencies:
        - list_files

Pass the VPC from earlier using --network-configuration at creation time.

aws mwaa-serverless create-workflow \
  --name ssm-config \
  --definition-s3-location '{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"definitions/ssm-config.yaml"}' \
  --code '{"S3Location":{"Bucket":"amzn-s3-demo-bucket","ObjectKey":"code/ssm-config.zip"}}' \
  --role-arn arn:aws:iam::123456789012:role/mwaa-serverless-exec \
  --network-configuration '{"SecurityGroupIds":["sg-xxxxxxxx"],"SubnetIds":["subnet-private-a","subnet-private-c"]}'

When executed, all three tasks completed with SUCCESS. The task log of get_config shows the result of the bulk retrieval.

result = {'data_bucket': 'amzn-s3-demo-bucket', 'output_prefix': 'definitions/', 'table_name': 'my-config-table'}

The XCom of list_files contained the result of listing S3 using the retrieved bucket name and prefix. The output of bash_show was also as expected.

table=my-config-table first_file=definitions/ssm-config.yaml

There is no longer any need to split Lambda tasks per parameter, use fixed-width padding, or route through Step Functions. Even if the number of parameters increases, the number of tasks remains one.

Time-based billing for VPC endpoints

The biggest issue with the workarounds was that MWAA Serverless billing has a minimum of 1 minute per task instance. Splitting tasks per parameter causes billing time to increase proportionally with the number of parameters. By consolidating into a single task, billing time no longer increases as parameters are added.

Instead, time-based billing for VPC endpoints applies. Interface endpoints cost 0.014 USD/hour/AZ in Tokyo. Placing SSM and CloudWatch Logs endpoints in 2 AZs comes to approximately 41 USD/month. If you create a dedicated VPC for the workflow, this cost is added, but if the existing VPC already has these endpoints, there is almost no additional cost.

The option of switching the store to DynamoDB

The remaining cost in the SSM configuration is the time-based billing for interface endpoints. Switching the configuration store to DynamoDB can reduce the SSM portion. Like S3, DynamoDB supports gateway endpoints, and gateway endpoints have no charge.

Even if you switch to DynamoDB, a VPC itself is still required. As confirmed in the reachability article mentioned earlier, tasks without a VPC cannot reach DynamoDB either. What changes is the type of endpoint and the billing.

Store configuration values as multiple attributes in a single item.

In the retrieval code, read a single item with get_item and return the attributes other than the partition key as a dictionary.

# ddbcfg.py
import boto3


def get_config():
    try:
        table = boto3.resource("dynamodb").Table("my-config-table")
        item = table.get_item(Key={"config_id": "app"})["Item"]
        result = {k: v for k, v in item.items() if k != "config_id"}
    except Exception as e:
        result = {"error": f"{e.__class__.__name__}: {e}"}
    print(f"result = {result}")
    return result

There are two key points.

• Use boto3.resource. With the low-level client, attribute values come in the typed format {"S": "..."}, making the template extraction two levels deep as ['data_bucket']['S']. With resource, you get a plain string dictionary.
• Catch exceptions, wrap them in a dictionary, and return. The reason is described below.

Add dynamodb:GetItem scoped to the target table to the execution role.

I ran this in a configuration with only two VPC endpoints — an S3 gateway and a DynamoDB gateway — meaning no interface endpoints at all. All three tasks completed with SUCCESS, and values could be extracted with bracket access just like in the SSM version.

However, because there is no CloudWatch Logs endpoint, task logs are not retained. The log group itself is created (the documentation states that if no log group name is specified, /aws/mwaa-serverless/<workflow-name>/ is used), but the log streams remained empty. Since execution succeeds, it is easy to overlook the absence of logs. That is why exceptions are wrapped in a dictionary and returned via XCom. Even in a configuration where logs are not available, you can check the details of a failure from the XCom of get-task-instance.

I compare the monthly cost of four patterns, including the per-parameter Lambda task splits that were used as workarounds, excluding common downstream tasks. The assumptions are 3 parameters, task billing at 0.08 USD/hour (minimum 1 minute billing at 0.0013 USD/task), and interface endpoints at 0.014 USD/hour/AZ in 2 AZs for 730 hours. Lambda costs, SSM standard parameters, and DynamoDB reads are negligible and not included.

The per-run task billing is 3x for the Lambda approach, but in monthly terms the fixed cost of interface endpoints is larger. The SSM + Logs configuration becomes cheaper than the Lambda approach only when exceeding 15,000 runs per month (once every 3 minutes) for 3 parameters, or around 3,400 runs per month (once every 13 minutes) for 10 parameters. If you cannot reuse existing VPC endpoints, choosing purely on cost points to either the Lambda approach or the DynamoDB-only configuration.

Note that SSM uses "N parameters" while DynamoDB uses "N attributes in a single item" — the granularity of how configuration values are stored changes. Features of Parameter Store such as per-parameter access control and revision history are no longer available, so please do not decide to switch based solely on a simple cost comparison.

Summary

• A dictionary returned by python_callable in PythonOperator is stored in XCom as a dictionary type and can be extracted using bracket access in downstream templates
• This makes it possible to retrieve multiple SSM parameters at once in a single task. While per-parameter task billing is eliminated, the fixed cost of interface endpoints is larger, and for 3 parameters running once per hour, the Lambda approach is cheaper
• A custom VPC is required to reach SSM. In addition to the SSM interface endpoint, do not forget the CloudWatch Logs endpoint for task logs
• Switching the store to DynamoDB allows a configuration using only free gateway endpoints. However, omitting the Logs endpoint means task logs are not retained, so design the system to allow failure diagnosis via XCom

I hope this is useful to someone.

Reference Links

• Amazon MWAA Serverless - Using Python and Bash operators
• Amazon MWAA Serverless - Networking