症状
错误文本直接陈述它所执行的规则:asyncio.run() cannot be called from a running event loop。上下文不对称就是指纹:纯 python 脚本里正常(没有循环在跑),Jupyter 单元格或 FastAPI 处理器里失败(框架已在本线程持有循环)。在失败上下文里调用 asyncio.get_running_loop() 得到 running=True 的循环对象,证明宿主是环境而不是你的代码——守卫禁止的正是嵌套第二个循环。
诊断思路
先确认活跃循环存在;再看当前任务与循环对象,通常会暴露持有执行权的框架集成(内核、Web 框架、测试夹具);最后 grep 所有 asyncio.run( 与 run_until_complete 调用点,能从异步上下文到达的就是需要重构的重入点。库暴露同步包装、内部却调 asyncio.run,是隐蔽变体:问题代码在依赖里而不是你这里,需要换调用方式或换库版本。
根因
asyncio.run 会创建新事件循环、跑完协程再关闭循环。在另一个循环运行中调用它等于在同一线程嵌套两个循环,asyncio 以 RuntimeError 拒绝——这是正确性守卫而非故障。CPython 事件循环并非线程安全,也不是为嵌套设计的。
修复与验证
调用形状匹配所在上下文:已在异步上下文就直接 await 协程;必须从同步代码驱动异步代码,就在工作线程里给它一个私有循环(threading.Thread 内 asyncio.run);笔记本单元格用顶层 await 或把阻塞工作推到 executor 线程;asyncio.run 只留给真正的程序入口——CLI 或脚本的第一个同步帧。验证时从每个上下文(笔记本单元、框架处理器、CLI)调用此前失败的路径,断言无 RuntimeError 完成;并确认每线程循环数保持为一。nest_asyncio 只给笔记本当最后手段——它是猴子补丁,会与库的清理逻辑互相干扰。
代码示例
RuntimeError: asyncio.run() cannot be called from a
running event loop
import asyncio
asyncio.get_running_loop()
# <_UnixSelectorEventLoop running=True closed=False debug=False>
async def handler():
result = await fetch_all(urls) # no run() anywhere
import asyncio, threading
def run_async(coro):
result = {}
def target():
result["value"] = asyncio.run(coro)
t = threading.Thread(target=target)
t.start(); t.join()
return result["value"]