FastAPI Webhook监听器触发422 Unprocessable Entity错误求助

问题场景 #

使用FastAPI的APIRouter搭建Webhook监听器，期望接收JSON格式POST请求并打印数据到控制台，但持续触发422错误。

执行的curl请求 #

curl -X 'POST' \
'http://127.0.0.1:8010/webhook' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{"data": {"UK":"test"}}'

监听器核心代码 #

import hmac
from logging import Logger, info
from fastapi import APIRouter, FastAPI, Header, Request, Response
from pydantic import BaseModel

WEBHOOK_SECRET = '123-456-789'

class WebhookResponse(BaseModel):
    result: str

class WebhookData(BaseModel):
    body: dict

app = FastAPI()
rout = APIRouter()

@rout.post("/webhook", response_model=WebhookResponse, status_code=200)
async def webhook(
    webhook_input: WebhookData,
    request: Request,
    response: Response,
    content_length: int = Header(...),
    x_hook_signature: str = Header(None)
):
    # 签名校验等逻辑
    printLog(webhook_input)
    return {"result": "ok"}

app.include_router(rout)

错误信息 #

curl返回：

{
    "detail": [
        {
            "loc": [
                "body"
            ],
            "msg": "value is not a valid dict",
            "type": "type_error.dict"
        }
    ]
}

控制台日志：

INFO:     127.0.0.1:50192 - "POST /webhook/ HTTP/1.1" 422 Unprocessable Entity

错误原因 #

核心问题是Pydantic模型定义与请求体结构不匹配：

• 定义的WebhookData模型要求请求体根字段为body，类型是dict；
• 但curl发送的请求体根字段是data，没有body字段，导致FastAPI无法将请求体解析为WebhookData实例，触发类型校验错误。

解决方法 #

提供两种可行方案，根据实际需求选择：

方案一：修改Pydantic模型匹配请求结构 #

如果需要保留当前curl的请求格式，将WebhookData模型改为匹配请求的字段结构：

class WebhookData(BaseModel):
    data: dict  # 对应请求体中的data字段

或者如果需要接收任意JSON格式的请求体，直接省略模型，用dict接收：

# 去掉WebhookData定义，修改接口参数
async def webhook(
    webhook_input: dict,  # 直接接收整个请求体为dict
    request: Request,
    response: Response,
    content_length: int = Header(...),
    x_hook_signature: str = Header(None)
):

方案二：修改curl请求匹配模型结构 #

如果要保留当前WebhookData模型定义，调整curl的请求体，添加body根字段：

curl -X 'POST' \
'http://127.0.0.1:8010/webhook' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{"body": {"data": {"UK":"test"}}}'

额外优化点 #

代码中存在两处小问题建议修复：

1. Logger使用错误：直接调用Logger.error是错误的，应该先获取日志实例：

   import logging
   logger = logging.getLogger(__name__)
   # 替换原有Logger.error/info为logger.error/info
   

2. 冗余导入：from router.router1 import rout后续被rout = APIRouter()覆盖，可删除该导入语句。

内容的提问来源于stack exchange，提问作者joe_bill.dollar