# Integrating with Labeling Using Webhooks ## Motivation A common usecase for Aquarium datasets is to identify problem labels and group them using *segments*, downloading a json or csv representation of a segment's *elements*, and then using scripting or a manual process to reformat the data to submit to a labeling service, whether an external vendor or an internal tool. By leveraging webhook configurations, the reformatting and submitting steps can be automated by a single handler. **This allows an Aquarium user to directly send data to a labeling service directly from the UI without writing code each time.** The *issue-exported* event will POST the issue's elements to a webhook. A full schema can be found in the [Event Schemas section](/aquarium/concepts/webhooks.md#issue-exported) of the general Webhooks page. {% code title="Schema with Example Map Keys" %} ```json { event: "issue-exported", project: str, issue: { id: str, elements: [{ dataset: str, inference_set: str, issue_name: str, element_id: str, element_type: "frame" | "crop", frame_id: str, frame_data: { coordinate_frames: [{ coordinate_frame_id: str, coordinate_frame_metadata: Optional[Dict], coordinate_frame_type: str, }], custom_metrics: { [custom_metric_type]: int[][] | float, ... }, date_captured: str, device_id: str, geo_data: { [coordinate_field]: float, ... }, label_data: [{ attributes: { confidence: float, ... }, label: str, label_coorindate_frame: str, label_type: str, linked_labels: str[], uuid: str, }, ...], sensor_data: [{ coordinate_frame: str, data_urls: { image_url: str, ... }, date_captured: str, sensor_id: str, sensor_metadata: Dict, sensor_type: str, }, ...], task_id: str, [user__metadata_field]: str, }, }, ...] } } ``` {% endcode %} ## Shaping the Webhook Payload To Send to Labeling You can then use the elements in `issue` to create a new payload that is accepted by a labeling service; we've provided some sample code that serves a webhook endpoint and transforms an Aquarium payload to some common formats. {% tabs %} {% tab title="GraphQL" %} {% code title="server.py" %} ```python from flask import Flask, request import os from python_graphql_client import GraphqlClient AQ_WEBHOOK_SECRET = os.getenv("AQ_WEBHOOK_SECRET") LABELING_API_KEY = os.getenv("LABELING_API_KEY") LABELING_API_ENDPOINT = os.getenv("LABELING_API_ENDPOINT") labeling_api_headers = { # Replace with the proper API key header if any for your service "Authorization": f"Bearer {LABELING_API_KEY}" } client = GraphqlClient(endpoint=LABELING_API_ENDPOINT, headers=labeling_api_headers) # Replace with appropriate graphql mutation. # In this example, the way to requeue a label is to remove it and mark as a template relabel_mutation_fragment = """ mutation BulkDeleteLabels ( $projectId: ID!, $makeTemplates: Boolean = true, $labelIds: [ID!]) { project (where: {id: $projectId}) { bulkDeleteLabels ( where: { id_in: $labelIds }, makeTemplates: $makeTemplates, waitForQueue: true ) { count } } } """ app = Flask(__name__) @app.route("/webhook", methods=["POST"]) def handle_webhook_payload(): # Optionally verify that the payload came from Aquarium aq_secret = request.headers.get("x-aquarium-secret") if aq_secret != AQ_WEBHOOK_SECRET: return f"Bad Request: {msg}", 400 payload_envelope = request.get_json() if not payload_envelope: msg = "no payload body received" print(f"error: {msg}") return f"Bad Request: {msg}", 400 if not isinstance(payload_envelope, dict) or not payload_envelope.get("event"): msg = "invalid webhook payload format" print(f"error: {msg}") return f"Bad Request: {msg}", 400 event_type = payload_envelope["event"] if event_type == "issue-exported": if not payload_envelope.get("issue"): msg = "webhook payload did not contain expected key: issue" print(f"error: {msg}") return f"Bad Request: {msg}", 400 _format_and_export_to_graphql_api(payload_envelope["project_name"], payload_envelope["issue"]) else: msg = f"endpoint not setup to handle {event_type} events yet" print(f"error: {msg}") return f"Bad Request: {msg}", 400 return ("", 204) def _format_and_export_to_graphql_api(project_name, issue): label_ids = set() for element in issue["elements"]: if element["element_type"] == "crop": label_ids.add(element["crop_data"]["uuid"]) else: for label in element["frame_data"]["label_data"]: label_ids.add(label["uuid"]) variables = { "projectId": project_name, "labelIds": list(label_ids) } client.execute( query=relabel_mutation_fragment, variables=variables, ) if __name__ == "__main__": PORT = int(os.getenv("PORT")) if os.getenv("PORT") else 8080 app.run(host="127.0.0.1", port=PORT, debug=True) ``` {% endcode %} {% endtab %} {% tab title="REST" %} {% code title="server.py" %} ```python from flask import Flask, request import os import requests AQ_WEBHOOK_SECRET = os.getenv("AQ_WEBHOOK_SECRET") LABELING_API_KEY = os.getenv("LABELING_API_KEY") LABELING_API_ENDPOINT = os.getenv("LABELING_API_ENDPOINT") labeling_api_headers = { # Substitute with the proper API key header if any for your service "Authorization": f"Bearer {LABELING_API_KEY}" } app = Flask(__name__) @app.route("/webhook", methods=["POST"]) def handle_webhook_payload(): # Optionally verify that the payload came from Aquarium aq_secret = request.headers.get("x-aquarium-secret") if aq_secret != AQ_WEBHOOK_SECRET: return f"Bad Request: {msg}", 400 payload_envelope = request.get_json() if not payload_envelope: msg = "no payload body received" print(f"error: {msg}") return f"Bad Request: {msg}", 400 if not isinstance(payload_envelope, dict) or not payload_envelope.get("event"): msg = "invalid webhook payload format" print(f"error: {msg}") return f"Bad Request: {msg}", 400 event_type = payload_envelope["event"] if event_type == "issue-exported": if not payload_envelope.get("issue"): msg = "webhook payload did not contain expected key: issue" print(f"error: {msg}") return f"Bad Request: {msg}", 400 _format_and_export_to_rest_api(payload_envelope["project_name"], payload_envelope["issue"]) else: msg = f"endpoint not setup to handle {event_type} events yet" print(f"error: {msg}") return f"Bad Request: {msg}", 400 return ("", 204) def _format_and_export_to_rest_api(project_name, issue): relabel_frames = [] for element in issue["elements"]: relabel_frames.append({ "id": element["frame_id"], "url": element["frame_data"]["sensor_data"]["data_urls"]["image_url"] # select the right key for your media type }) # Replace with appropriate post body. # In this example, we assume the way to requeue a label is to resubmit the frame(s) as a new batch new_dataset_payload = { "name": f"{issue['dataset']}_relabel_{issue['issue_name']}", "project": project_name, "frames": relabel_frames } requests.post(LABELING_API_ENDPOINT, json=new_dataset_payload, headers=labeling_api_headers) if __name__ == "__main__": PORT = int(os.getenv("PORT")) if os.getenv("PORT") else 8080 app.run(host="127.0.0.1", port=PORT, debug=True) ``` {% endcode %} {% endtab %} {% endtabs %} ## Triggering a Segment Export To trigger an export, click the Export Issue button on an issue's page. If there aren't any webhooks configured to handle the *issue-exported* event on the issue's project, it will prompt you to create one. ![Export to Labeling](/files/DaCvcLK6pG7dc9xtyTJ7) ![Confirmation step where you can preview the payload](/files/-M_gsqYlHEhZIvvkMTeN) --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://legacy-docs.aquariumlearning.com/aquarium/concepts/webhooks/integrating-with-labeling-using-webhooks.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.