概要

表題のとおりのSlackアプリケーションを作成する。
まともに書くと長文になるので、以下の要点ポイントのみを重点的に記載する。

Slackアプリケーションの留意点
Slack側のアプリケーション設定（権限周り）
SlackのHomeView（Homeタブ）やattachmentに関して
Lambda（python）周りの実装とライブラリ利用
APIGateWay設定

なお、シンプルに試すだけならWebHookURLを設定してLambdaとかをスケジュール実行してPOSTすれば終わるが、SlackClientライブラリを利用したインタラクティブなリクエストに対するメッセージ通知ができるようなシーンを想定している。

手順通りにやれば動くものが作れるというよりは、
調べるのが面倒だったり、案外ハマったポイントを断片的に紹介する点はご承知おきください。

Slackアプリケーションを作成するにあたっての留意点

レスポンスは3秒以内に一旦返す

ボタンをクリックし、レスポンスを返す場合などは３秒を超えるとエラーになる。

これは、処理でひとまずHTTPコード200を返して、
その後非同期で処理を行い、改めて結果をチャンネルに通知することで簡単に回避できる。

なお、本記事では深く触れないが
実際はリクエストパラメータのresponse_urlは3秒経過しても数分以上有効なので
タイムアウト覚悟のまま、response_urlに直接POSTすることもできなくはない。

UI（ホームタブ、フォーム）やメッセージの装飾にはBlock Kitと呼ばれるJSONの組み立てが必要

Block Kit Builderで試せるのだが、非常に面倒。

最低限、チャンネルIDとユーザIDを理解する

ユーザIDはrequestBodyのPayLoadに入ってる。ユーザ名ではないので注意。
チャンネルIDもrequestBodyのPayLoadにあるが、特定のパターンでは入ってこないこともあるし
別チャンネルへの通知を行いたい場合もあると思うので、対象のIDを、定数で埋め込んでおくのが良いだろう。

なお、チャンネルIDはブラウザで対象チャンネルを開くとアドレスから簡単に判別できる。

/app.slack.com/client/<ここは組織ID>/<ここがチャンネルID>/details/top

メッセージ装飾のマークアップが、テーブル（表）をサポートしていない

できないものは仕方ないですね。
私はdividerと引用マークアップとスペース埋めを駆使して強引に整形。。

Slack側のアプリケーション設定（権限周り）

これだけセットしときゃ大丈夫ってところだけ。意外に限られてる。
今回はWebHookURLは使わないが、一応設定入れる。

設定後は
[Settings] -> [Installed App Settings] -> [(Re)Install to WorkSpace]
での反映を忘れずに。

表示情報

[Settings] -> [Basic Infomation] -> [Display Information]
アプリ名とか説明とか、ロゴとか。適当に。

WebHookURL

[Features] -> [Incoming Webhooks]

[Activate Incoming Webhooks]をオンにする
[Webhook URLs for Your Workspace]で[Add New Webhook to Workspace]で追加

認証と許可

[Settings] -> [OAuth & Permissions]

[Bot User OAuth Access Token]

ここの値はlambdaからメッセージなどを送るのに必要になるので控えておく

[Scope] -> [Bot Token Scopes]

必須なのは
chat:write, chat:write.customize, channels:history くらいだと思う。
WebHook使うなら、incoming-webhookも。
私は他にapp_mentions:read, commands, reactions:read とか入れてる。

イベントの活性化とリクエストURLの設定

[Features] -> [Event Subscriptions]

APIを作成して、GateWayを設定し終わったら（方法は後述）
最後にここの[Enable Events]をオンにして、各種設定を入れる。

ここに作成したAPIのURL等を設定することで
実際に特定のアクションを実施した時にリクエストが飛ぶようにしておく。

リクエスト（API）URLの設定

現段階ではできないので、後で設定したら。暫定で適当なURLを入れても認証でNGになる。
最低限、API側の処理でチャレンジレスポンスが実装されていないと認証が通らないので。

最後にAPIのURLを入れておく場所と認識しておいてください。

どのイベントをHookするかの設定

[Subscribe to bot events]で設定する。

とりあえず、ホームタブのUI表示用に app_home_opened と
メッセージからのボタンイベントをとるために message.channels を追加しておけば大体OK。

SlackアプリケーションのUIやメッセージ装飾に関して

以下に簡単な例とイメージを記載。
コードでの実装イメージはSlackClientAPIの利用サンプルとともに後述する。

ホームタブ

HomeTabSample.json

{
	"type": "home",
	"blocks": [
		{
			"type": "header",
			"text": {
				"type": "plain_text",
				"text": "これは超シンプルなアプリテストです。"
			}
		},
		{
			"type": "divider"
		},
		{
			"type": "actions",
			"elements": [
				{
					"type": "button",
					"text": {
						"type": "plain_text",
						"text": "現在の時刻を表示"
					},
					"style": "primary",
					"value": "click_from_home_tab"
				}
			]
		}
	]
}

