FastAPI：后端管道 – ModelRiver Docs

概述

FastAPI 的异步优先架构和内置的 Pydantic 验证使其成为处理 ModelRiver 后端管道 Webhooks 的绝佳选择。您可以在后台任务中处理 AI 结果，使用类型安全的模型验证负载，并执行非阻塞的回调。

您将构建的内容：

一个具有 Pydantic 负载验证功能的异步 Webhook 端点
HMAC 签名验证依赖项 (Dependency)
使用 BackgroundTasks 进行后台任务处理
向 ModelRiver 发起异步 HTTP 回调

快速开始

安装依赖

Bash

pip install fastapi uvicorn httpx python-dotenv

配置

PYTHON

1# config.py
2from pydantic_settings import BaseSettings
3 
4 
5class Settings(BaseSettings):
6    modelriver_api_key: str
7    modelriver_webhook_secret: str
8 
9    class Config:
10        env_file = ".env"
11 
12 
13settings = Settings()

Webhook 处理程序

PYTHON

1# main.py
2import hmac
3import hashlib
4from typing import Any, Optional
5from datetime import datetime
6 
7import httpx
8from fastapi import FastAPI, Request, BackgroundTasks, HTTPException, Depends
9from pydantic import BaseModel
10 
11from config import settings
12 
13app = FastAPI()
14 
15 
16# --- Pydantic 模型 ---
17 
18class AIResponse(BaseModel):
19    data: dict[str, Any]
20 
21 
22class WebhookPayload(BaseModel):
23    """事件载荷，位于信封的 "data" 键之下。"""
24    type: Optional[str] = None
25    event: Optional[str] = None
26    channel_id: str
27    ai_response: Optional[AIResponse] = None
28    callback_url: Optional[str] = None
29    callback_required: Optional[bool] = None
30    customer_data: dict[str, Any] = {}
31    meta: dict[str, Any] = {}
32 
33 
34class WebhookEnvelope(BaseModel):
35    callback_url: Optional[str] = None
36    channel_id: str
37    data: WebhookPayload
38    timestamp: int
39 
40 
41class CallbackPayload(BaseModel):
42    data: dict[str, Any]
43    task_id: Optional[str] = None
44    metadata: dict[str, Any] = {}
45 
46 
47# --- 签名验证 ---
48 
49async def verify_signature(request: Request):
50    """FastAPI 依赖项，用于验证 Webhook 签名。"""
51    signature = request.headers.get("x-modelriver-signature", "")
52    timestamp = request.headers.get("x-modelriver-timestamp", "")
53    raw_body = await request.body()
54 
55    # 签名字符串："<timestamp>.<请求体中 "data" 值的原始 JSON 文本>"，
56    # 直接从原始请求体中截取（切勿对解析后的 JSON 重新编码）
57    start = raw_body.find(b'"data":')
58    end = raw_body.rfind(b',"timestamp":')
59    if start == -1 or end == -1:
60        raise HTTPException(status_code=401, detail="Invalid signature")
61 
62    expected = hmac.new(
63        settings.modelriver_webhook_secret.encode("utf-8"),
64        timestamp.encode("utf-8") + b"." + raw_body[start + len(b'"data":'):end],
65        hashlib.sha256,
66    ).hexdigest()
67 
68    if not hmac.compare_digest(expected, signature):
69        raise HTTPException(status_code=401, detail="Invalid signature")
70 
71    return raw_body
72 
73 
74# --- Webhook 端点 ---
75 
76@app.post("/webhooks/modelriver")
77async def handle_webhook(
78    envelope: WebhookEnvelope,
79    background_tasks: BackgroundTasks,
80    _verified: bytes = Depends(verify_signature),
81):
82    payload = envelope.data  # 事件载荷
83    # 处理后端管道的工作流
84    if payload.type == "task.ai_generated" and payload.callback_url:
85        background_tasks.add_task(
86            process_and_callback,
87            event=payload.event,
88            ai_response=payload.ai_response,
89            callback_url=payload.callback_url,
90            customer_data=payload.customer_data,
91        )
92        return {"received": True}
93 
94    # 标准 Webhook
95    return {"received": True}
96 
97 
98# --- 后台处理 ---
99 
100async def process_and_callback(
101    event: str | None,
102    ai_response: AIResponse | None,
103    callback_url: str,
104    customer_data: dict[str, Any],
105):
106    """处理 AI 响应并回调 ModelRiver。"""
107    async with httpx.AsyncClient() as client:
108        try:
109            enriched_data = {**(ai_response.data if ai_response else {})}
110 
111            # 您的自定义业务逻辑
112            if event == "content_ready":
113                # 保存到数据库、生成 Slug 等
114                enriched_data["processed"] = True
115                enriched_data["saved_at"] = datetime.now().isoformat()
116 
117            if event == "entities_extracted":
118                # 验证并存储实体
119                entities = enriched_data.get("entities", [])
120                enriched_data["entities_validated"] = len(entities)
121 
122            # 回调 ModelRiver
123            callback = CallbackPayload(
124                data=enriched_data,
125                task_id=f"fastapi_{event}_{int(datetime.now().timestamp())}",
126                metadata={"processed_by": "fastapi"},
127            )
128 
129            response = await client.post(
130                callback_url,
131                json=callback.model_dump(),
132                headers={
133                    "Authorization": f"Bearer {settings.modelriver_api_key}",
134                    "Content-Type": "application/json",
135                },
136                timeout=10.0,
137            )
138            response.raise_for_status()
139            print(f"✅ 已针对事件发送回调: {event}")
140 
141        except httpx.HTTPError as exc:
142            print(f"❌ 回调失败: {exc}")
143 
144            # 发送错误回调
145            error = CallbackPayload(
146                data={
147                    "processing_status": "failed",
148                    "processing_error": str(exc),
149                },
150            )
151            try:
152                await client.post(
153                    callback_url,
154                    json=error.model_dump(),
155                    headers={
156                        "Authorization": f"Bearer {settings.modelriver_api_key}",
157                        "Content-Type": "application/json",
158                    },
159                    timeout=10.0,
160                )
161            except Exception:
162                pass

