AWS Lambdaを使ってSlack通知ツールを作ってみた

2023年1月20日掲載

サーバレスサービスは非常に便利です。
仮想コンピューティングを準備しなくても、開発者は好きな言語による関数コードを綴るだけで、サーバレスがHTTP Requestやタイマートリガー、APIアクセス等に応じて自動処理し、その結果を返答するため、仮想コンピューティングのランニングコストを削減しつつ、様々な組み合わせやシナリオを実現することができます。

今回はそのシナリオの一つとしてAWSのサーバレスプロダクトサービス AWS Lambda がタイマートリガーをベースに外部からJSONデータを取得、その結果をSlackへ通知する仕組みを作ってみます。

1. 全体構成図

全体構成図は次の通りになります。ゴールは一定時刻になると天気予報を通知するものです。

2. 参考：Slack - サーバレスでできることの例

上記は非常に簡単な例ですが、この流れの基本を押さえておけば、次のようなシナリオを実現することができます。

Stable Diffusionを使って、チャットでBotが画像を生成して返却するサービス

レシートをGoogle スプレッドシートの家計簿へ自動反映の最中に異常検知アラート発信

クラウドサービス等の料金通知

・参考リンク
　GitHub: Node.js 14 以降向けの AWS 当月利用料金 Slack 通知スクリプト

Slackは対話型チャットベースでさまざまなサービスと連携できるため、上記を含め、さまざまなシナリオを簡単に実現することができます。コツは、Slack入力を検知するPythonスクリプト、およびSlackへ通知するPythonスクリプトを上手く生かすことです。今回はその一歩として、サーバレスからPythonスクリプトによって通知する簡単な仕組みを作ってみます。

3. Slackの設定

今回はSlack上にBotを作成しますので、Slackアカウント、利用したいサービスのワークスペース、およびslack連携には Slack AppとIncoming webhook URL の発行が必要です。

もし、Slackアカウント、利用したいサービスのワークスペース、Slack AppとIncoming webhook URLを持っている場合は、「4. Lambdaサービスの準備」へお進みください。

3-1. Slackアカウント＆ワークスペース作成

Slackアカウントを準備します。

https://slack.com/

Slackアカウントの準備ができたら以下のリンクを開き、ワークスペースを作成します。

https://slack.com/get-started#/createnew

今回はワークスペース名を「ServerlessToSlack」という名前にします。

表示された指示に従い、ワークスペースを作成し、起動します。

チャンネルを作成します。チャンネル名は何でも良いので、今回は「weatherchannel」にします。

3-2. Slackアプリ設定

ワークスペースが完成したら、今度はそのワークスペースとFunctionComputeを連携するための、Slackアプリ設定をします。

https://api.slack.com/apps にアクセスし、「Create an APP」をクリックしてSlackアプリを作成します。

「From scratch」をクリックします。

AppNameを設定し、ワークスペースを選択します。

「Incoming Webhooks」を選択します。

「Incoming Webhooks」をオンにします。

「Incoming Webhooks」をオンにすると、メニューの表示内容が変わります。

Webhooksをワークスペースに追加します。

チャンネルを選択し「許可する」ボタンをクリックして許可します。

Incoming Webhooksを保存します。これでSlack App設定は完了です。

最後に、Slack App設定にてWebhook URLがありますが、こちらはサーバレス側から通知するために必要なURLなので手元に置くようにします。

4. Lambdaサービスの準備

Slack側の準備は完了したので、今度はサーバレス側の準備を進めます。AWSコンソール画面からLambdaを開き、ウィザードでNode.jsをデプロイする専用のFunctionリソースを新規で作成します。

流れとして、①「Create Function」で新規Functionを立ち上げ、②Function名を入力。ランタイムを「Python」として選定します。ランタイムはコードを実行するために必要な言語設定等環境のことを指します。③Function立ち上げに必要な情報の入力が完了すれば「Create Function」ボタンを押すと、Functionが作成されます。以降はこのFunctionという名のリソースでさまざまな関数やコードをデプロイすることができます。

この設定が順調であれば、AWS Lambdaコンソール画面のモーダルウィンドウにて、Lambdaで実行したいコード関数を入力することができます。新規作成直後はデフォルトのサンプルコードで表示されます。

5. Slackと連携するPythonコード関数作成

初期状態として「Hello world」を返却するサンプルの関数ソースコードがありますが、これを全部消して、次の関数ソースコードを入力します。
Webhook URLは上記slack app設定で入手したWebhook URLを使用します。

import base64
import urllib3
import json
 
http = urllib3.PoolManager()
 
slack_endpoint = "Slack App設定から入手したWebhook URL"
weather_endpoint = "https://www.jma.go.jp/bosai/forecast/data/overview_forecast/130000.json"
 
 
def lambda_handler(event, context):
    # get weather result
    weather_result = get_weather()
    if weather_result["success"]:
        # compose weather info to markdown message
        weather_msg = compose_weather_message(weather_result["data"])
        forward_slack_message(weather_msg)
    else:
        weather_msg = weather_result["error_message"]
    forward_slack_message(weather_msg)
 
 
def forward_slack_message(message_content):
    msg = {
        "channel": "#weatherchannel",
        "username": "nancy",
        "text": message_content,
        "icon_emoji": ""
    }
    encoded_msg = json.dumps(msg).encode('utf-8')
    resp = http.request('POST', slack_endpoint, body=encoded_msg)
    # print({
    #     "message": "test message: {0}".format(message_content),
    #     "status_code": resp.status,
    #     "response": resp.data
    # })
 
 
def compose_weather_message(weather):
    return "*{0}  _{1}_  {2} 地域の天気情報です*\n{3}\n{4}".format(
        weather['publishingOffice'],
        weather['reportDatetime'],
        weather['targetArea'],
        weather['headlineText'],
        weather['text']
    )
 
 
def get_weather():
    result = {"success": True, "error_message": "OK"}
    try:
        resp = http.request(method="GET", url=weather_endpoint, headers={"Content-Type": "application/json"})
        if resp.status == 200 and resp.data:
            weather = json.loads(resp.data)
            # print(weather)
            result["data"] = weather
        else:
            msg = "Failed get weather, response code: {0}, response: {1}".format(resp.status, resp)
            print(msg)
            result["success"] = False
            result["error_message"] = msg
        return result
    except Exception as e:
        result["success"] = False
        result["error_message"] = str(e)
        print("Failed getting weather information.")
    return result

コード入力が完了したら、最後に「Deploy」ボタンをクリックします。Delopyボタンを押さない限り、Web上で入力したPythonコードは反映されません。