ボタン付きメッセージ

MessageSample.json

{
	"blocks": [
		{
			"type": "section",
			"text": {
				"type": "mrkdwn",
				"text": "こんにちは、サンプルアプリケーションです！"
			}
		},
		{
			"type": "actions",
			"elements": [
				{
					"type": "button",
					"text": {
						"type": "plain_text",
						"emoji": true,
						"text": "現在時刻を教えて"
					},
					"style": "primary",
					"value": "click_from_message_button"
				}
			]
		}
	]
}

実際のAPIをlambdaで構築（Python）

ポイントだけ紹介。
ちなみにログ関係はAPIGatewayのログで十分なので、lambda側では特に設定してない。

必要な構成（非同期実装のため）

Slackアプリケーション（前述）
↓↑
AWS APIGateway（後述）
↓↑
lambda slack-request.py API（ここで説明)
　> 3秒ルールがあるため、処理を非同期lambdaに委譲し一旦レスポンスを返す
　> payloadの解析などはここでほぼ行う
↓↑
lambda async-notice.py API（ここで説明)
　> 依頼内容や属性情報を全てもらった状態で処理を行い結果をSlackへ通知
　> 今回はスケジュール実行の例もここで

必要なライブラリの準備（Layer設定）

地味にてこずる部分。ハマると数時間たったりするという。
毎回忘れた頃にlambdaを作るので、
pythonフォルダで包むの忘れてアップロードして動かなくて戸惑うという。

まあ、これをアップロードして、両方のlambdaのlayerに追加する。
必要なのはslack-clientだけのような気もする（作り方は別でググってください）。

あとは今回はいらないけど、いろいろやるならついでにdjangoやpandasの最新版も入れておくといい感じ。
お試し版では上記のslackレイヤーだけあれば大丈夫。

実装のポイント：リクエスト処理のlambda

チャレンジ・レスポンスの実装
リクエストURLは最低でも、以下の実装をど先頭に入れる。
これがないと、アプリケーション設定で認証されない。

slack-request.py

    if "challenge" in event:
        return event["challenge"]

通知はSlackClientAPIで行う

SlackClientの生成

# アプリケーション設定のBot User OAuth Access Tokenの値
import slack
client = slack.WebClient(token="xoxb-XXXXXXXXXX....")

ホームタブのUI描画
JSONで指定する

ホームタブの描画を行う

    """
        HomeTabメニューの表示
    """
    if ('event' in event):
        slack_event = event['event']
        if (slack_event['type'] == 'app_home_opened') :
            if ('tab' in slack_event) and (slack_event['tab'] == 'home'):
                user = slack_event['user']
                channel = slack_event['channel']
                blocks = <ここがJSON>
                views = {"type": "home", "blocks": blocks}
                client.views_publish(user_id=user, view=views)
                return {'statusCode': 200}

ホームタブのJSON(再掲)

ホームタブのJSON

{
	"type": "home",
	"blocks": [
		{
			"type": "header",
			"text": {
				"type": "plain_text",
				"text": "これは超シンプルなアプリテストです。"
			}
		},
		{
			"type": "divider"
		},
		{
			"type": "actions",
			"elements": [
				{
					"type": "button",
					"text": {
						"type": "plain_text",
						"text": "現在の時刻を表示"
					},
					"style": "primary",
					"value": "click_from_home_tab"
				}
			]
		}
	]
}

イベントハンドル（ホームボタンやインタラクティブメッセージボタンの処理）
厳密には振り分けもう少し整理した方が良いが、これでも動く。

最大のポイントは、
リクエストは非同期で別のlambdaに依頼して先にリクエストを受け付けましたレスポンスを返すこと。

また、通知の度に他のSlackユーザに応答を返したくない場合もあるのでサンプルでは
メッセージ通知を「あなただけに表示されます」のEphemeralにしている。

