0Pricing
AI Agents · 课时

监听事件与斜杠命令

Slack Bolt 中的 app_mention、斜杠命令和操作处理器。

监听事件与斜杠命令 是 CoddyKit 上的免费 AI Agents 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

Slack 事件订阅概览

Slack 会在发生各种情况时向您的应用发送 事件，例如有人发布消息、提及您的机器人，或用户加入频道。您可以在 Slack 应用控制面板中订阅特定事件类型，然后在 Bolt 中使用 @app.event() 装饰器注册处理程序。

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"]}')

处理 app_mention 事件

当有人在频道中输入 @YourBot 时，app_mention 事件会触发。event['text'] 包含包括提及内容在内的完整消息。请去除提及前缀，以获取用户的实际查询内容。

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)

访问事件负载

每个事件处理程序都会接收 event 字典，其中包含原始 Slack 事件负载。关键字段包括：event['user']（用户 ID）、event['channel']（频道 ID）、event['text']（消息内容）、event['ts']（时间戳/消息 ID）。

@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}!')

斜杠命令 — 注册与响应

斜杠命令让用户可以从任意 Slack 频道触发代理操作。请在 Slack 应用控制面板中注册命令 URL（位于 斜杠命令下），然后使用 @app.command('/command-name') 进行处理。请始终立即调用 ack()——如果不这样做，Slack 会在 3 秒后超时。

@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() — 3 秒规则

Slack 要求您的应用在 3 秒内对每个传入的斜杠命令和交互式负载调用 ack()（确认）。如果不这样做，Slack 会向用户显示错误。对于长时间运行的操作，请立即确认，在后台线程中开始处理，然后使用 respond() 返回结果。

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() 与 respond() — 何时使用

有两个函数可以向 Slack 发布消息：

  • say() — 发布到事件发生的频道；所有人都能看到
  • respond() — 仅在斜杠命令处理程序中可用；可以发布临时消息，仅命令发起者可见

请使用 respond(response_type='in_channel') 发布公开响应，使用 respond(response_type='ephemeral') 发布私密响应。

@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')

消息事件与子类型

通用的 message 事件会在所有消息发生时触发，包括机器人消息、编辑和删除操作。请使用 subtype 字段进行筛选。常见子类型包括：bot_message、message_changed、message_deleted。如果不存在 subtype，则表示这是普通用户消息。

@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}')

监听反应

当有人向消息添加表情反应时，reaction_added 事件会触发。您可以使用它来触发代理操作——例如，添加 📌 反应来保存消息，或添加 ✅ 反应来标记任务已完成。

@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]}')

交互式组件的操作负载

当用户点击按钮或选择菜单项时，Slack 会发送操作负载。请使用 @app.action('action_id') 进行处理。操作 ID 是您创建 Block Kit 组件时设置的字符串。请始终立即调用 ack()。

@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')

按频道或用户筛选事件

在大型工作区中，您的机器人可能会接收来自许多频道的事件。请在处理程序的早期进行筛选，只处理相关事件。将 event['channel'] 与允许列表进行比对，或检查 event['user'] 以忽略特定用户（例如其他机器人）。

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]}')

支持线程回复

若要在线程中回复，而不是发布到主频道，请将 thread_ts 传递给 say()。使用 event.get('thread_ts', event['ts']) 获取线程时间戳——只有当消息已经位于线程中时才存在 thread_ts；否则，请使用消息自身的 ts 来开始新线程。

@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
    )

快速检查：ack() 的时机

测试您对 Slack 事件处理的理解。

事件与斜杠命令回顾

现在，您的 Slack 代理可以响应各种用户交互：

  • @app.event('app_mention') — 处理对 @bot 的提及；使用正则表达式去除提及前缀
  • @app.command('/cmd') — 处理斜杠命令；始终在 3 秒内调用 ack()
  • say() — 公开发布到频道；respond() — 回复斜杠命令（可以是临时消息）
  • @app.action('id') — 处理按钮点击和交互式组件操作
  • @app.event('reaction_added') — 在收到表情反应时触发
  • 在 say() 中使用 thread_ts，让回复保留在线程中
  • 尽早按频道/用户进行筛选，避免处理无关事件

常见问题解答

「监听事件与斜杠命令」课时是免费的吗？

是的 — 「监听事件与斜杠命令」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「监听事件与斜杠命令」这节课中我会学到什么？

Slack Bolt 中的 app_mention、斜杠命令和操作处理器。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。

「监听事件与斜杠命令」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. Slack Bolt SDK 基础
  2. 监听事件与斜杠命令
  3. 发送消息与富文本块
  4. 构建团队通知机器人
← 返回 AI Agents