メッセージのリスニング

あなたのアプリがアクセス権限を持つメッセージの投稿イベントをリッスンするには message() メソッドを利用します。このメソッドは typemessage ではないイベントを処理対象から除外します。

message() の引数には str 型または re.Pattern オブジェクトを指定できます。この条件のパターンに一致しないメッセージは除外されます。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
# '👋' が含まれるすべてのメッセージに一致
@app.message(":wave:")
def say_hello(message, say):
    user = message['user']
    say(f"Hi there, <@{user}>!")

正規表現パターンの利用

文字列の代わりに re.compile() メソッドを使用すれば、より細やかな条件指定ができます。

1
2
3
4
5
6
7
import re

@app.message(re.compile("(hi|hello|hey)"))
def say_hello_regex(say, context):
    # 正規表現のマッチ結果は context.matches に設定される
    greeting = context['matches'][0]
    say(f"{greeting}, how are you?")

メッセージの送信

リスナー関数内では、関連づけられた会話(例:リスナー実行のトリガーとなったイベントまたはアクションの発生元の会話)がある場合はいつでも say() を使用できます。say() には文字列または JSON ペイロードを指定できます。文字列の場合、送信できるのはテキストベースの単純なメッセージです。より複雑なメッセージを送信するには JSON ペイロードを指定します。指定したメッセージのペイロードは、関連づけられた会話内のメッセージとして送信されます。

リスナー関数の外でメッセージを送信したい場合や、より高度な処理(特定のエラーの処理など)を実行したい場合は、Bolt インスタンスにアタッチされたクライアントclient.chat_postMessage を呼び出します。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
# 'knock knock' が含まれるメッセージをリッスンし、イタリック体で 'Who's there?' と返信
@app.message("knock knock")
def ask_who(message, say):
    say("_Who's there?_")

ブロックを用いたメッセージの送信

say() は、より複雑なメッセージペイロードを受け付けるので、メッセージに機能やリッチな構造を与えることが容易です。

リッチなメッセージレイアウトをアプリに追加する方法については、API サイトのガイドを参照してください。また、Block Kit ビルダーの一般的なアプリフローのテンプレートも見てみてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# ユーザーが 📅 のリアクションをつけたら、日付ピッカーのついた section ブロックを送信
@app.event("reaction_added")
def show_datepicker(event, say):
    reaction = event["reaction"]
    if reaction == "calendar":
        blocks = [{
          "type": "section",
          "text": {"type": "mrkdwn", "text":"Pick a date for me to remind you"},
          "accessory": {
              "type": "datepicker",
              "action_id": "datepicker_remind",
              "initial_date":"2020-05-04",
              "placeholder": {"type": "plain_text", "text":"Select a date"}
          }
        }]
        say(
            blocks=blocks,
            text="Pick a date for me to remind you"
        )

イベントのリスニング

event() メソッドを使うと、Events API の任意のイベントをリッスンできます。リッスンするイベントは、アプリの設定であらかじめサブスクライブしておく必要があります。これを利用することで、アプリがインストールされたワークスペースで何らかのイベント(例:ユーザーがメッセージにリアクションをつけた、ユーザーがチャンネルに参加した)が発生したときに、アプリに何らかのアクションを実行させることができます。

event() メソッドには str 型の eventType を指定する必要があります。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
# ユーザーがワークスペースに参加した際に、自己紹介を促すメッセージを指定のチャンネルに送信
@app.event("team_join")
def ask_for_introduction(event, say):
    welcome_channel_id = "C12345"
    user_id = event["user"]
    text = f"Welcome to the team, <@{user_id}>! 🎉 You can introduce yourself in this channel."
    say(text=text, channel=welcome_channel_id)

メッセージのサブタイプのフィルタリング

message() リスナーは event("message") と等価の機能を提供します。

subtype という追加のキーを指定して、イベントのサブタイプでフィルタリングすることもできます。よく使われるサブタイプには、bot_messagemessage_replied があります。詳しくはメッセージイベントページを参照してください。

1
2
3
4
5
6
7
8
# 変更されたすべてのメッセージに一致
@app.event({
    "type": "message",
    "subtype": "message_changed"
})
def log_message_change(logger, event):
    user, text = event["user"], event["text"]
    logger.info(f"The user {user} changed the message to {text}")

Web API の使い方

app.client、またはミドルウェア・リスナーの引数 client として Bolt アプリに提供されている WebClient は必要な権限を付与されており、これを利用することであらゆる Web API メソッドを呼び出すことができます。このクライアントのメソッドを呼び出すと SlackResponse という Slack からの応答情報を含むオブジェクトが返されます。

Bolt の初期化に使用するトークンは context オブジェクトに設定されます。このトークンは、多くの Web API メソッドを呼び出す際に必要となります。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
@app.message("wake me up")
def say_hello(client, message):
    # 2020 年 9 月 30 日午後 11:59:59 を示す Unix エポック秒
    when_september_ends = 1601510399
    channel_id = message["channel"]
    client.chat_scheduleMessage(
        channel=channel_id,
        post_at=when_september_ends,
        text="Summer has come and passed"
    )

アクションのリスニング

Bolt アプリは action メソッドを用いて、ボタンのクリック、メニューの選択、メッセージショートカットなどのユーザーのアクションをリッスンすることができます。

アクションは str 型または re.Pattern 型の action_id でフィルタリングできます。action_id は、Slack プラットフォーム上のインタラクティブコンポーネントを区別する一意の識別子として機能します。

action() を使ったすべての例で ack() が使用されていることに注目してください。アクションのリスナー内では、Slack からのリクエストを受信したことを確認するために、ack() 関数を呼び出す必要があります。これについては、リクエストの確認セクションで説明しています。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
# 'approve_button' という action_id のブロックエレメントがトリガーされるたびに、このリスナーが呼び出させれる
@app.action("approve_button")
def update_message(ack):
    ack()
    # アクションへの反応としてメッセージを更新

制約付きオブジェクトを使用したアクションのリスニング

制約付きのオブジェクトを使用すると、callback_idblock_id、および action_id をそれぞれ、または任意に組み合わせてリッスンできます。オブジェクト内の制約は、str 型または re.Pattern 型で指定できます。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# この関数は、block_id が 'assign_ticket' に一致し
# かつ action_id が 'select_user' に一致する場合にのみ呼び出される
@app.action({
    "block_id": "assign_ticket",
    "action_id": "select_user"
})
def update_message(ack, body, client):
    ack()

    if "container" in body and "message_ts" in body["container"]:
        client.reactions_add(
            name="white_check_mark",
            channel=body["channel"]["id"],
            timestamp=body["container"]["message_ts"],
        )

アクションへの応答

アクションへの応答には、主に 2 つの方法があります。1 つ目の最も一般的なやり方は say() を使用する方法です。そのリクエストが発生した会話(チャンネルや DM)にメッセージを返します。

2 つ目は、respond() を使用する方法です。これは、アクションに関連づけられた response_url を使ったメッセージ送信を行うためのユーティリティです。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
# 'approve_button' という action_id のインタラクティブコンポーネントがトリガーされると、このリスナーが呼ばれる
@app.action("approve_button")
def approve_request(ack, say):
    # アクションのリクエストを確認
    ack()
    say("Request approved 👍")

