@@ -538,7 +538,7 @@ struct _py_func_state {
    If you add a new static type to the standard library, you may have to
    update one of these numbers.
    */
-#define _Py_NUM_MANAGED_PREINITIALIZED_TYPES 122
+#define _Py_NUM_MANAGED_PREINITIALIZED_TYPES 120
 #define _Py_MAX_MANAGED_STATIC_BUILTIN_TYPES \
     (_Py_NUM_MANAGED_PREINITIALIZED_TYPES + 83)
 #define _Py_MAX_MANAGED_STATIC_EXT_TYPES 10
 
@@ -8,20 +8,12 @@ extern "C" {
 #  error "this header requires Py_BUILD_CORE define"
 #endif
 
-extern PyTypeObject _PyACallIter_Type;
-extern PyTypeObject _PyACallIterAwaitable_Type;
-
 // Like PyCallIter_New(), but the iteration also stops when *callable* raises
 // an exception matching *stop_exc* (an exception class or a tuple of exception
 // classes).  *sentinel* can be NULL; NULL *stop_exc* means StopIteration.
 extern PyObject *_PyCallIter_NewEx(PyObject *callable, PyObject *sentinel,
                                    PyObject *stop_exc);
 
-// The asynchronous counterpart of _PyCallIter_NewEx(): the result of
-// *callable* is awaited, and NULL *stop_exc* means StopAsyncIteration.
-extern PyObject *_PyACallIter_New(PyObject *callable, PyObject *sentinel,
-                                  PyObject *stop_exc);
-
 #ifdef __cplusplus
 }
 #endif
 
@@ -5,11 +5,96 @@
 copied into the ``builtins`` module.
 """
 
-__all__ = ['anext']
+__all__ = ['aiter', 'anext']
 
 _NOT_GIVEN = sentinel("_NOT_GIVEN")
 
 
+def aiter(object, /, stop_value=_NOT_GIVEN, *, stop_exception=_NOT_GIVEN):
+    """Return an AsyncIterator for an AsyncIterable object.
+
+    In the second form, the callable is called and its result is awaited
+    until it returns the stop value or raises the specified exception.
+    """
+    if stop_value is _NOT_GIVEN and stop_exception is _NOT_GIVEN:
+        cls = type(object)
+        try:
+            # Looked up on the type, like the C slot am_aiter.
+            aiter_method = cls.__aiter__
+        except AttributeError:
+            raise TypeError(
+                f"'{cls.__name__}' object is not an async iterable"
+            ) from None
+        iterator = aiter_method(object)
+        if not hasattr(type(iterator), '__anext__'):
+            raise TypeError(
+                f"{cls.__name__}.__aiter__() must return an async iterator, "
+                f"not {type(iterator).__name__}"
+            )
+        return iterator
+
+    if not callable(object):
+        raise TypeError("aiter(): the first argument must be callable")
+    if stop_exception is _NOT_GIVEN:
+        stop_exception = StopAsyncIteration
+    else:
+        # Like _PyEval_CheckExceptTypeValid(): an exception class or a tuple
+        # of exception classes.
+        if isinstance(stop_exception, tuple):
+            items = stop_exception
+        else:
+            items = (stop_exception,)
+        for item in items:
+            if not (isinstance(item, type) and issubclass(item, BaseException)):
+                raise TypeError(
+                    "catching classes that do not inherit from BaseException "
+                    "is not allowed"
+                )
+    return async_callable_iterator(object, stop_value, stop_exception)
+
+
+class async_callable_iterator:
+    """The asynchronous counterpart of callable_iterator.
+
+    The callable is called and its result is awaited for every
+    __anext__().
+    """
+
+    __slots__ = ('_callable', '_stop_value', '_stop_exception')
+    __module__ = 'builtins'
+
+    def __init__(self, callable, stop_value, stop_exception):
+        # _callable is set to None when the iterator is exhausted.
+        self._callable = callable
+        self._stop_value = stop_value
+        self._stop_exception = stop_exception
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        callable = self._callable
+        if callable is None:
+            raise StopAsyncIteration
+        try:
+            value = await callable()
+        except self._stop_exception:
+            self._exhaust()
+            raise StopAsyncIteration from None
+        except (StopIteration, StopAsyncIteration) as exc:
+            raise RuntimeError(
+                f"callable raised {type(exc).__name__}") from exc
+        stop_value = self._stop_value
+        if stop_value is not _NOT_GIVEN and stop_value == value:
+            self._exhaust()
+            raise StopAsyncIteration
+        return value
+
+    def _exhaust(self):
+        self._callable = None
+        self._stop_value = _NOT_GIVEN
+
+
 def anext(async_iterator, default=_NOT_GIVEN, /):
     """Return the next item from the async iterator.
 
 
@@ -148,6 +148,45 @@ async def main():
             'async generator CallStackTestBase.test_stack_async_gen.<locals>.gen()',
             stack_for_gen_nested_call[1])
 
