slackflare 0.2.0

Lightweight Python Slack Bot framework with Socket Mode and decorator-style API

slackflare

A lightweight Python Slack Bot framework built on Socket Mode with a decorator-style API.

Installation

uv sync

Quick Start

Copy .env.example to .env and fill in your tokens:

SLACK_BOT_TOKEN=xoxb-...
SLACK_APP_TOKEN=xapp-...

from slackflare import SlackFlare, Message, Command

app = SlackFlare()

@app.on_startup()
async def startup() -> None:
    await app.send("#general", "Bot is online!")

@app.on_mention()
async def mentioned(msg: Message) -> str:
    return f"Hi <@{msg.user}>! You said: {msg.text}"

@app.on_message(channel="#general")
async def echo(msg: Message) -> str:
    return f"You said: {msg.text}"

@app.command("!hello")
async def hello(cmd: Command) -> str:
    return f"Hello, <@{cmd.message.user}>!"

app.run()

Slack App Configuration

Bot Token Scopes

Scope	Purpose
chat:write	Send messages
channels:read	Read public channel info
groups:read	Read private channel info
im:read	Read DM info
mpim:read	Read group DM info
app_mentions:read	Receive @mention events (required for on_mention)
reactions:read	Receive emoji reaction events
commands	Receive slash command invocations

App-Level Token Scopes

Scope	Purpose
connections:write	Establish Socket Mode connections

Event Subscriptions (Bot Events)

Event	Description
message.channels	Public channel messages
message.groups	Private channel messages
message.im	Direct messages
message.mpim	Group DM messages
app_mention	@mention events
reaction_added	Emoji reaction events

Interactivity & Shortcuts

Enable Features → Interactivity & Shortcuts → Enable. No Request URL is needed in Socket Mode. Once enabled, the app can receive block_actions (buttons, menus) and view_submission (Modal forms) events.

Slash Commands

Register each command (e.g., /deploy) under Features → Slash Commands → Create New Command. No Request URL is needed in Socket Mode.

---

Features

Event Handlers

@app.on_message(channel=None)

Listen for channel messages. channel accepts "#name" or a channel ID; None listens to all channels.

@app.on_message(channel="#support")
async def handle(msg: Message) -> str:
    return f"Received: {msg.text}"

@app.on_mention(channel=None)

Listen for @botname events. The msg.text automatically strips the <@BOTID> prefix.

@app.on_mention()
async def handle(msg: Message) -> str:
    return f"You called? You said: {msg.text}"

@app.command(name)

Listen for !command-style text commands with argument support.

@app.command("!ping")
async def ping(cmd: Command) -> str:
    # cmd.name    → "ping"
    # cmd.args    → ["arg1", "arg2", ...]
    # cmd.message → original Message object
    return "pong"

@app.on_action(action_id)

Listen for Block Kit interactive events (button clicks, menu selections, etc.).

from slackflare import Action

@app.on_action("approve_btn")
async def on_approve(action: Action) -> str:
    # action.action_id   → "approve_btn"
    # action.value       → button value (may be None)
    # action.user        → user ID of the person who clicked
    # action.channel     → Channel ID
    # action.message_ts  → original message timestamp
    # action.thread_ts   → thread timestamp (may be None)
    # action.raw         → full interactive payload
    return f"<@{action.user}> approved!"

@app.on_reaction(emoji=None)

Listen for emoji reaction events. emoji can specify a particular emoji name; None listens to all.

from slackflare import Reaction

@app.on_reaction(emoji="thumbsup")
async def on_thumbsup(reaction: Reaction) -> str:
    # reaction.emoji       → "thumbsup"
    # reaction.user        → user ID of the person who reacted
    # reaction.channel     → Channel ID
    # reaction.message_ts  → timestamp of the reacted message
    # reaction.raw         → full event payload
    return f"<@{reaction.user}> gave a thumbs up!"

@app.on_reaction()  # listen to all emojis
async def on_any_reaction(reaction: Reaction) -> None:
    print(f"{reaction.emoji} from {reaction.user}")

@app.slash("/command")

Listen for Slack slash commands. The command name can include or omit the leading /.

from slackflare import SlashCommand

@app.slash("/deploy")
async def deploy(cmd: SlashCommand) -> str:
    # cmd.name         → "deploy"
    # cmd.text         → text after the command (e.g., "/deploy prod" → "prod")
    # cmd.user         → sender's User ID
    # cmd.channel      → Channel ID
    # cmd.trigger_id   → can be used to open a Modal
    # cmd.response_url → delayed response URL
    # cmd.raw          → full slash command payload
    return f"Deploying {cmd.text}..."

@app.on_view(callback_id)

Listen for Modal form submission events. callback_id corresponds to the value set when opening the Modal.

from slackflare import ViewSubmission

@app.on_view("feedback_form")
async def on_feedback(view: ViewSubmission) -> str:
    # view.callback_id → "feedback_form"
    # view.view_id     → Modal's View ID
    # view.user        → submitter's User ID
    # view.values      → flattened form values {action_id: value_obj}
    # view.trigger_id  → Trigger ID
    # view.raw         → full view_submission payload
    name = view.values["name_input"]["value"]
    return f"Received feedback from {name}, thanks!"

@app.on_startup()

Executed after the bot connects successfully. Useful for sending online notifications or initialization.

@app.on_startup()
async def startup() -> None:
    await app.send("#ops", "Bot started")

@app.on_shutdown()

Executed before the bot disconnects. Useful for cleanup, saving state, or sending offline notifications. Multiple shutdown handlers run in registration order; errors in one handler do not block the others.