respond() の利用

respond()response_url を使って送信するときに便利なメソッドで、これらと同じような動作をします。投稿するメッセージのペイロードには JSON オブジェクトを渡すことができ、メッセージはやり取りの発生元に反映されます。オプションのプロパティとして response_type(値は in_channel または ephemeral)、replace_originaldelete_original などを指定できます。

1
2
3
4
5
# 'user_select' という action_id を持つアクションのトリガーをリッスン
@app.action("user_select")
def select_user(ack, action, respond):
    ack()
    respond(f"You selected <@{action['selected_user']}>")

リクエストの確認

アクション(action)、コマンド(command)、およびオプション(options)の各リクエストは、必ず ack() 関数を使って確認を行う必要があります。これによってリクエストが受信されたことが Slack に認識され、Slack のユーザーインターフェイスが適切に更新されます。

リクエストの種類によっては、確認で通知方法が異なる場合があります。例えば、外部データソースを使用する選択メニューのオプションのリクエストに対する確認では、適切なオプションのリストとともに ack() を呼び出します。

確認までの猶予は 3 秒しかないため、新しいメッセージの送信や、データベースからの情報の取得は、ack() を呼び出した後で行うことをおすすめします。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 外部データを使用する選択メニューオプションに応答するサンプル
@app.options("menu_selection")
def show_menu_options(ack):
    options = [
        {
            "text": {"type": "plain_text", "text":"Option 1"},
            "value":"1-1",
        },
        {
            "text": {"type": "plain_text", "text":"Option 2"},
            "value":"1-2",
        },
    ]
    ack(options=options)

ショートカットのリスニングと応答

shortcut() メソッドは、グローバルショートカットメッセージショートカットの 2 つをサポートしています。

ショートカットは、いつでも呼び出せるアプリのエントリーポイントを提供するものです。グローバルショートカットは Slack のテキスト入力エリアや検索ウィンドウからアクセスできます。メッセージショートカットはメッセージのコンテキストメニューからアクセスできます。アプリは、ショートカットリクエストをリッスンするために shortcut() メソッドを使用します。このメソッドには str 型または re.Pattern 型の callback_id パラメーターを指定します。

ショートカットリクエストがアプリによって確認されたことを Slack に伝えるため、ack() を呼び出す必要があります。

ショートカットのペイロードには trigger_id が含まれます。アプリはこれを使って、ユーザーにやろうとしていることを確認するためのモーダルを開くことができます。

アプリの設定でショートカットを登録する際は、他の URL と同じように、リクエスト URL の末尾に /slack/events をつけます。

⚠️ グローバルショートカットのペイロードにはチャンネル ID が 含まれません。アプリでチャンネル ID を取得する必要がある場合は、モーダル内に conversations_select エレメントを配置します。メッセージショートカットにはチャンネル ID が含まれます。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# 'open_modal' という callback_id のショートカットをリッスン
@app.shortcut("open_modal")
def open_modal(ack, shortcut, client):
    # ショートカットのリクエストを確認
    ack()
    # 組み込みのクライアントを使って views_open メソッドを呼び出す
    client.views_open(
        trigger_id=shortcut["trigger_id"],
        # モーダルで表示するシンプルなビューのペイロード
        view={
            "type": "modal",
            "title": {"type": "plain_text", "text":"My App"},
            "close": {"type": "plain_text", "text":"Close"},
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text":"About the simplest modal you could conceive of :smile:\n\nMaybe <https://api.slack.com/reference/block-kit/interactive-components|*make the modal interactive*> or <https://api.slack.com/surfaces/modals/using#modifying|*learn more advanced modal use cases*>."
                    }
                },
                {
                    "type": "context",
                    "elements": [
                        {
                            "type": "mrkdwn",
                            "text":"Psssst this modal was designed using <https://api.slack.com/tools/block-kit-builder|*Block Kit Builder*>"
                        }
                    ]
                }
            ]
        }
    )

制約付きオブジェクトを使用したショートカットのリスニング

制約付きオブジェクトを使って callback_idtype によるリッスンできます。オブジェクト内の制約は str 型または re.Pattern オブジェクトを使用できます。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
# このリスナーが呼び出されるのは、callback_id が 'open_modal' と一致し
# かつ type が 'message_action' と一致するときのみ
@app.shortcut({"callback_id": "open_modal", "type": "message_action"})
def open_modal(ack, shortcut, client):
    # ショートカットのリクエストを確認
    ack()
    # 組み込みのクライアントを使って views_open メソッドを呼び出す
    client.views_open(
        trigger_id=shortcut["trigger_id"],
        view={
            "type": "modal",
            "title": {"type": "plain_text", "text":"My App"},
            "close": {"type": "plain_text", "text":"Close"},
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text":"About the simplest modal you could conceive of :smile:\n\nMaybe <https://api.slack.com/reference/block-kit/interactive-components|*make the modal interactive*> or <https://api.slack.com/surfaces/modals/using#modifying|*learn more advanced modal use cases*>."
                    }
                },
                {
                    "type": "context",
                    "elements": [
                        {
                            "type": "mrkdwn",
                            "text":"Psssst this modal was designed using <https://api.slack.com/tools/block-kit-builder|*Block Kit Builder*>"
                        }
                    ]
                }
            ]
        }
    )

コマンドのリスニングと応答

スラッシュコマンドが実行されたリクエストをリッスンするには、command() メソッドを使用します。このメソッドでは str 型の command_name の指定が必要です。

コマンドリクエストをアプリが受信し確認したことを Slack に通知するため、ack() を呼び出す必要があります。

スラッシュコマンドに応答する方法は 2 つあります。1 つ目は say() を使う方法で、文字列または JSON のペイロードを渡すことができます。2 つ目は respond() を使う方法です。これは response_url がある場合に活躍します。これらの方法はアクションへの応答セクションで詳しく説明しています。

アプリの設定でコマンドを登録するときは、リクエスト URL の末尾に /slack/events をつけます。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
# echoコマンドは受け取ったコマンドをそのまま返す
@app.command("/echo")
def repeat_text(ack, respond, command):
    # command リクエストを確認
    ack()
    respond(f"{command['text']}")

モーダルの開始

モーダルは、ユーザーからのデータの入力を受け付けたり、動的な情報を表示したりするためのインターフェイスです。組み込みの APIクライアントの views.open メソッドに、有効な trigger_idビューのペイロードを指定してモーダルを開始します。

ショートカットの実行、ボタンを押下、選択メニューの操作などの操作の場合、Request URL に送信されるペイロードには trigger_id が含まれます。

