diff --git a/Doc/library/_interpreters.rst b/Doc/library/_interpreters.rst
new file mode 100644
index 00000000000000..bbc945016e2880
--- /dev/null
+++ b/Doc/library/_interpreters.rst
@@ -0,0 +1,90 @@
+:mod:`_interpreters` --- Low-level interpreters API
+===================================================
+
+.. module:: _interpreters
+   :synopsis: Low-level interpreters API.
+
+.. versionadded:: 3,7
+
+--------------
+
+This module provides low-level primitives for working with multiple
+Python interpreters in the same runtime in the current process.
+
+More information about (sub)interpreters is found at
+:ref:`sub-interpreter-support`, including what data is shared between
+interpreters and what is unique.  Note particularly that interpreters
+aren't inherently threaded, even though they track and manage Python
+threads.  To run code in an interpreter in a different OS thread, call
+:func:`run_string` in a function that you run in a new Python thread.
+For example::
+
+   id = _interpreters.create()
+   def f():
+       _interpreters.run_string(id, 'print("in a thread")')
+
+   t = threading.Thread(target=f)
+   t.start()
+
+This module is optional.  It is provided by Python implementations which
+support multiple interpreters.
+
+It defines the following functions:
+
+.. function:: enumerate()
+
+   Return a list of the IDs of every existing interpreter.
+
+
+.. function:: get_current()
+
+   Return the ID of the currently running interpreter.
+
+
+.. function:: get_main()
+
+   Return the ID of the main interpreter.
+
+
+.. function:: is_running(id)
+
+   Return whether or not the identified interpreter is currently
+   running any code.
+
+
+.. function:: create()
+
+   Initialize a new Python interpreter and return its identifier.  The
+   interpreter will be created in the current thread and will remain
+   idle until something is run in it.
+
+
+.. function:: destroy(id)
+
+   Finalize and destroy the identified interpreter.
+
+
+.. function:: run_string(id, command)
+
+   A wrapper around :c:func:`PyRun_SimpleString` which runs the provided
+   Python program in the main thread of the identified interpreter.
+   Providing an invalid or unknown ID results in a RuntimeError,
+   likewise if the main interpreter or any other running interpreter
+   is used.
+
+   Any value returned from the code is thrown away, similar to what
+   threads do.  If the code results in an exception then that exception
+   is raised in the thread in which run_string() was called, similar to
+   how :func:`exec` works.  This aligns with how interpreters are not
+   inherently threaded.  Note that SystemExit (as raised by sys.exit())
+   is not treated any differently and will result in the process ending
+   if not caught explicitly.
+
+
+.. function:: run_string_unrestricted(id, command, ns=None)
+
+   Like :c:func:`run_string` but returns the dict in which the code
+   was executed.  It also supports providing a namespace that gets
+   merged into the execution namespace before execution.  Note that
+   this allows objects to leak between interpreters, which may not
+   be desirable.
diff --git a/Doc/library/concurrency.rst b/Doc/library/concurrency.rst
index 0de281bd149531..76b9c52f593555 100644
--- a/Doc/library/concurrency.rst
+++ b/Doc/library/concurrency.rst
@@ -29,3 +29,4 @@ The following are support modules for some of the above services:
    dummy_threading.rst
    _thread.rst
    _dummy_thread.rst
