Python异步编程报错：asyncio.run()无法在运行中事件循环调用

问题重现 #

运行以下异步Python代码时触发RuntimeError：

import asyncio
import aiohttp

async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.text()
            return data

async def main():
    urls = [
        'https://example.com',
        'https://google.com',
        'https://openai.com'
    ]

    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

报错回溯：

---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
File c:\Users\Steven\stock.investing.decision.making\my-app\management\real.time.price2.py:21
     18     results = await asyncio.gather(*tasks)
     19     print(results)
---> 21 asyncio.run(main())

File c:\Python312\Lib\asyncio\runners.py:190, in run(main, debug, loop_factory)
    161 """Execute the coroutine and return the result.
    162 
    163 This function runs the passed coroutine, taking care of
   (...)
    186     asyncio.run(main())
    187 """
    188 if events._get_running_loop() is not None:
    189     # fail fast with short traceback
--> 190     raise RuntimeError(
    191         "asyncio.run() cannot be called from a running event loop")
    193 with Runner(debug=debug, loop_factory=loop_factory) as runner:
    194     return runner.run(main)

RuntimeError: asyncio.run() cannot be called from a running event loop

错误原因 #

asyncio.run()的设计是创建一个全新的事件循环，运行传入的协程后关闭循环。如果当前环境已经存在一个正在运行的事件循环（比如Jupyter笔记本、IPython交互式环境、Django/Flask异步框架等），调用asyncio.run()就会触发这个错误——同一进程中不能同时运行多个事件循环。

解决方案 #

1. 普通独立脚本环境 #

如果代码在普通Python脚本中运行，原代码本身逻辑正确。出现错误大概率是因为脚本被其他已经启动事件循环的代码调用，检查调用链，确保没有嵌套的事件循环启动逻辑即可。

2. 已有运行事件循环的环境（如Jupyter/IPython） #

直接使用await关键字调用协程，复用当前已有的事件循环，无需通过asyncio.run()创建新循环：

修改后的代码：

import asyncio
import aiohttp

async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.text()
            return data

async def main():
    urls = [
        'https://example.com',
        'https://google.com',
        'https://openai.com'
    ]

    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    print(results)

# 直接await协程，复用当前运行的事件循环
await main()

3. 手动管理循环的场景 #

如果需要在已有循环的环境中异步执行任务且不立即等待结果，可以获取当前循环并创建任务：

loop = asyncio.get_running_loop()
task = loop.create_task(main())
# 后续若需要获取结果，可使用 await task

内容的提问来源于stack exchange，提问作者Steven