Lambdaを使ってSlackに自動でスレッドを立ててくれるアプリを入れよう！

ご挨拶

はじめまして、今年新卒でエンジニアとして入社しました。若林です。
今回からディーネットの技術ブログを書くことにしました。名前だけでも覚えていってください。

初めてこういったブログを書くので拙いところもあるとは思いますが、ぜひ最後まで見ていただけますと嬉しいです。
今回の構築手順はそこそこ噛み砕いてご説明しますので既存知識の部分は随時飛ばしていただければと思います。

自己紹介

まず、簡単に私の自己紹介をさせていただきます。

名前　若林
出身地　埼玉県
趣味　エレキギター、散歩・サイクリング、サウナ
保有資格　AWS CLF-C02
最近の悩み　靴ひもがすぐに解ける
将来の夢　上司の身長を7cmいただく。
エンジニアになろうと思ったきっかけ　中学生のころにPythonとCobraを使ってDiscordのBotを作って完成した時達成感があって楽しかったから

特にエレキギターは今とても旬な時期です。休日は6時間ほど触っています。
(こうして自分を振り返るとインドアなのかアウトドアなのか分からなくなる時がある)

前準備

さて、ここからが本題です。
今回はタイトルの通り、Lambdaを使ってSlackに自動的にスレッドを立てていきます。

まず前準備として以下のものが必要になります。

AWS Lambda　関数
→サーバーレスにコードを実行してくれる。
Amazon EventBridge　スケジューラ
→Lambdaへ「この時間に実行して」って指示してくれる。
Systems Manager　パラメータ
→Lambdaが動くときにSlackの Web hookURLを取得できるようになる。
IAM ロール　許可ポリシー
→Lambdaへ権限を付与(詳しくは後述)
Slack Webhook　URL
→SlackのチャンネルとAWSを繋げるために必要

構成図だと以下のようになります。

非常にシンプルな構成ですので私のような初心者の方でも作れると思います。

Slackテスト用の環境を作成

まず、おそらく皆さん構築するにあたり、使用したいSlackのチャンネルがあると思います。
しかし、いきなりメインのチャンネルで構築を進めてしまいますと、何が起こるか分かりませんし、テストをする際にもスレッドが勝手に立ってしまいます。

それらを防ぐために、テスト用のチャンネルを作成していきます。
まず、ログインしたSlackのホーム画面にチャンネルという項目があると思います。

そこをクリックしていただくと、作成と表示されますので、その中にあるチャンネルを作成するをクリックしてください。

すると作成するチャンネル名を付ける画面に行きますので、そこで任意の名前を付けてください。(今回はchatテストと名付けます)

そのあと次へを押していただくとチャンネルの可視性を決める画面に行くと思います。
今回はテスト用で誰にも見せるわけでもありませんし、社内の人が間違えて入ることを防ぐためにプライベートで作成します。

しかし、今回一人で作成することを前提としてますが、複数人で構築する場合は作成後にユーザーを招待する画面に行くと思いますので、そこでチャンネル招待していただければと思います。

作成が完了しました。
このチャンネルは今回の検証環境です。

続きまして、Slackに自動でメッセージを送信するための土台作りをしていきます。
まず、以下のURLにアクセスをしてください。
おそらくWeb版の初回はログインが必要になると思いますのでご自身のSlackアカウントのメールアドレスでログインしてください。

https://slack.com/services/new/incoming-webhook

ログインが完了すると、Incoming Webhookっていうアプリが出てくると思います。
これは簡単に言えばSlackにスレッドを立ててくれるアプリで、AWS→Slack Webhookを通じてSlack上にスレッドを立ててくれます。

チャンネルへの投稿の横にある選択バーで先ほど作成したチャンネルを選択してIncoming Webhookインテグレーションの追加を押してください。

すると、Webhook URLというのが表示されると思います。

こちらのURLを後程使用しますのでメモ帳アプリなどに残しておきましょう。

SystemsManagerにWebhook URLを保存

ここからはいよいよAWSに触れていきます。

まず、AWSのコンソールにログインをしていただいたら、左上の検索からSystemsManagerと検索してください。
開いていただきますと、SystemsManagerのホームが表示されますので、左側のメニューからアプリケーションツールの中にあるパラメータストアを開いて下さい。

