LoginSignup
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@goran_bahamut(Tadashi Takahashi)

FastAPI AsyncエンドポイントでのSlack連携ガイド

Last updated at Posted at 2025-07-21

FastAPI AsyncエンドポイントでのSlack連携ガイド

このドキュメントでは、自社開発アプリ（Daily Standup）のSlack連携の手順と、主要な連携コード例をまとめます。CursorやChatGPTを頼ってもハルシネーションが多かったので記事にしました。

前提

本システムはマルチテナント対応であり、全てのSlack連携処理は非同期で実装されています。同期処理の実装とは違いますのでご注意ください。同期処理を選ぶべきか、非同期処理を選ぶべきかは、後続のサービス（データベース等）へのアクセスが同期か非同期かで決めれば良いかと思います。また、同期実装の方が学習データが多いせいかCursorやChatGPTはまずは同期処理での実装を例示してくることが多かったので、実装に不安があれば同期処理から始めることをお勧めします。LLM（や人間の部下）に非同期処理で実装を指示する時は、 非同期処理で と明示した方が良いでしょう。

1. Slack API画面での準備

  1. Slack API: Your Apps にアクセスし、「Create New App」から新規アプリを作成します
  2. App Name と Development Slack Workspace を入力し、作成します。
  3. 左メニューの「OAuth & Permissions」で以下のBot Token Scopesを追加します：
    • commands
    • chat:write
    • users:read
    • users:read.email
    • im:write
  4. 「Interactivity & Shortcuts」を有効化し、Request URLにAPIサーバーの /api/slack/events エンドポイントURLを設定します
  5. 「Slash Commands」から /plan など必要なコマンドを追加し、Request URLに /api/slack/events を指定します
  6. 「Basic Information」→「App Credentials」から Client ID, Client Secret, Signing Secret を控えておきます

環境変数の設定

取得した認証情報を .env ファイルに設定します：

# Slack設定
SLACK_CLIENT_ID=your_slack_client_id
SLACK_CLIENT_SECRET=your_slack_client_secret
SLACK_SIGNING_SECRET=your_slack_signing_secret

環境変数読み込みコードの実装例（Pydantic Settings）

環境変数の読み込みには Pydantic Settings を用いています。

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # ...
    # Slack設定
    SLACK_SIGNING_SECRET: str = Field(default="", description="Slack Signing Secret")
    SLACK_CLIENT_ID: str = Field(default="", description="Slack Client ID")
    SLACK_CLIENT_SECRET: str = Field(default="", description="Slack Client Secret")
    # ...
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=True,
        extra="ignore",
    )


settings = Settings()

後続のコードで settings. とあった場合は、このコードを実装して import して用いていると解釈してください。

2. Bot Appのインストール

  1. 「OAuth & Permissions」画面で「Install App to Workspace」をクリックし、ワークスペースにBotをインストールします
  2. インストール後、Bot User OAuth Token（xoxb-...）が発行されます
    1. こちらはSlack APIの管理コンソールからも確認できますが、プログラムで取得できるので控える必要はありません
    2. デバッグ目的で使用するときに用います。Postman等で試験をしてもいいですが、もっとお手軽に参考資料に載せたWeb API methodsの各APIのTesterで試験をすることもできます
  3. 必要に応じてBotを投稿したいチャンネルに招待します

3. OAuth認証

Slack連携ではOAuth 2.0フローを利用します。

OAuth設定の初期化

from slack_bolt.async_app import AsyncApp
from slack_bolt.oauth.async_oauth_settings import AsyncOAuthSettings
from slack_sdk.signature import SignatureVerifier

# OAuth 設定とインストール情報ストアを用意
oauth_settings = AsyncOAuthSettings(
    client_id=settings.SLACK_CLIENT_ID,
    client_secret=settings.SLACK_CLIENT_SECRET,
    scopes=[
        "commands",
        "chat:write",
        "users:read",
        "users:read.email",
        "im:write",
    ],
    installation_store_bot_only=True,
    installation_store=InstallationStore(),
)

bolt_app = AsyncApp(
    signing_secret=settings.SLACK_SIGNING_SECRET, oauth_settings=oauth_settings
)

InstallationStore()の実装例はお見せできないので公式のドキュメントURLをご紹介します。
Module slack_sdk.oauth.installation_store.sqlalchemy

scopes はWeb API methodsで使用するAPIの仕様をを確認して最低限のものを選択します。

認証用エンドポイント

from fastapi import APIRouter
from fastapi.responses import RedirectResponse


@router.get("/join")
async def handle_join(code: str, state: str | None = None) -> RedirectResponse:
    slack_usecase = SlackChatUsecase()
    result = await slack_usecase.handle_join_event(code, state)
    return RedirectResponse(url=settings.FRONTEND_URL)

OAuthトークン取得処理

トークン交換を実装：

from slack_sdk import WebClient
from pydantic import ValidationError

def get_oauth_access_token(self, code: str) -> SlackOAuthResponse:
    """Slack OAuth2.0でアクセストークンを取得する"""
    slack_client = WebClient()
    try:
        resp = slack_client.oauth_v2_access(
            client_id=settings.SLACK_CLIENT_ID,
            client_secret=settings.SLACK_CLIENT_SECRET,
            code=code,
        )
        slack_oauth_response = SlackOAuthResponse.model_validate(resp.data)
        return slack_oauth_response

