send_event.py

"""Sends one signed event to a Subcontext inbound custom integration, in Python 3.9
or later with the standard library only. Here it's a sale from your CRM, which
Subcontext records as an outcome on the lead.

Set these first (all three are on the integration's page in Subcontext):
  SUBCONTEXT_DELIVERY_URL      the Delivery URL
  SUBCONTEXT_SIGNING_SECRET    the value of the secret variable chosen as the Signing secret variable
  SUBCONTEXT_SIGNATURE_HEADER  the Signature header (X-Integrations-Signature unless you changed it)

Run it:  python3 send_event.py
"""

import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.request
import uuid
from datetime import datetime, timezone

DELIVERY_URL = os.environ.get("SUBCONTEXT_DELIVERY_URL")
SECRET = os.environ.get("SUBCONTEXT_SIGNING_SECRET")
HEADER = os.environ.get("SUBCONTEXT_SIGNATURE_HEADER", "X-Integrations-Signature")


def signature_header(raw_body: bytes, secret: str, timestamp: int = None) -> str:
    timestamp = timestamp or int(time.time())
    digest = hmac.new(secret.encode(), f"{timestamp}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return f"t={timestamp},v1={digest}"


def main() -> None:
    if not DELIVERY_URL or not SECRET:
        raise SystemExit("Set SUBCONTEXT_DELIVERY_URL and SUBCONTEXT_SIGNING_SECRET.")

    # The shape is up to you. The integration's settings say where to find each part.
    # This one fits an Event type path of "type" and the outcome mapping on the
    # Integrations guide.
    event = {
        "event_id": str(uuid.uuid4()),  # send the same id again if you retry, so it's recorded once
        "type": "deal.won",
        "deal": {
            "id": "DEAL-123",
            "contact_id": "YOUR_CRM_CONTACT_ID",  # your id for the lead, or send ours (contact_id) instead
            "closed_at": datetime.now(timezone.utc).isoformat(),
            "commission": "450.00",
        },
    }

    # Sign the exact bytes you send.
    body = json.dumps(event).encode()
    request = urllib.request.Request(DELIVERY_URL, data=body, method="POST", headers={
        "Content-Type": "application/json",
        HEADER: signature_header(body, SECRET),
    })
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            print(response.status, "accepted")
    except urllib.error.HTTPError as error:
        # 400: not JSON. 403: signature, secret or clock problem. 404: wrong URL.
        print(error.code, "not accepted")


if __name__ == "__main__":
    main()