diff --git a/pygeoapi/api/__init__.py b/pygeoapi/api/__init__.py
index b47541f231..b2a1ee944d 100644
--- a/pygeoapi/api/__init__.py
+++ b/pygeoapi/api/__init__.py
@@ -43,7 +43,7 @@
 import asyncio
 from collections import OrderedDict
 from copy import deepcopy
-from datetime import datetime
+from datetime import (datetime, timezone)
 from functools import partial
 from gzip import compress
 from http import HTTPStatus
@@ -220,6 +220,31 @@ def apply_gzip(headers: dict, content: Union[str, bytes]) -> Union[str, bytes]:
     return content
 
 
+def pre_load_colls(func):
+    """
+    Decorator that makes sure the loaded collections in memory are updated
+    before the function is executed.
+
+    :param func: decorated function
+
+    :returns: `func`
+    """
+
+    def inner(*args, **kwargs):
+        cls = args[0]
+
+        # Validation on the method name for the provided class instance on this
+        # decoration function
+        if hasattr(cls, 'reload_resources_if_necessary'):
+            # Validate the resources are up to date
+            cls.reload_resources_if_necessary()
+
+        # Continue
+        return func(*args, **kwargs)
+
+    return inner
+
+
 class APIRequest:
     """
     Transforms an incoming server-specific Request into an object
@@ -682,9 +707,74 @@ def __init__(self, config, openapi):
         self.tpl_config = deepcopy(self.config)
         self.tpl_config['server']['url'] = self.base_url
 
+        # Now that the basic configuration is read, call the load_resources function.  # noqa
+        # This call enables the api engine to load resources dynamically.
+        # This pattern allows for loading resources coming from another
+        # source (e.g. a database) rather than from the yaml file.
+        # This, along with the @pre_load_colls decorative function, enables
+        # resources management on multiple distributed pygeoapi instances.
+        self.load_resources()
+
         self.manager = get_manager(self.config)
         LOGGER.info('Process manager plugin loaded')
 
+    def on_load_resources(self, resources: dict) -> dict:
+        """
+        Overridable function to load the available resources dynamically.
+        By default, this function simply returns the provided resources
+        as-is. This is the native behavior of the API; expecting
+        resources to be already configured correctly per the yaml config
+        file.
+
+        :param resources: the resources as currently configured
+        (self.config['resources'])
+        :returns: the resources dictionary that's available in the API.
+        """
+
+        # By default, return the same resources object, unchanged.
+        return resources
+
+    def on_load_resources_check(self, last_loaded_resources: datetime) -> bool:  # noqa
+        """
+        Overridable function to check if the resources should be reloaded.
+        This implementation depends on your environment and messaging broker.
+        Natively, the resources used by the pygeoapi instance are strictly
+        the ones from the yaml configuration file. It doesn't support
+        resources changing on-the-fly. Therefore, False is returned here
+        and they are never reloaded.
+        """
+
+        # By default, return False to not reload the resources.
+        return False
+
+    def load_resources(self) -> None:
+        """
+        Calls on_load_resources and reassigns the resources configuration.
+        """
+
+        # Call on_load_resources sending the current resources configuration.
+        self.config['resources'] = self.on_load_resources(self.config['resources'])  # noqa
+
+        # Copy over for the template config (this is something that got added
+        # after a rebase of pending PR.. to be investigated if still
+        # necessary to do so..)
+        # self.tpl_config['resources'] = deepcopy(self.config['resources'])
+
+        # Keep track of UTC date of last time resources were loaded
+        self.last_loaded_resources = datetime.now(timezone.utc)
+
+    def reload_resources_if_necessary(self) -> None:
+        """
+        Checks if the resources should be reloaded by calling overridable
+        function 'on_load_resources_check' and then, when necessary, calls
+        'load_resources'.
+        """
+
+        # If the resources should be reloaded
+        if self.on_load_resources_check(self.last_loaded_resources):
+            # Reload the resources
+            self.load_resources()
+
     @gzip
     @pre_process
     @jsonldify
@@ -898,6 +988,7 @@ def conformance(self,
     @gzip
     @pre_process
     @jsonldify
+    @pre_load_colls
     def describe_collections(self, request: Union[APIRequest, Any],
                              dataset=None) -> Tuple[dict, int, str]:
         """
diff --git a/pygeoapi/api/coverages.py b/pygeoapi/api/coverages.py
index c6687042c8..0e3fcdb1eb 100644
--- a/pygeoapi/api/coverages.py
+++ b/pygeoapi/api/coverages.py
@@ -51,7 +51,7 @@
 
 from . import (
     APIRequest, API, F_JSON, SYSTEM_LOCALE, validate_bbox, validate_datetime,
-    validate_subset
+    validate_subset, pre_load_colls
 )
 
 LOGGER = logging.getLogger(__name__)
@@ -68,6 +68,7 @@
 ]
 
 
+@pre_load_colls
 def get_collection_coverage(
         api: API, request: APIRequest, dataset) -> Tuple[dict, int, str]:
     """
diff --git a/pygeoapi/api/itemtypes.py b/pygeoapi/api/itemtypes.py
index 008b28cb7d..d9320075f3 100644
--- a/pygeoapi/api/itemtypes.py
+++ b/pygeoapi/api/itemtypes.py
@@ -63,7 +63,7 @@
 
 from . import (
     APIRequest, API, SYSTEM_LOCALE, F_JSON, FORMAT_TYPES, F_HTML, F_JSONLD,
-    validate_bbox, validate_datetime
+    validate_bbox, validate_datetime, pre_load_colls
 )
 
 LOGGER = logging.getLogger(__name__)
@@ -100,6 +100,7 @@
 ]
 
 
+@pre_load_colls
 def get_collection_queryables(api: API, request: Union[APIRequest, Any],
                               dataset=None) -> Tuple[dict, int, str]:
     """
@@ -194,6 +195,7 @@ def get_collection_queryables(api: API, request: Union[APIRequest, Any],
     return headers, HTTPStatus.OK, to_json(queryables, api.pretty_print)
 
 
+@pre_load_colls
 def get_collection_items(
         api: API, request: Union[APIRequest, Any],
         dataset) -> Tuple[dict, int, str]:
@@ -631,6 +633,7 @@ def get_collection_items(
     return headers, HTTPStatus.OK, to_json(content, api.pretty_print)
 
 
+@pre_load_colls
 def post_collection_items(
         api: API, request: APIRequest, dataset) -> Tuple[dict, int, str]:
     """
@@ -916,6 +919,7 @@ def post_collection_items(
     return headers, HTTPStatus.OK, to_json(content, api.pretty_print)
 
 
+@pre_load_colls
 def manage_collection_item(
         api: API, request: APIRequest,
         action, dataset, identifier=None) -> Tuple[dict, int, str]:
@@ -1027,6 +1031,7 @@ def manage_collection_item(
         return headers, HTTPStatus.OK, ''
 
 
+@pre_load_colls
 def get_collection_item(api: API, request: APIRequest,
                         dataset, identifier) -> Tuple[dict, int, str]:
     """