An experimental guest mode for asyncio (inspired by Trio)

asyncio_guest.py

#!/usr/bin/env python3

# SPDX-FileCopyrightText: 2026 Ilya Egorov <0x42005e1f@gmail.com>
# SPDX-License-Identifier: ISC

# requires-python = ">=3.13"
# dependencies = [
#     "greenlet>=1.0.0",
#     "outcome>=1.0.0",
#     "sniffio>=1.3.0",
#     "wrapt>=2.0.0",
# ]

import asyncio

from concurrent.futures import ThreadPoolExecutor
from contextvars import copy_context

import outcome

from greenlet import getcurrent as current_greenlet, greenlet
from sniffio import thread_local as current_async_library_tlocal
from wrapt import AutoObjectProxy


class _ThreadedSelector(AutoObjectProxy):
    __slots__ = "__executor"

    def select(self, /, timeout=None):
        if timeout is None or timeout:  # timeout is non-zero
            try:
                executor = self.__executor
            except AttributeError:
                self.__executor = executor = ThreadPoolExecutor(1)

            future = executor.submit(self.__wrapped__.select, timeout)

            if (parent := current_greenlet().parent) is not None:
                greenlet.switch(parent, future)

            try:
                return future.result()
            finally:
                del future  # break reference cycles

        result = self.__wrapped__.select(timeout)

        if (parent := current_greenlet().parent) is not None:
            greenlet.switch(parent, None)

        return result

    def close(self, /):
        try:
            self.__wrapped__.close()
        finally:
            try:
                executor = self.__executor
            except AttributeError:
                pass
            else:
                executor.shutdown()


def _patch_loop(loop, /, *, host_uses_signal_set_wakeup_fd=False):
    loop_selector = loop._selector = _ThreadedSelector(loop._selector)
    if getattr(loop, "_proactor", None) is loop_selector.__wrapped__:
        loop._proactor = loop_selector  # Windows

    return loop


def start_guest_run(
    async_fn,
    /,
    *args,  # to `async_fn()`
    done_callback,
    run_sync_soon_threadsafe,
    run_sync_soon_not_threadsafe=None,
    host_uses_signal_set_wakeup_fd=False,  # ignored
    **kwargs,  # to `asyncio.Runner`
):
    kwargs.setdefault("loop_factory", asyncio.EventLoop)

    if run_sync_soon_not_threadsafe is None:
        run_sync_soon_not_threadsafe = run_sync_soon_threadsafe

    async def wrapper(*args):
        return await async_fn(*args)

    guest_runner = asyncio.Runner(**kwargs)
    guest_loop = _patch_loop(guest_runner.get_loop())

    gr = greenlet(outcome.capture)
    gr.gr_context = copy_context()

    def guest_callback(future):
        run_sync_soon_threadsafe(host_callback)

    def host_callback():
        host_library = current_async_library_tlocal.name
        host_loop = asyncio._get_running_loop()
        try:
            asyncio.set_event_loop(None)
            asyncio._set_running_loop(None)
            current_async_library_tlocal.name = "asyncio"

            gr.parent = current_greenlet()
            if gr:  # running
                guest_loop._run_forever_setup()

                obj = gr.switch()
            else:  # not yet started
                obj = gr.switch(guest_runner.run, wrapper(*args))
            if gr:  # running
                guest_loop._run_forever_cleanup()
        finally:
            current_async_library_tlocal.name = host_library
            asyncio._set_running_loop(host_loop)
            asyncio.set_event_loop(host_loop)

        if gr:  # running
            if obj is None:  # timeout is zero
                run_sync_soon_not_threadsafe(host_callback)
            else:  # timeout is non-zero
                obj.add_done_callback(guest_callback)
        else:  # completed
            run_sync_soon_not_threadsafe(lambda: done_callback(obj))

    run_sync_soon_threadsafe(host_callback)

    return guest_loop

See fleming79/async-kernel#320 (comment).

---

A brief note regarding host_uses_signal_set_wakeup_fd(for the 46th revision). This parameter is only relevant if the host lives longer than the guest (or if there is another guest; or if the host uses signal.set_wakeup_fd() in some special way). Otherwise, it does not matter who exactly will be woken up by signals received by the process: waking up the guest will also wake up the host due to the subsequent run_sync_soon_threadsafe().

---

The 52nd revision is the latest version that uses the greenlet library. It relies on python/cpython#110773, and the trade-off is that it requires Python ≥3.13. You can use it if the latest revision is not suitable for you for some reason.

There is also oremanj/aioguest, which also uses the greenlet approach. But it does not localize the asyncio event loop execution context, does not support multiple asyncio event loops, and has other implementation nuances. Сonceptually, however, it adheres to the same solutions as Trio.