To listen to messages that your app has access to receive, you can use the message() method which filters out events that aren’t of type message.
message() accepts an argument of type str or re.Pattern object that filters out any messages that don’t match the pattern.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
# This will match any message that contains 👋
@app.message(":wave:")
def say_hello(message, say):
user = message['user']
say(f"Hi there, <@{user}>!")
The re.compile() method can be used instead of a string for more granular matching.
1
2
3
4
5
6
7
import re
@app.message(re.compile("(hi|hello|hey)"))
def say_hello_regex(say, context):
# regular expression matches are inside of context.matches
greeting = context['matches'][0]
say(f"{greeting}, how are you?")
Within your listener function, say() is available whenever there is an associated conversation (for example, a conversation where the event or action which triggered the listener occurred). say() accepts a string to post simple messages and JSON payloads to send more complex messages. The message payload you pass in will be sent to the associated conversation.
In the case that you’d like to send a message outside of a listener or you want to do something more advanced (like handle specific errors), you can call client.chat_postMessage using the client attached to your Bolt instance.
Refer to the module document to learn the available listener arguments.
1
2
3
4
# Listens for messages containing "knock knock" and responds with an italicized "who's there?"
@app.message("knock knock")
def ask_who(message, say):
say("_Who's there?_")
say() accepts more complex message payloads to make it easy to add functionality and structure to your messages.
To explore adding rich message layouts to your app, read through the guide on our API site and look through templates of common app flows in the Block Kit Builder.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# Sends a section block with datepicker when someone reacts with a 📅 emoji
@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"
)
You can listen to any Events API event using the event() method after subscribing to it in your app configuration. This allows your app to take action when something happens in a workspace where it’s installed, like a user reacting to a message or joining a channel.
The event() method requires an eventType of type str.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
6
7
# When a user joins the workspace, send a message in a predefined channel asking them to introduce themselves
@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)
The message() listener is equivalent to event("message").
You can filter on subtypes of events by passing in the additional key subtype. Common message subtypes like bot_message and message_replied can be found on the message event page.
1
2
3
4
5
6
7
8
# Matches all modified messages
@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}")
You can call any Web API method using the WebClient provided to your Bolt app as either app.client or client in middleware/listener arguments (given that your app has the appropriate scopes). When you call one the client’s methods, it returns a SlackResponse which contains the response from Slack.
The token used to initialize Bolt can be found in the context object, which is required to call most Web API methods.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
6
7
8
9
10
@app.message("wake me up")
def say_hello(client, message):
# Unix Epoch time for September 30, 2020 11:59:59 PM
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"
)
Your app can listen to user actions, like button clicks, and menu selects, using the action method.
Actions can be filtered on an action_id of type str or re.Pattern. action_ids act as unique identifiers for interactive components on the Slack platform.
You’ll notice in all action() examples, ack() is used. It is required to call the ack() function within an action listener to acknowledge that the event was received from Slack. This is discussed in the acknowledging events section.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
# Your listener will be called every time a block element with the action_id "approve_button" is triggered
@app.action("approve_button")
def update_message(ack):
ack()
# Update the message to reflect the action
You can use a constraints object to listen to callback_ids, block_ids, and action_ids (or any combination of them). Constraints in the object can be of type str or re.Pattern.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# Your function will only be called when the action_id matches 'select_user' AND the block_id matches 'assign_ticket'
@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"],
)
There are two main ways to respond to actions. The first (and most common) way is to use say(), which sends a message back to the conversation where the incoming event took place.
The second way to respond to actions is using respond(), which is a utility to use the response_url associated with the action.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
6
# Your listener will be called every time an interactive component with the action_id “approve_button” is triggered
@app.action("approve_button")
def approve_request(ack, say):
# Acknowledge action request
ack()
say("Request approved 👍")
Since respond() is a utility for calling the response_url, it behaves in the same way. You can pass a JSON object with a new message payload that will be published back to the source of the original interaction with optional properties like response_type (which has a value of in_channel or ephemeral), replace_original, and delete_original.
1
2
3
4
5
# Listens to actions triggered with action_id of “user_select”
@app.action("user_select")
def select_user(ack, action, respond):
ack()
respond(f"You selected <@{action['selected_user']}>")
Actions, commands, and options events must always be acknowledged using the ack() function. This lets Slack know that the event was received and updates the Slack user interface accordingly.
Depending on the type of event, your acknowledgement may be different. For example, when acknowledging a menu selection associated with an external data source, you would call ack() with a list of relevant options.
We recommend calling ack() right away before sending a new message or fetching information from your database since you only have 3 seconds to respond.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# Example of responding to an external_select options request
@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)
The shortcut() method supports both global shortcuts and message shortcuts.
Shortcuts are invokable entry points to apps. Global shortcuts are available from within search and text composer area in Slack. Message shortcuts are available in the context menus of messages. Your app can use the shortcut() method to listen to incoming shortcut events. The method requires a callback_id parameter of type str or re.Pattern.
Shortcuts must be acknowledged with ack() to inform Slack that your app has received the event.
Shortcuts include a trigger_id which an app can use to open a modal that confirms the action the user is taking.
When setting up shortcuts within your app configuration, as with other URLs, you’ll append /slack/events to your request URL.
⚠️ Note that global shortcuts do not include a channel ID. If your app needs access to a channel ID, you may use a conversations_select element within a modal. Message shortcuts do include a channel ID.
Refer to the module document to learn the available listener arguments.
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
# The open_modal shortcut listens to a shortcut with the callback_id "open_modal"
@app.shortcut("open_modal")
def open_modal(ack, shortcut, client):
# Acknowledge the shortcut request
ack()
# Call the views_open method using the built-in WebClient
client.views_open(
trigger_id=shortcut["trigger_id"],
# A simple view payload for a modal
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*>"
}
]
}
]
}
)
You can use a constraints object to listen to callback_ids, and types. Constraints in the object can be of type str or 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
# Your listener will only be called when the callback_id matches 'open_modal' AND the type matches 'message_action'
@app.shortcut({"callback_id": "open_modal", "type": "message_action"})
def open_modal(ack, shortcut, client):
# Acknowledge the shortcut request
ack()
# Call the views_open method using one of the built-in WebClients
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*>"
}
]
}
]
}
)
Your app can use the command() method to listen to incoming slash command events. The method requires a command_name of type str.
Commands must be acknowledged with ack() to inform Slack your app has received the event.
There are two ways to respond to slash commands. The first way is to use say(), which accepts a string or JSON payload. The second is respond() which is a utility for the response_url. These are explained in more depth in the responding to actions section.
When setting up commands within your app configuration, you’ll append /slack/events to your request URL.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
6
# The echo command simply echoes on command
@app.command("/echo")
def repeat_text(ack, say, command):
# Acknowledge command request
ack()
say(f"{command['text']}")
Modals are focused surfaces that allow you to collect user data and display dynamic information. You can open a modal by passing a valid trigger_id and a view payload to the built-in client’s views.open method.
Your app receives trigger_ids in payloads sent to your Request URL that are triggered by user invocations, like a shortcut, button press, or interaction with a select menu.
Read more about modal composition in the API documentation.
Refer to the module document to learn the available listener arguments.
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
# Listen for a shortcut invocation
@app.shortcut("open_modal")
def open_modal(ack, body, client):
# Acknowledge the command request
ack()
# Call views_open with the built-in client
client.views_open(
# Pass a valid trigger_id within 3 seconds of receiving it
trigger_id=body["trigger_id"],
# View payload
view={
"type": "modal",
# View identifier
"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
}
}
]
}
)
Modals contain a stack of views. When you call views_open, you add the root view to the modal. After the initial call, you can dynamically update a view by calling views_update, or stack a new view on top of the root view by calling views_push.
views_update
To update a view, you can use the built-in client to call views_update with the view_id that was generated when you opened the view, and a new view including the updated blocks list. If you’re updating the view when a user interacts with an element inside of an existing view, the view_id will be available in the body of the request.
views_push
To push a new view onto the view stack, you can use the built-in client to call views_push with a valid trigger_id a new view payload. The arguments for views_push is the same as opening modals. After you open a modal, you may only push two additional views onto the view stack.
Learn more about updating and pushing views in our API documentation.
Refer to the module document to learn the available listener arguments.
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
# Listen for a button invocation with action_id `button_abc` (assume it's inside of a modal)
@app.action("button_abc")
def update_modal(ack, body, client):
# Acknowledge the button request
ack()
# Call views_update with the built-in client
client.views_update(
# Pass the view_id
view_id=body["view"]["id"],
# String that represents view state to protect against race conditions
hash=body["view"]["hash"],
# View payload with updated blocks
view={
"type": "modal",
# View identifier
"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"
}
]
}
)
If a view payload contains any input blocks, you must listen to view_submission events to receive their values. To listen to view_submission events, you can use the built-in view() method. view() requires a callback_id of type str or re.Pattern.
You can access the value of the input blocks by accessing the state object. state contains a values object that uses the block_id and unique action_id to store the input values.
Read more about view submissions in our API documentation.
Refer to the module document to learn the available listener arguments.
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
# Handle a view_submission event
@app.view("view_1")
def handle_submission(ack, body, client, view):
# Assume there's an input block with `block_c` as the block_id and `dreamy_input`
hopes_and_dreams = view["state"]["values"]["block_c"]["dreamy_input"]
user = body["user"]["id"]
# Validate the inputs
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
# Acknowledge the view_submission event and close the modal
ack()
# Do whatever you want with the input data - here we're saving it to a DB
# then sending the user a verification of their submission
# Message to send user
msg = ""
try:
# Save to DB
msg = f"Your submission of {hopes_and_dreams} was successful"
except Exception as e:
# Handle error
msg = "There was an error with your submission"
finally:
# Message the user
client.chat_postMessage(channel=user, text=msg)
Home tabs are customizable surfaces accessible via the sidebar and search that allow apps to display views on a per-user basis. After enabling App Home within your app configuration, home tabs can be published and updated by passing a user_id and view payload to the views.publish method.
You can subscribe to the app_home_opened event to listen for when users open your App Home.
Refer to the module document to learn the available listener arguments.
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:
# Call views.publish with the built-in client
client.views_publish(
# Use the user ID associated with the event
user_id=event["user"],
# Home tabs must be enabled in your app configuration
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}")
The options() method listens for incoming option request payloads from Slack. Similar to action(),
an action_id or constraints object is required. In order to load external data into your select menus, you must provide an options load URL in your app configuration, appended with /slack/events.
While it’s recommended to use action_id for external_select menus, dialogs do not support Block Kit so you’ll have to use the constraints object to filter on a callback_id.
To respond to options requests, you’ll need to call ack() with a valid options or option_groups list. Both external select response examples and dialog response examples can be found on our API site.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# Example of responding to an external_select options request
@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)
Slack apps installed on multiple workspaces will need to implement OAuth, then store installation information (like access tokens) securely. By providing client_id, client_secret, scopes, installation_store, and state_store when initializing App, Bolt for Python will handle the work of setting up OAuth routes and verifying state. If you’re implementing a custom adapter, you can make use of our OAuth library, which is what Bolt for Python uses under the hood.
Bolt for Python will create a Redirect URL slack/oauth_redirect, which Slack uses to redirect users after they complete your app’s installation flow. You will need to add this Redirect URL in your app configuration settings under OAuth and Permissions. This path can be configured in the OAuthSettings argument described below.
Bolt for Python will also create a slack/install route, where you can find an Add to Slack button for your app to perform direct installs of your app. If you need any additional authorizations (user tokens) from users inside a team when your app is already installed or a reason to dynamically generate an install URL, you can pass your own custom URL generator to oauth_settings as authorize_url_generator.
Bolt for Python automatically includes support for org wide installations in version 1.1.0+. Org wide installations can be enabled in your app configuration settings under Org Level Apps.
To learn more about the OAuth installation flow with Slack, read the API documentation.
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
)
You can override the default OAuth using oauth_settings, which can be passed in during the initialization of App. You can override the following:
install_path: Override default path for “Add to Slack” buttonredirect_uri: Override default redirect url pathcallback_options: Provide custom success and failure pages at the end of the OAuth flowstate_store: Provide a custom state store instead of using the built in FileOAuthStateStore
installation_store: Provide a custom installation store instead of the built-in 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, # you can redirect users too
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,
),
)
With the introduction of Socket Mode, Bolt for Python introduced support in version 1.2.0. With Socket Mode, instead of creating a server with endpoints that Slack sends payloads too, the app will instead connect to Slack via a WebSocket connection and receive data from Slack over the socket connection. Make sure to enable Socket Mode in your app configuration settings.
To use the Socket Mode, add SLACK_APP_TOKEN as an environment variable. You can get your App Token in your app configuration settings under the Basic Information section.
While we recommend using the built-in Socket Mode adapter, there are a few other 3rd party library based implementations. Here is the list of available adapters.
| PyPI Project | Bolt Adapter |
|---|---|
| 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
# Install the Slack app and get xoxb- token in advance
app = App(token=os.environ["SLACK_BOT_TOKEN"])
# Add middleware / listeners here
if __name__ == "__main__":
# export SLACK_APP_TOKEN=xapp-***
# export SLACK_BOT_TOKEN=xoxb-***
handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
handler.start()
To use the asyncio-based adapters such as aiohttp, your whole app needs to be compatible with asyncio’s async/await programming model. AsyncSocketModeHandler is available for running AsyncApp and its async middleware and listeners.
To learn how to use AsyncApp, checkout the Using Async document and relevant examples.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from slack_bolt.app.async_app import AsyncApp
# The default is the aiohttp based implementation
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
app = AsyncApp(token=os.environ["SLACK_BOT_TOKEN"])
# Add middleware / listeners here
async def main():
handler = AsyncSocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
await handler.start_async()
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Workflow Steps from apps allow your app to create and process custom workflow steps that users can add using Workflow Builder.
A workflow step is made up of three distinct user events:
All three events must be handled for a workflow step to function.
Read more about workflow steps from apps in the API documentation.
To create a workflow step, Bolt provides the WorkflowStep class.
When instantiating a new WorkflowStep, pass in the step’s callback_id and a configuration object.
The configuration object contains three keys: edit, save, and execute. Each of these keys must be a single callback or a list of callbacks. All callbacks have access to a step object that contains information about the workflow step event.
After instantiating a WorkflowStep, you can pass it into app.step(). Behind the scenes, your app will listen and respond to the workflow step’s events using the callbacks provided in the configuration object.
Refer to the module documents (common / step-specific) to learn the available arguments.
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
# Initiate the Bolt app as you normally would
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
# Create a new WorkflowStep instance
ws = WorkflowStep(
callback_id="add_task",
edit=edit,
save=save,
execute=execute,
)
# Pass Step to set up listeners
app.step(ws)
When a builder adds (or later edits) your step in their workflow, your app will receive a workflow_step_edit event. The edit callback in your WorkflowStep configuration will be run when this event is received.
Whether a builder is adding or editing a step, you need to send them a workflow step configuration modal. This modal is where step-specific settings are chosen, and it has more restrictions than typical modals—most notably, it cannot include title, submit, or close properties. By default, the configuration modal’s callback_id will be the same as the workflow step.
Within the edit callback, the configure() utility can be used to easily open your step’s configuration modal by passing in the view’s blocks with the corresponding blocks argument. To disable saving the configuration before certain conditions are met, you can also pass in submit_disabled with a value of True.
To learn more about opening configuration modals, read the documentation.
Refer to the module documents (common / step-specific) to learn the available arguments.
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)
After the configuration modal is opened, your app will listen for the view_submission event. The save callback in your WorkflowStep configuration will be run when this event is received.
Within the save callback, the update() method can be used to save the builder’s step configuration by passing in the following arguments:
inputs is a dictionary representing the data your app expects to receive from the user upon workflow step execution.outputs is a list of objects containing data that your app will provide upon the workflow step’s completion. Outputs can then be used in subsequent steps of the workflow.step_name overrides the default Step namestep_image_url overrides the default Step imageTo learn more about how to structure these parameters, read the documentation.
Refer to the module documents (common / step-specific) to learn the available arguments.
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)
When your workflow step is executed by an end user, your app will receive a workflow_step_execute event. The execute callback in your WorkflowStep configuration will be run when this event is received.
Using the inputs from the save callback, this is where you can make third-party API calls, save information to a database, update the user’s Home tab, or decide the outputs that will be available to subsequent workflow steps by mapping values to the outputs object.
Within the execute callback, your app must either call complete() to indicate that the step’s execution was successful, or fail() to indicate that the step’s execution failed.
Refer to the module documents (common / step-specific) to learn the available arguments.
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"]
# if everything was successful
outputs = {
"task_name": inputs["task_name"]["value"],
"task_description": inputs["task_description"]["value"],
}
complete(outputs=outputs)
# if something went wrong
error = {"message": "Just testing step failure!"}
fail(error=error)
ws = WorkflowStep(
callback_id="add_task",
edit=edit,
save=save,
execute=execute,
)
app.step(ws)
Adapters are responsible for handling and parsing incoming events from Slack to conform to BoltRequest, then dispatching those events to your Bolt app.
By default, Bolt will use the built-in HTTPServer adapter. While this is okay for local development, it is not recommended for production. Bolt for Python includes a collection of built-in adapters that can be imported and used with your app. The built-in adapters support a variety of popular Python frameworks including Flask, Django, and Starlette among others. Adapters support the use of any production-ready web server of your choice.
To use an adapter, you’ll create an app with the framework of your choosing and import its corresponding adapter. Then you’ll initialize the adapter instance and call its function that handles and parses incoming events.
The full list adapters, as well as configuration and sample usage, can be found within the repository’s examples folder.
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")
)
# There is nothing specific to Flask here!
# App is completely framework/runtime agnostic
@app.command("/hello-bolt")
def hello(body, ack):
ack(f"Hi <@{body['user_id']}>!")
# Initialize Flask app
from flask import Flask, request
flask_app = Flask(__name__)
# SlackRequestHandler translates WSGI requests to Bolt's interface
# and builds WSGI response from Bolt's response.
from slack_bolt.adapter.flask import SlackRequestHandler
handler = SlackRequestHandler(app)
# Register routes to Flask app
@flask_app.route("/slack/events", methods=["POST"])
def slack_events():
# handler runs App's dispatch method
return handler.handle(request)
Adapters are flexible and can be adjusted based on the framework you prefer. There are two necessary components of adapters:
__init__(app: App): Constructor that accepts and stores an instance of the Bolt App.handle(req: Request): Function (typically named handle()) that receives incoming Slack requests, parses them to conform to an instance of BoltRequest, then dispatches them to the stored Bolt app.BoltRequest instantiation accepts four parameters:
| Parameter | Description | Required? |
|---|---|---|
body: str |
The raw request body | Yes |
query: any |
The query string data | No |
headers: Dict[str, Union[str, List[str]]] |
Request headers | No |
context: BoltContext |
Any context for the request | No |
Your adapter will return an instance of BoltResponse from the Bolt app.
For more in-depth examples of custom adapters, look at the implementations of the built-in adapters.
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
# Necessary imports for 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
# This example is a simplified version of the Flask adapter
# For a more detailed, complete example, look in the adapter folder
# github.com/slackapi/bolt-python/blob/main/slack_bolt/adapter/flask/handler.py
# Takes in an HTTP request and converts it to a standard 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,
)
# Takes in a BoltResponse and converts it to a standard Flask response
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
# Instantiated from your app
# Accepts a Flask app
class SlackRequestHandler:
def __init__(self, app: App):
self.app = app
# handle() will be called from your Flask app
# when you receive a request from Slack
def handle(self, req: Request) -> Response:
# This example does not cover OAuth
if req.method == "POST":
# Dispatch the request for Bolt to handle and route
bolt_resp: BoltResponse = self.app.dispatch(to_bolt_request(req))
return to_flask_response(bolt_resp)
return make_response("Not Found", 404)
To use the async version of Bolt, you can import and initialize an AsyncApp instance (rather than App). AsyncApp relies on AIOHTTP to make API requests, which means you’ll need to install aiohttp (by adding to requirements.txt or running pip install aiohttp).
Sample async projects can be found within the repository’s examples folder.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# Requirement: install aiohttp
from slack_bolt.async_app import AsyncApp
app = AsyncApp()
@app.event("app_mention")
async def handle_mentions(event, client, say): # async function
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)
Internally AsyncApp#start() implements a AIOHTTP web server. If you prefer, you can use a framework other than AIOHTTP to handle incoming requests.
This example uses Sanic, but the full list of adapters are in the adapter folder.
The following commands install the necessary requirements and starts the Sanic server on port 3000.
1
2
3
4
# Install requirements
pip install slack_bolt sanic uvicorn
# Save the source as 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()
# There is nothing specific to Sanic here!
# AsyncApp is completely framework/runtime agnostic
@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
# Create an adapter for Sanic with the App instance
app_handler = AsyncSlackRequestHandler(app)
# Create a Sanic app
api = Sanic(name="awesome-slack-app")
@api.post("/slack/events")
async def endpoint(req: Request):
# app_handler internally runs the App's dispatch method
return await app_handler.handle(req)
if __name__ == "__main__":
api.run(host="0.0.0.0", port=int(os.environ.get("PORT", 3000)))
If an error occurs in a listener, you can handle it directly using a try/except block. Errors associated with your app will be of type BoltError. Errors associated with calling Slack APIs will be of type SlackApiError.
By default, the global error handler will log all non-handled exceptions to the console. To handle global errors yourself, you can attach a global error handler to your app using the app.error(fn) function.
1
2
3
4
@app.error
def custom_error_handler(error, body, logger):
logger.exception(f"Error: {error}")
logger.info(f"Request body: {body}")
By default, Bolt will log information from your app to the output destination. After you’ve imported the logging module, you can customize the root log level by passing the level parameter to basicConfig(). The available log levels in order of least to most severe are debug, info, warning, error, and critical.
Outside of a global context, you can also log a single message corresponding to a specific level. Because Bolt uses Python’s standard logging module, you can use any its features.
1
2
3
4
5
6
7
8
9
10
11
12
13
import logging
# logger in a global context
# requires importing logging
logging.basicConfig(level=logging.DEBUG)
@app.event("app_mention")
def handle_mention(body, say, logger):
user = body["event"]["user"]
# single logger call
# global logger is passed to listener
logger.debug(body)
say(f"{user} mentioned your app")
Authorization is the process of determining which Slack credentials should be available while processing an incoming Slack event.
Apps installed on a single workspace can simply pass their bot token into the App constructor using the token parameter. However, if your app will be installed on multiple workspaces, you have two options. The easier option is to use the built-in OAuth support. This will handle setting up OAuth routes and verifying state. Read the section on authenticating with OAuth for details.
For a more custom solution, you can set the authorize parameter to a function upon App instantiation. The authorize function should return an instance of AuthorizeResult, which contains information about who and where the event is coming from.
AuthorizeResult should have a few specific properties, all of type str:
bot_token (xoxb) or user_token (xoxp) are required. Most apps will use bot_token by default. Passing a token allows built-in functions (like say()) to work.bot_user_id and bot_id, if using a bot_token.enterprise_id and team_id, which can be found in events sent to your app.user_id only when using 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
# Import the AuthorizeResult class
from slack_bolt.authorization import AuthorizeResult
# This is just an example (assumes there are no user tokens)
# You should store authorizations in a secure 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):
# You can implement your own logic to fetch token here
for team in installations:
# enterprise_id doesn't exist for some teams
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)):
# Return an instance of AuthorizeResult
# If you don't store bot_id and bot_user_id, could also call `from_auth_test_response` with your bot_token to automatically fetch them
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
)
Listener middleware is only run for the listener in which it’s passed. You can pass any number of middleware functions to the listener using the middleware parameter, which must be a list that contains one to many middleware functions.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
6
7
8
9
10
# Listener middleware which filters out messages with "bot_message" subtype
def no_bot_messages(message, next):
subtype = message.get("subtype")
if subtype != "bot_message":
next()
# This listener only receives messages from humans
@app.event(event="message", middleware=[no_bot_messages])
def log_message(logger, event):
logger.info(f"(MSG) User: {event['user']}\nMessage: {event['text']}")
Global middleware is run for all incoming events, before any listener middleware. You can add any number of global middleware to your app by passing middleware functions to app.use(). Middleware functions are called with the same arguments as listeners, with an additional next() function.
Both global and listener middleware must call next() to pass control of the execution chain to the next middleware.
Refer to the module document to learn the available listener arguments.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@app.use
def auth_acme(client, context, logger, payload, next):
slack_user_id = payload["user"]
help_channel_id = "C12345"
try:
# Look up user in external system using their Slack user ID
user = acme.lookup_by_id(slack_user_id)
# Add that to 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 Acme or there was an error with authentication. Please post in <#{help_channel_id}> for assistance"
)
# Pass control to the next middleware
next()
All listeners have access to a context dictionary, which can be used to enrich events with additional information. Bolt automatically attaches information that is included in the incoming event, like user_id, team_id, channel_id, and enterprise_id.
context is just a dictionary, so you can directly modify it.
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
# Listener middleware to fetch tasks from external system using user ID
def fetch_tasks(context, event, next):
user = event["user"]
try:
# Assume get_tasks fetchs list of tasks from DB corresponding to user ID
user_tasks = db.get_tasks(user)
tasks = user_tasks
except Exception:
# get_tasks() raises exception because no tasks are found
tasks = []
finally:
# Put user's tasks in context
context["tasks"] = tasks
next()
# Listener middleware to create a list of section blocks
def create_sections(context, next):
task_blocks = []
# Loops through tasks added to context in previous middleware
for task in context["tasks"]:
task_blocks.append(
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*{task['title']}*\n{task['body']}"
},
"accessory": {
"type": "button",
"text": {
"type": "plain_text",
"text": "See task"
},
"url": task["url"],
}
}
)
# Put list of blocks in context
context["blocks"] = task_blocks
next()
# Listen for user opening app home
# Include fetch_tasks middleware
@app.event(
event = "app_home_opened",
middleware = [fetch_tasks, create_sections]
)
def show_tasks(event, client, context):
# Publish view to user's home tab
client.views_publish(
user_id=event["user"],
view={
"type": "home",
"blocks": context["blocks"]
}
)
⚠️ Lazy listener functions are a beta feature to make it easier to deploy Bolt for Python apps to FaaS environments. As the feature is developed, Bolt for Python’s API is subject to change.
Typically you’d call ack() as the first step of your listener functions. Calling ack() tells Slack that you’ve received the event and are handling it in within reasonable amount of time (3 seconds).
However, apps running on FaaS or similar runtimes that don’t allow you to run threads or processes after returning an HTTP response cannot follow this pattern. Instead, you should set the process_before_response flag to True. This allows you to create a listener that calls ack() and handles the event safely, though you still need to complete everything within 3 seconds. For events, while a listener doesn’t need ack() method call as you normally would, the listener needs to complete within 3 seconds, too.
Lazy listeners can be a solution for this issue. Rather than acting as a decorator, lazy listeners take two keyword args:
ack: Callable: Responsible for calling ack()
lazy: List[Callable]: Responsible for handling any time-consuming processes related to the event. The lazy function does not have access to 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) # longer than 3 seconds
respond(f"Completed! (task: {body['text']})")
app.command("/start-process")(
# ack() is still called within 3 seconds
ack=respond_to_slack_within_3_seconds,
# Lazy function is responsible for processing the event
lazy=[run_long_process]
)
This example deploys the code to AWS Lambda. There are more examples within the examples folder.
1
2
3
4
5
6
7
8
9
10
11
12
13
pip install slack_bolt
# Save the source code as main.py
# and refer handler as `handler: main.handler` in config.yaml
# https://pypi.org/project/python-lambda/
pip install python-lambda
# Configure config.yml properly
# lambda:InvokeFunction & lambda:GetFunction are required for running lazy listeners
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
# process_before_response must be True when running on FaaS
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) # longer than 3 seconds
respond(f"Completed! (task: {body['text']})")
app.command("/start-process")(
ack=respond_to_slack_within_3_seconds, # responsible for calling `ack()`
lazy=[run_long_process] # unable to call `ack()` / can have multiple functions
)
def handler(event, context):
slack_handler = SlackRequestHandler(app=app)
return slack_handler.handle(event, context)
Please note that the following IAM permissions would be required for running this example app.
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": "*"
}
]
}