@app.on_shutdown()
async def cleanup() -> None:
    await app.send("#ops", "Bot shutting down")

@app.cron(expression)

Run a coroutine on a cron schedule. The cron expression is validated at decoration time.

@app.cron("*/5 * * * *")  # every 5 minutes
async def periodic_report() -> None:
    await app.send("#ops", "Periodic health check: OK")

@app.interval(seconds)

Run a coroutine at a fixed interval. The first execution happens after the initial delay.

@app.interval(seconds=60)
async def heartbeat() -> None:
    await app.send("#ops", "Heartbeat")

@app.on_error()

Register a global error handler that is called whenever a handler (or startup/shutdown hook) raises an exception. Receives the exception and the event object (or None for lifecycle hooks).

@app.on_error()
async def handle_error(exc: Exception, event) -> None:
    print(f"Error: {exc} during {event}")

@app.middleware()

Register middleware that wraps every event handler. Middleware receives the event and a call_next async callable. Call call_next(event) to invoke the next middleware or the final handler. Middleware can short-circuit by returning a value without calling call_next. Middleware runs in registration order (first registered = outermost).

import time

@app.middleware()
async def log_timing(event, call_next):
    start = time.time()
    result = await call_next(event)
    print(f"Handler took {time.time() - start:.3f}s")
    return result

@app.middleware()
async def auth_check(event, call_next):
    if hasattr(event, "user") and event.user == "UBANNED":
        return "You are not allowed."  # short-circuit
    return await call_next(event)

Block Kit Builder

slackflare re-exports slack-sdk Block Kit components and provides factory functions for easier construction:

from slackflare import Reply, SectionBlock, DividerBlock, ButtonElement
from slackflare.blocks import section, divider, button, actions, header, context, md

# Using factory functions
@app.on_message()
async def rich_reply(msg: Message) -> Reply:
    return Reply(
        text="fallback text",
        blocks=[
            header("Announcement"),
            section("This is an important message", accessory=button("Learn More", "btn_more", value="info")),
            divider(),
            actions([
                button("Approve", "btn_approve", value="yes", style="primary"),
                button("Reject", "btn_reject", value="no", style="danger"),
            ]),
            context("Auto-generated by Bot"),
        ],
    )

Available Block types: SectionBlock, DividerBlock, HeaderBlock, ImageBlock, ActionsBlock, ContextBlock, InputBlock

Available Element types: ButtonElement, StaticSelectElement, PlainTextInputElement, UserSelectElement, ConversationSelectElement, ChannelSelectElement

Factory functions: section(), divider(), header(), image(), actions(), context(), button(), md(), plain(), text_input(), static_select(), user_select(), conversation_select(), channel_select(), input_block(), option(), confirm()

When a handler returns Reply(text, blocks=...), block objects are automatically serialized and reply() is called.

WebhookSender

Send messages without a Bot Token — ideal for CI/CD notifications or external system integrations:

from slackflare import WebhookSender
from slackflare.blocks import section, divider

sender = WebhookSender("https://hooks.slack.com/services/T.../B.../xxx")

# Plain text
await sender.send("Deployment complete!")

# With Block Kit
await sender.send("Deployment notification", blocks=[
    section("*Production* deployment succeeded"),
    divider(),
])

Proactive Messaging

await app.send(channel, text, **kwargs)

Proactively send messages to any channel or user. Supports all chat.postMessage parameters.

await app.send("#general", "Scheduled notification")
await app.send("U12345678", "Direct message")
await app.send("#general", "Thread reply", thread_ts="1234567890.123456")

Message Object

Property	Type	Description
msg.text	str	Message content (mention prefix stripped)
msg.user	str	Sender's User ID
msg.channel	str	Channel ID
msg.channel_name	str	Channel name
msg.ts	str	Message timestamp
msg.thread_ts	str | None	Thread timestamp
msg.mention	bool	Whether triggered by @mention
msg.files	list[File]	Attached files
msg.raw	dict	Raw Slack event payload

await msg.reply(text)

Reply within the same thread (automatically includes thread_ts).

await msg.reply_with_file(file, title="")

Upload a file to the same channel.

Handler Return Values

Returning str from a handler automatically calls reply(); returning dict allows text and blocks fields; returning Reply supports Block Kit blocks; returning None sends no reply.

@app.on_message()
async def handler(msg: Message) -> str | None:
    if "hello" in msg.text:
        return "Hi!"       # auto-reply
    # no return → no reply

---

Testing

# Unit tests (no network required)
uv run pytest

# Integration tests (requires real tokens in .env)
uv run pytest -m integration -v

# With coverage report
uv run pytest --cov=slackflare

Integration tests are automatically skipped when .env is missing or tokens are placeholders. Set TEST_SLACK_CHANNEL=#channel-name in .env to specify the test channel (defaults to #general).

Architecture

src/slackflare/
├── app.py              # SlackFlare main class, decorator API
├── router.py           # Message routing (message / mention / command / action / reaction / slash / view)
├── models.py           # Message, Command, Action, Reaction, SlashCommand, ViewSubmission data models
├── blocks.py           # Block Kit builder (re-exports slack-sdk + factory functions)
├── webhook.py          # WebhookSender — send messages via Incoming Webhooks
├── scheduler.py        # Cron and interval task scheduler (pure asyncio + croniter)
├── config.py           # Pydantic-settings environment configuration
├── _version.py         # Version string
├── py.typed            # PEP 561 type marker
└── transports/
    ├── base.py         # BaseTransport abstract interface
    └── slack.py        # Slack Socket Mode implementation

Routing priority: mention → command (!xxx) → message

Actions, Reactions, Slash Commands, and View Submissions are dispatched independently.