Agno AgentOS 接入 Slack：从应用配置、流式体验到多机器人协作与人工介入的完整指南

Agno AgentOS 接入 Slack从应用配置、流式体验到多机器人协作与人工介入的完整指南【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南聚焦 agno 开源仓库中cookbook/05_agent_os/17_slack目录的 Slack 集成示例系统讲解如何通过 AgentOS 的Slack接口把一个Agent、Team或Workflow挂载到 Slack使其能响应私信与频道 提及。读者学完后将掌握Slack App 从创建到上线的完整配置流程、消息/线程/用户身份解析机制、流式回复与任务卡片等体验增强、工作区搜索与文件工具、多机器人挂载与对等 App 协作以及基于 Slack 交互式卡片的四类 Human-in-the-Loop人工介入流程。文中所有运行示例与配置参数均以当前仓库为事实依据示例源码位于 cookbook/05_agent_os/17_slackSlack 接口的底层实现位于 libs/agno/agno/os/interfaces/slack。这个示例目录教你什么README.md 开篇就点明了 Slack 接口的能力边界它通过带签名的事件events与交互interactionsWebhook将 Agent、Team 或 Workflow 连接到 Slack支持流式回复、plan 模式工具卡片、建议提示词、文件收发、工作区搜索、按线程隔离的会话per-thread sessions以及人工介入表单。目录内共 12 个示例每个示例的职责划分如下文件教学重点basic.py单常驻 AgentDM 与频道 提及过滤的区别streaming_ux.py建议提示词、loading 提示、plan 模式任务卡片slack_tools.py频道历史、线程、工作区搜索与文件传输user_memory.py基于已解析 Slack 身份的跨线程用户记忆team.py带工作区搜索能力的专家支持 Teamworkflow.py“先研究后写作”的顺序 Workflowmultiple_bots.py一个 AgentOS 挂载两个独立凭据的 Slack Apppeer_agents.py对另一个 Slack App 的安全、非对称响应hitl_confirmation.py破坏性工具调用前的人工确认hitl_user_input.py在 Slack 中收集结构化用户输入hitl_external_execution.py展示工具参数并收集外部执行结果hitl_incident_commander.py组合全部暂停类型的复合故障响应流程版本要求README 明确说明若要使用 plan 模式任务卡片进行流式回复需要slack_sdk 3.40.0演示环境demo会安装这些受支持的 Slack 依赖。从源码看若未安装 Slack 相关依赖接口模块在导入时会直接抛出提示router.py 在ImportError时提示执行pip install agno[slack]。创建一个可用的 Slack App八步配置清单README 用八个步骤完整覆盖了 Slack App 的生命周期。以下逐步展开并补充源码层面的注意事项。1. 创建 App打开 https://api.slack.com/apps点击Create New App选择From scratch命名并选择目标 workspace在Basic Information页面复制Signing Secret。该 Secret 正是服务端校验 Webhook 请求签名的密钥——security.py 中的verify_slack_signature负责完成验签配置错误时会出现 README 故障排查表中“Webhook returns 403”的症状。2. 启用 Agents AI Apps这是流式回复与工作区搜索的前提侧边栏进入Agents AI Apps将Agent or Assistant开关设为On在Suggested Prompts下选择Dynamic点击Save。启用后 Slack 会自动添加assistant:write权限范围。3. 添加 OAuth Scopes在OAuth Permissions Bot Token Scopes中添加你打算运行示例所需的权限范围。README 给出的完整对照表如下Scope用途app_mentions:read接收频道内的 提及assistant:write流式回复、设置状态、提供动态建议提示词chat:write发送回复im:history接收并读取私信历史channels:read、groups:read解析公开/私密频道的元数据channels:history、groups:history读取公开/私密频道历史files:read、files:write下载收到的文件与上传结果文件users:read解析 Slack 用户users:read.email为user_memory.py解析稳定的邮件身份search:read.public搜索公开工作区消息search:read.files在工作区搜索中包含文件search:read.users在工作区搜索中解析人物其中常见的流式回复配置组合是app_mentions:read、assistant:write、chat:write、im:history。每个 Python 示例文件的 docstring 头部都有一行Slack scopes:明确列出该示例所需的精确 scope 集合例如 slack_tools.py 列出的是包含文件与搜索在内的 13 项完整集合而 basic.py 只有 4 项。以哪个示例为准就按哪个文件头部的列表来配。配置完成后把 App 安装Install到 workspace并复制Bot User OAuth Tokenxoxb-...。修改 scope 之后必须重新安装ReinstallApp否则新权限不生效。4. 订阅事件在Event Subscriptions中启用事件将Request URL设置为https://YOUR-PUBLIC-HOST/slack/events订阅 App 需要的 bot 事件Event用途app_mention接收频道 提及message.im接收私信message.channels接收公开频道消息包括对等 Apppeer app的消息message.groups接收私密频道消息assistant_thread_started设置建议提示词并接收工作区搜索所需的 action tokenassistant_thread_context_changed刷新 Assistant 线程上下文保存更改并重新安装 App。值得强调的是当你在 Slack 后台配置这个端点时Slack 会立即发送一次签名 URL 验证请求URL verification challenge因此AgentOS 服务必须先通过公网可达否则端点无法通过验证。这一点也是第 7 步“开隧道”的原因。5. 启用 Interactivity四个hitl_*.py示例必需在Interactivity Shortcuts中启用交互将Request URL设置为https://YOUR-PUBLIC-HOST/slack/interactions保存更改。README 特别提醒没有这个端点审批按钮与被提交的输入就无法恢复一个被暂停的 run。可以理解为事件端点负责“接收任务”交互端点负责“回收人工结论”。6. 设置环境变量export SLACK_TOKENxoxb-... export SLACK_SIGNING_SECRET... export OPENAI_API_KEYsk-...SLACK_TOKEN即 Bot User OAuth TokenSLACK_SIGNING_SECRET即 Basic Information 页面里的 Signing Secret。注意从源码看Slack接口在启动时会执行一次auth_test()来获取自身 bot 身份用于过滤自己发出的消息、避免回声循环见 router.py。失败时仅降级own_bot_id为空不会直接崩溃但真正收到事件前需要凭据有效。7. 开启公网隧道Slack 需要一个公网 HTTPS URL。本地开发可用如下任一方式ngrok http 7777 # 或 cloudflared tunnel --url http://localhost:7777把得到的公网 URL 分别填入“事件”与“交互”两个 Request URL。如果隧道主机名变化两处设置都必须同步更新。8. 运行一个示例.venvs/demo/bin/python cookbook/05_agent_os/17_slack/basic.py然后给 App 发私信或在频道里 它即可。以basic.py为例尝试方式是在一个线程里先说 “My project is Cedar”再问 “what the project is”用来验证历史会话是否跟随线程持久化。路由与接口挂载一个默认 Slack 接口长什么样默认情况下一个 Slack 接口会挂载以下路由方法路由用途POST/slack/eventsURL 验证与接收 Slack 事件POST/slack/interactionsHITL 按钮与表单提交此外 AgentOS 还暴露GET /health与GET /config。Slack 接口没有独立的 status 路由。自定义前缀custom prefix可以替换默认的/slack在多机器人场景中每个 App 必须把事件与交互 URL 指向各自的独立前缀例如/research/events、/analyst/events。这一设计对应源码中通过闭包隔离每个实例配置的做法见 router.py每个实体会用实体名生成唯一的op_suffix避免多个 Slack 实例挂载在同一 FastAPI 应用上时 operation_id 冲突。来看最简示例 basic.py 的完整骨架from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.openai import OpenAIResponses from agno.os import AgentOS from agno.os.interfaces.slack import Slack db SqliteDb( idslack-basic-db, db_filetmp/slack_basic.db, ) assistant Agent( idslack-assistant, nameSlack Assistant, modelOpenAIResponses(idgpt-5.5), dbdb, instructions[ You are a helpful assistant in Slack., Keep answers concise and easy to scan., ], add_history_to_contextTrue, num_history_runs3, markdownTrue, ) agent_os AgentOS( idslack-basic-os, descriptionAgentOS serving one persistent Slack assistant., agents[assistant], interfaces[ Slack( agentassistant, reply_to_mentions_onlyTrue, ) ], ) app agent_os.get_app() if __name__ __main__: agent_os.serve(appapp)这段代码揭示了几个关键模式Agent 通过SqliteDb持久化会话Slack接口既可以挂 Agent也可以像 team.py 那样挂Team、像 workflow.py 那样挂Workflow三选一router.py中据此判断entity_type来分发事件最后调用agent_os.get_app()拿到 FastAPI 应用。从 router.py 可以看到Slack接口承载的默认参数与限制整理为下表参数默认值说明reply_to_mentions_onlyTrue频道内是否只响应 提及streamingTrue是否开启流式回复task_display_modeplan工具活动如何展示plan 卡片loading_textThinking...运行期间 Assistant 状态文本loading_messagesNone可轮换的 loading 消息列表suggested_promptsNone新 Assistant 线程的建议提示词token/user_token/signing_secretNoneApp 凭据多机器人时显式传入resolve_user_identityFalse是否通过users.info解析稳定用户身份respond_to_other_appsFalse是否接收对等 App 的消息buffer_size100流式缓冲大小max_file_size1_073_741_8241GB文件传输上限markdown/unfurl_links/unfurl_mediaTrue消息呈现与链接展开选项消息、线程与会话用户身份消息过滤的两条路径当reply_to_mentions_onlyTrue时接口在频道中只处理app_mention事件并抑制普通非提及的频道消息私信仍会被回答。当该开关关闭时普通频道消息也会被处理而重复的app_mention事件则会被忽略。这一逻辑在 helpers.py 的should_respond函数中实现若开启该标志且事件类型为message且不是私信则返回 False不响应若关闭标志且事件类型为app_mention且不是私信同样返回 False防止同一条消息同时以 message 与 app_mention 形式到达时重复响应。线程即会话每一个 Slack 线程就是一个 AgentOS session。当前会话键的格式为{entity_id}:{channel_id}:{thread_ts}顶层消息的时间戳ts即线程起点回复复用父消息的thread_ts频道 ID 用于防止不同频道间时间戳碰撞解析器会先检查旧的{entity_id}:{thread_ts}形式保证升级后旧历史不被孤儿化。由于会话键不包含用户 ID在同一个频道线程里回复的所有人共享该会话上下文这也解释了为什么user_memory.py要单独解决“跨线程记住个人”的问题。用户身份member ID 还是 email默认情况下run 的user_id是 Slack member ID。当设置resolve_user_identityTrue时接口会调用users.info当 Slack 返回成员邮箱时用邮箱作为稳定 ID把成员显示名display name加入 run 元数据查询失败时回退到 Slack ID。user_memory.py 正是依赖此开关让记忆能跨线程跟随同一人因此它要求users:read与users:read.email两个 scope。事件处理器侧的相关分支见 event_handler.pyif self.resolve_user_identity:触发users.info调用。流式 UXloading、建议提示词与 plan 模式任务卡片Slack 流式回复默认开启。接口会打开chat_stream随 run 进展追加文本并把工具活动渲染成任务卡片。streaming_ux.py 把这些控件的用法显式化researcher Agent( idslack-streaming-researcher, nameSlack Streaming Researcher, modelOpenAIResponses(idgpt-5.5), dbdb, tools[WebSearchTools()], instructions[ Research current questions with web search., Use more than one source when the answer benefits from comparison., Return a concise synthesis with source links., ], add_history_to_contextTrue, num_history_runs3, markdownTrue, ) agent_os AgentOS( idslack-streaming-os, descriptionAgentOS demonstrating Slack streaming presentation controls., agents[researcher], interfaces[ Slack( agentresearcher, streamingTrue, task_display_modeplan, loading_textResearching..., loading_messages[ Searching current sources..., Comparing the evidence..., Preparing a concise answer..., ], suggested_prompts[ { title: Technology brief, message: Summarize todays most important AI infrastructure news., }, { title: Compare approaches, message: Compare two current approaches to Python dependency management., }, ], ) ], )各控件的作用loading_text与loading_messages在运行期间更新 Assistant 状态栏loading_messages支持多条轮换展示suggested_prompts填充新 Assistant 线程的动态建议提示词每条由titlemessage组成task_display_modeplan把工具调用过程以 plan 任务卡片的形式展示给用户避免“黑盒”等待。要完整呈现这套体验需要在 Slack 侧开启Agents AI Apps、订阅assistant_thread_started事件并保持slack_sdk处于较新版本README 明确要求 3.40.0才有 plan 模式任务卡片。运行该示例时可尝试问 “What changed in Python packaging this year? Cite sources.” 来观察网络搜索工具被渲染成卡片的效果。SlackTools频道历史、线程展开、工作区搜索与文件传输slack_tools.py 在一个 Agent 上组合了频道历史、线程展开、工作区搜索与文件下载/上传四类能力并通过SlackTools的布尔开关精确控制暴露哪些工具from agno.tools.slack import SlackTools workspace_tools SlackTools( output_directorytmp/slack_downloads, enable_send_messageFalse, enable_send_message_threadFalse, enable_list_channelsTrue, enable_get_channel_historyTrue, enable_upload_fileTrue, enable_download_fileTrue, enable_search_workspaceTrue, enable_get_threadTrue, enable_list_usersFalse, enable_get_user_infoFalse, enable_get_channel_infoTrue, )对应 Agent 的指令则把 Slack 定位成“工作区唯一事实来源”主题类问题走search_workspaceaction token 来自 Slack 事件已知频道走get_channel_history重要回复用get_thread展开需要内容分析时下载共享文件仅在用户明确要求时才上传结果文件。该示例需要 README 表格中几乎全部 13 项 scope。工作区搜索用的不是旧式搜索 APISlackTools.search_workspace走的是 Slack 的assistant.search.contextaction而非传统的 message-search API。工作机制如下Slack 在 Assistant 线程事件上提供一个短时有效的action_token接口把它放入 run 元数据工具包在调用时从 run 元数据中读取该 token。由此得出三点使用约束README 原文要点必须在 Slack Assistant 线程中调用它而不能在控制台 run 里调用需要授予search:read.public、search:read.files、search:read.users本课示例不需要SLACK_USER_TOKEN用户级 token搜索可见性遵循 Slack 用户与工作区的既有权限。team.py 把同样的搜索能力交给了 Team 中的一名专家一个“技术专家”负责诊断代码/API/基础设施问题携带 WebSearchTools另一个“文档专家”负责检索工作区历史讨论仅携带只开了enable_search_workspaceTrue的SlackTools。支持 Team 则按需把任务分派给对应专家并在回答中附带工具返回的证据链接。从源码看当挂载对象是 Team 时接口会自动把team.store_member_responses置为 Truerouter.py否则continue_run无法可靠地从数据库重载成员的工具状态。Workflow 也能被 Slack 唤起workflow.py 演示了一个“先研究、后写作”的两步顺序 Workflow 挂在 Slack 线程里运行并用 SQLite 持久化 workflow 历史content_workflow Workflow( idslack-content-workflow, nameSlack Content Workflow, descriptionResearch a topic, then write a concise brief., dbdb, steps[ Step(nameResearch, agentresearcher), Step(nameWrite, agentwriter), ], add_workflow_history_to_stepsTrue, num_history_runs3, ) agent_os AgentOS( idslack-workflow-os, descriptionAgentOS serving a sequential content Workflow through Slack., workflows[content_workflow], interfaces[Slack(workflowcontent_workflow)], )尝试提问示例Research passkeys for SaaS apps and write a short adoption brief.它会先让 Research Agent 带回带来源链接的调研结论再由 Writer Agent 产出 Slack 友好的简报。多机器人挂载一个 AgentOS、两个 Slack AppSlack 会把每个 App 的事件发送到一个已配置的 URL。multiple_bots.py 演示了在同一个服务器上挂载两个拥有独立 token、签名密钥与前缀的 Slack Appexport RESEARCH_SLACK_TOKENxoxb-... export RESEARCH_SLACK_SIGNING_SECRET... export ANALYST_SLACK_TOKENxoxb-... export ANALYST_SLACK_SIGNING_SECRET...对应两个 App 在 Slack 后台的事件/交互 URL 要分别指向自己的前缀AppEvents URLInteractions URLResearch/research/events/research/interactionsAnalyst/analyst/events/analyst/interactions代码层面两个Slack实例各自显式传入prefix、token、signing_secretinterfaces[ Slack( agentresearcher, prefix/research, tokenresearch_token, signing_secretresearch_secret, streamingTrue, ), Slack( agentanalyst, prefix/analyst, tokenanalyst_token, signing_secretanalyst_secret, streamingTrue, ), ],由于会话键携带entity_id即使两个 App 在同一个 workspace 里它们的会话也互不干扰。该文件用required_env辅助函数在读不到环境变量时直接抛错避免凭据缺省导致的隐性问题。对等 App安全、非对称的机器人间委托默认情况下机器人bot自己发出的消息会被丢弃。设置respond_to_other_appsTrue才会让某个接口接收来自对等 App 的消息即便如此来自该接口自身 bot 身份的消息仍会被丢弃防止自我回声。对应过滤逻辑在 event_handler.pyif is_bot and not self.respond_to_other_apps:直接跳过。peer_agents.py 刻意采用非对称拓扑协调者coordinator保持respond_to_other_appsFalse只听取真人研究员researcher设置为True可以听到协调者。这样既实现了单向委托又不会形成自动“乒乓”循环。README 警告除非你的应用另有显式循环保护否则不要把两个 App 对称地都开启 peer 响应。该示例需要的环境变量export COORDINATOR_SLACK_TOKENxoxb-... export COORDINATOR_SLACK_SIGNING_SECRET... export RESEARCHER_SLACK_TOKENxoxb-... export RESEARCHER_SLACK_SIGNING_SECRET... export RESEARCHER_SLACK_USER_IDU...最后一项RESEARCHER_SLACK_USER_ID是必须的协调者需要真实的研究员用户 ID才能发出USER_ID这种真实的 Slack 提及从而把研究任务投递给研究员 App。协调者的 SlackTools 开启了send_message_thread会在当前 Slack 线程内 研究员并附上一段自包含的研究请求研究员完成任务后在人类可读的线程里直接回答不再 协调者避免触发循环。URL 前缀配置协调者指向/coordinator/events、/coordinator/interactions研究员指向/researcher/events、/researcher/interactions。跨线程用户记忆让记忆跟着人走user_memory.py 解决“用户在 A 线程说过的偏好B 线程也要知道”的问题。它把第 4 节的身份解析与 AgentOS 的 MemoryManager 组合起来from agno.memory import MemoryManager memory_manager MemoryManager( idslack-user-memory-manager, modelOpenAIResponses(idgpt-5.5), dbdb, memory_capture_instructions( Capture the users name, role, communication preferences, current projects, and durable likes or dislikes. ), ) personal_assistant Agent( idslack-personal-assistant, nameSlack Personal Assistant, modelOpenAIResponses(idgpt-5.5), dbdb, memory_managermemory_manager, update_memory_on_runTrue, ... ) agent_os AgentOS( idslack-memory-os, descriptionAgentOS with email-resolved Slack user memory., agents[personal_assistant], interfaces[ Slack( agentpersonal_assistant, resolve_user_identityTrue, ) ], )关键联动点在于默认会话键按线程隔离user_id是 Slack member ID而 member ID 未必稳定地代表同一个自然人开启resolve_user_identityTrue后接口尽量用邮箱做稳定 ID见第 4 节记忆才能跨线程正确归人。运行该示例需要users:read与users:read.email两个额外 scope。Human-in-the-LoopSlack 里的四类暂停当 run 因等待人工而暂停时Slack 接口会把被暂停的工具需求渲染成交互式卡片并通过/slack/interactions恢复持久化的 run。这一课保持每个暂停类型一个聚焦示例另加一个复合的故障响应流程示例暂停类型交互内容hitl_confirmation.py确认破坏性工具调用前弹审批按钮hitl_user_input.py用户输入在 Slack 内收集结构化输入hitl_external_execution.py外部执行展示工具参数由运维在系统外执行后回填结果hitl_incident_commander.py复合组合全部暂停类型的故障响应流程外部执行external execution的语义以 hitl_external_execution.py 为例它暂停在 Agent 无法执行的 Kubernetes 诊断上。定义工具时加上external_executionTruefrom agno.tools import tool tool(external_executionTrue) def run_kubectl(command: str) - str: Represent a kubectl command that the operator executes outside AgentOS. return command核心语义README 原文 示例指令双重印证对于external_executionTrue的工具Python 入口点在暂停之前不会执行Slack 会显示工具名称与参数该示例把运维要执行的完整kubectl命令放进可见的command参数里运维在 AgentOS 之外自行执行该操作运维提交的值成为被恢复 run 的工具结果。示例还配了一个内部 runbook 查找工具lookup_runbook(symptom)把CrashLoopBackOff、ImagePullBackOff、Pending等症状映射到排查指引整体尝试命令是 “Check api-gateway pods in the production namespace.”。团队级审批在哪里README 特别说明团队级审批team-level approval与成员暂停传播属于通用的 AgentOS 行为示例位于 05_human_in_the_loop/team_approval.py。Slack 接口会把这类 Team 需求经由同一个 interaction 路由渲染出来因此不需要在 Slack 示例目录里重复实现。测试范围这个目录的验证边界目录内的 TEST_LOG.md 记录了凭据门控的构建冒烟测试construction smokes。这些测试的边界非常清晰使用哨兵凭据sentinel credentials只 patch 掉 Slack 启动时的auth_test验证内容为app 生命周期、/health、/config、以及精确的 events/interactions 路由对不声称覆盖真实的 Slack 安装、事件投递、交互恢复、工具请求、模型推理或外发消息。也就是说该测试只证明“服务能按预期挂载与暴露路由”真实链路仍需在本地用 ngrok/cloudflared 配合真实 Slack App 做端到端验证。Troubleshooting症状与对策速查README 末尾给出的排障表是上线时最直接的检查清单症状可能原因Bot 无响应事件 URL 未通过验证、App 未被邀请进频道、必需事件未订阅私信正常但频道消息不响应缺少app_mention/message.channels或 App 不在该频道中流式回复返回internal_errorAgents AI Apps 未开启、缺少assistant:write、或 App 修改后未重新安装没有任务卡片slack_sdk版本低于 3.40.0没有建议提示词未订阅assistant_thread_started或 Suggested Prompts 未设为 Dynamic工作区搜索提示无 action token该 run 并非从 Slack Assistant 线程发起HITL 按钮无效果Interactivity 未启用或其 Request URL 配置错误Webhook 返回 403App 的签名密钥与接口配置不一致小结把 Agent 交给 Slack 的完整心智模型回顾整条链路Slack接口的价值在于它把 Slack 从“消息收发通道”升级成了“会话状态、用户身份、工具呈现与人工介入”的统一载体配置层创建 App、开启 Agents AI Apps、按示例文件的Slack scopes:行配置 OAuth scopes、订阅事件、按需开启 Interactivity最后用隧道把/slack/events与/slack/interactions暴露到公网挂载层AgentOS(agents[...], interfaces[Slack(agent...)])一行即可把单个 Agent 挂上Team 与 Workflow 同理多 App 场景各自指定prefix、token、signing_secret运行时层每线程一个会话、reply_to_mentions_only控制频道过滤、resolve_user_identity决定记忆如何归人、respond_to_other_apps决定是否聆听对等机器人体验层默认开启的流式回复配合 plan 任务卡片、loading 消息与动态建议提示词让长任务不再“失联”控制层Interactivity 端点回收确认/输入/外部执行三类暂停的产物让破坏性操作与人工作业始终处于人的监督之下。把这套清单与实际运行 basic.py、streaming_ux.py、peer_agents.py 等示例结合起来即可把本课的每一项机制落到真实可用的 Slack 机器人上。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考