Documentation

Backend pipeline with FastAPI

Receive AI webhooks with async FastAPI handlers, process with background tasks, and call back to ModelRiver: built-in Pydantic validation included.

Overview

FastAPI's async-first architecture and built-in Pydantic validation make it an excellent choice for handling ModelRiver backend pipeline webhooks. Process AI results in background tasks, validate payloads with type-safe models, and execute non-blocking callbacks.

What you'll build:

  • An async webhook endpoint with Pydantic payload validation
  • HMAC signature verification dependency
  • Background task processing with BackgroundTasks
  • Async HTTP callback to ModelRiver

Quick start

Install dependencies

Bash
pip install fastapi uvicorn httpx python-dotenv

Configuration

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 handler

PYTHON
1# main.py
2import hmac
3import hashlib
4from typing import Any, Optional
5 
6import httpx
7from fastapi import FastAPI, Request, BackgroundTasks, HTTPException, Depends
8from pydantic import BaseModel
9 
10from config import settings
11 
12app = FastAPI()
13 
14 
15# --- Pydantic models ---
16 
17class AIResponse(BaseModel):
18 data: dict[str, Any]
19 
20 
21class WebhookPayload(BaseModel):
22 """The event payload, delivered under the envelope's "data" key."""
23 type: Optional[str] = None
24 event: Optional[str] = None
25 channel_id: str
26 ai_response: Optional[AIResponse] = None
27 callback_url: Optional[str] = None
28 callback_required: Optional[bool] = None
29 customer_data: dict[str, Any] = {}
30 meta: dict[str, Any] = {}
31 
32 
33class WebhookEnvelope(BaseModel):
34 callback_url: Optional[str] = None
35 channel_id: str
36 data: WebhookPayload
37 timestamp: int
38 
39 
40class CallbackPayload(BaseModel):
41 data: dict[str, Any]
42 task_id: Optional[str] = None
43 metadata: dict[str, Any] = {}
44 
45 
46# --- Signature verification ---
47 
48async def verify_signature(request: Request):
49 """FastAPI dependency that verifies the webhook signature."""
50 signature = request.headers.get("x-modelriver-signature", "")
51 timestamp = request.headers.get("x-modelriver-timestamp", "")
52 raw_body = await request.body()
53 
54 # Signed string: "<timestamp>.<exact JSON text of the body's "data" value>",
55 # sliced from the raw body (never re-encode parsed JSON)
56 start = raw_body.find(b'"data":')
57 end = raw_body.rfind(b',"timestamp":')
58 if start == -1 or end == -1:
59 raise HTTPException(status_code=401, detail="Invalid signature")
60 
61 expected = hmac.new(
62 settings.modelriver_webhook_secret.encode("utf-8"),
63 timestamp.encode("utf-8") + b"." + raw_body[start + len(b'"data":'):end],
64 hashlib.sha256,
65 ).hexdigest()
66 
67 if not hmac.compare_digest(expected, signature):
68 raise HTTPException(status_code=401, detail="Invalid signature")
69 
70 return raw_body
71 
72 
73# --- Webhook endpoint ---
74 
75@app.post("/webhooks/modelriver")
76async def handle_webhook(
77 envelope: WebhookEnvelope,
78 background_tasks: BackgroundTasks,
79 _verified: bytes = Depends(verify_signature),
80):
81 payload = envelope.data # the event payload
82 # Handle backend pipeline workflow
83 if payload.type == "task.ai_generated" and payload.callback_url:
84 background_tasks.add_task(
85 process_and_callback,
86 event=payload.event,
87 ai_response=payload.ai_response,
88 callback_url=payload.callback_url,
89 customer_data=payload.customer_data,
90 )
91 return {"received": True}
92 
93 # Standard webhook
94 return {"received": True}
95 
96 
97# --- Background processing ---
98 
99async def process_and_callback(
100 event: str | None,
101 ai_response: AIResponse | None,
102 callback_url: str,
103 customer_data: dict[str, Any],
104):
105 """Process the AI response and call back to ModelRiver."""
106 async with httpx.AsyncClient() as client:
107 try:
108 enriched_data = {**(ai_response.data if ai_response else {})}
109 
110 # Your custom business logic
111 if event == "content_ready":
112 # Save to database, generate slugs, etc.
113 enriched_data["processed"] = True
114 enriched_data["saved_at"] = datetime.now().isoformat()
115 
116 if event == "entities_extracted":
117 # Validate and store entities
118 entities = enriched_data.get("entities", [])
119 enriched_data["entities_validated"] = len(entities)
120 
121 # Call back to ModelRiver
122 callback = CallbackPayload(
123 data=enriched_data,
124 task_id=f"fastapi_{event}_{int(datetime.now().timestamp())}",
125 metadata={"processed_by": "fastapi"},
126 )
127 
128 response = await client.post(
129 callback_url,
130 json=callback.model_dump(),
131 headers={
132 "Authorization": f"Bearer {settings.modelriver_api_key}",
133 "Content-Type": "application/json",
134 },
135 timeout=10.0,
136 )
137 response.raise_for_status()
138 print(f"✅ Callback sent for event: {event}")
139 
140 except httpx.HTTPError as exc:
141 print(f"❌ Callback failed: {exc}")
142 
143 # Send error callback
144 error = CallbackPayload(
145 data={
146 "processing_status": "failed",
147 "processing_error": str(exc),
148 },
149 )
150 try:
151 await client.post(
152 callback_url,
153 json=error.model_dump(),
154 headers={
155 "Authorization": f"Bearer {settings.modelriver_api_key}",
156 "Content-Type": "application/json",
157 },
158 timeout=10.0,
159 )
160 except Exception:
161 pass
162 
163 
164from datetime import datetime

Triggering async requests

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 """Trigger a backend pipeline request."""
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# Usage in an endpoint
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 proxy

If your frontend can't connect to ModelRiver's WebSocket directly, proxy through 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 # Get ws_token from query params
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 # Join the ModelRiver channel
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

Best practices

  1. Use httpx.AsyncClient: FastAPI is async-first; avoid blocking requests library.
  2. Validate with Pydantic: Type-safe webhook payloads catch malformed data early.
  3. Use BackgroundTasks: Return 200 immediately and process in the background.
  4. Inject dependencies: Use Depends(verify_signature) for clean, reusable verification.
  5. Handle timeouts: Set explicit timeout on all HTTP calls.

Next steps