Authoring Mwaa Workflow

AWSOfficialPublished by AWS

Skill
Workflow
Cloud and DevOps
Apache-2.0
Low risk

From the author

Authors and deploys MWAA workflow artifacts: Python Airflow DAGs for provisioned environments or YAML workflow files for Serverless. Covers operator selection, timeout design, retry strategy, scheduling, failure notifications, idempotency, and MWAA Serverless schema compliance. Deploys the artifact (S3 DAG upload or Serverless CreateWorkflow/UpdateWorkflow), creates an environment inline when approved, and redeploys fixes, then optionally hands off to testing-mwaa-workflow. Triggers on: create a DAG, write a pipeline, build a workflow, orchestrate tasks, Airflow DAG, data pipeline, schedule a job, deploy a DAG, deploy a workflow, YAML workflow. Not applicable to converting or migrating existing DAGs between provisioned and serverless (conversion is out of scope), running or smoke-testing a deployed workflow (handled by testing-mwaa-workflow) or diagnosing a failed run (handled by debugging-mwaa-workflow).

— the description by AWS

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

This skill provides a structured path for creating and deploying Amazon Managed Workflows for Apache Airflow (MWAA). It handles the distinct requirements for provisioned environments (Python DAGs) and MWAA Serverless (YAML workflows). The skill covers the entire lifecycle from operator selection, retry strategies, and scheduling to environment discovery and deployment. It includes built-in safety checks for state-mutating operations and integrates with testing workflows to ensure reliable pipeline delivery.

What it does

Routes requests between provisioned Python DAGs and Serverless YAML workflows.
Validates operator selection and configuration against MWAA-specific constraints.
Automates S3 uploads for DAGs and Create/UpdateWorkflow calls for Serverless.
Handles inline environment creation and redeployment of fixes.
Enforces security best practices for IAM, secrets management, and resource defaults.
Provides a clear hand-off mechanism to testing-mwaa-workflow for verification.

When to use it

When you need to create a new data pipeline or workflow for Amazon MWAA.
When you are unsure whether your project requires a provisioned environment or Serverless.
When you need to deploy a validated DAG or YAML workflow to an AWS environment.
When you need to update an existing workflow or redeploy a fix after testing.
When you require guidance on operator compatibility and retry strategies for MWAA.

Advantages

Reduces configuration errors by enforcing schema compliance and operator allowlists.
Ensures production safety through explicit confirmation of state-mutating operations.
Simplifies complex deployment steps into a guided, repeatable process.
Maintains consistency across different MWAA deployment models.
Integrates seamlessly with testing and debugging workflows for a complete lifecycle.

Usage examples

“Create a new Python DAG for my provisioned MWAA environment that runs a daily data sync.”: The agent will guide you through gathering context, selecting operators, and deploying the DAG to your S3 bucket.
“I need to deploy a new YAML workflow to MWAA Serverless.”: The agent will validate your YAML schema, check operator support, and execute the CreateWorkflow command.
“Deploy my DAG and run a test immediately.”: The agent will deploy the artifact and automatically invoke the testing-mwaa-workflow skill to verify the run.

Install

Download the skill as a ZIP:
https://autogineer.com/agent-skills/aws/authoring-mwaa-workflow/download

Unzip it into the cross-client skills folder of your project:

mkdir -p .agents/skills && unzip authoring-mwaa-workflow.zip -d .agents/skills

Documentation by AWS

Authoring MWAA Workflows

AWS MCP server (optional but recommended): running the AWS CLI commands in this skill through the AWS MCP server gives sandboxed execution and audit logging. Every command here also works with the plain AWS CLI, so the skill does not require the MCP server or any MCP-only tools.

Author production-grade workflow artifacts for Amazon MWAA. Routes to one of two paths: Python DAG (provisioned) or YAML workflow (Serverless).

Execution note — poll in discrete steps: whenever you wait for an AWS operation to reach a terminal or ready state, issue one status check per call and decide in your own loop whether to check again. Never block a single command or script on the wait (no while+sleep until done), regardless of the operation or how long it takes.