開いたら、右上にあるパラメータの作成をクリックしてください。

開いていただくと、パラメータを作成できる画面に行きます。
そこで任意の名前と必要に応じて説明を記入します。(今回はSlackchat-testという名前にします)

利用枠とタイプとデータ型はデフォルトで問題ございません。
その下にある値という欄に先程取得したSlack WebhookのURLを記入してください。
下の方にあるタグもそのままで大丈夫です。

全て記入が完了したらパラメータを作成してください。
このパラメータも後程利用するので覚えるかメモを取ってください。

IAM ロールを作成

次はIAMロールを作成していきます。

まず、検索欄からIAMを入力して開いてください。
開けましたら、左側のメニューからアクセス管理の中にあるロールを開いてください。

開けましたら、右上にあるロールを作成をクリックしてください。

ロール作成画面まで来たら、信頼されたエンティティを選択と出ますので、その中にあるユースケースからLambdaを選択して次へを押してください。

次にロールにアタッチする許可ポリシーを選択します。
そこで以下のポリシーを追加してください

AWSLambdaBasicExecutionRole
AmazonSSMReadOnlyAccess

詳しい説明は省きますが、AWSLambdaBasicExecutionRoleはCloudWatch Logsへの書き込み権限を付与します。これでCloudWatch Logsへログが出力されるようになり、テスト時や本番の時にエラーが出た場合原因が一目瞭然になります。

AmazonSSMReadOnlyAccessはSystems Managerへの読み取りを許可しますってことです。これにより、先ほど設定したSlack WebhookへLambdaが読み取れるようになります。

では、構築の方に戻ります。

許可ポリシーを追加したら、パラメータを作成した時と同様に任意の名前と必要に応じて説明を入力してください。(今回はSlackroles-testという名前にします)

最後に、選択したエンティティや許可したポリシーの確認をしていきます。
おそらく信頼ポリシーというものの中にコードが書いてあると思います。
EventBridge用に下記のように書き換えれば問題ないかと思います。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com",
                    "scheduler.amazonaws.com"
                ]
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

あとは先程許可したポリシーが表示されているはずですので、問題が無ければロールを作成してください。

Lambdaを作成

続きまして、本日のメイン部分を担うLambdaの構築を進めていきます。

ここがおかしいと全体的に動かなくなるのでしっかり見ながら構築をしていきましょう。

まず、左上の検索欄からLambdaと入力します。
開いていただくと、左側のメニュー欄に関数がありますのでそちらを開きます。

開けたら右上に関数を作成がありますのでクリックしてください。

開けたら、まず基本的な情報を入力していきます。

今回も名前は任意で付けてください。(今回はSlackfn-testという名前にします)
続きまして、ランタイムという項目はPythonを選んでください。

こちらのバージョンは何でも良いですが、上の方に最新サポート対象と出てると思いますのでそこにあるPythonを選びます。

アーキテクチャはx86_64にしてください。

そうしたら下の方にデフォルト実行ロールの変更という項目がありますので、先ほど作ったIAMロール(今回の場合Slackroles-test)を選択してください。

※更に下のその他の構成はデフォルトのままで大丈夫です。

完了しましたら関数の作成を押してください

これでLambdaの作業は完了です。
このあともLambdaは触るのですが、一旦EventBridgeの方を構築します。

EventBridgeでスケジュールを作成

ここからは曜日ごとに自動でスレッドを立てるスケジュール作りを行います。
今回は下記の要件で構築していきます。

平日毎日同じ時間にスレッドを投稿
土日祝日は投稿しない
時間は朝の9時にセット

こちらで進めていきます。

ではまず、左上の検索からEventBridgeと入力します。
開いたら、左側のメニューにあるスケジューラの中にあるスケジュールを開きます。

開けましたら右下にあるスケジュールを作成をクリックしてください。

開いたらスケジュールの作成に進みます。
まず、任意の名前と必要に応じて説明を入力してください(今回はSlackEB-testという名前にします)
スケジュールグループはdefaultで大丈夫です。