モーダルの生成方法についての詳細は、API ドキュメントを参照してください。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# ショートカットの呼び出しをリッスン
@app.shortcut("open_modal")
def open_modal(ack, body, client):
    # コマンドのリクエストを確認
    ack()
    # 組み込みのクライアントで views_open を呼び出し
    client.views_open(
        # 受け取りから 3 秒以内に有効な trigger_id を渡す
        trigger_id=body["trigger_id"],
        # ビューのペイロード
        view={
            "type": "modal",
            # ビューの識別子
            "callback_id": "view_1",
            "title": {"type": "plain_text", "text":"My App"},
            "submit": {"type": "plain_text", "text":"Submit"},
            "blocks": [
                {
                    "type": "section",
                    "text": {"type": "mrkdwn", "text":"Welcome to a modal with _blocks_"},
                    "accessory": {
                        "type": "button",
                        "text": {"type": "plain_text", "text":"Click me!"},
                        "action_id": "button_abc"
                    }
                },
                {
                    "type": "input",
                    "block_id": "input_c",
                    "label": {"type": "plain_text", "text":"What are your hopes and dreams?"},
                    "element": {
                        "type": "plain_text_input",
                        "action_id": "dreamy_input",
                        "multiline":True
                    }
                }
            ]
        }
    )

モーダルの更新と多重表示

モーダル内では、複数のモーダルをスタックのように重ねることができます。views_open という APIを呼び出すと、親となるとなるモーダルビューが追加されます。この最初の呼び出しの後、views_update を呼び出すことでそのビューを更新することができます。また、views_push を呼び出すと、親のモーダルの上にさらに新しいモーダルビューを重ねることもできます。

views_update
モーダルの更新は、組み込みのクライアントで views_update API を呼び出します。この API呼び出しでは、ビューを開いた時に生成された view_id と、更新後の blocks のリストを含む新しい view を指定します。既存のモーダルに含まれるエレメントをユーザーが操作した時にビューを更新する場合は、リクエストの body に含まれる view_id が利用できます。

views_push
既存のモーダルの上に新しいモーダルをスタックのように追加する場合は、組み込みのクライアントで views_push API を呼び出します。この API 呼び出しでは、有効な trigger_id と新しいビューのペイロードを指定します。views_push の引数は モーダルの開始 と同じです。モーダルを開いた後、このモーダルのスタックに追加できるモーダルビューは 2 つまでです。

モーダルの更新と多重表示に関する詳細は、API ドキュメントを参照してください。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# モーダルに含まれる、`button_abc` という action_id のボタンの呼び出しをリッスン
@app.action("button_abc")
def update_modal(ack, body, client):
    # ボタンのリクエストを確認
    ack()
    # 組み込みのクライアントで views_update を呼び出し
    client.views_update(
        # view_id を渡すこと
        view_id=body["view"]["id"],
        # 競合状態を防ぐためのビューの状態を示す文字列
        hash=body["view"]["hash"],
        # 更新後の blocks を含むビューのペイロード
        view={
            "type": "modal",
            # ビューの識別子
            "callback_id": "view_1",
            "title": {"type": "plain_text", "text":"Updated modal"},
            "blocks": [
                {
                    "type": "section",
                    "text": {"type": "plain_text", "text":"You updated the modal!"}
                },
                {
                    "type": "image",
                    "image_url": "https://media.giphy.com/media/SVZGEcYt7brkFUyU90/giphy.gif",
                    "alt_text":"Yay!The modal was updated"
                }
            ]
        }
    )

モーダルの送信のリスニング

モーダルのペイロードに input ブロックを含める場合、その入力値を受け取るためにview_submission リクエストをリッスンする必要があります。view_submission リクエストのリッスンには、組み込みのview() メソッドを利用することができます。view() の引数には、str 型または re.Pattern 型の callback_id を指定します。

input ブロックの値にアクセスするには state オブジェクトを参照します。state 内には values というオブジェクトがあり、block_id と一意の action_id に紐づける形で入力値を保持しています。

モーダルの送信について詳しくは、API ドキュメントを参照してください。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# view_submission リクエストを処理
@app.view("view_1")
def handle_submission(ack, body, client, view, logger):
    # `block_c`という block_id に `dreamy_input` を持つ input ブロックがある場合
    hopes_and_dreams = view["state"]["values"]["block_c"]["dreamy_input"]
    user = body["user"]["id"]
    # 入力値を検証
    errors = {}
    if hopes_and_dreams is not None and len(hopes_and_dreams) <= 5:
        errors["block_c"] = "The value must be longer than 5 characters"
    if len(errors) > 0:
        ack(response_action="errors", errors=errors)
        return
    # view_submission リクエストの確認を行い、モーダルを閉じる
    ack()
    # 入力されたデータを使った処理を実行。このサンプルでは DB に保存する処理を行う
    # そして入力値の検証結果をユーザーに送信

    # ユーザーに送信するメッセージ
    msg = ""
    try:
        # DB に保存
        msg = f"Your submission of {hopes_and_dreams} was successful"
    except Exception as e:
        # エラーをハンドリング
        msg = "There was an error with your submission"

    # ユーザーにメッセージを送信
    try:
        client.chat_postMessage(channel=user, text=msg)
    except e:
        logger.exception(f"Failed to post a message {e}")


ホームタブの更新

ホームタブは、サイドバーや検索画面からアクセス可能なサーフェスエリアです。アプリはこのエリアを使ってユーザーごとのビューを表示することができます。アプリ設定ページで App Home の機能を有効にすると、views.publish API メソッドの呼び出しで user_idビューのペイロードを指定して、ホームタブを公開・更新することができるようになります。

app_home_opened イベントをサブスクライブすると、ユーザーが App Home を開く操作をリッスンできます。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@app.event("app_home_opened")
def update_home_tab(client, event, logger):
    try:
        # 組み込みのクライアントを使って views.publish を呼び出す
        client.views_publish(
            # イベントに関連づけられたユーザー ID を使用
            user_id=event["user"],
            # アプリの設定で予めホームタブが有効になっている必要がある
            view={
                "type": "home",
                "blocks": [
                    {
                        "type": "section",
                        "text": {
                            "type": "mrkdwn",
                            "text": "*Welcome home, <@" + event["user"] + "> :house:*"
                        }
                    },
                    {
                        "type": "section",
                        "text": {
                          "type": "mrkdwn",
                          "text":"Learn how home tabs can be more useful and interactive <https://api.slack.com/surfaces/tabs/using|*in the documentation*>."
                        }
                    }
                ]
            }
        )
    except Exception as e:
        logger.error(f"Error publishing home tab: {e}")

オプションのリスニングと応答

options() メソッドは、Slack からのオプション(セレクトメニュー内の動的な選択肢)をリクエストするペイロードをリッスンします。 action() と同様に、文字列型の action_id または制約付きオブジェクトが必要です。

外部データソースを使って選択メニューをロードするためには、末部に /slack/events が付加された URL を Options Load URL として予め設定しておく必要があります。

external_select メニューでは action_id を指定することをおすすめしています。ただし、ダイアログを利用している場合、ダイアログが Block Kit に対応していないため、callback_id をフィルタリングするための制約オブジェクトを使用する必要があります。

