企业官网建设流程全解析

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_slack，Slack 接口的底层实现位于 libs/agno/agno/os/interfaces/slack。

这个示例目录教你什么

README.md 开篇就点明了 Slack 接口的能力边界：它通过带签名的事件（events）与交互（interactions）Webhook将 Agent、Team 或 Workflow 连接到 Slack，支持流式回复、plan 模式工具卡片、建议提示词、文件收发、工作区搜索、按线程隔离的会话（per-thread sessions）以及人工介入表单。目录内共 12 个示例，每个示例的职责划分如下：

版本要求：README 明确说明，若要使用 plan 模式任务卡片进行流式回复，需要slack_sdk >= 3.40.0；演示环境（demo）会安装这些受支持的 Slack 依赖。从源码看，若未安装 Slack 相关依赖，接口模块在导入时会直接抛出提示：router.py 在ImportError时提示执行pip install 'agno[slack]'。

创建一个可用的 Slack App：八步配置清单

README 用八个步骤完整覆盖了 Slack App 的生命周期。以下逐步展开，并补充源码层面的注意事项。

1. 创建 App

1. 打开 https://api.slack.com/apps，点击Create New App；
2. 选择From scratch，命名并选择目标 workspace；
3. 在Basic Information页面复制Signing Secret。

该 Secret 正是服务端校验 Webhook 请求签名的密钥——security.py 中的verify_slack_signature负责完成验签，配置错误时会出现 README 故障排查表中“Webhook returns 403”的症状。

2. 启用 Agents & AI Apps

这是流式回复与工作区搜索的前提：

1. 侧边栏进入Agents & AI Apps；
2. 将Agent or Assistant开关设为On；
3. 在Suggested Prompts下选择Dynamic；
4. 点击Save。

启用后 Slack 会自动添加assistant:write权限范围。

3. 添加 OAuth Scopes

在OAuth & Permissions > Bot Token Scopes中添加你打算运行示例所需的权限范围。README 给出的完整对照表如下：

其中常见的流式回复配置组合是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 Token（xoxb-...）。修改 scope 之后必须重新安装（Reinstall）App，否则新权限不生效。

4. 订阅事件

1. 在Event Subscriptions中启用事件；

2. 将Request URL设置为：

   https://YOUR-PUBLIC-HOST/slack/events

3. 订阅 App 需要的 bot 事件：

4. 保存更改并重新安装 App。

值得强调的是：当你在 Slack 后台配置这个端点时，Slack 会立即发送一次签名 URL 验证请求（URL verification challenge），因此AgentOS 服务必须先通过公网可达，否则端点无法通过验证。这一点也是第 7 步“开隧道”的原因。

5. 启用 Interactivity（四个hitl_*.py示例必需）

1. 在Interactivity & Shortcuts中启用交互；

2. 将Request URL设置为：

   https://YOUR-PUBLIC-HOST/slack/interactions

3. 保存更改。

README 特别提醒：没有这个端点，审批按钮与被提交的输入就无法恢复一个被暂停的 run。可以理解为事件端点负责“接收任务”，交互端点负责“回收人工结论”。

6. 设置环境变量

export SLACK_TOKEN="xoxb-..." export SLACK_SIGNING_SECRET="..." export OPENAI_API_KEY="sk-..."

• SLACK_TOKEN即 Bot User OAuth Token；
• SLACK_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 接口会挂载以下路由：

此外 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( id="slack-basic-db", db_file="tmp/slack_basic.db", ) assistant = Agent( id="slack-assistant", name="Slack Assistant", model=OpenAIResponses(id="gpt-5.5"), db=db, instructions=[ "You are a helpful assistant in Slack.", "Keep answers concise and easy to scan.", ], add_history_to_context=True, num_history_runs=3, markdown=True, ) agent_os = AgentOS( id="slack-basic-os", description="AgentOS serving one persistent Slack assistant.", agents=[assistant], interfaces=[ Slack( agent=assistant, reply_to_mentions_only=True, ) ], ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app=app)

这段代码揭示了几个关键模式：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_only=True时，接口在频道中只处理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_identity=True时，接口会调用users.info：

1. 当 Slack 返回成员邮箱时，用邮箱作为稳定 ID；
2. 把成员显示名（display name）加入 run 元数据；
3. 查询失败时回退到 Slack ID。

user_memory.py 正是依赖此开关让记忆能跨线程跟随同一人，因此它要求users:read与users:read.email两个 scope。事件处理器侧的相关分支见 event_handler.py（if self.resolve_user_identity:触发users.info调用）。

流式 UX：loading、建议提示词与 plan 模式任务卡片

