Rossum Developer Hub

Rossum Data Capture for Developers and Integrators

Welcome to the Rossum developer hub. You'll find comprehensive guides and documentation to help you implement Rossum as quickly as possible, as well as support if you get stuck.

Let's jump right in!

Developer Guides    API Reference    User Help Center

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. Webhooks can be used for sending notifications to external systems or to modify annotation data and display messages to the user. Webhook extensions are similar to Connectors, but they are more flexible and easier to use.

How Webhooks Work

In order to start receiving notification 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 exported state

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 HTTP status larger than 400, it is retried within 30 seconds for a maximum of 10 attempts.

Webhook Requirements

To use a Webhook, you need to 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. If you don’t have your own server, 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 that was present before Webhooks were implemented. 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 names (/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 just 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 a month ago

How To Use Webhooks


Suggested Edits are limited on API Reference Pages

You can only suggest edits to Markdown body content, but not to the API spec.