diff --git a/include/forcing/ForcingsEngineDataProvider.hpp b/include/forcing/ForcingsEngineDataProvider.hpp
new file mode 100644
index 0000000000..5dcf8f8f69
--- /dev/null
+++ b/include/forcing/ForcingsEngineDataProvider.hpp
@@ -0,0 +1,219 @@
+#pragma once
+
+#include <NGenConfig.h>
+
+#if NGEN_WITH_PYTHON
+
+#include <cmath>
+#include <chrono>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "DataProvider.hpp"
+#include "bmi/Bmi_Py_Adapter.hpp"
+
+namespace data_access {
+
+//! Python module name for NextGen Forcings Engine
+static constexpr auto forcings_engine_python_module = "NextGen_Forcings_Engine";
+
+//! Python classname for NextGen Forcings Engine BMI class
+static constexpr auto forcings_engine_python_class  = "NWMv3_Forcing_Engine_BMI_model";
+
+//! Full Python classpath for Forcings Engine BMI class
+static constexpr auto forcings_engine_python_classpath = "NextGen_Forcings_Engine.NWMv3_Forcing_Engine_BMI_model";
+
+//! Default time format used for parsing timestamp strings
+static constexpr auto default_time_format = "%Y-%m-%d %H:%M:%S";
+
+namespace detail {
+
+//! Parse time string from format.
+//! Utility function for ForcingsEngineLumpedDataProvider constructor.
+time_t parse_time(const std::string& time, const std::string& fmt);
+
+//! Check that requirements for running the forcings engine
+//! are available at runtime. If requirements are not available,
+//! then this function throws.
+void assert_forcings_engine_requirements();
+
+//! Storage for Forcings Engine-specific BMI instances.
+struct ForcingsEngineStorage {
+    //! Key type for Forcings Engine storage, storing file paths to initialization files.
+    using key_type = std::string;
+
+    //! BMI adapter type used by the Python-based Forcings Engine.
+    using bmi_type = models::bmi::Bmi_Py_Adapter;
+
+    //! Value type stored, shared pointer to BMI adapter.
+    using value_type = std::shared_ptr<bmi_type>;
+
+    static ForcingsEngineStorage instances;
+
+    //! Get a Forcings Engine instance.
+    //! @param key Initialization file path for Forcings Engine instance.
+    //! @return Shared pointer to a Forcings Engine BMI instance, or @c nullptr if it has not
+    //!         been created yet.
+    value_type get(const key_type& key)
+    {
+        auto pos = data_.find(key);
+        if (pos == data_.end()) {
+          return nullptr;
+        }
+
+        return pos->second;
+    }
+
+    //! Associate a Forcings Engine instance to a file path.
+    //! @param key Initialization file path for Forcings Engine instance.
+    //! @param value Shared pointer to a Forcings Engine BMI instance.
+    void set(const key_type& key, value_type value)
+    {
+        data_[key] = value;
+    }
+
+    //! Clear all references to Forcings Engine instances.
+    //! @note This will not necessarily destroy the Forcings Engine instances. Since they
+    //!       are reference counted, it will only decrement their instance by one.
+    void clear()
+    {
+        data_.clear();
+    }
+
+  private:
+    //! Instance map of underlying BMI models.
+    std::unordered_map<key_type, value_type> data_;
+};
+
+} // namespace detail
+
+
+//! Forcings Engine Data Provider
+//! @tparam DataType Data type for values returned from derived classes
+//! @tparam SelectionType Selector type for querying data from derived classes
+template<typename DataType, typename SelectionType>
+struct ForcingsEngineDataProvider
+  : public DataProvider<DataType, SelectionType>
+{
+    using data_type = DataType;
+    using selection_type = SelectionType;
+    using clock_type = std::chrono::system_clock;
+
+    ~ForcingsEngineDataProvider() override = default;
+
+    boost::span<const std::string> get_available_variable_names() override
+    {
+        return var_output_names_;
+    }
+
+    long get_data_start_time() override
+    {
+        return clock_type::to_time_t(time_begin_);
+    }
+
+    long get_data_stop_time() override
+    {
+        return clock_type::to_time_t(time_end_);
+    }
+
+    long record_duration() override
+    {
+        return std::chrono::duration_cast<std::chrono::seconds>(time_step_).count();
+    }
+
+    size_t get_ts_index_for_time(const time_t& epoch_time) override
+    {
+        const auto epoch = clock_type::from_time_t(epoch_time);
+
+        if (epoch < time_begin_ || epoch > time_end_) {
+          throw std::out_of_range{
+            "epoch " + std::to_string(epoch.time_since_epoch().count()) +
+            " out of range of " + std::to_string(time_begin_.time_since_epoch().count()) + ", "
+            + std::to_string(time_end_.time_since_epoch().count())
+          };
+        }
+
+        return (epoch - time_begin_) / time_step_;
+    }
+
+    std::shared_ptr<models::bmi::Bmi_Py_Adapter> model() noexcept
+    {
+        return bmi_;
+    }
+
+    /* Remaining virtual member functions from DataProvider must be implemented
+       by derived classes. */
+
+    data_type get_value(
+      const selection_type& selector,
+      data_access::ReSampleMethod m
+    ) override = 0;
+
+    std::vector<data_type> get_values(
+      const selection_type& selector,
+      data_access::ReSampleMethod m
+    ) override = 0;
+
+  protected:
+    using storage_type = detail::ForcingsEngineStorage;
+
+    //! Forcings Engine Data Provider Constructor
+    //!
+    //! @note Derived implementations should delegate to this constructor
+    //!       to acquire a shared forcings engine instance.
+    ForcingsEngineDataProvider(
+        const std::string& init,
+        std::size_t time_begin_seconds,
+        std::size_t time_end_seconds
+    )
+      : time_begin_(std::chrono::seconds{time_begin_seconds})
+      , time_end_(std::chrono::seconds{time_end_seconds})
+    {
+        // Get a forcings engine instance if it exists for this initialization file
+        bmi_ = storage_type::instances.get(init);
+
+        // If it doesn't exist, create it and assign it to the storage map
+        if (bmi_ == nullptr) {
+            // Outside of this branch, this->bmi_ != nullptr after this
+            bmi_ = std::make_shared<models::bmi::Bmi_Py_Adapter>(
+                "ForcingsEngine",
+                init,
+                forcings_engine_python_classpath,
+                /*has_fixed_time_step=*/true
+            );
+
+            storage_type::instances.set(init, bmi_);
+        }
+
+        // Now, initialize the BMI dependent instance members
+        // NOTE: using std::lround instead of static_cast will prevent potential UB
+        time_step_ = std::chrono::seconds{std::lround(bmi_->GetTimeStep())};
+        var_output_names_ = bmi_->GetOutputVarNames();
+    }
+
+    //! Forcings Engine instance
+    std::shared_ptr<models::bmi::Bmi_Py_Adapter> bmi_ = nullptr;
+
+    //! Output variable names
+    std::vector<std::string> var_output_names_{};
+
+  private:
+    //! Initialization config file path
+    std::string init_;
+
+    //! Calendar time for simulation beginning
+    clock_type::time_point time_begin_{};
+
+    //! Calendar time for simulation end
+    clock_type::time_point time_end_{};
+
+    //! Duration of a single simulation tick
+    clock_type::duration time_step_{};
+};
+
+} // namespace data_access
+
+#endif // NGEN_WITH_PYTHON
diff --git a/include/forcing/GenericDataProvider.hpp b/include/forcing/GenericDataProvider.hpp
index 6bdeeb4f05..d4c6bd7ae9 100644
--- a/include/forcing/GenericDataProvider.hpp
+++ b/include/forcing/GenericDataProvider.hpp
@@ -6,12 +6,7 @@
 
 namespace data_access
 {
-    class GenericDataProvider : public DataProvider<double, CatchmentAggrDataSelector>
-    {
-        public:
-
-        private:
-    };
+    using GenericDataProvider = DataProvider<double, CatchmentAggrDataSelector>;
 }
 
