How to develop a Kili Webhook
Context
This notebook is an end-to-end example that you can follow to create a project, register a first webhook and activate it on this project for the corresponding events in Kili.
For more information on the available events, please refer to the documentation.
Webhooks are really similar to the plugins, except they are self hosted, and will require a webservice deployed on your premise, callable by Kili (You can implement a header-based security).
If you are a Europe SaaS user, plugins & webhooks are available for paying-customers in beta for now. If you are a US/ On Premise customer, only webhooks are available as of version 1.128.0.
Webhook allow you to easily access your custom code, manage the CI/CD of the version of the plugin deployed, and easily integrate it with your own stack.
If you are looking for a more of-the-shelf capability, you can have a look at the plugins development tutorial.
NB: The webhook capabilities of Kili are under active development, and compatible with version 2.128.0 and later of Kili. Don't hesitate to reach out via Github or the Kili support to provide feedback.
Step 1: Instantiate Kili
%pip install kili
from kili.client import Kili
kili = Kili(
# api_endpoint="https://cloud.kili-technology.com/api/label/v2/graphql",
# the line above can be uncommented and changed if you are working with an on-premise version of Kili
)
Step 2: Create the project
First, we need to create a new project. In our example, we will use an IMAGE type project with the following jsonInterace:
json_interface = {
"jobs": {
"JOB_0": {
"content": {
"categories": {
"OBJECT_A": {
"children": [],
"name": "Object A",
"color": "#733AFB",
"id": "category1",
},
"OBJECT_B": {
"children": [],
"name": "Object B",
"color": "#3CD876",
"id": "category2",
},
},
"input": "radio",
},
"instruction": "Categories",
"isChild": False,
"tools": ["rectangle"],
"mlTask": "OBJECT_DETECTION",
"models": {},
"isVisible": True,
"required": 1,
"isNew": False,
}
}
}
title = "[Kili SDK Notebook]: Webhooks example test project"
description = "My first project with a webhook"
input_type = "IMAGE"
project = kili.create_project(
title=title, description=description, input_type=input_type, json_interface=json_interface
)
project_id = project["id"]
print(f"Created project {project_id}")
Created project clfcblkni05pq0jrq8wgib142
Upload an asset:
content_array = ["https://storage.googleapis.com/label-public-staging/car/car_1.jpg"]
names_array = ["car"]
kili.append_many_to_dataset(
project_id=project_id,
content_array=content_array,
external_id_array=names_array,
disable_tqdm=True,
)
asset_id = kili.assets(project_id=project_id, fields=["id"], disable_tqdm=True)[0]["id"]
This project has one job of bounding box creation with two categories.
With our plugin, we want to make sure that the labelers don't create more than one bounding box of category A.
To iterate on the plugin code, you can refer to the plugins development tutorial.
Step 3: Write & host the webhook
The webhook rely on the same handlers provided by the plugins. For maximum compatibility, we encourage you to define it with the same base class. Below is an example with FastAPI webservice.
# file plugin.py
"""
Custom module with basic plugin example
"""
from typing import Dict
from kili.plugins import PluginCore
def check_rules_on_label(label: Dict):
"""
Custom function to check rules on label.
For basic rules, a handy object is `search` that \
provides various analytics on objects
For more advanced use-cases, you might need to \
fetch the complete `jsonResponse`
"""
issues_array = []
for job_dot_category, nb_of_objects in label['search']['numberOfAnnotationsByObject'].items():
if job_dot_category == "JOB_0.OBJECT_A":
if nb_of_objects > 1:
issues_array.append({
'text': f'There are too many BBox ({nb_of_objects}) - Only 1 BBox of Object A accepted',
'mid': None}
)
return issues_array
def _get_area(bounding_box):
"""
Custom helper to compute size of Kili Bounding boxes
"""
x_array = [point['x'] for point in bounding_box]
y_array = [point['y'] for point in bounding_box]
width = max(x_array) - min(x_array)
height = max(y_array) - min(y_array)
return width * height
class PluginHandler(PluginCore):
"""
Custom plugin instance
"""
def check_complex_rules_on_label(self, asset_id: str):
"""
Custom method to check if a box is larger than 33% of the image
For basic rules, a handy object is `search` that \
provides various analytics on objects
In this more complex use-case, we will \
fetch the complete `jsonResponse`
"""
json_response = self.kili.labels(
asset_id=asset_id,
project_id=self.project_id,
fields=['jsonResponse'],
disable_tqdm=True
)[0]['jsonResponse']
issues_array = []
for annotation in json_response['JOB_0']['annotations']:
bounding_box = annotation['boundingPoly'][0]['normalizedVertices']
area = _get_area(bounding_box)
# Refuse bounding boxes larger than 0.33
if area > 0.33:
issues_array.append({
'text': 'BBox too large',
'mid': annotation["mid"]
})
return issues_array
def on_submit(self, label: Dict, asset_id: str) -> None:
"""
Dedicated handler for Submit action
"""
self.logger.info("On submit called")
issues_array = check_rules_on_label(label)
issues_array += self.check_complex_rules_on_label(asset_id)
project_id = self.project_id
if len(issues_array) > 0:
print(f'Creating {len(issues_array)} issues...')
for issue in issues_array:
print(issue)
self.kili.create_issues(
label_id_array=[label['id']],
project_id=project_id,
text_array=[issue['text']],
object_mid_array=[issue['mid']]
)
self.logger.warning("Issue created!")
self.kili.send_back_to_queue(asset_ids=[asset_id])
else:
self.logger.info('No issues encountered')
You will need to deploy this on your premise for this to work. Easy solutions are FastAPI, with a few lines of codes, and to quickly test your code, we recommend ngrok that allows to quickly expose your local server.
For this demo, we will also display the use of https://webhook.site that will enable us to explore the payload of the calls.
You can also add a custom Authorization header when creating the webhook in Kili, and then verify that header in your deployed webhook. As an example, you can see the verify_token function below.
"""
Basic app for kili webhook
Note: Don't host it locally, it won't work as Kili can't call your localhost
"""
# file main.py
import os
from typing import Dict
from fastapi import FastAPI, HTTPException, Depends, Request
from kili.client import Kili
# Assuming your plugin is in a file `plugin.py` in the same folder
from plugin import PluginHandler
app = FastAPI()
kili = Kili()
API_KEY = "secret-api-key"
# Define the token verification, here we assume we only check if the header is equal
# to a secret value that can be hard-coded / defined from an environment variable, etc.
def verify_token(req: Request):
"""
Verifies the request token
Parameters
----------
req: request
"""
print('Verifying token...')
token = req.headers.get('Authorization')
if token != API_KEY:
print('Token different from API_KEY...')
raise HTTPException(
status_code=401,
detail='Unauthorized'
)
print('Token ok.')
return True
@app.post("/")
def main(raw_payload: Dict, authorized: bool = Depends(verify_token)):
"""
Basic endpoint to receive kili events
Parameters
----------
- raw_payload: webhook payload
- authorized: bool
Has the request been authorized
"""
if not authorized:
print('Not authorized, early return')
return None
event_type = raw_payload.get('eventType')
project_id = raw_payload.get('logPayload').get('projectId')
if not project_id:
print('Invalid projectId')
return
plugin = PluginHandler(kili, project_id)
if not event_type:
print('Invalid event')
return
payload = raw_payload.get('payload')
label = payload.get('label')
asset_id = payload.get('asset_id')
if event_type == 'onSubmit':
plugin.on_submit(label, asset_id)
if event_type == 'onReview':
plugin.on_review(label, asset_id)
Local dev webhook
To quickly get started, setup the following folder:
├── local_webhook
├── main.py
└── plugin.py
To start your fastapi app, just run uvicorn main:app --reload to have live reload in case your code changes.
To start exposing your app, just run ngrok http 8000. You will need to register on ngrok to be able to request a public url that redirects to your computer.
Session Status online
Account *** (Plan: Free)
Update update available (version 2.3.41, Ctrl-U to update)
Version 2.3.40
Region United States (us)
Web Interface http://127.0.0.1:4040
Forwarding http://your-unique-id.ngrok-free.app -> http://localhost:8000
Forwarding https://your-unique-id.ngrok-free.app -> http://localhost:8000
HTTP Requests
-------------
POST / 200 OK
Then, follow the rest of the tutorial to register the webhook with the https url returned by ngrok.
Step 4: Register & activate the webhook
import requests
from kili.exceptions import GraphQLError
# we get a new webhook listener
res = requests.post("https://webhook.site/token")
uuid = res.json()["uuid"]
webhook_url_from_browser = f"https://webhook.site/#!/{uuid}"
webhook_name = "Webhook bbox count"
webhook_url = f"https://webhook.site/{uuid}"
print(webhook_url_from_browser)
# The Authorization header that will be used when calling your deployed webhook
webhook_security_header = "secret-api-key"
try:
kili.create_webhook(
plugin_name=webhook_name, webhook_url=webhook_url, header=webhook_security_header
)
except GraphQLError as error:
print(str(error))
https://webhook.site/#!/f81dfe6a-****-****-****-4b6dfe4b0721
kili.activate_plugin_on_project(plugin_name=webhook_name, project_id=project_id)
Plugin with name "Webhook bbox count" activated on project "clfcblkni05pq0jrq8wgib142"
INFO:kili.services.plugins:Plugin with name "Webhook bbox count" activated on project "clfcblkni05pq0jrq8wgib142"
'Plugin with name Webhook bbox count successfully activated'
Note: Similar to plugins, you have access to the methods kili.update_webhook & kili.deactivate_plugin_on_project for iterations on your code.
Step 5: Webhook in action
After that, you can test it by labelling in the Kili interface or just by uploading the following label.
When you add the label that contains errors, you will see a new issue automatically created in the Kili app, if you have deployed the webhook. Else, you can visit the webhook site to check incoming events.
json_response = {
"JOB_0": {
"annotations": [
{
"boundingPoly": [
{
"normalizedVertices": [
{"x": 0.15, "y": 0.84},
{"x": 0.15, "y": 0.31},
{"x": 0.82, "y": 0.31},
{"x": 0.82, "y": 0.84},
]
}
],
"categories": [{"name": "OBJECT_A"}],
"children": {},
"mid": "20221124161451411-13314",
"type": "rectangle",
},
{
"boundingPoly": [
{
"normalizedVertices": [
{"x": 0.79, "y": 0.20},
{"x": 0.79, "y": 0.13},
{"x": 0.91, "y": 0.13},
{"x": 0.91, "y": 0.20},
]
}
],
"categories": [{"name": "OBJECT_A"}],
"children": {},
"mid": "20221124161456406-47055",
"type": "rectangle",
},
{
"boundingPoly": [
{
"normalizedVertices": [
{"x": 0.87, "y": 0.36},
{"x": 0.87, "y": 0.27},
{"x": 0.99, "y": 0.27},
{"x": 0.99, "y": 0.36},
]
}
],
"categories": [{"name": "OBJECT_A"}],
"children": {},
"mid": "20221124161459298-45160",
"type": "rectangle",
},
]
}
}
kili.append_labels(
json_response_array=[json_response], asset_id_array=[asset_id], label_type="DEFAULT"
)
[{'id': 'clfcblncs0h550js5golxg96s'}]
If you used & hosted the base webhook provided, the webhook should:
- Create an issue with information that three bboxes were found, instead of one
- Create an issue with info that the first bbox is too large
- Send the asset back to the labeling queue (status
ONGOING)
If you haven't deployed your webhook just yet, you can still visit the address here :
print(f"Go to my webhook: {webhook_url_from_browser}")
try:
# If your webhook is live !
kili.issues(project_id=project_id, fields=["comments.text", "objectMid"])
except GraphQLError as error:
print(str(error))
Go to my webhook: https://webhook.site/#!/f81dfe6a-****-****-****-4b6dfe4b0721
Woah! Amazing! Well done :) 🚀
Let's test now to post a proper label, this one for example:
json_response = {
"JOB_0": {
"annotations": [
{
"boundingPoly": [
{
"normalizedVertices": [
{"x": 0.15, "y": 0.84},
{"x": 0.15, "y": 0.31},
{"x": 0.82, "y": 0.31},
{"x": 0.82, "y": 0.84},
]
}
],
"categories": [{"name": "OBJECT_A"}],
"children": {},
"mid": "20221124161451411-13314",
"type": "rectangle",
}
]
}
}
kili.append_labels(
json_response_array=[json_response], asset_id_array=[asset_id], label_type="DEFAULT"
)
print(f"Go to my webhook: {webhook_url_from_browser}")
Go to my webhook: https://webhook.site/#!/f81dfe6a-****-****-****-4b6dfe4b0721
The status of your asset should have now changed to LABELED. In this webhook, previous issues remain but you can solve them through the API as well.
Well done! You can now iterate on the script. To learn how to avoid latency when building and deploying your plugin, refer to the plugins development tutorial.
kili.delete_project(project_id)