触发异步请求

PYTHON

1# ai/client.py
2import httpx
3from config import settings
4 
5 
6async def trigger_async_ai(workflow: str, prompt: str, metadata: dict = None) -> dict:
7    """触发后端管道的 AI 请求。"""
8    async with httpx.AsyncClient() as client:
9        response = await client.post(
10            "https://api.modelriver.com/v1/ai/async",
11            headers={
12                "Authorization": f"Bearer {settings.modelriver_api_key}",
13                "Content-Type": "application/json",
14            },
15            json={
16                "model": workflow,
17                "messages": [{"role": "user", "content": prompt}],
18                "metadata": metadata or {},
19            },
20            timeout=10.0,
21        )
22        response.raise_for_status()
23        return response.json()
24 
25 
26# 在端点中使用
27@app.post("/ai/generate")
28async def generate_content(prompt: str):
29    result = await trigger_async_ai(
30        workflow="content_generator",
31        prompt=prompt,
32        metadata={"source": "api"},
33    )
34    return {
35        "channel_id": result["channel_id"],
36        "ws_token": result["ws_token"],
37        "websocket_channel": result["websocket_channel"],
38    }

WebSocket 代理

如果您的前端无法直接连接到 ModelRiver 的 WebSocket，可以通过 FastAPI 进行代理：

PYTHON

1from fastapi import WebSocket, WebSocketDisconnect
2import asyncio
3import websockets
4import json
5 
6 
7@app.websocket("/ws/ai/{channel_id}")
8async def ai_websocket_proxy(websocket: WebSocket, channel_id: str):
9    await websocket.accept()
10 
11    # 从查询参数获取 ws_token
12    ws_token = websocket.query_params.get("token", "")
13 
14    async with websockets.connect(
15        f"wss://api.modelriver.com/socket/websocket?token={ws_token}"
16    ) as mr_ws:
17        # 加入 ModelRiver 频道
18        await mr_ws.send(json.dumps({
19            "topic": f"ai_response:{channel_id}",
20            "event": "phx_join",
21            "payload": {},
22            "ref": "1",
23        }))
24 
25        try:
26            async for message in mr_ws:
27                data = json.loads(message)
28                if data.get("event") == "response":
29                    await websocket.send_json(data["payload"])
30                    break
31        except WebSocketDisconnect:
32            pass

最佳实践

使用 httpx.AsyncClient：FastAPI 是异步优先的；请避免使用会产生阻塞的 requests 库。
使用 Pydantic 进行验证：类型安全的 Webhook 负载可以及早捕获格式错误的数据。
使用 BackgroundTasks：立即返回 200 并在后台进行处理。
注入依赖项：使用 Depends(verify_signature) 以实现整洁、可复用的验证逻辑。
处理超时：在所有 HTTP 调用上设置显式的 timeout。

下一步

Laravel 后端管道指南：PHP 备选方案
FastAPI 集成：标准 ModelRiver + FastAPI 的用法
Webhooks 参考：重试策略和投递监控

FastAPI 的后端管道

概述

快速开始

安装依赖

配置

Webhook 处理程序

触发异步请求

WebSocket 代理

最佳实践

下一步