オプションのリクエストに応答するときは、有効なオプションを含む options または option_groups のリストとともに ack() を呼び出す必要があります。API サイトにある外部データを使用する選択メニューに応答するサンプル例と、ダイアログでの応答例を参考にしてください。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 外部データを使用する選択メニューオプションに応答するサンプル例
@app.options("external_action")
def show_options(ack):
    options = [
        {
            "text": {"type": "plain_text", "text":"Option 1"},
            "value":"1-1",
        },
        {
            "text": {"type": "plain_text", "text":"Option 2"},
            "value":"1-2",
        },
    ]
    ack(options=options)

OAuth を使った認証

Slack アプリを複数のワークスペースにインストールできるようにするためには、OAuth フローを実装した上で、アクセストークンなどのインストールに関する情報をセキュアな方法で保存する必要があります。アプリを初期化する際に client_idclient_secretscopesinstallation_storestate_store を指定することで、OAuth のエンドポイントのルート情報や stateパラメーターの検証をBolt for Python にハンドリングさせることができます。カスタムのアダプターを実装する場合は、SDK が提供する組み込みのOAuth ライブラリを利用するのが便利です。これは Slack が開発したモジュールで、Bolt for Python 内部でも利用しています。

Bolt for Python によって slack/oauth_redirect というリダイレクト URL が生成されます。Slack はアプリのインストールフローを完了させたユーザーをこの URL にリダイレクトします。このリダイレクト URL は、アプリの設定の「OAuth and Permissions」であらかじめ追加しておく必要があります。この URL は、後ほど説明するように OAuthSettings というコンストラクタの引数で指定することもできます。

Bolt for Python は slack/install というルートも生成します。これはアプリを直接インストールするための「Add to Slack」ボタンを表示するために使われます。すでにワークスペースへのアプリのインストールが済んでいる場合に追加で各ユーザーのユーザートークンなどの情報を取得する場合や、カスタムのインストール用の URL を動的に生成したい場合などは、oauth_settingsauthorize_url_generator でカスタムの URL ジェネレーターを指定することができます。

バージョン 1.1.0 以降の Bolt for Python では、OrG 全体へのインストールがデフォルトでサポートされています。OrG 全体へのインストールは、アプリの設定の「Org Level Apps」で有効化できます。

Slack での OAuth を使ったインストールフローについて詳しくは、API ドキュメントを参照してください

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import os
from slack_bolt import App
from slack_bolt.oauth.oauth_settings import OAuthSettings
from slack_sdk.oauth.installation_store import FileInstallationStore
from slack_sdk.oauth.state_store import FileOAuthStateStore

oauth_settings = OAuthSettings(
    client_id=os.environ["SLACK_CLIENT_ID"],
    client_secret=os.environ["SLACK_CLIENT_SECRET"],
    scopes=["channels:read", "groups:read", "chat:write"],
    installation_store=FileInstallationStore(base_dir="./data"),
    state_store=FileOAuthStateStore(expiration_seconds=600, base_dir="./data")
)

app = App(
    signing_secret=os.environ["SIGNING_SECRET"],
    oauth_settings=oauth_settings
)

OAuth デフォルト設定をカスタマイズ

oauth_settings を使って OAuth モジュールのデフォルト設定を上書きすることができます。このカスタマイズされた設定は App の初期化時に渡します。以下の情報を変更可能です:

  • install_path : 「Add to Slack」ボタンのデフォルトのパスを上書きするために使用
  • redirect_uri : リダイレクト URL のデフォルトのパスを上書きするために使用
  • callback_options : OAuth フローの最後に表示するカスタムの成功ページと失敗ページの表示処理を提供するために使用
  • state_store : 組み込みの FileOAuthStateStore に代わる、カスタムの stateに関するデータストアを指定するために使用
  • installation_store : 組み込みの FileInstallationStore に代わる、カスタムのデータストアを指定するために使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
from slack_bolt.oauth.callback_options import CallbackOptions, SuccessArgs, FailureArgs
from slack_bolt.response import BoltResponse

def success(args:SuccessArgs) -> BoltResponse:
    assert args.request is not None
    return BoltResponse(
        status=200,  # ユーザーをリダイレクトすることも可能
        body="Your own response to end-users here"
    )

def failure(args:FailureArgs) -> BoltResponse:
    assert args.request is not None
    assert args.reason is not None
    return BoltResponse(
        status=args.suggested_status_code,
        body="Your own response to end-users here"
    )

callback_options = CallbackOptions(success=success, failure=failure)

import os
from slack_bolt import App
from slack_bolt.oauth.oauth_settings import OAuthSettings
from slack_sdk.oauth.installation_store import FileInstallationStore
from slack_sdk.oauth.state_store import FileOAuthStateStore

app = App(
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET"),
    installation_store=FileInstallationStore(base_dir="./data"),
    oauth_settings=OAuthSettings(
        client_id=os.environ.get("SLACK_CLIENT_ID"),
        client_secret=os.environ.get("SLACK_CLIENT_SECRET"),
        scopes=["app_mentions:read", "channels:history", "im:history", "chat:write"],
        user_scopes=[],
        redirect_uri=None,
        install_path="/slack/install",
        redirect_uri_path="/slack/oauth_redirect",
        state_store=FileOAuthStateStore(expiration_seconds=600, base_dir="./data"),
        callback_options=callback_options,
    ),
)

ソケットモードの利用

ソケットモードは、アプリに WebSocket での接続と、そのコネクション経由でのデータ受信を可能とします。Bolt for Python は、バージョン 1.2.0 からこれに対応しています。

ソケットモードでは、Slack からのペイロード送信を受け付けるエンドポイントをホストする HTTP サーバーを起動する代わりに WebSocket で Slack に接続し、そのコネクション経由でデータを受信します。ソケットモードを使う前に、アプリの管理画面でソケットモードの機能が有効になっていることを確認しておいてください。

ソケットモードを使用するには、環境変数に SLACK_APP_TOKEN を追加します。アプリのトークン(App-Level Token)は、アプリの設定の「Basic Information」セクションで確認できます。

組み込みのソケットモードアダプターを使用するのがおすすめですが、サードパーティ製ライブラリを使ったアダプターの実装もいくつか存在しています。利用可能なアダプターの一覧です。

内部的に利用する PyPI プロジェクト名 Bolt アダプター
slack_sdk slack_bolt.adapter.socket_mode
websocket_client slack_bolt.adapter.socket_mode.websocket_client
aiohttp (asyncio-based) slack_bolt.adapter.socket_mode.aiohttp
websockets (asyncio-based) slack_bolt.adapter.socket_mode.websockets
1
2
3
4
5
6
7
8
9
10
11
12
13
14
import os
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

# 事前に Slack アプリをインストールし 'xoxb-' で始まるトークンを入手
app = App(token=os.environ["SLACK_BOT_TOKEN"])

# ここでミドルウェアとリスナーの追加を行います

if __name__ == "__main__":
    # export SLACK_APP_TOKEN=xapp-***
    # export SLACK_BOT_TOKEN=xoxb-***
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    handler.start()

Async (asyncio) の利用

