diff --git a/taskflow/engines/helpers.py b/taskflow/engines/helpers.py
index c70c8f1e..c9000b80 100644
--- a/taskflow/engines/helpers.py
+++ b/taskflow/engines/helpers.py
@@ -19,6 +19,7 @@ import contextlib
 import six
 import stevedore.driver
 
+from taskflow import exceptions as exc
 from taskflow.openstack.common import importutils
 from taskflow.persistence import backends as p_backends
 from taskflow.utils import misc
@@ -98,15 +99,19 @@ def load(flow, store=None, flow_detail=None, book=None,
         flow_detail = p_utils.create_flow_detail(flow, book=book,
                                                  backend=backend)
 
-    mgr = stevedore.driver.DriverManager(
-        namespace, engine_name,
-        invoke_on_load=True,
-        invoke_args=(flow, flow_detail, backend, engine_conf),
-        invoke_kwds=kwargs)
-    engine = mgr.driver
-    if store:
-        engine.storage.inject(store)
-    return engine
+    try:
+        mgr = stevedore.driver.DriverManager(
+            namespace, engine_name,
+            invoke_on_load=True,
+            invoke_args=(flow, flow_detail, backend, engine_conf),
+            invoke_kwds=kwargs)
+        engine = mgr.driver
+    except RuntimeError as e:
+        raise exc.NotFound("Could not find engine %s" % (engine_name), e)
+    else:
+        if store:
+            engine.storage.inject(store)
+        return engine
 
 
 def run(flow, store=None, flow_detail=None, book=None,
diff --git a/taskflow/tests/unit/test_engine_helpers.py b/taskflow/tests/unit/test_engine_helpers.py
index 6f3d3386..da0a276b 100644
--- a/taskflow/tests/unit/test_engine_helpers.py
+++ b/taskflow/tests/unit/test_engine_helpers.py
@@ -16,6 +16,8 @@
 
 import mock
 
+from taskflow import exceptions as exc
+from taskflow.patterns import linear_flow
 from taskflow import test
 from taskflow.tests import utils as test_utils
 from taskflow.utils import persistence_utils as p_utils
@@ -23,6 +25,20 @@ from taskflow.utils import persistence_utils as p_utils
 import taskflow.engines
 
 
+class EngineLoadingTestCase(test.TestCase):
+    def test_default_load(self):
+        f = linear_flow.Flow('test')
+        f.add(test_utils.TaskOneReturn("run-1"))
+        e = taskflow.engines.load(f)
+        self.assertIsNotNone(e)
+
+    def test_unknown_load(self):
+        f = linear_flow.Flow('test')
+        f.add(test_utils.TaskOneReturn("run-1"))
+        self.assertRaises(exc.NotFound, taskflow.engines.load, f,
+                          engine_conf='not_really_any_engine')
+
+
 class FlowFromDetailTestCase(test.TestCase):
     def test_no_meta(self):
         _lb, flow_detail = p_utils.temporary_flow_detail()