Как исправить RuntimeError: asyncio.run() в Jupyter Notebook

В Jupyter Notebook не вызывайте asyncio.run() непосредственно из ячейки. Ядро IPython уже выполняет цикл событий, поэтому второй цикл нельзя запустить в том же потоке. Замените вызов asyncio.run(main()) на await main().

import asyncio

async def main():
    await asyncio.sleep(1)
    return "Готово"

result = await main()
print(result)

Это основное исправление для обычного Jupyter Notebook, JupyterLab и IPython с поддержкой верхнеуровневого await.

Почему возникает ошибка

Типичный проблемный код выглядит так:

import asyncio

async def main():
    await asyncio.sleep(1)
    print("Готово")

asyncio.run(main())

При выполнении в блокноте он может завершиться исключением:

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

Функция asyncio.run() предназначена для запуска корутины из обычного синхронного кода. Она создаёт новый цикл событий, выполняет переданную корутину, завершает асинхронные генераторы и закрывает созданный цикл.

Согласно документации Python, asyncio.run() нельзя вызывать, когда в том же потоке уже работает другой цикл событий. В Jupyter таким циклом управляет ядро IPython. Он нужен для обработки сообщений между интерфейсом блокнота и выполняемым Python-процессом, поэтому останавливать или заменять его пользовательским вызовом обычно не требуется.

В ячейке Jupyter используйте await coroutine(). В обычном Python-скрипте используйте asyncio.run(coroutine()). Эти варианты предназначены для разных сред запуска.

Исправление для кода в ячейке

Шаг 1. Найдите вызов asyncio.run()

Например:

response = asyncio.run(load_data())

Шаг 2. Оставьте асинхронную функцию без изменений

import asyncio

async def load_data():
    await asyncio.sleep(0.5)
    return {"status": "ok"}

Шаг 3. Вызовите корутину через await

response = await load_data()
print(response)

Верхнеуровневый await поддерживается IPython: его можно использовать прямо в ячейке, не помещая вызов внутрь дополнительной функции.

Шаг 4. Проверьте результат

Проверяйте не отсутствие исключения само по себе, а ожидаемое значение или состояние:

response = await load_data()

assert response["status"] == "ok"
print(response)

Такой тест подтверждает, что корутина действительно завершилась и вернула ожидаемые данные.

Если функция main() уже написана

Переписывать её не нужно. Меняется только точка входа.

В Python-скрипте:

import asyncio

async def main():
    await asyncio.sleep(1)
    print("Завершено")

if __name__ == "__main__":
    asyncio.run(main())

В Jupyter Notebook:

await main()

Не переносите конструкцию if __name__ == "__main__" в ячейку ради запуска корутины. В блокноте она не решает конфликт циклов событий: вызов asyncio.run() останется вложенным в уже работающий цикл.

Несколько асинхронных операций

Если задачи независимы, их можно запустить совместно через asyncio.gather():

import asyncio

async def fetch_item(item_id):
    await asyncio.sleep(0.2)
    return {"id": item_id}

results = await asyncio.gather(
    fetch_item(1),
    fetch_item(2),
    fetch_item(3),
)

print(results)

Здесь asyncio.run() также не нужен: gather() возвращает объект, который можно ожидать через await.

Когда операции должны выполняться строго последовательно, используйте несколько вызовов await:

first = await fetch_item(1)
second = await fetch_item(2)

print(first, second)

Создание отдельной задачи

Для фонового выполнения внутри текущего цикла событий можно создать задачу:

task = asyncio.create_task(fetch_item(10))

# В этой же или следующей ячейке:
result = await task
print(result)

asyncio.create_task() требует работающего цикла событий. В Jupyter он обычно уже есть, поэтому вызов из асинхронного контекста ячейки допустим. Важно сохранить ссылку на задачу, если результат будет получен позже.

Для проверки состояния доступны стандартные методы:

print(task.done())

if task.done() and not task.cancelled():
    print(task.result())

Метод result() нельзя использовать как замену await до завершения задачи: для незавершённой задачи он вызывает исключение InvalidStateError.

Как отменить запущенную задачу

При повторном выполнении ячеек старые задачи могут продолжить работу. Сохранённую задачу можно отменить:

task.cancel()

try:
    await task
except asyncio.CancelledError:
    print("Задача отменена")

Отмена доставляется в корутину при ближайшей точке ожидания. Если корутина содержит длительный синхронный участок без await, она не сможет обработать отмену немедленно.

Когда ошибка возникает внутри сторонней библиотеки

Иногда в пользовательском коде нет asyncio.run(), но исключение появляется после вызова библиотечной функции:

