Documentation

Backend pipeline with Django

Receive AI webhooks in Django views, process with Celery background tasks, and call back to ModelRiver: production-ready webhook handling.

Overview

Django's batteries-included approach makes it straightforward to handle ModelRiver backend pipeline webhooks. Use Django views for the webhook endpoint, Celery for background processing, and Django REST Framework for structured request handling.

What you'll build:

  • A webhook endpoint that receives AI-generated data from ModelRiver
  • HMAC signature verification decorator
  • Celery tasks for async processing and callback
  • Django ORM integration for data persistence

Quick start

Install dependencies

Bash
pip install django djangorestframework celery requests python-dotenv

Configuration

PYTHON
1# settings.py
2import os
3 
4MODELRIVER_API_KEY = os.environ.get("MODELRIVER_API_KEY", "")
5MODELRIVER_WEBHOOK_SECRET = os.environ.get("MODELRIVER_WEBHOOK_SECRET", "")

Webhook handler

View

PYTHON
1# webhooks/views.py
2import hmac
3import hashlib
4import json
5import os
6import requests
7from django.http import JsonResponse
8from django.views.decorators.csrf import csrf_exempt
9from django.views.decorators.http import require_POST
10from django.conf import settings
11 
12 
13def verify_webhook_signature(raw_body, timestamp, signature, secret):
14 """Verify x-modelriver-signature: HMAC-SHA256 of "<timestamp>.<raw JSON text of data>"."""
15 # Slice the exact "data" JSON text out of the raw body (never re-encode parsed JSON)
16 start = raw_body.find(b'"data":')
17 end = raw_body.rfind(b',"timestamp":')
18 if start == -1 or end == -1:
19 return False
20 signed = timestamp.encode("utf-8") + b"." + raw_body[start + len(b'"data":'):end]
21 expected = hmac.new(
22 secret.encode("utf-8"),
23 signed,
24 hashlib.sha256,
25 ).hexdigest()
26 return hmac.compare_digest(expected, signature)
27 
28 
29@csrf_exempt
30@require_POST
31def modelriver_webhook(request):
32 signature = request.headers.get("x-modelriver-signature", "")
33 timestamp = request.headers.get("x-modelriver-timestamp", "")
34 raw_body = request.body
35 
36 # 1. Verify webhook signature
37 if not verify_webhook_signature(raw_body, timestamp, signature, settings.MODELRIVER_WEBHOOK_SECRET):
38 return JsonResponse({"error": "Invalid signature"}, status=401)
39 
40 payload = json.loads(raw_body).get("data", {}) # the event payload is under "data"
41 event_type = payload.get("type")
42 callback_url = payload.get("callback_url")
43 
44 # 2. Handle backend pipeline workflow
45 if event_type == "task.ai_generated" and callback_url:
46 from .tasks import process_ai_webhook
47 
48 # Queue background task with Celery
49 process_ai_webhook.delay(
50 event=payload.get("event"),
51 ai_response=payload.get("ai_response"),
52 callback_url=callback_url,
53 customer_data=payload.get("customer_data", {}),
54 )
55 return JsonResponse({"received": True}, status=200)
56 
57 # Standard webhook
58 return JsonResponse({"received": True}, status=200)

URL configuration

PYTHON
1# webhooks/urls.py
2from django.urls import path
3from . import views
4 
5urlpatterns = [
6 path("webhooks/modelriver/", views.modelriver_webhook, name="modelriver_webhook"),
7]

Celery background task