slack-request.py

    """
        ActionをHookして処理を振り分け、SlackへのResponseを返す
        ChannelIdは固定不変なのでリクエストからではなく定数を利用する
    """
    if ('body-json' in event):
        
        # payloadの内容を抽出
        action_message = urllib.parse.unquote(event['body-json'].split('=')[1])
        action_message = json.loads(action_message)

        # 通知先となるユーザIDを取得        
        user_id = action_message['user']['id']

        # Message Button == Interactive Messageか否かを判定
		isInteractiveMessage　= False
        if('type' in action_message ) and (action_message['type'] == 'interactive_message'):
            isInteractiveMessage = True

        # actionがあればイベントをHook
        if ('actions' in action_message):

            """
                値の取得と通知依頼は別のlambdaに非同期で依頼する
                非同期依頼は複数あって良いが、依頼は主にPrimaryModeのみで判定する
            """
            delegateLambda_primaryMode = []
            
            # payloadからaction message部分のみを抽出する
            action_message = action_message['actions'][0]
            # 非同期送信なので同期レスポンスメッセージのDefaultを先に設定しておく
            send_message = "リクエストを受け付けました。少しお待ちください。"
            
            if (action_message['type'] == 'button') :

                if (action_message['value'] == 'click_from_home_tab'):
                    """
                        ホームタブのボタンがクリックされた
                    """
                    delegateLambda_primaryMode.append('click_from_home_tab')
					
                elif (action_message['value'] == 'click_from_message_button'):
                    """
                        ホームタブのボタンがクリックされた
                    """
                    delegateLambda_primaryMode.append('click_from_message_button')

                
                """
                    設定した非同期依頼をかける
					PrimaryとSecondaryのほか、user_idを渡す仕様にした
					非同期処理は後述する
                """
                for p in delegateLambda_primaryMode:
                    input_event = {
                        "primaryMode": p,
                        "secondaryMode": "now no mean",
                        "user_id": user_id
                    }
                    lambdaMediator.callNortifyAsync(json.dumps(input_event))
                
                """
                    一旦、応答メッセージを返却する
                """
                if isSendEphemeral == True:
                    response = client.chat_postEphemeral(
                        channel=notice_channel_id,
                        text=send_message,
                        user=user_id
                    )
                else:
                    response = client.chat_postMessage(
                        channel=notice_channel_id,
                        text=send_message
                    )

    if isInteractiveMessage == True:
        pass
    else:
        return {'statusCode': 200}

lambdaMediator.py

import boto3

def callNortifyAsync(payload):
    """
    非同期で委譲先のlambda(async-notice.py固定)を呼ぶ

    Parameters
    ----------
    event : 委譲先lambdaのevent json
        PrimaryMode: 依頼内容を表す第一Key文字列
        SecondaryMode: 依頼内容を表す第二Key文字列
        UserId: 依頼者のSlackユーザID

    Returns
    ----------
    response
    """
    
    response = boto3.client('lambda').invoke(
        FunctionName='async-notice',
        InvocationType='Event', 
        Payload=payload
    )
    return response

実装のポイント：非同期処理のlambda

attachmentは利用しなくてもいいが、凝ったことをやりたい場合は
メッセージではなくattachmentでやるので、サンプルはそうしている。

今回は両方のlambdaの実行RoleにAWSLambda_FullAccessを付与している。
また、サンプルなのでPrimaryModeをslack-request.pyからもらっているが分岐は入れていない。

attachmentを利用してメッセージを送信

async-notice.py

import json
import datetime
import time
import boto3
import slack

from libs import mylib
from libs import slack_info

def lambda_handler(event, context):
    """
    このlambdaはslack-requestからのdelegate処理（非同期）、およびスケジューラからのみKickされる
    
    Parameters
    ----------
    event : 
        PrimaryMode: 依頼内容を表す第一Key文字列
        SecondaryMode: 依頼内容を表す第二Key文字列（未利用：拡張用）
        UserId: 依頼者のSlackユーザID
        
    Returns
    ----------
    httpResponseBody : statusCode:200 , body -> SlackClientからの送信を行うため200を返すだけ
    
    ToDo
    ----------

    """
    
	# ユーザIDの取得
    user_id = ""
    if 'user_id' in event:
        user_id = event['user_id']
		
    """
        パラメータなしはスケジューラからの起動なので全体通知
		パラメータありは非同期依頼なのでEphemeral(あなただけに見えてます)通知
    """
	isEphemeral = True
    if ('primaryMode' in event) == False:
        isEphemeral = False
        
    """
        メッセージ通知を行う
		attachmentsはメッセージJSON
    """
    postDataMessage(attachments, False, isEphemeral, user_id, "ここに本文メッセージを入れる。")

    return {
        'statusCode': 200,
        'body': json.dumps('OK')
    }
    
    