次にスケジュールパターンですが、ここは大事な部分ですのでしっかり見てください。
まず頻度を定期的なスケジュールにします。これで毎週同じ頻度で実行されるようになります。
その下のタイムゾーンは(UTC+09:00)Asia/Tokyoを選んでください。こちらは日本で使用するならこちらで問題ございません。

次はスケジュールの種類ですが、こちらはcronベースのスケジュールを選んでください。
選択すると、cron式を入力する項目がありますので以下のように入力してください。

cron(0 9 ? * MON-FRI *)

これで日本時間の平日AM9:00に実行できるようになります。

その下にあるフレックスタイムウィンドウですが、オフが推奨です。
こちらは設定した時間の間にランダムで実行されるものです。

例えば、フレックスタイムウィンドウを15分に設定したら、朝9:00～9:15の間にランダムで実行されるようになります。

ですので、今回の要件には不向きですのでこちらはオフで進めます。

その下の時間枠ですが、これは季節によってスケジュールを変更したいときに利用しますが、今回は1年中朝9時にセットするので必要ないかと思いますので省きます。

これでスケジュールの設定は完了です。
次はこちらのスケジュールを先程構築したLambdaに設定します。

まずターゲットの選択ですべてのAPIを選択します。
そこでサービスを検索する事が出来ますのでLambdaと入力してください。
Lambdaを選択したら、その中にInvokeという欄がありますのでそちらを選択します。

そうしたら先程構築したLamdbaの関数(今回の場合Slackfn-test)を選択します。
バージョン/エイリアスはデフォルトで問題ございません。
その下にあるペイロードも空白で大丈夫です。

その次は詳細設定を一気に進めます。
まずスケジュールの設定ですが、ここは基本同じように設定していただければいいと思います。

スケジュールの状態は、スケジュールの有効化をオンにしてください。
オンにすると作成直後にスケジュールが動きます。構築後に忘れててEventBridgeが動かないってことにならないようにするため、オンが推奨です。

次にスケジュール完了後のアクションですが、必ずNONEにしてください。
DELETEを選択してしまうとせっかくスケジュール設定を作ったのに完了になったら自動的に削除されてしまいます。

次に再試行ポリシーとDLQですが、再試行はオン、DLQはなしで大丈夫です。
再試行はスケジューラがうまく動かない時に再試行をデフォルトで24時間再試行してくれます。DLQもSlackの通知ならそこまで必要ないので今回は無しで大丈夫です。

次に暗号化ですが、これはオフで大丈夫です。
今回の要件なら基本的にはAWSの管理キーだけで問題ございません。

最後、実行ロールは新しく作成してください。ロール名は任意で問題ございません。
先程作った既存ロールに新たに権限付与でもよろしいんですが、新しく自動で作った方が確実です。

一気に書いて長文になってしまったのでまとめます。

スケジュールの状態 → 有効化
完了後アクション → NONE
再試行ポリシー → デフォルト
DLQ → なし
暗号化 → デフォルト
アクセス許可 → 新しいロールの作成

このような状態になってればOKです。

最後に設定した内容を確認出来るのでよく確認してからスケジュールを作成してください。

関数にコードを記述

いよいよラストスパートです。あと少しで完成しますので頑張りましょう。

先程作成したLambdaの関数を開くとコードというものがあります。
この中には作成時に設定したPythonのデフォルトのコードが入っていますので、こちらを以下に書き換えます。

import datetime
import json
import urllib.request
import boto3
import time

def is_holiday_or_weekend():

    today = datetime.date.today()

    # 土日
    if today.weekday() >= 5:
        print(f"[INFO] 今日は週末です: {today}")
        return True

    # 祝日
    try:
        url = f"https://holidays-jp.github.io/api/v1/{today.year}/date.json"
        req = urllib.request.Request(
            url, headers={"User-Agent": "Slackchat-test/1.0"}
        )

        with urllib.request.urlopen(req, timeout=3) as response:
            if response.getcode() == 200:
                holidays = json.loads(response.read().decode("utf-8"))
                if str(today) in holidays:
                    print(f"[INFO] 今日は祝日です: {holidays[str(today)]}")
                    return True
    except Exception as e:
        print(f"[WARN] 祝日API呼び出し失敗（平日扱い）: {e}")

    print(f"[INFO] 今日は平日です: {today}")
    return False

