diff --git a/python/ray/_raylet.pyx b/python/ray/_raylet.pyx
index 0953b02a2..806e7b5a3 100644
--- a/python/ray/_raylet.pyx
+++ b/python/ray/_raylet.pyx
@@ -486,10 +486,10 @@ cdef execute_task(
             if isinstance(error, RayTaskError):
                 # Avoid recursive nesting of RayTaskError.
                 failure_object = RayTaskError(function_name, backtrace,
-                                              error.cause_cls, proctitle=title)
+                                              error.cause, proctitle=title)
             else:
                 failure_object = RayTaskError(function_name, backtrace,
-                                              error.__class__, proctitle=title)
+                                              error, proctitle=title)
             errors = []
             for _ in range(c_return_ids.size()):
                 errors.append(failure_object)
diff --git a/python/ray/exceptions.py b/python/ray/exceptions.py
index 8adc46044..75d382fd7 100644
--- a/python/ray/exceptions.py
+++ b/python/ray/exceptions.py
@@ -41,17 +41,12 @@ class RayTaskError(RayError):
     retrieved from the object store, the Python method that retrieved it checks
     to see if the object is a RayTaskError and if it is then an exception is
     thrown propagating the error message.
-
-    Attributes:
-        function_name (str): The name of the function that failed and produced
-            the RayTaskError.
-        traceback_str (str): The traceback from the exception.
     """
 
     def __init__(self,
                  function_name,
                  traceback_str,
-                 cause_cls,
+                 cause,
                  proctitle=None,
                  pid=None,
                  ip=None):
@@ -64,34 +59,43 @@ class RayTaskError(RayError):
         self.ip = ip or ray.services.get_node_ip_address()
         self.function_name = function_name
         self.traceback_str = traceback_str
-        self.cause_cls = cause_cls
+        # TODO(edoakes): should we handle non-serializable exception objects?
+        self.cause = cause
         assert traceback_str is not None
 
     def as_instanceof_cause(self):
-        """Returns copy that is an instance of the cause's Python class.
+        """Returns an exception that is an instance of the cause's class.
 
         The returned exception will inherit from both RayTaskError and the
-        cause class.
+        cause class and will contain all of the attributes of the cause
+        exception.
         """
 
-        if issubclass(RayTaskError, self.cause_cls):
+        cause_cls = self.cause.__class__
+        if issubclass(RayTaskError, cause_cls):
             return self  # already satisfied
 
-        if issubclass(self.cause_cls, RayError):
+        if issubclass(cause_cls, RayError):
             return self  # don't try to wrap ray internal errors
 
-        class cls(RayTaskError, self.cause_cls):
-            def __init__(self, function_name, traceback_str, cause_cls,
-                         proctitle, pid, ip):
-                RayTaskError.__init__(self, function_name, traceback_str,
-                                      cause_cls, proctitle, pid, ip)
+        cause = self.cause
+        error_msg = str(self)
 
-        name = "RayTaskError({})".format(self.cause_cls.__name__)
+        class cls(RayTaskError, cause_cls):
+            def __init__(self):
+                pass
+
+            def __getattr__(self, name):
+                return getattr(cause, name)
+
+            def __str__(self):
+                return error_msg
+
+        name = "RayTaskError({})".format(cause_cls.__name__)
         cls.__name__ = name
         cls.__qualname__ = name
 
-        return cls(self.function_name, self.traceback_str, self.cause_cls,
-                   self.proctitle, self.pid, self.ip)
+        return cls()
 
     def __str__(self):
         """Format a RayTaskError as a string."""
diff --git a/python/ray/tests/test_failure.py b/python/ray/tests/test_failure.py
index 7813caa55..7ce16a5f0 100644
--- a/python/ray/tests/test_failure.py
+++ b/python/ray/tests/test_failure.py
@@ -63,7 +63,12 @@ def test_failed_task(ray_start_regular):
             assert False
 
     class CustomException(ValueError):
-        pass
+        def __init__(self, msg):
+            super().__init__(msg)
+            self.field = 1
+
+        def f(self):
+            return 2
 
     @ray.remote
     def f():
@@ -73,9 +78,12 @@ def test_failed_task(ray_start_regular):
         ray.get(f.remote())
     except Exception as e:
         assert "This function failed." in str(e)
+        assert isinstance(e, ValueError)
         assert isinstance(e, CustomException)
         assert isinstance(e, ray.exceptions.RayTaskError)
         assert "RayTaskError(CustomException)" in repr(e)
+        assert e.field == 1
+        assert e.f() == 2
     else:
         # ray.get should throw an exception.
         assert False