Slack Integration

Overview

This guide shows how to build a webhook receiver that verifies LevelFour event signatures and posts formatted messages to Slack. Examples are provided in Python (FastAPI) and TypeScript (Express).

Prerequisites

A Slack app with the chat:write scope (create one here)
The Bot User OAuth Token (xoxb-...) from your Slack app
Your LevelFour webhook signing secret (whsec_...)
The Slack bot invited to the target channel

Environment Variables

Variable	Description
`LEVELFOUR_WEBHOOK_SECRET`	Webhook signing secret from LevelFour (`whsec_...`)
`SLACK_BOT_TOKEN`	Slack Bot User OAuth Token (`xoxb-...`)
`SLACK_CHANNEL`	Target Slack channel (default: `#cloud-costs`)

Event Types

All five LevelFour webhook events are handled:

Event	Slack Message
`recommendation.accepted`	Shows recommendation ID, status, who accepted
`recommendation.rejected`	Shows recommendation ID, status, rejection reason
`optimization.started`	Shows recommendation ID, implementation method
`optimization.completed`	Shows recommendation ID, final status
`optimization.failed`	Shows recommendation ID with warning

Python (FastAPI)

Dependencies

pip install levelfour fastapi uvicorn httpx

Server

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

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

TypeScript (Express)

Dependencies

npm install levelfour express @slack/web-api

Server

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

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

Deployment

Expose your webhook endpoint publicly (e.g., via a cloud function, container, or tunnel like ngrok for testing). Then register it with LevelFour:

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",
    ],
)

Slack Integration

On this page