Nasłuchiwanie zdarzeń i poleceń slash

Przegląd subskrypcji zdarzeń Slack

from slack_bolt import App
import os

app = App(
    token=os.environ['SLACK_BOT_TOKEN'],
    signing_secret=os.environ['SLACK_SIGNING_SECRET']
)

# Register event handlers with @app.event()
# The string argument must match the Slack event type exactly

@app.event('app_mention')
def handle_mention(event, say, logger):
    logger.info(f'Mention event: {event}')
    say(f'Hello <@{event["user"]}>!')

@app.event('message')
def handle_message(event, say):
    # Fires for every message in subscribed channels
    if event.get('subtype') is None:  # ignore bot messages
        print(f'Message: {event["text"]}')

Obsługa zdarzeń app_mention

import re

@app.event('app_mention')
def handle_mention(event, say, client):
    # event['text'] example: '<@U0123BOT> summarize this'
    text = event.get('text', '')
    user_id = event['user']
    channel = event['channel']

    # Remove the @mention to get the clean query
    clean_text = re.sub(r'<@[A-Z0-9]+>', '', text).strip()
    print(f'User {user_id} asked: {clean_text}')

    if not clean_text:
        say(f'Hi <@{user_id}>! How can I help you today?')
        return

    # Process the query
    response = process_user_query(clean_text, user_id)
    say(response)

Dostęp do payloadu zdarzenia

@app.event('app_mention')
def handle_mention(event, say, client):
    print('Event type:', event.get('type'))
    print('User ID:', event.get('user'))     # e.g. 'U0123ABC'
    print('Channel:', event.get('channel'))   # e.g. 'C0456DEF'
    print('Text:', event.get('text'))          # full message text
    print('Timestamp:', event.get('ts'))       # '1234567890.123456'
    print('Thread TS:', event.get('thread_ts')) # if in a thread

    # Get full user info from the user ID
    user_info = client.users_info(user=event['user'])
    real_name = user_info['user']['real_name']
    email = user_info['user']['profile'].get('email', '')
    say(f'Hello {real_name}!')

Polecenia ukośnikowe — rejestrowanie i odpowiadanie

@app.command('/summarize')
def handle_summarize(ack, body, say, respond):
    # CRITICAL: ack() must be called within 3 seconds
    ack()  # acknowledge the command immediately

    # body contains the command payload
    user_id = body['user_id']
    channel_id = body['channel_id']
    text = body.get('text', '').strip()  # text after the command

    print(f'User {user_id} ran /summarize with: "{text}"')

    if not text:
        respond('Usage: /summarize <text or URL to summarize>')
        return

    # Process and respond
    summary = generate_summary(text)
    say(f'Summary by <@{user_id}>:\n{summary}')

ack() — reguła 3 sekund

import threading

@app.command('/analyze')
def handle_analyze(ack, body, respond):
    ack()  # Must be within 3 seconds!

    text = body.get('text', '').strip()
    if not text:
        respond('Please provide text to analyze.')
        return

    # For slow operations: run in background thread
    def process_in_background():
        result = slow_ai_analysis(text)  # may take 10+ seconds
        respond(f'Analysis complete:\n{result}')

    thread = threading.Thread(target=process_in_background)
    thread.daemon = True
    thread.start()

    # respond() is safe to call from a different thread
    # ack() already sent, Slack won't time out

say() a respond() — kiedy używać poszczególnych funkcji

@app.command('/status')
def handle_status(ack, respond, body):
    ack()

    # Ephemeral: only visible to the user who ran the command
    respond(
        text='Agent status: Running | Queue: 3 tasks | Uptime: 4h 22m',
        response_type='ephemeral'  # private to command user
    )

@app.command('/broadcast')
def handle_broadcast(ack, say, respond, body):
    ack()
    text = body.get('text', '')

    # Public: visible to everyone in the channel
    say(
        text=f'<@{body["user_id"]}> broadcast: {text}',
        channel=body['channel_id']
    )

    # Confirm privately to the sender
    respond('Broadcast sent!', response_type='ephemeral')

Zdarzenia wiadomości i podtypy

@app.event('message')
def handle_message(event, say, client):
    subtype = event.get('subtype')

    # Ignore bot messages to prevent loops
    if subtype == 'bot_message':
        return

    # Ignore message edits and deletes
    if subtype in ('message_changed', 'message_deleted'):
        return

    # Only process direct messages (DMs) to the bot
    channel_type = event.get('channel_type', '')
    if channel_type == 'im':
        text = event.get('text', '').strip()
        user = event['user']
        print(f'DM from {user}: {text}')
        say(f'You said: {text}')

Nasłuchiwanie reakcji

@app.event('reaction_added')
def handle_reaction(event, client, say):
    reaction = event['reaction']  # e.g. 'thumbsup', 'white_check_mark'
    user_id = event['user']  # who reacted
    item = event['item']  # what was reacted to

    print(f'User {user_id} reacted :{reaction}: to {item["type"]}')

    if reaction == 'white_check_mark' and item['type'] == 'message':
        # Fetch the original message
        result = client.conversations_history(
            channel=item['channel'],
            oldest=item['ts'],
            latest=item['ts'],
            inclusive=True,
            limit=1
        )
        messages = result.get('messages', [])
        if messages:
            text = messages[0].get('text', '')
            print(f'Task completed: {text[:100]}')

Payloady działań z komponentów interaktywnych

@app.action('approve_task')
def handle_approve(ack, body, respond, client):
    ack()  # acknowledge within 3 seconds

    action = body['actions'][0]  # the button that was clicked
    action_id = action['action_id']  # 'approve_task'
    value = action.get('value', '')  # data attached to the button
    user_id = body['user']['id']

    print(f'User {user_id} clicked {action_id} with value: {value}')

    # Update the original message to show it was approved
    client.chat_update(
        channel=body['container']['channel_id'],
        ts=body['container']['message_ts'],
        text=f'Task approved by <@{user_id}>',
        blocks=[]  # remove buttons after action
    )
    respond('Task approved!', response_type='ephemeral')

Filtrowanie zdarzeń według kanału lub użytkownika

import os

# Only respond in designated channels
ALLOWED_CHANNELS = set(
    os.environ.get('ALLOWED_CHANNELS', '').split(',')
)

BOT_USER_IDS = set()  # will be populated at startup

@app.event('app_mention')
def handle_mention(event, say, client):
    channel = event.get('channel', '')
    user = event.get('user', '')

    # Skip if channel not in allowed list (if list is configured)
    if ALLOWED_CHANNELS and channel not in ALLOWED_CHANNELS:
        return

    # Skip if the 'user' is actually a bot
    if user in BOT_USER_IDS:
        return

    text = event.get('text', '').strip()
    say(f'Processing: {text[:50]}')

Obsługa odpowiedzi w wątkach

@app.event('app_mention')
def handle_mention_in_thread(event, say):
    user = event['user']
    text = event.get('text', '').strip()

    # Reply in the same thread (or start a new one)
    thread_ts = event.get('thread_ts') or event.get('ts')

    response_text = f'<@{user}>, processing your request...'

    say(
        text=response_text,
        thread_ts=thread_ts  # keeps the reply in the thread
    )

    # Do the actual work
    result = do_agent_work(text)

    say(
        text=f'Done! Result:\n{result}',
        thread_ts=thread_ts
    )

Szybki test: czas wywołania ack()

Podsumowanie zdarzeń i poleceń ukośnikowych