Speckle Webhooks

Build a simple Discord integration that will send notifications when stream events occur on your Speckle Server - and learn how to use Speckle's webhooks in the process!

a WebhookServer that will listen for webhook requests from your stream a webhook_called function that will verify that the signature of incoming requests a Discord message template an on_stream_update function which populates your Discord message template with data from your webhook payload an add_author function which handles the author's info and avatar (whether it's a URL or a base64 image that needs to be decoded and attached) a send_to_discord method which puts together and sends a POST request to Discord to send your message

import cherrypy
import requests
import base64
import json
import hmac


# It's not a good practice to hardcode the secret or the URL. It's best to get it from an environment variable.
# See the finished project for an example.
SECRET = ""
DISCORD_URL = "https://discord.com/api/webhooks/1023657743040659466/oyr5N5hhxRScNcqI4ISPNnRd8RFzbOC4bX9dCCkv0z-kfVJAtHKtSoRg5Cja0oPn3mqz"


def get_message_template():
    return {
        "username": "Speckle",
        "avatar_url": "https://avatars.githubusercontent.com/u/65039012?s=200&v=4",
        "embeds": [
            {
                "author": {
                    "name": "speckle user",
                    "url": "",
                    "icon_url": "https://avatars.githubusercontent.com/u/65039012?s=200&v=4",
                },
                "title": "",
                "url": "",
                "description": "",
                "color": 295163,
                "image": {"url": None},
            }
        ],
        "files": [],
    }


def send_to_discord(message: str):
    files = message.pop("files")  # any images we're attaching (user avatar)
    # dump our message to the `payload_json` field
    files.append(("payload_json", (None, json.dumps(message))))
    res = requests.post(DISCORD_URL, files=files)


def add_author(msg, user_info, server_url):
    avatar = user_info["avatar"]  # get the user's avatar
    if avatar and not avatar.startswith("http"):
        type = avatar[5:].split(";")[0]
        filename = f"avatar.{type.split('/')[1]}"
        # decode and prepare it for being uploaded
        msg["files"].append(
            (
                "file",
                (
                    filename,
                    (base64.b64decode(avatar.split(",")[1])),
                    type,
                ),
            )
        )
        avatar = f"attachment://{filename}"  # attachment syntax for files

    msg["embeds"][0]["author"].update(
        {
            "name": user_info["name"],
            "url": f"{server_url}/profile/{user_info['id']}",
            "icon_url": avatar,
        }
    )


def on_stream_update(server_info, user_info, stream_info, webhook_info, event_info):
    server_url = server_info["canonicalUrl"].rstrip("/")
    msg = get_message_template()
    msg["embeds"][0].update(
        {
            "title": f"Stream Updated: [{stream_info['name']}]",
            "url": f"{server_url}/streams/{stream_info['id']}",
            "description": f"{user_info['name']} updated stream `{stream_info['id']}`",
            "fields": [
                {
                    "name": "Old",
                    "value": f"**Name:** {event_info['old']['name']}\n**Description:** {event_info['old']['description'] if len(event_info['old']['description']) < 30 else event_info['old']['description'][:30] + '...'}\n**Is Public:** {event_info['old']['isPublic']}",
                    "inline": True,
                },
                {
                    "name": "Updated",
                    "value": f"**Name:** {event_info['new']['name']}\n**Description:** {event_info['new']['description'] if len(event_info['new']['description']) < 30 else event_info['new']['description'][:30] + '...'}\n**Is Public:** {event_info['new']['isPublic']}",
                    "inline": True,
                },
            ],
            "image": {"url": f"{server_url}/preview/{stream_info['id']}"},
        }
    )
    add_author(msg, user_info, server_url)  # * see next section for more info on this
    send_to_discord(msg)


# Web server:
class WebhookServer(object):
    @cherrypy.expose
    @cherrypy.tools.json_in()
    def webhook(self, *args, **kwargs):
        payload_json = cherrypy.request.json.get("payload", "{}")
        signature = cherrypy.request.headers["X-WEBHOOK-SIGNATURE"]
        self.webhook_called(payload_json, signature)

    # This function will be called each time the webhook endpoint is accessed
    def webhook_called(self, payload_json: str, signature: str):
        # verify the signature
        expected_signature = hmac.new(
            SECRET.encode(), payload_json.encode(), "sha256"
        ).hexdigest()
        if not hmac.compare_digest(expected_signature, signature):
            print("Ignoring request with invalid signature")
            return
        payload = json.loads(payload_json)
        print("Received webhook payload:\n" + json.dumps(payload, indent=4))

        # check the event and invoke the correct response
        event_name = payload.get("event", {}).get("event_name", "UNKNOWN")
        if event_name == "stream_update":
            on_stream_update(
                payload["server"],
                payload["user"],
                payload["stream"],
                payload["webhook"],
                payload["event"].get("data", {}),
            )


cherrypy.config.update(
    {
        # TODO: uncomment the following line after finishing development
        # 'environment': 'production',
        "server.socket_host": "0.0.0.0",
        "server.socket_port": 8003,
    }
)

# start the server
cherrypy.quickstart(WebhookServer())

---

ngrok server: https://dashboard.ngrok.com/get-started/setup webhook discord-speckle: https://speckle.systems/tutorials/webhooks-discord-tutorial/