Guardrail — where this skill's own files live (MCP vs local install)

This skill can be loaded two ways, and they resolve the skill's own bundled files from different places. Determine how the skill was loaded before reading a reference:

Loaded through the AWS MCP retrieve_skill tool: The skill is not installed on the local filesystem. You MUST fetch each reference via retrieve_skill with the file parameter (e.g. file="references/authoring-provisioned-dag.md") and read the returned content. Do NOT file_read these paths locally — they do not exist on disk.
Installed locally (e.g. .kiro/skills/authoring-mwaa-workflow/ or ~/.claude/skills/authoring-mwaa-workflow/): Read the files from the local skill directory using relative paths.

This distinction applies only to the skill's own packaged files. User data and session artifacts are always read from and written to the user's working directory. Never fetch or write customer data through retrieve_skill.

Step 0: Route to Path

Evaluate in this order:

Resolvable target provided? A target reference is a definitive routing signal regardless of other keywords:
- A provisioned environment (ARN like arn:aws:airflow:<region>:<account>:environment/<name>, or a name resolvable via aws mwaa get-environment) → go to Path A.
- A Serverless workflow ARN (arn:aws:airflow-serverless:<region>:<account>:workflow/<name>) → go to Path B.
Both-path keywords present? If the request contains keywords from both paths and no resolvable target, disambiguate by intent:
- PythonOperator is supported on Serverless, so an operator-level python cue (PythonOperator, python_callable, "Python function/task") alongside a Path B signal (yaml, serverless, workflow ARN) is NOT ambiguous → go to Path B.
- Conversion context ("convert my Python DAG to serverless") → this skill does not apply; conversion is out of scope.
- A genuine provisioned cue (Python DAG, provisioned) alongside a Serverless cue with no target → ask the clarifying question.
Exactly one path keyword? Treat only deployment-target terms as Path A signals: provisioned, Python DAG, or "a .py for my environment" → go to Path A. yaml or serverless → go to Path B. Operator-level Python mentions are not Path A signals.
No routing signal? "DAG" or "Workflow" alone is ambiguous — it does NOT indicate a path. Ask: is the target MWAA provisioned (Python DAG) or MWAA Serverless (YAML)?

Paths

