SSE 路径带上key，messages/接口如何获取到这个key，进行message消息里的auth认证

[ipfred]

http://127.0.0.1:8020/sse?key=123456

我想生成这样一个sse server
message 接口里能获取到key，进行鉴权，如何实现

[lvyhhyvl]

@app.middleware("http")
async def key_auth_middleware(request: Request, call_next):
    """基于URL参数key的鉴权中间件"""
    global current_key

    # 只对/sse路径的请求进行鉴权
    if request.url.path == "/sse":
        # 从URL参数获取key
        key = request.query_params.get("key")
        if not key:
            # 既记录日志又返回错误
            print("缺少key参数")  # 实际项目应该用logging
            return JSONResponse(
                status_code=401,
                content={"detail": "缺少key参数"}
            )

        # 保存key到全局变量
        current_key = key

    # 继续处理请求
    return await call_next(request)
def main():
    """服务主入口"""
    print("Server starting...")
    # 创建SSE应用并添加中间件
    sse_app = mcp.sse_app()
    app.middleware("http")(key_auth_middleware)
    # 挂载处理后的SSE应用
    app.mount("/", sse_app)
    uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
    main()  # 保留直接运行能力（如 `python -m server.main`）

可以参考下这个实现，可以获取到key，进行鉴权。

[Kludex]

Using a middleware as pointed above is an option.