aiohttp のような asyncio をベースとしたアダプターを使う場合、アプリケーション全体が asyncio の async/await プログラミングモデルで実装されている必要があります。AsyncApp を動作させるためには AsyncSocketModeHandler とその async なミドルウェアやリスナーを利用します。

AsyncApp の使い方についての詳細は、Async (asyncio) の利用や、関連するサンプルコード例を参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from slack_bolt.app.async_app import AsyncApp
# デフォルトは aiohttp を使った実装
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler

app = AsyncApp(token=os.environ["SLACK_BOT_TOKEN"])

# ここでミドルウェアとリスナーの追加を行います

async def main():
    handler = AsyncSocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    await handler.start_async()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

ワークフローステップの概要

(アプリによる)ワークフローステップでは、処理をアプリ側で行うカスタムのワークフローステップを提供することができます。ユーザーはワークフロービルダーを使ってこれらのステップをワークフローに追加できます。

ワークフローステップは、次の 3 つのユーザーイベントで構成されます。

  • ワークフローステップをワークフローに追加・変更する
  • ワークフロー内のステップの設定内容を更新する
  • エンドユーザーがそのステップを実行する

ワークフローステップを機能させるためには、これら 3 つのイベントすべてに対応する必要があります。

アプリを使ったワークフローステップに関する詳細は、API ドキュメントを参照してください。


ステップの定義

ワークフローステップの作成には、Bolt が提供する WorkflowStep クラスを利用します。

ステップの callback_id と設定オブジェクトを指定して、WorkflowStep の新しいインスタンスを作成します。

設定オブジェクトは、editsaveexecute という 3 つのキーを持ちます。それぞれのキーは、単一のコールバック、またはコールバックのリストである必要があります。すべてのコールバックは、ワークフローステップのイベントに関する情報を保持する step オブジェクトにアクセスできます。

WorkflowStep のインスタンスを作成したら、それをapp.step() メソッドに渡します。これによって、アプリがワークフローステップのイベントをリッスンし、設定オブジェクトで指定されたコールバックを使ってそれに応答できるようになります。

また、デコレーターとして利用できる WorkflowStepBuilder クラスを使ってワークフローステップを定義することもできます。 詳細は、こちらのドキュメントのコード例などを参考にしてください。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください(共通 / ステップ用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import os
from slack_bolt import App
from slack_bolt.workflows.step import WorkflowStep

# いつも通りBolt アプリを起動する
app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET")
)

def edit(ack, step, configure):
    pass

def save(ack, view, update):
    pass

def execute(step, complete, fail):
    pass

# WorkflowStep の新しいインスタンスを作成する
ws = WorkflowStep(
    callback_id="add_task",
    edit=edit,
    save=save,
    execute=execute,
)
# ワークフローステップを渡してリスナーを設定する
app.step(ws)

ステップの追加・編集

作成したワークフローステップがワークフローに追加またはその設定を変更されるタイミングで、workflow_step_edit イベントがアプリに送信されます。このイベントがアプリに届くと、WorkflowStep で設定した edit コールバックが実行されます。

ステップの追加と編集のどちらが行われるときも、ワークフローステップの設定モーダルをビルダーに送信する必要があります。このモーダルは、そのステップ独自の設定を選択するための場所です。通常のモーダルより制限が強く、例えば titlesubmitclose のプロパティを含めることができません。設定モーダルの callback_id は、デフォルトではワークフローステップと同じものになります。

edit コールバック内で configure() ユーティリティを使用すると、対応する blocks 引数にビューのblocks 部分だけを渡して、ステップの設定モーダルを簡単に表示させることができます。必要な入力内容が揃うまで設定の保存を無効にするには、True の値をセットした submit_disabled を渡します。

