bolt-python/docs/reference/async_app.html at main · https-saml-salesforce-com/bolt-python

History

5600 lines (5116 loc) · 286 KB

Raw

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>

<head>

<title>slack_bolt.async_app API documentation</title>

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>

<h1 class="title">Module <code>slack_bolt.async_app</code></h1>

</header>

<p>Module for creating asyncio based apps</p>

<h3 id="creating-an-async-app">Creating an async app</h3>

<p>If you'd prefer to build your app with <a href="https://docs.python.org/3/library/asyncio.html">asyncio</a>, you can import the <a href="https://docs.aiohttp.org/en/stable/">AIOHTTP</a> library and call the <code><a title="slack_bolt.async_app.AsyncApp" href="#slack_bolt.async_app.AsyncApp">AsyncApp</a></code> constructor. Within async apps, you can use the async/await pattern.</p>

<pre><code class="language-bash"># Python 3.7+ required

python -m venv .venv

source .venv/bin/activate

pip install -U pip

# aiohttp is required

pip install slack_bolt aiohttp

</code></pre>

<p>In async apps, all middleware/listeners must be async functions. When calling utility methods (like <code>ack</code> and <code>say</code>) within these functions, it's required to use the <code>await</code> keyword.</p>

<pre><code class="language-python"># Import the async app instead of the regular one

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?")

@app.command("/hello-bolt-python")

async def command(ack, body, respond):

await ack()

await respond(f"Hi <@{body['user_id']}>!")

if __name__ == "__main__":

app.start(3000)

</code></pre>

<p>If you want to use another async Web framework (e.g., Sanic, FastAPI, Starlette), take a look at the built-in adapters and their examples.</p>

<ul>

<li><a href="https://github.com/slackapi/bolt-python/tree/main/examples">The Bolt app examples</a></li>

<li><a href="https://github.com/slackapi/bolt-python/tree/main/slack_bolt/adapter">The built-in adapters</a>

Apps can be run the same way as the synchronous example above. If you'd prefer another async Web framework (e.g., Sanic, FastAPI, Starlette), take a look at <a href="https://github.com/slackapi/bolt-python/tree/main/slack_bolt/adapter">the built-in adapters</a> and their corresponding <a href="https://github.com/slackapi/bolt-python/tree/main/examples">examples</a>.</li>

</ul>

<p>Refer to <code><a title="slack_bolt.app.async_app" href="app/async_app.html">slack_bolt.app.async_app</a></code> for more details.</p>

</section>

</section>

</section>

</section>

<h2 class="section-title" id="header-classes">Classes</h2>

<dl>

<span>class <span class="ident">AsyncAck</span></span>

</code></dt>

<dd>

<span>Expand source code</span>

</summary>

<pre><code class="python">class AsyncAck:

response: Optional[BoltResponse]

def __init__(self):

self.response: Optional[BoltResponse] = None

async def __call__(

self,

text: Union[str, dict] = "", # text: str or whole_response: dict

blocks: Optional[Sequence[Union[dict, Block]]] = None,

attachments: Optional[Sequence[Union[dict, Attachment]]] = None,

unfurl_links: Optional[bool] = None,

unfurl_media: Optional[bool] = None,

response_type: Optional[str] = None, # in_channel / ephemeral

# block_suggestion / dialog_suggestion

options: Optional[Sequence[Union[dict, Option]]] = None,

option_groups: Optional[Sequence[Union[dict, OptionGroup]]] = None,

# view_submission

response_action: Optional[str] = None, # errors / update / push / clear

errors: Optional[Dict[str, str]] = None,

view: Optional[Union[dict, View]] = None,

) -> BoltResponse:

return _set_response(

self,

text_or_whole_response=text,

blocks=blocks,

attachments=attachments,

unfurl_links=unfurl_links,

unfurl_media=unfurl_media,

response_type=response_type,

options=options,

option_groups=option_groups,

response_action=response_action,

errors=errors,

view=view,

)</code></pre>

</details>

<h3>Class variables</h3>

<dl>

<dt id="slack_bolt.async_app.AsyncAck.response"><code class="name">var <span class="ident">response</span> : <a title="slack_bolt.response.response.BoltResponse" href="response/response.html#slack_bolt.response.response.BoltResponse">BoltResponse</a> | None</code></dt>

<dd>

<div class="desc"><p>The type of the None singleton.</p></div>

</dd>

</dl>

</dd>

