Module slack_bolt.app.async_app
Expand source code
import inspect
import logging
import os
import time
from typing import Optional, List, Union, Callable, Pattern, Dict, Awaitable, Sequence
from aiohttp import web
from slack_bolt.app.async_server import AsyncSlackAppServer
from slack_bolt.listener.async_builtins import AsyncTokenRevocationListeners
from slack_bolt.listener.async_listener_completion_handler import (
AsyncDefaultListenerCompletionHandler,
)
from slack_bolt.listener.asyncio_runner import AsyncioListenerRunner
from slack_bolt.middleware.message_listener_matches.async_message_listener_matches import (
AsyncMessageListenerMatches,
)
from slack_bolt.oauth.async_internals import select_consistent_installation_store
from slack_bolt.util.utils import get_name_for_callable
from slack_bolt.workflows.step.async_step import (
AsyncWorkflowStep,
AsyncWorkflowStepBuilder,
)
from slack_bolt.workflows.step.async_step_middleware import AsyncWorkflowStepMiddleware
from slack_sdk.oauth.installation_store.async_installation_store import (
AsyncInstallationStore,
)
from slack_sdk.web.async_client import AsyncWebClient
from slack_bolt.authorization import AuthorizeResult
from slack_bolt.authorization.async_authorize import (
AsyncAuthorize,
AsyncCallableAuthorize,
AsyncInstallationStoreAuthorize,
)
from slack_bolt.error import BoltError, BoltUnhandledRequestError
from slack_bolt.logger.messages import (
warning_client_prioritized_and_token_skipped,
warning_token_skipped,
error_token_required,
warning_unhandled_request,
debug_checking_listener,
debug_running_listener,
error_unexpected_listener_middleware,
error_listener_function_must_be_coro_func,
error_client_invalid_type_async,
error_authorize_conflicts,
error_oauth_settings_invalid_type_async,
error_oauth_flow_invalid_type_async,
warning_bot_only_conflicts,
debug_return_listener_middleware_response,
info_default_oauth_settings_loaded,
error_installation_store_required_for_builtin_listeners,
warning_unhandled_by_global_middleware,
)
from slack_bolt.lazy_listener.asyncio_runner import AsyncioLazyListenerRunner
from slack_bolt.listener.async_listener import AsyncListener, AsyncCustomListener
from slack_bolt.listener.async_listener_error_handler import (
AsyncDefaultListenerErrorHandler,
AsyncCustomListenerErrorHandler,
)
from slack_bolt.listener_matcher import builtins as builtin_matchers
from slack_bolt.listener_matcher.async_listener_matcher import (
AsyncListenerMatcher,
AsyncCustomListenerMatcher,
)
from slack_bolt.logger import get_bolt_logger, get_bolt_app_logger
from slack_bolt.middleware.async_builtins import (
AsyncSslCheck,
AsyncRequestVerification,
AsyncIgnoringSelfEvents,
AsyncUrlVerification,
)
from slack_bolt.middleware.async_custom_middleware import (
AsyncMiddleware,
AsyncCustomMiddleware,
)
from slack_bolt.middleware.authorization.async_multi_teams_authorization import (
AsyncMultiTeamsAuthorization,
)
from slack_bolt.middleware.authorization.async_single_team_authorization import (
AsyncSingleTeamAuthorization,
)
from slack_bolt.oauth.async_oauth_flow import AsyncOAuthFlow
from slack_bolt.oauth.async_oauth_settings import AsyncOAuthSettings
from slack_bolt.request.async_request import AsyncBoltRequest
from slack_bolt.response import BoltResponse
from slack_bolt.util.async_utils import create_async_web_client
class AsyncApp:
def __init__(
self,
*,
logger: Optional[logging.Logger] = None,
# Used in logger
name: Optional[str] = None,
# Set True when you run this app on a FaaS platform
process_before_response: bool = False,
# Set True if you want to handle an unhandled request as an exception
raise_error_for_unhandled_request: bool = False,
# Basic Information > Credentials > Signing Secret
signing_secret: Optional[str] = None,
# for single-workspace apps
token: Optional[str] = None,
client: Optional[AsyncWebClient] = None,
# for multi-workspace apps
installation_store: Optional[AsyncInstallationStore] = None,
# for either only bot scope usage or v1.0.x compatibility
installation_store_bot_only: Optional[bool] = None,
authorize: Optional[Callable[..., Awaitable[AuthorizeResult]]] = None,
# for the OAuth flow
oauth_settings: Optional[AsyncOAuthSettings] = None,
oauth_flow: Optional[AsyncOAuthFlow] = None,
# No need to set (the value is used only in response to ssl_check requests)
verification_token: Optional[str] = None,
):
"""Bolt App that provides functionalities to register middleware/listeners.
import os
from slack_bolt.async_app import AsyncApp
# Initializes your app with your bot token and signing secret
app = AsyncApp(
token=os.environ.get("SLACK_BOT_TOKEN"),
signing_secret=os.environ.get("SLACK_SIGNING_SECRET")
)
# Listens to incoming messages that contain "hello"
@app.message("hello")
async def message_hello(message, say): # async function
# say() sends a message to the channel where the event was triggered
await say(f"Hey there <@{message['user']}>!")
# Start your app
if __name__ == "__main__":
app.start(port=int(os.environ.get("PORT", 3000)))
Refer to https://slack.dev/bolt-python/concepts#async for details.
If yoy would like to build an OAuth app for enabling the app to run with multiple workspaces,
refer to https://slack.dev/bolt-python/concepts#authenticating-oauth to learn how to configure the app.
Args:
logger: The custom logger that can be used in this app.
name: The application name that will be used in logging. If absent, the source file name will be used.
process_before_response: True if this app runs on Function as a Service. (Default: False)
raise_error_for_unhandled_request: True if you want to raise exceptions for unhandled requests
and use @app.error listeners instead of
the built-in handler, which pints warning logs and returns 404 to Slack (Default: False)
signing_secret: The Signing Secret value used for verifying requests from Slack.
token: The bot/user access token required only for single-workspace app.
client: The singleton `slack_sdk.web.async_client.AsyncWebClient` instance for this app.
authorize: The function to authorize an incoming request from Slack
by checking if there is a team/user in the installation data.
installation_store: The module offering save/find operations of installation data
installation_store_bot_only: Use `AsyncInstallationStore#async_find_bot()` if True (Default: False)
oauth_settings: The settings related to Slack app installation flow (OAuth flow)
oauth_flow: Instantiated `slack_bolt.oauth.AsyncOAuthFlow`. This is always prioritized over oauth_settings.
verification_token: Deprecated verification mechanism. This can used only for ssl_check requests.
"""
signing_secret = signing_secret or os.environ.get("SLACK_SIGNING_SECRET")
token = token or os.environ.get("SLACK_BOT_TOKEN")
self._name: str = name or inspect.stack()[1].filename.split(os.path.sep)[-1]
self._signing_secret: str = signing_secret
self._verification_token: Optional[str] = verification_token or os.environ.get(
"SLACK_VERIFICATION_TOKEN", None
)
self._framework_logger = logger or get_bolt_logger(AsyncApp)
self._raise_error_for_unhandled_request = raise_error_for_unhandled_request
self._token: Optional[str] = token
if client is not None:
if not isinstance(client, AsyncWebClient):
raise BoltError(error_client_invalid_type_async())
self._async_client = client
self._token = client.token
if token is not None:
self._framework_logger.warning(
warning_client_prioritized_and_token_skipped()
)
else:
self._async_client = create_async_web_client(
# NOTE: the token here can be None
token=token,
logger=self._framework_logger,
)
# --------------------------------------
# Authorize & OAuthFlow initialization
# --------------------------------------
self._async_authorize: Optional[AsyncAuthorize] = None
if authorize is not None:
if oauth_settings is not None or oauth_flow is not None:
raise BoltError(error_authorize_conflicts())
self._async_authorize = AsyncCallableAuthorize(
logger=self._framework_logger, func=authorize
)
self._async_installation_store: Optional[
AsyncInstallationStore
] = installation_store
if self._async_installation_store is not None and self._async_authorize is None:
self._async_authorize = AsyncInstallationStoreAuthorize(
installation_store=self._async_installation_store,
logger=self._framework_logger,
bot_only=installation_store_bot_only,
)
self._async_oauth_flow: Optional[AsyncOAuthFlow] = None
if (
oauth_settings is None
and os.environ.get("SLACK_CLIENT_ID") is not None
and os.environ.get("SLACK_CLIENT_SECRET") is not None
):
# initialize with the default settings
oauth_settings = AsyncOAuthSettings()
if oauth_flow is None and installation_store is None:
# show info-level log for avoiding confusions
self._framework_logger.info(info_default_oauth_settings_loaded())
if oauth_flow:
if not isinstance(oauth_flow, AsyncOAuthFlow):
raise BoltError(error_oauth_flow_invalid_type_async())
self._async_oauth_flow = oauth_flow
installation_store = select_consistent_installation_store(
client_id=self._async_oauth_flow.client_id,
app_store=self._async_installation_store,
oauth_flow_store=self._async_oauth_flow.settings.installation_store,
logger=self._framework_logger,
)
self._async_installation_store = installation_store
self._async_oauth_flow.settings.installation_store = installation_store
if self._async_oauth_flow._async_client is None:
self._async_oauth_flow._async_client = self._async_client
if self._async_authorize is None:
self._async_authorize = self._async_oauth_flow.settings.authorize
elif oauth_settings is not None:
if not isinstance(oauth_settings, AsyncOAuthSettings):
raise BoltError(error_oauth_settings_invalid_type_async())
installation_store = select_consistent_installation_store(
client_id=oauth_settings.client_id,
app_store=self._async_installation_store,
oauth_flow_store=oauth_settings.installation_store,
logger=self._framework_logger,
)
self._async_installation_store = installation_store
oauth_settings.installation_store = installation_store
self._async_oauth_flow = AsyncOAuthFlow(
client=self._async_client, logger=self.logger, settings=oauth_settings
)
if self._async_authorize is None:
self._async_authorize = self._async_oauth_flow.settings.authorize
if (
self._async_installation_store is not None
or self._async_authorize is not None
) and self._token is not None:
self._token = None
self._framework_logger.warning(warning_token_skipped())
# after setting bot_only here, __init__ cannot replace authorize function
if (
installation_store_bot_only is not None
and self._async_oauth_flow is not None
):
app_bot_only = installation_store_bot_only or False
oauth_flow_bot_only = (
self._async_oauth_flow.settings.installation_store_bot_only
)
if app_bot_only != oauth_flow_bot_only:
self.logger.warning(warning_bot_only_conflicts())
self._async_oauth_flow.settings.installation_store_bot_only = (
app_bot_only
)
self._async_authorize.bot_only = app_bot_only
self._async_tokens_revocation_listeners: Optional[
AsyncTokenRevocationListeners
] = None
if self._async_installation_store is not None:
self._async_tokens_revocation_listeners = AsyncTokenRevocationListeners(
self._async_installation_store
)
# --------------------------------------
# Middleware Initialization
# --------------------------------------
self._async_middleware_list: List[Union[Callable, AsyncMiddleware]] = []
self._async_listeners: List[AsyncListener] = []
self._process_before_response = process_before_response
self._async_listener_runner = AsyncioListenerRunner(
logger=self._framework_logger,
process_before_response=process_before_response,
listener_error_handler=AsyncDefaultListenerErrorHandler(
logger=self._framework_logger
),
listener_completion_handler=AsyncDefaultListenerCompletionHandler(
logger=self._framework_logger
),
lazy_listener_runner=AsyncioLazyListenerRunner(
logger=self._framework_logger,
),
)
self._init_middleware_list_done = False
self._init_async_middleware_list()
self._server: Optional[AsyncSlackAppServer] = None
def _init_async_middleware_list(self):
if self._init_middleware_list_done:
return
self._async_middleware_list.append(
AsyncSslCheck(verification_token=self._verification_token)
)
self._async_middleware_list.append(
AsyncRequestVerification(self._signing_secret)
)
if self._async_oauth_flow is None:
if self._token:
self._async_middleware_list.append(AsyncSingleTeamAuthorization())
elif self._async_authorize is not None:
self._async_middleware_list.append(
AsyncMultiTeamsAuthorization(authorize=self._async_authorize)
)
else:
raise BoltError(error_token_required())
else:
self._async_middleware_list.append(
AsyncMultiTeamsAuthorization(authorize=self._async_authorize)
)
self._async_middleware_list.append(AsyncIgnoringSelfEvents())
self._async_middleware_list.append(AsyncUrlVerification())
self._init_middleware_list_done = True
# -------------------------
# accessors
@property
def name(self) -> str:
"""The name of this app (default: the filename)"""
return self._name
@property
def oauth_flow(self) -> Optional[AsyncOAuthFlow]:
"""Configured `OAuthFlow` object if exists."""
return self._async_oauth_flow
@property
def client(self) -> AsyncWebClient:
"""The singleton `slack_sdk.web.async_client.AsyncWebClient` instance in this app."""
return self._async_client
@property
def logger(self) -> logging.Logger:
"""The logger this app uses."""
return self._framework_logger
@property
def installation_store(self) -> Optional[AsyncInstallationStore]:
"""The `slack_sdk.oauth.AsyncInstallationStore` that can be used in the `authorize` middleware."""
return self._async_installation_store
@property
def listener_runner(self) -> AsyncioListenerRunner:
"""The asyncio-based executor for asynchronously running listeners."""
return self._async_listener_runner
@property
def process_before_response(self) -> bool:
return self._process_before_response or False
# -------------------------
# standalone server
from .async_server import AsyncSlackAppServer
def server(
self, port: int = 3000, path: str = "/slack/events"
) -> AsyncSlackAppServer:
"""Configure a web server using AIOHTTP.
Refer to https://docs.aiohttp.org/ for more details about AIOHTTP.
Args:
port: The port to listen on (Default: 3000)
path:The path to handle request from Slack (Default: `/slack/events`)
"""
if (
self._server is None
or self._server.port != port
or self._server.path != path
):
self._server = AsyncSlackAppServer(
port=port,
path=path,
app=self,
)
return self._server
def web_app(self, path: str = "/slack/events") -> web.Application:
"""Returns a `web.Application` instance for aiohttp-devtools users.
from slack_bolt.async_app import AsyncApp
app = AsyncApp()
@app.event("app_mention")
async def event_test(body, say, logger):
logger.info(body)
await say("What's up?")
def app_factory():
return app.web_app()
# adev runserver --port 3000 --app-factory app_factory async_app.py
Args:
path: The path to receive incoming requests from Slack
"""
return self.server(path=path).web_app
def start(self, port: int = 3000, path: str = "/slack/events") -> None:
"""Start a web server using AIOHTTP.
Refer to https://docs.aiohttp.org/ for more details about AIOHTTP.
Args:
port: The port to listen on (Default: 3000)
path:The path to handle request from Slack (Default: `/slack/events`)
"""
self.server(port=port, path=path).start()
# -------------------------
# main dispatcher
async def async_dispatch(self, req: AsyncBoltRequest) -> BoltResponse:
"""Applies all middleware and dispatches an incoming request from Slack to the right code path.
Args:
req: An incoming request from Slack.
Returns:
The response generated by this Bolt app.
"""
starting_time = time.time()
self._init_context(req)
resp: BoltResponse = BoltResponse(status=200, body="")
middleware_state = {"next_called": False}
async def async_middleware_next():
middleware_state["next_called"] = True
for middleware in self._async_middleware_list:
middleware_state["next_called"] = False
if self._framework_logger.level <= logging.DEBUG:
self._framework_logger.debug(f"Applying {middleware.name}")
resp = await middleware.async_process(
req=req, resp=resp, next=async_middleware_next
)
if not middleware_state["next_called"]:
if resp is None:
# next() method was not called without providing the response to return to Slack
# This should not be an intentional handling in usual use cases.
resp = BoltResponse(
status=404, body={"error": "no next() calls in middleware"}
)
if self._raise_error_for_unhandled_request is True:
await self._async_listener_runner.listener_error_handler.handle(
error=BoltUnhandledRequestError(
request=req,
current_response=resp,
last_global_middleware_name=middleware.name,
),
request=req,
response=resp,
)
return resp
self._framework_logger.warning(
warning_unhandled_by_global_middleware(middleware.name, req)
)
return resp
return resp
for listener in self._async_listeners:
listener_name = get_name_for_callable(listener.ack_function)
self._framework_logger.debug(debug_checking_listener(listener_name))
if await listener.async_matches(req=req, resp=resp):
# run all the middleware attached to this listener first
(
middleware_resp,
next_was_not_called,
) = await listener.run_async_middleware(req=req, resp=resp)
if next_was_not_called:
if middleware_resp is not None:
if self._framework_logger.level <= logging.DEBUG:
debug_message = debug_return_listener_middleware_response(
listener_name,
middleware_resp.status,
middleware_resp.body,
starting_time,
)
self._framework_logger.debug(debug_message)
return middleware_resp
# The last listener middleware didn't call next() method.
# This means the listener is not for this incoming request.
continue
if middleware_resp is not None:
resp = middleware_resp
self._framework_logger.debug(debug_running_listener(listener_name))
listener_response: Optional[
BoltResponse
] = await self._async_listener_runner.run(
request=req,
response=resp,
listener_name=listener_name,
listener=listener,
)
if listener_response is not None:
return listener_response
if resp is None:
resp = BoltResponse(status=404, body={"error": "unhandled request"})
if self._raise_error_for_unhandled_request is True:
await self._async_listener_runner.listener_error_handler.handle(
error=BoltUnhandledRequestError(
request=req,
current_response=resp,
),
request=req,
response=resp,
)
return resp
return self._handle_unmatched_requests(req, resp)
def _handle_unmatched_requests(
self, req: AsyncBoltRequest, resp: BoltResponse
) -> BoltResponse:
# TODO: provide more info like suggestion of listeners
# e.g., You can handle this type of message with @app.event("app_mention")
self._framework_logger.warning(warning_unhandled_request(req))
return resp
# -------------------------
# middleware
def use(self, *args) -> Optional[Callable]:
"""Refer to `AsyncApp#middleware()` method's docstring for details."""
return self.middleware(*args)
def middleware(self, *args) -> Optional[Callable]:
"""Registers a new middleware to this app.
This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.middleware
async def middleware_func(logger, body, next):
logger.info(f"request body: {body}")
await next()
# Pass a function to this method
app.middleware(middleware_func)
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
*args: A function that works as a global middleware.
"""
if len(args) > 0:
middleware_or_callable = args[0]
if isinstance(middleware_or_callable, AsyncMiddleware):
self._async_middleware_list.append(middleware_or_callable)
elif isinstance(middleware_or_callable, Callable):
self._async_middleware_list.append(
AsyncCustomMiddleware(
app_name=self.name, func=middleware_or_callable
)
)
return middleware_or_callable
else:
raise BoltError(
f"Unexpected type for a middleware ({type(middleware_or_callable)})"
)
return None
# -------------------------
# Workflows: Steps from Apps
def step(
self,
callback_id: Union[str, Pattern, AsyncWorkflowStep, AsyncWorkflowStepBuilder],
edit: Optional[
Union[
Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable]
]
] = None,
save: Optional[
Union[
Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable]
]
] = None,
execute: Optional[
Union[
Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable]
]
] = None,
):
"""
Registers a new Workflow Step listener.
Unlike others, this method doesn't behave as a decorator.
If you want to register a workflow step by a decorator, use `AsyncWorkflowStepBuilder`'s methods.
# Create a new WorkflowStep instance
from slack_bolt.workflows.async_step import AsyncWorkflowStep
ws = AsyncWorkflowStep(
callback_id="add_task",
edit=edit,
save=save,
execute=execute,
)
# Pass Step to set up listeners
app.step(ws)
Refer to https://api.slack.com/workflows/steps for details of Steps from Apps.
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
For further information about AsyncWorkflowStep specific function arguments
such as `configure`, `update`, `complete`, and `fail`,
refer to the `async` prefixed ones in `slack_bolt.workflows.step.utilities` API documents.
Args:
callback_id: The Callback ID for this workflow step
edit: The function for displaying a modal in the Workflow Builder
save: The function for handling configuration in the Workflow Builder
execute: The function for handling the step execution
"""
step = callback_id
if isinstance(callback_id, (str, Pattern)):
step = AsyncWorkflowStep(
callback_id=callback_id,
edit=edit,
save=save,
execute=execute,
)
elif isinstance(step, AsyncWorkflowStepBuilder):
step = step.build()
elif not isinstance(step, AsyncWorkflowStep):
raise BoltError(f"Invalid step object ({type(step)})")
self.use(AsyncWorkflowStepMiddleware(step, self._async_listener_runner))
# -------------------------
# global error handler
def error(
self, func: Callable[..., Awaitable[Optional[BoltResponse]]]
) -> Callable[..., Awaitable[Optional[BoltResponse]]]:
"""Updates the global error handler. This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.error
async def custom_error_handler(error, body, logger):
logger.exception(f"Error: {error}")
logger.info(f"Request body: {body}")
# Pass a function to this method
app.error(custom_error_handler)
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
func: The function that is supposed to be executed
when getting an unhandled error in Bolt app.
"""
if not inspect.iscoroutinefunction(func):
name = get_name_for_callable(func)
raise BoltError(error_listener_function_must_be_coro_func(name))
self._async_listener_runner.listener_error_handler = (
AsyncCustomListenerErrorHandler(
logger=self._framework_logger,
func=func,
)
)
return func
# -------------------------
# events
def event(
self,
event: Union[
str, Pattern, Dict[str, Union[str, Sequence[Optional[Union[str, Pattern]]]]]
],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new event listener. This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.event("team_join")
async def ask_for_introduction(event, say):
welcome_channel_id = "C12345"
user_id = event["user"]
text = f"Welcome to the team, <@{user_id}>! :tada: You can introduce yourself in this channel."
await say(text=text, channel=welcome_channel_id)
# Pass a function to this method
app.event("team_join")(ask_for_introduction)
Refer to https://api.slack.com/apis/connections/events-api for details of Events API.
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
event: The conditions that match a request payload.
If you pass a dict for this, you can have type, subtype in the constraint.
matchers: A list of listener matcher functions.
Only when all the matchers return True, the listener function can be invoked.
middleware: A list of lister middleware functions.
Only when all the middleware call `next()` method, the listener function can be invoked.
"""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.event(event, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware, True
)
return __call__
def message(
self,
keyword: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new message event listener. This method can be used as either a decorator or a method.
Check the `App#event` method's docstring for details.
# Use this method as a decorator
@app.message(":wave:")
async def say_hello(message, say):
user = message['user']
await say(f"Hi there, <@{user}>!")
# Pass a function to this method
app.message(":wave:")(say_hello)
Refer to https://api.slack.com/events/message for details of `message` events.
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
keyword: The keyword to match
matchers: A list of listener matcher functions.
Only when all the matchers return True, the listener function can be invoked.
middleware: A list of lister middleware functions.
Only when all the middleware call `next()` method, the listener function can be invoked.
"""
matchers = list(matchers) if matchers else []
middleware = list(middleware) if middleware else []
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
# As of Jan 2021, most bot messages no longer have the subtype bot_message.
# By contrast, messages posted using classic app's bot token still have the subtype.
constraints = {"type": "message", "subtype": (None, "bot_message")}
primary_matcher = builtin_matchers.event(
constraints=constraints, asyncio=True
)
middleware.insert(0, AsyncMessageListenerMatches(keyword))
return self._register_listener(
list(functions), primary_matcher, matchers, middleware, True
)
return __call__
# -------------------------
# slash commands
def command(
self,
command: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new slash command listener.
This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.command("/echo")
async def repeat_text(ack, say, command):
# Acknowledge command request
await ack()
await say(f"{command['text']}")
# Pass a function to this method
app.command("/echo")(repeat_text)
Refer to https://api.slack.com/interactivity/slash-commands for details of Slash Commands.
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
command: The conditions that match a request payload
matchers: A list of listener matcher functions.
Only when all the matchers return True, the listener function can be invoked.
middleware: A list of lister middleware functions.
Only when all the middleware call `next()` method, the listener function can be invoked.
"""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.command(command, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
# -------------------------
# shortcut
def shortcut(
self,
constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new shortcut listener.
This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.shortcut("open_modal")
async def open_modal(ack, body, client):
# Acknowledge the command request
await ack()
# Call views_open with the built-in client
await client.views_open(
# Pass a valid trigger_id within 3 seconds of receiving it
trigger_id=body["trigger_id"],
# View payload
view={ ... }
)
# Pass a function to this method
app.shortcut("open_modal")(open_modal)
Refer to https://api.slack.com/interactivity/shortcuts for details about Shortcuts.
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
constraints: The conditions that match a request payload.
matchers: A list of listener matcher functions.
Only when all the matchers return True, the listener function can be invoked.
middleware: A list of lister middleware functions.
Only when all the middleware call `next()` method, the listener function can be invoked.
"""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.shortcut(constraints, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def global_shortcut(
self,
callback_id: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new global shortcut listener."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.global_shortcut(callback_id, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def message_shortcut(
self,
callback_id: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new message shortcut listener."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.message_shortcut(callback_id, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
# -------------------------
# action
def action(
self,
constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new action listener. This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.action("approve_button")
async def update_message(ack):
await ack()
# Pass a function to this method
app.action("approve_button")(update_message)
* Refer to https://api.slack.com/reference/interaction-payloads/block-actions for actions in `blocks`.
* Refer to https://api.slack.com/legacy/message-buttons for actions in `attachments`.
* Refer to https://api.slack.com/dialogs for actions in dialogs.
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
constraints: The conditions that match a request payload
matchers: A list of listener matcher functions.
Only when all the matchers return True, the listener function can be invoked.
middleware: A list of lister middleware functions.
Only when all the middleware call `next()` method, the listener function can be invoked.
"""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.action(constraints, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def block_action(
self,
constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `block_actions` action listener.
Refer to https://api.slack.com/reference/interaction-payloads/block-actions for details.
"""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.block_action(constraints, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def attachment_action(
self,
callback_id: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `interactive_message` action listener.
Refer to https://api.slack.com/legacy/message-buttons for details."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.attachment_action(callback_id, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def dialog_submission(
self,
callback_id: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `dialog_submission` listener.
Refer to https://api.slack.com/dialogs for details."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.dialog_submission(callback_id, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def dialog_cancellation(
self,
callback_id: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `dialog_submission` listener.
Refer to https://api.slack.com/dialogs for details."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.dialog_cancellation(callback_id, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
# -------------------------
# view
def view(
self,
constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `view_submission`/`view_closed` event listener.
This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.view("view_1")
async 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:
await ack(response_action="errors", errors=errors)
return
# Acknowledge the view_submission event and close the modal
await ack()
# Do whatever you want with the input data - here we're saving it to a DB
# Pass a function to this method
app.view("view_1")(handle_submission)
Refer to https://api.slack.com/reference/interaction-payloads/views for details of payloads.
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
constraints: The conditions that match a request payload
matchers: A list of listener matcher functions.
Only when all the matchers return True, the listener function can be invoked.
middleware: A list of lister middleware functions.
Only when all the middleware call `next()` method, the listener function can be invoked.
"""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.view(constraints, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def view_submission(
self,
constraints: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `view_submission` listener.
Refer to https://api.slack.com/reference/interaction-payloads/views#view_submission for details."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.view_submission(constraints, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def view_closed(
self,
constraints: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `view_closed` listener.
Refer to https://api.slack.com/reference/interaction-payloads/views#view_closed for details."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.view_closed(constraints, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
# -------------------------
# options
def options(
self,
constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new options listener.
This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.options("menu_selection")
async 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",
},
]
await ack(options=options)
# Pass a function to this method
app.options("menu_selection")(show_menu_options)
Refer to the following documents for details:
* https://api.slack.com/reference/block-kit/block-elements#external_select
* https://api.slack.com/reference/block-kit/block-elements#external_multi_select
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document.
Args:
matchers: A list of listener matcher functions.
Only when all the matchers return True, the listener function can be invoked.
middleware: A list of lister middleware functions.
Only when all the middleware call `next()` method, the listener function can be invoked.
"""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.options(constraints, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def block_suggestion(
self,
action_id: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `block_suggestion` listener."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.block_suggestion(action_id, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
def dialog_suggestion(
self,
callback_id: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None,
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
"""Registers a new `dialog_suggestion` listener.
Refer to https://api.slack.com/dialogs for details."""
def __call__(*args, **kwargs):
functions = self._to_listener_functions(kwargs) if kwargs else list(args)
primary_matcher = builtin_matchers.dialog_suggestion(callback_id, True)
return self._register_listener(
list(functions), primary_matcher, matchers, middleware
)
return __call__
# -------------------------
# built-in listener functions
def default_tokens_revoked_event_listener(
self,
) -> Callable[..., Awaitable[Optional[BoltResponse]]]:
if self._async_tokens_revocation_listeners is None:
raise BoltError(error_installation_store_required_for_builtin_listeners())
return self._async_tokens_revocation_listeners.handle_tokens_revoked_events
def default_app_uninstalled_event_listener(
self,
) -> Callable[..., Awaitable[Optional[BoltResponse]]]:
if self._async_tokens_revocation_listeners is None:
raise BoltError(error_installation_store_required_for_builtin_listeners())
return self._async_tokens_revocation_listeners.handle_app_uninstalled_events
def enable_token_revocation_listeners(self) -> None:
self.event("tokens_revoked")(self.default_tokens_revoked_event_listener())
self.event("app_uninstalled")(self.default_app_uninstalled_event_listener())
# -------------------------
def _init_context(self, req: AsyncBoltRequest):
req.context["logger"] = get_bolt_app_logger(self.name)
req.context["token"] = self._token
if self._token is not None:
# This AsyncWebClient instance can be safely singleton
req.context["client"] = self._async_client
else:
# Set a new dedicated instance for this request
client_per_request: AsyncWebClient = AsyncWebClient(
token=None, # the token will be set later
base_url=self._async_client.base_url,
timeout=self._async_client.timeout,
ssl=self._async_client.ssl,
proxy=self._async_client.proxy,
session=self._async_client.session,
trust_env_in_session=self._async_client.trust_env_in_session,
headers=self._async_client.headers,
team_id=req.context.team_id,
)
req.context["client"] = client_per_request
@staticmethod
def _to_listener_functions(
kwargs: dict,
) -> Optional[Sequence[Callable[..., Awaitable[Optional[BoltResponse]]]]]:
if kwargs:
functions = [kwargs["ack"]]
for sub in kwargs["lazy"]:
functions.append(sub)
return functions
return None
def _register_listener(
self,
functions: Sequence[Callable[..., Awaitable[Optional[BoltResponse]]]],
primary_matcher: AsyncListenerMatcher,
matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]],
middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]],
auto_acknowledgement: bool = False,
) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]:
value_to_return = None
if not isinstance(functions, list):
functions = list(functions)
if len(functions) == 1:
# In the case where the function is registered using decorator,
# the registration should return the original function.
value_to_return = functions[0]
for func in functions:
if not inspect.iscoroutinefunction(func):
name = get_name_for_callable(func)
raise BoltError(error_listener_function_must_be_coro_func(name))
listener_matchers = [
AsyncCustomListenerMatcher(app_name=self.name, func=f)
for f in (matchers or [])
]
listener_matchers.insert(0, primary_matcher)
listener_middleware = []
for m in middleware or []:
if isinstance(m, AsyncMiddleware):
listener_middleware.append(m)
elif isinstance(m, Callable) and inspect.iscoroutinefunction(m):
listener_middleware.append(
AsyncCustomMiddleware(app_name=self.name, func=m)
)
else:
raise ValueError(error_unexpected_listener_middleware(type(m)))
self._async_listeners.append(
AsyncCustomListener(
app_name=self.name,
ack_function=functions.pop(0),
lazy_functions=functions,
matchers=listener_matchers,
middleware=listener_middleware,
auto_acknowledgement=auto_acknowledgement,
)
)
return value_to_returnClasses
class AsyncApp (*, logger: Optional[logging.Logger] = None, name: Optional[str] = None, process_before_response: bool = False, raise_error_for_unhandled_request: bool = False, signing_secret: Optional[str] = None, token: Optional[str] = None, client: Optional[slack_sdk.web.async_client.AsyncWebClient] = None, installation_store: Optional[slack_sdk.oauth.installation_store.async_installation_store.AsyncInstallationStore] = None, installation_store_bot_only: Optional[bool] = None, authorize: Optional[Callable[..., Awaitable[AuthorizeResult]]] = None, oauth_settings: Optional[AsyncOAuthSettings] = None, oauth_flow: Optional[AsyncOAuthFlow] = None, verification_token: Optional[str] = None)-
Bolt App that provides functionalities to register middleware/listeners.
import os from slack_bolt.async_app import AsyncApp # Initializes your app with your bot token and signing secret app = AsyncApp( token=os.environ.get("SLACK_BOT_TOKEN"), signing_secret=os.environ.get("SLACK_SIGNING_SECRET") ) # Listens to incoming messages that contain "hello" @app.message("hello") async def message_hello(message, say): # async function # say() sends a message to the channel where the event was triggered await say(f"Hey there <@{message['user']}>!") # Start your app if __name__ == "__main__": app.start(port=int(os.environ.get("PORT", 3000)))Refer to https://slack.dev/bolt-python/concepts#async for details.
If yoy would like to build an OAuth app for enabling the app to run with multiple workspaces, refer to https://slack.dev/bolt-python/concepts#authenticating-oauth to learn how to configure the app.
Args
logger- The custom logger that can be used in this app.
name- The application name that will be used in logging. If absent, the source file name will be used.
process_before_response- True if this app runs on Function as a Service. (Default: False)
raise_error_for_unhandled_request- True if you want to raise exceptions for unhandled requests and use @app.error listeners instead of the built-in handler, which pints warning logs and returns 404 to Slack (Default: False)
signing_secret- The Signing Secret value used for verifying requests from Slack.
token- The bot/user access token required only for single-workspace app.
client- The singleton
slack_sdk.web.async_client.AsyncWebClientinstance for this app. authorize- The function to authorize an incoming request from Slack by checking if there is a team/user in the installation data.
installation_store- The module offering save/find operations of installation data
installation_store_bot_only- Use
AsyncInstallationStore#async_find_bot()if True (Default: False) oauth_settings- The settings related to Slack app installation flow (OAuth flow)
oauth_flow- Instantiated
slack_bolt.oauth.AsyncOAuthFlow. This is always prioritized over oauth_settings. verification_token- Deprecated verification mechanism. This can used only for ssl_check requests.
Expand source code
class AsyncApp: def __init__( self, *, logger: Optional[logging.Logger] = None, # Used in logger name: Optional[str] = None, # Set True when you run this app on a FaaS platform process_before_response: bool = False, # Set True if you want to handle an unhandled request as an exception raise_error_for_unhandled_request: bool = False, # Basic Information > Credentials > Signing Secret signing_secret: Optional[str] = None, # for single-workspace apps token: Optional[str] = None, client: Optional[AsyncWebClient] = None, # for multi-workspace apps installation_store: Optional[AsyncInstallationStore] = None, # for either only bot scope usage or v1.0.x compatibility installation_store_bot_only: Optional[bool] = None, authorize: Optional[Callable[..., Awaitable[AuthorizeResult]]] = None, # for the OAuth flow oauth_settings: Optional[AsyncOAuthSettings] = None, oauth_flow: Optional[AsyncOAuthFlow] = None, # No need to set (the value is used only in response to ssl_check requests) verification_token: Optional[str] = None, ): """Bolt App that provides functionalities to register middleware/listeners. import os from slack_bolt.async_app import AsyncApp # Initializes your app with your bot token and signing secret app = AsyncApp( token=os.environ.get("SLACK_BOT_TOKEN"), signing_secret=os.environ.get("SLACK_SIGNING_SECRET") ) # Listens to incoming messages that contain "hello" @app.message("hello") async def message_hello(message, say): # async function # say() sends a message to the channel where the event was triggered await say(f"Hey there <@{message['user']}>!") # Start your app if __name__ == "__main__": app.start(port=int(os.environ.get("PORT", 3000))) Refer to https://slack.dev/bolt-python/concepts#async for details. If yoy would like to build an OAuth app for enabling the app to run with multiple workspaces, refer to https://slack.dev/bolt-python/concepts#authenticating-oauth to learn how to configure the app. Args: logger: The custom logger that can be used in this app. name: The application name that will be used in logging. If absent, the source file name will be used. process_before_response: True if this app runs on Function as a Service. (Default: False) raise_error_for_unhandled_request: True if you want to raise exceptions for unhandled requests and use @app.error listeners instead of the built-in handler, which pints warning logs and returns 404 to Slack (Default: False) signing_secret: The Signing Secret value used for verifying requests from Slack. token: The bot/user access token required only for single-workspace app. client: The singleton `slack_sdk.web.async_client.AsyncWebClient` instance for this app. authorize: The function to authorize an incoming request from Slack by checking if there is a team/user in the installation data. installation_store: The module offering save/find operations of installation data installation_store_bot_only: Use `AsyncInstallationStore#async_find_bot()` if True (Default: False) oauth_settings: The settings related to Slack app installation flow (OAuth flow) oauth_flow: Instantiated `slack_bolt.oauth.AsyncOAuthFlow`. This is always prioritized over oauth_settings. verification_token: Deprecated verification mechanism. This can used only for ssl_check requests. """ signing_secret = signing_secret or os.environ.get("SLACK_SIGNING_SECRET") token = token or os.environ.get("SLACK_BOT_TOKEN") self._name: str = name or inspect.stack()[1].filename.split(os.path.sep)[-1] self._signing_secret: str = signing_secret self._verification_token: Optional[str] = verification_token or os.environ.get( "SLACK_VERIFICATION_TOKEN", None ) self._framework_logger = logger or get_bolt_logger(AsyncApp) self._raise_error_for_unhandled_request = raise_error_for_unhandled_request self._token: Optional[str] = token if client is not None: if not isinstance(client, AsyncWebClient): raise BoltError(error_client_invalid_type_async()) self._async_client = client self._token = client.token if token is not None: self._framework_logger.warning( warning_client_prioritized_and_token_skipped() ) else: self._async_client = create_async_web_client( # NOTE: the token here can be None token=token, logger=self._framework_logger, ) # -------------------------------------- # Authorize & OAuthFlow initialization # -------------------------------------- self._async_authorize: Optional[AsyncAuthorize] = None if authorize is not None: if oauth_settings is not None or oauth_flow is not None: raise BoltError(error_authorize_conflicts()) self._async_authorize = AsyncCallableAuthorize( logger=self._framework_logger, func=authorize ) self._async_installation_store: Optional[ AsyncInstallationStore ] = installation_store if self._async_installation_store is not None and self._async_authorize is None: self._async_authorize = AsyncInstallationStoreAuthorize( installation_store=self._async_installation_store, logger=self._framework_logger, bot_only=installation_store_bot_only, ) self._async_oauth_flow: Optional[AsyncOAuthFlow] = None if ( oauth_settings is None and os.environ.get("SLACK_CLIENT_ID") is not None and os.environ.get("SLACK_CLIENT_SECRET") is not None ): # initialize with the default settings oauth_settings = AsyncOAuthSettings() if oauth_flow is None and installation_store is None: # show info-level log for avoiding confusions self._framework_logger.info(info_default_oauth_settings_loaded()) if oauth_flow: if not isinstance(oauth_flow, AsyncOAuthFlow): raise BoltError(error_oauth_flow_invalid_type_async()) self._async_oauth_flow = oauth_flow installation_store = select_consistent_installation_store( client_id=self._async_oauth_flow.client_id, app_store=self._async_installation_store, oauth_flow_store=self._async_oauth_flow.settings.installation_store, logger=self._framework_logger, ) self._async_installation_store = installation_store self._async_oauth_flow.settings.installation_store = installation_store if self._async_oauth_flow._async_client is None: self._async_oauth_flow._async_client = self._async_client if self._async_authorize is None: self._async_authorize = self._async_oauth_flow.settings.authorize elif oauth_settings is not None: if not isinstance(oauth_settings, AsyncOAuthSettings): raise BoltError(error_oauth_settings_invalid_type_async()) installation_store = select_consistent_installation_store( client_id=oauth_settings.client_id, app_store=self._async_installation_store, oauth_flow_store=oauth_settings.installation_store, logger=self._framework_logger, ) self._async_installation_store = installation_store oauth_settings.installation_store = installation_store self._async_oauth_flow = AsyncOAuthFlow( client=self._async_client, logger=self.logger, settings=oauth_settings ) if self._async_authorize is None: self._async_authorize = self._async_oauth_flow.settings.authorize if ( self._async_installation_store is not None or self._async_authorize is not None ) and self._token is not None: self._token = None self._framework_logger.warning(warning_token_skipped()) # after setting bot_only here, __init__ cannot replace authorize function if ( installation_store_bot_only is not None and self._async_oauth_flow is not None ): app_bot_only = installation_store_bot_only or False oauth_flow_bot_only = ( self._async_oauth_flow.settings.installation_store_bot_only ) if app_bot_only != oauth_flow_bot_only: self.logger.warning(warning_bot_only_conflicts()) self._async_oauth_flow.settings.installation_store_bot_only = ( app_bot_only ) self._async_authorize.bot_only = app_bot_only self._async_tokens_revocation_listeners: Optional[ AsyncTokenRevocationListeners ] = None if self._async_installation_store is not None: self._async_tokens_revocation_listeners = AsyncTokenRevocationListeners( self._async_installation_store ) # -------------------------------------- # Middleware Initialization # -------------------------------------- self._async_middleware_list: List[Union[Callable, AsyncMiddleware]] = [] self._async_listeners: List[AsyncListener] = [] self._process_before_response = process_before_response self._async_listener_runner = AsyncioListenerRunner( logger=self._framework_logger, process_before_response=process_before_response, listener_error_handler=AsyncDefaultListenerErrorHandler( logger=self._framework_logger ), listener_completion_handler=AsyncDefaultListenerCompletionHandler( logger=self._framework_logger ), lazy_listener_runner=AsyncioLazyListenerRunner( logger=self._framework_logger, ), ) self._init_middleware_list_done = False self._init_async_middleware_list() self._server: Optional[AsyncSlackAppServer] = None def _init_async_middleware_list(self): if self._init_middleware_list_done: return self._async_middleware_list.append( AsyncSslCheck(verification_token=self._verification_token) ) self._async_middleware_list.append( AsyncRequestVerification(self._signing_secret) ) if self._async_oauth_flow is None: if self._token: self._async_middleware_list.append(AsyncSingleTeamAuthorization()) elif self._async_authorize is not None: self._async_middleware_list.append( AsyncMultiTeamsAuthorization(authorize=self._async_authorize) ) else: raise BoltError(error_token_required()) else: self._async_middleware_list.append( AsyncMultiTeamsAuthorization(authorize=self._async_authorize) ) self._async_middleware_list.append(AsyncIgnoringSelfEvents()) self._async_middleware_list.append(AsyncUrlVerification()) self._init_middleware_list_done = True # ------------------------- # accessors @property def name(self) -> str: """The name of this app (default: the filename)""" return self._name @property def oauth_flow(self) -> Optional[AsyncOAuthFlow]: """Configured `OAuthFlow` object if exists.""" return self._async_oauth_flow @property def client(self) -> AsyncWebClient: """The singleton `slack_sdk.web.async_client.AsyncWebClient` instance in this app.""" return self._async_client @property def logger(self) -> logging.Logger: """The logger this app uses.""" return self._framework_logger @property def installation_store(self) -> Optional[AsyncInstallationStore]: """The `slack_sdk.oauth.AsyncInstallationStore` that can be used in the `authorize` middleware.""" return self._async_installation_store @property def listener_runner(self) -> AsyncioListenerRunner: """The asyncio-based executor for asynchronously running listeners.""" return self._async_listener_runner @property def process_before_response(self) -> bool: return self._process_before_response or False # ------------------------- # standalone server from .async_server import AsyncSlackAppServer def server( self, port: int = 3000, path: str = "/slack/events" ) -> AsyncSlackAppServer: """Configure a web server using AIOHTTP. Refer to https://docs.aiohttp.org/ for more details about AIOHTTP. Args: port: The port to listen on (Default: 3000) path:The path to handle request from Slack (Default: `/slack/events`) """ if ( self._server is None or self._server.port != port or self._server.path != path ): self._server = AsyncSlackAppServer( port=port, path=path, app=self, ) return self._server def web_app(self, path: str = "/slack/events") -> web.Application: """Returns a `web.Application` instance for aiohttp-devtools users. from slack_bolt.async_app import AsyncApp app = AsyncApp() @app.event("app_mention") async def event_test(body, say, logger): logger.info(body) await say("What's up?") def app_factory(): return app.web_app() # adev runserver --port 3000 --app-factory app_factory async_app.py Args: path: The path to receive incoming requests from Slack """ return self.server(path=path).web_app def start(self, port: int = 3000, path: str = "/slack/events") -> None: """Start a web server using AIOHTTP. Refer to https://docs.aiohttp.org/ for more details about AIOHTTP. Args: port: The port to listen on (Default: 3000) path:The path to handle request from Slack (Default: `/slack/events`) """ self.server(port=port, path=path).start() # ------------------------- # main dispatcher async def async_dispatch(self, req: AsyncBoltRequest) -> BoltResponse: """Applies all middleware and dispatches an incoming request from Slack to the right code path. Args: req: An incoming request from Slack. Returns: The response generated by this Bolt app. """ starting_time = time.time() self._init_context(req) resp: BoltResponse = BoltResponse(status=200, body="") middleware_state = {"next_called": False} async def async_middleware_next(): middleware_state["next_called"] = True for middleware in self._async_middleware_list: middleware_state["next_called"] = False if self._framework_logger.level <= logging.DEBUG: self._framework_logger.debug(f"Applying {middleware.name}") resp = await middleware.async_process( req=req, resp=resp, next=async_middleware_next ) if not middleware_state["next_called"]: if resp is None: # next() method was not called without providing the response to return to Slack # This should not be an intentional handling in usual use cases. resp = BoltResponse( status=404, body={"error": "no next() calls in middleware"} ) if self._raise_error_for_unhandled_request is True: await self._async_listener_runner.listener_error_handler.handle( error=BoltUnhandledRequestError( request=req, current_response=resp, last_global_middleware_name=middleware.name, ), request=req, response=resp, ) return resp self._framework_logger.warning( warning_unhandled_by_global_middleware(middleware.name, req) ) return resp return resp for listener in self._async_listeners: listener_name = get_name_for_callable(listener.ack_function) self._framework_logger.debug(debug_checking_listener(listener_name)) if await listener.async_matches(req=req, resp=resp): # run all the middleware attached to this listener first ( middleware_resp, next_was_not_called, ) = await listener.run_async_middleware(req=req, resp=resp) if next_was_not_called: if middleware_resp is not None: if self._framework_logger.level <= logging.DEBUG: debug_message = debug_return_listener_middleware_response( listener_name, middleware_resp.status, middleware_resp.body, starting_time, ) self._framework_logger.debug(debug_message) return middleware_resp # The last listener middleware didn't call next() method. # This means the listener is not for this incoming request. continue if middleware_resp is not None: resp = middleware_resp self._framework_logger.debug(debug_running_listener(listener_name)) listener_response: Optional[ BoltResponse ] = await self._async_listener_runner.run( request=req, response=resp, listener_name=listener_name, listener=listener, ) if listener_response is not None: return listener_response if resp is None: resp = BoltResponse(status=404, body={"error": "unhandled request"}) if self._raise_error_for_unhandled_request is True: await self._async_listener_runner.listener_error_handler.handle( error=BoltUnhandledRequestError( request=req, current_response=resp, ), request=req, response=resp, ) return resp return self._handle_unmatched_requests(req, resp) def _handle_unmatched_requests( self, req: AsyncBoltRequest, resp: BoltResponse ) -> BoltResponse: # TODO: provide more info like suggestion of listeners # e.g., You can handle this type of message with @app.event("app_mention") self._framework_logger.warning(warning_unhandled_request(req)) return resp # ------------------------- # middleware def use(self, *args) -> Optional[Callable]: """Refer to `AsyncApp#middleware()` method's docstring for details.""" return self.middleware(*args) def middleware(self, *args) -> Optional[Callable]: """Registers a new middleware to this app. This method can be used as either a decorator or a method. # Use this method as a decorator @app.middleware async def middleware_func(logger, body, next): logger.info(f"request body: {body}") await next() # Pass a function to this method app.middleware(middleware_func) To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: *args: A function that works as a global middleware. """ if len(args) > 0: middleware_or_callable = args[0] if isinstance(middleware_or_callable, AsyncMiddleware): self._async_middleware_list.append(middleware_or_callable) elif isinstance(middleware_or_callable, Callable): self._async_middleware_list.append( AsyncCustomMiddleware( app_name=self.name, func=middleware_or_callable ) ) return middleware_or_callable else: raise BoltError( f"Unexpected type for a middleware ({type(middleware_or_callable)})" ) return None # ------------------------- # Workflows: Steps from Apps def step( self, callback_id: Union[str, Pattern, AsyncWorkflowStep, AsyncWorkflowStepBuilder], edit: Optional[ Union[ Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable] ] ] = None, save: Optional[ Union[ Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable] ] ] = None, execute: Optional[ Union[ Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable] ] ] = None, ): """ Registers a new Workflow Step listener. Unlike others, this method doesn't behave as a decorator. If you want to register a workflow step by a decorator, use `AsyncWorkflowStepBuilder`'s methods. # Create a new WorkflowStep instance from slack_bolt.workflows.async_step import AsyncWorkflowStep ws = AsyncWorkflowStep( callback_id="add_task", edit=edit, save=save, execute=execute, ) # Pass Step to set up listeners app.step(ws) Refer to https://api.slack.com/workflows/steps for details of Steps from Apps. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. For further information about AsyncWorkflowStep specific function arguments such as `configure`, `update`, `complete`, and `fail`, refer to the `async` prefixed ones in `slack_bolt.workflows.step.utilities` API documents. Args: callback_id: The Callback ID for this workflow step edit: The function for displaying a modal in the Workflow Builder save: The function for handling configuration in the Workflow Builder execute: The function for handling the step execution """ step = callback_id if isinstance(callback_id, (str, Pattern)): step = AsyncWorkflowStep( callback_id=callback_id, edit=edit, save=save, execute=execute, ) elif isinstance(step, AsyncWorkflowStepBuilder): step = step.build() elif not isinstance(step, AsyncWorkflowStep): raise BoltError(f"Invalid step object ({type(step)})") self.use(AsyncWorkflowStepMiddleware(step, self._async_listener_runner)) # ------------------------- # global error handler def error( self, func: Callable[..., Awaitable[Optional[BoltResponse]]] ) -> Callable[..., Awaitable[Optional[BoltResponse]]]: """Updates the global error handler. This method can be used as either a decorator or a method. # Use this method as a decorator @app.error async def custom_error_handler(error, body, logger): logger.exception(f"Error: {error}") logger.info(f"Request body: {body}") # Pass a function to this method app.error(custom_error_handler) To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: func: The function that is supposed to be executed when getting an unhandled error in Bolt app. """ if not inspect.iscoroutinefunction(func): name = get_name_for_callable(func) raise BoltError(error_listener_function_must_be_coro_func(name)) self._async_listener_runner.listener_error_handler = ( AsyncCustomListenerErrorHandler( logger=self._framework_logger, func=func, ) ) return func # ------------------------- # events def event( self, event: Union[ str, Pattern, Dict[str, Union[str, Sequence[Optional[Union[str, Pattern]]]]] ], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new event listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.event("team_join") async def ask_for_introduction(event, say): welcome_channel_id = "C12345" user_id = event["user"] text = f"Welcome to the team, <@{user_id}>! :tada: You can introduce yourself in this channel." await say(text=text, channel=welcome_channel_id) # Pass a function to this method app.event("team_join")(ask_for_introduction) Refer to https://api.slack.com/apis/connections/events-api for details of Events API. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: event: The conditions that match a request payload. If you pass a dict for this, you can have type, subtype in the constraint. matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.event(event, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware, True ) return __call__ def message( self, keyword: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new message event listener. This method can be used as either a decorator or a method. Check the `App#event` method's docstring for details. # Use this method as a decorator @app.message(":wave:") async def say_hello(message, say): user = message['user'] await say(f"Hi there, <@{user}>!") # Pass a function to this method app.message(":wave:")(say_hello) Refer to https://api.slack.com/events/message for details of `message` events. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: keyword: The keyword to match matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ matchers = list(matchers) if matchers else [] middleware = list(middleware) if middleware else [] def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) # As of Jan 2021, most bot messages no longer have the subtype bot_message. # By contrast, messages posted using classic app's bot token still have the subtype. constraints = {"type": "message", "subtype": (None, "bot_message")} primary_matcher = builtin_matchers.event( constraints=constraints, asyncio=True ) middleware.insert(0, AsyncMessageListenerMatches(keyword)) return self._register_listener( list(functions), primary_matcher, matchers, middleware, True ) return __call__ # ------------------------- # slash commands def command( self, command: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new slash command listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.command("/echo") async def repeat_text(ack, say, command): # Acknowledge command request await ack() await say(f"{command['text']}") # Pass a function to this method app.command("/echo")(repeat_text) Refer to https://api.slack.com/interactivity/slash-commands for details of Slash Commands. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: command: The conditions that match a request payload matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.command(command, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ # ------------------------- # shortcut def shortcut( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new shortcut listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.shortcut("open_modal") async def open_modal(ack, body, client): # Acknowledge the command request await ack() # Call views_open with the built-in client await client.views_open( # Pass a valid trigger_id within 3 seconds of receiving it trigger_id=body["trigger_id"], # View payload view={ ... } ) # Pass a function to this method app.shortcut("open_modal")(open_modal) Refer to https://api.slack.com/interactivity/shortcuts for details about Shortcuts. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: constraints: The conditions that match a request payload. matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.shortcut(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def global_shortcut( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new global shortcut listener.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.global_shortcut(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def message_shortcut( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new message shortcut listener.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.message_shortcut(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ # ------------------------- # action def action( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new action listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.action("approve_button") async def update_message(ack): await ack() # Pass a function to this method app.action("approve_button")(update_message) * Refer to https://api.slack.com/reference/interaction-payloads/block-actions for actions in `blocks`. * Refer to https://api.slack.com/legacy/message-buttons for actions in `attachments`. * Refer to https://api.slack.com/dialogs for actions in dialogs. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: constraints: The conditions that match a request payload matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.action(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def block_action( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `block_actions` action listener. Refer to https://api.slack.com/reference/interaction-payloads/block-actions for details. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.block_action(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def attachment_action( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `interactive_message` action listener. Refer to https://api.slack.com/legacy/message-buttons for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.attachment_action(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def dialog_submission( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `dialog_submission` listener. Refer to https://api.slack.com/dialogs for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.dialog_submission(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def dialog_cancellation( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `dialog_submission` listener. Refer to https://api.slack.com/dialogs for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.dialog_cancellation(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ # ------------------------- # view def view( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `view_submission`/`view_closed` event listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.view("view_1") async 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: await ack(response_action="errors", errors=errors) return # Acknowledge the view_submission event and close the modal await ack() # Do whatever you want with the input data - here we're saving it to a DB # Pass a function to this method app.view("view_1")(handle_submission) Refer to https://api.slack.com/reference/interaction-payloads/views for details of payloads. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: constraints: The conditions that match a request payload matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.view(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def view_submission( self, constraints: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `view_submission` listener. Refer to https://api.slack.com/reference/interaction-payloads/views#view_submission for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.view_submission(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def view_closed( self, constraints: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `view_closed` listener. Refer to https://api.slack.com/reference/interaction-payloads/views#view_closed for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.view_closed(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ # ------------------------- # options def options( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new options listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.options("menu_selection") async 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", }, ] await ack(options=options) # Pass a function to this method app.options("menu_selection")(show_menu_options) Refer to the following documents for details: * https://api.slack.com/reference/block-kit/block-elements#external_select * https://api.slack.com/reference/block-kit/block-elements#external_multi_select To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.options(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def block_suggestion( self, action_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `block_suggestion` listener.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.block_suggestion(action_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def dialog_suggestion( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `dialog_suggestion` listener. Refer to https://api.slack.com/dialogs for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.dialog_suggestion(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ # ------------------------- # built-in listener functions def default_tokens_revoked_event_listener( self, ) -> Callable[..., Awaitable[Optional[BoltResponse]]]: if self._async_tokens_revocation_listeners is None: raise BoltError(error_installation_store_required_for_builtin_listeners()) return self._async_tokens_revocation_listeners.handle_tokens_revoked_events def default_app_uninstalled_event_listener( self, ) -> Callable[..., Awaitable[Optional[BoltResponse]]]: if self._async_tokens_revocation_listeners is None: raise BoltError(error_installation_store_required_for_builtin_listeners()) return self._async_tokens_revocation_listeners.handle_app_uninstalled_events def enable_token_revocation_listeners(self) -> None: self.event("tokens_revoked")(self.default_tokens_revoked_event_listener()) self.event("app_uninstalled")(self.default_app_uninstalled_event_listener()) # ------------------------- def _init_context(self, req: AsyncBoltRequest): req.context["logger"] = get_bolt_app_logger(self.name) req.context["token"] = self._token if self._token is not None: # This AsyncWebClient instance can be safely singleton req.context["client"] = self._async_client else: # Set a new dedicated instance for this request client_per_request: AsyncWebClient = AsyncWebClient( token=None, # the token will be set later base_url=self._async_client.base_url, timeout=self._async_client.timeout, ssl=self._async_client.ssl, proxy=self._async_client.proxy, session=self._async_client.session, trust_env_in_session=self._async_client.trust_env_in_session, headers=self._async_client.headers, team_id=req.context.team_id, ) req.context["client"] = client_per_request @staticmethod def _to_listener_functions( kwargs: dict, ) -> Optional[Sequence[Callable[..., Awaitable[Optional[BoltResponse]]]]]: if kwargs: functions = [kwargs["ack"]] for sub in kwargs["lazy"]: functions.append(sub) return functions return None def _register_listener( self, functions: Sequence[Callable[..., Awaitable[Optional[BoltResponse]]]], primary_matcher: AsyncListenerMatcher, matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]], middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]], auto_acknowledgement: bool = False, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: value_to_return = None if not isinstance(functions, list): functions = list(functions) if len(functions) == 1: # In the case where the function is registered using decorator, # the registration should return the original function. value_to_return = functions[0] for func in functions: if not inspect.iscoroutinefunction(func): name = get_name_for_callable(func) raise BoltError(error_listener_function_must_be_coro_func(name)) listener_matchers = [ AsyncCustomListenerMatcher(app_name=self.name, func=f) for f in (matchers or []) ] listener_matchers.insert(0, primary_matcher) listener_middleware = [] for m in middleware or []: if isinstance(m, AsyncMiddleware): listener_middleware.append(m) elif isinstance(m, Callable) and inspect.iscoroutinefunction(m): listener_middleware.append( AsyncCustomMiddleware(app_name=self.name, func=m) ) else: raise ValueError(error_unexpected_listener_middleware(type(m))) self._async_listeners.append( AsyncCustomListener( app_name=self.name, ack_function=functions.pop(0), lazy_functions=functions, matchers=listener_matchers, middleware=listener_middleware, auto_acknowledgement=auto_acknowledgement, ) ) return value_to_returnClass variables
var AsyncSlackAppServer
Instance variables
var client : slack_sdk.web.async_client.AsyncWebClient-
The singleton
slack_sdk.web.async_client.AsyncWebClientinstance in this app.Expand source code
@property def client(self) -> AsyncWebClient: """The singleton `slack_sdk.web.async_client.AsyncWebClient` instance in this app.""" return self._async_client var installation_store : Optional[slack_sdk.oauth.installation_store.async_installation_store.AsyncInstallationStore]-
The
slack_sdk.oauth.AsyncInstallationStorethat can be used in theauthorizemiddleware.Expand source code
@property def installation_store(self) -> Optional[AsyncInstallationStore]: """The `slack_sdk.oauth.AsyncInstallationStore` that can be used in the `authorize` middleware.""" return self._async_installation_store var listener_runner : AsyncioListenerRunner-
The asyncio-based executor for asynchronously running listeners.
Expand source code
@property def listener_runner(self) -> AsyncioListenerRunner: """The asyncio-based executor for asynchronously running listeners.""" return self._async_listener_runner var logger : logging.Logger-
The logger this app uses.
Expand source code
@property def logger(self) -> logging.Logger: """The logger this app uses.""" return self._framework_logger var name : str-
The name of this app (default: the filename)
Expand source code
@property def name(self) -> str: """The name of this app (default: the filename)""" return self._name var oauth_flow : Optional[AsyncOAuthFlow]-
Configured
OAuthFlowobject if exists.Expand source code
@property def oauth_flow(self) -> Optional[AsyncOAuthFlow]: """Configured `OAuthFlow` object if exists.""" return self._async_oauth_flow var process_before_response : bool-
Expand source code
@property def process_before_response(self) -> bool: return self._process_before_response or False
Methods
def action(self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new action listener. This method can be used as either a decorator or a method.
# Use this method as a decorator @app.action("approve_button") async def update_message(ack): await ack() # Pass a function to this method app.action("approve_button")(update_message)- Refer to https://api.slack.com/reference/interaction-payloads/block-actions for actions in
blocks. - Refer to https://api.slack.com/legacy/message-buttons for actions in
attachments. - Refer to https://api.slack.com/dialogs for actions in dialogs.
To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
constraints- The conditions that match a request payload
matchers- A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked.
middleware- A list of lister middleware functions.
Only when all the middleware call
next()method, the listener function can be invoked.
Expand source code
def action( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new action listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.action("approve_button") async def update_message(ack): await ack() # Pass a function to this method app.action("approve_button")(update_message) * Refer to https://api.slack.com/reference/interaction-payloads/block-actions for actions in `blocks`. * Refer to https://api.slack.com/legacy/message-buttons for actions in `attachments`. * Refer to https://api.slack.com/dialogs for actions in dialogs. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: constraints: The conditions that match a request payload matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.action(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ - Refer to https://api.slack.com/reference/interaction-payloads/block-actions for actions in
async def async_dispatch(self, req: AsyncBoltRequest) ‑> BoltResponse-
Applies all middleware and dispatches an incoming request from Slack to the right code path.
Args
req- An incoming request from Slack.
Returns
The response generated by this Bolt app.
Expand source code
async def async_dispatch(self, req: AsyncBoltRequest) -> BoltResponse: """Applies all middleware and dispatches an incoming request from Slack to the right code path. Args: req: An incoming request from Slack. Returns: The response generated by this Bolt app. """ starting_time = time.time() self._init_context(req) resp: BoltResponse = BoltResponse(status=200, body="") middleware_state = {"next_called": False} async def async_middleware_next(): middleware_state["next_called"] = True for middleware in self._async_middleware_list: middleware_state["next_called"] = False if self._framework_logger.level <= logging.DEBUG: self._framework_logger.debug(f"Applying {middleware.name}") resp = await middleware.async_process( req=req, resp=resp, next=async_middleware_next ) if not middleware_state["next_called"]: if resp is None: # next() method was not called without providing the response to return to Slack # This should not be an intentional handling in usual use cases. resp = BoltResponse( status=404, body={"error": "no next() calls in middleware"} ) if self._raise_error_for_unhandled_request is True: await self._async_listener_runner.listener_error_handler.handle( error=BoltUnhandledRequestError( request=req, current_response=resp, last_global_middleware_name=middleware.name, ), request=req, response=resp, ) return resp self._framework_logger.warning( warning_unhandled_by_global_middleware(middleware.name, req) ) return resp return resp for listener in self._async_listeners: listener_name = get_name_for_callable(listener.ack_function) self._framework_logger.debug(debug_checking_listener(listener_name)) if await listener.async_matches(req=req, resp=resp): # run all the middleware attached to this listener first ( middleware_resp, next_was_not_called, ) = await listener.run_async_middleware(req=req, resp=resp) if next_was_not_called: if middleware_resp is not None: if self._framework_logger.level <= logging.DEBUG: debug_message = debug_return_listener_middleware_response( listener_name, middleware_resp.status, middleware_resp.body, starting_time, ) self._framework_logger.debug(debug_message) return middleware_resp # The last listener middleware didn't call next() method. # This means the listener is not for this incoming request. continue if middleware_resp is not None: resp = middleware_resp self._framework_logger.debug(debug_running_listener(listener_name)) listener_response: Optional[ BoltResponse ] = await self._async_listener_runner.run( request=req, response=resp, listener_name=listener_name, listener=listener, ) if listener_response is not None: return listener_response if resp is None: resp = BoltResponse(status=404, body={"error": "unhandled request"}) if self._raise_error_for_unhandled_request is True: await self._async_listener_runner.listener_error_handler.handle( error=BoltUnhandledRequestError( request=req, current_response=resp, ), request=req, response=resp, ) return resp return self._handle_unmatched_requests(req, resp) def attachment_action(self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
interactive_messageaction listener. Refer to https://api.slack.com/legacy/message-buttons for details.Expand source code
def attachment_action( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `interactive_message` action listener. Refer to https://api.slack.com/legacy/message-buttons for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.attachment_action(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def block_action(self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
block_actionsaction listener. Refer to https://api.slack.com/reference/interaction-payloads/block-actions for details.Expand source code
def block_action( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `block_actions` action listener. Refer to https://api.slack.com/reference/interaction-payloads/block-actions for details. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.block_action(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def block_suggestion(self, action_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
block_suggestionlistener.Expand source code
def block_suggestion( self, action_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `block_suggestion` listener.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.block_suggestion(action_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def command(self, command: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new slash command listener. This method can be used as either a decorator or a method.
# Use this method as a decorator @app.command("/echo") async def repeat_text(ack, say, command): # Acknowledge command request await ack() await say(f"{command['text']}") # Pass a function to this method app.command("/echo")(repeat_text)Refer to https://api.slack.com/interactivity/slash-commands for details of Slash Commands.
To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
command- The conditions that match a request payload
matchers- A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked.
middleware- A list of lister middleware functions.
Only when all the middleware call
next()method, the listener function can be invoked.
Expand source code
def command( self, command: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new slash command listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.command("/echo") async def repeat_text(ack, say, command): # Acknowledge command request await ack() await say(f"{command['text']}") # Pass a function to this method app.command("/echo")(repeat_text) Refer to https://api.slack.com/interactivity/slash-commands for details of Slash Commands. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: command: The conditions that match a request payload matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.command(command, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def default_app_uninstalled_event_listener(self) ‑> Callable[..., Awaitable[Optional[BoltResponse]]]-
Expand source code
def default_app_uninstalled_event_listener( self, ) -> Callable[..., Awaitable[Optional[BoltResponse]]]: if self._async_tokens_revocation_listeners is None: raise BoltError(error_installation_store_required_for_builtin_listeners()) return self._async_tokens_revocation_listeners.handle_app_uninstalled_events def default_tokens_revoked_event_listener(self) ‑> Callable[..., Awaitable[Optional[BoltResponse]]]-
Expand source code
def default_tokens_revoked_event_listener( self, ) -> Callable[..., Awaitable[Optional[BoltResponse]]]: if self._async_tokens_revocation_listeners is None: raise BoltError(error_installation_store_required_for_builtin_listeners()) return self._async_tokens_revocation_listeners.handle_tokens_revoked_events def dialog_cancellation(self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
dialog_submissionlistener. Refer to https://api.slack.com/dialogs for details.Expand source code
def dialog_cancellation( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `dialog_submission` listener. Refer to https://api.slack.com/dialogs for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.dialog_cancellation(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def dialog_submission(self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
dialog_submissionlistener. Refer to https://api.slack.com/dialogs for details.Expand source code
def dialog_submission( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `dialog_submission` listener. Refer to https://api.slack.com/dialogs for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.dialog_submission(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def dialog_suggestion(self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
dialog_suggestionlistener. Refer to https://api.slack.com/dialogs for details.Expand source code
def dialog_suggestion( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `dialog_suggestion` listener. Refer to https://api.slack.com/dialogs for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.dialog_suggestion(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def enable_token_revocation_listeners(self) ‑> NoneType-
Expand source code
def enable_token_revocation_listeners(self) -> None: self.event("tokens_revoked")(self.default_tokens_revoked_event_listener()) self.event("app_uninstalled")(self.default_app_uninstalled_event_listener()) def error(self, func: Callable[..., Awaitable[Optional[BoltResponse]]]) ‑> Callable[..., Awaitable[Optional[BoltResponse]]]-
Updates the global error handler. This method can be used as either a decorator or a method.
# Use this method as a decorator @app.error async def custom_error_handler(error, body, logger): logger.exception(f"Error: {error}") logger.info(f"Request body: {body}") # Pass a function to this method app.error(custom_error_handler)To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
func- The function that is supposed to be executed when getting an unhandled error in Bolt app.
Expand source code
def error( self, func: Callable[..., Awaitable[Optional[BoltResponse]]] ) -> Callable[..., Awaitable[Optional[BoltResponse]]]: """Updates the global error handler. This method can be used as either a decorator or a method. # Use this method as a decorator @app.error async def custom_error_handler(error, body, logger): logger.exception(f"Error: {error}") logger.info(f"Request body: {body}") # Pass a function to this method app.error(custom_error_handler) To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: func: The function that is supposed to be executed when getting an unhandled error in Bolt app. """ if not inspect.iscoroutinefunction(func): name = get_name_for_callable(func) raise BoltError(error_listener_function_must_be_coro_func(name)) self._async_listener_runner.listener_error_handler = ( AsyncCustomListenerErrorHandler( logger=self._framework_logger, func=func, ) ) return func def event(self, event: Union[str, Pattern, Dict[str, Union[str, Sequence[Union[str, Pattern, NoneType]]]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new event listener. This method can be used as either a decorator or a method.
# Use this method as a decorator @app.event("team_join") async def ask_for_introduction(event, say): welcome_channel_id = "C12345" user_id = event["user"] text = f"Welcome to the team, <@{user_id}>! :tada: You can introduce yourself in this channel." await say(text=text, channel=welcome_channel_id) # Pass a function to this method app.event("team_join")(ask_for_introduction)Refer to https://api.slack.com/apis/connections/events-api for details of Events API.
To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
event- The conditions that match a request payload. If you pass a dict for this, you can have type, subtype in the constraint.
matchers- A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked.
middleware- A list of lister middleware functions.
Only when all the middleware call
next()method, the listener function can be invoked.
Expand source code
def event( self, event: Union[ str, Pattern, Dict[str, Union[str, Sequence[Optional[Union[str, Pattern]]]]] ], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new event listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.event("team_join") async def ask_for_introduction(event, say): welcome_channel_id = "C12345" user_id = event["user"] text = f"Welcome to the team, <@{user_id}>! :tada: You can introduce yourself in this channel." await say(text=text, channel=welcome_channel_id) # Pass a function to this method app.event("team_join")(ask_for_introduction) Refer to https://api.slack.com/apis/connections/events-api for details of Events API. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: event: The conditions that match a request payload. If you pass a dict for this, you can have type, subtype in the constraint. matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.event(event, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware, True ) return __call__ def global_shortcut(self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new global shortcut listener.
Expand source code
def global_shortcut( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new global shortcut listener.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.global_shortcut(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def message(self, keyword: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new message event listener. This method can be used as either a decorator or a method. Check the
App#eventmethod's docstring for details.# Use this method as a decorator @app.message(":wave:") async def say_hello(message, say): user = message['user'] await say(f"Hi there, <@{user}>!") # Pass a function to this method app.message(":wave:")(say_hello)Refer to https://api.slack.com/events/message for details of
messageevents.To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
keyword- The keyword to match
matchers- A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked.
middleware- A list of lister middleware functions.
Only when all the middleware call
next()method, the listener function can be invoked.
Expand source code
def message( self, keyword: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new message event listener. This method can be used as either a decorator or a method. Check the `App#event` method's docstring for details. # Use this method as a decorator @app.message(":wave:") async def say_hello(message, say): user = message['user'] await say(f"Hi there, <@{user}>!") # Pass a function to this method app.message(":wave:")(say_hello) Refer to https://api.slack.com/events/message for details of `message` events. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: keyword: The keyword to match matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ matchers = list(matchers) if matchers else [] middleware = list(middleware) if middleware else [] def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) # As of Jan 2021, most bot messages no longer have the subtype bot_message. # By contrast, messages posted using classic app's bot token still have the subtype. constraints = {"type": "message", "subtype": (None, "bot_message")} primary_matcher = builtin_matchers.event( constraints=constraints, asyncio=True ) middleware.insert(0, AsyncMessageListenerMatches(keyword)) return self._register_listener( list(functions), primary_matcher, matchers, middleware, True ) return __call__ def message_shortcut(self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new message shortcut listener.
Expand source code
def message_shortcut( self, callback_id: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new message shortcut listener.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.message_shortcut(callback_id, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def middleware(self, *args) ‑> Optional[Callable]-
Registers a new middleware to this app. This method can be used as either a decorator or a method.
# Use this method as a decorator @app.middleware async def middleware_func(logger, body, next): logger.info(f"request body: {body}") await next() # Pass a function to this method app.middleware(middleware_func)To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
*args- A function that works as a global middleware.
Expand source code
def middleware(self, *args) -> Optional[Callable]: """Registers a new middleware to this app. This method can be used as either a decorator or a method. # Use this method as a decorator @app.middleware async def middleware_func(logger, body, next): logger.info(f"request body: {body}") await next() # Pass a function to this method app.middleware(middleware_func) To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: *args: A function that works as a global middleware. """ if len(args) > 0: middleware_or_callable = args[0] if isinstance(middleware_or_callable, AsyncMiddleware): self._async_middleware_list.append(middleware_or_callable) elif isinstance(middleware_or_callable, Callable): self._async_middleware_list.append( AsyncCustomMiddleware( app_name=self.name, func=middleware_or_callable ) ) return middleware_or_callable else: raise BoltError( f"Unexpected type for a middleware ({type(middleware_or_callable)})" ) return None def options(self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new options listener. This method can be used as either a decorator or a method.
# Use this method as a decorator @app.options("menu_selection") async 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", }, ] await ack(options=options) # Pass a function to this method app.options("menu_selection")(show_menu_options)Refer to the following documents for details:
- https://api.slack.com/reference/block-kit/block-elements#external_select
- https://api.slack.com/reference/block-kit/block-elements#external_multi_select
To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
matchers- A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked.
middleware- A list of lister middleware functions.
Only when all the middleware call
next()method, the listener function can be invoked.
Expand source code
def options( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new options listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.options("menu_selection") async 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", }, ] await ack(options=options) # Pass a function to this method app.options("menu_selection")(show_menu_options) Refer to the following documents for details: * https://api.slack.com/reference/block-kit/block-elements#external_select * https://api.slack.com/reference/block-kit/block-elements#external_multi_select To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.options(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def server(self, port: int = 3000, path: str = '/slack/events') ‑> AsyncSlackAppServer-
Configure a web server using AIOHTTP. Refer to https://docs.aiohttp.org/ for more details about AIOHTTP.
Args
port- The port to listen on (Default: 3000)
path:The path to handle request from Slack (Default:
/slack/events)Expand source code
def server( self, port: int = 3000, path: str = "/slack/events" ) -> AsyncSlackAppServer: """Configure a web server using AIOHTTP. Refer to https://docs.aiohttp.org/ for more details about AIOHTTP. Args: port: The port to listen on (Default: 3000) path:The path to handle request from Slack (Default: `/slack/events`) """ if ( self._server is None or self._server.port != port or self._server.path != path ): self._server = AsyncSlackAppServer( port=port, path=path, app=self, ) return self._server def shortcut(self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new shortcut listener. This method can be used as either a decorator or a method.
# Use this method as a decorator @app.shortcut("open_modal") async def open_modal(ack, body, client): # Acknowledge the command request await ack() # Call views_open with the built-in client await client.views_open( # Pass a valid trigger_id within 3 seconds of receiving it trigger_id=body["trigger_id"], # View payload view={ ... } ) # Pass a function to this method app.shortcut("open_modal")(open_modal)Refer to https://api.slack.com/interactivity/shortcuts for details about Shortcuts.
To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
constraints- The conditions that match a request payload.
matchers- A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked.
middleware- A list of lister middleware functions.
Only when all the middleware call
next()method, the listener function can be invoked.
Expand source code
def shortcut( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new shortcut listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.shortcut("open_modal") async def open_modal(ack, body, client): # Acknowledge the command request await ack() # Call views_open with the built-in client await client.views_open( # Pass a valid trigger_id within 3 seconds of receiving it trigger_id=body["trigger_id"], # View payload view={ ... } ) # Pass a function to this method app.shortcut("open_modal")(open_modal) Refer to https://api.slack.com/interactivity/shortcuts for details about Shortcuts. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: constraints: The conditions that match a request payload. matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.shortcut(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def start(self, port: int = 3000, path: str = '/slack/events') ‑> NoneType-
Start a web server using AIOHTTP. Refer to https://docs.aiohttp.org/ for more details about AIOHTTP.
Args
port- The port to listen on (Default: 3000)
path:The path to handle request from Slack (Default:
/slack/events)Expand source code
def start(self, port: int = 3000, path: str = "/slack/events") -> None: """Start a web server using AIOHTTP. Refer to https://docs.aiohttp.org/ for more details about AIOHTTP. Args: port: The port to listen on (Default: 3000) path:The path to handle request from Slack (Default: `/slack/events`) """ self.server(port=port, path=path).start() def step(self, callback_id: Union[str, Pattern, AsyncWorkflowStep, AsyncWorkflowStepBuilder], edit: Union[Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable], NoneType] = None, save: Union[Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable], NoneType] = None, execute: Union[Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable], NoneType] = None)-
Registers a new Workflow Step listener. Unlike others, this method doesn't behave as a decorator. If you want to register a workflow step by a decorator, use
AsyncWorkflowStepBuilder's methods.# Create a new WorkflowStep instance from slack_bolt.workflows.async_step import AsyncWorkflowStep ws = AsyncWorkflowStep( callback_id="add_task", edit=edit, save=save, execute=execute, ) # Pass Step to set up listeners app.step(ws)Refer to https://api.slack.com/workflows/steps for details of Steps from Apps.
To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document. For further information about AsyncWorkflowStep specific function arguments such asconfigure,update,complete, andfail, refer to theasyncprefixed ones inslack_bolt.workflows.step.utilitiesAPI documents.Args
callback_id- The Callback ID for this workflow step
edit- The function for displaying a modal in the Workflow Builder
save- The function for handling configuration in the Workflow Builder
execute- The function for handling the step execution
Expand source code
def step( self, callback_id: Union[str, Pattern, AsyncWorkflowStep, AsyncWorkflowStepBuilder], edit: Optional[ Union[ Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable] ] ] = None, save: Optional[ Union[ Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable] ] ] = None, execute: Optional[ Union[ Callable[..., Optional[BoltResponse]], AsyncListener, Sequence[Callable] ] ] = None, ): """ Registers a new Workflow Step listener. Unlike others, this method doesn't behave as a decorator. If you want to register a workflow step by a decorator, use `AsyncWorkflowStepBuilder`'s methods. # Create a new WorkflowStep instance from slack_bolt.workflows.async_step import AsyncWorkflowStep ws = AsyncWorkflowStep( callback_id="add_task", edit=edit, save=save, execute=execute, ) # Pass Step to set up listeners app.step(ws) Refer to https://api.slack.com/workflows/steps for details of Steps from Apps. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. For further information about AsyncWorkflowStep specific function arguments such as `configure`, `update`, `complete`, and `fail`, refer to the `async` prefixed ones in `slack_bolt.workflows.step.utilities` API documents. Args: callback_id: The Callback ID for this workflow step edit: The function for displaying a modal in the Workflow Builder save: The function for handling configuration in the Workflow Builder execute: The function for handling the step execution """ step = callback_id if isinstance(callback_id, (str, Pattern)): step = AsyncWorkflowStep( callback_id=callback_id, edit=edit, save=save, execute=execute, ) elif isinstance(step, AsyncWorkflowStepBuilder): step = step.build() elif not isinstance(step, AsyncWorkflowStep): raise BoltError(f"Invalid step object ({type(step)})") self.use(AsyncWorkflowStepMiddleware(step, self._async_listener_runner)) def use(self, *args) ‑> Optional[Callable]-
Refer to
AsyncApp#middleware()method's docstring for details.Expand source code
def use(self, *args) -> Optional[Callable]: """Refer to `AsyncApp#middleware()` method's docstring for details.""" return self.middleware(*args) def view(self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
view_submission/view_closedevent listener. This method can be used as either a decorator or a method.# Use this method as a decorator @app.view("view_1") async def handle_submission(ack, body, client, view): # Assume there's an input block with <code>block\_c</code> as the block_id and <code>dreamy\_input</code> 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: await ack(response_action="errors", errors=errors) return # Acknowledge the view_submission event and close the modal await ack() # Do whatever you want with the input data - here we're saving it to a DB # Pass a function to this method app.view("view_1")(handle_submission)Refer to https://api.slack.com/reference/interaction-payloads/views for details of payloads.
To learn available arguments for middleware/listeners, see
slack_bolt.kwargs_injection.async_args's API document.Args
constraints- The conditions that match a request payload
matchers- A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked.
middleware- A list of lister middleware functions.
Only when all the middleware call
next()method, the listener function can be invoked.
Expand source code
def view( self, constraints: Union[str, Pattern, Dict[str, Union[str, Pattern]]], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `view_submission`/`view_closed` event listener. This method can be used as either a decorator or a method. # Use this method as a decorator @app.view("view_1") async 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: await ack(response_action="errors", errors=errors) return # Acknowledge the view_submission event and close the modal await ack() # Do whatever you want with the input data - here we're saving it to a DB # Pass a function to this method app.view("view_1")(handle_submission) Refer to https://api.slack.com/reference/interaction-payloads/views for details of payloads. To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.async_args`'s API document. Args: constraints: The conditions that match a request payload matchers: A list of listener matcher functions. Only when all the matchers return True, the listener function can be invoked. middleware: A list of lister middleware functions. Only when all the middleware call `next()` method, the listener function can be invoked. """ def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.view(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def view_closed(self, constraints: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
view_closedlistener. Refer to https://api.slack.com/reference/interaction-payloads/views#view_closed for details.Expand source code
def view_closed( self, constraints: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `view_closed` listener. Refer to https://api.slack.com/reference/interaction-payloads/views#view_closed for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.view_closed(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def view_submission(self, constraints: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None) ‑> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]-
Registers a new
view_submissionlistener. Refer to https://api.slack.com/reference/interaction-payloads/views#view_submission for details.Expand source code
def view_submission( self, constraints: Union[str, Pattern], matchers: Optional[Sequence[Callable[..., Awaitable[bool]]]] = None, middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None, ) -> Optional[Callable[..., Awaitable[Optional[BoltResponse]]]]: """Registers a new `view_submission` listener. Refer to https://api.slack.com/reference/interaction-payloads/views#view_submission for details.""" def __call__(*args, **kwargs): functions = self._to_listener_functions(kwargs) if kwargs else list(args) primary_matcher = builtin_matchers.view_submission(constraints, True) return self._register_listener( list(functions), primary_matcher, matchers, middleware ) return __call__ def web_app(self, path: str = '/slack/events') ‑> aiohttp.web_app.Application-
Returns a
web.Applicationinstance for aiohttp-devtools users.from slack_bolt.async_app import AsyncApp app = AsyncApp() @app.event("app_mention") async def event_test(body, say, logger): logger.info(body) await say("What's up?") def app_factory(): return app.web_app() # adev runserver --port 3000 --app-factory app_factory async_app.pyArgs
path- The path to receive incoming requests from Slack
Expand source code
def web_app(self, path: str = "/slack/events") -> web.Application: """Returns a `web.Application` instance for aiohttp-devtools users. from slack_bolt.async_app import AsyncApp app = AsyncApp() @app.event("app_mention") async def event_test(body, say, logger): logger.info(body) await say("What's up?") def app_factory(): return app.web_app() # adev runserver --port 3000 --app-factory app_factory async_app.py Args: path: The path to receive incoming requests from Slack """ return self.server(path=path).web_app