+    async def test_stack_anext(self):
+        # anext() without a default returns the __anext__() coroutine
+        # itself, so nothing is inserted between it and main().
+
+        loop = asyncio.get_running_loop()
+        blocker = loop.create_future()
+
+        async def inner():
+            await blocker
+
+        class AIter:
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                await inner()
+                return 1
+
+        async def main():
+            await anext(AIter())
+
+        task = asyncio.create_task(main(), name='anext task')
+        await asyncio.sleep(0)
+        try:
+            stack = capture_test_stack(fut=task)
+        finally:
+            blocker.set_result(None)
+            await task
+
+        self.assertEqual(stack[0], [
+            'T<anext task>',
+            [
+                'a inner',
+                'a __anext__',
+                'a main',
+            ],
+            []
+        ])
+
     async def test_stack_anext_default(self):
         # anext() with a default wraps the awaitable in a coroutine, so the
         # call graph of a suspended task sees through it into __anext__().
@@ -188,6 +227,84 @@ async def main():
             []
         ])
 
+    async def test_stack_aiter(self):
+        # The one-argument form of aiter() returns the object's own
+        # iterator, so it adds no frame of its own to the call graph.
+
+        loop = asyncio.get_running_loop()
+        blocker = loop.create_future()
+
+        async def inner():
+            await blocker
+
+        class AIter:
+            def __aiter__(self):
+                return self
+
+            async def __anext__(self):
+                await inner()
+                return 1
+
+        async def main():
+            async for _ in aiter(AIter()):
+                break
+
+        task = asyncio.create_task(main(), name='aiter task')
+        await asyncio.sleep(0)
+        try:
+            stack = capture_test_stack(fut=task)
+        finally:
+            blocker.set_result(None)
+            await task
+
+        self.assertEqual(stack[0], [
+            'T<aiter task>',
+            [
+                'a inner',
+                'a __anext__',
+                'a main',
+            ],
+            []
+        ])
+
+    async def test_stack_aiter_callable(self):
+        # aiter(callable, ...) returns an iterator whose __anext__() is a
+        # coroutine, so the call graph of a suspended task sees through it
+        # into the awaited result of the callable.
+
+        loop = asyncio.get_running_loop()
+        blocker = loop.create_future()
+
+        async def inner():
+            await blocker
+
+        async def spam():
+            await inner()
+            return 1
+
+        async def main():
+            async for _ in aiter(spam, 1):
+                pass
+
+        task = asyncio.create_task(main(), name='aiter task')
+        await asyncio.sleep(0)
+        try:
+            stack = capture_test_stack(fut=task)
+        finally:
+            blocker.set_result(None)
+            await task
+
+        self.assertEqual(stack[0], [
+            'T<aiter task>',
+            [
+                'a inner',
+                'a spam',
+                'a __anext__',
+                'a main',
+            ],
+            []
+        ])
+
     def test_ag_frame_used_for_async_generator(self):
         # Regression test for gh-148736: the ag_await branch of
         # _build_graph_for_future must read ag_frame, not cr_frame.
 
@@ -6229,7 +6229,7 @@ def test_builtins_have_signatures(self):
         needs_groups = {"range", "slice", "getattr", "next"}
         no_signature |= needs_groups
         # These have unrepresentable parameter default values of NULL
-        unsupported_signature = {"aiter", "iter", "dir", "vars"}
+        unsupported_signature = {"iter", "dir", "vars"}
         # These need *args support in Argument Clinic
         needs_varargs = {"min", "max", "__build_class__"}
         no_signature |= needs_varargs
 
@@ -0,0 +1,5 @@
+Implement :func:`aiter` in Python instead of C, in the frozen ``_pybuiltins``
+module. The async iterator returned by ``aiter(callable, ...)`` now has a
+plain coroutine ``__anext__()``, so introspection tools such as
+:func:`asyncio.print_call_graph` can see through it into the awaited
+callable result.