-#endif
\ No newline at end of file
+#endif
diff --git a/include/forcing/NullForcingProvider.hpp b/include/forcing/NullForcingProvider.hpp
index 1b1f9c7caa..a7d3c8b6d8 100644
--- a/include/forcing/NullForcingProvider.hpp
+++ b/include/forcing/NullForcingProvider.hpp
@@ -3,8 +3,6 @@
 
 #include <vector>
 #include <string>
-#include <stdexcept>
-#include <limits>
 #include "GenericDataProvider.hpp"
 
 /**
@@ -14,43 +12,25 @@ class NullForcingProvider : public data_access::GenericDataProvider
 {
     public:
 
-    NullForcingProvider(){}
+    NullForcingProvider();
 
     // BEGIN DataProvider interface methods
 
-    long get_data_start_time() override {
-        return 0;
-    }
+    long get_data_start_time() override;
 
-    long get_data_stop_time() override {
-        return LONG_MAX;
-    }
+    long get_data_stop_time() override;
 
-    long record_duration() override {
-        return 1;
-    }
+    long record_duration() override;
 
-    size_t get_ts_index_for_time(const time_t &epoch_time) override {
-        return 0;
-    }
+    size_t get_ts_index_for_time(const time_t &epoch_time) override;
 
-    double get_value(const CatchmentAggrDataSelector& selector, data_access::ReSampleMethod m) override
-    {
-        throw std::runtime_error("Called get_value function in NullDataProvider");
-    }
+    double get_value(const CatchmentAggrDataSelector& selector, data_access::ReSampleMethod m) override;
 
-    virtual std::vector<double> get_values(const CatchmentAggrDataSelector& selector, data_access::ReSampleMethod m) override
-    {
-        throw std::runtime_error("Called get_values function in NullDataProvider");
-    }
+    std::vector<double> get_values(const CatchmentAggrDataSelector& selector, data_access::ReSampleMethod m) override;
 
-    inline bool is_property_sum_over_time_step(const std::string& name) override {
-        throw std::runtime_error("Got request for variable " + name + " but no such variable is provided by NullForcingProvider." + SOURCE_LOC);
-    }
+    inline bool is_property_sum_over_time_step(const std::string& name) override;
 
-    boost::span<const std::string> get_available_variable_names() override {
-        return {};
-    }
+    boost::span<const std::string> get_available_variable_names() override;
 };
 
 #endif // NGEN_NULLFORCING_H
diff --git a/src/forcing/CMakeLists.txt b/src/forcing/CMakeLists.txt
index 38924215ea..51ec7d22a8 100644
--- a/src/forcing/CMakeLists.txt
+++ b/src/forcing/CMakeLists.txt
@@ -1,8 +1,7 @@
-include(${PROJECT_SOURCE_DIR}/cmake/dynamic_sourced_library.cmake)
-dynamic_sourced_cxx_library(forcing "${CMAKE_CURRENT_SOURCE_DIR}")
-
+add_library(forcing)
 add_library(NGen::forcing ALIAS forcing)
 
+
 find_package(Threads REQUIRED)
 
 target_include_directories(forcing PUBLIC
@@ -14,13 +13,26 @@ target_include_directories(forcing PUBLIC
         )
 
 target_link_libraries(forcing PUBLIC
+        NGen::config_header
+        NGen::core
         Boost::boost                # Headers-only Boost
         NGen::config_header
         Threads::Threads
 )
 
+target_sources(forcing PRIVATE "${CMAKE_CURRENT_LIST_DIR}/NullForcingProvider.cpp")
+
 if(NGEN_WITH_NETCDF)
+    target_sources(forcing PRIVATE "${CMAKE_CURRENT_LIST_DIR}/NetCDFPerFeatureDataProvider.cpp")
     target_link_libraries(forcing PUBLIC NetCDF)
 endif()
 
+if(NGEN_WITH_PYTHON)
+    target_sources(forcing
+      PRIVATE
+        "${CMAKE_CURRENT_LIST_DIR}/ForcingsEngineDataProvider.cpp"
+    )
+    target_link_libraries(forcing PUBLIC pybind11::embed NGen::ngen_bmi)
+endif()
+
 #target_compile_options(forcing PUBLIC -std=c++14 -Wall)
diff --git a/src/forcing/ForcingsEngineDataProvider.cpp b/src/forcing/ForcingsEngineDataProvider.cpp
new file mode 100644
index 0000000000..5eef9a90f2
--- /dev/null
+++ b/src/forcing/ForcingsEngineDataProvider.cpp
@@ -0,0 +1,48 @@
+#include <forcing/ForcingsEngineDataProvider.hpp>
+#include <utilities/python/InterpreterUtil.hpp>
+
+#include <ctime> // timegm
+#include <iomanip> // std::get_time
+
+namespace data_access {
+namespace detail {
+
+// Initialize instance storage
+ForcingsEngineStorage ForcingsEngineStorage::instances{};
+
+time_t parse_time(const std::string& time, const std::string& fmt)
+{
+    std::tm tm_ = {};
+    std::stringstream tmstr{time};
+    tmstr >> std::get_time(&tm_, fmt.c_str());
+
+    // Note: `timegm` is available for Linux and macOS via time.h, but not Windows.
+    return timegm(&tm_);
+}
+
+void assert_forcings_engine_requirements()
+{
+    // Check that the python module is installed.
+    {
+        auto interpreter_ = utils::ngenPy::InterpreterUtil::getInstance();
+        try {
+            auto mod = interpreter_->getModule(forcings_engine_python_module);
+            auto cls = mod.attr(forcings_engine_python_class).cast<py::object>();
+        } catch(std::exception& e) {
+            throw std::runtime_error{
+                "Failed to initialize ForcingsEngine: ForcingsEngine python module is not installed or is not properly configured. (" + std::string{e.what()} + ")"
+            };
+        }
+    }
+
+    // Check that the WGRIB2 environment variable is defined
+    {
+        const auto* wgrib2_exec = std::getenv("WGRIB2");
+        if (wgrib2_exec == nullptr) {
+            throw std::runtime_error{"Failed to initialize ForcingsEngine: $WGRIB2 is not defined"};
+        }
+    }
+}
+
+} // namespace detail
+} // namespace data_access
diff --git a/src/forcing/NullForcingProvider.cpp b/src/forcing/NullForcingProvider.cpp
new file mode 100644
index 0000000000..6307f7fbfa
--- /dev/null
+++ b/src/forcing/NullForcingProvider.cpp
@@ -0,0 +1,43 @@
+#include "NullForcingProvider.hpp"
+
+#include <stdexcept>
+#include <limits>
+
+#include <all.h>
+
+NullForcingProvider::NullForcingProvider() = default;
+
+long NullForcingProvider::get_data_start_time()
+{
+    return 0;
+}
+
+long NullForcingProvider::get_data_stop_time() {
+    return std::numeric_limits<long>::max();
+}
+
+long NullForcingProvider::record_duration() {
+    return 1;
+}
+
+size_t NullForcingProvider::get_ts_index_for_time(const time_t &epoch_time) {
+    return 0;
+}
+
+double NullForcingProvider::get_value(const CatchmentAggrDataSelector& selector, data_access::ReSampleMethod m)
+{
+    throw std::runtime_error("Called get_value function in NullDataProvider");
+}
+
+std::vector<double> NullForcingProvider::get_values(const CatchmentAggrDataSelector& selector, data_access::ReSampleMethod m)
+{
+    throw std::runtime_error("Called get_values function in NullDataProvider");
+}
+
+inline bool NullForcingProvider::is_property_sum_over_time_step(const std::string& name) {
+    throw std::runtime_error("Got request for variable " + name + " but no such variable is provided by NullForcingProvider." + SOURCE_LOC);
+}
+
+boost::span<const std::string> NullForcingProvider::get_available_variable_names() {
+    return {};
+}
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index fbde394e70..80f68377b8 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -183,6 +183,16 @@ ngen_add_test(
         NGen::geojson
 )
 
+ngen_add_test(
+    test_forcings_engine
+    OBJECTS
+        forcing/ForcingsEngineDataProvider_Test.cpp
+    LIBRARIES
+        NGen::forcing
+    REQUIRES
+        NGEN_WITH_PYTHON
+)
+
 ########################## Series Unit Tests
 ngen_add_test(
     test_mdarray
diff --git a/test/forcing/ForcingsEngineDataProvider_Test.cpp b/test/forcing/ForcingsEngineDataProvider_Test.cpp
new file mode 100644
index 0000000000..e5fb709e75
--- /dev/null
+++ b/test/forcing/ForcingsEngineDataProvider_Test.cpp
@@ -0,0 +1,5 @@
+#include <forcing/ForcingsEngineDataProvider.hpp>
+
+#include "DataProviderSelectors.hpp"
+
+template struct data_access::ForcingsEngineDataProvider<double, CatchmentAggrDataSelector>;