Slack Integration

LevelFour signs every webhook it sends. This receiver checks the signature, turns the event into Slack blocks, and posts it to a channel. Webhooks owns the headers, the signature algorithm, the timestamp tolerance and the retry rules the code below depends on.

What lands in the channel

Every message opens with the event name, then carries the recommendation id and the status. The event decides what goes underneath.

Event	What the message adds
`recommendation.accepted`	Who accepted it
`recommendation.rejected`	The rejection reason
`optimization.started`	The implementation method
`optimization.completed`	Nothing beyond the status
`optimization.failed`	A warning line

Events has what each one means, and Payloads has the fields each carries.

Build it

Create the Slack app

Create an app at api.slack.com/apps with the chat:write scope, then copy its Bot User OAuth Token. It starts with xoxb-.

Invite the bot to the target channel before the first event arrives. The receiver posts to SLACK_CHANNEL, and the bot has to be in it.

Set the environment

The receiver reads its configuration from the environment.

Variable	Description
`LEVELFOUR_WEBHOOK_SECRET`	Webhook signing secret from LevelFour (`whsec_...`), the one described in Verify the signature
`SLACK_BOT_TOKEN`	Slack Bot User OAuth Token (`xoxb-...`)
`SLACK_CHANNEL`	Target Slack channel (default: `#cloud-costs`)
`PORT`	Port the server listens on (default: `3000`), read by the TypeScript version only

Write the receiver

Pick a language. The Python receiver is FastAPI, the TypeScript one Express, and both do the same three things: install the dependencies, save the server file, and run it.

pip install levelfour fastapi uvicorn httpx

main.py

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"]
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_CHANNEL = os.environ.get("SLACK_CHANNEL", "#cloud-costs")

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_slack_blocks(event_type: str, payload: dict) -> list[dict]:
    label = EVENT_LABELS.get(event_type, event_type)
    rec_id = payload.get("recommendation_id", "unknown")
    status = payload.get("status", "unknown")

    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": f"LevelFour: {label}"},
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*Recommendation:*\n{rec_id}"},
                {"type": "mrkdwn", "text": f"*Status:*\n{status}"},
            ],
        },
    ]

    if event_type == "recommendation.accepted":
        accepted_by = payload.get("saving_accepted_by", "unknown")
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": f"Accepted by {accepted_by}"}],
        })

    if event_type == "recommendation.rejected":
        reason = payload.get("rejection_reason", "No reason provided")
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": f"Reason: {reason}"}],
        })

    if event_type == "optimization.started":
        method = payload.get("implementation_method", "unknown")
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": f"Method: {method}"}],
        })

    if event_type == "optimization.failed":
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": ":warning: Optimization failed"}],
        })

    return blocks


async def post_to_slack(blocks: list[dict]) -> None:
    async with httpx.AsyncClient() as client:
        await client.post(
            "https://slack.com/api/chat.postMessage",
            headers={"Authorization": f"Bearer {SLACK_BOT_TOKEN}"},
            json={"channel": SLACK_CHANNEL, "blocks": blocks},
        )


@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", "")
    blocks = build_slack_blocks(event_type, payload)
    await post_to_slack(blocks)

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

Run it with the variables in the environment:

LEVELFOUR_WEBHOOK_SECRET="whsec_..." \
SLACK_BOT_TOKEN="xoxb-..." \
SLACK_CHANNEL="#cloud-costs" \
uvicorn main:app --port 8000

npm install levelfour express @slack/web-api

server.ts

import express from "express";
import { WebClient, type KnownBlock } from "@slack/web-api";
import { WebhookVerifier, WebhookVerificationError } from "levelfour";

const WEBHOOK_SECRET = process.env.LEVELFOUR_WEBHOOK_SECRET!;
const SLACK_BOT_TOKEN = process.env.SLACK_BOT_TOKEN!;
const SLACK_CHANNEL = process.env.SLACK_CHANNEL || "#cloud-costs";
const PORT = parseInt(process.env.PORT || "3000", 10);

const verifier = new WebhookVerifier(WEBHOOK_SECRET);
const slack = new WebClient(SLACK_BOT_TOKEN);

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 buildSlackBlocks(eventType: string, payload: WebhookPayload) {
    const label = EVENT_LABELS[eventType] || eventType;
    const recId = payload.recommendation_id || "unknown";
    const status = payload.status || "unknown";

    const blocks: KnownBlock[] = [
        {
            type: "header",
            text: { type: "plain_text", text: `LevelFour: ${label}` },
        },
        {
            type: "section",
            fields: [
                { type: "mrkdwn", text: `*Recommendation:*\n${recId}` },
                { type: "mrkdwn", text: `*Status:*\n${status}` },
            ],
        },
    ];

    if (eventType === "recommendation.accepted" && payload.saving_accepted_by) {
        blocks.push({
            type: "context",
            elements: [{ type: "mrkdwn", text: `Accepted by ${payload.saving_accepted_by}` }],
        });
    }

    if (eventType === "recommendation.rejected") {
        const reason = payload.rejection_reason || "No reason provided";
        blocks.push({
            type: "context",
            elements: [{ type: "mrkdwn", text: `Reason: ${reason}` }],
        });
    }

    if (eventType === "optimization.started" && payload.implementation_method) {
        blocks.push({
            type: "context",
            elements: [{ type: "mrkdwn", text: `Method: ${payload.implementation_method}` }],
        });
    }

    if (eventType === "optimization.failed") {
        blocks.push({
            type: "context",
            elements: [{ type: "mrkdwn", text: ":warning: Optimization failed" }],
        });
    }

    return blocks;
}

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 blocks = buildSlackBlocks(eventType, payload);

    await slack.chat.postMessage({
        channel: SLACK_CHANNEL,
        blocks: blocks,
    });

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

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

Run it the same way, with PORT added:

LEVELFOUR_WEBHOOK_SECRET="whsec_..." \
SLACK_BOT_TOKEN="xoxb-..." \
SLACK_CHANNEL="#cloud-costs" \
PORT=3000 \
npx tsx server.ts

Verify the bytes as they arrived. The signature covers the exact body LevelFour sent, so express.raw has to stay: swap in express.json() and every event fails verification. await request.body() is the FastAPI equivalent.

A 400 from this endpoint means the signature did not verify. Check the whsec_ value first, then whether the delivery sat longer than the verifier's timestamp tolerance. Verify the signature has the headers and the algorithm.

Expose it and register the URL

LevelFour posts to a URL it can reach, so the receiver needs a public one: a cloud function, a container, or an ngrok tunnel while you are testing. Register that URL:

from levelfour import LevelFour

client = LevelFour()
client.webhooks.register(
    url="https://your-domain.com/webhook",
    event_types=[
        "recommendation.accepted",
        "recommendation.rejected",
        "optimization.started",
        "optimization.completed",
        "optimization.failed",
    ],
)

A verified recommendation.accepted event posts a message headed LevelFour: Recommendation Accepted, with the recommendation id and the status in the fields beneath it. That message proves both secrets: the signature verified against LEVELFOUR_WEBHOOK_SECRET, and Slack accepted the post with SLACK_BOT_TOKEN.

This handler posts to Slack inside the request and reads webhook-id only to verify the signature, never to deduplicate on it, so a slow post means a retry and a retry means the same event in the channel twice. Before you rely on it, deduplicate on webhook-id and move the Slack post off the request. See Retries.

Webhooks is the reference: payloads, retries, timestamp tolerance and the signing algorithm
Google Chat integration is the same receiver against a Chat space