Add MwaaServerlessStartWorkflowRunOperator

[john-jac]

What

Add an operator for Amazon MWAA Serverless to start workflow runs.

Motivation

TriggerDagRunOperator triggers a DAG run within the same Airflow environment. MwaaServerlessStartWorkflowRunOperator serves the analogous purpose for MWAA Serverless — it triggers a workflow execution on a remote MWAA Serverless environment from any Airflow DAG. This enables patterns like:

• Cross-environment orchestration: A central Airflow instance triggers serverless workflows that run on MWAA Serverless with automatic scaling and task isolation
• Hybrid pipelines: Combine provisioned MWAA for long-running orchestration with MWAA Serverless for burst workloads
• CI/CD for workflows: A deployment DAG validates and triggers workflow runs on MWAA Serverless as part of a release pipeline

MWAA Serverless is a separate service from MWAA (similar to how EMR Serverless is separate from EMR), with its own API (mwaa-serverless) and resource model (workflows instead of environments).

How

• New MwaaServerlessStartWorkflowRunOperator using AwsBaseOperator[AwsBaseHook] with client_type="mwaa-serverless"
• Supports override_parameters for runtime customization and workflow_version for version pinning
• Returns the workflow run ID
• Follows EMR/EMR Serverless pattern: separate integration entry, separate operator file, shared logo

Files

• operators/mwaa_serverless.py — operator
• test_mwaa_serverless.py — unit tests (3 tests)
• example_mwaa_serverless.py — system test
• mwaa_serverless.rst — docs
• provider.yaml + get_provider_info.py — new MWAA Serverless integration + operator entries

Testing

• Unit tests: 3 passed
• mypy: no issues
• Static checks: 43 passed, 0 PR-related failures

[john-jac]

Closing to resubmit with clean branch.

[john-jac]

Thanks @vincbeck for the review! Addressed both points:

1. IAM role as external parameter — now uses SystemTestContextBuilder().add_variable("ROLE_ARN") instead of creating roles in the test (following the example_batch.py pattern).

2. Split into individual tasks — replaced the monolithic setup/teardown with separate @task functions: create_bucket (using S3 operators), upload_workflow_yaml, create_workflow, stop_workflow_run, delete_workflow, delete_bucket (also using S3 operators). Each can be replaced by an operator later.

[john-jac]

Good call @vincbeck — moved all @task functions outside the DAG context to module level, matching the pattern in example_batch.py and other system tests.