I implemented something like this not long ago: class Api(AsyncApi): def __init__(self, token): try: self._loop = async

I implemented something like this not long ago: class Api(AsyncApi): def __init__(self, token): try: self._loop = asyncio.get_event_loop() except RuntimeError: # when running in a thread, get_event_loop doesn't create another one self._loop = asyncio.new_event_loop() asyncio.set_event_loop(self._loop) self._session = aiohttp.ClientSession(loop=self._loop) super().__init__(token, self._loop, self._session) @lru_cache(maxsize=None) def __getattribute__(self, name): attr = super().__getattribute__(name) if name.startswith("_") or not asyncio.iscoroutinefunction(attr): return attr def call_sync(*args, **kwargs): coro = attr(*args, **kwargs) return self._loop.run_until_complete(coro) return call_sync It can run when there is no loop or even when a loop is already running.