def postDataMessage(attachment_data, isSendMultiple=False, isEphemeral=True, userId="", textMessage=""):
    """
    メッセージを送信する
    
    Parameters
    ----------
    attachment : アタッチメントデータ
    isSendMultiple : アタッチメントデータが複数あるか否か
    isEphemeral : リクエストユーザのみ見えるように送信するか否か
    userId : リクエストユーザのID
    textMessage : テキストメッセージ

    Returns
    ----------

    ToDo
    ----------

    """
    
    client = slack.WebClient(token="xoxb-XXXXXXXXX.....")

    # 複数送信対応のため、単一送信であってもリストで包み直す
    if isSendMultiple == False:
        attachment_data = [attachment_data]
    
	# TARGET_CHANNEL_IDは通知先のチャネルIDを別途設定しておくこと
    for attachment in attachment_data:
        if isEphemeral == True:
            response = client.chat_postEphemeral(
                channel=TARGET_CHANNEL_ID,
                attachments = attachment,
                user=userId
            ) 
        else:
            response = client.chat_postMessage(
                text=textMessage,
                channel=TARGET_CHANNEL_ID,
                attachments = attachment
            )

ボタン付きメッセージのJSON（再掲）

attachmentのjson例

{
	"type": "home",
	"blocks": [
		{
			"type": "header",
			"text": {
				"type": "plain_text",
				"text": "これは超シンプルなアプリテストです。"
			}
		},
		{
			"type": "divider"
		},
		{
			"type": "actions",
			"elements": [
				{
					"type": "button",
					"text": {
						"type": "plain_text",
						"text": "現在の時刻を表示"
					},
					"style": "primary",
					"value": "click_from_home_tab"
				}
			]
		}
	]
}

おまけ：スケジュール実行したい場合

Amazon EventBridgeのルール作成で対象をasync-noticeのlambdaとcron設定をブラウザから入れるだけ。

月曜から金曜の9時に実行（GMT時間なので-9時間で設定）

cron(0 0 ? * MON-FRI *)

おまけ : RedShiftへの接続

boto3でredshift-dataを指定するだけで
データアクセスはクエリ実行時に指定したRedshiftアカウント内でのロール権限で実施されるため非常に楽。

サンプルコード

RedShift接続

import json
import time
import boto3
        
# Redshift接続情報
CLUSTER_NAME='cluster名を入れる'
DATABASE_NAME='データベース名を入れる'
DB_USER='ユーザ名を入れる'

def getDateTimeSample():
    """
    時刻の取得
    """

    sql = '''
		select getdate();
	'''
    return _getData(sql)

def _getData(sql):
    """
    RedShiftへクエリを発行して結果をそのまま返す
    
    Parameters
    ----------
    String : sql文

    Returns
    ----------
    statement : boto3の取得結果そのまま
    """
    
    # Redshiftにクエリを投げる。非同期なのですぐ返ってくる
    data_client = boto3.client('redshift-data')
    result = data_client.execute_statement(
        ClusterIdentifier=CLUSTER_NAME,
        Database=DATABASE_NAME,
        DbUser=DB_USER,
        Sql=sql,
    )

    # 実行IDを取得
    id = result['Id']

    # クエリが終わるのを待つ
    statement = ''
    status = ''
    while status != 'FINISHED' and status != 'FAILED' and status != 'ABORTED':
        statement = data_client.describe_statement(Id=id)
        #print(statement)
        status = statement['Status']
        time.sleep(1)

    # 結果の表示
    if status == 'FINISHED':
        if int(statement['ResultSize']) > 0:
            # select文等なら戻り値を表示
            statement = data_client.get_statement_result(Id=id)
        else:
            # 戻り値がないものはFINISHだけ出力して終わり
            print('QUERY FINSHED')
    elif status == 'FAILED':
        # 失敗時
        print('QUERY FAILED\n{}'.format(statement))
    elif status == 'ABORTED':
        # ユーザによる停止時
        print('QUERY ABORTED: The query run was stopped by the user.') 
    
    return statement

APIGateWayの設定

作成したlambda APIにURLを与えてアクセスできるようにする。

大まかな手順

REST APIで作成

lambdaを統合

マッピングテンプレートでapplication/x-www-form-urlencodedの追加
一番重要。ホームタブのオープンイベントではこのフォーマットで来るので、ここいれないと動作しない。
また、他のリクエストはapplication/jsonで来るものがあるので、合わせて追記するのもポイント。

あとはステージをデプロイするだけ

ログ設定
この設定で十分デバッグできる。
ログはCloudWatchにステージ名のロググループで出力される。

その他のポイント・気付き

3秒ルールはしんどい

Slackの性質を考えればまあ、そうなるのは理解できるが凝ったこともしたくなる場合もある。

非同期でデータベースアクセス含めた処理を行って
単純にレスポンス200を返すだけでも、lambdaのコールドスタートなどを考えるとギリギリ間に合わない場合も稀にある。

lambdaにはProvisioned Concurrencyもあり、料金はかかるが場合によってはこれも活用できる。

マークアップにテーブルが欲しい

結構需要はあると思うが。
引用を使うと引用内では半角スペースを忠実に再現するのでこれと、division線を使うと表っぽくはなる。
ただ、無理やり感は拭えない。