Slackとの連携をPythonで実現するための公式ライブラリslack-sdk
について、基本から応用まで徹底的に解説します。
はじめに:Slack SDKとは? 🤔
Slackは、ビジネスコミュニケーションツールとして広く利用されていますが、その真価はAPIを通じたカスタマイズや連携にあります。PythonでSlack APIを利用する際、最も推奨されるのが公式のslack-sdk
ライブラリです。
以前はslackclient
というライブラリがありましたが、現在はメンテナンスモードとなっており、slack-sdk
がその後継としてアクティブに開発されています。slack-sdk
を利用することで、requestsのようなHTTPクライアントを直接使うよりも、以下のようなメリットがあります。
- 新しいAPIや機能への迅速な対応
- 非推奨パラメータなどの変更に対する警告
- 認証、リトライ処理、エラーハンドリングなどの定型処理の簡略化
- Block KitなどのリッチなUIコンポーネント作成支援
このslack-sdk
は、Slackが提供する様々なAPIに対応する複数のモジュール(slack_sdk.web
, slack_sdk.webhook
, slack_sdk.socket_mode
など)で構成されており、それぞれ単独でも、組み合わせて使うことも可能です。
Bolt for Pythonとの関係 ⚡
slack-sdk
とよく一緒に語られるライブラリにslack_bolt
(Bolt for Python)があります。slack-sdk
がSlack APIを直接叩くための低レベルな機能を提供するのに対し、slack_bolt
はイベントリスニングやインタラクティブな応答など、Slackアプリ開発をより簡単にするための高レベルなフレームワークです。slack_bolt
は内部でslack-sdk
を利用しています。単純なメッセージ送信や情報取得であればslack-sdk
のみで十分ですが、イベント駆動型のアプリを作る場合はslack_bolt
の利用が推奨されます。
インストール 🛠️
slack-sdk
のインストールはpipを使って簡単に行えます。Python 3.6以上が必要です。
pip install slack_sdk
もし古いslackclient
やslack
パッケージがインストールされている環境でAttributeError: module 'slack' has no attribute 'WebClient'
のようなエラーが出た場合は、古いパッケージをアンインストールしてからslack_sdk
を再インストールしてみてください。
pip uninstall slack slackclient
pip install slack_sdk
認証:トークンの準備 🔑
Slack APIを利用するには、認証のためのトークンが必要です。主に以下の種類があります。
- Bot User OAuth Token (
xoxb-
): ボットユーザーとしてAPIを呼び出すためのトークン。ほとんどのアプリ開発で利用します。 - User OAuth Token (
xoxp-
): アプリをインストールしたユーザーとしてAPIを呼び出すためのトークン。 - App-Level Token (
xapp-
): Socket Mode接続など、特定の機能で利用されるトークン。
これらのトークンは、Slack App管理画面でアプリを作成し、必要な権限(Scope)を設定した後に発行されます。例えば、メッセージを投稿するにはchat:write
スコープが必要です。
🚨 トークンの安全な管理
トークンはパスワードと同様に機密情報です。ソースコードに直接書き込んだり、バージョン管理システムにコミットしたりしないでください。環境変数や安全な設定管理サービスを利用して管理することが強く推奨されます。
import os
# 環境変数からトークンを読み込む
slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")
環境変数の設定例 (bash):
export SLACK_BOT_TOKEN="xoxb-your-token-here"
python your_script.py
アプリを単一のワークスペースにインストールする場合は、管理画面から直接インストールしてトークンを取得できます。複数のワークスペースに対応させる場合は、OAuthフローを実装する必要があります。slack-sdk
にはOAuthフローをサポートするモジュール (slack_sdk.oauth
) も含まれています。
基本的な使い方:WebClientによるメッセージ送信 💬
slack-sdk
の中心的な機能の一つが、WebClient
を通じたWeb APIの呼び出しです。これにより、チャンネルへのメッセージ投稿、ユーザー情報の取得、ファイルのアップロードなど、数百種類以上のAPIメソッドを利用できます。
最も基本的な例として、チャンネルにメッセージを投稿する方法を見てみましょう。
import os
import logging
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
# ロギングの設定 (デバッグ用)
logging.basicConfig(level=logging.DEBUG)
# 環境変数からボットトークンを取得
slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")
if not slack_bot_token:
print("エラー: 環境変数 SLACK_BOT_TOKEN が設定されていません。")
exit()
# WebClientの初期化
client = WebClient(token=slack_bot_token)
# メッセージを投稿したいチャンネルID
channel_id = "#general" # または "C0XXXXXXXXX" のような形式
try:
# chat.postMessage APIを呼び出す
response = client.chat_postMessage(
channel=channel_id,
text="こんにちは! slack-sdk からのテストメッセージです 👋"
)
# 成功した場合、レスポンスには投稿されたメッセージの情報が含まれる
# logger.info(f"メッセージ送信成功: {response['ts']}") # tsはタイムスタンプ
print(f"メッセージがチャンネル {channel_id} に送信されました。")
except SlackApiError as e:
# API呼び出しでエラーが発生した場合
# エラーレスポンスは e.response で確認できる
print(f"エラーが発生しました: {e.response['error']}") # 例: 'channel_not_found', 'invalid_auth'
このコードでは、まず環境変数からボットトークンを取得し、それを使ってWebClient
のインスタンスを作成しています。そして、client.chat_postMessage
メソッドを呼び出して、指定したチャンネルID (#general
やC0123456789
など) にテキストメッセージを送信しています。
try...except SlackApiError
ブロックでAPI呼び出し時のエラーを捕捉しています。SlackApiError
が発生した場合、e.response['error']
でエラーコード(例: 'channel_not_found'
, 'invalid_auth'
, 'not_in_channel'
など)を確認できます。not_in_channel
エラーが出た場合は、ボットが対象チャンネルに参加しているか確認してください。
エフェメラルメッセージ(Ephemeral Message)の送信
エフェメラルメッセージは、特定のユーザーにのみ表示され、他のチャンネルメンバーには見えない一時的なメッセージです。chat.postEphemeral
メソッドを使用します。
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")
client = WebClient(token=slack_bot_token)
channel_id = "#general"
user_id = "U0XXXXXXXXX" # メッセージを表示させたいユーザーのID
try:
response = client.chat_postEphemeral(
channel=channel_id,
user=user_id,
text="これはあなただけに表示されるメッセージです🤫"
)
print(f"エフェメラルメッセージがユーザー {user_id} に送信されました。")
except SlackApiError as e:
print(f"エラーが発生しました: {e.response['error']}")
chat.postMessage
との主な違いは、メッセージを表示させたいユーザーのIDをuser
引数で指定する点です。
Block Kitによるリッチなメッセージ表現 🧱✨
Slackのメッセージは、単なるテキストだけでなく、ボタン、画像、セレクトメニュー、日付ピッカーなど、様々なUIコンポーネントを組み合わせたリッチな表現が可能です。これを実現するのがBlock Kitです。
Block Kitは、JSON形式でUI構造を定義します。slack-sdk
では、chat.postMessage
などのメソッドのblocks
引数に、このJSON構造に対応するPythonのリスト(辞書のリスト)を渡すことで利用できます。
Block Kit Builderという公式ツールを使うと、GUIでインタラクティブにUIを組み立て、対応するJSONを生成できるため非常に便利です。
Block Kitを使ったメッセージ送信例
以下は、セクション、画像、ボタンを含むメッセージを送信する例です。
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")
client = WebClient(token=slack_bot_token)
channel_id = "#general"
try:
response = client.chat_postMessage(
channel=channel_id,
text="Block Kitを使ったメッセージのフォールバックテキスト", # 通知や古いクライアント用
blocks=[
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "これは *Block Kit* を使ったメッセージです! 🎉"
}
},
{
"type": "divider"
},
{
"type": "image",
"title": {
"type": "plain_text",
"text": "かわいい犬の画像 🐶",
"emoji": True
},
"image_url": "https://hips.hearstapps.com/hmg-prod/images/dog-puppy-on-garden-royalty-free-image-1586966191.jpg?crop=1xw:0.74975xh;center,top&resize=1200:*", # 画像URLは適切なものに置き換えてください
"alt_text": "庭にいるかわいい子犬"
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {
"type": "plain_text",
"text": "クリックしてね",
"emoji": True
},
"value": "click_me_123",
"action_id": "button_click_action" # インタラクションを処理する際に使用
}
]
}
]
)
print(f"Block Kitメッセージがチャンネル {channel_id} に送信されました。")
except SlackApiError as e:
print(f"エラーが発生しました: {e.response['error']}")
ポイントは以下の通りです。
blocks
引数にリストを渡します。リストの各要素がひとつのブロック(辞書)です。- 各ブロック辞書には
type
キーがあり、ブロックの種類(section
,image
,actions
,divider
など)を指定します。 section
ブロックではtext
オブジェクトで表示テキストを指定します。type
をmrkdwn
にするとMarkdown形式の装飾が使えます。image
ブロックではimage_url
とalt_text
を指定します。actions
ブロックにはボタンなどのインタラクティブ要素(elements
)を含めることができます。- ボタンなどのインタラクティブ要素には
action_id
を指定し、ユーザーが操作した際にどの要素が操作されたかを識別できるようにします。(インタラクションの処理は通常slack_bolt
で行います) text
引数は、Block Kitが表示できない環境(プッシュ通知など)でのフォールバックテキストとして機能するため、設定することが推奨されます。
slack-sdk
には、これらのBlock Kit要素をPythonオブジェクトとして簡単に構築するためのモデル (slack_sdk.models
) も用意されています。これにより、JSONライクな辞書を直接書く代わりに、よりPythonicな方法でブロックを組み立てることも可能です。
ファイルアップロード 📂
slack-sdk
を使ってチャンネルにファイルをアップロードすることもできます。以前はfiles.upload
メソッドが使われていましたが、このメソッドは2025年3月11日に廃止される予定です。
新しい推奨方法はfiles.upload_v2
メソッドを使用することです。このメソッドは内部的に複数ステップのAPI呼び出し(files.getUploadURLExternal
と files.completeUploadExternal
)を行いますが、SDKがそれらを抽象化してくれます。
⚠️ files.uploadの廃止と移行
既存のコードでfiles.upload
を使用している場合は、早めにfiles.upload_v2
への移行を計画してください。2024年5月8日以降に作成された新しいSlackアプリは、すでにfiles.upload
を利用できません。
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import logging
logging.basicConfig(level=logging.INFO) # INFOレベルに変更
logger = logging.getLogger(__name__)
slack_bot_token = os.environ.get("SLACK_BOT_TOKEN")
client = WebClient(token=slack_bot_token)
channel_id = "#general"
filepath = "./my_report.txt" # アップロードするファイルのパス
initial_comment = "こちらがレポートファイルです📄"
# アップロードするファイルを作成(サンプル用)
try:
with open(filepath, "w") as f:
f.write("これはテストレポートファイルの内容です。\n")
f.write("Generated by slack-sdk example.\n")
except Exception as e:
logger.error(f"テストファイルの作成に失敗しました: {e}")
exit()
try:
# files.upload_v2 メソッドを使用してファイルをアップロード
# このメソッドは必要な権限 (files:read, files:write) を要求します
response = client.files_upload_v2(
channel=channel_id,
filepath=filepath,
initial_comment=initial_comment,
title="レポートファイル" # 省略可能: ファイルのタイトル
)
logger.info(f"ファイルアップロード成功: {response.get('file', {}).get('name')}")
# アップロード成功時のレスポンス構造は複雑なので、必要に応じて確認してください
except SlackApiError as e:
logger.error(f"ファイルアップロードエラー: {e.response['error']}")
# logger.error(f"エラー詳細: {e.response}") # 詳細なエラー情報
except FileNotFoundError:
logger.error(f"ファイルが見つかりません: {filepath}")
except Exception as e:
# その他の予期せぬエラー
logger.error(f"予期せぬエラーが発生しました: {e}")
# サンプル用に作成したファイルを削除
# if os.path.exists(filepath):
# os.remove(filepath)
files.upload_v2
を使用するには、アプリにfiles:read
とfiles:write
のスコープが必要です。filepath
引数でローカルファイルのパスを指定するか、content
引数でファイルの内容をバイト列または文字列として直接渡すこともできます。
その他の主要なWeb APIメソッド例 📚
WebClient
を通じて利用できるAPIは非常に多岐にわたります。ここではよく使われる例をいくつか紹介します。
チャンネル情報の取得 (conversations.list)
参加可能なパブリックチャンネルや参加しているプライベートチャンネルの一覧を取得します。
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))
try:
# typesで取得するチャンネルの種類を指定 (public_channel, private_channel, mpim, im)
# limitで一度に取得する件数を指定 (デフォルト100, 最大1000)
result = client.conversations_list(
types="public_channel,private_channel",
limit=200
)
channels = result.get("channels", [])
print(f"{len(channels)} 件のチャンネルが見つかりました:")
for channel in channels:
channel_id = channel.get("id")
channel_name = channel.get("name")
is_private = channel.get("is_private", False)
print(f"- ID: {channel_id}, 名前: {channel_name}, プライベート: {is_private}")
# TODO: ページネーションの処理 (結果が1000件を超える場合)
# result['response_metadata']['next_cursor'] を使って次のページを取得
except SlackApiError as e:
print(f"チャンネルリスト取得エラー: {e.response['error']}")
必要なスコープ: channels:read
(パブリック), groups:read
(プライベート)
チャンネル履歴の取得 (conversations.history)
特定のチャンネルのメッセージ履歴を取得します。
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))
channel_id = "C0XXXXXXXXX" # 履歴を取得したいチャンネルID
try:
# limitで取得件数を指定 (デフォルト100)
result = client.conversations_history(channel=channel_id, limit=10)
messages = result.get("messages", [])
print(f"チャンネル {channel_id} の最新メッセージ {len(messages)} 件:")
for msg in messages:
user = msg.get("user", "N/A")
text = msg.get("text", "")
ts = msg.get("ts", "") # タイムスタンプ
print(f" [{ts}] User {user}: {text[:50]}...") # 長すぎる場合は省略
# TODO: ページネーション (古いメッセージを取得する場合)
except SlackApiError as e:
print(f"チャンネル履歴取得エラー: {e.response['error']}")
必要なスコープ: channels:history
(パブリック), groups:history
(プライベート), mpim:history
(複数人DM), im:history
(個人DM)
ユーザー情報の取得 (users.info)
特定のユーザーIDに関する詳細情報を取得します。
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))
user_id = "U0XXXXXXXXX" # 情報を取得したいユーザーID
try:
result = client.users_info(user=user_id)
user_info = result.get("user")
if user_info:
name = user_info.get("real_name", user_info.get("name", "N/A"))
email = user_info.get("profile", {}).get("email", "N/A")
is_bot = user_info.get("is_bot", False)
print(f"ユーザー情報 ({user_id}):")
print(f" 名前: {name}")
print(f" Email: {email}")
print(f" ボット: {is_bot}")
else:
print(f"ユーザー {user_id} の情報が見つかりませんでした。")
except SlackApiError as e:
print(f"ユーザー情報取得エラー: {e.response['error']}")
必要なスコープ: users:read
, users:read.email
(Email取得時)
メッセージの更新 (chat.update)
投稿済みのメッセージの内容を更新します。
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))
channel_id = "C0XXXXXXXXX"
message_ts = "1678886400.123456" # 更新したいメッセージのタイムスタンプ (ts)
try:
result = client.chat_update(
channel=channel_id,
ts=message_ts,
text="メッセージ内容を更新しました! ✨",
# blocks=[...] # Block Kitで更新も可能
)
print(f"メッセージ (ts: {message_ts}) が更新されました。")
except SlackApiError as e:
print(f"メッセージ更新エラー: {e.response['error']}")
必要なスコープ: chat:write
メッセージの削除 (chat.delete)
投稿済みのメッセージを削除します。
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))
channel_id = "C0XXXXXXXXX"
message_ts = "1678886400.123456" # 削除したいメッセージのタイムスタンプ (ts)
try:
result = client.chat_delete(
channel=channel_id,
ts=message_ts
)
print(f"メッセージ (ts: {message_ts}) が削除されました。")
except SlackApiError as e:
print(f"メッセージ削除エラー: {e.response['error']}")
必要なスコープ: chat:write
非同期処理 (Asyncio) 対応 async
slack-sdk
はPythonの非同期フレームワークasyncio
にも対応しています。WebClient
に対応するAsyncWebClient
クラスなどが提供されており、async/await
構文を使って効率的なI/O処理を行うことができます。
特に、Webフレームワーク(FastAPI, aiohttpなど)と組み合わせてSlackアプリを開発する場合や、多数のAPIリクエストを並行して処理したい場合に有効です。
import os
import asyncio
from slack_sdk.web.async_client import AsyncWebClient
from slack_sdk.errors import SlackApiError
async def send_async_message():
client = AsyncWebClient(token=os.environ.get("SLACK_BOT_TOKEN"))
channel_id = "#general"
try:
response = await client.chat_postMessage(
channel=channel_id,
text="非同期クライアントからのメッセージです!🚀"
)
print(f"非同期メッセージ送信成功: {response['ts']}")
except SlackApiError as e:
print(f"非同期メッセージ送信エラー: {e.response['error']}")
async def main():
await send_async_message()
# 他の非同期タスクも実行可能
# await asyncio.gather(send_async_message(), fetch_user_data())
if __name__ == "__main__":
asyncio.run(main())
AsyncWebClient
のメソッドは基本的にWebClient
と同じですが、呼び出し時にawait
が必要です。slack_sdk
には他にもAsyncWebhookClient
, AsyncSocketModeClient
などが用意されています。
Socket Mode Client 🔌
通常、Slackからのイベント(メッセージ投稿、インタラクションなど)を受け取るには、公開されたHTTPSエンドポイントを用意し、Slack App設定でそのURL(Request URL)を指定する必要があります。しかし、ファイアウォールの内側で開発している場合や、HTTPSサーバーを立てるのが難しい場合があります。
Socket Modeは、このような場合にHTTPSエンドポイントの代わりにWebSocket接続を使ってSlack APIと通信する方法を提供します。アプリがSlackに接続しに行き、そのWebSocket接続を通じてイベントデータを受信したり、インタラクションに応答したりできます。
slack-sdk
にはSocketModeClient
(slack_sdk.socket_mode
) が含まれており、これを利用してSocket Mode接続を確立し、イベントを処理することができます。Socket Modeを利用するには、アプリにconnections:write
スコープと、アプリレベルトークン (xapp-
) が必要です。
slack_bolt
フレームワークを使用すると、Socket Modeの利用がさらに簡単になります。Boltは内部でSocketModeClient
を適切に管理してくれます。
# これはBoltを使ったSocket Modeの例です (SDK単体より推奨)
import os
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
# アプリレベルトークンとボットトークンが必要
app = App(token=os.environ.get("SLACK_BOT_TOKEN"))
@app.message("hello bolt")
def message_hello(message, say):
say(f"Hey there <@{message['user']}>!")
@app.event("app_mention")
def handle_app_mention_events(body, say, logger):
logger.info(body)
say("私にメンションしましたね!")
# アプリを起動
if __name__ == "__main__":
handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
handler.start()
この例では、SocketModeHandler
を使ってアプリを起動しています。これにより、HTTPSサーバーを公開することなく、メッセージイベントやメンションイベントを処理できます。
高度な設定と考慮事項 ⚙️
- レートリミット: Slack APIにはメソッドごとに呼び出し回数制限(レートリミット)があります。SDKは基本的なリトライ処理(
Retry-After
ヘッダーに基づく)を行いますが、大量のリクエストを行う場合はAPIドキュメントを確認し、適切な待機処理を実装する必要があります。 - ページネーション:
conversations.list
やconversations.history
のように大量の結果を返す可能性のあるAPIでは、一度に全てのデータを返しません。レスポンスに含まれるresponse_metadata.next_cursor
を確認し、次のページのデータを取得するためのリクエストを繰り返す必要があります。 - プロキシ設定: プロキシ経由でインターネットに接続する必要がある場合、
WebClient
の初期化時にproxy
引数を指定できます。 - SSL検証: デフォルトではSSL証明書の検証が行われますが、必要に応じて無効化することも可能です(非推奨)。
ssl
引数にカスタムのSSLコンテキストを渡すこともできます。 - タイムアウト: APIリクエストのタイムアウト時間を
timeout
引数で指定できます(デフォルトは30秒)。
from slack_sdk import WebClient
import ssl
# プロキシとカスタムSSLコンテキストの例 (通常は不要)
ssl_context = ssl.create_default_context()
# ssl_context.check_hostname = False # 非推奨
# ssl_context.verify_mode = ssl.CERT_NONE # 非推奨
client = WebClient(
token="your-token",
proxy="http://proxy.example.com:8080",
ssl=ssl_context,
timeout=60 # タイムアウトを60秒に設定
)
まとめ 🎉
Python用Slack公式SDKであるslack-sdk
は、Slack APIとの連携を容易にし、堅牢なアプリケーション開発をサポートする強力なツールです。基本的なメッセージ送信から、Block KitによるリッチなUI作成、ファイルアップロード、非同期処理、Socket Modeまで、幅広い機能を提供します。
この記事では主要な機能と使い方を解説しましたが、slack-sdk
には他にもAudit Logs APIクライアントやSCIM APIクライアントなど、様々な機能が含まれています。
ぜひ公式ドキュメント (https://slack.dev/python-slack-sdk/) やGitHubリポジトリ (https://github.com/slackapi/python-slack-sdk) も参照し、slack-sdk
を活用して、あなたのSlack体験をより便利で効率的なものにしてください!😊
コメント