def lambda_handler(event, context):
    """Slackに投稿するメイン処理"""
    start_time = time.time()
    print(f"[INFO] 実行開始: {datetime.datetime.now()}")

    # 祝日・週末チェック
    if is_holiday_or_weekend():
        elapsed = time.time() - start_time
        print(f"[INFO] 投稿スキップ（{elapsed:.2f}秒）")
        return {"statusCode": 200, "body": "Skipped (holiday/weekend)"}

    try:
        # Parameter Store から Slack Webhook URL 取得
        ssm = boto3.client("ssm", region_name="ap-northeast-1")
        response = ssm.get_parameter(Name="Slackchat-test", WithDecryption=True)
        webhook_url = response["Parameter"]["Value"]

        # メッセージ作成
        today = datetime.date.today().strftime("%Y-%m-%d")
        data = {
            "text": f"{today} - 本日分のスレッドです。今日も頑張りましょう！",
        }

        # Slackに送信
        req = urllib.request.Request(
            webhook_url,
            data=json.dumps(data, ensure_ascii=False).encode("utf-8"),
            headers={"Content-Type": "application/json; charset=utf-8"},
        )

        with urllib.request.urlopen(req, timeout=8) as res:
            elapsed = time.time() - start_time
            print(f"[INFO] Slack投稿成功: {res.getcode()} ({elapsed:.2f}秒)")
            return {"statusCode": res.getcode(), "body": "Message sent"}

    except Exception as e:
        elapsed = time.time() - start_time
        print(f"[ERROR] 投稿失敗: {e} ({elapsed:.2f}秒)")
        return {"statusCode": 500, "body": str(e)}

こちらを入力していただき、左側にあるDeployを押してください。
そうすると最初に作ったSlackチャンネルに【このチャンネルに次のインテグレーションを追加しました : incoming-webhook】っていうのが表示されるはずです。

表示されてれば成功です。

しかし、このままではおそらくエラーになると思います。
CloudWatch Logsが正常に動作するかテストもかねて手動で動作確認をしてみます。

手動の動作確認方法は、Lambda関数の中にあるテストという項目がありますので、入るとテストって押せると思います。押してみましょう。

やっぱり失敗しました。ログを確認するためにCloudWatch Logsの中に入ります。

ログにはこう書かれてました。

2025-09-09T16:45:42.612+09:00
INIT_START Runtime Version: python:3.13.v60 Runtime Version ARN: arn:aws:lambda:ap-northeast-1::runtime *********************************************
2025-09-09T16:45:42.946+09:00
START RequestId: e5f754ea-c0b7-4897-a34c-16efbe20bad9 Version: $LATEST
2025-09-09T16:45:42.947+09:00
[INFO] 実行開始: 2025-09-09 07:45:42.947450
2025-09-09T16:45:43.373+09:00
[INFO] 今日は平日です: 2025-09-09
2025-09-09T16:45:45.977+09:00
END RequestId: e5f754ea-c0b7-4897-a34c-16efbe20bad9
2025-09-09T16:45:45.977+09:00
REPORT RequestId: e5f754ea-c0b7-4897-a34c-16efbe20bad9 Duration: 3000.00 ms Billed Duration: 3331 ms Memory Size: 128 MB Max Memory Used: 88 MB Init Duration: 330.32 ms Status: timeout

これを見る限り、Slackへの送信処理がタイムアウトしてます。

原因は、Lambdaのデフォルトタイムアウト時間が3秒なのですが、この3秒間で処理が終わらなかったという事みたいです。

これを解決するにはタイムアウト時間を設定する必要があります。
先程のLambda関数の中に設定がありますので開きます。
次に一般設定の中にタイムアウトという欄がありますので1分に変えてください。

ついでに関数内のコードも少し書き換えましょう。

import datetime
import json
import urllib.request
import boto3
import time

