Ask or search…
⌃K
Links
📗

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. 1.
    Request for one to be generated and shared with you.
  2. 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 3mo ago
Copyright © Unitary Ltd