📗

Integrating Webhooks

Process Unitary's API responses in a scalable way

Webhooks are currently the best and fastest way to access results and act on them as they come.
Whenever a job finishes processing, a POST request will be sent to the callback_url provided in the body of the job request. (please check the API Integration reference for more details on how to send a job request).
The body of the request is defined in the Webhook Payloads guide.
Example: Simple webhook receiver FastAPI
The following Python (FastAPI) server will listen for results and log the jobs that are having at least one GARM category detected.
from fastapi import Request, FastAPI
​
app = FastAPI()
​
@app.post("/results")
async def get_body(request: Request):
    result = await request.json()
    if result["is_error"] is False:
        garm_categories = result["result"].get("garm_categories", [])
        if len(garm_categories) > 0:
            print("GARM category detected")
​
    return "OK"
Validating payloads
Please read Webhook Payloads to learn more about how outputs are validated.
There are currently two ways to set up the secret key:
1.Request for one to be generated and shared with you.
2.Set up a custom one and share it with us.
Please send the chosen option to your dedicated Account Manager.
Example: Validating payloads
In the following example, the payload is validated according to the secret key that has previously been set up.
from fastapi import Header, Request, FastAPI
​
import base64
import hashlib
import hmac
​
SECRET_KEY = "test_key"
​
app = FastAPI()
​
def compute_signature(payload: str, secret_key: str) -> str:
    digest = hmac.new(
        secret_key.encode(), msg=payload.encode("utf-8"), digestmod=hashlib.sha256
    ).digest()
    signature = base64.b64encode(digest).decode()
    return signature
​
@app.post("/results")
async def get_body(request: Request, x_hub_signature_256: str | None = Header(default=None)):
    result = await request.body()
    if compute_signature(result.decode(), SECRET_KEY) == x_hub_signature_256:
        print("Body validated")
        # .. do processing
    else:
        print("Body not validated")
    return "OK"

Last modified 1mo ago