samraj2k · Sep 15, 2026 · Sep 17, 2026 · Sep 21, 2026 · Sep 21, 2026 · Sep 21, 2026
diff --git a/airflow-core/docs/administration-and-deployment/dag-bundles.rst b/airflow-core/docs/administration-and-deployment/dag-bundles.rst
@@ -64,6 +64,59 @@ Configuring Dag bundles
 
 Dag bundles are configured in :ref:`config:dag_processor__dag_bundle_config_list`. You can add one or more Dag bundles here.
 
+Writing a bundle source
+~~~~~~~~~~~~~~~~~~~~~~~
+
+Set ``[dag_processor] dag_bundle_source`` to the import path of a class that extends
+``DagBundleSource``. The default, ``ConfigDagBundleSource``, reads ``dag_bundle_config_list``.
+This is a class extension point; it does not require a new Airflow provider package.
+
+The class must support construction without arguments and implement two methods:
+
+* ``get_configured_bundle_metadata()`` returns the complete list of configured bundle names and,
+  in multi-team mode, their team names. Airflow manages validation, example bundles, and database state.
+* ``get_bundle(name, version, version_data)`` constructs the requested bundle. The source chooses
+  its class and settings and forwards both version arguments to the bundle.
+
+For example, save this minimal source in an importable module named ``my_deployment.bundles``:
+
+.. code-block:: python
+
+    from airflow.dag_processing.bundles.local import LocalDagBundle
+    from airflow.dag_processing.bundles.source import DagBundleMetadata, DagBundleSource
+
+
+    class DeploymentDagBundleSource(DagBundleSource):
+        def get_configured_bundle_metadata(self):
+            return [DagBundleMetadata(name="team-dags")]
+
+        def get_bundle(self, name, version=None, version_data=None):
+            if name != "team-dags":
+                raise ValueError(f"Unknown bundle: {name}")
+            return LocalDagBundle(
+                name=name,
+                version=version,
+                version_data=version_data,
+                path="/opt/airflow/team-dags",
+            )
+
+Configure it with:
+
+.. code-block:: ini
+
+    [dag_processor]
+    dag_bundle_source = my_deployment.bundles.DeploymentDagBundleSource
+
+Install the module in each Airflow process that resolves bundles, including Dag processors and workers.
+A source can read deployment-controlled configuration from a file or service instead of using the fixed
+values above. Raise an exception when discovery fails; return an empty list only when no bundles are
+configured. Do not include Airflow's example bundles in that list.
+
+Keep a bundle name tied to the same implementation and construction settings. If a bundle leaves the
+configured list, its name must remain resolvable while retained Dag runs can still need its code.
+Keep access to the configuration source under Deployment Manager control and reference credentials
+through Airflow Connections rather than embedding them in bundle settings.
+
 .. warning:: Reference credentials through a Connection — do not inline them
 
     Bundle ``kwargs`` are stored in the ``[dag_processor] dag_bundle_config_list``

@@ -112,7 +112,7 @@ def _dag_processor_instance_health(job: Job) -> dict[str, Any]:
 # behind forever and a restarted replica adds a second one. The denominator is therefore taken from
 # the declared work partition instead, which is unaffected by how replicas come and go:
 #
-# * Dag processor -- the bundles in ``[dag_processor] dag_bundle_config_list``.
+# * Dag processor -- the bundles returned by the configured Dag bundle source.
 # * Triggerer -- the team scopes those bundles declare, since a triggerer only picks up triggers for
 #   its own team (see ``Trigger.ids_for_triggerer``).
 # * Scheduler -- schedulers are symmetric, so there is no partition and no partial state to report.

@@ -215,6 +215,10 @@ def uses_default_dag_discovery(self) -> bool:
         unset there; an environment variable counts as a choice whatever its value. Other bundles are
         left alone, since refreshing a remote bundle fetches from it.
         """
+        if conf.get("dag_processor", "dag_bundle_source") != conf.get_default_value(
+            "dag_processor", "dag_bundle_source"
+        ):
+            return False
         if REFRESH_INTERVAL_ENV_VAR in os.environ:
             return False
         if conf.get("dag_processor", "refresh_interval") != conf.get_default_value(

@@ -218,7 +218,7 @@ def team_sync(args, *, session=NEW_SESSION):
 
     dag_bundle_teams = {
         bundle.team_name
-        for bundle in DagBundlesManager()._bundle_config.values()
+        for bundle in DagBundlesManager().get_active_bundle_metadata()
         if bundle.team_name is not None
     }
     existing_teams = Team.get_all_team_names(session=session)
@@ -311,9 +311,9 @@ def team_verify(args, *, session=NEW_SESSION):
 
     existing_teams = {team.name for team in teams}
 
-    for bundle_name, bundle in DagBundlesManager()._bundle_config.items():
+    for bundle in DagBundlesManager().get_active_bundle_metadata():
         if bundle.team_name and bundle.team_name not in existing_teams:
-            issues.append(f"DAG bundle '{bundle_name}' references unknown team '{bundle.team_name}'.")
+            issues.append(f"DAG bundle '{bundle.name}' references unknown team '{bundle.team_name}'.")
 
     if issues:
         print("Verification failed.\n")

diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml
@@ -3277,6 +3277,19 @@ dag_processor:
       example: "/tmp/some-place"
       default: ~
 
+    dag_bundle_source:
+      description: |
+        Import path for the class that provides configured Dag bundle metadata and constructs Dag bundles.
+        The class must extend ``airflow.dag_processing.bundles.source.DagBundleSource`` and have a
+        no-argument constructor.
+
+        The default implementation reads ``dag_bundle_config_list``. A custom implementation can read
+        bundle configuration from another source and must return the complete set of configured bundles.
+      version_added: 3.5.0
+      type: string
+      example: ~
+      default: "airflow.dag_processing.bundles.source.ConfigDagBundleSource"
+
     dag_bundle_config_list:
       description: |
         List of backend configs.  Must supply name, classpath, and kwargs for each backend.