python asyncio run: How to Execute Coroutines

When you write async Python code, you need an event loop to drive your coroutines. The asyncio.run() function is the simplest and most reliable way to start that loop, execute a coroutine, and shut everything down cleanly. Introduced in Python 3.7, it replaces the older pattern of manually creating a loop and calling run_until_complete(). This article explains how python asyncio run works, what it does behind the scenes, and where it can cause problems if used incorrectly.

What asyncio.run() Actually Does

asyncio.run(coro) creates a new event loop, runs the given coroutine until it completes, and then closes the loop. It also cancels any remaining tasks and waits for them to finish before closing. This is a higher-level convenience function that hides the loop management details.

The function signature is simple:

import asyncio

async def main():
    print('Hello')
    await asyncio.sleep(1)
    print('World')

asyncio.run(main())

When you call asyncio.run(), it performs these steps internally:

Creates a new event loop using asyncio.new_event_loop().
Sets it as the current loop for the current thread.
Runs the coroutine until it completes.
Cancels any pending tasks and waits for their cancellation to finish.
Closes the loop and sets the current loop back to None.

This means you don't have to manually manage the loop lifecycle. For most applications, asyncio.run() is the correct entry point.

Using asyncio.run() in a Main Function

The most common pattern is to define an async def main() and call it with asyncio.run(). This works well for scripts, CLI tools, and small services. The main coroutine can spawn other tasks and await them.

import asyncio

async def fetch_data(url):
    # Simulate network I/O
    await asyncio.sleep(1)
    return f'Data from {url}'

async def main():
    tasks = [
        asyncio.create_task(fetch_data('https://api.example.com/1')),
        asyncio.create_task(fetch_data('https://api.example.com/2')),
    ]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

Because asyncio.run() always creates a fresh event loop, you can call it multiple times in the same program, though that is rarely needed. Each call is independent and cleans up after itself.

How asyncio.run() Handles the Event Loop

One important detail is that asyncio.run() always creates a new event loop, even if one already exists in the current thread. This is intentional: it prevents accidental reuse of a loop that may be in an unknown state. If you call asyncio.run() from within an already-running loop, you will get a RuntimeError. For example, this fails:

import asyncio

async def inner():
    pass

async def outer():
    asyncio.run(inner())  # RuntimeError: asyncio.run() cannot be called from a running event loop

asyncio.run(outer())

The error message is clear: asyncio.run() cannot be called from a running event loop. This is a common mistake when trying to run a coroutine from inside another coroutine. Instead, you should await the inner coroutine directly or use asyncio.create_task().

Error Handling and Resource Cleanup

When a coroutine raises an exception, asyncio.run() propagates that exception after cleaning up the loop. This means you can use normal try/except blocks around the call.

import asyncio

async def fail():
    raise ValueError('Something went wrong')

try:
    asyncio.run(fail())
except ValueError as e:
    print(f'Caught: {e}')

More importantly, asyncio.run() ensures that all pending tasks are cancelled before the loop closes. If your coroutine creates background tasks and returns without awaiting them, those tasks will be cancelled. This prevents resource leaks but can also surprise developers who expect tasks to continue running.

import asyncio

async def background():
    while True:
        await asyncio.sleep(1)
        print('tick')

async def main():
    asyncio.create_task(background())
    print('main done')

asyncio.run(main())
# Only prints 'main done', then the task is cancelled silently

If you need background tasks to complete, you must explicitly await them, for example with asyncio.gather() or by storing the task and awaiting it later.

When Not to Use asyncio.run()

asyncio.run() is not suitable for every situation. Avoid it when you need to manage an event loop that persists across multiple coroutines or when you are integrating with an existing loop, such as in a Jupyter notebook or a web framework like FastAPI. In those contexts, the loop is already running, and you should use asyncio.create_task() or await directly.

Another limitation is that asyncio.run() cannot be used in a thread that already has a running event loop. If you are writing multi-threaded code and each thread needs its own loop, you can create a loop in that thread manually with asyncio.new_event_loop() and loop.run_until_complete(), but that is an advanced pattern.

For long-running services that need to handle many connections, you typically don't call asyncio.run() yourself. Instead, a framework like aiohttp or FastAPI manages the loop for you. The framework's entry point will start the loop and then run your application code as coroutines.

Performance and Overhead of asyncio.run()

Creating a new event loop each time you call asyncio.run() has a small overhead. For a single script that runs once, this is negligible. However, if you are calling asyncio.run() in a loop, for example to process a batch of tasks sequentially, you might want to reuse a single loop instead to avoid the setup and teardown cost.

import asyncio

async def process(item):
    # do something
    return item * 2

# Inefficient: creates a new loop for each item
for i in range(10):
    result = asyncio.run(process(i))

# Better: create one loop and run multiple coroutines
async def main():
    results = [await process(i) for i in range(10)]
    return results

results = asyncio.run(main())

The second approach is more efficient because it avoids repeated loop creation. For most use cases, the overhead is minor, but it becomes relevant in tight loops or when the coroutine itself is very short.

Compatibility and Version Notes

asyncio.run() was added in Python 3.7. If you are supporting older Python versions, you need to use the older pattern:

loop = asyncio.new_event_loop()
try:
    loop.run_until_complete(main())
finally:
    loop.close()

This pattern is still valid in modern Python, but asyncio.run() is preferred because it also handles task cancellation and loop cleanup automatically. The behavior of asyncio.run() has remained stable since its introduction, so code written today will work in future Python versions as long as you avoid deprecated loop APIs.

One subtle point: asyncio.run() sets the event loop policy for the current thread. If you are using custom event loop policies or running in a multi-threaded environment, be aware that the loop is created according to the current policy. In most applications, the default policy is fine.

Practical Example: A Simple Async CLI Tool

To see asyncio.run() in action, here is a small command-line tool that fetches multiple URLs concurrently:

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

if __name__ == '__main__':
    urls = ['https://example.com', 'https://httpbin.org/get']
    results = asyncio.run(main(urls))
    for url, content in zip(urls, results):
        print(f'{url}: {len(content)} bytes')

This pattern is idiomatic for async scripts: define an async main(), set up resources inside it, and call asyncio.run() at the bottom. The asyncio.run() call ensures that the event loop is properly closed even if an exception occurs.

Avoiding Common Pitfalls

A frequent mistake is trying to use asyncio.run() inside a coroutine that is already being executed by a loop. As shown earlier, this raises a RuntimeError. Another pitfall is forgetting that asyncio.run() cancels pending tasks. If you create tasks and don't keep references to them, they may be garbage collected or cancelled prematurely. Always store references to tasks that you intend to await later.

Also, be careful when mixing asyncio.run() with libraries that expect a running loop, such as asyncio.Queue or asyncio.Lock. These primitives must be created inside the coroutine that runs within the loop, not outside. For example, creating a Lock at module level and then using it inside asyncio.run() will fail because the lock is not bound to the loop until it is used. The correct pattern is to create the lock inside the main() coroutine.

Final Consideration: Loop Lifecycle and Debugging

When debugging async code, it helps to enable asyncio's debug mode by setting the environment variable PYTHONASYNCIODEBUG=1 or by calling asyncio.run(main(), debug=True). The debug parameter, available since Python 3.7, enables slow-callback detection and other diagnostics. This can reveal issues like unawaited coroutines or tasks that are never completed.

asyncio.run(main(), debug=True)

In production, you might leave debug mode off for performance, but during development it is invaluable. Understanding how asyncio.run() manages the loop lifecycle helps you write more predictable async code and avoid the subtle bugs that arise from incorrect loop management.