Skip to main content

webhooks

Note!

Webhooks is an advanced feature that is currently in beta.

Introduction​

The SDK supports sending notifications when certain events happen in the system. The notifications are sent via HTTP POST and the messages are signed with HMAC-SHA256 using a shared secret.

Registering endpoint​

While in beta you will need to contact Treble to register your endpoint, endpoint registration will become available to organization administrators via the management portal soon.

Message structure​

Each POST message will contain a JSON body in the following format:

{
   "messageId":"00000000-0000-0000-0000-000000000000",
   "createdAt":"2025-01-01T00:00:00.000Z",
   "payload": ...,
   "signature":"...."
}

Where messageId is a 'unique' guid of this particular message and createdAt is the timestamp when the event message was created, these should be the same values as the messageId and createdAt fields within the payload. The content of the payload is different between events (See available events). Signature contains the HMAC-SHA256 hash of the payload.

Note

All timestamps are in UTC and on ISO 8601 format (YYYY-MM-ddTHH:mm:ss.fffZ).

Note

We try to send the event messages in the order that they happen but there is no guarantee that the ordering of messages will be intact. E.g. a simulation.completed message may be received before a simulation.running message for very short simulations.

IMPORTANT!

Once the endpoint has received the message it should return the HTTP status code 200. Any other status code will be considered as a failed delivery which will trigger the retry mechanism of the webhooks system.

Note

Requests from the webhooks system will include the following headers:

{
    "Content-Type": "application/json", 
    "User-Agent": "Treble-Webhooks/1.0"
}
Note

Requests made by the webhooks system have a 10 second timeout.

Note

If the client server does not respond with 200 within the request timeout the request is considered to have failed and will be retried. The webhooks notification system will try to send the request up to 3 times with a delay of around 1 minute after each try.

Validating message signature​

It is important to validate the message signature to make sure that the message originated from Treble and that data has not been tampered with.

Example validate signature code​

import hmac
import hashlib

def verify_signature(payload_str: str, received_signature: str, secret: str) -> bool:
    try:
        calculated_signature = hmac.new(secret.encode("utf-8"), payload_str.encode("utf-8"), hashlib.sha256).hexdigest()
        return hmac.compare_digest(calculated_signature, received_signature)
    except Exception as e:
        print(f"Error verifying signature: {e}")
        return False

Available events​

Following are descriptions of available events and an example message for each event. You can test validating the signature by computing the HMAC-SHA256 of the payload using the secret value "test".

simulation.completed​

This message triggers when a simulation has completed processing, the content of the message will contain information if the simulation completed successfully or not.

{
  "version": "1.0",
  "eventType": "simulation.completed",
  "messageId": "a81af274-f61c-4780-bb44-b493873f700a",
  "createdAt": "2025-09-11T13:02:07.883Z",
  "objectType": "Simulation",
  "objectId": "fc31f4e8-e457-42e9-baba-86122d5c6c1e",
  "organizationId": "00000000-0000-0000-0000-000000000001",
  "data": {
    "projectId": "a407a28c-4450-46a4-acd7-e4170330ad89",
    "status": "Completed",
    "message": null,
    "billedTokens": 0.0101,
    "billable": true,
    "tasks": [
      {
        "id": "d424f8b1-d6d0-4054-bf19-79a17ff32d61",
        "taskType": "GA",
        "taskSubType": "None",
        "status": "Completed",
        "tokenCost": 0.010045,
        "startedAt": "2025-09-11T13:01:39.981Z",
        "completedAt": "2025-09-11T13:02:07.396Z"
      }
    ]
  }
}

simulation.running​

This message triggers when a simulation task has begun solving in the cloud. E.g. Any simulation task has started processing in the cloud after any queue and/or cloud provisioning time, if any.

{
  "version": "1.0",
  "eventType": "simulation.running",
  "messageId": "89dae419-e3ad-4ff7-b2cd-f8a21c12fdd2",
  "createdAt": "2025-09-11T13:01:50.868Z",
  "objectType": "Simulation",
  "objectId": "fc31f4e8-e457-42e9-baba-86122d5c6c1e",
  "organizationId": "00000000-0000-0000-0000-000000000001",
  "data": {
    "simulationId": "fc31f4e8-e457-42e9-baba-86122d5c6c1e",
    "name": "treble_lab_1757595525",
    "projectId": "a407a28c-4450-46a4-acd7-e4170330ad89",
    "status": "InProgress",
    "message": null
  }
}

model.processing.completed​

This message triggers when a model has finished processing and contains information if the model is valid or not.

{
  "version": "1.0",
  "eventType": "model.processing.completed",
  "messageId": "8cf6ffb2-a493-4dc8-8948-d2913d2c7738",
  "createdAt": "2025-09-11T13:52:09.236Z",
  "objectType": "Model",
  "objectId": "5e9477f3-1a57-460e-a10c-099df9dbe980",
  "organizationId": "00000000-0000-0000-0000-000000000001",
  "data": {
    "modelId": "5e9477f3-1a57-460e-a10c-099df9dbe980",
    "name": "Treble lab IIII",
    "projectId": "bd8fb2c9-b66b-466e-987c-47209d060640",
    "status": "Valid",
    "message": "OK",
    "isWatertight": true
  }
}

Example endpoint client​

Following is a primitive example of a webservice that accepts and validates webhook notifications using Python and Flask.

from flask import Flask, request
import json
import logging
import hmac
import hashlib

app = Flask(__name__)

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Expected secret for signature verification (in production, this should come from secure config)
EXPECTED_SECRET = "test"


def verify_signature(payload_str: str, received_signature: str, secret: str) -> bool:
    """Verify the HMAC signature of the payload."""
    try:
        calculated_signature = hmac.new(secret.encode("utf-8"), payload_str.encode("utf-8"), hashlib.sha256).hexdigest()
        return hmac.compare_digest(calculated_signature, received_signature)
    except Exception as e:
        logger.error(f"Error verifying signature: {e}")
        return False


@app.route("/webhook", methods=["POST"])
def receive_webhook():
    """Receive and log webhook notifications"""
    try:
        # Get the request data
        data = request.get_json()
        headers = dict(request.headers)

        # Log the received webhook
        logger.info("=" * 50)
        logger.info("WEBHOOK RECEIVED")
        logger.info(f"Body: {json.dumps(data, indent=2)}")

        # Extract signature and payload for verification
        received_signature = data.get("signature")
        message_payload_str = data.get("payload")
        message_payload = json.loads(message_payload_str)
        logger.info(f"Message payload structured: {json.dumps(message_payload, indent=2)}")

        if received_signature and message_payload:
            # Verify the signature
            is_valid = verify_signature(message_payload_str, received_signature, EXPECTED_SECRET)
            logger.info(f"SIGNATURE VERIFICATION: {'VALID' if is_valid else 'INVALID'}")

            if not is_valid:
                logger.warning("Webhook signature verification failed")
        else:
            logger.warning("Missing signature or payload in webhook data")

        logger.info("=" * 50)

    except Exception as e:
        logger.error(f"Error processing webhook: {e}")

    return "OK", 200


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