Your library has an async function, fetch_user. Some callers are plain
scripts, and they want a regular function to call. The usual answer is a small
wrapper:
import asyncio
async def fetch_user(user_id: int) -> dict[str, int]:
await asyncio.sleep(0.1) # stands in for a network call
return {"id": user_id}
def fetch_user_sync(user_id: int) -> dict[str, int]:
return asyncio.run(fetch_user(user_id))
print(fetch_user_sync(42)) # {'id': 42}
This works in a script. It breaks as soon as async code calls it. This article
shows what asyncio.run does for you, and how to make the wrapper fail with a
clear message.
asyncio.run starts, runs and cleans up
asyncio.run(coro) does three jobs:
- It creates a new event loop.
- It runs
corountil it finishes, then returns its result or raises its exception. - It cleans up. It cancels any tasks that are still running, waits for them to stop, and closes the loop.
Every call gets a fresh loop, and asyncio.run closes that loop before it
returns. That is why you call it once, at the top of a program, and write
everything below it as async def functions.
Share one loop between calls with a Runner
Sometimes regular code needs to call async code several times, for example in
a test suite or a command-line tool. Separate asyncio.run calls run each
piece on a new loop. Some objects, such as an async database connection pool,
belong to the loop that opened them and fail on any other loop.
asyncio.Runner keeps one loop open across several calls:
import asyncio
async def which_loop() -> int:
return id(asyncio.get_running_loop())
with asyncio.Runner() as runner:
first = runner.run(which_loop())
second = runner.run(which_loop())
print(first == second) # True, both calls ran on the same loop
A Runner is an object that owns one event loop. runner.run(coro) works
like asyncio.run. It runs one coroutine and returns its result. The
difference is that the loop stays open between calls, so an object opened in
the first call still works in the second. When the with block ends, the
runner does the same cleanup as asyncio.run and closes the loop.
The loop runs only while a run call is in progress. A task you start in one
call pauses until the next call.
Why the wrapper breaks inside async code
Now an async web handler calls fetch_user_sync(42). The handler is already
running on a loop, and asyncio.run refuses to start a second loop in the same
thread:
RuntimeError: asyncio.run() cannot be called from a running event loop
RuntimeWarning: coroutine 'fetch_user' was never awaited
The warning comes from the wrapper. It created the fetch_user coroutine, and
then nothing ran it.
The rule is one running loop per thread. Inside async code, write
await fetch_user(42) instead, and keep the wrapper for plain scripts.
To give async callers a clear error, check before you start a loop.
asyncio.get_running_loop() returns the loop that is running in the current
thread, and raises RuntimeError if there is none:
def fetch_user_sync(user_id: int) -> dict[str, int]:
try:
asyncio.get_running_loop()
except RuntimeError:
return asyncio.run(fetch_user(user_id)) # no loop here, so start one
raise RuntimeError("fetch_user_sync() called from async code; await fetch_user()")
The check is per thread. A worker thread starts without a loop, so the wrapper works there even while the main thread runs one.
Common mistakes
asyncio.run(main)passes the function instead of a coroutine, and raises an error. Writeasyncio.run(main()). In the same way,await mainraisesTypeError.asyncio.get_event_loop()can return a loop that is set up but idle, so it cannot tell you whether you are inside async code. Useasyncio.get_running_loop()instead.- Put only the
get_running_loop()call inside thetry. Ifasyncio.run(...)is inside it too, aRuntimeErrorraised by your own coroutine looks like "a loop is running".
Check your understanding
An async web handler calls the version of fetch_user_sync with the check. What happens, and what should the handler do instead?
Show answer
Hide answer
The check finds the handler's running loop. The wrapper raises its own
RuntimeError before it creates a coroutine, so Python has nothing to warn
about. The handler is already inside async code, so it should call
await fetch_user(42) directly.
Going further
On Python 3.14, asyncio.run and Runner.run accept any awaitable. On 3.12
and 3.13 they need a coroutine.
asyncio.get_event_loop() behaves differently across versions, and on 3.14 it
raises RuntimeError when no loop is set. Python 3.14 also deprecates the
event loop policy system, which is due for removal in 3.16. To choose how
asyncio creates the loop, pass loop_factory to asyncio.run or asyncio.Runner.
See the asyncio runners documentation
and get_running_loop.