result = library_function()

Это означает, что библиотека или её обёртка может запускать собственную корутину через asyncio.run(). Проверьте полный traceback: в нём должен быть указан файл и строка, где сделан конфликтующий вызов.

Предпочтительный порядок исправления:

Найдите асинхронный вариант API библиотеки, например метод с суффиксом async, отдельный асинхронный клиент или функцию, возвращающую корутину.
Вызовите этот API через await.
Если библиотека находится под вашим контролем, уберите asyncio.run() из внутренней функции и сделайте её асинхронной.
Если доступен только синхронный API, не предполагайте, что его безопасно вызывать в текущем цикле. Проверьте официальную документацию конкретной библиотеки.

Пример неправильной обёртки:

def load_sync():
    return asyncio.run(load_data())

В блокноте такую обёртку лучше заменить асинхронной:

async def load_async():
    return await load_data()

response = await load_async()

Универсальная функция для скрипта и Jupyter

Не стоит создавать функцию, которая пытается автоматически вызвать asyncio.run() или вернуть задачу в зависимости от наличия цикла событий. У такой функции меняется тип результата: в одном случае она возвращает готовое значение, в другом — Task. Это усложняет обработку ошибок и делает поведение неочевидным.

Надёжнее разделить асинхронную логику и точки входа:

import asyncio

async def application():
    await asyncio.sleep(0.1)
    return 42

def run_script():
    return asyncio.run(application())

В скрипте:

value = run_script()
print(value)

В Jupyter:

value = await application()
print(value)

Так асинхронная функция остаётся общей, а способ запуска явно соответствует среде.

Проверка активного цикла событий

Для диагностики внутри ячейки можно запросить текущий работающий цикл:

import asyncio

loop = asyncio.get_running_loop()
print(type(loop).__name__)
print(loop.is_running())

Если вызов выполнен в контексте активного цикла, get_running_loop() возвращает объект цикла. В синхронном коде без работающего цикла функция вызывает RuntimeError.

Не используйте эту проверку как основание для вызова loop.run_until_complete() в Jupyter. Метод run_until_complete() также пытается управлять циклом и не предназначен для повторного запуска уже работающего цикла.

Почему не следует вручную закрывать цикл Jupyter

Код следующего вида не является исправлением:

loop = asyncio.get_event_loop()
loop.close()

Цикл событий принадлежит среде выполнения блокнота. Его принудительное закрытие может нарушить дальнейшую работу ядра, асинхронных библиотек и уже созданных задач. Пользовательский код должен работать внутри предоставленного цикла через await, а не заменять его.

Что делать с предупреждением о неожиданной корутине

После неудачного вызова asyncio.run(main()) рядом с основным исключением может появиться предупреждение:

RuntimeWarning: coroutine 'main' was never awaited

Оно означает, что объект корутины был создан вызовом main(), но не был выполнен через await. Исправление то же:

await main()

Не подавляйте предупреждение через фильтры warnings. Оно указывает на реальную ошибку управления корутиной.

Минимальный диагностический пример

Выполните ячейки по порядку.

Первая ячейка:

import asyncio

async def check_asyncio():
    await asyncio.sleep(0.1)
    return "asyncio работает"

Вторая ячейка:

message = await check_asyncio()
assert message == "asyncio работает"
print(message)

Если этот пример выполняется, поддержка верхнеуровневого await работает, а исходную ошибку следует искать в конкретном вызове asyncio.run() или во внутреннем коде используемой библиотеки.

Ограничения решения

Решение относится к средам Jupyter/IPython, где доступен верхнеуровневый await.
Конкретная интеграция цикла событий может зависеть от ядра и выбранного async-бэкенда IPython.
Для сторонних библиотек нужно учитывать их официальный асинхронный API и правила управления ресурсами.
Замена asyncio.run() на await не исправляет блокирующий синхронный код внутри корутины.
Сетевые клиенты, файловые объекты и другие асинхронные ресурсы по-прежнему нужно корректно закрывать согласно документации соответствующей библиотеки.

Итоговый чек-лист

Найдите asyncio.run(...) в ячейке или traceback.
Вызовите корутину через await.
Для нескольких независимых корутин используйте await asyncio.gather(...).
Для отдельно управляемой операции используйте asyncio.create_task() и затем await task.
Не вызывайте run_until_complete() для уже работающего цикла.
Не закрывайте цикл событий, которым управляет Jupyter.
Разделяйте общую асинхронную функцию и точки входа для скрипта и блокнота.
Проверяйте результат через возвращаемое значение, assert или ожидаемое состояние.