あなたのアプリがアクセス権限を持つメッセージの投稿イベントをリッスンするには message()
メソッドを利用します。このメソッドは type
が message
ではないイベントを処理対象から除外します。
message()
の引数には str
型または re.Pattern
オブジェクトを指定できます。この条件のパターンに一致しないメッセージは除外されます。
あなたのアプリがアクセス権限を持つメッセージの投稿イベントをリッスンするには message()
メソッドを利用します。このメソッドは type
が message
ではないイベントを処理対象から除外します。
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_message
や message_replied
があります。詳しくはメッセージイベントページを参照してください。サブタイプなしのイベントだけにフルターするために明に None
を指定することもできます。
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}")
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()
# アクションへの反応としてメッセージを更新
制約付きのオブジェクトを使用すると、block_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()
は response_url
を使って送信するときに便利なメソッドで、これらと同じような動作をします。投稿するメッセージのペイロードには、全てのメッセージペイロードのプロパティとオプションのプロパティとして response_type
(値は "in_channel"
または "ephemeral"
)、replace_original
、delete_original
、unfurl_links
、unfurl_media
などを指定できます。こうすることによってアプリから送信されるメッセージは、やり取りの発生元に反映されます。
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)、ショートカット(shortcut)、オプション(options)、およびモーダルからのデータ送信(view_submission)の各リクエストは、必ず ack()
関数を使って確認を行う必要があります。これによってリクエストが受信されたことが Slack に認識され、Slack のユーザーインターフェイスが適切に更新されます。
リクエストの種類によっては、確認で通知方法が異なる場合があります。例えば、外部データソースを使用する選択メニューのオプションのリクエストに対する確認では、適切なオプションのリストとともに ack()
を呼び出します。モーダルからのデータ送信に対する確認では、 response_action
を渡すことでモーダルの更新などを行えます。
確認までの猶予は 3 秒しかないため、新しいメッセージの送信やデータベースからの情報の取得といった時間のかかる処理は、ack()
を呼び出した後で行うことをおすすめします。
FaaS / serverless 環境を使う場合、 ack()
するタイミングが異なります。 これに関する詳細は Lazy listeners (FaaS) を参照してください。
指定可能な引数の一覧はモジュールドキュメントを参考にしてください。
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_id
や type
によるリッスンできます。オブジェクト内の制約は 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
に紐づける形で入力値を保持しています。
view_submission
リクエストに対してモーダルを更新するには、リクエストの確認の中で update
という response_action
と新しく作成した view
を指定します。
1
2
3
4
5
6
7
# モーダル送信でのビューの更新
@app.view("view_1")
def handle_submission(ack, body):
# build_new_view() method はモーダルビューを返します
# モーダルの構築には Block Kit Builder を試してみてください:
# https://app.slack.com/block-kit-builder/#%7B%22type%22:%22modal%22,%22callback_id%22:%22view_1%22,%22title%22:%7B%22type%22:%22plain_text%22,%22text%22:%22My%20App%22,%22emoji%22:true%7D,%22blocks%22:%5B%5D%7D
ack(response_action="update", view=build_new_view(body))
この例と同様に、モーダルでの送信リクエストに対して、エラーを表示するためのオプションもあります。
モーダルの送信について詳しくは、API ドキュメントを参照してください。
view_closed
リクエストをリッスンするためには callback_id
を指定して、かつ notify_on_close
属性をモーダルのビューに設定する必要があります。以下のコード例をご覧ください。
よく詳しい情報は、API ドキュメントを参照してください。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
client.views_open(
trigger_id=body.get("trigger_id"),
view={
"type": "modal",
"callback_id": "modal-id", # view_closed の処理時に必要
"title": {
"type": "plain_text",
"text": "Modal title"
},
"blocks": [],
"close": {
"type": "plain_text",
"text": "Cancel"
},
"notify_on_close": True, # この属性は必須
}
)
# view_closed リクエストを処理する
@app.view_closed("modal-id")
def handle_view_closed(ack, body, logger):
ack()
logger.info(body)
指定可能な引数の一覧はモジュールドキュメントを参考にしてください。
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):
# `input_c`という block_id に `dreamy_input` を持つ input ブロックがある場合
hopes_and_dreams = view["state"]["values"]["input_c"]["dreamy_input"]
user = body["user"]["id"]
# 入力値を検証
errors = {}
if hopes_and_dreams is not None and len(hopes_and_dreams) <= 5:
errors["input_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 サイトにある外部データを使用する選択メニューに応答するサンプル例と、ダイアログでの応答例を参考にしてください。
さらに、ユーザーが入力したキーワードに基づいたオプションを返すようフィルタリングロジックを適用することもできます。 これは payload
という引数の ` value` の値に基づいて、それぞれのパターンで異なるオプションの一覧を返すように実装することができます。 Bolt for Python のすべてのリスナーやミドルウェアでは、多くの有用な引数にアクセスすることができますので、チェックしてみてください。
指定可能な引数の一覧はモジュールドキュメントを参考にしてください。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 外部データを使用する選択メニューオプションに応答するサンプル例
@app.options("external_action")
def show_options(ack, payload):
options = [
{
"text": {"type": "plain_text", "text": "Option 1"},
"value": "1-1",
},
{
"text": {"type": "plain_text", "text": "Option 2"},
"value": "1-2",
},
]
keyword = payload.get("value")
if keyword is not None and len(keyword) > 0:
options = [o for o in options if keyword in o["text"]["text"]]
ack(options=options)
Slack アプリを複数のワークスペースにインストールできるようにするためには、OAuth フローを実装した上で、アクセストークンなどのインストールに関する情報をセキュアな方法で保存する必要があります。アプリを初期化する際に client_id
、client_secret
、scopes
、installation_store
、state_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_settings
の authorize_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/installations"),
state_store=FileOAuthStateStore(expiration_seconds=600, base_dir="./data/states")
)
app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
oauth_settings=oauth_settings
)
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/installations"),
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/states"),
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()
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())
アダプターは 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("SLACK_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)
非同期バージョンの 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のログレベルを変更することができます。指定できるログレベルは、重要度の低い方から debug
、info
、warning
、error
、および 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)は、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_id
、team_id
、channel_id
、enterprise_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 環境への Slack アプリのデプロイを容易にする機能です。この機能は Bolt for Python でのみ利用可能で、他の Bolt フレームワークでこの機能に対応することは予定していません。
通常、アクション(action)、コマンド(command)、ショートカット(shortcut)、オプション(options)、およびモーダルからのデータ送信(view_submission)をハンドルするとき、 ack()
を呼び出し、Slack からのリクエストを 3 秒以内に確認する必要があります。ack()
を呼び出すと Slack に HTTP ステータスが 200 OK の応答が返されます。こうすることで、アプリがリクエストの応答を処理中であることを Slack に伝えられます。通常であれば、この確認処理を処理関数の最初のステップとして行うことを推奨しています。
しかし、FaaS 環境や類似のランタイムで実行されるアプリでは、 HTTP レスポンスを返したあとにスレッドやプロセスの実行を続けることができない ため、確認の応答を送信した後で時間のかかる処理をするという通常のパターンに従うことができません。こうした環境で動作させるためには、 process_before_response
フラグを True
に設定します。このフラグが True
に設定されている場合、Bolt はリスナー関数での処理が完了するまで HTTP レスポンスの送信を遅延させます。そのため 3 秒以内にリスナーのすべての処理が完了しなかった場合は Slack 上でタイムアウトのエラー表示となってしまいます。また、Events API に応答するリスナーでは明示的な ack()
メソッドの呼び出しを必要としませんが、この設定を有効にしている場合、リスナーの処理を 3 秒以内に完了させる必要があることにも注意してください。
処理関数の中で時間のかかる処理を実行できるようにするために、私たちは Lazy リスナーという関数を実行する仕組みを導入しました。Lazy リスナーは、デコレーターとして動作させるのではなく、以下の 2 つのキーワード引数を受け取ります。
ack: Callable
: 3 秒以内での 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 にコードをデプロイします。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' > 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": "*"
}
]
}
(アプリによる)ワークフローステップでは、処理をアプリ側で行うカスタムのワークフローステップを提供することができます。ユーザーはワークフロービルダーを使ってこれらのステップをワークフローに追加できます。
ワークフローステップは、次の 3 つのユーザーイベントで構成されます。
ワークフローステップを機能させるためには、これら 3 つのイベントすべてに対応する必要があります。
アプリを使ったワークフローステップに関する詳細は、API ドキュメントを参照してください。
ワークフローステップの作成には、Bolt が提供する WorkflowStep
クラスを利用します。
ステップの callback_id
と設定オブジェクトを指定して、WorkflowStep
の新しいインスタンスを作成します。
設定オブジェクトは、edit
、save
、execute
という 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
コールバックが実行されます。
ステップの追加と編集のどちらが行われるときも、ワークフローステップの設定モーダルをビルダーに送信する必要があります。このモーダルは、そのステップ独自の設定を選択するための場所です。通常のモーダルより制限が強く、例えば title
、submit
、close
のプロパティを含めることができません。設定モーダルの 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)