Google Chat Integration

pip install levelfour fastapi uvicorn httpx

import os

import httpx
from fastapi import FastAPI, Request, Response

from levelfour.webhooks.verifier import WebhookVerificationError, WebhookVerifier

app = FastAPI()

WEBHOOK_SECRET = os.environ["LEVELFOUR_WEBHOOK_SECRET"]
GOOGLE_CHAT_WEBHOOK_URL = os.environ["GOOGLE_CHAT_WEBHOOK_URL"]

verifier = WebhookVerifier(WEBHOOK_SECRET)

EVENT_LABELS = {
    "recommendation.accepted": "Recommendation Accepted",
    "recommendation.rejected": "Recommendation Rejected",
    "optimization.started": "Optimization Started",
    "optimization.completed": "Optimization Completed",
    "optimization.failed": "Optimization Failed",
}


def build_chat_message(event_type: str, payload: dict) -> dict:
    label = EVENT_LABELS.get(event_type, event_type)
    rec_id = payload.get("recommendation_id", "unknown")
    status = payload.get("status", "unknown")

    text = f"*LevelFour: {label}*\nRecommendation: `{rec_id}`\nStatus: `{status}`"

    if event_type == "recommendation.accepted":
        accepted_by = payload.get("saving_accepted_by", "unknown")
        text += f"\nAccepted by: {accepted_by}"

    if event_type == "recommendation.rejected":
        reason = payload.get("rejection_reason", "No reason provided")
        text += f"\nReason: {reason}"

    if event_type == "optimization.started":
        method = payload.get("implementation_method", "unknown")
        text += f"\nMethod: {method}"

    if event_type == "optimization.failed":
        text += "\n⚠️ Optimization failed"

    return {"text": text}


async def post_to_google_chat(message: dict) -> None:
    async with httpx.AsyncClient() as client:
        await client.post(GOOGLE_CHAT_WEBHOOK_URL, json=message)


@app.post("/webhook")
async def handle_webhook(request: Request) -> Response:
    body = await request.body()
    headers = {
        "webhook-id": request.headers.get("webhook-id", ""),
        "webhook-timestamp": request.headers.get("webhook-timestamp", ""),
        "webhook-signature": request.headers.get("webhook-signature", ""),
    }

    try:
        payload = verifier.verify(payload=body, headers=headers)
    except WebhookVerificationError:
        return Response(status_code=400, content="Invalid signature")

    event_type = payload.get("type", "")
    message = build_chat_message(event_type, payload)
    await post_to_google_chat(message)

    return Response(status_code=200, content="OK")

LEVELFOUR_WEBHOOK_SECRET="whsec_..." \
GOOGLE_CHAT_WEBHOOK_URL="https://chat.googleapis.com/v1/spaces/..." \
uvicorn main:app --port 8000

npm install levelfour express

import express from "express";
import { WebhookVerifier, WebhookVerificationError } from "levelfour";

const WEBHOOK_SECRET = process.env.LEVELFOUR_WEBHOOK_SECRET!;
const GOOGLE_CHAT_WEBHOOK_URL = process.env.GOOGLE_CHAT_WEBHOOK_URL!;
const PORT = parseInt(process.env.PORT || "3000", 10);

const verifier = new WebhookVerifier(WEBHOOK_SECRET);

const EVENT_LABELS: Record<string, string> = {
    "recommendation.accepted": "Recommendation Accepted",
    "recommendation.rejected": "Recommendation Rejected",
    "optimization.started": "Optimization Started",
    "optimization.completed": "Optimization Completed",
    "optimization.failed": "Optimization Failed",
};

interface WebhookPayload {
    type?: string;
    recommendation_id?: string;
    status?: string;
    saving_accepted_by?: string;
    rejection_reason?: string;
    implementation_method?: string;
    [key: string]: unknown;
}

function buildChatMessage(eventType: string, payload: WebhookPayload): { text: string } {
    const label = EVENT_LABELS[eventType] || eventType;
    const recId = payload.recommendation_id || "unknown";
    const status = payload.status || "unknown";

    let text = `*LevelFour: ${label}*\nRecommendation: \`${recId}\`\nStatus: \`${status}\``;

    if (eventType === "recommendation.accepted" && payload.saving_accepted_by) {
        text += `\nAccepted by: ${payload.saving_accepted_by}`;
    }

    if (eventType === "recommendation.rejected") {
        const reason = payload.rejection_reason || "No reason provided";
        text += `\nReason: ${reason}`;
    }

    if (eventType === "optimization.started" && payload.implementation_method) {
        text += `\nMethod: ${payload.implementation_method}`;
    }

    if (eventType === "optimization.failed") {
        text += "\n⚠️ Optimization failed";
    }

    return { text };
}

const app = express();
app.use(express.raw({ type: "application/json" }));

app.post("/webhook", async (req, res) => {
    const body = req.body as Buffer;
    const headers: Record<string, string> = {
        "webhook-id": req.headers["webhook-id"] as string || "",
        "webhook-timestamp": req.headers["webhook-timestamp"] as string || "",
        "webhook-signature": req.headers["webhook-signature"] as string || "",
    };

    let payload: WebhookPayload;
    try {
        payload = verifier.verify(body, headers) as WebhookPayload;
    } catch (err) {
        if (err instanceof WebhookVerificationError) {
            res.status(400).send("Invalid signature");
            return;
        }
        throw err;
    }

    const eventType = payload.type || "unknown";
    const message = buildChatMessage(eventType, payload);

    await fetch(GOOGLE_CHAT_WEBHOOK_URL, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify(message),
    });

    res.status(200).send("OK");
});

app.listen(PORT, () => {
    console.log(`Listening on port ${PORT}`);
});

LEVELFOUR_WEBHOOK_SECRET="whsec_..." \
GOOGLE_CHAT_WEBHOOK_URL="https://chat.googleapis.com/v1/spaces/..." \
PORT=3000 \
npx tsx server.ts