+   _interpreters.rst
diff --git a/Doc/library/interpreters.rst b/Doc/library/interpreters.rst
new file mode 100644
index 00000000000000..bef62d30a673fe
--- /dev/null
+++ b/Doc/library/interpreters.rst
@@ -0,0 +1,64 @@
+:mod:`interpreters` --- high-level interpreters API
+===================================================
+
+.. module:: interpreters
+   :synopsis: High-level interpreters API.
+
+**Source code:** :source:`Lib/interpreters.py`
+
+.. versionadded:: 3,7
+
+--------------
+
+This module provides high-level interfaces for interacting with
+the Python interpreters.  It is built on top of the lower- level
+:mod:`_interpreters` module.
+
+.. XXX Summarize :ref:`_sub-interpreter-support` here.
+
+This module defines the following functions:
+
+.. function:: enumerate()
+
+   Return a list of all existing interpreters.
+
+
+.. function:: get_current()
+
+   Return the currently running interpreter.
+
+
+.. function:: get_main()
+
+   Return the main interpreter.
+
+
+.. function:: create()
+
+   Initialize a new Python interpreter and return it.  The
+   interpreter will be created in the current thread and will remain
+   idle until something is run in it.
+
+
+This module also defines the following class:
+
+.. class:: Interpreter(id)
+
+   ``id`` is the interpreter's ID.
+
+   .. property:: id
+
+      The interpreter's ID.
+
+   .. method:: is_running()
+
+      Return whether or not the interpreter is currently running.
+
+   .. method:: destroy()
+
+      Finalize and destroy the interpreter.
+
+   .. method:: run(code)
+
+      Run the provided Python code in the interpreter, in the current
+      OS thread.  Supported code: source text.
diff --git a/Lib/interpreters.py b/Lib/interpreters.py
new file mode 100644
index 00000000000000..3bfc567b6a79b0
--- /dev/null
+++ b/Lib/interpreters.py
@@ -0,0 +1,79 @@
+"""Interpreters module providing a high-level API to Python interpreters."""
+
+import _interpreters
+
+
+__all__ = ['enumerate', 'current', 'main', 'create', 'Interpreter']
+
+
+def enumerate():
+    """Return a list of all existing interpreters."""
+    return [Interpreter(id)
+            for id in _interpreters.enumerate()]
+
+
+def current():
+    """Return the currently running interpreter."""
+    id = _interpreters.get_current()
+    return Interpreter(id)
+
+
+def main():
+    """Return the main interpreter."""
+    id = _interpreters.get_main()
+    return Interpreter(id)
+
+
+def create():
+    """Return a new Interpreter.
+
+    The interpreter is created in the current OS thread.  It will remain
+    idle until its run() method is called.
+    """
+    id = _interpreters.create()
+    return Interpreter(id)
+
+
+class Interpreter:
+    """A single Python interpreter in the current runtime."""
+
+    def __init__(self, id):
+        self._id = id
+
+    def __repr__(self):
+        return '{}(id={!r})'.format(type(self).__name__, self._id)
+
+    def __eq__(self, other):
+        try:
+            other_id = other.id
+        except AttributeError:
+            return False
+        return self._id == other_id
+
+    @property
+    def id(self):
+        """The interpreter's ID."""
+        return self._id
+
+    def is_running(self):
+        """Return whether or not the interpreter is currently running."""
+        return _interpreters.is_running(self._id)
+
+    def destroy(self):
+        """Finalize and destroy the interpreter.
+        
+        After calling destroy(), all operations on this interpreter
+        will fail.
+        """
+        _interpreters.destroy(self._id)
+
+    def run(self, code):
+        """Run the provided Python code in this interpreter.
+
+        If the code is a string then it will be run as it would by
+        calling exec().  Other kinds of code are not supported.
+        """
+        if isinstance(code, str):
+            _interpreters.run_string(self._id, code)
+        else:
+            raise NotImplementedError
diff --git a/Lib/test/test__interpreters.py b/Lib/test/test__interpreters.py
new file mode 100644
index 00000000000000..29d04aa4d00cfb
--- /dev/null
+++ b/Lib/test/test__interpreters.py
@@ -0,0 +1,582 @@
+import contextlib
+import os
+import os.path
+import shutil
+import tempfile
+from textwrap import dedent, indent
+import threading
+import unittest
+
+from test import support
+from test.support import script_helper
+
+interpreters = support.import_module('_interpreters')
+
+
+SCRIPT_THREADED_INTERP = """\
+from textwrap import dedent
+import threading
+import _interpreters
+def f():
+    _interpreters.run_string(id, dedent('''
+        {}
+        '''))
+
+t = threading.Thread(target=f)
+t.start()
+"""
+
+
+@contextlib.contextmanager
+def _blocked(dirname):
+    filename = os.path.join(dirname, '.lock')
+    wait_script = dedent("""
+        import os.path
+        import time
+        while not os.path.exists('{}'):
+            time.sleep(0.1)
+        """).format(filename)
+    try:
+        yield wait_script
+    finally:
+        support.create_empty_file(filename)
+
+
+class InterpreterTests(unittest.TestCase):
+
+    def setUp(self):
+        self.dirname = tempfile.mkdtemp()
+
+    def tearDown(self):
+        shutil.rmtree(self.dirname)
+
+    def test_still_running_at_exit(self):
+        subscript = dedent("""
+            import time
+            # Give plenty of time for the main interpreter to finish.
+            time.sleep(1_000_000)
+            """)
+        script = SCRIPT_THREADED_INTERP.format(indent(subscript, '        '))
+        filename = script_helper.make_script(self.dirname, 'interp', script)
+        with script_helper.spawn_python(filename) as proc:
+            retcode = proc.wait()
+
+        self.assertEqual(retcode, 0)
+
+
+class TestBase(unittest.TestCase):
+
+    def tearDown(self):
+        for id in interpreters.enumerate():
+            if id == 0:  # main
+                continue
+            try:
+                interpreters.destroy(id)
+            except RuntimeError:
+                pass  # already destroyed
+
+
+class EnumerateTests(TestBase):
+
+    def test_multiple(self):
+        main, = interpreters.enumerate()
+        id1 = interpreters.create()
+        id2 = interpreters.create()
+        ids = interpreters.enumerate()
+
+        self.assertEqual(set(ids), {main, id1, id2})
+
+    def test_main_only(self):
+        main, = interpreters.enumerate()
+
+        self.assertEqual(main, 0)
+
+
+class GetCurrentTests(TestBase):
+
+    def test_main(self):
+        main, = interpreters.enumerate()
+        id = interpreters.get_current()
+
+        self.assertEqual(id, main)
+
+    def test_sub(self):
+        id1 = interpreters.create()
+        ns = interpreters.run_string_unrestricted(id1, dedent("""
+            import _interpreters
+            id = _interpreters.get_current()
+            """))
+        id2 = ns['id']
+
+        self.assertEqual(id2, id1)
+
+
+class GetMainTests(TestBase):
+
+    def test_main(self):
+        expected, = interpreters.enumerate()
+        main = interpreters.get_main()
+
+        self.assertEqual(main, 0)
+        self.assertEqual(main, expected)
+
+
+class IsRunningTests(TestBase):
+
+    def test_main_running(self):
+        main, = interpreters.enumerate()
+        sub = interpreters.create()
+        main_running = interpreters.is_running(main)
+        sub_running = interpreters.is_running(sub)
+
+        self.assertTrue(main_running)
+        self.assertFalse(sub_running)
+
+    def test_sub_running(self):
+        main, = interpreters.enumerate()
+        sub1 = interpreters.create()
+        sub2 = interpreters.create()
+        ns = interpreters.run_string_unrestricted(sub1, dedent(f"""
+            import _interpreters
+            main = _interpreters.is_running({main})
+            sub1 = _interpreters.is_running({sub1})
+            sub2 = _interpreters.is_running({sub2})
+            """))
+        main_running = ns['main']
+        sub1_running = ns['sub1']
+        sub2_running = ns['sub2']
+
+        self.assertTrue(main_running)
+        self.assertTrue(sub1_running)
+        self.assertFalse(sub2_running)
+
+
+class CreateTests(TestBase):
+
+    def test_in_main(self):
+        id = interpreters.create()
+
+        self.assertIn(id, interpreters.enumerate())
+
+    @unittest.skip('enable this test when working on pystate.c')
+    def test_unique_id(self):
+        seen = set()
+        for _ in range(100):
+            id = interpreters.create()
+            interpreters.destroy(id)
+            seen.add(id)
+
+        self.assertEqual(len(seen), 100)
+
+    def test_in_thread(self):
+        lock = threading.Lock()
+        id = None
+        def f():
+            nonlocal id
+            id = interpreters.create()
+            lock.acquire()
+            lock.release()
+
+        t = threading.Thread(target=f)
+        with lock:
+            t.start()
+        t.join()
+        self.assertIn(id, interpreters.enumerate())
+
+    def test_in_subinterpreter(self):
+        main, = interpreters.enumerate()
+        id1 = interpreters.create()
+        ns = interpreters.run_string_unrestricted(id1, dedent("""
+            import _interpreters
+            id = _interpreters.create()
+            """))
+        id2 = ns['id']
+
+        self.assertEqual(set(interpreters.enumerate()), {main, id1, id2})
+
+    def test_in_threaded_subinterpreter(self):
+        main, = interpreters.enumerate()
+        id1 = interpreters.create()
+        ns = None
+        script = dedent("""
+            import _interpreters
+            id = _interpreters.create()
+            """)
+        def f():
+            nonlocal ns
+            ns = interpreters.run_string_unrestricted(id1, script)
+
+        t = threading.Thread(target=f)
+        t.start()
+        t.join()
+        id2 = ns['id']
+
+        self.assertEqual(set(interpreters.enumerate()), {main, id1, id2})
+
+
+    def test_after_destroy_all(self):
+        before = set(interpreters.enumerate())
+        # Create 3 subinterpreters.
+        ids = []
+        for _ in range(3):
+            id = interpreters.create()
+            ids.append(id)
+        # Now destroy them.
+        for id in ids:
+            interpreters.destroy(id)
+        # Finally, create another.
+        id = interpreters.create()
+        self.assertEqual(set(interpreters.enumerate()), before | {id})
+
+    def test_after_destroy_some(self):
+        before = set(interpreters.enumerate())
+        # Create 3 subinterpreters.
+        id1 = interpreters.create()
+        id2 = interpreters.create()
+        id3 = interpreters.create()
+        # Now destroy 2 of them.
+        interpreters.destroy(id1)
+        interpreters.destroy(id3)
+        # Finally, create another.
+        id = interpreters.create()
+        self.assertEqual(set(interpreters.enumerate()), before | {id, id2})
+
+
+class DestroyTests(TestBase):
+
+    def test_one(self):
+        id1 = interpreters.create()
+        id2 = interpreters.create()
+        id3 = interpreters.create()
+        self.assertIn(id2, interpreters.enumerate())
+        interpreters.destroy(id2)
+        self.assertNotIn(id2, interpreters.enumerate())
+        self.assertIn(id1, interpreters.enumerate())
+        self.assertIn(id3, interpreters.enumerate())
+
+    def test_all(self):
+        before = set(interpreters.enumerate())
+        ids = set()
+        for _ in range(3):
+            id = interpreters.create()
+            ids.add(id)
+        self.assertEqual(set(interpreters.enumerate()), before | ids)
+        for id in ids:
+            interpreters.destroy(id)
+        self.assertEqual(set(interpreters.enumerate()), before)
+
+    def test_main(self):
+        main, = interpreters.enumerate()
+        with self.assertRaises(RuntimeError):
+            interpreters.destroy(main)
+
+        def f():
+            with self.assertRaises(RuntimeError):
+                interpreters.destroy(main)
+
+        t = threading.Thread(target=f)
+        t.start()
+        t.join()
+
+    def test_already_destroyed(self):
+        id = interpreters.create()
+        interpreters.destroy(id)
+        with self.assertRaises(RuntimeError):
+            interpreters.destroy(id)
+
+    def test_does_not_exist(self):
+        with self.assertRaises(RuntimeError):
+            interpreters.destroy(1_000_000)
+
+    def test_bad_id(self):
+        with self.assertRaises(RuntimeError):
+            interpreters.destroy(-1)
+
+    def test_from_current(self):
+        main, = interpreters.enumerate()
+        id = interpreters.create()
+        script = dedent("""
+            import _interpreters
+            _interpreters.destroy({})
+            """).format(id)
+
+        with self.assertRaises(RuntimeError):
+            interpreters.run_string(id, script)
+        self.assertEqual(set(interpreters.enumerate()), {main, id})
+
+    def test_from_sibling(self):
+        main, = interpreters.enumerate()
+        id1 = interpreters.create()
+        id2 = interpreters.create()
+        script = dedent("""
+            import _interpreters
+            _interpreters.destroy({})
+            """).format(id2)
+        interpreters.run_string(id1, script)
+
+        self.assertEqual(set(interpreters.enumerate()), {main, id1})
+
+    def test_from_other_thread(self):
+        id = interpreters.create()
+        def f():
+            interpreters.destroy(id)
+
+        t = threading.Thread(target=f)
+        t.start()
+        t.join()
+
+    def test_still_running(self):
+        # XXX Rewrite this test without files by using
+        # run_string_unrestricted().
+        main, = interpreters.enumerate()
+        id = interpreters.create()
+        def f():
+            interpreters.run_string(id, wait_script)
+
+        dirname = tempfile.mkdtemp()
+        t = threading.Thread(target=f)
+        with _blocked(dirname) as wait_script:
+            t.start()
+            with self.assertRaises(RuntimeError):
+                interpreters.destroy(id)
+
+        t.join()
+        self.assertEqual(set(interpreters.enumerate()), {main, id})
+
+
+class RunStringTests(TestBase):
+
+    SCRIPT = dedent("""
+        with open('{}', 'w') as out:
+            out.write('{}')
+        """)
+    FILENAME = 'spam'
+
+    def setUp(self):
+        self.id = interpreters.create()
+        self.dirname = None
+        self.filename = None
+
+    def tearDown(self):
+        if self.dirname is not None:
+            try:
+                shutil.rmtree(self.dirname)
+            except FileNotFoundError:
+                pass  # already deleted
+        super().tearDown()
+
+    def _resolve_filename(self, name=None):
+        if name is None:
+            name = self.FILENAME
+        if self.dirname is None:
+            self.dirname = tempfile.mkdtemp()
+        return os.path.join(self.dirname, name)
+
+    def _empty_file(self):
+        self.filename = self._resolve_filename()
+        support.create_empty_file(self.filename)
+        return self.filename
+
+    def assert_file_contains(self, expected, filename=None):
+        if filename is None:
+            filename = self.filename
+        self.assertIsNotNone(filename)
+        with open(filename) as out:
+            content = out.read()
+        self.assertEqual(content, expected)
+
+    def test_success(self):
+        filename = self._empty_file()
+        expected = 'spam spam spam spam spam'
+        script = self.SCRIPT.format(filename, expected)
+        interpreters.run_string(self.id, script)
+
+        self.assert_file_contains(expected)
+
+    def test_in_thread(self):
+        filename = self._empty_file()
+        expected = 'spam spam spam spam spam'
+        script = self.SCRIPT.format(filename, expected)
+        def f():
+            interpreters.run_string(self.id, script)
+
+        t = threading.Thread(target=f)
+        t.start()
+        t.join()
+
+        self.assert_file_contains(expected)
+
+    def test_create_thread(self):
+        filename = self._empty_file()
+        expected = 'spam spam spam spam spam'
+        script = dedent("""
+            import threading
+            def f():
+                with open('{}', 'w') as out:
+                    out.write('{}')
+
+            t = threading.Thread(target=f)
+            t.start()
+            t.join()
+            """).format(filename, expected)
+        interpreters.run_string(self.id, script)
+
+        self.assert_file_contains(expected)
+
+    @unittest.skipUnless(hasattr(os, 'fork'), "test needs os.fork()")
+    def test_fork(self):
+        filename = self._empty_file()
+        expected = 'spam spam spam spam spam'
+        script = dedent("""
+            import os
+            r, w = os.pipe()
+            pid = os.fork()
+            if pid == 0:  # child
+                import sys
+                filename = '{}'
+                with open(filename, 'w') as out:
+                    out.write('{}')
+                os.write(w, b'done!')
+
+                # Kill the unittest runner in the child process.
+                os._exit(1)
+            else:
+                import select
+                try:
+                    select.select([r], [], [])
+                finally:
+                    os.close(r)
+                    os.close(w)
+            """).format(filename, expected)
+        interpreters.run_string(self.id, script)
+        self.assert_file_contains(expected)
+
+    def test_already_running(self):
+        def f():
+            interpreters.run_string(self.id, wait_script)
+
+        t = threading.Thread(target=f)
+        dirname = tempfile.mkdtemp()
+        with _blocked(dirname) as wait_script:
+            t.start()
+            with self.assertRaises(RuntimeError):
+                interpreters.run_string(self.id, 'print("spam")')
+        t.join()
+
+    def test_does_not_exist(self):
+        id = 0
+        while id in interpreters.enumerate():
+            id += 1
+        with self.assertRaises(RuntimeError):
+            interpreters.run_string(id, 'print("spam")')
+
+    def test_error_id(self):
+        with self.assertRaises(RuntimeError):
+            interpreters.run_string(-1, 'print("spam")')
+
+    def test_bad_id(self):
+        with self.assertRaises(TypeError):
+            interpreters.run_string('spam', 'print("spam")')
+
+    def test_bad_code(self):
+        with self.assertRaises(TypeError):
+            interpreters.run_string(self.id, 10)
+
+    def test_bytes_for_code(self):
+        with self.assertRaises(TypeError):
+            interpreters.run_string(self.id, b'print("spam")')
+
+    def test_invalid_syntax(self):
+        with self.assertRaises(SyntaxError):
+            # missing close paren
+            interpreters.run_string(self.id, 'print("spam"')
+
+    def test_failure(self):
+        with self.assertRaises(Exception) as caught:
+            interpreters.run_string(self.id, 'raise Exception("spam")')
+        self.assertEqual(str(caught.exception), 'spam')
+
+    def test_sys_exit(self):
+        with self.assertRaises(SystemExit) as cm:
+            interpreters.run_string(self.id, dedent("""
+                import sys
+                sys.exit()
+                """))
+        self.assertIsNone(cm.exception.code)
+
+        with self.assertRaises(SystemExit) as cm:
+            interpreters.run_string(self.id, dedent("""
+                import sys
+                sys.exit(42)
+                """))
+        self.assertEqual(cm.exception.code, 42)
+
+    def test_SystemError(self):
+        with self.assertRaises(SystemExit) as cm:
+            interpreters.run_string(self.id, 'raise SystemExit(42)')
+        self.assertEqual(cm.exception.code, 42)
+
+
+class RunStringUnrestrictedTests(TestBase):
+
+    def setUp(self):
+        self.id = interpreters.create()
+
+    def test_without_ns(self):
+        script = dedent("""
+            spam = 42
+            """)
+        ns = interpreters.run_string_unrestricted(self.id, script)
+
+        self.assertEqual(ns['spam'], 42)
+
+    def test_with_ns(self):
+        updates = {'spam': 'ham', 'eggs': -1}
+        script = dedent("""
+            spam = 42
+            result = spam + eggs
+            """)
+        ns = interpreters.run_string_unrestricted(self.id, script, updates)
+
+        self.assertEqual(ns['spam'], 42)
+        self.assertEqual(ns['eggs'], -1)
+        self.assertEqual(ns['result'], 41)
+
+    def test_ns_does_not_overwrite(self):
+        updates = {'__name__': 'not __main__'}
+        script = dedent("""
+            spam = 42
+            """)
+        ns = interpreters.run_string_unrestricted(self.id, script, updates)
+
+        self.assertEqual(ns['__name__'], '__main__')
+
+    def test_main_not_shared(self):
+        ns1 = interpreters.run_string_unrestricted(self.id, 'spam = True')
+        ns2 = interpreters.run_string_unrestricted(self.id, 'eggs = False')
+
+        self.assertIn('spam', ns1)
+        self.assertNotIn('eggs', ns1)
+        self.assertIn('eggs', ns2)
+        self.assertNotIn('spam', ns2)
+
+    def test_return_execution_namespace(self):
+        script = dedent("""
+            spam = 42
+            """)
+        ns = interpreters.run_string_unrestricted(self.id, script)
+
+        ns.pop('__builtins__')
+        ns.pop('__loader__')
+        self.assertEqual(ns, {
+            '__name__': '__main__',
+            '__annotations__': {},
+            '__doc__': None,
+            '__package__': None,
+            '__spec__': None,
+            'spam': 42,
+            })
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/Lib/test/test_interpreters.py b/Lib/test/test_interpreters.py
new file mode 100644
index 00000000000000..3e020139e0bdae
--- /dev/null
+++ b/Lib/test/test_interpreters.py
@@ -0,0 +1,167 @@
+import ast
+import os.path
+import shutil
+import tempfile
+from textwrap import dedent
+import threading
+import unittest
+
+from test import support
+
+_interpreters = support.import_module('_interpreters')
+interpreters = support.import_module('interpreters')
+
+
+class TestBase(unittest.TestCase):
+
+    def setUp(self):
+        self.dirname = None
+
+    def tearDown(self):
+        if self.dirname is not None:
+            try:
+                shutil.rmtree(self.dirname)
+            except FileNotFoundError:
+                pass  # already deleted
+
+        for id in _interpreters.enumerate():
+            if id == 0:  # main
+                continue
+            try:
+                _interpreters.destroy(id)
+            except RuntimeError:
+                pass  # already destroyed
+
+    def _resolve_filename(self, name):
+        if self.dirname is None:
+            self.dirname = tempfile.mkdtemp()
+        return os.path.join(self.dirname, name)
+
+    def _empty_file(self, name):
+        filename = self._resolve_filename(name)
+        support.create_empty_file(filename)
+        return filename
+
+    def assert_file_contains(self, filename, expected):
+        with open(filename) as out:
+            content = out.read()
+        self.assertEqual(content, expected)
+
+
+class ModuleTests(TestBase):
+
+    def test_enumerate(self):
+        main_id = _interpreters.get_main()
+        got = interpreters.enumerate()
+        self.assertEqual(set(i.id for i in got), {main_id})
+
+        id1 = _interpreters.create()
+        id2 = _interpreters.create()
+        got = interpreters.enumerate()
+        self.assertEqual(set(i.id for i in got), {main_id, id1, id2})
+
+    def test_current(self):
+        main_id = _interpreters.get_main()
+        current = interpreters.current()
+        self.assertEqual(current.id, main_id)
+
+        id = _interpreters.create()
+        script = dedent("""
+            import interpreters
+            interp = interpreters.current()
+            """)
+        ns = _interpreters.run_string_unrestricted(id, script)
+        current = ns['interp']
+        self.assertEqual(current.id, id)
+
+    def test_main(self):
+        expected = _interpreters.get_main()
+        main = interpreters.main()
+
+        self.assertEqual(main.id, expected)
+
+    def test_create(self):
+        main_id = _interpreters.get_main()
+        interp = interpreters.create()
+
+        self.assertIsInstance(interp, interpreters.Interpreter)
+        self.assertGreater(interp.id, main_id)
+
+
+class InterpreterTests(TestBase):
+
+    def test_repr(self):
+        interp = interpreters.Interpreter(10)
+        result = repr(interp)
+
+        self.assertEqual(result, 'Interpreter(id=10)')
+
+    def test_equality(self):
+        interp1 = interpreters.Interpreter(0)
+        interp2 = interpreters.Interpreter(10)
+        interp3 = interpreters.Interpreter(10)
+        different = (interp1 == interp2)
+        same = (interp2 == interp3)
+        identity = (interp1 == interp1)
+
+        self.assertFalse(different)
+        self.assertTrue(same)
+        self.assertTrue(identity)
+
+    def test_id(self):
+        interp = interpreters.Interpreter(10)
+        id = interp.id
+
+        self.assertEqual(id, 10)
+
+    def test_is_running(self):
+        interp = interpreters.create()
+        is_running = interp.is_running()
+        self.assertFalse(is_running)
+
+        script = 'is_running = interp.is_running()'
+        ns = _interpreters.run_string_unrestricted(interp.id, script, {
+                'interp': interp,
+                })
+        is_running = ns['is_running']
+        self.assertTrue(is_running)
+
+    def test_destroy(self):
+        before = set(_interpreters.enumerate())
+        interp = interpreters.create()
+        self.assertEqual(set(_interpreters.enumerate()), before | {interp.id})
+
+        interp.destroy()
+        after = set(_interpreters.enumerate())
+
+        self.assertEqual(before, after)
+
+    def test_run_string(self):
+        filename = self._empty_file('spam')
+        data = 'success!'
+        script = dedent(f"""
+            with open('{filename}', 'w') as out:
+                out.write('{data}')
+            """)
+        interp = interpreters.create()
+
+        interp.run(script)
+        self.assert_file_contains(filename, data)
+
+    def test_run_unsupported(self):
+        script = 'raise Exception'
+        interp = interpreters.create()
+
+        with self.assertRaises(NotImplementedError):
+            interp.run(None)
+        with self.assertRaises(NotImplementedError):
+            interp.run(10)
+        node = compile(script, '<script>', 'exec', ast.PyCF_ONLY_AST)
+        with self.assertRaises(NotImplementedError):
+            interp.run(node)
+        code = compile(script, '<script>', 'exec')
+        with self.assertRaises(NotImplementedError):
+            interp.run(code)
+        f = lambda: None
+        with self.assertRaises(NotImplementedError):
+            interp.run(f)
diff --git a/Modules/_interpretersmodule.c b/Modules/_interpretersmodule.c
new file mode 100644
index 00000000000000..5eef04d5d755a5
--- /dev/null
+++ b/Modules/_interpretersmodule.c
@@ -0,0 +1,515 @@
+
+/* interpreters module */
+/* low-level access to interpreter primitives */
+
+#include "Python.h"
+#include "frameobject.h"
+
+
+static PyInterpreterState *
+_get_current(void)
+{
+    PyThreadState *tstate;
+
+    tstate = PyThreadState_Get();
+    if (tstate == NULL)
+        return NULL;
+    return tstate->interp;
+}
+
+static PyInterpreterState *
+_look_up_int64(PY_INT64_T requested_id)
+{
+    if (requested_id < 0)
+        goto error;
+
+    PyInterpreterState *interp = PyInterpreterState_Head();
+    while (interp != NULL) {
+        PY_INT64_T id = PyInterpreterState_GetID(interp);
+        if (id < 0)
+            return NULL;
+        if (requested_id == id)
+            return interp;
+        interp = PyInterpreterState_Next(interp);
+    }
+
+error:
+    PyErr_Format(PyExc_RuntimeError,
+                 "unrecognized interpreter ID %lld", requested_id);
+    return NULL;
+}
+
+static PyInterpreterState *
+_look_up(PyObject *requested_id)
+{
+    long long id = PyLong_AsLongLong(requested_id);
+    if (id == -1 && PyErr_Occurred() != NULL)
+        return NULL;
+    assert(id <= INT64_MAX);
+    return _look_up_int64(id);
+}
+
+static PyObject *
+_get_id(PyInterpreterState *interp)
+{
+    PY_INT64_T id = PyInterpreterState_GetID(interp);
+    if (id < 0)
+        return NULL;
+    return PyLong_FromLongLong(id);
+}
+
+static int
+_is_running(PyInterpreterState *interp)
+{
+    PyThreadState *tstate = PyInterpreterState_ThreadHead(interp);
+    if (PyThreadState_Next(tstate) != NULL) {
+        PyErr_SetString(PyExc_RuntimeError,
+                        "interpreter has more than one thread");
+        return -1;
+    }
+    PyFrameObject *frame = _PyThreadState_GetFrame(tstate);
+    if (frame == NULL) {
+        if (PyErr_Occurred() != NULL)
+            return -1;
+        return 0;
+    }
+    return (int)(frame->f_executing);
+}
+
+static int
+_ensure_not_running(PyInterpreterState *interp)
+{
+    int is_running = _is_running(interp);
+    if (is_running < 0)
+        return -1;
+    if (is_running) {
+        PyErr_Format(PyExc_RuntimeError, "interpreter already running");
+        return -1;
+    }
+    return 0;
+}
+
+static PyObject *
+_copy_module(PyObject *m, PyObject *updates)
+{
+    PyObject *orig = PyModule_GetDict(m);  // borrowed
+    if (orig == NULL) {
+        return NULL;
+    }
+
+    PyObject *copy = PyModule_New("__main__");
+    if (copy == NULL) {
+        return NULL;
+    }
+    PyObject *ns = PyModule_GetDict(copy);  // borrowed
+    if (ns == NULL)
+        goto error;
+
+    if (PyDict_Merge(ns, orig, 1) < 0)
+        goto error;
+    if (updates != NULL && PyDict_Merge(ns, updates, 0) < 0)
+        goto error;
+    return copy;
+
+error:
+    Py_DECREF(copy);
+    return NULL;
+}
+
+static int
+_run_string(PyInterpreterState *interp, const char *codestr, PyObject *ns)
+{
+    // Switch to interpreter.
+    PyThreadState *tstate = PyInterpreterState_ThreadHead(interp);
+    PyThreadState *save_tstate = PyThreadState_Swap(tstate);
+
+    // Run the string (see PyRun_SimpleStringFlags).
+    PyObject *exc = NULL, *value = NULL, *tb = NULL;
+    PyObject *result = PyRun_StringFlags(codestr, Py_file_input, ns, ns, NULL);
+    if (result == NULL) {
+        // Get the exception from the subinterpreter.
+        PyErr_Fetch(&exc, &value, &tb);
+    } else {
+        Py_DECREF(result);  // We throw away the result.
+    }
+
+    // Switch back.
+    if (save_tstate != NULL)
+        PyThreadState_Swap(save_tstate);
+
+    // Propagate any exception out to the caller.
+    PyErr_Restore(exc, value, tb);
+
+    return result == NULL ? -1 : 0;
+}
+
+static PyObject *
+_run_string_in_main(PyInterpreterState *interp, const char *codestr,
+                    PyObject *updates)
+{
+    if (_ensure_not_running(interp) < 0)
+        return NULL;
+
+    // Get the namespace in which to execute.  This involves creating
+    // a new module, updating it from the __main__ module and the given
+    // updates (if any), replacing the __main__ with the new module in
+    // sys.modules, and then using the new module's __dict__.  At the
+    // end we restore the original __main__ module.
+    PyObject *main_mod = PyMapping_GetItemString(interp->modules, "__main__");
+    if (main_mod == NULL)
+        return NULL;
+    PyObject *m = _copy_module(main_mod, updates);
+    if (m == NULL) {
+        Py_DECREF(main_mod);
+        return NULL;
+    }
+    if (PyMapping_SetItemString(interp->modules, "__main__", m) < 0) {
+        Py_DECREF(main_mod);
+        Py_DECREF(m);
+        return NULL;
+    }
+    PyObject *ns = PyModule_GetDict(m);  // borrowed
+    Py_INCREF(ns);
+    Py_DECREF(m);
+
+    // Run the string.
+    PyObject *result = ns;
+    if (_run_string(interp, codestr, ns) < 0) {
+        result = NULL;
+        Py_DECREF(ns);
+    }
+
+    // Restore __main__.
+    PyObject *exc = NULL, *value = NULL, *tb = NULL;
+    PyErr_Fetch(&exc, &value, &tb);
+    if (PyMapping_SetItemString(interp->modules, "__main__", main_mod) < 0) {
+        _PyErr_ChainExceptions(exc, value, tb);
+    } else {
+        PyErr_Restore(exc, value, tb);
+    }
+    Py_DECREF(main_mod);
+
+    return result;
+}
+
+
+/* module level code ********************************************************/
+
+static PyObject *
+interp_create(PyObject *self, PyObject *args)
+{
+    if (!PyArg_UnpackTuple(args, "create", 0, 0))
+        return NULL;
+
+    // Create and initialize the new interpreter.
+    PyThreadState *tstate, *save_tstate;
+    save_tstate = PyThreadState_Swap(NULL);
+    tstate = Py_NewInterpreter();
+    PyThreadState_Swap(save_tstate);
+    if (tstate == NULL) {
+        /* Since no new thread state was created, there is no exception to
+           propagate; raise a fresh one after swapping in the old thread
+           state. */
+        PyErr_SetString(PyExc_RuntimeError, "interpreter creation failed");
+        return NULL;
+    }
+    return _get_id(tstate->interp);
+}
+
+PyDoc_STRVAR(create_doc,
+"create() -> ID\n\
+\n\
+Create a new interpreter and return a unique generated ID.");
+
+
+static PyObject *
+interp_destroy(PyObject *self, PyObject *args)
+{
+    PyObject *id;
+    if (!PyArg_UnpackTuple(args, "destroy", 1, 1, &id))
+        return NULL;
+    if (!PyLong_Check(id)) {
+        PyErr_SetString(PyExc_TypeError, "ID must be an int");
+        return NULL;
+    }
+
+    // Look up the interpreter.
+    PyInterpreterState *interp = _look_up(id);
+    if (interp == NULL)
+        return NULL;
+
+    // Ensure we don't try to destroy the current interpreter.
+    PyInterpreterState *current = _get_current();
+    if (current == NULL)
+        return NULL;
+    if (interp == current) {
+        PyErr_SetString(PyExc_RuntimeError,
+                        "cannot destroy the current interpreter");
+        return NULL;
+    }
+
+    // Ensure the interpreter isn't running.
+    /* XXX We *could* support destroying a running interpreter but
+       aren't going to worry about it for now. */
+    if (_ensure_not_running(interp) < 0)
+        return NULL;
+
+    // Destroy the interpreter.
+    //PyInterpreterState_Delete(interp);
+    PyThreadState *tstate, *save_tstate;
+    tstate = PyInterpreterState_ThreadHead(interp);
+    save_tstate = PyThreadState_Swap(tstate);
+    Py_EndInterpreter(tstate);
+    PyThreadState_Swap(save_tstate);
+
+    Py_RETURN_NONE;
+}
+
+PyDoc_STRVAR(destroy_doc,
+"destroy(ID)\n\
+\n\
+Destroy the identified interpreter.\n\
+\n\
+Attempting to destroy the current interpreter results in a RuntimeError.\n\
+So does an unrecognized ID.");
+
+
+static PyObject *
+interp_enumerate(PyObject *self)
+{
+    PyObject *ids, *id;
+    PyInterpreterState *interp;
+
+    ids = PyList_New(0);
+    if (ids == NULL)
+        return NULL;
+
+    interp = PyInterpreterState_Head();
+    while (interp != NULL) {
+        id = _get_id(interp);
+        if (id == NULL)
+            return NULL;
+        // insert at front of list
+        if (PyList_Insert(ids, 0, id) < 0)
+            return NULL;
+
+        interp = PyInterpreterState_Next(interp);
+    }
+
+    return ids;
+}
+
+PyDoc_STRVAR(enumerate_doc,
+"enumerate() -> [ID]\n\
+\n\
+Return a list containing the ID of every existing interpreter.");
+
+
+static PyObject *
+interp_get_current(PyObject *self)
+{
+    PyInterpreterState *interp =_get_current();
+    if (interp == NULL)
+        return NULL;
+    return _get_id(interp);
+}
+
+PyDoc_STRVAR(get_current_doc,
+"get_current() -> ID\n\
+\n\
+Return the ID of current interpreter.");
+
+
+static PyObject *
+interp_get_main(PyObject *self)
+{
+    // Currently, 0 is always the main interpreter.
+    return PyLong_FromLongLong(0);
+}
+
+PyDoc_STRVAR(get_main_doc,
+"get_main() -> ID\n\
+\n\
+Return the ID of main interpreter.");
+
+
+static PyObject *
+interp_run_string(PyObject *self, PyObject *args)
+{
+    PyObject *id, *code;
+    if (!PyArg_UnpackTuple(args, "run_string", 2, 2, &id, &code))
+        return NULL;
+    if (!PyLong_Check(id)) {
+        PyErr_SetString(PyExc_TypeError, "first arg (ID) must be an int");
+        return NULL;
+    }
+    if (!PyUnicode_Check(code)) {
+        PyErr_SetString(PyExc_TypeError,
+                        "second arg (code) must be a string");
+        return NULL;
+    }
+
+    // Look up the interpreter.
+    PyInterpreterState *interp = _look_up(id);
+    if (interp == NULL)
+        return NULL;
+
+    // Extract code.
+    Py_ssize_t size;
+    const char *codestr = PyUnicode_AsUTF8AndSize(code, &size);
+    if (codestr == NULL)
+        return NULL;
+    if (strlen(codestr) != (size_t)size) {
+        PyErr_SetString(PyExc_ValueError,
+                        "source code string cannot contain null bytes");
+        return NULL;
+    }
+
+    // Run the code in the interpreter.
+    PyObject *ns = _run_string_in_main(interp, codestr, NULL);
+    if (ns == NULL)
+        return NULL;
+    else
+        Py_RETURN_NONE;
+}
+
+PyDoc_STRVAR(run_string_doc,
+"run_string(ID, sourcetext)\n\
+\n\
+Execute the provided string in the identified interpreter.\n\
+\n\
+See PyRun_SimpleStrings.");
+
+
+static PyObject *
+interp_run_string_unrestricted(PyObject *self, PyObject *args)
+{
+    PyObject *id, *code, *ns = NULL;
+    if (!PyArg_UnpackTuple(args, "run_string_unrestricted", 2, 3,
+                           &id, &code, &ns))
+        return NULL;
+    if (!PyLong_Check(id)) {
+        PyErr_SetString(PyExc_TypeError, "first arg (ID) must be an int");
+        return NULL;
+    }
+    if (!PyUnicode_Check(code)) {
+        PyErr_SetString(PyExc_TypeError,
+                        "second arg (code) must be a string");
+        return NULL;
+    }
+    if (ns == Py_None)
+        ns = NULL;
+
+    // Look up the interpreter.
+    PyInterpreterState *interp = _look_up(id);
+    if (interp == NULL)
+        return NULL;
+
+    // Extract code.
+    Py_ssize_t size;
+    const char *codestr = PyUnicode_AsUTF8AndSize(code, &size);
+    if (codestr == NULL)
+        return NULL;
+    if (strlen(codestr) != (size_t)size) {
+        PyErr_SetString(PyExc_ValueError,
+                        "source code string cannot contain null bytes");
+        return NULL;
+    }
+
+    // Run the code in the interpreter.
+    return _run_string_in_main(interp, codestr, ns);
+}
+
+PyDoc_STRVAR(run_string_unrestricted_doc,
+"run_string_unrestricted(ID, sourcetext, ns=None) -> main module ns\n\
+\n\
+Execute the provided string in the identified interpreter.  Return the\n\
+dict in which the code executed.  If the ns arg is provided then it is\n\
+merged into the execution namespace before the code is executed.\n\
+\n\
+See PyRun_SimpleStrings.");
+
+
+static PyObject *
+interp_is_running(PyObject *self, PyObject *args)
+{
+    PyObject *id;
+    if (!PyArg_UnpackTuple(args, "is_running", 1, 1, &id))
+        return NULL;
+    if (!PyLong_Check(id)) {
+        PyErr_SetString(PyExc_TypeError, "ID must be an int");
+        return NULL;
+    }
+
+    PyInterpreterState *interp = _look_up(id);
+    if (interp == NULL)
+        return NULL;
+    int is_running = _is_running(interp);
+    if (is_running < 0)
+        return NULL;
+    if (is_running)
+        Py_RETURN_TRUE;
+    Py_RETURN_FALSE;
+}
+
+PyDoc_STRVAR(is_running_doc,
+"is_running(id) -> bool\n\
+\n\
+Return whether or not the identified interpreter is running.");
+
+
+static PyMethodDef module_functions[] = {
+    {"create",                  (PyCFunction)interp_create,
+     METH_VARARGS, create_doc},
+    {"destroy",                 (PyCFunction)interp_destroy,
+     METH_VARARGS, destroy_doc},
+
+    {"enumerate",               (PyCFunction)interp_enumerate,
+     METH_NOARGS, enumerate_doc},
+    {"get_current",             (PyCFunction)interp_get_current,
+     METH_NOARGS, get_current_doc},
+    {"get_main",                (PyCFunction)interp_get_main,
+     METH_NOARGS, get_main_doc},
+    {"is_running",              (PyCFunction)interp_is_running,
+     METH_VARARGS, is_running_doc},
+
+    {"run_string",              (PyCFunction)interp_run_string,
+     METH_VARARGS, run_string_doc},
+    {"run_string_unrestricted", (PyCFunction)interp_run_string_unrestricted,
+     METH_VARARGS, run_string_unrestricted_doc},
+
+    {NULL,                      NULL}           /* sentinel */
+};
+
+
+/* initialization function */
+
+PyDoc_STRVAR(module_doc,
+"This module provides primitive operations to manage Python interpreters.\n\
+The 'interpreters' module provides a more convenient interface.");
+
+static struct PyModuleDef interpretersmodule = {
+    PyModuleDef_HEAD_INIT,
+    "_interpreters",
+    module_doc,
+    -1,
+    module_functions,
+    NULL,
+    NULL,
+    NULL,
+    NULL
+};
+
+
+PyMODINIT_FUNC
+PyInit__interpreters(void)
+{
+    PyObject *module;
+
+    module = PyModule_Create(&interpretersmodule);
+    if (module == NULL)
+        return NULL;
+
+
+    return module;
+}
diff --git a/setup.py b/setup.py
index 6a05643838bb4e..d95015721e008b 100644
--- a/setup.py
+++ b/setup.py
@@ -694,6 +694,9 @@ def detect_modules(self):
         # syslog daemon interface
         exts.append( Extension('syslog', ['syslogmodule.c']) )
 
+        # Python interface to subinterpreter C-API.
+        exts.append(Extension('_interpreters', ['_interpretersmodule.c']))
+
         #
         # Here ends the simple stuff.  From here on, modules need certain
         # libraries, are platform-specific, or present other surprises.