Follow the reference for the path you routed to (you do not need the other path's reference):

Path A — Python DAG (MWAA Provisioned): references/authoring-provisioned-dag.md
Path B — YAML Workflow (MWAA Serverless): references/authoring-serverless-workflow.md

After the routed path's Write step, continue with Deploy & Test below.

Deploy & Test (optional, after Write)

Authoring owns all deployment and redeployment. Detail in references/deploying-mwaa.md.

Steps

Ask — present options based on whether the artifact has a schedule. Frame the question using path-appropriate language:
- Provisioned: "deploy this DAG to an environment" (DAGs are uploaded to an environment's S3 bucket).
- Serverless: "deploy this workflow" (workflows are standalone resources — never say "deploy to an environment").
If the DAG/workflow has a schedule:
- Deploy and test — deploy, unpause, trigger a run now
- Deploy and unpause — deploy, unpause, let it run on schedule (no immediate trigger)
- Deploy only — upload to S3, leave paused
If the DAG/workflow has no schedule (manual-trigger only):
- Deploy and test — deploy, trigger a run now
- Deploy only — upload to S3, leave paused (no "unpause" option — nothing to schedule)
The user may also decline all options.
Deploy:
- Provisioned: upload the DAG to the environment's SourceBucketArn/ DagS3Path. Run post-deploy verification (see deploying-mwaa.md) to confirm the scheduler parsed the new file without import errors or dag_id conflicts. If no environment exists and the user approves, create one inline (plan-validate-execute + explicit confirmation), then poll CREATING -> AVAILABLE (~20-40 min). The user may instead supply an existing environment.
- Serverless: CreateWorkflow (new) or UpdateWorkflow (redeploy); the YAML is validated synchronously here. If the workflow uses PythonOperator/BashOperator, first build and upload the code package to S3 and pass it via --code (see references/serverless-code-packaging.md and references/deploying-mwaa.md). The user may instead supply an existing ARN.
- Redeploy (fix loop): the same upload / UpdateWorkflow path, reused when testing-mwaa-workflow delegates an ARTIFACT or ENVIRONMENT fix.
If "Deploy and unpause" selected — deploy per step 2, then unpause. Do not trigger a run or invoke testing-mwaa-workflow.
If "Deploy and test" selected — deploy per step 2, unpause if applicable, then invoke testing-mwaa-workflow with the resolved target (env name + dag_id, or workflow ARN). That hand-off is testing's delegated invocation mode.

HARD GATE: If testing is requested — whether upfront ("deploy and test") or later in the conversation ("test it", "run it", "try it") — you MUST invoke testing-mwaa-workflow. Do NOT trigger, monitor, or verify DAG runs manually. "Deploy and unpause" is NOT a test request — it is a deploy-only action.

Production safety: create-environment, update-environment, create-workflow, and update-workflow mutate state — confirm each with its impact stated. Warn on prod-named targets.

Troubleshooting

Error	Cause	Fix
dagrun_timeout kills DAG early	< timeout set in service called	Raise dagrun_timeout or lower service timeout
YAML validation rejects workflow	Wrong type or param	Use timedelta format; check allowlist
Operator not found in Serverless	Not allowlisted	Use a supported operator, PythonOperator/BashOperator, or Lambda
Serverless run: cannot extract code / corrupt env	Bad code package	Files at zip root, no `__pycache__`, ≤250 MB; repackage
Serverless Python task ImportError	Missing dep or wrong-platform wheel	Bundle as `manylinux2014_x86_64` / Py3.12 wheel; don't bundle pre-installed packages
Template variable undefined	Version mismatch	Check vars for exact Airflow version

References

references/dag-patterns.md — Python DAG templates
references/yaml-schema.md — MWAA Serverless format
references/deploying-mwaa.md — deploy, inline env creation, and redeploy
references/serverless-code-packaging.md — Python/Bash code packaging, pre-installed packages, limits
references/authoring-provisioned-dag.md — Path A: provisioned Python-DAG authoring steps (A1–A7)
references/authoring-serverless-workflow.md — Path B: Serverless YAML authoring steps (B1–B5)

Security Considerations

State-mutating operations (create/update-environment, create/update-workflow, S3 DAG upload) require explicit confirmation with impact stated; warn on prod-named targets (see the Deploy HARD-GATE).
Least-privilege IAM: the A5 check adds only the exact Action/Resource pairs the artifact needs — never *FullAccess or service:*.
No hardcoded secrets/endpoints: use Airflow Variables/Connections backed by Secrets Manager or SSM Parameter Store; never emit credentials in DAG code or CLI examples.
Serverless code packages ship only the user's own modules plus pinned, platform-matched wheels — no unreviewed third-party binaries.
Data protection: keep sensitive data out of SNS/CloudWatch notification payloads and logs; rely on their encryption.
Secure defaults for inline-created resources: encrypt and lock down any S3/MWAA/SNS/CloudWatch resource this skill creates — see deploying-mwaa.md "Secure defaults".
AWS security best practices: verify the security posture against the MWAA User Guide's Security best practices page (and the MWAA Serverless equivalent) at runtime — AWS updates them over time; see deploying-mwaa.md "Secure defaults".

Files

ATTRIBUTION.mdAdded by AutoGineer926 B
LICENSELicence file added by AutoGineer9.9 KiB
NOTICELicence file added by AutoGineer67 B
SKILL.md11.0 KiB
references/authoring-provisioned-dag.md13.9 KiB
references/authoring-serverless-workflow.md7.4 KiB
references/dag-patterns.md7.9 KiB
references/deploying-mwaa.md10.0 KiB
references/serverless-code-packaging.md3.8 KiB
references/yaml-schema.md4.7 KiB

references/dag-patterns.md

DAG Patterns Reference

Package Availability

MWAA pre-installs only Airflow core, the Amazon provider package, and their transitive dependencies. Do not assume any other Python package (pandas, pyarrow, numpy, scikit-learn, etc.) is available at task runtime. If DAG code needs a package not in the base image:

Add it to requirements.txt (installed at environment startup). Pin every package to an exact version and install against the Apache Airflow constraints file that matches the environment's Airflow and Python version (--constraint "https://raw.githubusercontent.com/apache/airflow/constraints-<AIRFLOW_VERSION>/constraints-<PY_VERSION>.txt", with the exact URL fetched at runtime). Unpinned or unconstrained requirements are the most common cause of a failed environment update — a transitive dependency resolves to a version incompatible with the pinned Airflow.
Or delegate the work to a managed service (Glue for Spark/pandas, SageMaker for ML, Lambda for lightweight transforms) that has its own runtime.

A ModuleNotFoundError at task execution (not at parse time) means the import is inside a task callable and the package is not installed. The fix is requirements.txt or delegation — not inlining the install in the DAG.

Timeout Chain

sensor.timeout <= dagrun_timeout
service_timeout <= dagrun_timeout
execution_timeout = safety net per non-deferred window (15min typical)
dagrun_timeout >= sum(service timeouts) + retry buffer

For deferrable operators, execution_timeout resets on each deferred-to-active cycle. It need not exceed the service timeout.

Retry Categories

Category	retries	retry_delay	Notes
Sensor	0	--	Has own timeout
API call	3	5min	Short, transient failures
Compute job (expensive)	1	5min	Costly reruns; minimize
Compute job (cheap idempotent)	2	5min	Safe to retry
DQ gate	2	5min	AirflowFailException non-retryable
Teardown	2	5min	trigger_rule=all_done

Always: retry_exponential_backoff=True, max_retry_delay=timedelta(minutes=30).

Failure Callback Template

def notify_failure(context):
    import json
    import boto3
    from datetime import datetime, timezone

    task_instance = context["task_instance"]
    dag_id = context["dag"].dag_id
    task_id = task_instance.task_id
    start_date = task_instance.start_date
    end_date = task_instance.end_date or datetime.now(timezone.utc)
    exception = context.get("exception", "")

    message = {
        "dag_id": dag_id,
        "task_id": task_id,
        "start_date": str(start_date),
        "end_date": str(end_date),
        # Publish only the exception TYPE (carries no secret) + the log_url;
        # never the message/traceback/args, which can carry secrets or PII
        # (CWE-532). Authorized viewers get full detail via log_url.
        "error_type": type(exception).__name__ if exception else "",
        "log_url": task_instance.log_url,
    }

    sns = boto3.client("sns", region_name=REGION)
    sns.publish(
        TopicArn=SNS_TOPIC_ARN,
        Subject=f"Airflow task failed: {dag_id}.{task_id}",
        Message=json.dumps(message, indent=2),
    )

Security: enable server-side encryption (SSE) on the SNS topic, restrict Publish to the execution role, and confirm SNS_TOPIC_ARN is an expected in-account recipient — the callback sends operational metadata off the environment.

Waiting for AWS Resource State Changes

Exhaust Airflow-native constructs before writing custom wait code. An operator or sensor that owns the wait is almost always preferable to a hand-written boto3 poll — it frees the worker slot and encodes the polling and error handling for you. Work down this ladder and stop at the first rung that applies:

Deferrable operator. If a deferrable operator exists for the action, use it with deferrable=True (e.g., EmrServerlessStartJobOperator(deferrable=True)). It hands the wait to the triggerer, freeing the worker slot for the whole wait and resuming on the completion event.
Submit + deferrable sensor. If no single operator both submits and waits, submit with a non-blocking operator (e.g., wait_for_completion=False) and wait with a deferrable sensor (deferrable=True). Deferrable support is per-sensor and per-provider-version — verify it for the specific sensor. Some AWS sensors are NOT deferrable (e.g. AthenaSensor, EmrServerlessJobSensor); for those, use the deferrable operator form instead (AthenaOperator(deferrable=True), EmrServerlessStartJobOperator(deferrable=True)). If no deferrable path exists, a sensor in mode="reschedule" still frees the worker slot between pokes (no triggerer needed, but it only resumes on a fixed interval).
Bounded in-task poll loop. Only when no operator or sensor fits and you must wait inside a @task with boto3. Do not assume client.get_waiter("<state>") exists — not every service/operation ships a waiter, and a missing one raises ValueError: Waiter does not exist at runtime. Verify waiter availability in the boto3 docs for your runtime's botocore version before using get_waiter; otherwise use a bounded loop:

for _ in range(max_attempts):
    response = client.get_<resource>(<id>=resource_id)
    state = response["<resource>"]["state"]
    if state == target_state:
        break
    if state in terminal_failure_states:
        raise RuntimeError(f"Resource reached {state}")
    time.sleep(delay_seconds)
else:
    raise RuntimeError(f"Resource did not reach {target_state}")

Deferrable operators and sensors require a running triggerer; MWAA runs a managed triggerer on all MWAA-supported Airflow versions, so the deferrable rungs work out of the box on both Airflow 2 and 3. The mode="reschedule" fallback needs no triggerer. See A6 rule 5 in authoring-provisioned-dag.md for the submit + @task sensor split.

start_date

Use a fixed, timezone-aware datetime from stdlib. Never use days_ago() (deprecated in 2.8, removed in 3.x) or naive datetimes (rejected in 3.x). Always pair a fixed past start_date with catchup=False: on Airflow 2.x catchup defaults to True, so a scheduled DAG with a past start_date backfills every missed interval in one burst the moment it is unpaused. AF3 defaults catchup=False, but set it explicitly so the DAG behaves the same on both.

from datetime import datetime, timezone

with DAG(
    dag_id="example",
    start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
    schedule="<mm> <h> * * *",  # off-peak cron DERIVED from dag_id (A6 rule 15): mm 1-59, h 1-6 UTC — recompute per DAG, never a fixed literal or @preset
    catchup=False,  # REQUIRED with a fixed past start_date — avoids a backfill storm
):
    ...

Do not use pendulum.datetime(...) — it adds a top-level dependency on pendulum's API which changed between v2 and v3. Stdlib timezone.utc works on all MWAA-supported Python versions (3.10+) and all Airflow versions (2.x and 3.x).

Idempotency

Job names with "{{ ds_nodash }}" suffixes for date uniqueness.

Cross-DAG Triggering

Event-driven coupling. The object type and its in-task event key differ by Airflow major, so gate the example by the target Airflow version (per A3 "verify APIs exist in the Airflow version"):

# Airflow 2.4-2.x (Datasets):
from airflow.datasets import Dataset
DATASET = Dataset("s3://bucket/prefix/")
# Producer: outlets=[DATASET] on final task
# Consumer: schedule=[DATASET]
# In-task events: context["triggering_dataset_events"]

# Airflow 3.x (Assets replace Datasets):
from airflow.sdk import Asset
ASSET = Asset("s3://bucket/prefix/")
# Producer: outlets=[ASSET] on final task
# Consumer: schedule=[ASSET]
# In-task events: context["triggering_asset_events"]

Cross-environment: MwaaTriggerDagRunOperator if support available in Airflow version in use.

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs an MCP server, Needs Python

Security

What our scanner noticed· Low risk

Instructions that use the network

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Published by

AWS

Publisher

AWS (Official)

Source

aws/agent-toolkit-for-aws · plugins/aws-data-analytics/skills/authoring-mwaa-workflow · commit 0a218f8

Licence

Apache License 2.0 (redistribution allowed)

LICENSE (added by AutoGineer from LICENSE)
NOTICE (added by AutoGineer from NOTICE)
ATTRIBUTION.md (written by AutoGineer)

Synced

2026-11-09

Also distributed in

aws/agent-toolkit-for-aws

AutoGineer is not affiliated with or endorsed by AWS or AWS.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download authoring-mwaa-workflow.zip