設定モーダルの開き方に関する詳細は、こちらのドキュメントを参照してください。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください(共通 / ステップ用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def edit(ack, step, configure):
    ack()

    blocks = [
        {
            "type": "input",
            "block_id": "task_name_input",
            "element": {
                "type": "plain_text_input",
                "action_id": "name",
                "placeholder": {"type": "plain_text", "text":"Add a task name"},
            },
            "label": {"type": "plain_text", "text":"Task name"},
        },
        {
            "type": "input",
            "block_id": "task_description_input",
            "element": {
                "type": "plain_text_input",
                "action_id": "description",
                "placeholder": {"type": "plain_text", "text":"Add a task description"},
            },
            "label": {"type": "plain_text", "text":"Task description"},
        },
    ]
    configure(blocks=blocks)

ws = WorkflowStep(
    callback_id="add_task",
    edit=edit,
    save=save,
    execute=execute,
)
app.step(ws)

ステップの設定の保存

設定モーダルを開いた後、アプリは view_submission イベントをリッスンします。このイベントがアプリに届くと、WorkflowStep で設定した save コールバックが実行されます。

save コールバック内では、update() メソッドを使って、ワークフローに追加されたステップの設定を保存することができます。このメソッドには次の引数を指定します。

  • inputs : ユーザーがワークフローステップを実行したときにアプリが受け取る予定のデータを表す辞書型の値です。
  • outputs : ワークフローステップの完了時にアプリが出力するデータが設定されたオブジェクトのリストです。この outputs は、ワークフローの後続のステップで利用することができます。
  • step_name : ステップのデフォルトの名前をオーバーライドします。
  • step_image_url : ステップのデフォルトの画像をオーバーライドします。

これらのパラメータの構成方法に関する詳細は、こちらのドキュメントを参照してください。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください(共通 / ステップ用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def save(ack, view, update):
    ack()

    values = view["state"]["values"]
    task_name = values["task_name_input"]["name"]
    task_description = values["task_description_input"]["description"]

    inputs = {
        "task_name": {"value": task_name["value"]},
        "task_description": {"value": task_description["value"]}
    }
    outputs = [
        {
            "type": "text",
            "name": "task_name",
            "label":"Task name",
        },
        {
            "type": "text",
            "name": "task_description",
            "label":"Task description",
        }
    ]
    update(inputs=inputs, outputs=outputs)

ws = WorkflowStep(
    callback_id="add_task",
    edit=edit,
    save=save,
    execute=execute,
)
app.step(ws)

ステップの実行

エンドユーザーがワークフローステップを実行すると、アプリに workflow_step_execute イベントが送信されます。このイベントがアプリに届くと、WorkflowStep で設定した execute コールバックが実行されます。

save コールバックで取り出した inputs を使って、サードパーティの API を呼び出す、情報をデータベースに保存する、ユーザーのホームタブを更新するといった処理を実行することができます。また、ワークフローの後続のステップで利用する出力値を outputs オブジェクトに設定します。

execute コールバック内では、complete() を呼び出してステップの実行が成功したことを示すか、fail() を呼び出してステップの実行が失敗したことを示す必要があります。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください(共通 / ステップ用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def execute(step, complete, fail):
    inputs = step["inputs"]
    # すべての処理が成功した場合
    outputs = {
        "task_name": inputs["task_name"]["value"],
        "task_description": inputs["task_description"]["value"],
    }
    complete(outputs=outputs)

    # 失敗した処理がある場合
    error = {"message":"Just testing step failure!"}
    fail(error=error)

ws = WorkflowStep(
    callback_id="add_task",
    edit=edit,
    save=save,
    execute=execute,
)
app.step(ws)

アダプター

アダプターは Slack から届く受信リクエストの受付とパーズを担当し、それらのリクエストを BoltRequest の形式に変換して Bolt アプリに引き渡します。

デフォルトでは、Bolt の組み込みの HTTPServer アダプターが使われます。このアダプターは、ローカルで開発するのには問題がありませんが、本番環境での利用は推奨されていません。Bolt for Python には複数の組み込みのアダプターが用意されており、必要に応じてインポートしてアプリで使用することができます。組み込みのアダプターは Flask、Django、Starlette をはじめとする様々な人気の Python フレームワークをサポートしています。これらのアダプターは、あなたが選択した本番環境で利用可能な Webサーバーとともに利用することができます。

アダプターを使用するには、任意のフレームワークを使ってアプリを開発し、そのコードに対応するアダプターをインポートします。その後、アダプターのインスタンスを初期化して、受信リクエストの受付とパーズを行う関数を呼び出します。

すべてのアダプターの一覧と、設定や使い方のサンプルは、リポジトリの examples フォルダをご覧ください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from slack_bolt import App
app = App(
    signing_secret=os.environ.get("SIGNING_SECRET"),
    token=os.environ.get("SLACK_BOT_TOKEN")
)

# ここには Flask 固有の記述はありません
# App はフレームワークやランタイムに一切依存しません
@app.command("/hello-bolt")
def hello(body, ack):
    ack(f"Hi <@{body['user_id']}>!")

# Flask アプリを初期化します
from flask import Flask, request
flask_app = Flask(__name__)

# SlackRequestHandler は WSGI のリクエストを Bolt のインターフェイスに合った形に変換します
# Bolt レスポンスからの WSGI レスポンスの作成も行います
from slack_bolt.adapter.flask import SlackRequestHandler
handler = SlackRequestHandler(app)

# Flask アプリへのルートを登録します
@flask_app.route("/slack/events", methods=["POST"])
def slack_events():
    # handler はアプリのディスパッチメソッドを実行します
    return handler.handle(request)

カスタムのアダプター

アダプターはフレキシブルで、あなたが使用したいフレームワークに合わせた調整も可能です。アダプターでは、次の 2 つの要素が必須となっています。

  • __init__(app:App) : コンストラクター。Bolt の App のインスタンスを受け取り、保持します。
  • handle(req:Request) : Slack からの受信リクエストを受け取り、解析を行う関数。通常は handle() という名前です。リクエストを BoltRequest のインスタンスに合った形にして、保持している Bolt アプリに引き渡します。

BoltRequest のインスタンスの作成では、以下の 4 種類のパラメーターを指定できます。

パラメーター 説明 必須
body: str そのままのリクエストボディ Yes
query: any クエリストリング No
headers:Dict[str, Union[str, List[str]]] リクエストヘッダー No
context:BoltContext リクエストのコンテキスト情報 No

アダプターは、Bolt アプリからの BoltResponse のインスタンスを返します。

カスタムのアダプターに関連した詳しいサンプルについては、組み込みのアダプターの実装を参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
# Flask で必要なパッケージをインポートします
from flask import Request, Response, make_response

from slack_bolt.app import App
from slack_bolt.request import BoltRequest
from slack_bolt.response import BoltResponse

# この例は Flask アダプターを簡略化したものです
# もう少し詳しい完全版のサンプルは、adapter フォルダをご覧ください
# github.com/slackapi/bolt-python/blob/main/slack_bolt/adapter/flask/handler.py

# HTTP リクエストを取り込み、標準の BoltRequest に変換します
def to_bolt_request(req:Request) -> BoltRequest:
    return BoltRequest(
        body=req.get_data(as_text=True),
        query=req.query_string.decode("utf-8"),
        headers=req.headers,
    )

# BoltResponse を取り込み、標準の Flask レスポンスに変換します
def to_flask_response(bolt_resp:BoltResponse) -> Response:
    resp:Response = make_response(bolt_resp.body, bolt_resp.status)
    for k, values in bolt_resp.headers.items():
        for v in values:
            resp.headers.add_header(k, v)
    return resp

# アプリからインスタンス化します
# Flask アプリを受け取ります
class SlackRequestHandler:
    def __init__(self, app:App):
        self.app = app

    # Slack からリクエストが届いたときに
    # Flask アプリの handle() を呼び出します
    def handle(self, req:Request) -> Response:
        # この例では OAuth に関する部分は扱いません
        if req.method == "POST":
            # Bolt へのリクエストをディスパッチし、処理とルーティングを行います
            bolt_resp:BoltResponse = self.app.dispatch(to_bolt_request(req))
            return to_flask_response(bolt_resp)

        return make_response("Not Found", 404)

Async(asyncio)の使用

非同期バージョンの Bolt を使用する場合は、App の代わりに AsyncApp インスタンスをインポートして初期化します。AsyncApp では AIOHTTP を使って API リクエストを行うため、aiohttp をインストールする必要があります(requirements.txt に追記するか、pip install aiohttp を実行します)。

非同期バージョンのプロジェクトのサンプルは、リポジトリの examples フォルダにあります。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# aiohttp のインストールが必要です
from slack_bolt.async_app import AsyncApp
app = AsyncApp()

@app.event("app_mention")
async def handle_mentions(event, client, say):  # 非同期関数
    api_response = await client.reactions_add(
        channel=event["channel"],
        timestamp=event["ts"],
        name="eyes",
    )
    await say("What's up?")

if __name__ == "__main__":
    app.start(3000)

他のフレームワークを使用する

AsyncApp#start() では内部的に AIOHTTP のWebサーバーが実装されています。必要に応じて、受信リクエストの処理に AIOHTTP 以外のフレームワークを使用することができます。

この例では Sanic を使用しています。すべてのアダプターのリストについては、adapter フォルダ を参照してください。

以下のコマンドを実行すると、必要なパッケージをインストールして、Sanic サーバーをポート 3000 で起動します。

1
2
3
4
# 必要なパッケージをインストールします
pip install slack_bolt sanic uvicorn
# ソースファイルを async_app.py として保存します
uvicorn async_app:api --reload --port 3000 --log-level debug
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from slack_bolt.async_app import AsyncApp
app = AsyncApp()

# ここには Sanic に固有の記述はありません
# AsyncApp はフレームワークやランタイムに依存しません
@app.event("app_mention")
async def handle_app_mentions(say):
    await say("What's up?")

import os
from sanic import Sanic
from sanic.request import Request
from slack_bolt.adapter.sanic import AsyncSlackRequestHandler

# App のインスタンスから Sanic 用のアダプターを作成します
app_handler = AsyncSlackRequestHandler(app)
# Sanic アプリを作成します
api = Sanic(name="awesome-slack-app")

@api.post("/slack/events")
async def endpoint(req: Request):
    # app_handler では内部的にアプリのディスパッチメソッドが実行されます
    return await app_handler.handle(req)

if __name__ == "__main__":
    api.run(host="0.0.0.0", port=int(os.environ.get("PORT", 3000)))

エラーの処理

リスナー内でエラーが発生した場合に try/except ブロックを使用して直接エラーを処理することができます。アプリに関連するエラーは、BoltError 型です。Slack API の呼び出しに関連するエラーは、SlackApiError 型となります。

デフォルトでは、すべての処理されなかった例外のログはグローバルのエラーハンドラーによってコンソールに出力されます。グローバルのエラーを開発者自身で処理するには、app.error(fn) 関数を使ってグローバルのエラーハンドラーをアプリに設定します。

1
2
3
4
@app.error
def custom_error_handler(error, body, logger):
    logger.exception(f"Error: {error}")
    logger.info(f"Request body: {body}")

ロギング

デフォルトでは、アプリからのログ情報は、既定の出力先に出力されます。logging モジュールをインポートすれば、basicConfig()level パラメーターでrootのログレベルを変更することができます。指定できるログレベルは、重要度の低い方から debuginfowarningerror、および critical です。

グローバルのコンテキストとは別に、指定のログレベルに応じて単一のメッセージをログ出力することもできます。Bolt では Python 標準の logging モジュールが使われているため、このモジュールが持つすべての機能を利用できます。

1
2
3
4
5
6
7
8
9
10
11
12
13
import logging

# グローバルのコンテキストの logger です
# logging をインポートする必要があります
logging.basicConfig(level=logging.DEBUG)

@app.event("app_mention")
def handle_mention(body, say, logger):
    user = body["event"]["user"]
    # 単一の logger の呼び出しです
    # グローバルの logger がリスナーに渡されています
    logger.debug(body)
    say(f"{user} mentioned your app")

認可(Authorization)

認可(Authorization)は、Slack からの受信リクエストを処理するにあたって、どのようなSlack クレデンシャル (ボットトークンなど) を使用可能にするかを決定するプロセスです。

単一のワークスペースにインストールされるアプリでは、token パラメーターを使って App のコンストラクターにボットトークンを渡すという、シンプルな方法が使えます。それに対して、複数のワークスペースにインストールされるアプリでは、次の 2 つの方法のいずれかを使用する必要があります。簡単なのは、組み込みの OAuth サポートを使用する方法です。OAuth サポートは、OAuth フロー用のURLのセットアップとstateの検証を行います。詳細は「OAuth を使った認証」セクションを参照してください。

よりカスタマイズできる方法として、App をインスタンス化する関数にauthorize パラメーターを指定する方法があります。authorize 関数から返される AuthorizeResult のインスタンスには、どのユーザーがどこで発生させたリクエストかを示す情報が含まれます。

AuthorizeResult には、いくつか特定のプロパティを指定する必要があり、いずれも str 型です。

  • bot_token(xoxb)または user_token(xoxp): どちらか一方が必須です。ほとんどのアプリでは、デフォルトの bot_token を使用すればよいでしょう。トークンを渡すことで、say() などの組み込みの関数を機能させることができます。
  • bot_user_id および bot_id : bot_token を使用する場合に指定します。
  • enterprise_id および team_id : アプリに届いたリクエストから見つけることができます。
  • user_id : user_token を使用する場合に必須です。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import os
from slack_bolt import App
# AuthorizeResult クラスをインポートします
from slack_bolt.authorization import AuthorizeResult

# これはあくまでサンプル例です(ユーザートークンがないことを想定しています)
# 実際にはセキュアな DB に認可情報を保存してください
installations = [
    {
      "enterprise_id":"E1234A12AB",
      "team_id":"T12345",
      "bot_token": "xoxb-123abc",
      "bot_id":"B1251",
      "bot_user_id":"U12385"
    },
    {
      "team_id":"T77712",
      "bot_token": "xoxb-102anc",
      "bot_id":"B5910",
      "bot_user_id":"U1239",
      "enterprise_id":"E1234A12AB"
    }
]

def authorize(enterprise_id, team_id, logger):
    # トークンを取得するためのあなたのロジックをここに記述します
    for team in installations:
        # 一部のチームは enterprise_id を持たない場合があります
        is_valid_enterprise = True if (("enterprise_id" not in team) or (enterprise_id == team["enterprise_id"])) else False
        if ((is_valid_enterprise == True) and (team["team_id"] == team_id)):
          # AuthorizeResult のインスタンスを返します
          # bot_id と bot_user_id を保存していない場合、bot_token を使って `from_auth_test_response` を呼び出すと、自動的に取得できます
          return AuthorizeResult(
              enterprise_id=enterprise_id,
              team_id=team_id,
              bot_token=team["bot_token"],
              bot_id=team["bot_id"],
              bot_user_id=team["bot_user_id"]
          )

    logger.error("No authorization information was found")

app = App(
    signing_secret=os.environ["SLACK_SIGNING_SECRET"],
    authorize=authorize
)

トークンのローテーション

Bolt for Python v1.7.0 から、アクセストークンのさらなるセキュリティ強化のレイヤーであるトークンローテーションの機能に対応しています。トークンローテーションは OAuth V2 の RFC で規定されているものです。

既存の Slack アプリではアクセストークンが無期限に存在し続けるのに対して、トークンローテーションを有効にしたアプリではアクセストークンが失効するようになります。リフレッシュトークンを利用して、アクセストークンを長期間にわたって更新し続けることができます。

Bolt for Python の組み込みの OAuth 機能 を使用していれば、Bolt for Python が自動的にトークンローテーションの処理をハンドリングします。

トークンローテーションに関する詳細は API ドキュメントを参照してください。


リスナーミドルウェア

リスナーミドルウェアは、それを渡したリスナーでのみ実行されるミドルウェアです。リスナーには、middleware パラメーターを使ってミドルウェア関数をいくつでも渡すことができます。このパラメーターには、1 つまたは複数のミドルウェア関数からなるリストを指定します。

非常にシンプルなリスナーミドルウェアの場合であれば、next() メソッドを呼び出す代わりに bool 値(処理を継続したい場合は True)を返すだけで済む「リスナーマッチャー」を使うとよいでしょう。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# "bot_message" サブタイプのメッセージを抽出するリスナーミドルウェア
def no_bot_messages(message, next):
    subtype = message.get("subtype")
    if subtype != "bot_message":
       next()

# このリスナーは人間によって送信されたメッセージのみを受け取ります
@app.event(event="message", middleware=[no_bot_messages])
def log_message(logger, event):
    logger.info(f"(MSG) User: {event['user']}\nMessage: {event['text']}")

# リスナーマッチャー: 簡略化されたバージョンのリスナーミドルウェア
def no_bot_messages(message) -> bool:
    return message.get("subtype") != "bot_message"

@app.event(
    event="message", 
    matchers=[no_bot_messages]
    # or matchers=[lambda message: message.get("subtype") != "bot_message"]
)
def log_message(logger, event):
    logger.info(f"(MSG) User: {event['user']}\nMessage: {event['text']}")

グローバルミドルウェア

グローバルミドルウェアは、すべての受信リクエストに対して、リスナーミドルウェアが呼ばれる前に実行されるものです。ミドルウェア関数を app.use() に渡すことで、アプリにはグローバルミドルウェアをいくつでも追加できます。ミドルウェア関数で受け取れる引数はリスナー関数と同じものに加えてnext() 関数があります。

グローバルミドルウェアでもリスナーミドルウェアでも、次のミドルウェアに実行チェーンの制御をリレーするために、next() を呼び出す必要があります。

指定可能な引数の一覧はモジュールドキュメントを参考にしてください。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@app.use
def auth_abc(client, context, logger, payload, next):
    slack_user_id = payload["user"]
    help_channel_id = "C12345"

    try:
        # Slack のユーザー ID を使って外部のシステムでユーザーを検索します
        user = abc.lookup_by_id(slack_user_id)
        # 結果を context に保存します
        context["user"] = user
    except Exception:
        client.chat_postEphemeral(
            channel=payload["channel"],
            user=slack_user_id,
            text=f"Sorry <@{slack_user_id}>, you aren't registered in ABC or there was an error with authentication.Please post in <#{help_channel_id}> for assistance"
        )

    # 次のミドルウェアに実行権を渡します
    next()

コンテキストの追加

すべてのリスナーは context ディクショナリにアクセスできます。リスナーはこれを使ってリクエストの付加情報を得ることができます。受信リクエストに含まれる user_idteam_idchannel_identerprise_id などの情報は、Bolt によって自動的に設定されます。

context は単純なディクショナリで、変更を直接加えることもできます。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
# ユーザーID を使って外部のシステムからタスクを取得するリスナーミドルウェア
def fetch_tasks(context, event, next):
    user = event["user"]
    try:
        # get_tasks は、ユーザー ID に対応するタスクのリストを DB から取得します
        user_tasks = db.get_tasks(user)
        tasks = user_tasks
    except Exception:
        # タスクが見つからなかった場合 get_tasks() は例外を投げます
        tasks = []
    finally:
        # ユーザーのタスクを context に設定します
        context["tasks"] = tasks
        next()

# section のブロックのリストを作成するリスナーミドルウェア
def create_sections(context, next):
    task_blocks = []
    # 先ほどのミドルウェアを使って context に追加した各タスクについて、処理を繰り返します
    for task in context["tasks"]:
        task_blocks.append(
            {
              "type": "section",
              "text": {
                  "type": "mrkdwn",
                  "text": f"*{task['title']}*
{task['body']}"
              },
              "accessory": {
                  "type": "button",
                  "text": {
                    "type": "plain_text",
                    "text":"See task"
                  },
                  "url": task["url"],
                }
            }
        )
    # ブロックのリストを context に設定します
    context["blocks"] = task_blocks
    next()

# ユーザーがアプリのホームを開くのをリッスンします
# fetch_tasks ミドルウェアを含めます
@app.event(
  event = "app_home_opened",
  middleware = [fetch_tasks, create_sections]
)
def show_tasks(event, client, context):
    # ユーザーのホームタブにビューを表示します
    client.views_publish(
        user_id=event["user"],
        view={
            "type": "home",
            "blocks": context["blocks"]
        }
    )

Lazy リスナー(FaaS)

⚠️ Lazy リスナー関数は、FaaS 環境への Bolt for Python アプリのデプロイを容易にする、ベータ版の機能です。開発中の機能であるため、Bolt for Python の API は変更される可能性があります。

通常、リスナー関数では最初の手順として ack() を呼び出します。ack() を呼び出すことで、アプリがリクエストを受け取り、適切な時間内(3 秒間)に処理する予定であることが Slack に伝えられます。

しかし、FaaS 環境や類似のランタイムで実行されるアプリでは、HTTP レスポンスを返したあとにスレッドやプロセスの実行を続けることができないため、同じパターンに従うことはできません。代わりに、process_before_response フラグを True に設定します。この設定により、ack() の呼び出しとリクエストの処理を安全に行うリスナーを作成することができます。しかし、3 秒以内にすべての処理を完了させる必要があることは変わりません。Events APIに応答するリスナーではack() メソッドの呼び出しを必要としませんが、この設定では処理を 3 秒以内に完了させる必要があります。

Lazy リスナーは、この問題を解決するためのソリューションです。Lazy リスナーは、デコレーターとして動作するものではなく、次の 2 つのキーワード引数を指定することにより動作するものです。 * ack:Callable : ack() の呼び出しを行います。 * lazy:List[Callable] : リクエストに関係する、時間のかかるプロセスの処理を担当します。Lazy 関数からは ack() にアクセスできません。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def respond_to_slack_within_3_seconds(body, ack):
    text = body.get("text")
    if text is None or len(text) == 0:
        ack(f":x:Usage: /start-process (description here)")
    else:
        ack(f"Accepted! (task: {body['text']})")

import time
def run_long_process(respond, body):
    time.sleep(5)  # 3 秒より長い時間を指定します
    respond(f"Completed! (task: {body['text']})")

app.command("/start-process")(
    # この場合でも ack() は 3 秒以内に呼ばれます
    ack=respond_to_slack_within_3_seconds,
    # Lazy 関数がイベントの処理を担当します
    lazy=[run_long_process]
)

AWS Lambda を使用した例

このサンプルは、AWS Lambda にコードをデプロイします。examples フォルダにはほかにもサンプルが用意されています。

1
2
3
4
5
6
7
8
9
10
11
12
13
pip install slack_bolt
# ソースコードを main.py として保存します
# config.yaml を設定してハンドラーを `handler: main.handler` で参照できるようにします

# https://pypi.org/project/python-lambda/
pip install python-lambda

# config.yml を適切に設定します
# lazy リスナーの実行には lambda:InvokeFunction と lambda:GetFunction が必要です
export SLACK_SIGNING_SECRET=***
export SLACK_BOT_TOKEN=xoxb-***
echo 'slack_bolt' &gt; requirements.txt
lambda deploy --config-file config.yaml --requirements requirements.txt
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from slack_bolt import App
from slack_bolt.adapter.aws_lambda import SlackRequestHandler

# FaaS で実行するときは process_before_response を True にする必要があります
app = App(process_before_response=True)

def respond_to_slack_within_3_seconds(body, ack):
    text = body.get("text")
    if text is None or len(text) == 0:
        ack(":x: Usage: /start-process (description here)")
    else:
        ack(f"Accepted! (task: {body['text']})")

import time
def run_long_process(respond, body):
    time.sleep(5)  # 3 秒より長い時間を指定します
    respond(f"Completed! (task: {body['text']})")

app.command("/start-process")(
    ack=respond_to_slack_within_3_seconds,  # `ack()` の呼び出しを担当します
    lazy=[run_long_process]  # `ack()` の呼び出しはできません。複数の関数を持たせることができます。
)

def handler(event, context):
    slack_handler = SlackRequestHandler(app=app)
    return slack_handler.handle(event, context)

このサンプルアプリを実行するには、以下の IAM 権限が必要になります。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction",
                "lambda:GetFunction"
            ],
            "Resource": "*"
        }
    ]
}