def is_holiday_or_weekend():
    """日本の祝日・週末判定（フェイルセーフ機能付き）"""
    today = datetime.date.today()

    # 土日
    if today.weekday() >= 5:
        print(f"[INFO] 今日は週末です: {today}")
        return True

    # 祝日
    try:
        url = f"https://holidays-jp.github.io/api/v1/{today.year}/date.json"
        req = urllib.request.Request(
            url, headers={"User-Agent": "Slackchat-test/1.0"}
        )

        with urllib.request.urlopen(req, timeout=5) as response:
            if response.getcode() == 200:
                holidays = json.loads(response.read().decode("utf-8"))
                if str(today) in holidays:
                    print(f"[INFO] 今日は祝日です: {holidays[str(today)]}")
                    return True
    except Exception as e:
        print(f"[WARN] 祝日API呼び出し失敗（平日扱い）: {e}")

    print(f"[INFO] 今日は平日です: {today}")
    return False

def post_to_slack_with_retry(webhook_url, data, max_retries=3):
    """リトライ機能付きSlack投稿"""
    for attempt in range(max_retries):
        try:
            print(f"[INFO] Slack投稿試行 {attempt + 1}/{max_retries}")

            req = urllib.request.Request(
                webhook_url,
                data=json.dumps(data, ensure_ascii=False).encode("utf-8"),
                headers={"Content-Type": "application/json; charset=utf-8"},
            )

            timeout = 10 + (attempt * 5)
            print(f"[DEBUG] タイムアウト設定: {timeout}秒")

            with urllib.request.urlopen(req, timeout=timeout) as res:
                print(f"[INFO] Slack投稿成功: {res.getcode()} (試行: {attempt + 1}回)")
                return {
                    "success": True,
                    "status_code": res.getcode(),
                    "attempts": attempt + 1
                }

        except urllib.error.HTTPError as e:
            # HTTPエラーは即座に失敗
            print(f"[ERROR] HTTP エラー: {e.code} {e.reason}")
            return {
                "success": False,
                "error": f"HTTP {e.code}: {e.reason}",
                "attempts": attempt + 1
            }

        except (urllib.error.URLError, OSError) as e:
            # ネットワークエラー、タイムアウトはリトライ対象
            print(f"[WARN] 試行 {attempt + 1} 失敗: {e}")

            if attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"[INFO] {wait_time}秒後にリトライします...")
                time.sleep(wait_time)
            else:
                print(f"[ERROR] 全ての試行が失敗しました（{max_retries}回）")
                return {
                    "success": False,
                    "error": str(e),
                    "attempts": max_retries
                }

    return {
        "success": False,
        "error": "予期しないエラー",
        "attempts": max_retries
    }

def lambda_handler(event, context):
    """Slackに投稿するメイン処理"""
    start_time = time.time()
    print(f"[INFO] 実行開始: {datetime.datetime.now()}")

    # Lambda実行時間の監視
    remaining_time = context.get_remaining_time_in_millis() / 1000
    print(f"[DEBUG] Lambda残り時間: {remaining_time:.1f}秒")

    # 祝日・週末チェック
    if is_holiday_or_weekend():
        elapsed = time.time() - start_time
        print(f"[INFO] 投稿スキップ（{elapsed:.2f}秒）")
        return {"statusCode": 200, "body": "Skipped (holiday/weekend)"}

    try:
        # Parameter Store から Slack Webhook URL 取得
        print("[INFO] Parameter Store から URL取得開始")

        # boto3のタイムアウト設定
        config = boto3.session.Config(
            retries={'max_attempts': 2},
            read_timeout=10,
            connect_timeout=5
        )
        ssm = boto3.client("ssm", region_name="ap-northeast-1", config=config)

        response = ssm.get_parameter(Name="Slackchat-test", WithDecryption=True)
        webhook_url = response["Parameter"]["Value"]
        print("[INFO] Parameter Store から URL取得成功")

        # メッセージ作成
        today = datetime.date.today().strftime("%Y-%m-%d")
        data = {
            "text": f"{today} - 本日分のスレッドです。今日も頑張りましょう！",
        }

        # 残り時間チェック
        remaining_time = context.get_remaining_time_in_millis() / 1000
        if remaining_time < 30:
            print(f"[WARN] Lambda残り時間が少ない: {remaining_time:.1f}秒")

        # リトライ機能付きSlack投稿
        result = post_to_slack_with_retry(webhook_url, data)

        elapsed = time.time() - start_time

        if result["success"]:
            print(f"[INFO] 投稿完了: {result['status_code']} ({elapsed:.2f}秒, {result['attempts']}回試行)")
            return {
                "statusCode": result["status_code"], 
                "body": f"Message sent (attempts: {result['attempts']})"
            }
        else:
            print(f"[ERROR] 投稿失敗: {result['error']} ({elapsed:.2f}秒, {result['attempts']}回試行)")
            return {
                "statusCode": 500, 
                "body": f"Failed after {result['attempts']} attempts: {result['error']}"
            }

    except Exception as e:
        elapsed = time.time() - start_time
        print(f"[ERROR] 予期しないエラー: {e} ({elapsed:.2f}秒)")
        return {"statusCode": 500, "body": f"Unexpected error: {str(e)}"}

