How to use Replicate webhooks
PREREQUISITES
Python 3.8+Replicate account with API accesspip install flask requests
Setup
To use Replicate webhooks, you need a Replicate account and a publicly accessible HTTPS endpoint to receive webhook POST requests. You can create this endpoint using frameworks like Flask or FastAPI. Install necessary packages with pip install flask requests. Configure your webhook URL in the Replicate dashboard or via API.
pip install flask requests Collecting flask Collecting requests Successfully installed flask requests-2.31.0
Step by step
This example shows a simple Flask server that listens for Replicate webhook POST requests, parses the JSON payload, and prints the prediction result.
import os
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/replicate-webhook', methods=['POST'])
def replicate_webhook():
data = request.json
# Example payload fields: id, status, output
prediction_id = data.get('id')
status = data.get('status')
output = data.get('output')
print(f"Received webhook for prediction {prediction_id} with status {status}")
print(f"Output: {output}")
# TODO: Add your processing logic here
return jsonify({'received': True}), 200
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000) * Running on http://0.0.0.0:5000/ (Press CTRL+C to quit) Received webhook for prediction 123abc with status succeeded Output: ['Result data here']
Common variations
You can secure your webhook endpoint by verifying the X-Replicate-Signature header using your Replicate webhook secret. Also, you can use asynchronous frameworks like FastAPI for better performance. Webhooks can notify you about different events such as prediction start, success, or failure.
from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
import os
app = FastAPI()
WEBHOOK_SECRET = os.environ.get('REPLICATE_WEBHOOK_SECRET')
@app.post('/replicate-webhook')
async def replicate_webhook(request: Request):
body = await request.body()
signature = request.headers.get('X-Replicate-Signature')
if not signature or not WEBHOOK_SECRET:
raise HTTPException(status_code=400, detail='Missing signature or secret')
expected_signature = hmac.new(
WEBHOOK_SECRET.encode(), body, hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected_signature):
raise HTTPException(status_code=403, detail='Invalid signature')
data = await request.json()
# Process webhook payload
return {'received': True} INFO: Started server process [12345] INFO: Waiting for application startup. INFO: Application startup complete. INFO: 127.0.0.1:54321 - "POST /replicate-webhook HTTP/1.1" 200 OK
Troubleshooting
- If your webhook endpoint does not receive events, ensure it is publicly accessible over HTTPS.
- Check that the webhook URL is correctly configured in your Replicate dashboard or API.
- Verify webhook signatures if you use secret verification to avoid unauthorized requests.
- Use logging to debug incoming webhook payloads and HTTP status codes.
Key Takeaways
- Configure a public HTTPS endpoint to receive Replicate webhook POST requests.
- Use frameworks like Flask or FastAPI to implement webhook handlers in Python.
- Secure webhooks by verifying the X-Replicate-Signature header with your secret.
- Test webhook delivery by triggering model predictions and monitoring logs.
- Ensure webhook URLs are correctly set in the Replicate dashboard or API.