如何从FastAPI处理函数异步触发Python函数且不等待返回？

方法1：使用FastAPI内置的BackgroundTasks（推荐） #

FastAPI原生提供了BackgroundTasks组件，专门用来处理这类后台任务，无需手动管理事件循环，是最安全可靠的方案。

示例代码：

from fastapi import FastAPI, BackgroundTasks, Body

async def func_to_be_called():
    # 这里写耗时的异步操作，比如调用外部API、文件IO等
    await some_async_io_operation()

@app.post("/fast-api-endpoint")
def fast_api_func(request: dict = Body(...), background_tasks: BackgroundTasks = BackgroundTasks()):
    # 将异步任务加入后台队列
    background_tasks.add_task(func_to_be_called)
    return "Ok"

补充：处理同步阻塞函数 #

如果func_to_be_called是同步阻塞类型（比如CPU密集型计算），用asyncio.to_thread包装后再添加到后台任务即可：

import asyncio

def sync_blocking_func():
    # 耗时的同步操作，比如大量数据计算
    pass

@app.post("/fast-api-endpoint")
def fast_api_func(request: dict = Body(...), background_tasks: BackgroundTasks = BackgroundTasks()):
    background_tasks.add_task(asyncio.to_thread, sync_blocking_func)
    return "Ok"

方法2：使用asyncio.create_task（需适配异步接口） #

如果不想用BackgroundTasks，可以直接用asyncio.create_task，但要求接口函数本身是异步的：

import asyncio
from fastapi import FastAPI, Body

async def func_to_be_called():
    await some_async_io_operation()

@app.post("/fast-api-endpoint")
async def fast_api_func(request: dict = Body(...)):
    # 创建后台任务，无需await，直接交由事件循环调度
    asyncio.create_task(func_to_be_called())
    return "Ok"

注意：这种方法只适用于异步函数，同步阻塞函数必须用asyncio.to_thread包装后再创建任务，否则会阻塞整个事件循环。

为什么之前的尝试会失败？ #

• 直接调用func_to_be_called()：仅创建了协程对象，没有被事件循环调度执行，任务根本没启动。
• 用await func_to_be_called()：会等待任务执行完成才返回，不符合“立即响应”的需求。
• 手动操作事件循环但逻辑错误：如果接口是同步函数，无法正确获取FastAPI的异步事件循环，强行调用会导致任务无法正常运行。

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