diff --git a/taskflow/exceptions.py b/taskflow/exceptions.py
index b2b44193b..ae7baef48 100644
--- a/taskflow/exceptions.py
+++ b/taskflow/exceptions.py
@@ -15,17 +15,50 @@
 #    under the License.
 
 import os
+import sys
 import traceback
 
 import six
 
 
+def raise_with_cause(exc_cls, message, *args, **kwargs):
+    """Helper to raise + chain exceptions (when able) and associate a *cause*.
+
+    NOTE(harlowja): Since in py3.x exceptions can be chained (due to
+    :pep:`3134`) we should try to raise the desired exception with the given
+    *cause* (or extract a *cause* from the current stack if able) so that the
+    exception formats nicely in old and new versions of python. Since py2.x
+    does **not** support exception chaining (or formatting) our root exception
+    class has a :py:meth:`~taskflow.exceptions.TaskFlowException.pformat`
+    method that can be used to get *similar* information instead (and this
+    function makes sure to retain the *cause* in that case as well so
+    that the :py:meth:`~taskflow.exceptions.TaskFlowException.pformat` method
+    shows them).
+
+    :param exc_cls: the :py:class:`~taskflow.exceptions.TaskFlowException`
+                    class to raise.
+    :param message: the text/str message that will be passed to
+                    the exceptions constructor as its first positional
+                    argument.
+    :param args: any additional positional arguments to pass to the
+                 exceptions constructor.
+    :param kwargs: any additional keyword arguments to pass to the
+                   exceptions constructor.
+    """
+    if 'cause' not in kwargs:
+        exc_type, exc, exc_tb = sys.exc_info()
+        if exc is not None:
+            kwargs['cause'] = exc
+        del(exc_type, exc, exc_tb)
+    six.raise_from(exc_cls(message, *args, **kwargs), kwargs.get('cause'))
+
+
 class TaskFlowException(Exception):
     """Base class for *most* exceptions emitted from this library.
 
     NOTE(harlowja): in later versions of python we can likely remove the need
-    to have a cause here as PY3+ have implemented PEP 3134 which handles
-    chaining in a much more elegant manner.
+    to have a ``cause`` here as PY3+ have implemented :pep:`3134` which
+    handles chaining in a much more elegant manner.
 
     :param message: the exception message, typically some string that is
                     useful for consumers to view when debugging or analyzing
diff --git a/taskflow/tests/unit/test_exceptions.py b/taskflow/tests/unit/test_exceptions.py
new file mode 100644
index 000000000..d834ce7c1
--- /dev/null
+++ b/taskflow/tests/unit/test_exceptions.py
@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+
+#    Copyright (C) 2013 Yahoo! Inc. All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import six
+import testtools
+
+from taskflow import exceptions as exc
+from taskflow import test
+
+
+class TestExceptions(test.TestCase):
+    def test_cause(self):
+        capture = None
+        try:
+            raise exc.TaskFlowException("broken", cause=IOError("dead"))
+        except Exception as e:
+            capture = e
+        self.assertIsNotNone(capture)
+        self.assertIsInstance(capture, exc.TaskFlowException)
+        self.assertIsNotNone(capture.cause)
+        self.assertIsInstance(capture.cause, IOError)
+
+    def test_cause_pformat(self):
+        capture = None
+        try:
+            raise exc.TaskFlowException("broken", cause=IOError("dead"))
+        except Exception as e:
+            capture = e
+        self.assertIsNotNone(capture)
+        self.assertGreater(0, len(capture.pformat()))
+
+    def test_raise_with(self):
+        capture = None
+        try:
+            raise IOError('broken')
+        except Exception:
+            try:
+                exc.raise_with_cause(exc.TaskFlowException, 'broken')
+            except Exception as e:
+                capture = e
+        self.assertIsNotNone(capture)
+        self.assertIsInstance(capture, exc.TaskFlowException)
+        self.assertIsNotNone(capture.cause)
+        self.assertIsInstance(capture.cause, IOError)
+
+    @testtools.skipIf(not six.PY3, 'py3.x is not available')
+    def test_raise_with_cause(self):
+        capture = None
+        try:
+            raise IOError('broken')
+        except Exception:
+            try:
+                exc.raise_with_cause(exc.TaskFlowException, 'broken')
+            except Exception as e:
+                capture = e
+        self.assertIsNotNone(capture)
+        self.assertIsInstance(capture, exc.TaskFlowException)
+        self.assertIsNotNone(capture.cause)
+        self.assertIsInstance(capture.cause, IOError)
+        self.assertIsNotNone(capture.__cause__)
+        self.assertIsInstance(capture.__cause__, IOError)