-
Notifications
You must be signed in to change notification settings - Fork 287
Expand file tree
/
Copy pathapp.html
More file actions
3273 lines (2939 loc) · 172 KB
/
app.html
File metadata and controls
3273 lines (2939 loc) · 172 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, minimum-scale=1">
<meta name="generator" content="pdoc3 0.11.5">
<title>slack_bolt.app.app API documentation</title>
<meta name="description" content="">
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/13.0.0/sanitize.min.css" integrity="sha512-y1dtMcuvtTMJc1yPgEqF0ZjQbhnc/bFhyvIyVNb9Zk5mIGtqVaAB1Ttl28su8AvFMOY0EwRbAe+HCLqj6W7/KA==" crossorigin>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/13.0.0/typography.min.css" integrity="sha512-Y1DYSb995BAfxobCkKepB1BqJJTPrOp3zPL74AWFugHHmmdcvO+C48WLrUOlhGMc0QG7AE3f7gmvvcrmX2fDoA==" crossorigin>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.9.0/styles/default.min.css" crossorigin>
<style>:root{--highlight-color:#fe9}.flex{display:flex !important}body{line-height:1.5em}#content{padding:20px}#sidebar{padding:1.5em;overflow:hidden}#sidebar > *:last-child{margin-bottom:2cm}.http-server-breadcrumbs{font-size:130%;margin:0 0 15px 0}#footer{font-size:.75em;padding:5px 30px;border-top:1px solid #ddd;text-align:right}#footer p{margin:0 0 0 1em;display:inline-block}#footer p:last-child{margin-right:30px}h1,h2,h3,h4,h5{font-weight:300}h1{font-size:2.5em;line-height:1.1em}h2{font-size:1.75em;margin:2em 0 .50em 0}h3{font-size:1.4em;margin:1.6em 0 .7em 0}h4{margin:0;font-size:105%}h1:target,h2:target,h3:target,h4:target,h5:target,h6:target{background:var(--highlight-color);padding:.2em 0}a{color:#058;text-decoration:none;transition:color .2s ease-in-out}a:visited{color:#503}a:hover{color:#b62}.title code{font-weight:bold}h2[id^="header-"]{margin-top:2em}.ident{color:#900;font-weight:bold}pre code{font-size:.8em;line-height:1.4em;padding:1em;display:block}code{background:#f3f3f3;font-family:"DejaVu Sans Mono",monospace;padding:1px 4px;overflow-wrap:break-word}h1 code{background:transparent}pre{border-top:1px solid #ccc;border-bottom:1px solid #ccc;margin:1em 0}#http-server-module-list{display:flex;flex-flow:column}#http-server-module-list div{display:flex}#http-server-module-list dt{min-width:10%}#http-server-module-list p{margin-top:0}.toc ul,#index{list-style-type:none;margin:0;padding:0}#index code{background:transparent}#index h3{border-bottom:1px solid #ddd}#index ul{padding:0}#index h4{margin-top:.6em;font-weight:bold}@media (min-width:200ex){#index .two-column{column-count:2}}@media (min-width:300ex){#index .two-column{column-count:3}}dl{margin-bottom:2em}dl dl:last-child{margin-bottom:4em}dd{margin:0 0 1em 3em}#header-classes + dl > dd{margin-bottom:3em}dd dd{margin-left:2em}dd p{margin:10px 0}.name{background:#eee;font-size:.85em;padding:5px 10px;display:inline-block;min-width:40%}.name:hover{background:#e0e0e0}dt:target .name{background:var(--highlight-color)}.name > span:first-child{white-space:nowrap}.name.class > span:nth-child(2){margin-left:.4em}.inherited{color:#999;border-left:5px solid #eee;padding-left:1em}.inheritance em{font-style:normal;font-weight:bold}.desc h2{font-weight:400;font-size:1.25em}.desc h3{font-size:1em}.desc dt code{background:inherit}.source > summary,.git-link-div{color:#666;text-align:right;font-weight:400;font-size:.8em;text-transform:uppercase}.source summary > *{white-space:nowrap;cursor:pointer}.git-link{color:inherit;margin-left:1em}.source pre{max-height:500px;overflow:auto;margin:0}.source pre code{font-size:12px;overflow:visible;min-width:max-content}.hlist{list-style:none}.hlist li{display:inline}.hlist li:after{content:',\2002'}.hlist li:last-child:after{content:none}.hlist .hlist{display:inline;padding-left:1em}img{max-width:100%}td{padding:0 .5em}.admonition{padding:.1em 1em;margin:1em 0}.admonition-title{font-weight:bold}.admonition.note,.admonition.info,.admonition.important{background:#aef}.admonition.todo,.admonition.versionadded,.admonition.tip,.admonition.hint{background:#dfd}.admonition.warning,.admonition.versionchanged,.admonition.deprecated{background:#fd4}.admonition.error,.admonition.danger,.admonition.caution{background:lightpink}</style>
<style media="screen and (min-width: 700px)">@media screen and (min-width:700px){#sidebar{width:30%;height:100vh;overflow:auto;position:sticky;top:0}#content{width:70%;max-width:100ch;padding:3em 4em;border-left:1px solid #ddd}pre code{font-size:1em}.name{font-size:1em}main{display:flex;flex-direction:row-reverse;justify-content:flex-end}.toc ul ul,#index ul ul{padding-left:1em}.toc > ul > li{margin-top:.5em}}</style>
<style media="print">@media print{#sidebar h1{page-break-before:always}.source{display:none}}@media print{*{background:transparent !important;color:#000 !important;box-shadow:none !important;text-shadow:none !important}a[href]:after{content:" (" attr(href) ")";font-size:90%}a[href][title]:after{content:none}abbr[title]:after{content:" (" attr(title) ")"}.ir a:after,a[href^="javascript:"]:after,a[href^="#"]:after{content:""}pre,blockquote{border:1px solid #999;page-break-inside:avoid}thead{display:table-header-group}tr,img{page-break-inside:avoid}img{max-width:100% !important}@page{margin:0.5cm}p,h2,h3{orphans:3;widows:3}h1,h2,h3,h4,h5,h6{page-break-after:avoid}}</style>
<script defer src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.9.0/highlight.min.js" integrity="sha512-D9gUyxqja7hBtkWpPWGt9wfbfaMGVt9gnyCvYa+jojwwPHLCzUm5i8rpk7vD7wNee9bA35eYIjobYPaQuKS1MQ==" crossorigin></script>
<script>window.addEventListener('DOMContentLoaded', () => {
hljs.configure({languages: ['bash', 'css', 'diff', 'graphql', 'ini', 'javascript', 'json', 'plaintext', 'python', 'python-repl', 'rust', 'shell', 'sql', 'typescript', 'xml', 'yaml']});
hljs.highlightAll();
/* Collapse source docstrings */
setTimeout(() => {
[...document.querySelectorAll('.hljs.language-python > .hljs-string')]
.filter(el => el.innerHTML.length > 200 && ['"""', "'''"].includes(el.innerHTML.substring(0, 3)))
.forEach(el => {
let d = document.createElement('details');
d.classList.add('hljs-string');
d.innerHTML = '<summary>"""</summary>' + el.innerHTML.substring(3);
el.replaceWith(d);
});
}, 100);
})</script>
</head>
<body>
<main>
<article id="content">
<header>
<h1 class="title">Module <code>slack_bolt.app.app</code></h1>
</header>
<section id="section-intro">
</section>
<section>
</section>
<section>
</section>
<section>
</section>
<section>
<h2 class="section-title" id="header-classes">Classes</h2>
<dl>
<dt id="slack_bolt.app.app.App"><code class="flex name class">
<span>class <span class="ident">App</span></span>
<span>(</span><span>*,<br>logger: logging.Logger | None = None,<br>name: str | None = None,<br>process_before_response: bool = False,<br>raise_error_for_unhandled_request: bool = False,<br>signing_secret: str | None = None,<br>token: str | None = None,<br>token_verification_enabled: bool = True,<br>client: slack_sdk.web.client.WebClient | None = None,<br>before_authorize: <a title="slack_bolt.middleware.middleware.Middleware" href="../middleware/middleware.html#slack_bolt.middleware.middleware.Middleware">Middleware</a> | Callable[..., Any] | None = None,<br>authorize: Callable[..., <a title="slack_bolt.authorization.authorize_result.AuthorizeResult" href="../authorization/authorize_result.html#slack_bolt.authorization.authorize_result.AuthorizeResult">AuthorizeResult</a>] | None = None,<br>user_facing_authorize_error_message: str | None = None,<br>installation_store: slack_sdk.oauth.installation_store.installation_store.InstallationStore | None = None,<br>installation_store_bot_only: bool | None = None,<br>request_verification_enabled: bool = True,<br>ignoring_self_events_enabled: bool = True,<br>ignoring_self_assistant_message_events_enabled: bool = True,<br>ssl_check_enabled: bool = True,<br>url_verification_enabled: bool = True,<br>attaching_function_token_enabled: bool = True,<br>oauth_settings: <a title="slack_bolt.oauth.oauth_settings.OAuthSettings" href="../oauth/oauth_settings.html#slack_bolt.oauth.oauth_settings.OAuthSettings">OAuthSettings</a> | None = None,<br>oauth_flow: <a title="slack_bolt.oauth.oauth_flow.OAuthFlow" href="../oauth/oauth_flow.html#slack_bolt.oauth.oauth_flow.OAuthFlow">OAuthFlow</a> | None = None,<br>verification_token: str | None = None,<br>listener_executor: concurrent.futures._base.Executor | None = None,<br>assistant_thread_context_store: <a title="slack_bolt.context.assistant.thread_context_store.store.AssistantThreadContextStore" href="../context/assistant/thread_context_store/store.html#slack_bolt.context.assistant.thread_context_store.store.AssistantThreadContextStore">AssistantThreadContextStore</a> | None = None)</span>
</code></dt>
<dd>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">class App:
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,
token_verification_enabled: bool = True,
client: Optional[WebClient] = None,
# for multi-workspace apps
before_authorize: Optional[Union[Middleware, Callable[..., Any]]] = None,
authorize: Optional[Callable[..., AuthorizeResult]] = None,
user_facing_authorize_error_message: Optional[str] = None,
installation_store: Optional[InstallationStore] = None,
# for either only bot scope usage or v1.0.x compatibility
installation_store_bot_only: Optional[bool] = None,
# for customizing the built-in middleware
request_verification_enabled: bool = True,
ignoring_self_events_enabled: bool = True,
ignoring_self_assistant_message_events_enabled: bool = True,
ssl_check_enabled: bool = True,
url_verification_enabled: bool = True,
attaching_function_token_enabled: bool = True,
# for the OAuth flow
oauth_settings: Optional[OAuthSettings] = None,
oauth_flow: Optional[OAuthFlow] = None,
# No need to set (the value is used only in response to ssl_check requests)
verification_token: Optional[str] = None,
# Set this one only when you want to customize the executor
listener_executor: Optional[Executor] = None,
# for AI Agents & Assistants
assistant_thread_context_store: Optional[AssistantThreadContextStore] = None,
):
"""Bolt App that provides functionalities to register middleware/listeners.
import os
from slack_bolt import App
# Initializes your app with your bot token and signing secret
app = App(
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")
def message_hello(message, say):
# say() sends a message to the channel where the event was triggered
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/tutorial/getting-started for details.
If you 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.
token_verification_enabled: Verifies the validity of the given token if True.
client: The singleton `slack_sdk.WebClient` instance for this app.
before_authorize: A global middleware that can be executed right before authorize function
authorize: The function to authorize an incoming request from Slack
by checking if there is a team/user in the installation data.
user_facing_authorize_error_message: The user-facing error message to display
when the app is installed but the installation is not managed by this app's installation store
installation_store: The module offering save/find operations of installation data
installation_store_bot_only: Use `InstallationStore#find_bot()` if True (Default: False)
request_verification_enabled: False if you would like to disable the built-in middleware (Default: True).
`RequestVerification` is a built-in middleware that verifies the signature in HTTP Mode requests.
Make sure if it's safe enough when you turn a built-in middleware off.
We strongly recommend using RequestVerification for better security.
If you have a proxy that verifies request signature in front of the Bolt app,
it's totally fine to disable RequestVerification to avoid duplication of work.
Don't turn it off just for easiness of development.
ignoring_self_events_enabled: False if you would like to disable the built-in middleware (Default: True).
`IgnoringSelfEvents` is a built-in middleware that enables Bolt apps to easily skip the events
generated by this app's bot user (this is useful for avoiding code error causing an infinite loop).
ignoring_self_assistant_message_events_enabled: False if you would like to disable the built-in middleware.
`IgnoringSelfEvents` for this app's bot user message events within an assistant thread
This is useful for avoiding code error causing an infinite loop; Default: True
url_verification_enabled: False if you would like to disable the built-in middleware (Default: True).
`UrlVerification` is a built-in middleware that handles url_verification requests
that verify the endpoint for Events API in HTTP Mode requests.
attaching_function_token_enabled: False if you would like to disable the built-in middleware (Default: True).
`AttachingFunctionToken` is a built-in middleware that injects the just-in-time workflow-execution tokens
when your app receives `function_executed` or interactivity events scoped to a custom step.
ssl_check_enabled: bool = False if you would like to disable the built-in middleware (Default: True).
`SslCheck` is a built-in middleware that handles ssl_check requests from Slack.
oauth_settings: The settings related to Slack app installation flow (OAuth flow)
oauth_flow: Instantiated `slack_bolt.oauth.OAuthFlow`. This is always prioritized over oauth_settings.
verification_token: Deprecated verification mechanism. This can be used only for ssl_check requests.
listener_executor: Custom executor to run background tasks. If absent, the default `ThreadPoolExecutor` will
be used.
assistant_thread_context_store: Custom AssistantThreadContext store (Default: the built-in implementation,
which uses a parent message's metadata to store the latest context)
"""
if signing_secret is None:
signing_secret = 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)
# If a logger is explicitly passed when initializing, the logger works as the base logger.
# The base logger's logging settings will be propagated to all the loggers created by bolt-python.
self._base_logger = logger
# The framework logger is supposed to be used for the internal logging.
# Also, it's accessible via `app.logger` as the app's singleton logger.
self._framework_logger = logger or get_bolt_logger(App)
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, WebClient):
raise BoltError(error_client_invalid_type())
self._client = client
self._token = client.token
if token is not None:
self._framework_logger.warning(warning_client_prioritized_and_token_skipped())
else:
self._client = create_web_client(
# NOTE: the token here can be None
token=token,
logger=self._framework_logger,
)
# --------------------------------------
# Authorize & OAuthFlow initialization
# --------------------------------------
self._before_authorize: Optional[Middleware] = None
if before_authorize is not None:
if callable(before_authorize):
self._before_authorize = CustomMiddleware(
app_name=self._name,
func=before_authorize,
base_logger=self._framework_logger,
)
elif isinstance(before_authorize, Middleware):
self._before_authorize = before_authorize
self._authorize: Optional[Authorize] = None
if authorize is not None:
if isinstance(authorize, Authorize):
# As long as an advanced developer understands what they're doing,
# bolt-python should not prevent customizing authorize middleware
self._authorize = authorize
else:
if oauth_settings is not None or oauth_flow is not None:
# If the given authorize is a simple function,
# it does not work along with installation_store.
raise BoltError(error_authorize_conflicts())
self._authorize = CallableAuthorize(logger=self._framework_logger, func=authorize)
self._installation_store: Optional[InstallationStore] = installation_store
if self._installation_store is not None and self._authorize is None:
settings = oauth_flow.settings if oauth_flow is not None else oauth_settings
self._authorize = InstallationStoreAuthorize(
installation_store=self._installation_store,
client_id=settings.client_id if settings is not None else None,
client_secret=settings.client_secret if settings is not None else None,
logger=self._framework_logger,
bot_only=installation_store_bot_only or False,
client=self._client, # for proxy use cases etc.
user_token_resolution=(settings.user_token_resolution if settings is not None else "authed_user"),
)
self._oauth_flow: Optional[OAuthFlow] = 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 = OAuthSettings()
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 is not None:
self._oauth_flow = oauth_flow
installation_store = select_consistent_installation_store(
client_id=self._oauth_flow.client_id,
app_store=self._installation_store,
oauth_flow_store=self._oauth_flow.settings.installation_store,
logger=self._framework_logger,
)
self._installation_store = installation_store
if installation_store is not None:
self._oauth_flow.settings.installation_store = installation_store
if self._oauth_flow._client is None:
self._oauth_flow._client = self._client
if self._authorize is None:
self._authorize = self._oauth_flow.settings.authorize
elif oauth_settings is not None:
installation_store = select_consistent_installation_store(
client_id=oauth_settings.client_id,
app_store=self._installation_store,
oauth_flow_store=oauth_settings.installation_store,
logger=self._framework_logger,
)
self._installation_store = installation_store
if installation_store is not None:
oauth_settings.installation_store = installation_store
self._oauth_flow = OAuthFlow(client=self.client, logger=self.logger, settings=oauth_settings)
if self._authorize is None:
self._authorize = self._oauth_flow.settings.authorize
self._authorize.token_rotation_expiration_minutes = oauth_settings.token_rotation_expiration_minutes # type: ignore[attr-defined] # noqa: E501
if (self._installation_store is not None or self._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._oauth_flow is not None:
app_bot_only = installation_store_bot_only or False
oauth_flow_bot_only = self._oauth_flow.settings.installation_store_bot_only
if app_bot_only != oauth_flow_bot_only:
self.logger.warning(warning_bot_only_conflicts())
self._oauth_flow.settings.installation_store_bot_only = app_bot_only
self._authorize.bot_only = app_bot_only # type: ignore[union-attr]
self._tokens_revocation_listeners: Optional[TokenRevocationListeners] = None
if self._installation_store is not None:
self._tokens_revocation_listeners = TokenRevocationListeners(self._installation_store)
# --------------------------------------
# Middleware Initialization
# --------------------------------------
self._middleware_list: List[Middleware] = []
self._listeners: List[Listener] = []
if listener_executor is None:
listener_executor = ThreadPoolExecutor(max_workers=5)
self._assistant_thread_context_store = assistant_thread_context_store
self._process_before_response = process_before_response
self._listener_runner = ThreadListenerRunner(
logger=self._framework_logger,
process_before_response=process_before_response,
listener_error_handler=DefaultListenerErrorHandler(logger=self._framework_logger),
listener_start_handler=DefaultListenerStartHandler(logger=self._framework_logger),
listener_completion_handler=DefaultListenerCompletionHandler(logger=self._framework_logger),
listener_executor=listener_executor,
lazy_listener_runner=ThreadLazyListenerRunner(
logger=self._framework_logger,
executor=listener_executor,
),
)
self._middleware_error_handler: MiddlewareErrorHandler = DefaultMiddlewareErrorHandler(
logger=self._framework_logger,
)
self._init_middleware_list_done = False
self._init_middleware_list(
token_verification_enabled=token_verification_enabled,
request_verification_enabled=request_verification_enabled,
ignoring_self_events_enabled=ignoring_self_events_enabled,
ignoring_self_assistant_message_events_enabled=ignoring_self_assistant_message_events_enabled,
ssl_check_enabled=ssl_check_enabled,
url_verification_enabled=url_verification_enabled,
attaching_function_token_enabled=attaching_function_token_enabled,
user_facing_authorize_error_message=user_facing_authorize_error_message,
)
def _init_middleware_list(
self,
token_verification_enabled: bool = True,
request_verification_enabled: bool = True,
ignoring_self_events_enabled: bool = True,
ignoring_self_assistant_message_events_enabled: bool = True,
ssl_check_enabled: bool = True,
url_verification_enabled: bool = True,
attaching_function_token_enabled: bool = True,
user_facing_authorize_error_message: Optional[str] = None,
):
if self._init_middleware_list_done:
return
if ssl_check_enabled is True:
self._middleware_list.append(
SslCheck(
verification_token=self._verification_token,
base_logger=self._base_logger,
)
)
if request_verification_enabled is True:
self._middleware_list.append(RequestVerification(self._signing_secret, base_logger=self._base_logger))
if self._before_authorize is not None:
self._middleware_list.append(self._before_authorize)
# As authorize is required for making a Bolt app function, we don't offer the flag to disable this
if self._oauth_flow is None:
if self._token is not None:
try:
auth_test_result = None
if token_verification_enabled:
# This API call is for eagerly validating the token
auth_test_result = self._client.auth_test(token=self._token)
self._middleware_list.append(
SingleTeamAuthorization(
auth_test_result=auth_test_result,
base_logger=self._base_logger,
user_facing_authorize_error_message=user_facing_authorize_error_message,
)
)
except SlackApiError as err:
raise BoltError(error_auth_test_failure(err.response))
elif self._authorize is not None:
self._middleware_list.append(
MultiTeamsAuthorization(
authorize=self._authorize,
base_logger=self._base_logger,
user_facing_authorize_error_message=user_facing_authorize_error_message,
)
)
else:
raise BoltError(error_token_required())
elif self._authorize is not None:
self._middleware_list.append(
MultiTeamsAuthorization(
authorize=self._authorize,
base_logger=self._base_logger,
user_token_resolution=self._oauth_flow.settings.user_token_resolution,
user_facing_authorize_error_message=user_facing_authorize_error_message,
)
)
else:
raise BoltError(error_oauth_flow_or_authorize_required())
if ignoring_self_events_enabled is True:
self._middleware_list.append(
IgnoringSelfEvents(
base_logger=self._base_logger,
ignoring_self_assistant_message_events_enabled=ignoring_self_assistant_message_events_enabled,
)
)
if url_verification_enabled is True:
self._middleware_list.append(UrlVerification(base_logger=self._base_logger))
if attaching_function_token_enabled is True:
self._middleware_list.append(AttachingFunctionToken())
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[OAuthFlow]:
"""Configured `OAuthFlow` object if exists."""
return self._oauth_flow
@property
def logger(self) -> logging.Logger:
"""The logger this app uses."""
return self._framework_logger
@property
def client(self) -> WebClient:
"""The singleton `slack_sdk.WebClient` instance in this app."""
return self._client
@property
def installation_store(self) -> Optional[InstallationStore]:
"""The `slack_sdk.oauth.InstallationStore` that can be used in the `authorize` middleware."""
return self._installation_store
@property
def listener_runner(self) -> ThreadListenerRunner:
"""The thread executor for asynchronously running listeners."""
return self._listener_runner
@property
def process_before_response(self) -> bool:
return self._process_before_response or False
# -------------------------
# standalone server
def start(
self,
port: int = 3000,
path: str = "/slack/events",
http_server_logger_enabled: bool = True,
) -> None:
"""Starts a web server for local development.
# With the default settings, `http://localhost:3000/slack/events`
# is available for handling incoming requests from Slack
app.start()
This method internally starts a Web server process built with the `http.server` module.
For production, consider using a production-ready WSGI server such as Gunicorn.
Args:
port: The port to listen on (Default: 3000)
path: The path to handle request from Slack (Default: `/slack/events`)
http_server_logger_enabled: The flag to enable http.server logging if True (Default: True)
"""
self._development_server = SlackAppDevelopmentServer(
port=port,
path=path,
app=self,
oauth_flow=self.oauth_flow,
http_server_logger_enabled=http_server_logger_enabled,
)
self._development_server.start()
# -------------------------
# main dispatcher
def dispatch(self, req: BoltRequest) -> 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: Optional[BoltResponse] = BoltResponse(status=200, body="")
middleware_state = {"next_called": False}
def middleware_next():
middleware_state["next_called"] = True
try:
for middleware in self._middleware_list:
middleware_state["next_called"] = False
if self._framework_logger.level <= logging.DEBUG:
self._framework_logger.debug(debug_applying_middleware(middleware.name))
resp = middleware.process(req=req, resp=resp, next=middleware_next) # type: ignore[arg-type]
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:
try:
raise BoltUnhandledRequestError(
request=req,
current_response=resp,
last_global_middleware_name=middleware.name,
)
except BoltUnhandledRequestError as e:
self._listener_runner.listener_error_handler.handle(
error=e,
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._listeners:
listener_name = get_name_for_callable(listener.ack_function)
self._framework_logger.debug(debug_checking_listener(listener_name))
if listener.matches(req=req, resp=resp): # type: ignore[arg-type]
# run all the middleware attached to this listener first
middleware_resp, next_was_not_called = listener.run_middleware(
req=req, resp=resp # type: ignore[arg-type]
)
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] = self._listener_runner.run(
request=req,
response=resp, # type: ignore[arg-type]
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:
try:
raise BoltUnhandledRequestError(
request=req,
current_response=resp,
)
except BoltUnhandledRequestError as e:
self._listener_runner.listener_error_handler.handle(
error=e,
request=req,
response=resp,
)
return resp
return self._handle_unmatched_requests(req, resp)
except Exception as error:
resp = BoltResponse(status=500, body="")
self._middleware_error_handler.handle(
error=error,
request=req,
response=resp,
)
return resp
def _handle_unmatched_requests(self, req: BoltRequest, resp: BoltResponse) -> BoltResponse:
self._framework_logger.warning(warning_unhandled_request(req))
return resp
# -------------------------
# middleware
def use(self, *args) -> Optional[Callable]:
"""Registers a new global middleware to this app. This method can be used as either a decorator or a method.
Refer to `App#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
def middleware_func(logger, body, next):
logger.info(f"request body: {body}")
next()
# Pass a function to this method
app.middleware(middleware_func)
Refer to https://slack.dev/bolt-python/concepts#global-middleware for details.
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.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, Middleware):
middleware: Middleware = middleware_or_callable
self._middleware_list.append(middleware)
if isinstance(middleware, Assistant) and middleware.thread_context_store is not None:
self._assistant_thread_context_store = middleware.thread_context_store
elif callable(middleware_or_callable):
self._middleware_list.append(
CustomMiddleware(
app_name=self.name,
func=middleware_or_callable,
base_logger=self._base_logger,
)
)
return middleware_or_callable
else:
raise BoltError(f"Unexpected type for a middleware ({type(middleware_or_callable)})")
return None
# -------------------------
# AI Agents & Assistants
def assistant(self, assistant: Assistant) -> Optional[Callable]:
return self.middleware(assistant)
# -------------------------
# Workflows: Steps from apps
def step(
self,
callback_id: Union[str, Pattern, WorkflowStep, WorkflowStepBuilder],
edit: Optional[Union[Callable[..., Optional[BoltResponse]], Listener, Sequence[Callable]]] = None,
save: Optional[Union[Callable[..., Optional[BoltResponse]], Listener, Sequence[Callable]]] = None,
execute: Optional[Union[Callable[..., Optional[BoltResponse]], Listener, Sequence[Callable]]] = None,
):
"""
Deprecated:
Steps from apps for legacy workflows are now deprecated.
Use new custom steps: https://api.slack.com/automation/functions/custom-bolt
Registers a new step from app listener.
Unlike others, this method doesn't behave as a decorator.
If you want to register a step from app by a decorator, use `WorkflowStepBuilder`'s methods.
# Create a new WorkflowStep instance
from slack_bolt.workflows.step import WorkflowStep
ws = WorkflowStep(
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.args`'s API document.
For further information about WorkflowStep specific function arguments
such as `configure`, `update`, `complete`, and `fail`,
refer to `slack_bolt.workflows.step.utilities` API documents.
Args:
callback_id: The Callback ID for this step from app
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
"""
warnings.warn(
(
"Steps from apps for legacy workflows are now deprecated. "
"Use new custom steps: https://api.slack.com/automation/functions/custom-bolt"
),
category=DeprecationWarning,
)
step = callback_id
if isinstance(callback_id, (str, Pattern)):
step = WorkflowStep(
callback_id=callback_id,
edit=edit, # type: ignore[arg-type]
save=save, # type: ignore[arg-type]
execute=execute, # type: ignore[arg-type]
base_logger=self._base_logger,
)
elif isinstance(step, WorkflowStepBuilder):
step = step.build(base_logger=self._base_logger)
elif not isinstance(step, WorkflowStep):
raise BoltError(f"Invalid step object ({type(step)})")
self.use(WorkflowStepMiddleware(step))
# -------------------------
# global error handler
def error(self, func: Callable[..., Optional[BoltResponse]]) -> Callable[..., 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
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.args`'s API document.
Args:
func: The function that is supposed to be executed
when getting an unhandled error in Bolt app.
"""
self._listener_runner.listener_error_handler = CustomListenerErrorHandler(
logger=self._framework_logger,
func=func,
)
self._middleware_error_handler = CustomMiddlewareErrorHandler(
logger=self._framework_logger,
func=func,
)
return func
# -------------------------
# events
def event(
self,
event: Union[
str,
Pattern,
Dict[str, Optional[Union[str, Sequence[Optional[Union[str, Pattern]]]]]],
],
matchers: Optional[Sequence[Callable[..., bool]]] = None,
middleware: Optional[Sequence[Union[Callable, Middleware]]] = None,
) -> Callable[..., Optional[Callable[..., 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")
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."
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.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, base_logger=self._base_logger)
return self._register_listener(list(functions), primary_matcher, matchers, middleware, True)
return __call__
def message(
self,
keyword: Union[str, Pattern] = "",
matchers: Optional[Sequence[Callable[..., bool]]] = None,
middleware: Optional[Sequence[Union[Callable, Middleware]]] = None,
) -> Callable[..., Optional[Callable[..., 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:")
def say_hello(message, say):
user = message['user']
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.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)
constraints = {
"type": "message",
"subtype": (
# In most cases, new message events come with no subtype.
None,
# 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.
"bot_message",
# If an end-user posts a message with "Also send to #channel" checked,
# the message event comes with this subtype.
"thread_broadcast",
# If an end-user posts a message with attached files,
# the message event comes with this subtype.
"file_share",
),
}
primary_matcher = builtin_matchers.message_event(
keyword=keyword, constraints=constraints, base_logger=self._base_logger
)
middleware.insert(0, MessageListenerMatches(keyword))
return self._register_listener(list(functions), primary_matcher, matchers, middleware, True)
return __call__
def function(
self,
callback_id: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., bool]]] = None,
middleware: Optional[Sequence[Union[Callable, Middleware]]] = None,
auto_acknowledge: bool = True,
) -> Callable[..., Optional[Callable[..., Optional[BoltResponse]]]]:
"""Registers a new Function listener.
This method can be used as either a decorator or a method.
# Use this method as a decorator
@app.function("reverse")
def reverse_string(ack: Ack, inputs: dict, complete: Complete, fail: Fail):
try:
ack()
string_to_reverse = inputs["stringToReverse"]
complete(outputs={"reverseString": string_to_reverse[::-1]})
except Exception as e:
fail(f"Cannot reverse string (error: {e})")
raise e
# Pass a function to this method
app.function("reverse")(reverse_string)
To learn available arguments for middleware/listeners, see `slack_bolt.kwargs_injection.args`'s API document.
Args:
callback_id: The callback id to identify the function
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)
primary_matcher = builtin_matchers.function_executed(callback_id=callback_id, base_logger=self._base_logger)
return self._register_listener(functions, primary_matcher, matchers, middleware, auto_acknowledge)
return __call__
# -------------------------
# slash commands
def command(
self,
command: Union[str, Pattern],
matchers: Optional[Sequence[Callable[..., bool]]] = None,
middleware: Optional[Sequence[Union[Callable, Middleware]]] = None,
) -> Callable[..., Optional[Callable[..., 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")
def repeat_text(ack, say, command):
# Acknowledge command request
ack()
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.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, base_logger=self._base_logger)
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[..., bool]]] = None,
middleware: Optional[Sequence[Union[Callable, Middleware]]] = None,
) -> Callable[..., Optional[Callable[..., 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")
def open_modal(ack, body, client):
# Acknowledge the command request
ack()
# Call views_open with the built-in client
client.views_open(
# Pass a valid trigger_id within 3 seconds of receiving it
trigger_id=body["trigger_id"],
# View payload
view={ ... }
)
# 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.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, base_logger=self._base_logger)
return self._register_listener(list(functions), primary_matcher, matchers, middleware)
return __call__
def global_shortcut(