書き換えたら先ほどと同じ手順です。

Deploy→テストを実行してみましょう。

今度は実行に成功しました。

Slackの方も見てみましょう。

無事成功していました。
これでスレッドを立てることが出来ます。

あとは平日の朝9時まで待ってみてチャンネル内を見てみましょう。上手くいってれば自動的に投稿されているはずです。

もしすぐに検証したいとのことでしたら、先程スケジュールで設定したcron式を直近の時間に変更すれば確認できます。
例えば現在の時刻が朝の10時だとしたら、その10分後に検証するためには以下に設定してください。

cron(10 10 ? * MON-FRI *)

こうすれば朝の10:10に実行処理がされますので是非お試しください。
※検証後に元の時間に戻すことをお忘れなく。

完全自動で投稿が出来たのを確認したら、最初のSlack Webhookで導入したいチャンネルのURLをコピーして、パラメータの値を変更してください。そうすればテスト用のチャンネルを抜けて本番環境へ移行されます。

おまけ

アプリを入れられたけれども、名前とアイコンがデフォルトだと少し味気ないと感じると思います。
そんな時は名前とアイコンを変えてみましょう。

まずアイコンにしたい画像のURLを用意します。
今回はいらすとやのル〇ィーを使用します。

アイコンに適した画像を用意したらLambda関数内のコードから以下の行を探してください。

# メッセージ作成
        today = datetime.date.today().strftime("%Y-%m-%d")
        data = {
            "text": f"{today} - 本日分のスレッドです。今日も頑張りましょう！",
        }

次に名前を決めていきます。
(思いつかなかったので)テストアプリ名前変更中にします

名前が決まったら以下の行に追記していきます。

 # メッセージ作成
        today = datetime.date.today().strftime("%Y-%m-%d")
        data = {
            "text": f"{today} - 本日分のスレッドです。今日も頑張りましょう！",
            "username": "テストアプリ名前変更中",
            "icon_url": "https://blogger.googleusercontent.com/img/b/R29vZ2xl/AVvXsEh-Y7rgTcW5NdDkxvwMW4Gdj2Q3G3lZVBvHHC10A3T_Iwxj0257NbTbdhvWKFOqn7nxXw6-V4P_0VFuJZ_5cQSDPxlazFKTD9N-d1A0IrX0k7LoaVpG3X9IwQ48H0zfXTJOT1JntRr0Lq3o/s1048/onepiece01_luffy.png"

無事に変わっていることが確認できました。

職場で使用する際はチャンネル内の仲間と一緒に考えて名前やアイコンも面白くするのも楽しいかもしれません。

最後に

いかがだったでしょうか。

初めて完全一人で構成を考えるところからAWS環境構築だったので成功した時はとてもうれしかったです。

一つ注意点なのですが、途中で入れた祝日の部分ですが、今年のものを判定するので年末になったら来年のものに反映する必要があります。そこだけ手動ですのでご注意ください。

ここまで作っては何ですが、需要はあるかって言われたらそこまで無いかもしれません。

しかしAWSにまだ慣れていない私でも数時間ほどで構築できたので初心者の方には良い練習になりますし、料金も無料枠で収まるように考えたので気軽に出来る思います。

またこういったブログを上げていきますので見つけたら覗きに来て下さい。
それではまたどこかでお会いしましょう。

Wakabayashi.dayo

文系エンジニア見習いです。

ご挨拶