Slack 流式回复默认开启。接口会打开chat_stream，随 run 进展追加文本，并把工具活动渲染成任务卡片。streaming_ux.py 把这些控件的用法显式化：

researcher = Agent( id="slack-streaming-researcher", name="Slack Streaming Researcher", model=OpenAIResponses(id="gpt-5.5"), db=db, 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_context=True, num_history_runs=3, markdown=True, ) agent_os = AgentOS( id="slack-streaming-os", description="AgentOS demonstrating Slack streaming presentation controls.", agents=[researcher], interfaces=[ Slack( agent=researcher, streaming=True, task_display_mode="plan", loading_text="Researching...", loading_messages=[ "Searching current sources...", "Comparing the evidence...", "Preparing a concise answer...", ], suggested_prompts=[ { "title": "Technology brief", "message": "Summarize today's 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 线程的动态建议提示词，每条由title+message组成；
• task_display_mode="plan"：把工具调用过程以 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_directory="tmp/slack_downloads", enable_send_message=False, enable_send_message_thread=False, enable_list_channels=True, enable_get_channel_history=True, enable_upload_file=True, enable_download_file=True, enable_search_workspace=True, enable_get_thread=True, enable_list_users=False, enable_get_user_info=False, enable_get_channel_info=True, )

对应 Agent 的指令则把 Slack 定位成“工作区唯一事实来源”：主题类问题走search_workspace（action token 来自 Slack 事件）；已知频道走get_channel_history；重要回复用get_thread展开；需要内容分析时下载共享文件；仅在用户明确要求时才上传结果文件。该示例需要 README 表格中几乎全部 13 项 scope。

工作区搜索用的不是旧式搜索 API

SlackTools.search_workspace走的是 Slack 的assistant.search.contextaction，而非传统的 message-search API。工作机制如下：

• Slack 在 Assistant 线程事件上提供一个短时有效的action_token；
• 接口把它放入 run 元数据；
• 工具包在调用时从 run 元数据中读取该 token。

由此得出三点使用约束（README 原文要点）：

1. 必须在 Slack Assistant 线程中调用它，而不能在控制台 run 里调用；
2. 需要授予search:read.public、search:read.files、search:read.users；
3. 本课示例不需要SLACK_USER_TOKEN（用户级 token）；
4. 搜索可见性遵循 Slack 用户与工作区的既有权限。

team.py 把同样的搜索能力交给了 Team 中的一名专家：一个“技术专家”负责诊断代码/API/基础设施问题（携带 WebSearchTools），另一个“文档专家”负责检索工作区历史讨论（仅携带只开了enable_search_workspace=True的SlackTools）。支持 Team 则按需把任务分派给对应专家，并在回答中附带工具返回的证据链接。从源码看，当挂载对象是 Team 时，接口会自动把team.store_member_responses置为 True（router.py），否则continue_run无法可靠地从数据库重载成员的工具状态。

Workflow 也能被 Slack 唤起

workflow.py 演示了一个“先研究、后写作”的两步顺序 Workflow 挂在 Slack 线程里运行，并用 SQLite 持久化 workflow 历史：

content_workflow = Workflow( id="slack-content-workflow", name="Slack Content Workflow", description="Research a topic, then write a concise brief.", db=db, steps=[ Step(name="Research", agent=researcher), Step(name="Write", agent=writer), ], add_workflow_history_to_steps=True, num_history_runs=3, ) agent_os = AgentOS( id="slack-workflow-os", description="AgentOS serving a sequential content Workflow through Slack.", workflows=[content_workflow], interfaces=[Slack(workflow=content_workflow)], )

尝试提问示例："Research passkeys for SaaS apps and write a short adoption brief."它会先让 Research Agent 带回带来源链接的调研结论，再由 Writer Agent 产出 Slack 友好的简报。

多机器人挂载：一个 AgentOS、两个 Slack App

Slack 会把每个 App 的事件发送到一个已配置的 URL。multiple_bots.py 演示了在同一个服务器上挂载两个拥有独立 token、签名密钥与前缀的 Slack App：

export RESEARCH_SLACK_TOKEN="xoxb-..." export RESEARCH_SLACK_SIGNING_SECRET="..." export ANALYST_SLACK_TOKEN="xoxb-..." export ANALYST_SLACK_SIGNING_SECRET="..."

对应两个 App 在 Slack 后台的事件/交互 URL 要分别指向自己的前缀：

代码层面，两个Slack实例各自显式传入prefix、token、signing_secret：

interfaces=[ Slack( agent=researcher, prefix="/research", token=research_token, signing_secret=research_secret, streaming=True, ), Slack( agent=analyst, prefix="/analyst", token=analyst_token, signing_secret=analyst_secret, streaming=True, ), ],

由于会话键携带entity_id，即使两个 App 在同一个 workspace 里，它们的会话也互不干扰。该文件用required_env辅助函数在读不到环境变量时直接抛错，避免凭据缺省导致的隐性问题。

对等 App：安全、非对称的机器人间委托

默认情况下，机器人（bot）自己发出的消息会被丢弃。设置respond_to_other_apps=True才会让某个接口接收来自对等 App 的消息；即便如此，来自该接口自身 bot 身份的消息仍会被丢弃（防止自我回声）。对应过滤逻辑在 event_handler.py：if is_bot and not self.respond_to_other_apps:直接跳过。

peer_agents.py 刻意采用非对称拓扑：

• 协调者（coordinator）保持respond_to_other_apps=False，只听取真人；
• 研究员（researcher）设置为True，可以听到协调者。

这样既实现了单向委托，又不会形成自动“乒乓”循环。README 警告：除非你的应用另有显式循环保护，否则不要把两个 App 对称地都开启 peer 响应。

该示例需要的环境变量：

export COORDINATOR_SLACK_TOKEN="xoxb-..." export COORDINATOR_SLACK_SIGNING_SECRET="..." export RESEARCHER_SLACK_TOKEN="xoxb-..." export RESEARCHER_SLACK_SIGNING_SECRET="..." export RESEARCHER_SLACK_USER_ID="U..."

最后一项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( id="slack-user-memory-manager", model=OpenAIResponses(id="gpt-5.5"), db=db, memory_capture_instructions=( "Capture the user's name, role, communication preferences, current " "projects, and durable likes or dislikes." ), ) personal_assistant = Agent( id="slack-personal-assistant", name="Slack Personal Assistant", model=OpenAIResponses(id="gpt-5.5"), db=db, memory_manager=memory_manager, update_memory_on_run=True, ... ) agent_os = AgentOS( id="slack-memory-os", description="AgentOS with email-resolved Slack user memory.", agents=[personal_assistant], interfaces=[ Slack( agent=personal_assistant, resolve_user_identity=True, ) ], )

关键联动点在于：默认会话键按线程隔离，user_id是 Slack member ID，而 member ID 未必稳定地代表同一个自然人；开启resolve_user_identity=True后，接口尽量用邮箱做稳定 ID（见第 4 节），记忆才能跨线程正确归人。运行该示例需要users:read与users:read.email两个额外 scope。

Human-in-the-Loop：Slack 里的四类暂停

当 run 因等待人工而暂停时，Slack 接口会把被暂停的工具需求渲染成交互式卡片，并通过/slack/interactions恢复持久化的 run。这一课保持每个暂停类型一个聚焦示例，另加一个复合的故障响应流程：

外部执行（external execution）的语义

以 hitl_external_execution.py 为例：它暂停在 Agent 无法执行的 Kubernetes 诊断上。定义工具时加上external_execution=True：

from agno.tools import tool @tool(external_execution=True) def run_kubectl(command: str) -> str: """Represent a kubectl command that the operator executes outside AgentOS.""" return command

核心语义（README 原文 + 示例指令双重印证）：

• 对于external_execution=True的工具，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 末尾给出的排障表是上线时最直接的检查清单：

小结：把 Agent 交给 Slack 的完整心智模型

回顾整条链路，Slack接口的价值在于它把 Slack 从“消息收发通道”升级成了“会话状态、用户身份、工具呈现与人工介入”的统一载体：

1. 配置层：创建 App、开启 Agents & AI Apps、按示例文件的Slack scopes:行配置 OAuth scopes、订阅事件、按需开启 Interactivity，最后用隧道把/slack/events与/slack/interactions暴露到公网；
2. 挂载层：AgentOS(agents=[...], interfaces=[Slack(agent=...)])一行即可把单个 Agent 挂上，Team 与 Workflow 同理；多 App 场景各自指定prefix、token、signing_secret；
3. 运行时层：每线程一个会话、reply_to_mentions_only控制频道过滤、resolve_user_identity决定记忆如何归人、respond_to_other_apps决定是否聆听对等机器人；
4. 体验层：默认开启的流式回复配合 plan 任务卡片、loading 消息与动态建议提示词，让长任务不再“失联”；
5. 控制层：Interactivity 端点回收确认/输入/外部执行三类暂停的产物，让破坏性操作与人工作业始终处于人的监督之下。

把这套清单与实际运行 basic.py、streaming_ux.py、peer_agents.py 等示例结合起来，即可把本课的每一项机制落到真实可用的 Slack 机器人上。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno