@@ -1764,10 +1764,12 @@ _PyTuple_Resize
 PyTuple_SetItem
 PyTuple_Size
 PyTuple_Type
+PyType_AddWatcher
 _PyType_CalculateMetaclass
 _PyType_CheckConsistency
 PyType_ClearCache
 _PyType_ClearNoShadowingInstances
+PyType_ClearWatcher
 PyType_FromModuleAndSpec
 PyType_FromSpec
 PyType_FromSpecWithBases
@@ -1789,6 +1791,8 @@ PyType_Ready
 _PyType_SetNoShadowingInstances
 _PyTypes_InitSlotDefs
 PyType_Type
+PyType_Unwatch
+PyType_Watch
 _Py_Uid_Converter
 PyUnicode_Append
 PyUnicode_AppendAndDel
 
@@ -565,6 +565,14 @@ PyAPI_FUNC(int) _PyTrash_cond(PyObject *op, destructor dealloc);
 #define Py_TRASHCAN_SAFE_BEGIN(op) Py_TRASHCAN_BEGIN_CONDITION(op, 1)
 #define Py_TRASHCAN_SAFE_END(op) Py_TRASHCAN_END
 
+#define TYPE_MAX_WATCHERS 8
+
+typedef int(*PyType_WatchCallback)(PyTypeObject *);
+PyAPI_FUNC(int) PyType_AddWatcher(PyType_WatchCallback callback);
+PyAPI_FUNC(int) PyType_ClearWatcher(int watcher_id);
+PyAPI_FUNC(int) PyType_Watch(int watcher_id, PyObject *type);
+PyAPI_FUNC(int) PyType_Unwatch(int watcher_id, PyObject *type);
+
 /* Attempt to assign a version tag to the given type.
  *
  * Returns 1 if the type already had a valid version tag or a new one was
 
@@ -298,6 +298,7 @@ struct _is {
     struct atexit_state atexit;
 
     PyObject *audit_hooks;
+    PyType_WatchCallback type_watchers[TYPE_MAX_WATCHERS];
     PyCode_WatchCallback code_watchers[CODE_MAX_WATCHERS];
     // One bit is set for each non-NULL entry in code_watchers
     uint8_t active_code_watchers;
 
@@ -2,7 +2,7 @@
 # these are all functions _testcapi exports whose name begins with 'test_'.
 
 from collections import OrderedDict
-from contextlib import contextmanager
+from contextlib import contextmanager, ExitStack
 import importlib.machinery
 import importlib.util
 import os
@@ -1221,6 +1221,185 @@ def test_clear_unassigned_watcher_id(self):
             self.clear_watcher(1)
 
 
+class TestTypeWatchers(unittest.TestCase):
+    # types of watchers testcapimodule can add:
+    TYPES = 0    # appends modified types to global event list
+    ERROR = 1    # unconditionally sets and signals a RuntimeException
+    WRAP = 2     # appends modified type wrapped in list to global event list
+
+    # duplicating the C constant
+    TYPE_MAX_WATCHERS = 8
+
+    def add_watcher(self, kind=TYPES):
+        return _testcapi.add_type_watcher(kind)
+
+    def clear_watcher(self, watcher_id):
+        _testcapi.clear_type_watcher(watcher_id)
+
+    @contextmanager
+    def watcher(self, kind=TYPES):
+        wid = self.add_watcher(kind)
+        try:
+            yield wid
+        finally:
+            self.clear_watcher(wid)
+
+    def assert_events(self, expected):
+        actual = _testcapi.get_type_modified_events()
+        self.assertEqual(actual, expected)
+
+    def watch(self, wid, t):
+        _testcapi.watch_type(wid, t)
+
+    def unwatch(self, wid, t):
+        _testcapi.unwatch_type(wid, t)
+
+    def test_watch_type(self):
+        class C: pass
+        with self.watcher() as wid:
+            self.watch(wid, C)
+            C.foo = "bar"
+            self.assert_events([C])
+
+    def test_event_aggregation(self):
+        class C: pass
+        with self.watcher() as wid:
+            self.watch(wid, C)
+            C.foo = "bar"
+            C.bar = "baz"
+            # only one event registered for both modifications
+            self.assert_events([C])
+
+    def test_lookup_resets_aggregation(self):
+        class C: pass
+        with self.watcher() as wid:
+            self.watch(wid, C)
+            C.foo = "bar"
+            # lookup resets type version tag
+            self.assertEqual(C.foo, "bar")
+            C.bar = "baz"
+            # both events registered
+            self.assert_events([C, C])
+
+    def test_unwatch_type(self):
+        class C: pass
+        with self.watcher() as wid:
+            self.watch(wid, C)
+            C.foo = "bar"
+            self.assertEqual(C.foo, "bar")
+            self.assert_events([C])
+            self.unwatch(wid, C)
+            C.bar = "baz"
+            self.assert_events([C])
+
+    def test_clear_watcher(self):
+        class C: pass
+        # outer watcher is unused, it's just to keep events list alive
+        with self.watcher() as _:
+            with self.watcher() as wid:
+                self.watch(wid, C)
+                C.foo = "bar"
+                self.assertEqual(C.foo, "bar")
+                self.assert_events([C])
+            C.bar = "baz"
+            # Watcher on C has been cleared, no new event
+            self.assert_events([C])
+
+    def test_watch_type_subclass(self):
+        class C: pass
+        class D(C): pass
+        with self.watcher() as wid:
+            self.watch(wid, D)
+            C.foo = "bar"
+            self.assert_events([D])
+
+    def test_error(self):
+        class C: pass
+        with self.watcher(kind=self.ERROR) as wid:
+            self.watch(wid, C)
+            with catch_unraisable_exception() as cm:
+                C.foo = "bar"
+                self.assertIs(cm.unraisable.object, C)
+                self.assertEqual(str(cm.unraisable.exc_value), "boom!")
+            self.assert_events([])
+
+    def test_two_watchers(self):
+        class C1: pass
+        class C2: pass
+        with self.watcher() as wid1:
+            with self.watcher(kind=self.WRAP) as wid2:
+                self.assertNotEqual(wid1, wid2)
+                self.watch(wid1, C1)
+                self.watch(wid2, C2)
+                C1.foo = "bar"
+                C2.hmm = "baz"
+                self.assert_events([C1, [C2]])
+
+    def test_all_watchers(self):
+        class C: pass
+        with ExitStack() as stack:
+            last_wid = -1
+            # don't make assumptions about how many watchers are already
+            # registered, just go until we reach the max ID
+            while last_wid < self.TYPE_MAX_WATCHERS - 1:
+                last_wid = stack.enter_context(self.watcher())
+            self.watch(last_wid, C)
+            C.foo = "bar"
+            self.assert_events([C])
+
+    def test_watch_non_type(self):
+        with self.watcher() as wid:
+            with self.assertRaisesRegex(ValueError, r"Cannot watch non-type"):
+                self.watch(wid, 1)
+
+    def test_watch_out_of_range_watcher_id(self):
+        class C: pass
+        with self.assertRaisesRegex(ValueError, r"Invalid type watcher ID -1"):
+            self.watch(-1, C)
+        with self.assertRaisesRegex(ValueError, r"Invalid type watcher ID 8"):
+            self.watch(self.TYPE_MAX_WATCHERS, C)
+
+    def test_watch_unassigned_watcher_id(self):
+        class C: pass
+        with self.assertRaisesRegex(ValueError, r"No type watcher set for ID 1"):
+            self.watch(1, C)
+
+    def test_unwatch_non_type(self):
+        with self.watcher() as wid:
+            with self.assertRaisesRegex(ValueError, r"Cannot watch non-type"):
+                self.unwatch(wid, 1)
+
+    def test_unwatch_out_of_range_watcher_id(self):
+        class C: pass
+        with self.assertRaisesRegex(ValueError, r"Invalid type watcher ID -1"):
+            self.unwatch(-1, C)
+        with self.assertRaisesRegex(ValueError, r"Invalid type watcher ID 8"):
+            self.unwatch(self.TYPE_MAX_WATCHERS, C)
+
+    def test_unwatch_unassigned_watcher_id(self):
+        class C: pass
+        with self.assertRaisesRegex(ValueError, r"No type watcher set for ID 1"):
+            self.unwatch(1, C)
+
+    def test_clear_out_of_range_watcher_id(self):
+        with self.assertRaisesRegex(ValueError, r"Invalid type watcher ID -1"):
+            self.clear_watcher(-1)
+        with self.assertRaisesRegex(ValueError, r"Invalid type watcher ID 8"):
+            self.clear_watcher(self.TYPE_MAX_WATCHERS)
+
+    def test_clear_unassigned_watcher_id(self):
+        with self.assertRaisesRegex(ValueError, r"No type watcher set for ID 1"):
+            self.clear_watcher(1)
+
+    def test_no_more_ids_available(self):
+        contexts = [self.watcher() for i in range(self.TYPE_MAX_WATCHERS)]
+        with ExitStack() as stack:
+            for ctx in contexts:
+                stack.enter_context(ctx)
+            with self.assertRaisesRegex(RuntimeError, r"no more type watcher IDs"):
+                self.add_watcher()
+
+
 class TestCodeObjectWatchers(unittest.TestCase):
     @contextmanager
     def code_watcher(self, which_watcher):
 
@@ -6265,6 +6265,128 @@ get_dict_watcher_events(PyObject *self, PyObject *Py_UNUSED(args))
     return Py_NewRef(g_dict_watch_events);
 }
 
+// Test type watchers
+static PyObject *g_type_modified_events;
+static int g_type_watchers_installed;
+
+static int
+type_modified_callback(PyTypeObject *type)
+{
+    assert(PyList_Check(g_type_modified_events));
+    if(PyList_Append(g_type_modified_events, (PyObject *)type) < 0) {
+        return -1;
+    }
+    return 0;
+}
+
+static int
+type_modified_callback_wrap(PyTypeObject *type)
+{
+    assert(PyList_Check(g_type_modified_events));
+    PyObject *list = PyList_New(0);
+    if (list == NULL) {
+        return -1;
+    }
+    if (PyList_Append(list, (PyObject *)type) < 0) {
+        Py_DECREF(list);
+        return -1;
+    }
+    if (PyList_Append(g_type_modified_events, list) < 0) {
+        Py_DECREF(list);
+        return -1;
+    }
+    Py_DECREF(list);
+    return 0;
+}
+
+static int
+type_modified_callback_error(PyTypeObject *type)
+{
+    PyErr_SetString(PyExc_RuntimeError, "boom!");
+    return -1;
+}
+
+static PyObject *
+add_type_watcher(PyObject *self, PyObject *kind)
+{
+    int watcher_id;
+    assert(PyLong_Check(kind));
+    long kind_l = PyLong_AsLong(kind);
+    if (kind_l == 2) {
+        watcher_id = PyType_AddWatcher(type_modified_callback_wrap);
+    }
+    else if (kind_l == 1) {
+        watcher_id = PyType_AddWatcher(type_modified_callback_error);
+    }
+    else {
+        watcher_id = PyType_AddWatcher(type_modified_callback);
+    }
+    if (watcher_id < 0) {
+        return NULL;
+    }
+    if (!g_type_watchers_installed) {
+        assert(!g_type_modified_events);
+        if (!(g_type_modified_events = PyList_New(0))) {
+            return NULL;
+        }
+    }
+    g_type_watchers_installed++;
+    return PyLong_FromLong(watcher_id);
+}
+
+static PyObject *
+clear_type_watcher(PyObject *self, PyObject *watcher_id)
+{
+    if (PyType_ClearWatcher(PyLong_AsLong(watcher_id))) {
+        return NULL;
+    }
+    g_type_watchers_installed--;
+    if (!g_type_watchers_installed) {
+        assert(g_type_modified_events);
+        Py_CLEAR(g_type_modified_events);
+    }
+    Py_RETURN_NONE;
+}
+
+static PyObject *
+get_type_modified_events(PyObject *self, PyObject *Py_UNUSED(args))
+{
+    if (!g_type_modified_events) {
+        PyErr_SetString(PyExc_RuntimeError, "no watchers active");
+        return NULL;
+    }
+    return Py_NewRef(g_type_modified_events);
+}
+
+static PyObject *
+watch_type(PyObject *module, PyObject *args)
+{
+    int watcher_id;
+    PyObject *type;
+    if (!PyArg_ParseTuple(args, "iO", &watcher_id, &type)) {
+        return NULL;
+    }
+    if (PyType_Watch(watcher_id, type)) {
+        return NULL;
+    }
+    Py_RETURN_NONE;
+}
+
+static PyObject *
+unwatch_type(PyObject *module, PyObject *args)
+{
+    int watcher_id;
+    PyObject *type;
+    if (!PyArg_ParseTuple(args, "iO", &watcher_id, &type)) {
+        return NULL;
+    }
+    if (PyType_Unwatch(watcher_id, type)) {
+        return NULL;
+    }
+    Py_RETURN_NONE;
+}
+
+
 // Test code object watching
 
 #define NUM_CODE_WATCHERS 2
@@ -6925,6 +7047,11 @@ static PyMethodDef TestMethods[] = {
     {"watch_dict", watch_dict, METH_VARARGS},
     {"unwatch_dict", unwatch_dict, METH_VARARGS},
     {"get_dict_watcher_events", get_dict_watcher_events, METH_NOARGS},
+    {"add_type_watcher", add_type_watcher, METH_O},
+    {"clear_type_watcher", clear_type_watcher, METH_O},
+    {"watch_type", watch_type, METH_VARARGS},
+    {"unwatch_type", unwatch_type, METH_VARARGS},
+    {"get_type_modified_events", get_type_modified_events, METH_NOARGS},
     {"add_code_watcher", add_code_watcher, METH_O},
     {"clear_code_watcher", clear_code_watcher, METH_O},
     {"get_code_watcher_num_created_events",