Developer HubAPI ReferenceHelp CenterRossum UniversityStatus Page
Guides
« Back to Rossum

How To Use Webhooks

What Are Webhooks

Webhook extensions are user-defined callbacks that are triggered by a real-time notification from Rossum for a related Queue. You can use Webhooks to send notifications to external systems or modify annotation data and display messages to the user. Webhook extensions are similar to Connectors but are more flexible and easier to use.

How Webhooks Work

To start receiving notifications in your microservice, you have to create a Hook object of type “webhook.” This Hook object has to be assigned to your Rossum Queue to receive notifications on various events happening with a document. The Webhook can be created and assigned to a Queue within the Rossum application. Once it receives a notification from Rossum, it will forward it to the remote URL of your choice. On this remote URL, you would receive the data, use them to take a certain action, and send a response back to the Hook. One Webhook can be triggered on multiple events, and you can also have more Hooks attached to one Queue.

A Webhook can be notified on one of the following event actions:

  • Annotation status changed,
  • Annotation content was extracted for the first time,
  • User updated the annotation data,
  • Annotation is being moved to the exported state,
  • Email was sent to a Rossum queue.

The Webhooks that are notified on user update events are notified only once as they are interactive. In case the notification wasn’t successful, it is not retried. For the other Webhook events, if the notification call fails with an HTTP status larger than 400, it is retried within 30 seconds for a maximum of 10 attempts. The webhook timeout is 30 seconds, and it is currently global.

Webhook Requirements

To use a Webhook, you must have your own web server and remote URL ready to listen to the notifications. Read more about how to set up your own microservice here. Suppose you don’t have your own server. In that case, you can use third-party services such as Zapier to build the integration or use some of the existing Zapier integrations].

Webhooks vs. Connectors

You may be unsure about whether you should choose a Connector or a Webhook Extension for your integration. We definitely recommend choosing the Webhook Extension. The Connector Extension is a deprecated extension type present before we implemented Webhooks. But now, Webhooks have multiple advantages over Connectors, namely:

  • There may be multiple Webhooks defined for a single Queue.
  • There is no need to have a hard-coded endpoint name (/validate, /save) in your URL.
  • Webhooks come with a robust retry mechanism in case of the notification failure.

As you can see, a Webhook Extension will provide you with everything that the Connector Extension can and even more.

When To Use Webhooks

There are many use cases where Webhooks can be very helpful. The most frequent scenarios include situations when you need to:

  • Trigger some actions when an annotation status changes or would like to show the current annotation status to the user in your ERP system.
  • Get the initially extracted data to apply custom rules on them to do initial data validation or supplementary data validation in case of automation.
  • Match user input against the master data in your database (e.g., Vendor matching, Purchase Order matching, or Product matching).
  • Conduct some final validations before data is exported to your ERP to prevent exporting incorrect data.

You can read more detailed steps on how to set up a Webhook here or find more technical information in our API documentation. To see a practical example of a Webhook written in Python, check out the following code snippet:

import hashlib
import hmac
from functools import wraps
from typing import Tuple

import flask
from flask_apispec import use_kwargs
import jmespath as jmespath
from flask import jsonify, request
from marshmallow import Schema, fields, EXCLUDE
from werkzeug.exceptions import abort

app = flask.Flask(__name__)
SECRET_KEY = "Your secret key stored in hook.config.secret"  # never store this in code


class UserUpdateSchema(Schema):
   annotation = fields.Dict(required=True, location="json")

   class Meta:
       unknown = EXCLUDE


def hmac_signature_required(f):
   @wraps(f)
   def authorize_request(*args, **kwargs):
       digest = hmac.new(SECRET_KEY.encode(), request.data, hashlib.sha1).hexdigest()
       try:
           prefix, signature = request.headers["X-Elis-Signature"].split("=")
       except ValueError:
           abort(401, "Incorrect header format")
       if not (prefix == "sha1" and hmac.compare_digest(signature, digest)):
           abort(401, "Authorization failed.")
       return f(*args, **kwargs)
   return authorize_request


@app.route("/user_update", methods=["POST"])
@hmac_signature_required
@use_kwargs(UserUpdateSchema())
def api_validate(annotation: dict) -> flask.Response:
   messages, updated_data_fields = zero_string_to_float(annotation["content"])
   return jsonify(
       {
           "messages": messages,
           "operations": updated_data_fields,
       }
   )


def zero_string_to_float(annotation_tree_content: list) -> Tuple[list, list]:
   vat_rate_items = _search_annotation_tree(annotation_tree_content, "item_vat_rate")
   if not vat_rate_items:
       return [{"type": "error", "content": "Column with schema ID 'item_vat_rate' is missing."}], []

   updated_prices: list = []
   for item in vat_rate_items:
       if item["node_value"]:
           if "ZERO" in item["node_value"].upper():
               updated_prices.append({"op": "replace", "id": item["node_id"], "value": {"content": {"value": "0.0"}}})
   return [], updated_prices


def _search_annotation_tree(content: list, schema_id: str):
   return jmespath.search(
       "[].children[].children[].children[?schema_id=='%s'][].{node_id: id, node_value: content.value}"
       % schema_id,
       content,
   )


if __name__ == '__main__':
   app.run(host="0.0.0.0", port=5000)

The code above does the following:

  • Provisions an endpoint /user_update where the webhook listens on the notifications sent from Rossum (via @app.route("/user_update", methods=["POST"])
  • Validates received payloads to authorize the request coming from Rossum (in @hmac_signature_required)
  • Serializes the request from Rossum for easier handling (in UserUpdateSchema())
  • Searches the annotation data for the datapoint with id “item_vat_rate” (in _search_annotation_tree())
  • Changes “ZERO” string to float 0.0
  • Returns the updated data (in api_validate())

Updated about 2 years ago