slack_bolt.middleware.async_builtins API documentation

class AsyncAttachingConversationKwargs(AsyncMiddleware):

    thread_context_store: Optional[AsyncAssistantThreadContextStore]

    def __init__ (self, thread_context_store: Optional[AsyncAssistantThreadContextStore] = None):
        self.thread_context_store = thread_context_store

    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> Optional[BoltResponse]:
        event = to_event(req.body)
        if event is not None:
            if is_assistant_event(req.body):
                assistant = AsyncAssistantUtilities(
                    payload=event,
                    context=req.context,
                    thread_context_store=self.thread_context_store,
                )
                req.context["say"] = assistant.say
                req.context["set_title"] = assistant.set_title
                req.context["set_suggested_prompts"] = assistant.set_suggested_prompts
                req.context["get_thread_context"] = assistant.get_thread_context
                req.context["save_thread_context"] = assistant.save_thread_context

            # TODO: in the future we might want to introduce a "proper" extract_ts utility
            thread_ts = req.context.thread_ts or event.get("ts")
            if req.context.channel_id and thread_ts:
                req.context["set_status"] = AsyncSetStatus(
                    client=req.context.client,
                    channel_id=req.context.channel_id,
                    thread_ts=thread_ts,
                )
                req.context["say_stream"] = AsyncSayStream(
                    client=req.context.client,
                    channel=req.context.channel_id,
                    recipient_team_id=req.context.team_id or req.context.enterprise_id,
                    recipient_user_id=req.context.user_id,
                    thread_ts=thread_ts,
                )
        return await next()

class AsyncAttachingFunctionToken(AsyncMiddleware):
    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        # This method is not supposed to be invoked by bolt-python users
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        if req.context.function_bot_access_token is not None:
            req.context.client.token = req.context.function_bot_access_token

        return await next()

class AsyncIgnoringSelfEvents(IgnoringSelfEvents, AsyncMiddleware):
    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        auth_result = req.context.authorize_result
        # message events can have $.event.bot_id while it does not have its user_id
        bot_id = req.body.get("event", {}).get("bot_id")
        if self._is_self_event(auth_result, req.context.user_id, bot_id, req.body): # type: ignore[arg-type]
            if self.ignoring_self_assistant_message_events_enabled is False:
                if is_bot_message_event_in_assistant_thread(req.body):
                    # Assistant#bot_message handler acknowledges this pattern
                    return await next()

            self._debug_log(req.body)
            return await req.context.ack()
        else:
            return await next()

class AsyncMessageListenerMatches(AsyncMiddleware):
    def __init__ (self, keyword: Union[str, Pattern]):
        """Captures matched keywords and saves the values in context."""
        self.keyword = keyword

    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        # As this method is not supposed to be invoked by bolt-python users,
        # the naming conflict with the built-in one affects
        # only the internals of this method
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        text = req.body.get("event", {}).get("text", "")
        if text:
            m: Optional[Union[Sequence]] = re.findall(self.keyword, text)
            if m is not None and m != []:
                if type(m[0]) is not tuple:
                    m = tuple(m)
                else:
                    m = m[0]
                req.context["matches"] = m # tuple or list
                return await next()

        # As the text doesn't match, skip running the listener
        return resp

class AsyncRequestVerification(RequestVerification, AsyncMiddleware):
    """Verifies an incoming request by checking the validity of
    `x-slack-signature`, `x-slack-request-timestamp`, and its body data.

    Refer to https://docs.slack.dev/authentication/verifying-requests-from-slack/ for details.
    """

    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        # As this method is not supposed to be invoked by bolt-python users,
        # the naming conflict with the built-in one affects
        # only the internals of this method
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        if self._can_skip(req.mode, req.body):
            return await next()

        body = req.raw_body
        timestamp = req.headers.get("x-slack-request-timestamp", ["0"])[0]
        signature = req.headers.get("x-slack-signature", [""])[0]
        if self.verifier.is_valid(body, timestamp, signature):
            return await next()
        else:
            self._debug_log_error(signature, timestamp, body)
            return self._build_error_response()

class AsyncSslCheck(SslCheck, AsyncMiddleware):
    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        # As this method is not supposed to be invoked by bolt-python users,
        # the naming conflict with the built-in one affects
        # only the internals of this method
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        if self._is_ssl_check_request(req.body):
            if self._verify_token_if_needed(req.body):
                return self._build_error_response()
            return self._build_success_response()
        else:
            return await next()

class AsyncUrlVerification(UrlVerification, AsyncMiddleware):
    def __init__ (self, base_logger: Optional[Logger] = None):
        self.logger = get_bolt_logger(AsyncUrlVerification, base_logger=base_logger)

    async def async_process(
        self,
        *,
        req: AsyncBoltRequest,
        resp: BoltResponse,
        next: Callable[[], Awaitable[BoltResponse]],
    ) -> BoltResponse:
        if self._is_url_verification_request(req.body):
            return self._build_success_response(req.body)
        else:
            return await next()