npx skills add ...
npx skills add astronomer/agents --skill blueprint
Define reusable Airflow task group templates with Pydantic validation and compose DAGs from YAML. Use when creating blueprint templates, composing DAGs from YAML, declaring shared variables or per-environment profiles, validating configurations, sharing templates as an installable package, or enabling no-code DAG authoring for non-engineers.
npx skills add astronomer/agents --skill blueprint
You are helping a user work with Blueprint, a system for composing Airflow DAGs from YAML using reusable Python templates. Execute steps in order and prefer the simplest configuration that meets the user's needs.
Package:
airflow-blueprinton PyPI — this skill documents 0.5.0 Repo: https://github.com/astronomer/blueprint Requires: Python 3.10+, Airflow 2.5+ Cross-references: theairflowskill for Astro CLI, registry, and REST API discovery commands;authoring-dagsordag-factorywhen the user needs full Airflow flexibility instead of validated templates.
| User Request | Action |
|---|---|
| "Create a blueprint" / "Define a template" | Go to Creating Blueprints |
| "Build a template from other templates" | Go to Composing Templates |
| "Create a DAG from YAML" / "Compose steps" | Go to Composing DAGs in YAML |
| "Reuse a value across steps or DAGs" / "Different value per environment" | Go to Variables and Profiles |
| "Use a blueprint in an existing Python DAG" / "Generate DAGs in a loop" | Go to Blueprints in Python DAGs |
| "Customize DAG args" / "Add tags to DAG" / "Different DAG defaults per folder" | Go to Customizing DAG-Level Configuration |
| "Share templates across repos" / "Install blueprints from a package" | Go to Sharing Blueprints as a Package |
| "Override config at runtime" / "Trigger with params" | Go to Runtime Parameter Overrides |
| "Post-process DAGs" / "Add callback" / "Don't let one bad file break everything" | Go to Loader Options |
| "Validate my YAML" / "Lint blueprint" | Go to Validation Commands |
| "Set up blueprint in my project" | Go to Project Setup |
| "Version my blueprint" | Go to Versioning |
| "Generate schema" / "Astro IDE setup" | Go to Schema Generation |
| Blueprint errors / troubleshooting | Go to Troubleshooting |
If the user is starting fresh, guide them through setup:
Add airflow-blueprint>=0.5.0 to requirements.txt.
Create dags/loader.py:
The function name matters. Airflow's safe-mode DAG file processor only parses files containing both
airflowanddag, so the import line itself is what makes the loader discoverable.build_allandbuild_all_dagsstill work as deprecated aliases that emitDeprecationWarning; migrate existing loaders tobuild_all_airflow_dags.
DAG-level configuration (schedule, description, tags, default_args, etc.) is handled via YAML fields and BlueprintDagArgs templates — see Customizing DAG-Level Configuration.
Run blueprint list from the project root. If no blueprints are found, the user needs to create blueprint classes first.
Config model, generic base class, and a render() returning a task or group keyed on self.step_id. Adapt this rather than inventing a different structure:
| Element | Requirement |
|---|---|
| Config class | Must inherit from BaseModel |
| Blueprint class | Must inherit from Blueprint[ConfigClass] |
render() method | Must return TaskGroup or BaseOperator |
| Task IDs | Use self.step_id for the group/task ID |
| Field types | Must be single-typed and YAML-compatible (see below) |
Config fields must be single-typed. Multi-type unions like str | int or Union[A, B] are rejected at class-definition time (raises TypeError) because they produce ambiguous YAML parsing and anyOf schemas. The check recurses through nested models, list items, and dict values.
str, int, float, bool), Literal[...], list[X], dict[str, V], nested BaseModel, and Optional[X] / X | None (the nullable pattern).str | int, Union[A, B], or any union with more than one non-None arm. Bare Any and dict[str, Any] are rejected for the same reason — use an explicit single type for the value.Use Field(default=..., init=False) for fields used inside render() that should not be overridable from YAML. They are excluded from the constructor and omitted from JSON Schema output:
A step config model inherits Pydantic's default extra="ignore", so a misspelled field in a step's YAML is silently dropped rather than reported. Suggest model_config = ConfigDict(extra="forbid") to turn those typos into errors:
DAG args config models are the opposite and need no such setting — Blueprint makes them strict for you (see Customizing DAG-Level Configuration).
A blueprint can instantiate and render other blueprints inside its render() method, letting you build higher-level templates from lower-level building blocks while exposing a single, flat config to YAML authors.
Inside render(), instantiate each child blueprint, set its step_id, call render(...) with a config you construct, and wire the results together inside a parent TaskGroup:
YAML authors then see a single step with a flat config, and the composed children stay invisible to them.
By default, only schedule and description are supported as DAG-level fields (via the built-in DefaultDagArgs). For other fields like tags, default_args, catchup, etc., see Customizing DAG-Level Configuration.
| Key | Purpose |
|---|---|
blueprint | Template name (required) |
depends_on | List of upstream step names |
version | Pin to specific blueprint version |
trigger_rule | Airflow trigger rule for the step; validated against the installed Airflow version |
Everything else passes to the blueprint's config.
Use trigger_rule to control when a step runs relative to its upstream dependencies — for example, to run a notification step even if an upstream step failed:
Values are validated dynamically against the installed Airflow's TriggerRule enum, so the accepted set follows your Airflow version rather than this skill. When the step's blueprint renders a TaskGroup, the rule applies only to the group's root tasks (those with no internal upstream), preserving the blueprint author's internal wiring.
YAML supports Jinja2 templating with access to environment variables, Airflow variables/connections, and runtime context:
Available template variables:
env — environment variablesvar — Airflow Variablesconn — Airflow Connectionscontext — proxy that generates Airflow template expressions for runtime macros (e.g. context.ds_nodash, context.dag_run.conf, context.task_instance.xcom_pull(...))profile — the active variable profile name, or nothing when none is selected. Useful for deriving a value from the profile rather than enumerating it per profile: dag_id: "pipeline_{{ profile }}"For values that are fixed at parse time and shared across steps or DAGs, prefer Variables and Profiles over a Jinja {% set %} block — variables are scoped, shareable, and visible to blueprint lint.
DAG YAML can declare variables and reference them as ${name}. Use this to stop repeating a value across steps and DAGs.
Declare them in a blueprint.vars.yaml shared by every DAG beneath it, in a DAG's own vars: block, or both — nearer declarations override further ones:
Substitution runs after YAML parsing, so expiration_days stays an int rather than becoming the string "90". Values are scalars or lists, and variables may compose (base: ${db}.${schema}).
Variable names match ^[A-Za-z_][A-Za-z0-9_-]*$ — hyphens are allowed, and periods are reserved so dotted namespaces can be added later without ambiguity.
${...}is always a variable reference. Anything else that uses that syntax — most often a shell variable in abash_command— must be escaped as$${...}, or Blueprint tries to resolve it as a variable. Only$$immediately before{is treated as an escape, so a bare$$(a shell PID, an awk field) needs no change.blueprint lintreports each unescaped occurrence and names the escape in the error, so lint the project after adopting variables.
A variable can carry a different value per named profile, selected at build time. Environments are the obvious use, but the mechanism is just named selection:
Every profile a DAG declares must give the variable a value; a partial mapping is an error rather than a silent fallback.
blueprint vars <path> shows the resolved value of each variable and where it came from, and flags variables a DAG never references. blueprint lint validates every declared profile unless --profile narrows it to one. Pass --root to match the path the loader builds from, or resolution differs between lint and runtime.
Blueprints aren't tied to the YAML composition flow. Two patterns let you use them from Python — useful for incremental adoption or data-driven DAG generation.
Instantiate the Blueprint class, set its step_id, call render(), and wire it in with >>:
The step_id you set determines the task_id / group_id the blueprint renders under.
Builder / DAGConfigFor data-driven DAG generation (one DAG per region, tenant, etc.), build DAGs in a loop and register each in globals() so Airflow discovers them:
DAGConfig accepts the same fields you would write in YAML. Pass source_path=__file__ so the DAG args template is resolved from this file's directory the same way a YAML file's would be — without it, resolution falls back to the project-wide default (see Customizing DAG-Level Configuration).
By default, Blueprint supports schedule and description as DAG-level YAML fields. To use other DAG constructor arguments (tags, default_args, catchup, etc.), define a BlueprintDagArgs subclass. Its render() returns a dict of kwargs passed to the Airflow DAG() constructor, so the accepted keys are whatever your Airflow version's DAG accepts.
The declared fields then become valid DAG-level YAML keys, validated by the config model.
A project may define more than one template. Each DAG uses the template defined closest above it: resolution starts in the DAG file's own directory and walks up parent directories, so a subdirectory overrides its parents.
A DAG with no template above it falls back to the one declared default=True, then to the sole registered template, then to the built-in DefaultDagArgs. A template is scoped to the directory holding the .py file that defines it, so moving that file rescopes it — the most common surprise in this feature.
Nothing in the DAG YAML changes: a DAG never names its template. Run blueprint list to see which template applies to which path, which one is the fallback, and where each is defined; blueprint lint names the resolved template per DAG.
A template registers under the snake_case form of its class name — ProjectDagArgs becomes project_dag_args — which is the name blueprint schema --dag-args <name> expects. Setting name = "..." overrides it, and must itself be snake_case.
A DAG args config model defines the DAG YAML's top-level surface, so Blueprint applies extra="forbid" to it automatically — an undeclared top-level key is an error rather than a silently ignored one. This is the opposite default from step configs, which ignore unknown keys unless you opt in.
This shows up in the generated schema as additionalProperties: false, so editors and the Astro IDE reject unknown top-level keys too.
Setting extra yourself on the model leaves your choice intact. To defer to the model's own policy instead, pass allow_extra=True on the class:
name = "..." on the class renames one) or more than one declaring default=True.DefaultDagArgs is used (schedule and description only).Blueprints can be shared across repositories as an installable package instead of copied files. The package advertises itself under the airflow_blueprint.blueprints entry-point group, and Blueprint discovers it once installed, with no per-repo configuration:
The value must be a plain dotted module or package path. The advertised module — and every submodule, if it is a package — is scanned exactly like a locally discovered file, so any Blueprint or BlueprintDagArgs subclass defined in it is registered.
Consumers install the package and the blueprints appear in blueprint list alongside local ones, with the source column distinguishing them. A package that fails to import raises EntryPointLoadError rather than silently vanishing from the registry.
To turn discovery off, pass discover_entry_points=False to the loader, or the corresponding --no-entry-points flag to the CLI (--help confirms which commands accept it).
Blueprint config fields can be overridden at DAG trigger time using Airflow params, letting users customize behavior when manually triggering DAGs.
supports_params = TrueA blueprint must set the class attribute supports_params = True for its config fields to register as Airflow params (namespaced as {step}__{field}). Without it, self.param() / self.resolve_config() do nothing and no fields appear in the trigger form. Only opt in for blueprints that actually use those methods — otherwise dead params clutter the form with no effect.
Use self.param() in operator template fields, where Airflow renders the value at execution time; use self.resolve_config() in Python callables, where you need a validated config object. Both can appear in one blueprint:
step_name__field)ValidationError at execution timeconf with the namespaced names to the DAG run endpoint (af api ls --filter dagRun finds the current path — see the airflow skill)Pydantic field schema flows through to Airflow's trigger form; json_schema_extra controls how each field renders (format values such as multiline and date pickers, examples, values_display, description_md). The Airflow version determines which are honoured, so check against the form rather than assuming.
Validation nuance: only Field constraints that map to JSON Schema (ge, le, pattern, min_length, max_length, Literal enums) are enforced in the trigger form. Custom @field_validator / @model_validator logic does not map to JSON Schema, so it runs only at build time and inside resolve_config(). If custom validators enforce important constraints, call self.resolve_config() in your @task function so they run on overridden values.
build_all_airflow_dags() takes the options that govern a whole project. The ones that change behaviour materially:
| Option | Effect |
|---|---|
profile= | Selects which profile's values the ${...} variables resolve to (see Variables and Profiles) |
skip_invalid_dags=True | Renders the valid YAML files and skips faulty ones instead of failing the import |
discover_entry_points=False | Turns off discovery of blueprints installed as packages |
on_dag_built= | Callback to post-process each DAG after construction |
discover_entry_points is ignored when bp_registry is supplied directly, since that registry has already run discovery.
The rest of the signature is plumbing that rarely needs changing: search_path and pattern control YAML discovery, register_globals overrides the caller's globals(), render_templates and template_context govern Jinja, and bp_registry supplies a pre-built registry.
.airflowignoreYAML discovery honours .airflowignore, using Airflow's own ignore-file walker — so the syntax, the core.dag_ignore_file_syntax setting, and nested ignore files behave exactly as they do for the DAG processor. blueprint lint honours it too when scanning a directory, so a draft excluded from Airflow is also excluded from lint; passing that file explicitly still lints it, which is how you check a draft on purpose.
Patterns are matched against the tail of each path, so name patterns like *.dag.yaml behave as with rglob. ** is not supported before Python 3.13.
Use on_dag_built to post-process DAGs after construction — adding tags, access controls, or audit metadata:
The callback receives the constructed Airflow DAG (mutable) and the Path of the YAML file that defined it.
skip_invalid_dags=True stops one bad YAML file from taking down every other DAG in the folder. Explain both costs before recommending it:
Pair it with blueprint lint in CI, so invalid files are caught somewhere visible.
Run CLI commands with uvx:
| Command | When to Use |
|---|---|
list | Show available blueprints, versions, sources, and DAG args templates |
describe <name> | Show config schema for a blueprint |
lint | Validate DAG YAML — bare to scan recursively, or pass one file |
vars <path> | Show resolved variables for a DAG and where each came from |
schema | Generate JSON Schema for a blueprint or for DAG-level fields |
new | Interactive DAG YAML creation. --output-dir picks where the file lands, which also selects the DAG args template it is validated against |
Every command takes --help, and -h / -v work as shorthands for --help / --version.
Run them from the project root, not from inside dags/ — a bare invocation resolves dags/ relative to the working directory, so running from within it finds no blueprints. Use --template-dir for any other layout.
Provider operators in the CLI. The
uvx --from airflow-blueprintenvironment is isolated and does not include the Airflow provider packages your Astro Runtime project has. If templates import provider operators, add--with <provider-package>so the CLI can import them — otherwiselist/lint/schemafail withModuleNotFoundError:
Versions are separate classes with a V{N} suffix: Extract is v1, ExtractV2 is v2, and each carries its own config model. A blueprint's discovered versions must form a contiguous 1..N sequence.
When the class name doesn't follow the convention, set them directly:
An explicit name must be snake_case (^[a-z][a-z0-9_]*$) or the class raises ValueError at definition time. Without one, the name is the snake_case form of the class name.
Omit version to get the latest; pin it to hold a step on an older one:
blueprint list shows the discovered versions of each blueprint.
Generate JSON schemas for editor autocompletion or external tooling. blueprint schema <name> emits a step template's config; blueprint schema --dag-args emits the DAG-level fields (dag_id, steps, and whatever your BlueprintDagArgs exposes). With multiple DAG args templates, --dag-args takes an optional template name.
Each emitted schema includes a top-level templateType field — "blueprint" for a step template, "dag_args" for DAG-level fields — so consumers can tell them apart. The command emits raw JSON when piped or written with -o/--output, and pretty, highlighted JSON when run interactively.
Write with
-o/--output, not>. Importing a template can print warnings to stdout — an Airflow deprecation warning from an operator import is the common case — and those interleave with the JSON, leaving redirected output unparseable.-owrites the schema alone.
Optional fields emit a plain type. An optional config field is published as
{"type": "string"}, not ananyOfwith a null branch — optionality is carried by the schema'srequiredarray alone. This keeps generated clients and form renderers from producing a union wrapper type for every optional field. Airflow params deliberately differ and keep a nullable type, because an unset optional param is an explicit null rather than an absent key; do not "fix" one to match the other.
After creating or modifying a blueprint, automatically check whether the project is an Astro project by looking for a .astro/ directory (created by astro dev init).
If it is, automatically regenerate schemas without prompting, writing one file per blueprint from blueprint list plus the DAG-level args schema, into blueprint/generated-schemas/. The Astro IDE reads that directory to render configuration forms, so keeping it in sync ensures the visual builder reflects the latest configs.
If you cannot determine whether the project is an Astro project, ask the user once and remember for the rest of the session.
Error messages carry their own remediation hints; read the message before applying anything here.
Cause: Blueprint class not in Python path.
Fix: Point the CLI at the right directory with --template-dir, and check blueprint list for what is actually discovered. If the blueprint is meant to come from an installed package, confirm entry-point discovery is on.
Cause: YAML field name typo with extra="forbid" enabled.
Fix: Run blueprint describe <name> to see valid field names.
Cause: Missing or broken loader — including a loader that imports a deprecated alias, which Airflow safe-mode may skip.
Fix: Ensure dags/loader.py calls build_all_airflow_dags(). If skip_invalid_dags=True is set, the file parses even when a DAG is broken, so check the DAG processor log rather than the import errors view.
Cause: The standalone uvx --from airflow-blueprint environment doesn't include the Airflow provider packages your project has, so a template importing provider operators can't be imported. This is the CLI's isolated environment, not your project.
Fix: Add --with apache-airflow-providers-X to the uvx invocation.
${...}Cause: A ${...} that is not a declared variable — commonly a shell variable in a bash_command, or a variable declared in a blueprint.vars.yaml outside the search root.
Fix: Escape non-variable occurrences as $${...}. For genuinely missing variables, run blueprint vars <path> to see what resolves and blueprint lint for the full list; check that --root matches the path the loader uses.
Cause: Variables that reference each other in a loop, or a composition chain deeper than the resolver's limit.
Fix: The error names the cycle or the chain. Break it by inlining one value; blueprint vars <path> shows what each variable resolves to.
Cause: Not that several templates exist — that is supported. These fire when resolution is ambiguous: two templates in one directory, two sharing a name, or more than one declaring default=True.
Fix: Move one template to the directory whose DAGs should use it, rename one with name = "...", or leave only one default=True. blueprint list shows which template applies where.
Cause: A named DAG args template was requested that isn't registered.
Fix: Check the name against blueprint list.
Cause: An installed package advertising blueprints failed to import.
Fix: Import the module directly to see the real traceback, and confirm the package and its dependencies are installed in the same environment as Airflow.
Cause: A blueprint's versions don't form a contiguous 1..N sequence, or YAML pins a version that doesn't exist.
Fix: Ensure versions increment by one with no gaps; run blueprint list to see available versions.
Cause: A config field uses a type Blueprint rejects — a multi-type union (e.g. str | int), bare Any, or dict[str, Any].
Fix: Use a single, explicit type. Optional[X] / X | None is still allowed. See Creating Blueprints → Config Field Types Must Be YAML-Compatible.
Cause: Circular depends_on references.
Fix: Review step dependencies and remove cycles.
Every Blueprint task has extra fields in Rendered Template:
blueprint_step_config — resolved YAML configblueprint_step_code — Python source of the blueprintBefore finishing, verify with the user:
blueprint list shows their templates, and the expected DAG args template applies to each pathblueprint lint passes (bare to scan all *.dag.yaml recursively, or pass a specific file — passing a directory path fails with Is a directory)blueprint vars resolves as expected, if variables or profiles are in usedags/loader.py exists and calls build_all_airflow_dags()