PYTHON
1# webhooks/tasks.py
2import requests
3from celery import shared_task
4from django.conf import settings
5from datetime import datetime
6 
7 
8@shared_task(bind=True, max_retries=3, default_retry_delay=10)
9def process_ai_webhook(self, event, ai_response, callback_url, customer_data):
10 """Process AI response and call back to ModelRiver."""
11 try:
12 enriched_data = {**ai_response.get("data", {})}
13 
14 # 3. Your custom business logic based on the event
15 if event == "content_ready":
16 from myapp.models import Content
17 
18 content = Content.objects.create(
19 title=enriched_data.get("title", ""),
20 body=enriched_data.get("description", ""),
21 category=customer_data.get("category", "general"),
22 source="modelriver",
23 )
24 enriched_data["id"] = str(content.id)
25 enriched_data["slug"] = content.slug
26 enriched_data["saved_at"] = datetime.now().isoformat()
27 
28 if event == "entities_extracted":
29 from myapp.models import Entity
30 
31 for entity in enriched_data.get("entities", []):
32 Entity.objects.create(
33 name=entity["name"],
34 entity_type=entity["type"],
35 source_id=customer_data.get("document_id"),
36 )
37 enriched_data["entities_saved"] = len(enriched_data.get("entities", []))
38 
39 # 4. Call back to ModelRiver
40 response = requests.post(
41 callback_url,
42 json={
43 "data": enriched_data,
44 "task_id": f"django_{event}_{datetime.now().strftime('%Y%m%d%H%M%S')}",
45 "metadata": {
46 "processed_by": "django",
47 "processed_at": datetime.now().isoformat(),
48 },
49 },
50 headers={
51 "Authorization": f"Bearer {settings.MODELRIVER_API_KEY}",
52 "Content-Type": "application/json",
53 },
54 timeout=10,
55 )
56 response.raise_for_status()
57 print(f"✅ Callback sent for event: {event}")
58 
59 except requests.exceptions.RequestException as exc:
60 print(f"❌ Callback failed: {exc}")
61 # Send error callback
62 try:
63 requests.post(
64 callback_url,
65 json={
66 "data": {
67 "processing_status": "failed",
68 "processing_error": str(exc),
69 },
70 },
71 headers={
72 "Authorization": f"Bearer {settings.MODELRIVER_API_KEY}",
73 "Content-Type": "application/json",
74 },
75 timeout=10,
76 )
77 except Exception:
78 pass
79 raise self.retry(exc=exc)

Django REST Framework

For a more structured approach using DRF serializers:

PYTHON
1# webhooks/serializers.py
2from rest_framework import serializers
3 
4 
5class AIResponseSerializer(serializers.Serializer):
6 data = serializers.DictField()
7 
8 
9class WebhookPayloadSerializer(serializers.Serializer):
10 """Validates the event payload found under the envelope's "data" key."""
11 type = serializers.CharField(required=False)
12 event = serializers.CharField(required=False)
13 channel_id = serializers.CharField()
14 ai_response = AIResponseSerializer(required=False)
15 callback_url = serializers.URLField(required=False)
16 callback_required = serializers.BooleanField(required=False)
17 customer_data = serializers.DictField(required=False, default={})
18 
19 
20# webhooks/views.py
21from rest_framework.views import APIView
22from rest_framework.response import Response
23from rest_framework import status
24 
25 
26class ModelRiverWebhookView(APIView):
27 authentication_classes = [] # Webhook uses signature verification instead
28 permission_classes = []
29 
30 def post(self, request):
31 signature = request.headers.get("x-modelriver-signature", "")
32 timestamp = request.headers.get("x-modelriver-timestamp", "")
33 raw_body = request.body # read the raw bytes before request.data parses them
34 
35 if not verify_webhook_signature(raw_body, timestamp, signature, settings.MODELRIVER_WEBHOOK_SECRET):
36 return Response({"error": "Invalid signature"}, status=status.HTTP_401_UNAUTHORIZED)
37 
38 serializer = WebhookPayloadSerializer(data=request.data.get("data", {}))
39 serializer.is_valid(raise_exception=True)
40 
41 data = serializer.validated_data
42 if data.get("type") == "task.ai_generated" and data.get("callback_url"):
43 process_ai_webhook.delay(
44 event=data.get("event"),
45 ai_response=data.get("ai_response"),
46 callback_url=data["callback_url"],
47 customer_data=data.get("customer_data", {}),
48 )
49 
50 return Response({"received": True}, status=status.HTTP_200_OK)

Triggering async requests

PYTHON
1# ai/client.py
2import requests
3from django.conf import settings
4 
5 
6def trigger_async_ai(workflow: str, prompt: str, metadata: dict = None) -> dict:
7 """Trigger a backend pipeline request."""
8 response = requests.post(
9 "https://api.modelriver.com/v1/ai/async",
10 headers={
11 "Authorization": f"Bearer {settings.MODELRIVER_API_KEY}",
12 "Content-Type": "application/json",
13 },
14 json={
15 "model": workflow,
16 "messages": [{"role": "user", "content": prompt}],
17 "metadata": metadata or {},
18 },
19 timeout=10,
20 )
21 response.raise_for_status()
22 return response.json()

Best practices

  1. Use Celery for processing: Never block the webhook response with heavy logic.
  2. Set task retries: Use max_retries and default_retry_delay on Celery tasks.
  3. Verify signatures first: Always check x-modelriver-signature before processing.
  4. Use Django ORM transactions: Wrap database writes in transaction.atomic() for consistency.
  5. Monitor with Request Logs: Track backend pipeline flows in Observability.

Next steps