--- a/taskcluster/taskgraph/generator.py
+++ b/taskcluster/taskgraph/generator.py
@@ -5,16 +5,17 @@
 from __future__ import absolute_import, print_function, unicode_literals
 import logging
 import os
 import yaml
 
 from .graph import Graph
 from .taskgraph import TaskGraph
 from .optimize import optimize_task_graph
+from .util.python_path import find_object
 
 logger = logging.getLogger(__name__)
 
 
 class Kind(object):
 
     def __init__(self, name, path, config):
         self.name = name
@@ -22,28 +23,17 @@ class Kind(object):
         self.config = config
 
     def _get_impl_class(self):
         # load the class defined by implementation
         try:
             impl = self.config['implementation']
         except KeyError:
             raise KeyError("{!r} does not define implementation".format(self.path))
-        if impl.count(':') != 1:
-            raise TypeError('{!r} implementation does not have the form "module:object"'
-                            .format(self.path))
-
-        impl_module, impl_object = impl.split(':')
-        impl_class = __import__(impl_module)
-        for a in impl_module.split('.')[1:]:
-            impl_class = getattr(impl_class, a)
-        for a in impl_object.split('.'):
-            impl_class = getattr(impl_class, a)
-
-        return impl_class
+        return find_object(impl)
 
     def load_tasks(self, parameters, loaded_tasks):
         impl_class = self._get_impl_class()
         return impl_class.load_tasks(self.name, self.path, self.config,
                                      parameters, loaded_tasks)
 
 
 class TaskGraphGenerator(object):

new file mode 100644
--- /dev/null
+++ b/taskcluster/taskgraph/test/test_util_python_path.py
@@ -0,0 +1,31 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+from __future__ import absolute_import, print_function, unicode_literals
+
+import unittest
+from ..util import python_path
+
+
+class TestObject(object):
+
+    testClassProperty = object()
+
+
+class TestPythonPath(unittest.TestCase):
+
+    def test_find_object_no_such_module(self):
+        """find_object raises ImportError for a nonexistent module"""
+        self.assertRaises(ImportError, python_path.find_object, "no_such_module:someobj")
+
+    def test_find_object_no_such_object(self):
+        """find_object raises AttributeError for a nonexistent object"""
+        self.assertRaises(AttributeError, python_path.find_object,
+                          "taskgraph.test.test_util_python_path:NoSuchObject")
+
+    def test_find_object_exists(self):
+        """find_object finds an existing object"""
+        obj = python_path.find_object(
+            "taskgraph.test.test_util_python_path:TestObject.testClassProperty")
+        self.assertIs(obj, TestObject.testClassProperty)

new file mode 100644
--- /dev/null
+++ b/taskcluster/taskgraph/util/python_path.py
@@ -0,0 +1,27 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+from __future__ import absolute_import, print_function, unicode_literals
+
+
+def find_object(path):
+    """
+    Find a Python object given a path of the form <modulepath>:<objectpath>.
+    Conceptually equivalent to
+
+        def find_object(modulepath, objectpath):
+            import <modulepath> as mod
+            return mod.<objectpath>
+    """
+    if path.count(':') != 1:
+        raise ValueError(
+            'python path {!r} does not have the form "module:object"'.format(path))
+
+    modulepath, objectpath = path.split(':')
+    obj = __import__(modulepath)
+    for a in modulepath.split('.')[1:]:
+        obj = getattr(obj, a)
+    for a in objectpath.split('.'):
+        obj = getattr(obj, a)
+    return obj