Краткий ответ

Нет, напрямую создавать задачу из другого потока нельзя считать безопасным: объекты asyncio обычно привязаны к потоку, в котором работает их event loop. Для передачи корутины в loop из другого потока используют asyncio.run_coroutine_threadsafe, а для постановки обычного callback — loop.call_soon_threadsafe.

Исторический контекст

asyncio строится вокруг однопоточного event loop, который последовательно выполняет готовые callbacks и корутины. Такой подход уменьшает стоимость переключения между большим числом операций ввода-вывода и позволяет не защищать каждую внутреннюю операцию event loop блокировками.

Обратная сторона — большинство объектов asyncio не являются потокобезопасными. Межпоточное взаимодействие вынесено в специальные API, которые безопасно будят и уведомляют нужный event loop.

Постановка проблемы

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

Неверный подход особенно опасен тем, что ошибка может проявляться только при нагрузке: редкое одновременное изменение очереди задач или состояния loop приводит к непредсказуемому поведению. Сам факт, что loop уже запущен, не делает его API универсально доступным из любого потока.

Подробное решение

asyncio.run_coroutine_threadsafe принимает корутину и конкретный event loop, безопасно передаёт её на выполнение в поток loop и возвращает объект concurrent.futures.Future. Результат этого future можно получить в исходном потоке, но блокирующее ожидание следует применять осторожно, чтобы не создать взаимную блокировку.

Для callback без корутины используют loop.call_soon_threadsafe. Этот метод добавляет callback в потокобезопасном режиме и будит event loop, если тот сейчас ожидает события.

import asyncio
import threading

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

def submit(loop):
    future = asyncio.run_coroutine_threadsafe(work(), loop)
    print(future.result(timeout=1))

async def main():
    thread = threading.Thread(target=submit, args=(asyncio.get_running_loop(),))
    thread.start()
    await asyncio.sleep(0.2)
    thread.join()

asyncio.run(main())

Вызов create_task предназначен для планирования корутины в текущем работающем loop и не заменяет межпоточный механизм передачи. Кроме того, вызов future.result() из потока event loop опасен: если корутина ждёт продолжения, которое может выполнить только этот же loop, loop будет заблокирован.

Нужно также различать потокобезопасность event loop и потокобезопасность пользовательских данных. Даже если задача правильно передана через run_coroutine_threadsafe, общее состояние между потоками всё равно может требовать threading.Lock, очереди или другого подходящего механизма синхронизации.

Ситуация из практики

Сервис обрабатывает сетевые запросы в event loop, а сторонняя библиотека запускает callback в собственном рабочем потоке. После завершения операции callback должен уведомить asyncio-код.

Вариант с прямым вызовом создания задачи прост, но некорректен: он нарушает потоковую модель asyncio. Вариант с общей переменной и периодическим опросом из event loop работает, однако добавляет задержку, лишнюю нагрузку и отдельную синхронизацию.

Выбранное решение — передавать результат через loop.call_soon_threadsafe, если достаточно callback, либо через asyncio.run_coroutine_threadsafe, если необходимо запустить корутину и получить её результат. Это явно обозначает границу между потоками, не требует ручного доступа к внутренней очереди loop и обеспечивает своевременное пробуждение event loop.

Что кандидаты часто упускают

Чем отличается asyncio.run_coroutine_threadsafe от asyncio.to_thread?

run_coroutine_threadsafe передаёт корутину из другого потока в уже работающий event loop. Она продолжает выполняться в потоке loop и использует кооперативное планирование asyncio.

to_thread, наоборот, переносит синхронную функцию из event loop в рабочий поток, чтобы блокирующая операция не остановила loop. Эти API решают противоположные задачи: первое направляет работу в loop, второе выводит синхронную работу из loop.
Можно ли вызвать loop.call_soon вместо loop.call_soon_threadsafe из другого потока?

Нельзя полагаться на это как на безопасное решение. call_soon рассчитан на вызов из потока самого event loop, тогда как call_soon_threadsafe синхронизирует постановку callback и пробуждение loop.

В отладочном режиме asyncio некоторые нарушения потоковой модели может обнаружить раньше, но это не превращает обычный call_soon в потокобезопасный API.
Что произойдёт с результатом run_coroutine_threadsafe, если корутина завершится исключением?

Исключение будет сохранено в возвращённом concurrent.futures.Future. При вызове его result() в другом потоке исключение будет повторно выброшено там; его можно также проверить через методы future или обработать с помощью тайм-аута.

Если сам event loop остановится до выполнения или завершения корутины, future может не получить ожидаемый результат. Поэтому жизненный цикл loop и обработку отмены нужно согласовывать с кодом, который отправляет работу из другого потока.