如何将SocketIO AsyncServer实例作为依赖传入FastAPI路由？

错误原因 #

你直接将socketio.AsyncServer类传入Depends()，FastAPI会尝试自动实例化这个类，但AsyncServer初始化需要必填参数（如async_mode），框架无法找到所需参数，因此抛出"Field required"错误。

解决方案 #

核心思路是：创建一个依赖函数，返回已初始化完成的sio实例，而非让FastAPI尝试实例化AsyncServer类。同时调整路由和路由构建逻辑，使用这个依赖函数。

步骤1：抽离SocketIO实例到单独模块（避免循环导入） #

创建src/core/socketio.py文件，统一管理SocketIO实例和依赖函数：

import socketio

# 初始化AsyncServer实例
sio = socketio.AsyncServer(cors_allowed_origins="*", async_mode="asgi")

# 定义依赖函数，返回已创建的sio实例
def get_sio():
    return sio

步骤2：修改main.py，使用抽离的SocketIO实例 #

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from src.api.rest.api_v1.route_builder import build_routes
from src.core.socketio import sio, get_sio

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],  # 补充允许的headers，避免跨域问题
)

app = build_routes(app)
sio_asgi_app = socketio.ASGIApp(sio, app)

@app.get("/")
async def root():
    return {"message": "Hello from main server !"}

def start_server():
    api_host = "0.0.0.0"  # 修正你原代码中变量名写反的问题
    api_port = 8000
    uvicorn.run(
        "src.api.main:sio_asgi_app",
        host=api_host,
        port=api_port,
        log_level="info",
        reload=True,
    )

if __name__ == "__main__":
    start_server()

步骤3：修改product_routes.py，使用正确的依赖函数 #

from typing import Annotated
import socketio
from fastapi import APIRouter, Depends
from src.core.socketio import get_sio

router = APIRouter(prefix="/products")

@router.get("/emit")
async def emit_message(
    sio: Annotated[socketio.AsyncServer, Depends(get_sio)]
) -> None:
    await sio.emit("reply", {"message": "Hello from server !"})

步骤4：修正build_routes.py的路由依赖 #

from fastapi import FastAPI
from src.api.rest.api_v1.routes.product_routes import router as r_products
from src.core.socketio import get_sio

def build_routes(app: FastAPI) -> FastAPI:
    app.include_router(
        r_products,
        prefix="/v1",
        dependencies=[Depends(get_sio)],  # 使用依赖函数而非类
    )
    return app

验证 #

启动服务后，访问/v1/products/emit，FastAPI会正确注入已初始化的sio实例，调用emit方法发送消息。

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