署名検証の設定

from slack_sdk.signature import SignatureVerifier
from daily_standup.core.config import settings

verifier = SignatureVerifier(signing_secret=settings.SLACK_SIGNING_SECRET)

設定手順

  1. Slack Appの「OAuth & Permissions」→「Redirect URLs」に、
    https://your-domain/api/slack/join などを登録しておきます。
  2. 認証後、Botトークンやユーザートークン、ワークスペースIDなどをDBに保存します。

4. Slash コマンドの受付と返答

Slashコマンドの実装例

src/daily_standup/controller/slack_chat_controller.py でSlashコマンドを実装：

from collections.abc import Awaitable
from typing import Any, Callable
from slack_sdk.web.async_client import AsyncWebClient

@bolt_app.command("/plan")
async def open_plan(
    ack: Callable[..., Awaitable[None]], body: dict[str, Any], client: AsyncWebClient
) -> None:
    await ack()
    await client.views_open(
        trigger_id=body["trigger_id"],
        view={
            "type": "modal",
            "callback_id": "planning_blocker_view",
            "title": {"type": "plain_text", "text": "今日やること"},
            "submit": {"type": "plain_text", "text": "次へ"},
            "blocks": [
                {
                    "type": "input",
                    "block_id": "plan",
                    "label": {
                        "type": "plain_text",
                        "text": "今日予定しているタスクを入力してください。",
                    },
                    "element": {
                        "type": "plain_text_input",
                        "action_id": "plan_input",
                        "multiline": True,
                    },
                }
            ],
        },
    )

モーダル送信後のハンドラー

import json
from collections.abc import Awaitable
from typing import Any, Callable
from slack_sdk.web.async_client import AsyncWebClient

@bolt_app.view("planning_blocker_view")
async def handle_step1(
    ack: Callable[..., Awaitable[None]], body: dict[str, Any], client: AsyncWebClient
) -> None:
    plan = body["view"]["state"]["values"]["plan"]["plan_input"]["value"]
    metadata = {"plan": plan}

    # ackで新しいviewを返す（response_action="update"）
    await ack(
        response_action="update",
        view={
            "type": "modal",
            "callback_id": "planning_complete",
            "title": {"type": "plain_text", "text": "困っていること・ボトルネック"},
            "submit": {"type": "plain_text", "text": "完了"},
            "private_metadata": json.dumps(metadata),
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": plan,
                    },
                },
                {
                    "type": "input",
                    "block_id": "blocker",
                    "label": {
                        "type": "plain_text",
                        "text": "困っていることやブロックしている問題があれば入力してください",
                    },
                    "element": {
                        "type": "plain_text_input",
                        "action_id": "blocker_input",
                        "multiline": True,
                    },
                    "optional": True,
                },
            ],
        },
    )

イベント受信エンドポイント

from fastapi import APIRouter, Request, Response

@router.post("/events", response_model=None)
async def slack_events(request: Request) -> Response:
    return await handler.handle(request)

メッセージ送信処理

src/daily_standup/usecase/slack_chat_usecase.py でメッセージ送信を実装：

from slack_sdk.web.async_client import AsyncWebClient
from daily_standup.utils.logger import logger

async def review_plan(self, slack_id: str, client: AsyncWebClient, message: str) -> None:
    """SlackユーザーIDからユーザーを取得し、DMで振り返りメッセージを送信する"""
    user = await self.user_domain.get_by_slack_id(slack_id)

    im = await client.conversations_open(users=slack_id)
    channel_id: str = im["channel"]["id"]
    await client.chat_postMessage(
        channel=channel_id,
        text=message,
        blocks=[
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": message,
                },
            }
        ],
    )

参考資料

備考

Bot TokenとUser Tokenの違い

項目 Bot Token User Token
形式 xoxb-... xoxp-...
用途 Botユーザーとしての操作 特定ユーザーとしての操作
権限 Bot Token Scopesで制限 User Token Scopesで制限
操作対象 ワークスペース全体 特定ユーザーの権限範囲内
主な用途 チャンネルへの投稿、Slashコマンド処理 ユーザー情報取得、DM送信
取得方法 OAuth & Permissions画面 OAuth認証フロー
セキュリティ ワークスペース管理者が管理 ユーザー個人の認証

OAuthで必要なBot Token Scopes

Scope 説明 用途
commands Slashコマンドの作成と管理 /plan、/reviewなどのコマンド実行
chat:write メッセージの送信 チャンネルやDMへのメッセージ投稿
users:read ユーザー情報の読み取り ワークスペース内のユーザー情報取得
users:read.email ユーザーのメールアドレス読み取り ユーザー識別とメール連携
im:write ダイレクトメッセージの送信 ユーザーへのDM送信

注意事項

  • Botの権限やイベントサブスクリプション、リクエストURLの設定ミスに注意してください
  • Bot Tokenはワークスペース全体で共有されるため、機密性の高い操作にはUser Tokenを使用してください
  • 例示したソースコードが実装されている本バックエンドはFastAPI、PostgreSQL、TortoiseORMを用いて構築されています

以上

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?