The Bolt Framework for Python - a comprehensive framework for building Slack applications with decorator-based API for handling events, actions, and interactive components
—
OAuth 2.0 flow implementation for multi-workspace Slack app installation and management. Includes installation stores, token management, and authorization patterns for enterprise-grade Slack applications.
Configure OAuth flow settings for multi-workspace app installation.
class OAuthSettings:
"""OAuth configuration settings for multi-workspace apps."""
def __init__(
self,
*,
client_id: str,
client_secret: str,
scopes: Union[str, Sequence[str]],
user_scopes: Union[str, Sequence[str]] = None,
redirect_uri_path: str = "/slack/oauth_redirect",
install_path: str = "/slack/install",
success_url: str = None,
failure_url: str = None,
authorization_url: str = "https://slack.com/oauth/v2/authorize",
token_url: str = "https://slack.com/api/oauth.v2.access",
state_store=None,
installation_store=None,
install_page_rendering_enabled: bool = True,
redirect_uri_page_rendering_enabled: bool = True
):
"""
Initialize OAuth settings.
Args:
client_id (str): Slack app client ID
client_secret (str): Slack app client secret
scopes (Union[str, Sequence[str]]): Bot token scopes
user_scopes (Union[str, Sequence[str]], optional): User token scopes
redirect_uri_path (str): OAuth redirect endpoint path
install_path (str): Installation start endpoint path
success_url (str, optional): Success redirect URL
failure_url (str, optional): Error redirect URL
authorization_url (str): Slack authorization endpoint
token_url (str): Slack token exchange endpoint
state_store: Custom OAuth state store
installation_store: Custom installation data store
install_page_rendering_enabled (bool): Enable built-in install page
redirect_uri_page_rendering_enabled (bool): Enable built-in redirect page
"""Main OAuth flow handler that manages the complete installation process.
class OAuthFlow:
"""OAuth 2.0 flow handler for Slack app installation."""
def __init__(
self,
*,
settings: OAuthSettings,
logger: Logger = None
):
"""
Initialize OAuth flow handler.
Args:
settings (OAuthSettings): OAuth configuration
logger (Logger, optional): Custom logger instance
"""
def handle_installation(self, request):
"""
Handle installation request.
Args:
request: HTTP request object
Returns:
Response: Installation page or redirect response
"""
def handle_callback(self, request):
"""
Handle OAuth callback.
Args:
request: HTTP request with authorization code
Returns:
Response: Success/failure page or redirect
"""Stores for persisting installation data across workspace installations.
class InstallationStore:
"""Base class for installation data persistence."""
def save(self, installation):
"""
Save installation data.
Args:
installation: Installation data object
"""
raise NotImplementedError()
def find_installation(
self,
*,
enterprise_id: str = None,
team_id: str = None,
user_id: str = None,
is_enterprise_install: bool = None
):
"""
Find installation data.
Args:
enterprise_id (str, optional): Enterprise ID
team_id (str, optional): Team/workspace ID
user_id (str, optional): User ID
is_enterprise_install (bool, optional): Enterprise install flag
Returns:
Installation data or None
"""
raise NotImplementedError()
class FileInstallationStore(InstallationStore):
"""File-based installation store implementation."""
def __init__(self, base_dir: str = "./data/installations"):
"""
Initialize file-based store.
Args:
base_dir (str): Directory for installation data files
"""
class MemoryInstallationStore(InstallationStore):
"""In-memory installation store (for development only)."""
def __init__(self):
"""Initialize in-memory store."""Objects representing authorization data for requests.
class AuthorizeResult:
"""Result of authorization process."""
def __init__(
self,
*,
enterprise_id: str = None,
team_id: str = None,
user_id: str = None,
bot_id: str = None,
bot_user_id: str = None,
bot_token: str = None,
user_token: str = None,
bot_scopes: list = None,
user_scopes: list = None
):
"""
Initialize authorization result.
Args:
enterprise_id (str, optional): Enterprise Grid organization ID
team_id (str, optional): Workspace/team ID
user_id (str, optional): User ID
bot_id (str, optional): Bot app ID
bot_user_id (str, optional): Bot user ID
bot_token (str, optional): Bot access token (xoxb-...)
user_token (str, optional): User access token (xoxp-...)
bot_scopes (list, optional): Bot token scopes
user_scopes (list, optional): User token scopes
"""Implement custom authorization logic for advanced use cases.
def authorize_function(
*,
enterprise_id: str = None,
team_id: str = None,
user_id: str = None,
logger: Logger
) -> AuthorizeResult:
"""
Custom authorization function signature.
Args:
enterprise_id (str, optional): Enterprise ID from request
team_id (str, optional): Team ID from request
user_id (str, optional): User ID from request
logger (Logger): App logger instance
Returns:
AuthorizeResult: Authorization data or None if not authorized
"""import os
from slack_bolt import App
from slack_bolt.oauth import OAuthSettings
# Configure OAuth settings
oauth_settings = OAuthSettings(
client_id=os.environ["SLACK_CLIENT_ID"],
client_secret=os.environ["SLACK_CLIENT_SECRET"],
scopes=["chat:write", "commands", "app_mentions:read"],
install_path="/slack/install",
redirect_uri_path="/slack/oauth_redirect"
)
# Initialize app with OAuth
app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
oauth_settings=oauth_settings
)
@app.event("app_mention")
def handle_mention(event, say):
say(f"Thanks for mentioning me, <@{event['user']}>!")
# OAuth endpoints are automatically handled
app.start(port=3000)import os
import json
from slack_bolt import App
from slack_bolt.oauth import OAuthSettings
from slack_bolt.oauth.installation_store import InstallationStore
class DatabaseInstallationStore(InstallationStore):
"""Custom database-backed installation store."""
def __init__(self, database_url):
self.db = connect_to_database(database_url)
def save(self, installation):
"""Save installation to database."""
self.db.installations.insert_one({
"enterprise_id": installation.enterprise_id,
"team_id": installation.team_id,
"bot_token": installation.bot_token,
"bot_user_id": installation.bot_user_id,
"bot_scopes": installation.bot_scopes,
"user_id": installation.user_id,
"user_token": installation.user_token,
"user_scopes": installation.user_scopes,
"installed_at": installation.installed_at
})
def find_installation(self, *, enterprise_id=None, team_id=None, user_id=None, is_enterprise_install=None):
"""Find installation in database."""
query = {}
if enterprise_id:
query["enterprise_id"] = enterprise_id
if team_id:
query["team_id"] = team_id
if user_id:
query["user_id"] = user_id
result = self.db.installations.find_one(query)
if result:
return Installation(
app_id=result["app_id"],
enterprise_id=result.get("enterprise_id"),
team_id=result.get("team_id"),
bot_token=result.get("bot_token"),
bot_user_id=result.get("bot_user_id"),
# ... map other fields
)
return None
# Use custom installation store
oauth_settings = OAuthSettings(
client_id=os.environ["SLACK_CLIENT_ID"],
client_secret=os.environ["SLACK_CLIENT_SECRET"],
scopes=["chat:write", "commands"],
)
app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
oauth_settings=oauth_settings,
installation_store=DatabaseInstallationStore(os.environ["DATABASE_URL"])
)from slack_bolt import App
from slack_bolt.authorization import AuthorizeResult
def custom_authorize(*, enterprise_id, team_id, user_id, logger):
"""Custom authorization with business logic."""
# Check if workspace is allowed
if team_id not in ALLOWED_WORKSPACES:
logger.warning(f"Unauthorized workspace: {team_id}")
return None
# Get installation data from custom store
installation = get_installation_from_db(team_id, user_id)
if not installation:
logger.warning(f"No installation found for team {team_id}")
return None
# Check subscription status
if not check_subscription_active(team_id):
logger.warning(f"Inactive subscription for team {team_id}")
return None
# Return authorization result
return AuthorizeResult(
enterprise_id=enterprise_id,
team_id=team_id,
user_id=user_id,
bot_token=installation["bot_token"],
bot_user_id=installation["bot_user_id"],
bot_scopes=installation["bot_scopes"]
)
app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
authorize=custom_authorize
)# Request both bot and user scopes
oauth_settings = OAuthSettings(
client_id=os.environ["SLACK_CLIENT_ID"],
client_secret=os.environ["SLACK_CLIENT_SECRET"],
scopes=["chat:write", "commands"], # Bot scopes
user_scopes=["search:read", "files:read"] # User scopes
)
app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
oauth_settings=oauth_settings
)
@app.command("/search")
def search_command(ack, command, client, context):
"""Command that uses user token for search."""
ack()
# Use user token from context
user_client = WebClient(token=context.user_token)
try:
# Search using user permissions
search_result = user_client.search_messages(
query=command['text'],
count=5
)
messages = search_result["messages"]["matches"]
if messages:
response = "Search results:\n"
for msg in messages[:3]:
response += f"• {msg['text'][:100]}...\n"
else:
response = "No messages found matching your search."
client.chat_postEphemeral(
channel=command['channel_id'],
user=command['user_id'],
text=response
)
except Exception as e:
client.chat_postEphemeral(
channel=command['channel_id'],
user=command['user_id'],
text="Sorry, search failed. Make sure you've granted the necessary permissions."
)oauth_settings = OAuthSettings(
client_id=os.environ["SLACK_CLIENT_ID"],
client_secret=os.environ["SLACK_CLIENT_SECRET"],
scopes=["chat:write", "commands"],
# Enable org-wide installation for Enterprise Grid
install_path="/slack/install",
redirect_uri_path="/slack/oauth_redirect"
)
def enterprise_authorize(*, enterprise_id, team_id, user_id, logger):
"""Authorization for Enterprise Grid installations."""
if enterprise_id:
# This is an Enterprise Grid installation
logger.info(f"Enterprise installation: {enterprise_id}, team: {team_id}")
# Find org-wide or team-specific installation
installation = find_enterprise_installation(enterprise_id, team_id)
if installation:
return AuthorizeResult(
enterprise_id=enterprise_id,
team_id=team_id,
user_id=user_id,
bot_token=installation.bot_token,
bot_user_id=installation.bot_user_id
)
# Fallback to regular team installation
return find_team_installation(team_id, user_id)
app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
oauth_settings=oauth_settings,
authorize=enterprise_authorize
)from slack_bolt.oauth.oauth_settings import OAuthSettings
from slack_bolt.oauth.state_store import FileOAuthStateStore
# Custom state store for OAuth security
state_store = FileOAuthStateStore(
expiration_seconds=600, # 10 minutes
base_dir="./data/oauth_states"
)
oauth_settings = OAuthSettings(
client_id=os.environ["SLACK_CLIENT_ID"],
client_secret=os.environ["SLACK_CLIENT_SECRET"],
scopes=["chat:write", "commands"],
state_store=state_store # Custom state management
)
app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
oauth_settings=oauth_settings
)oauth_settings = OAuthSettings(
client_id=os.environ["SLACK_CLIENT_ID"],
client_secret=os.environ["SLACK_CLIENT_SECRET"],
scopes=["chat:write", "commands"],
success_url="https://yourapp.com/slack/success",
failure_url="https://yourapp.com/slack/error",
# Disable built-in pages to use custom redirects
install_page_rendering_enabled=False,
redirect_uri_page_rendering_enabled=False
)
app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
oauth_settings=oauth_settings
)from fastapi import FastAPI, Request
from slack_bolt import App
from slack_bolt.adapter.fastapi import SlackRequestHandler
from slack_bolt.oauth import OAuthSettings
oauth_settings = OAuthSettings(
client_id=os.environ["SLACK_CLIENT_ID"],
client_secret=os.environ["SLACK_CLIENT_SECRET"],
scopes=["chat:write", "commands"]
)
bolt_app = App(
signing_secret=os.environ["SLACK_SIGNING_SECRET"],
oauth_settings=oauth_settings
)
fastapi_app = FastAPI()
handler = SlackRequestHandler(bolt_app)
# OAuth endpoints
@fastapi_app.get("/slack/install")
async def install(request: Request):
return await handler.handle_async(request)
@fastapi_app.get("/slack/oauth_redirect")
async def oauth_redirect(request: Request):
return await handler.handle_async(request)
# Event endpoint
@fastapi_app.post("/slack/events")
async def endpoint(request: Request):
return await handler.handle_async(request)Install with Tessl CLI
npx tessl i tessl/pypi-slack-bolt