# Slack API 技巧使用 Slack 平台掌握 Slack 机器人开发和工作区自动化。该技能涵盖 Web API、事件 API、套接字模式、Block Kit UI 框架以及用于构建生产就绪的 Slack 应用程序的 Python Bolt SDK。 ## 何时使用此技能 ### 在以下情况下使用： * 构建 CI/CD 管道的通知系统 * 为团队工作流程创建交互式机器人 * 自动事件响应和警报 * 使用交互式消息构建审批工作流程 * 将外部服务与 Slack 渠道集成 * 创建常用操作的斜杠命令 * 使用模式对话框构建内部工具 * 实现定时消息自动化 ### 不要在以下情况下使用： * Microsoft Teams 是主要平台（使用 team-api） * 仅简单的单向通知（直接使用传入的 webhook） * 需要基于电子邮件的工作流程（不同域） * 具有复杂组织要求的 Slack Enterprise Grid * 实时游戏或高频更新（考虑WebSockets） ## 先决条件 ### Slack 应用程序设置 # 1. 在 https://api.slack.com/apps 创建一个 Slack 应用程序 # 2. 选择“从头开始”并选择您的工作区 # 所需的机器人令牌范围（OAuth 和权限）： # - 聊天：写 - 发布消息 # - chat:write.public - 无需加入即可发布到频道 # -channels:read - 列出公共频道 # - Channels:history - 读取频道消息 # - groups:read - 列出私人频道 # - im:read - 列出直接消息 # - users:read - 访问用户信息 # - files:write - 上传文件 # - 反应：写入 - 添加反应 # - 命令 - 添加斜杠命令 # 事件订阅（对于事件 API）： # - message.channels - 公共频道中的消息 # - message.groups - 私人频道中的消息 # - message.im - 直接消息 # - app_mention - 当提到机器人时 # 交互组件： # - 在应用程序设置中启用 # - 设置按钮/选择处理的请求 URL ### Python环境设置 # 创建虚拟环境 python -m venv slack-bot-env 源 slack-bot-env/bin/activate # Linux/macOS # slack-bot-env\Scripts\activate # Windows # 安装 Slack Bolt SDK pip 安装 slack-bolt slack-sdk # 安装额外的依赖项 pip install python-dotenv aiohttp 请求 # 创建需求.txt 猫 > 要求.txt << 'EOF' 松紧螺栓>=1.18.0 slack-sdk>=3.21.0 python-dotenv>=1.0.0 aiohttp>=3.9.0 请求>=2.31.0 EOF # 环境变量猫 > .env << 'EOF' SLACK_BOT_TOKEN=xoxb-your-bot-token SLACK_SIGNING_SECRET=您的签名秘密 SLACK_APP_TOKEN=xapp-your-app-token # 对于套接字模式 EOF ### 使用 ngrok 进行本地开发 # 安装 ngrok 酿造安装 ngrok # macOS # 或者从 https://ngrok.com/download 下载 # 验证ngrok ngrok 配置 add-authtoken YOUR_AUTH_TOKEN # 启动本地开发隧道恩格洛克 http 3000 # 使用 HTTPS URL： # - 事件订阅请求 URL # - 交互组件请求 URL # - 斜杠命令请求 URL ## 核心能力 ### 1\.带 Bolt 的基本 Slack 机器人 # 应用程序.py # 关于我：使用 Bolt 框架的基本 Slack 机器人 # ABOUTME：处理消息、提及和斜线命令导入操作系统从 dotenv 导入 load_dotenv 从 slack_bolt 导入应用程序从 slack_bolt.adapter.socket_mode 导入 SocketModeHandler 加载_dotenv() # 使用机器人令牌和签名密钥初始化应用程序应用程序 = 应用程序（ token=os.environ.get("SLACK_BOT_TOKEN"), signing_secret=os.environ.get("SLACK_SIGNING_SECRET") ） # 监听包含“hello”的消息 @app.message(“你好”) def message_hello（消息，说）： """回复包含“你好”的消息""" 用户 = 消息['用户'] 说（f“嘿，<@{user}>！”） # 监听应用程序的提及 @app.event(“app_mention”) def handle_app_mention（事件，比如说客户端）： """提及机器人时做出回应""" 用户 = 事件['用户'] 频道 = 事件['频道'] 文本 = 事件['文本'] # 获取用户信息 user_info = client.users_info(用户=用户) user_name = user_info['用户']['真实姓名'] say(f“嗨 {user_name}！您提到我时用的是：{text}”) # 处理消息事件 @app.event(“消息”) defhandle_message_events（主体，记录器）： """记录所有消息事件""" logger.info(f"消息事件：{body}") # 斜杠命令处理程序 @app.command(“/问候”) defhandle_greet_command（确认，说，命令）： """处理/问候斜杠命令""" ack() # 3秒内确认命令用户 = 命令['user_id'] 文本 = command.get('文本', '每个人') say(f"<@{user}> 向 {text} 发送问候！") # 错误处理程序 @应用程序错误 def custom_error_handler（错误，正文，记录器）： """优雅地处理错误""" logger.exception(f"错误: {error}") logger.info(f"请求正文：{body}") # 以套接字模式运行（不需要公共 URL）如果 __name__ == "__main__": 处理程序 = SocketModeHandler( 应用程序， os.environ.get("SLACK_APP_TOKEN") ） print("机器人正在运行...") 处理程序.start() ### 2\.阻止套件消息 # 块.py # ABOUTME: Block Kit message construction utilities # ABOUTME: Creates rich, interactive Slack messages 从 slack_bolt 导入应用程序导入操作系统 app = App(token=os.environ.get("SLACK_BOT_TOKEN")) def create_deployment_message( 环境：str，版本：str，状态：str，部署网址：str，日志网址：str ) -> 列表： """Create a deployment notification with Block Kit""" 状态表情符号= { “成功”：“：white_check_mark：”， “失败”：“：x：”， "in_progress": ":hourglass_flowing_sand:", “待处理”：“：时钟3：” } emoji = status_emoji.get(status, ":问题:") 块= [ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": f"{emoji} 部署 {status.title()}", “表情符号”：正确 } }, { “类型”：“部分”， “字段”：[ { “类型”：“mrkdwn”， "text": f"*环境:*\n{环境}" }, { “类型”：“mrkdwn”， "text": f"*版本:*\n{版本}" }, { “类型”：“mrkdwn”， "text": f"*状态:*\n{status.title()}" }, { “类型”：“mrkdwn”， "text": f"*Time:*\n" } ] }, { “类型”：“分隔符” }, { “类型”：“动作”， “元素”：[ { “类型”：“按钮”， “文本”：{ “类型”：“纯文本”， "text": "查看部署", “表情符号”：正确 }, “网址”：部署网址， “风格”：“主要” }, { “类型”：“按钮”， “文本”：{ “类型”：“纯文本”， "text": "查看日志", “表情符号”：正确 }, “url”：logs_url } ] }, { “类型”：“上下文”， “元素”：[ { “类型”：“mrkdwn”， "text": "由 CI/CD 管道部署" } ] } ] 返回块 def create_approval_message( 请求 ID：str，请求者：str，描述：str，详细信息：字典 ) -> 列表： """Create an approval request with interactive buttons""" 块= [ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": ":clipboard: 批准请求", “表情符号”：正确 } }, { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": f"*Request ID:* `{request_id}`\n*Requested by:* <@{requester}>\n\n{description}" } }, { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": "*详细信息:*\n" + "\n".join( f"- {k}: {v}" 代表details.items()中的k、v ） } }, { “类型”：“分隔符” }, { “类型”：“动作”， "block_id": f"approval_{request_id}", “元素”：[ { “类型”：“按钮”， “文本”：{ “类型”：“纯文本”， "text": "批准", “表情符号”：正确 }, “风格”：“主要”， "action_id": "批准请求", “值”：request_id }, { “类型”：“按钮”， “文本”：{ “类型”：“纯文本”， "text": "拒绝", “表情符号”：正确 }, “风格”：“危险”， "action_id": "拒绝请求", “值”：request_id }, { “类型”：“按钮”， “文本”：{ “类型”：“纯文本”， "text": "请求信息", “表情符号”：正确 }, "action_id": "request_info", “值”：request_id } ] } ] 返回块 def create_poll_message(问题: str, 选项: 列表) -> 列表: """使用单选按钮创建投票""" 选项元素 = [ { “文本”：{ “类型”：“纯文本”， “文本”：选项， “表情符号”：正确 }, “值”：f“选项_{i}” } 对于 i，枚举中的选项（选项） ] 块= [ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， “文本”：“：bar_chart：民意调查”， “表情符号”：正确 } }, { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， “文本”：f“*{问题}*” } }, { “类型”：“分隔符” }, { “类型”：“部分”， "block_id": "轮询选项", “文本”：{ “类型”：“mrkdwn”， "text": "选择您的选择：" }, “配件”：{ “类型”：“单选按钮”， "action_id": "投票投票", “选项”：选项元素 } }, { “类型”：“动作”， “元素”：[ { “类型”：“按钮”， “文本”：{ “类型”：“纯文本”， "text": "提交投票", “表情符号”：正确 }, “风格”：“主要”， “action_id”：“提交投票” } ] } ] 返回块 # 发送部署通知 def send_deployment_notification（通道：str）： """向频道发送部署通知""" 块= create_deployment_message( 环境=“生产”，版本=“v2.1.0”，状态=“成功”，部署_url =“https://app.example.com”， Logs_url =“https://logs.example.com” ） app.client.chat_postMessage( 频道=频道，块=块， text="部署通知" # 通知的后备） ### 3\.交互组件和动作 # 交互.py # ABOUTME：处理交互组件，如按钮、选择、模式 # ABOUTME：通过状态管理实现审批工作流程从 slack_bolt 导入应用程序从日期时间导入日期时间导入 json 导入操作系统应用程序=应用程序（令牌= os.environ.get（“SLACK_BOT_TOKEN”）） # 内存存储（在生产中使用数据库）批准请求 = {} @app.action("approve_request") defhandle_approve（ack，主体，客户端，记录器）： """处理批准按钮点击""" 确认（）用户 = 正文['用户']['id'] request_id = body['actions'][0]['value'] 频道=正文['频道']['id'] message_ts = body['消息']['ts'] # 更新消息以显示批准 update_blocks = body['message']['blocks'].copy() # 删除操作按钮 Updated_blocks = [b for b in Updated_blocks if b.get('type') != 'actions'] # 添加批准状态更新的_blocks.append({ “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": f":white_check_mark: *已由 <@{user}> 在 {datetime.now().strftime('%Y-%m-%d %H:%M')} 批准*" } }) # 更新原始消息客户端.chat_update( 频道=频道， ts=消息_ts，块=更新的块，文本=“请求已批准” ） # 商店批准批准请求[请求 ID] = { “状态”：“已批准”， “approved_by”：用户， “时间戳”：datetime.now().isoformat() } logger.info(f"请求 {request_id} 已被 {user} 批准") @app.action(“拒绝请求”) def handle_reject(ack, 主体, 客户端, 响应): """用原因模式处理拒绝""" 确认（） request_id = body['actions'][0]['value'] 触发器 ID = 正文['触发器 ID'] # 打开拒绝原因模式客户端.views_open( 触发器id=触发器id，视图={ “类型”：“模态”， "callback_id": f"reject_modal_{request_id}", “标题”：{ “类型”：“纯文本”， "text": "拒绝请求" }, “提交”：{ “类型”：“纯文本”， “文本”：“拒绝” }, “关闭”：{ “类型”：“纯文本”， “文本”：“取消” }, “块”：[ { “类型”：“输入”， "block_id": "reason_block", “元素”：{ “类型”：“纯文本输入”， "action_id": "拒绝原因", “多行”：正确， “占位符”：{ “类型”：“纯文本”， "text": "输入拒绝原因..." } }, “标签”：{ “类型”：“纯文本”， "text": "拒绝原因" } } ], “私有元数据”：json.dumps({ “频道”：主体['频道']['id']， “message_ts”：正文['消息']['ts']， “request_id”：request_id }) } ） @app.view_submission("reject_modal_.*") defhandle_reject_submission（ack，主体，客户端，视图，记录器）： """处理拒绝模式提交""" 确认（） # 从模态中获取值原因=视图['状态']['值']['reason_block']['rejection_reason']['value'] 元数据 = json.loads(view['private_metadata']) 用户 = 正文['用户']['id'] 频道 = 元数据['频道'] message_ts = 元数据['message_ts'] request_id = 元数据['request_id'] # 更新原始消息 client.chat_postMessage( 频道=频道， thread_ts=message_ts, text=f":x: *被<@{user}>拒绝*\n*原因:* {reason}" ） # 更新原来的消息块原始消息 = client.conversations_history( 频道=频道，最新=message_ts，包容=真，极限=1 ）如果原始消息['消息']： Updated_blocks = Original_message['messages'][0].get('blocks', []) Updated_blocks = [b for b in Updated_blocks if b.get('type') != 'actions'] 更新的_blocks.append({ “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": f":x: *被 <@{user}> 拒绝*" } }) 客户端.chat_update( 频道=频道， ts=消息_ts，块=更新的块， text="请求被拒绝" ） logger.info(f"请求 {request_id} 被 {user} 拒绝: {reason}") @app.action("poll_vote") def handle_poll_vote(ack, body, logger): """处理投票选择""" 确认（）所选=正文['动作'][0]['selected_option']['值'] logger.info(f"投票：{selected}") @app.action(“提交投票”) def handle_submit_vote(ack, 正文, 客户端, 响应): """处理投票提交""" 确认（）用户 = 正文['用户']['id'] # 从状态中获取选定的选项 state = body.get('state', {}).get('values', {}) 已选择 = 无对于 state.items() 中的 block_id、block_values：对于 block_values.items() 中的 action_id、action_value： if action_value.get('selected_option'): 所选 = action_value['selected_option'] 如果选择：响应（ text=f"<@{user}> 投票给: {selected['text']['text']}", 响应类型=“in_channel” ）其他：响应（ text="提交前请选择一个选项。", response_type =“短暂” ） ### 4\.模态和视图 # 模态.py # ABOUTME：用于复杂用户输入的模式对话框 # 关于我：具有视图更新的多步骤工作流程从 slack_bolt 导入应用程序导入 json 导入操作系统应用程序=应用程序（令牌= os.environ.get（“SLACK_BOT_TOKEN”）） @app.command(“/创建票”) def open_ticket_modal(ack, body, client): """打开工单创建模式""" 确认（）客户端.views_open( trigger_id=body['trigger_id'], 视图={ “类型”：“模态”， "callback_id": "create_ticket_modal", “标题”：{ “类型”：“纯文本”， "text": "创建票证" }, “提交”：{ “类型”：“纯文本”， “文本”：“创建” }, “关闭”：{ “类型”：“纯文本”， “文本”：“取消” }, “块”：[ { “类型”：“输入”， "block_id": "标题块", “元素”：{ “类型”：“纯文本输入”， "action_id": "标题_输入", “占位符”：{ “类型”：“纯文本”， "text": "问题的简要描述" } }, “标签”：{ “类型”：“纯文本”， “文本”：“标题” } }, { “类型”：“输入”， "block_id": "描述块", “元素”：{ “类型”：“纯文本输入”， "action_id": "描述输入", “多行”：正确， “占位符”：{ “类型”：“纯文本”， "text": "详细描述..." } }, “标签”：{ “类型”：“纯文本”， “文本”：“描述” } }, { “类型”：“输入”， "block_id": "优先级块", “元素”：{ “类型”：“静态选择”， "action_id": "优先级选择", “占位符”：{ “类型”：“纯文本”， "text": "选择优先级" }, “选项”：[ { "text": {"type": "plain_text", "text": "低"}, “价值”：“低” }, { "text": {"type": "plain_text", "text": "Medium"}, “价值”：“中等” }, { "text": {"type": "plain_text", "text": "高"}, “价值”：“高” }, { "text": {"type": "plain_text", "text": "Critical"}, “价值”：“关键” } ] }, “标签”：{ “类型”：“纯文本”， “文本”：“优先级” } }, { “类型”：“输入”， "block_id": "受让人_block", “元素”：{ “类型”：“用户选择”， "action_id": "受让人选择", “占位符”：{ “类型”：“纯文本”， "text": "选择受让人" } }, “标签”：{ “类型”：“纯文本”， “文本”：“受让人” }, “可选”：正确 }, { “类型”：“输入”， "block_id": "due_date_block", “元素”：{ “类型”：“日期选择器”， "action_id": "due_date_picker", “占位符”：{ “类型”：“纯文本”， "text": "选择日期" } }, “标签”：{ “类型”：“纯文本”， "text": "截止日期" }, “可选”：正确 } ] } ） @app.view("create_ticket_modal") def handle_ticket_submission(ack, body, client, view, logger): """处理工单模式提交""" # 提取值值=视图['状态']['值'] 标题 = 值['title_block']['title_input']['value'] 描述 = 值['描述块']['描述输入']['值'] 优先级=值['priority_block']['priority_select']['selected_option']['value'] 受让人=值['assignee_block']['assignee_select'].get('selected_user') due_date = 值['due_date_block']['due_date_picker'].get('selected_date') 用户 = 正文['用户']['id'] # 验证输入错误={} 如果 len(标题) < 5: error['title_block'] = "标题必须至少为 5 个字符" 如果 len(描述) < 10: error['description_block'] = "描述必须至少有 10 个字符" 如果出现错误： ack（response_action =“错误”，错误=错误）返回确认（） # 创建票证（在实际应用中，保存到数据库/API） Ticket_id = f"TICKET-{hash(title) % 10000:04d}" # 在频道中通知块= [ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": f":ticket: 新票已创建", “表情符号”：正确 } }, { “类型”：“部分”， “字段”：[ {"type": "mrkdwn", "text": f"*ID:*\n`{ticket_id}`"}, {"type": "mrkdwn", "text": f"*优先级:*\n{priority.title()}"}, {"type": "mrkdwn", "text": f"*创建者:*\n<@{user}>"}, {"type": "mrkdwn", "text": f"*受让人:*\n{'<@' + 受让人 + '>' if 受让人 else '未分配'}"} ] }, { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": f"*标题:*\n{标题}\n\n*描述:*\n{描述}" } } ] 如果到期日期：块.追加（{ “类型”：“上下文”， “元素”：[ {“type”：“mrkdwn”，“text”：f“：日历：截止日期：{due_date}”} ] }) # 发布到门票频道 client.chat_postMessage( 频道=“#门票”，块=块， text=f"新票：{title}" ） # DM 创建者确认 client.chat_postMessage( 频道=用户， text=f":white_check_mark: 您的票证 `{ticket_id}` 已创建！" ） logger.info(f"由 {user} 创建的票证 {ticket_id}") # 多步骤模式工作流程 @app.command(“/onboard”) def start_onboarding(ack, body, client): """启动多步骤入职工作流程""" 确认（）客户端.views_open( trigger_id=body['trigger_id'], 视图={ “类型”：“模态”， "callback_id": "onboard_step_1", "title": {"type": "plain_text", "text": "入门 (1/3)"}, “提交”：{“类型”：“plain_text”，“文本”：“下一步”}， "关闭": {"type": "plain_text", "text": "取消"}, “块”：[ { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": "欢迎！让我们设置您的个人资料。" } }, { “类型”：“输入”， "block_id": "name_block", “元素”：{ “类型”：“纯文本输入”， “action_id”：“全名” }, "label": {"type": "plain_text", "text": "全名"} }, { “类型”：“输入”， "block_id": "角色块", “元素”：{ “类型”：“纯文本输入”， “action_id”：“角色” }, "label": {"type": "plain_text", "text": "角色/头衔"} } ] } ） @app.view(“onboard_step_1”) def handle_step_1(ack, 主体, 客户端, 视图): """处理步骤 1 并显示步骤 2""" 值=视图['状态']['值'] 名称 = 值['name_block']['full_name']['value'] 角色 = 值['role_block']['role']['value'] # 更新到步骤2 确认（response_action =“更新”，视图= { “类型”：“模态”， "callback_id": "onboard_step_2", "title": {"type": "plain_text", "text": "入门 (2/3)"}, “提交”：{“类型”：“plain_text”，“文本”：“下一步”}， "关闭": {"type": "plain_text", "text": "取消"}, "private_metadata": json.dumps({"name": 名称, "role": 角色}), “块”：[ { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": f"太棒了，*{name}*！现在选择您的团队。" } }, { “类型”：“输入”， "block_id": "team_block", “元素”：{ “类型”：“静态选择”， "action_id": "team_select", “选项”：[ {“text”：{“type”：“plain_text”，“text”：“工程”}，“value”：“工程”}， {"text": {"type": "plain_text", "text": "设计"}, "value": "设计"}, {“文本”：{“类型”：“plain_text”，“文本”：“产品”}，“值”：“产品”}， {“text”：{“type”：“plain_text”，“text”：“操作”}，“value”：“操作”} ] }, “标签”：{“类型”：“plain_text”，“文本”：“团队”} } ] }) @app.view(“onboard_step_2”) def handle_step_2(ack, body, client, view): """处理步骤 2 并显示步骤 3（最终）""" 值=视图['状态']['值'] 前一个 = json.loads(view['private_metadata']) 团队=值['team_block']['team_select']['selected_option']['value'] 上一个['团队'] = 团队确认（response_action =“更新”，视图= { “类型”：“模态”， "callback_id": "onboard_step_3", "title": {"type": "plain_text", "text": "入门 (3/3)"}, "提交": {"type": "plain_text", "text": "完整"}, "关闭": {"type": "plain_text", "text": "取消"}, “private_metadata”：json.dumps（上一个）， “块”：[ { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": "快完成了！还有其他信息吗？" } }, { “类型”：“输入”， "block_id": "bio_block", “元素”：{ “类型”：“纯文本输入”， “action_id”：“生物”， “多行”：正确 }, "label": {"type": "plain_text", "text": "简短的个人简介"}, “可选”：正确 } ] }) @app.view(“onboard_step_3”) def handle_final_step(ack, body, client, view, logger): """完成入职""" 确认（）值=视图['状态']['值'] 前一个 = json.loads(view['private_metadata']) bio = values['bio_block']['bio'].get('value', '未提供生物') 用户 = 正文['用户']['id'] # 完成入职个人资料 = { **上一个， “生物”：生物， “user_id”：用户 } # 宣布新团队成员 client.chat_postMessage( 频道=“#通用”，块=[ { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": f":wave: 欢迎 <@{user}> 加入团队！\n\n*角色:* {profile['role']}\n*团队:* {profile['team'].title()}\n*Bio:* {bio}" } } ], text=f“欢迎{profile['name']}加入团队！” ） logger.info(f"{user} 的入职完成：{profile}") ### 5\.斜线命令 # 命令.py # 关于：斜杠命令实现 # 关于我：用于团队工作流程的各种实用命令从 slack_bolt 导入应用程序从日期时间导入日期时间，时间增量随机导入导入操作系统应用程序=应用程序（令牌= os.environ.get（“SLACK_BOT_TOKEN”）） @app.command(“/standup”) def handle_standup(ack, 主体, 客户端, 命令): """启动一个独立线程""" 确认（）频道 = 命令['channel_id'] 用户 = 命令['user_id'] # 创建独立线程结果 = client.chat_postMessage( 频道=频道，块=[ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": f":sunrise: 每日站立 - {datetime.now().strftime('%A, %B %d')}", “表情符号”：正确 } }, { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": "请在此线程中分享您的更新：\n\n1. :white_check_mark: 您昨天完成了什么？\n2. :calendar: 您今天在做什么？\n3. :construction: 有任何阻碍吗？" } }, { “类型”：“分隔符” }, { “类型”：“上下文”， “元素”：[ {"type": "mrkdwn", "text": f"由 <@{user}> 启动"} ] } ], text="每日站立" ） # 固定站立 client.pins_add(通道=通道, 时间戳=结果['ts']) @app.command(“/民意调查”) def handle_poll(ack, 主体, 客户端, 命令): """创建快速投票：/poll "问题" "选项 1" "选项 2" ...""" 确认（）文本 = command.get('文本', '') # 解析引用的参数进口再 parts = re.findall(r'"([^"]+)"', 文本) 如果 len(部分) < 3: client.chat_postEphemeral( 频道=命令['channel_id'], 用户=命令['user_id'], text='用法：/poll "问题" "选项 1" "选项 2" "选项 3"' ）返回问题=部分[0] 选项=部分[1:] # 创建轮询块选项块= [] emojis = [':一:', ':二:', ':三:', ':四:', ':五:', ':六:', ':七:', ':八:', ':九:'] 对于 i，枚举中的选项（选项[:9]）： option_blocks.append({ “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": f"{emojis[i]} {选项}" } }) 块= [ { “类型”：“标题”， “文本”：{“类型”：“plain_text”，“文本”：“：bar_chart：民意调查”，“表情符号”：True} }, { “类型”：“部分”， “文本”：{“类型”：“mrkdwn”，“文本”：f“* {问题} *”} }, {“类型”：“分隔符”}， *选项_块， {“类型”：“分隔符”}， { “类型”：“上下文”， “元素”：[ {"type": "mrkdwn", "text": f"由 <@{command['user_id']}> 进行投票 |反应投票！”} ] } ] 结果 = client.chat_postMessage( 频道=命令['channel_id'], 块=块， text=f"投票：{问题}" ） # 添加反应选项对于范围内的 i(len(选项[:9]))： emoji_names = ['一', '二', '三', '四', '五', '六', '七', '八', '九'] 客户端.reactions_add( 频道=命令['channel_id'], 时间戳=结果['ts'], 名称=表情符号名称[i] ） @app.command("/提醒团队") def handle_team_reminder(ack, body, client, command): """设置团队提醒：/remind-team 15m 检查部署状态""" 确认（）文本 = command.get('文本', '').strip() 零件 = text.split(' ', 1) 如果 len(部分) < 2: client.chat_postEphemeral( 频道=命令['channel_id'], 用户=命令['user_id'], text='用法：/remind-team 15m 您的提醒消息' ）返回 time_str = 零件[0] 消息=部分[1] # 解析时间 time_map = {'s': 1, 'm': 60, 'h': 3600} 单位=时间_str[-1] 如果单位不在 time_map 中： client.chat_postEphemeral( 频道=命令['channel_id'], 用户=命令['user_id'], text='时间格式：15s、15m 或 2h' ）返回尝试：金额 = int(time_str[:-1]) 秒 = 数量 * time_map[单位] 除了值错误： client.chat_postEphemeral( 频道=命令['channel_id'], 用户=命令['user_id'], text='时间格式无效' ）返回 # 安排消息 post_at = int(datetime.now().timestamp()) + 秒 client.chat_scheduleMessage( 频道=命令['channel_id'], post_at=post_at, text=f":bell: *提醒:* {消息}\n\n_由<@{command['user_id']}>_设置" ） client.chat_postEphemeral( 频道=命令['channel_id'], 用户=命令['user_id'], text=f"从现在开始安排在 {time_str} 进行提醒！" ） @app.command(“/随机选择”) def handle_random_pick(ack, body, client, command): """从选项中随机选择：/random-pick option1 option2 option3""" 确认（）文本 = command.get('文本', '').strip() 如果不是文字： client.chat_postEphemeral( 频道=命令['channel_id'], 用户=命令['user_id'], text='用法：/random-pick option1 option2 option3' ）返回选项 = text.split() 挑选=随机.选择（选项） client.chat_postMessage( 频道=命令['channel_id'], 块=[ { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， "text": f":game_die: <@{command['user_id']}> 让我从：{', '.join(options)}" 中随机选择 } }, { “类型”：“部分”， “文本”：{ “类型”：“mrkdwn”， “文本”：f“：point_right：* {picked} *” } } ], text=f"随机选择：{picked}" ） ### 6\. Webhook 和传入消息 # webhooks.py # 关于我：外部服务的传入 webhook 集成 # 关于我：CI/CD 通知、警报和外部触发器导入请求导入 json 从输入 import 可选、列表、字典导入hmac 导入哈希库导入时间 SlackWebhook 类： """Slack 的传入 Webhook 客户端""" def __init__(self, webhook_url: str): self.webhook_url = webhook_url 定义发送（自我, 文本：str，块：可选[List[Dict]] = None，附件：可选[List[Dict]] = None， thread_ts：可选[str] =无， unfurl_links: 布尔 = True, unfurl_media: 布尔 = True ) -> 字典： """通过 webhook 发送消息""" 有效负载={ “文本”：文本， “展开链接”：展开链接， “展开媒体”：展开媒体 } 如果块：有效负载[“块”] = 块如果是附件：有效负载[“附件”] = 附件如果线程_ts：有效负载[“thread_ts”] = thread_ts 响应 = requests.post( self.webhook_url, json=有效负载， headers={"Content-Type": "application/json"} ）响应.raise_for_status() 返回 {“ok”：True，“status”：response.status_code} def send_deployment_notification( 自我, 应用程序名称：str，环境：str，版本：str，状态：str， commit_sha：str，作者：str, url: 可选[str] = 无）： """发送部署通知""" 颜色图 = { “成功”：“#36a64f”， “失败”：“#ff0000”， “开始”：“#ffcc00”， “待处理”：“#808080” } 状态表情符号= { “成功”：“：white_check_mark：”， “失败”：“：x：”， “开始”：“：火箭：”， “待处理”：“：沙漏：” } 块= [ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": f"{status_emoji.get(status, ':grey_question:')} 部署 {status.title()}: {app_name}", “表情符号”：正确 } }, { “类型”：“部分”， “字段”：[ {"type": "mrkdwn", "text": f"*环境:*\n{环境}"}, {"type": "mrkdwn", "text": f"*版本:*\n{版本}"}, {"type": "mrkdwn", "text": f"*提交:*\n`{commit_sha[:8]}`"}, {"type": "mrkdwn", "text": f"*作者:*\n{author}"} ] } ] 如果网址：块.追加（{ “类型”：“动作”， “元素”：[ { “类型”：“按钮”， "text": {"type": "plain_text", "text": "查看部署"}, “网址”：网址， "style": "primary" 如果 status == "success" 否则无 } ] }) 附件=[ { "颜色": color_map.get(status, "#808080"), “块”：块 } ] 返回 self.send( text=f"部署 {status}：{app_name} 到 {environment}", 附件=附件） def 发送警报( 自我, 标题：str，消息：str，严重性：str =“警告”，来源：str =“系统”，详细信息：可选[Dict] = None ）： """发送警报通知""" 严重性配置 = { “关键”：{“表情符号”：“：rotating_light：”，“颜色”：“#ff0000”}， “错误”：{“表情符号”：“：x：”，“颜色”：“#ff4444”}， "警告": {"表情符号": ":警告:", "颜色": "#ffcc00"}, “信息”：{“表情符号”：“：information_source：”，“颜色”：“#0088ff”} } config=severity_config.get（严重性，severity_config [“信息”]）块= [ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": f"{config['emoji']} {title}", “表情符号”：正确 } }, { “类型”：“部分”， “文本”：{“类型”：“mrkdwn”，“文本”：消息} }, { “类型”：“上下文”， “元素”：[ {"type": "mrkdwn", "text": f"*来源:* {source} | *严重性:* {severity.upper()}"} ] } ] 如果详细： detail_text = "\n".join(f"*{k}:* {v}" for k, v indetails.items()) 块.插入(2, { “类型”：“部分”， “文本”：{“类型”：“mrkdwn”，“文本”：detail_text} }) 返回 self.send( text=f"[{severity.upper()}] {标题}", 附件=[{“颜色”：配置[“颜色”]，“块”：块}] ） # Webhook签名验证 def verify_slack_signature( 签名秘密：str，请求正文：str，时间戳：str，签名：str ) -> 布尔值： """验证 Slack 请求签名""" # 检查时间戳以防止重放攻击如果abs(time.time() - int(时间戳)) > 60 * 5: 返回错误 sig_basestring = f"v0:{时间戳}:{request_body}" 计算签名 = 'v0=' + hmac.new( 签名秘密.encode(), sig_basestring.encode(), 哈希库.sha256 ).hexdigest() 返回hmac.compare_digest（计算签名，签名） # 使用示例如果 __name__ == "__main__": webhook = SlackWebhook("https://hooks.slack.com/services/T00/B00/XXX") # 发送部署通知 webhook.send_deployment_notification（ app_name="我的服务", 环境=“生产”，版本=“v1.2.3”，状态=“成功”， commit_sha=“abc123def”，作者=“developer@example.com”， url =“https://deployments.example.com/123” ） # 发送警报 webhook.send_alert( title="CPU 使用率高", message="过去 5 分钟服务器 cpu 使用率已超过 90%。", 严重性=“警告”，来源=“监控”，详情={ “服务器”：“prod-web-01”， “当前使用情况”：“92%”， “阈值”：“90%” } ） ## 集成示例 ### GitHub 操作集成 # .github/workflows/slack-notify.yml 名称：Slack 通知于：推：分支：[主要] 拉请求：类型：[打开、关闭、合并] 工作流程_运行：工作流程：[“CI”] 类型：[已完成] 职位：通知部署：运行：ubuntu-latest if: github.event_name == 'push' && github.ref == 'refs/heads/main' 步骤： - 名称：通知 Slack 使用：slackapi/slack-github-action@v1 与：有效负载： | { “块”：[ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": ":rocket: 部署已开始" } }, { “类型”：“部分”， “字段”：[ {"type": "mrkdwn", "text": "*存储库:*\n${{ github.repository }}"}, {"type": "mrkdwn", "text": "*分支:*\n${{ github.ref_name }}"}, {"type": "mrkdwn", "text": "*提交:*\n`${{ github.sha }}`"}, {"type": "mrkdwn", "text": "*作者:*\n${{ github.actor }}"} ] }, { “类型”：“动作”， “元素”：[ { “类型”：“按钮”， "text": {"type": "plain_text", "text": "查看提交"}, "url": "${{ github.event.head_commit.url }}" } ] } ] } 环境： SLACK_WEBHOOK_URL：${{secrets.SLACK_WEBHOOK_URL }} SLACK_WEBHOOK_TYPE：INCOMING_WEBHOOK ### FastAPI 集成 # api_integration.py # 关于：用于 Slack 事件处理的 FastAPI 集成 # 关于我：Slack Events API 的 Webhook 端点从 fastapi 导入 FastAPI、请求、HTTPException 从 slack_bolt 导入应用程序从 slack_bolt.adapter.fastapi 导入 SlackRequestHandler 导入操作系统 # 初始化 Slack 应用程序 slack_app = 应用程序（ token=os.environ.get("SLACK_BOT_TOKEN"), signing_secret=os.environ.get("SLACK_SIGNING_SECRET") ） # 注册事件处理程序 @slack_app.event(“消息”) def handler_message(事件，比如说)：如果 event.get("text", "").lower() 中的“hello”：说（f“嗨 <@{event['user']}>！”） @slack_app.command("/api-status") def handle_status(ack, 响应): 确认（）响应（“API 是健康的！”） # FastAPI 设置应用程序= FastAPI（标题=“松弛机器人API”）处理程序 = SlackRequestHandler(slack_app) @app.post(“/slack/events”) 异步 def slack_events(请求: 请求): """处理 Slack 事件""" 返回等待处理程序.handle(请求) @app.post(“/slack/交互”) 异步 def slack_interactions(请求: 请求): """处理 Slack 交互组件""" 返回等待处理程序.handle(请求) @app.get(“/健康”) 异步 def health(): """健康检查端点""" 返回{“状态”：“健康”} ## 最佳实践 ### 1\.速率限制 # 速率限制处理导入时间从 functools 导入包装 defrate_limit_handler(max_retries=3, base_delay=1): """用于处理 Slack 速率限制的装饰器""" def 装饰器（函数）： @wraps(函数) def 包装器（*args，**kwargs）：对于范围内的尝试（max_retries）：尝试：返回 func(*args, **kwargs) 除了异常 e：如果 str(e) 中的“rate_limited”：延迟 = base_delay * (2 ** 尝试) 时间.睡眠（延迟）其他：提高引发异常（“超出最大重试次数”）返回包装器返回装饰器 @rate_limit_handler(最大重试次数=3) def send_message(客户端、通道、文本): 返回 client.chat_postMessage(channel=channel, text=text) ### 2\.错误处理 # 全面的错误处理从 slack_sdk.errors 导入 SlackApiError def safe_send_message(客户端、通道、文本、块=无): """发送带有错误处理的消息""" 尝试：结果 = client.chat_postMessage( 频道=频道，文本=文本，块=块）返回结果除了 SlackApiError 为 e： error_code = e.response.get("错误", "未知错误") 如果错误代码==“channel_not_found”： # 处理丢失的通道 raise ValueError(f"未找到频道 {channel}") elif error_code ==“not_in_channel”： # 先尝试加入频道 client.conversations_join(频道=频道) 返回 client.chat_postMessage(通道=通道，文本=文本，块=块) elif error_code ==“速率限制”： # 等待并重试 retry_after = int(e.response.headers.get("重试后", 1)) 时间.睡眠（retry_after）返回 safe_send_message(客户端、通道、文本、块) 其他：提高 ### 3\.消息格式 # 安全的消息格式 def escape_text(text: str) -> str: """转义 Slack 的特殊字符""" 文本 = 文本.replace("&", "&") 文本 = 文本.replace("<", "<") 文本 = 文本.replace(">", ">") 返回文本 def format_user_mention(user_id: str) -> str: """格式化用户提及""" 返回 f"<@{user_id}>" def format_channel_link(channel_id: str) -> str: """格式化频道链接""" 返回 f"<#{channel_id}>" def format_url(url: str, text: str = None) -> str: """使用可选文本设置 URL 格式""" 如果文本：返回 f"<{url}|{escape_text(text)}>" 返回 f"<{url}>" def format_code_block(代码: str, 语言: str = "") -> str: """格式化代码块""" return f"```{语言}\n{代码}\n```" ### 4\.令牌安全 # 安全令牌管理导入操作系统从 functools 导入 lru_cache @lru_cache() def get_slack_client(): """使用安全令牌获取缓存的 Slack 客户端""" 从 slack_sdk 导入 WebClient 令牌 = os.environ.get("SLACK_BOT_TOKEN") 如果不是令牌：引发 ValueError("SLACK_BOT_TOKEN 未设置") 如果不是 token.startswith("xoxb-"): 引发 ValueError("无效的机器人令牌格式") 返回WebClient（令牌=令牌） # 从不记录令牌导入日志记录类 TokenFilter(logging.Filter): def 过滤器（自身，记录）： if hasattr(记录, 'msg'): 记录.msg = str(记录.msg).替换( os.environ.get("SLACK_BOT_TOKEN", ""), "[已编辑]" ）返回真 ## 故障排除 ### 常见问题 **问题：机器人不响应消息** # 验证机器人具有正确的范围 # 检查事件订阅是否启用 # 确保请求 URL 已验证 # 使用日志记录进行调试导入日志记录日志记录.basicConfig（级别=日志记录.DEBUG） @app.event(“消息”) def debug_messages（正文，记录器）： logger.info(f"收到消息事件：{body}") **问题：交互式组件不起作用** # 确保交互组件 URL 已设置 # 检查action_id是否与处理程序匹配 @app.action("button_click") # 必须匹配块中的action_id def handle_click(ack, body, logger): 确认（） logger.info(f"点击按钮：{body}") **问题：套接字模式连接丢失** # 增加连接超时时间从 slack_bolt.adapter.socket_mode 导入 SocketModeHandler 处理程序 = SocketModeHandler( 应用程序，应用程序令牌， ping_interval=30 # 每 30 秒发送一次 ping ） **问题：消息未出现在频道中** # 检查通道ID格式 # 验证机器人是否在频道中 def Ensure_in_channel(客户端, 通道): 尝试： client.conversations_info(频道=频道) 除外： client.conversations_join(频道=频道) ### 调试命令 # 测试网络钩子卷曲-X POST \ -H“内容类型：application/json”\ -d '{"text": "测试消息"}' \ $SLACK_WEBHOOK_URL # 测试机器人令牌卷曲-X POST \ -H“授权：持有者$SLACK_BOT_TOKEN”\ https://slack.com/api/auth.test # 列出频道卷曲-X GET \ -H“授权：持有者$SLACK_BOT_TOKEN”\ “https://slack.com/api/conversations.list?limit=10” ## 版本历史版本 |日期 |变化 ---|---|--- 1.0.0 | 2026-01-17 |具有全面 Slack API 模式的初始版本 ## 资源 * [Slack API 文档]() * [Python 版 Bolt]() * [块套件生成器]() * [Slack 应用清单]() * [套接字模式]() * [事件 API]() ** ** _此技能为 Slack 机器人开发提供生产就绪模式，从而实现强大的团队自动化和交互式工作流程。_

松弛 API

包含技能