<span>class <span class="ident">AsyncApp</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>client: slack_sdk.web.async_client.AsyncWebClient | None = None,<br>before_authorize: <a title="slack_bolt.middleware.async_middleware.AsyncMiddleware" href="middleware/async_middleware.html#slack_bolt.middleware.async_middleware.AsyncMiddleware">AsyncMiddleware</a> | Callable[..., Awaitable[Any]] | None = None,<br>authorize: Callable[..., Awaitable[<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.async_installation_store.AsyncInstallationStore | 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.async_oauth_settings.AsyncOAuthSettings" href="oauth/async_oauth_settings.html#slack_bolt.oauth.async_oauth_settings.AsyncOAuthSettings">AsyncOAuthSettings</a> | None = None,<br>oauth_flow: <a title="slack_bolt.oauth.async_oauth_flow.AsyncOAuthFlow" href="oauth/async_oauth_flow.html#slack_bolt.oauth.async_oauth_flow.AsyncOAuthFlow">AsyncOAuthFlow</a> | None = None,<br>verification_token: str | None = None,<br>assistant_thread_context_store: <a title="slack_bolt.context.assistant.thread_context_store.async_store.AsyncAssistantThreadContextStore" href="context/assistant/thread_context_store/async_store.html#slack_bolt.context.assistant.thread_context_store.async_store.AsyncAssistantThreadContextStore">AsyncAssistantThreadContextStore</a> | None = None)</span>

</code></dt>

<dd>

<span>Expand source code</span>

</summary>

<pre><code class="python">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

before_authorize: Optional[Union[AsyncMiddleware, Callable[..., Awaitable[Any]]]] = None,

authorize: Optional[Callable[..., Awaitable[AuthorizeResult]]] = None,

user_facing_authorize_error_message: Optional[str] = None,

installation_store: Optional[AsyncInstallationStore] = 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[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,

# for AI Agents & Assistants

assistant_thread_context_store: Optional[AsyncAssistantThreadContextStore] = 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://docs.slack.dev/tools/bolt-python/concepts/async for details.

If you would like to build an OAuth app for enabling the app to run with multiple workspaces,

refer to https://docs.slack.dev/tools/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.

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 `AsyncInstallationStore#async_find_bot()` if True (Default: False)

request_verification_enabled: False if you would like to disable the built-in middleware (Default: True).

`AsyncRequestVerification` 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).

`AsyncIgnoringSelfEvents` 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).

`AsyncUrlVerification` is a built-in middleware that handles url_verification requests

that verify the endpoint for Events API in HTTP Mode requests.

ssl_check_enabled: bool = False if you would like to disable the built-in middleware (Default: True).

`AsyncSslCheck` is a built-in middleware that handles ssl_check requests from Slack.

attaching_function_token_enabled: False if you would like to disable the built-in middleware (Default: True).

`AsyncAttachingFunctionToken` is a built-in middleware that injects the just-in-time workflow-execution token

when your app receives `function_executed` or interactivity events scoped to a custom step.

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 be used only for ssl_check requests.

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(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_before_authorize: Optional[AsyncMiddleware] = None

if before_authorize is not None:

if callable(before_authorize):

self._async_before_authorize = AsyncCustomMiddleware(

app_name=self._name,

func=before_authorize,

base_logger=self._framework_logger,

)

elif isinstance(before_authorize, AsyncMiddleware):

self._async_before_authorize = before_authorize

self._async_authorize: Optional[AsyncAuthorize] = None

if authorize is not None:

if isinstance(authorize, AsyncAuthorize):

# As long as an advanced developer understands what they're doing,

# bolt-python should not prevent customizing authorize middleware

self._async_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._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:

settings = oauth_flow.settings if oauth_flow is not None else oauth_settings

self._async_authorize = AsyncInstallationStoreAuthorize(

installation_store=self._async_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._async_client, # for proxy use cases etc.

user_token_resolution=(settings.user_token_resolution if settings is not None else "authed_user"),

)

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

if installation_store is not None:

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

if installation_store is not None:

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

self._async_authorize.token_rotation_expiration_minutes = oauth_settings.token_rotation_expiration_minutes # type: ignore[attr-defined] # noqa: E501

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 # type: ignore[union-attr]

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[AsyncMiddleware] = []

self._async_listeners: List[AsyncListener] = []

self._assistant_thread_context_store = assistant_thread_context_store

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_start_handler=AsyncDefaultListenerStartHandler(logger=self._framework_logger),

listener_completion_handler=AsyncDefaultListenerCompletionHandler(logger=self._framework_logger),

lazy_listener_runner=AsyncioLazyListenerRunner(

logger=self._framework_logger,

)

self._async_middleware_error_handler: AsyncMiddlewareErrorHandler = AsyncDefaultMiddlewareErrorHandler(

logger=self._framework_logger,

)

self._init_middleware_list_done = False

self._init_async_middleware_list(

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,

)

self._server: Optional[AsyncSlackAppServer] = None

def _init_async_middleware_list(

self,

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._async_middleware_list.append(

AsyncSslCheck(

verification_token=self._verification_token,

base_logger=self._base_logger,

)

if request_verification_enabled is True:

self._async_middleware_list.append(AsyncRequestVerification(self._signing_secret, base_logger=self._base_logger))

if self._async_before_authorize is not None:

self._async_middleware_list.append(self._async_before_authorize)

# As authorize is required for making a Bolt app function, we don't offer the flag to disable this

if self._async_oauth_flow is None:

if self._token:

self._async_middleware_list.append(

AsyncSingleTeamAuthorization(

base_logger=self._base_logger,

user_facing_authorize_error_message=user_facing_authorize_error_message,

)

elif self._async_authorize is not None:

self._async_middleware_list.append(

AsyncMultiTeamsAuthorization(

authorize=self._async_authorize,

base_logger=self._base_logger,

user_facing_authorize_error_message=user_facing_authorize_error_message,

)

else:

raise BoltError(error_token_required())

elif self._async_authorize is not None:

self._async_middleware_list.append(

AsyncMultiTeamsAuthorization(

authorize=self._async_authorize,

base_logger=self._base_logger,

user_token_resolution=self._async_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._async_middleware_list.append(

AsyncIgnoringSelfEvents(

base_logger=self._base_logger,

ignoring_self_assistant_message_events_enabled=ignoring_self_assistant_message_events_enabled,

)

if url_verification_enabled is True:

self._async_middleware_list.append(AsyncUrlVerification(base_logger=self._base_logger))

if attaching_function_token_enabled is True:

self._async_middleware_list.append(AsyncAttachingFunctionToken())

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",

host: Optional[str] = None,

) -> 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`)

host: The hostname to serve the web endpoints. (Default: 0.0.0.0)

"""

if self._server is None or self._server.port != port or self._server.path != path:

self._server = AsyncSlackAppServer(

port=port,

path=path,

app=self,

host=host,

)

return self._server

def web_app(self, path: str = "/slack/events", port: int = 3000) -> 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

port: The port to listen on (Default: 3000)

"""

return self.server(path=path, port=port).web_app

def start(self, port: int = 3000, path: str = "/slack/events", host: Optional[str] = None) -> 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`)

host: The hostname to serve the web endpoints. (Default: 0.0.0.0)

"""

self.server(port=port, path=path, host=host).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: Optional[BoltResponse] = BoltResponse(status=200, body="")

middleware_state = {"next_called": False}

async def async_middleware_next():

middleware_state["next_called"] = True

try:

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 # 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:

await self._async_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

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): # type: ignore[arg-type]

# run all the middleware attached to this listener first

(middleware_resp, next_was_not_called) = await listener.run_async_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] = await self._async_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:

await self._async_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="")

await self._async_middleware_error_handler.handle(

error=error,

request=req,

response=resp,

)

return resp

def _handle_unmatched_requests(self, req: AsyncBoltRequest, resp: BoltResponse) -> BoltResponse:

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):

middleware: AsyncMiddleware = middleware_or_callable

self._async_middleware_list.append(middleware)

if isinstance(middleware, AsyncAssistant) and middleware.thread_context_store is not None:

self._assistant_thread_context_store = middleware.thread_context_store

elif callable(middleware_or_callable):

self._async_middleware_list.append(

AsyncCustomMiddleware(

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

def assistant(self, assistant: AsyncAssistant) -> Optional[Callable]:

return self.middleware(assistant)

# -------------------------

# 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,

"""

Deprecated:

Steps from apps for legacy workflows are now deprecated.

Use new custom steps: https://docs.slack.dev/workflows/workflow-steps/

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 `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://docs.slack.dev/legacy/legacy-steps-from-apps/ 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 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://docs.slack.dev/workflows/workflow-steps/"

category=DeprecationWarning,

)

step = callback_id

if isinstance(callback_id, (str, Pattern)):

step = AsyncWorkflowStep(

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, AsyncWorkflowStepBuilder):

step = step.build(base_logger=self._base_logger)

elif not isinstance(step, AsyncWorkflowStep):

raise BoltError(f"Invalid step object ({type(step)})")

self.use(AsyncWorkflowStepMiddleware(step))

# -------------------------

# 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 is_callable_coroutine(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,

)

self._async_middleware_error_handler = AsyncCustomMiddlewareErrorHandler(

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[..., Awaitable[bool]]]] = None,

middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,

) -> Callable[..., 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://docs.slack.dev/apis/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, 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[..., Awaitable[bool]]]] = None,

middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,

) -> Callable[..., 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://docs.slack.dev/reference/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)

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(

constraints=constraints,

keyword=keyword,

asyncio=True,

base_logger=self._base_logger,

)

middleware.insert(0, AsyncMessageListenerMatches(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[..., Awaitable[bool]]]] = None,

middleware: Optional[Sequence[Union[Callable, AsyncMiddleware]]] = None,

auto_acknowledge: bool = True,

ack_timeout: int = 3,

) -> Callable[..., Optional[Callable[..., Awaitable[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")

async def reverse_string(ack: AsyncAck, inputs: dict, complete: AsyncComplete, fail: AsyncFail):

try:

await ack()

string_to_reverse = inputs["stringToReverse"]

await complete({"reverseString": string_to_reverse[::-1]})

except Exception as e:

await fail(f"Cannot reverse string (error: {e})")

raise e

# Pass a function to this method

View remainder of file in raw view

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

FilesExpand file tree

async_app.html

Latest commit

History

async_app.html

File metadata and controls