OpenClaw 消息工具详解：多渠道消息发送实战指南

消息发送是 AI Agent 与用户交互的核心桥梁，OpenClaw 的 message 工具提供了强大而灵活的多渠道消息发送能力。本文深入剖析 message 工具的设计理念、核心参数、多渠道适配机制，以及富文本消息、卡片消息、消息回复与引用、消息撤回与编辑等高级功能。通过 Telegram、飞书、Discord、Slack、WhatsApp 等主流平台的实战案例，帮助读者掌握跨平台消息发送的最

七夜zippoe

1191人浏览 · 2026-04-05 17:00:17

七夜zippoe · 2026-04-05 17:00:17 发布

摘要

1. 引言：消息工具的重要性

在 AI Agent 的世界里，消息是连接智能与用户的纽带。一个设计良好的消息系统，不仅要能发送文本，还要支持富媒体、交互组件、跨平台适配等复杂需求。OpenClaw 的 message 工具正是为此而生。

1.1 为什么需要统一的消息工具

想象这样一个场景：你开发了一个 AI 助手，最初只在 Telegram 上运行。随着用户需求增长，你需要将其扩展到飞书、Discord、Slack 等多个平台。如果没有统一的消息抽象层，你将面临以下挑战：

挑战	具体问题	影响
API 差异	每个平台的消息 API 参数、格式各不相同	开发成本成倍增加
功能差异	有的平台支持卡片消息，有的不支持	功能实现不一致
限流策略	不同平台的频率限制、消息长度限制不同	需要针对每个平台适配
媒体处理	图片、文件的上传方式各不相同	媒体发送逻辑复杂
交互组件	按钮、表单等交互元素实现方式迥异	用户体验难以统一

OpenClaw 的 message 工具通过统一的抽象层解决了这些问题。开发者只需调用一个工具，指定目标渠道和消息内容，框架会自动处理平台差异，实现"一次编写，多端运行"。

1.2 OpenClaw 消息工具的设计理念

OpenClaw 的消息工具设计遵循以下核心原则：

统一接口，平台适配

无论目标平台是 Telegram、飞书还是 Discord，调用方式保持一致。框架内部根据渠道类型自动选择合适的发送策略。

功能丰富，扩展灵活

message 工具不仅支持基础文本发送，还涵盖：

富文本格式化（Markdown、HTML）
媒体文件发送（图片、视频、文档）
交互式卡片消息
消息回复与引用
消息撤回与编辑
投票与问卷
表情反应

安全可控，权限精细

通过 groupPolicy、allowFrom 等配置，可以精确控制消息发送权限，防止滥用。

2. message 工具核心概念

2.1 工具定义与参数

message 工具是 OpenClaw 提供的核心工具之一，用于发送、删除和管理消息。其定义包含丰富的参数，支持多种消息操作。

核心参数说明

参数名	类型	必填	说明
`action`	string	是	操作类型：send、broadcast 等
`target`	string	否	目标渠道/用户 ID，省略则回复当前会话
`message`	string	否	消息文本内容
`channel`	string	否	指定渠道类型，如 telegram、feishu
`channelId`	string	否	指定渠道内的目标 ID
`media`	string	否	媒体文件 URL 或本地路径
`replyTo`	string	否	回复的消息 ID
`silent`	boolean	否	静默发送，不触发通知

目标参数解析

message 工具支持多种目标指定方式，灵活适应不同场景：

# 方式一：直接指定目标（推荐）
target: "telegram:-1001234567890"  # Telegram 群组
target: "feishu:ou_xxxxx"          # 飞书用户
target: "discord:123456789"        # Discord 频道

# 方式二：分离指定
channel: "telegram"
channelId: "-1001234567890"

# 方式三：省略目标（回复当前会话）
# 在对话上下文中，AI 自动推断回复目标
message: "这是回复消息"

2.2 消息发送流程

当 AI Agent 调用 message 工具发送消息时，OpenClaw 内部会经历以下处理流程：

这个流程展示了 OpenClaw 如何通过统一的工具接口，实现对不同平台的无缝适配。对于 AI Agent 而言，它只需要调用 message 工具，无需关心底层平台差异。

3. 多渠道消息发送

OpenClaw 原生支持 20+ 消息平台，每个平台都有其独特的配置方式和发送特性。本节将详细介绍主流平台的接入与消息发送方法。

3.1 支持的渠道概览

渠道	类型	特点	适用场景
Telegram	即时通讯	Bot API 成熟，支持频道、群组、私聊	国际化应用、社区运营
飞书/Lark	企业协作	卡片消息丰富，支持文档、审批集成	企业内部应用、办公自动化
Discord	社区平台	支持服务器、频道、线程，游戏社区活跃	游戏社区、技术社区
Slack	企业协作	企业级功能完善，支持 App、Slash 命令	企业办公、团队协作
WhatsApp	即时通讯	用户基数大，支持个人和商业账号	客户服务、营销触达
微信企业号	企业协作	国内企业首选，与微信生态打通	国内企业办公
钉钉	企业协作	阿里生态，支持钉钉小程序	阿里系企业应用
Matrix	去中心化	开源协议，支持联邦部署	隐私敏感场景
Signal	加密通讯	端到端加密，隐私保护强	安全通信场景
IRC	传统聊天	轻量级，兼容性好	技术社区、开源项目

3.2 Telegram 消息发送

Telegram 是 OpenClaw 支持最完善的渠道之一，具有 API 稳定、功能丰富、文档完善等优点。

配置示例

channels:
  telegram:
    enabled: true
    botToken: "123456789:ABCdefGHIjklMNOpqrsTUVwxyz"  # Bot Token
    dmPolicy: "pairing"      # 私聊策略：pairing/open/allowlist
    groupPolicy: "open"      # 群聊策略
    allowFrom: []            # 允许的用户列表
    groups:                  # 群组配置
      "-1001234567890":
        requireMention: true # 需要 @ 提及才响应

上述配置展示了 Telegram 渠道的基本设置。botToken 是从 BotFather 获取的机器人令牌，是连接 Telegram API 的关键凭证。dmPolicy 控制私聊权限策略，pairing 表示需要用户先与机器人配对才能对话。groupPolicy 控制群聊权限，open 表示任何群都可以添加机器人。groups 配置允许对特定群组进行精细化控制，如设置 requireMention 要求用户必须 @ 机器人才会响应。

消息发送代码示例

# Telegram 消息发送核心逻辑
async def send_telegram_message(
    chat_id: str,
    text: str,
    token: str,
    reply_to_message_id: str = None,
    message_thread_id: int = None,
    silent: bool = False
):
    """
    发送 Telegram 消息
    
    Args:
        chat_id: 目标聊天 ID（用户 ID、群组 ID 或频道 ID）
        text: 消息文本内容（支持 Markdown 和 HTML 格式）
        token: Bot Token
        reply_to_message_id: 回复的消息 ID
        message_thread_id: 话题/线程 ID（用于论坛群组）
        silent: 是否静默发送（无通知）
    
    Returns:
        dict: 包含 messageId 和发送状态的字典
    """
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    
    payload = {
        "chat_id": chat_id,
        "text": text,
        "parse_mode": "Markdown"  # 支持 Markdown 格式
    }
    
    # 可选参数
    if reply_to_message_id:
        payload["reply_to_message_id"] = reply_to_message_id
    if message_thread_id:
        payload["message_thread_id"] = message_thread_id
    if silent:
        payload["disable_notification"] = True
    
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=payload) as resp:
            result = await resp.json()
            if result.get("ok"):
                message_id = result["result"]["message_id"]
                return {"messageId": str(message_id), "chatId": chat_id}
            else:
                raise Exception(f"Telegram send failed: {result.get('description')}")

上述代码展示了 Telegram 消息发送的核心实现。函数接收聊天 ID、消息文本和 Bot Token 作为主要参数，构造 API 请求并发送。parse_mode 设置为 Markdown 表示消息支持 Markdown 格式化。可选参数包括回复消息 ID、话题线程 ID 和静默发送标志。函数返回包含 messageId 和 chatId 的字典，便于后续操作如编辑或删除消息。

3.3 飞书消息发送

飞书（Lark）是字节跳动的企业协作平台，提供了丰富的消息能力，特别是交互式卡片消息。

配置示例

channels:
  feishu:
    enabled: true
    appId: "cli_xxxxxxxxxxxxxxxx"           # 应用 ID
    appSecret: "xxxxxxxxxxxxxxxxxxxxxxxx"   # 应用密钥
    encryptKey: ""                          # 消息加密密钥（可选）
    verificationToken: ""                   # 验证令牌（可选）
    domain: "feishu"                        # 域名：feishu 或 lark
    connectionMode: "websocket"             # 连接模式：websocket/webhook
    dmPolicy: "pairing"                     # 私聊策略
    groupPolicy: "open"                     # 群聊策略
    renderMode: "auto"                      # 渲染模式：auto/raw/card

飞书配置需要从飞书开放平台获取 appId 和 appSecret。domain 区分国内版（feishu）和国际版（lark）。connectionMode 支持两种模式：websocket 使用长连接，适合服务器环境；webhook 使用 HTTP 回调，需要公网可访问的地址。renderMode 控制消息渲染方式，auto 会根据内容自动选择文本或卡片格式。

消息发送代码示例

# 飞书消息发送核心逻辑
async def send_feishu_message(
    receive_id: str,
    text: str,
    client: FeishuClient,
    receive_id_type: str = "open_id",
    reply_to_message_id: str = None
):
    """
    发送飞书消息
    
    Args:
        receive_id: 接收者 ID（open_id、user_id 或 chat_id）
        text: 消息文本内容（支持 Markdown 格式）
        client: 飞书客户端实例
        receive_id_type: ID 类型（open_id/user_id/chat_id）
        reply_to_message_id: 回复的消息 ID
    
    Returns:
        dict: 包含 messageId 的发送结果
    """
    # 构建消息体（富文本格式）
    content = json.dumps({
        "zh_cn": {
            "content": [
                [
                    {
                        "tag": "md",
                        "text": text
                    }
                ]
            ]
        }
    })
    
    if reply_to_message_id:
        # 回复消息
        response = await client.im.message.reply(
            path={"message_id": reply_to_message_id},
            data={
                "content": content,
                "msg_type": "post"
            }
        )
    else:
        # 发送新消息
        response = await client.im.message.create(
            params={"receive_id_type": receive_id_type},
            data={
                "receive_id": receive_id,
                "content": content,
                "msg_type": "post"
            }
        )
    
    if response.code == 0:
        message_id = response.data.get("message_id")
        return {"messageId": message_id, "chatId": receive_id}
    else:
        raise Exception(f"Feishu send failed: {response.msg}")

上述代码展示了飞书消息发送的核心实现。飞书使用富文本格式（msg_type: "post"）发送消息，内容需要构造为 JSON 结构。receive_id_type 指定 ID 类型，open_id 是推荐的用户标识方式。回复消息使用 im.message.reply API，发送新消息使用 im.message.create API。飞书 API 返回 code 字段表示操作结果，0 表示成功。

3.4 Discord 消息发送

Discord 是游戏和技术社区的热门平台，支持服务器、频道、线程等多层次结构。

配置示例

channels:
  discord:
    enabled: true
    token: "MTk4NjIyNDgzNDc..."  # Bot Token
    dm:
      policy: "pairing"           # 私聊策略
      allowFrom: []               # 允许的用户 ID 列表
    groupPolicy: "open"           # 群组策略
    guilds:                       # 服务器配置
      "123456789012345678":
        channels:
          "987654321098765432":
            requireMention: false

Discord 配置需要从 Discord Developer Portal 创建应用并获取 Bot Token。guilds 配置允许对特定服务器的特定频道进行精细控制。Discord 的权限模型比较复杂，需要在 Developer Portal 中配置 Intents 才能接收消息事件。

消息发送代码示例

# Discord 消息发送核心逻辑
async def send_discord_message(
    channel_id: str,
    text: str,
    token: str,
    reply_to_id: str = None,
    silent: bool = False
):
    """
    发送 Discord 消息
    
    Args:
        channel_id: 目标频道 ID 或用户 ID（格式：user:ID）
        text: 消息文本内容（支持 Markdown 格式）
        token: Bot Token
        reply_to_id: 回复的消息 ID
        silent: 是否静默发送
    
    Returns:
        dict: 包含 messageId 的发送结果
    """
    # 判断是私聊还是频道
    if channel_id.startswith("user:"):
        # 私聊：先创建 DM 频道
        user_id = channel_id.replace("user:", "")
        dm_url = f"https://discord.com/api/v10/users/@me/channels"
        dm_payload = {"recipient_id": user_id}
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                dm_url,
                json=dm_payload,
                headers={"Authorization": f"Bot {token}"}
            ) as dm_resp:
                dm_data = await dm_resp.json()
                target_channel_id = dm_data["id"]
    else:
        target_channel_id = channel_id
    
    # 发送消息
    url = f"https://discord.com/api/v10/channels/{target_channel_id}/messages"
    payload = {"content": text}
    
    if reply_to_id:
        payload["message_reference"] = {"message_id": reply_to_id}
    if silent:
        payload["flags"] = 4096  # SUPPRESS_NOTIFICATIONS
    
    async with aiohttp.ClientSession() as session:
        async with session.post(
            url,
            json=payload,
            headers={"Authorization": f"Bot {token}"}
        ) as resp:
            result = await resp.json()
            message_id = result.get("id")
            return {"messageId": message_id, "channel": "discord"}

上述代码展示了 Discord 消息发送的实现。Discord 的私聊需要先创建 DM 频道，然后才能发送消息。message_reference 字段用于实现消息回复功能。flags 字段设置为 4096 可以静默发送消息，不触发接收者的通知。Discord API 使用 REST 风格，通过 Authorization 头传递 Bot Token。

3.5 Slack 消息发送

Slack 是企业协作的主流平台，支持丰富的消息格式和交互组件。

配置示例

channels:
  slack:
    enabled: true
    botToken: "xoxb-xxxxxxxxxxxx-xxxxxxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxx"  # Bot Token
    appToken: "xapp-x-x-xxxxxxxxxxxx-xxxxxxxxxxxx-xxxxxxxxxxxxxxxxxxxxxxxx"  # App Token
    dm:
      policy: "pairing"
      allowFrom: []
    groupPolicy: "open"
    channels:
      "C1234567890":
        requireMention: true

Slack 配置需要两种 Token：botToken（xoxb- 开头）用于 API 调用，appToken（xapp- 开头）用于 Socket Mode 连接。Slack 的权限系统比较复杂，需要在应用配置页面添加相应的 OAuth Scopes。

4. 消息格式与样式

不同平台对消息格式的支持程度不同，OpenClaw 通过智能适配，让开发者能够用统一的方式发送格式化消息。

4.1 文本格式化

Markdown 支持

大多数平台都支持 Markdown 格式，但具体语法略有差异：

格式	Markdown 语法	Telegram	飞书	Discord	Slack
粗体	`text`	✅	✅	✅	✅
斜体	`text` 或 `_text_`	✅	✅	✅	✅
删除线	`~~text~~`	✅	✅	✅	✅
代码	`code`	✅	✅	✅	✅
代码块	` ```code````	✅	✅	✅	✅
链接	`[text](url)`	✅	✅	✅	✅
引用	`> text`	✅	✅	✅	❌
列表	`- item`	✅	✅	✅	✅

平台特定格式

某些平台支持特定的格式化语法：

<!-- Telegram 特有 -->
[inline mention](tg://user?id=123456789)  <!-- 内联提及用户 -->

<!-- Discord 特有 -->
<@123456789>      <!-- 提及用户 -->
<#987654321>      <!-- 提及频道 -->
<@&111222333>     <!-- 提及角色 -->

<!-- Slack 特有 -->
<@U12345678>      <!-- 提及用户 -->
<#C12345678>      <!-- 提及频道 -->
<!here>           <!-- @here 提及 -->
<!channel>        <!-- @channel 提及 -->

4.2 表格处理

Markdown 表格在不同平台的渲染效果差异较大。OpenClaw 提供了智能表格处理机制：

飞书卡片表格示例

飞书支持在卡片消息中渲染表格，效果最佳：

{
  "schema": "2.0",
  "config": {
    "wide_screen_mode": true
  },
  "body": {
    "elements": [
      {
        "tag": "markdown",
        "content": "| 列1 | 列2 | 列3 |\n|-----|-----|-----|\n| A | B | C |\n| D | E | F |"
      }
    ]
  }
}

4.3 代码块处理

代码块是技术文档和错误信息展示的重要方式。OpenClaw 支持语法高亮和自动格式化：

# 代码块示例
def hello_world():
    """这是一个示例函数"""
    print("Hello, OpenClaw!")
    return True

if __name__ == "__main__":
    hello_world()

上述代码块展示了 Python 代码的格式化效果。OpenClaw 会根据目标平台自动选择最佳的代码块渲染方式，确保代码可读性。

5. 富文本与卡片消息

卡片消息是现代聊天平台的重要特性，支持丰富的布局、交互组件和动态内容。

5.1 卡片消息概述

卡片消息相比纯文本消息具有以下优势：

特性	纯文本消息	卡片消息
布局控制	单一流式布局	多列、分区布局
交互组件	仅链接可点击	按钮、选择器、输入框
媒体展示	单张图片	图片、视频、图标组合
动态更新	需编辑整条消息	可更新特定区域
品牌定制	无	自定义颜色、样式

5.2 飞书卡片消息

飞书的卡片消息功能最为完善，支持复杂的交互逻辑。

基础卡片结构

{
  "schema": "2.0",
  "config": {
    "wide_screen_mode": true
  },
  "header": {
    "title": {
      "tag": "plain_text",
      "content": "卡片标题"
    },
    "subtitle": {
      "tag": "plain_text",
      "content": "卡片副标题"
    },
    "template": "blue"
  },
  "body": {
    "elements": [
      {
        "tag": "div",
        "text": {
          "tag": "lark_md",
          "content": "**正文内容**，支持 *Markdown* 格式"
        }
      },
      {
        "tag": "action",
        "actions": [
          {
            "tag": "button",
            "text": {
              "tag": "plain_text",
              "content": "确认"
            },
            "type": "primary",
            "value": {
              "action": "confirm"
            }
          },
          {
            "tag": "button",
            "text": {
              "tag": "plain_text",
              "content": "取消"
            },
            "type": "default",
            "value": {
              "action": "cancel"
            }
          }
        ]
      }
    ]
  }
}

上述 JSON 结构展示了飞书卡片消息的基本框架。schema 指定卡片版本，config 配置显示模式。header 定义卡片头部，包含标题、副标题和主题色。body 包含卡片主体内容，div 元素用于展示文本，action 元素包含交互按钮。按钮的 value 字段会在用户点击时传递给回调处理函数。

卡片消息发送代码

async def send_feishu_card(
    receive_id: str,
    card: dict,
    client: FeishuClient,
    receive_id_type: str = "open_id"
):
    """
    发送飞书卡片消息
    
    Args:
        receive_id: 接收者 ID
        card: 卡片 JSON 结构
        client: 飞书客户端
        receive_id_type: ID 类型
    
    Returns:
        dict: 发送结果
    """
    content = json.dumps(card)
    
    response = await client.im.message.create(
        params={"receive_id_type": receive_id_type},
        data={
            "receive_id": receive_id,
            "content": content,
            "msg_type": "interactive"  # 卡片消息类型
        }
    )
    
    if response.code == 0:
        return {"messageId": response.data.get("message_id")}
    else:
        raise Exception(f"Card send failed: {response.msg}")

上述代码展示了飞书卡片消息的发送方法。关键点是将 msg_type 设置为 interactive，并将卡片 JSON 序列化为字符串作为 content。卡片消息支持动态更新，可以通过 im.message.patch API 更新已发送的卡片内容。

5.3 Discord Embed 消息

Discord 使用 Embed 结构实现富文本消息：

async def send_discord_embed(
    channel_id: str,
    embed: dict,
    token: str
):
    """
    发送 Discord Embed 消息
    
    Args:
        channel_id: 目标频道 ID
        embed: Embed 结构
        token: Bot Token
    
    Returns:
        dict: 发送结果
    """
    url = f"https://discord.com/api/v10/channels/{channel_id}/messages"
    
    payload = {
        "embeds": [embed]  # Discord 支持一次发送多个 Embed
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.post(
            url,
            json=payload,
            headers={"Authorization": f"Bot {token}"}
        ) as resp:
            result = await resp.json()
            return {"messageId": result.get("id")}

Embed 结构示例

{
  "title": "Embed 标题",
  "description": "Embed 描述内容",
  "url": "https://example.com",
  "color": 5814783,
  "fields": [
    {
      "name": "字段1",
      "value": "字段值",
      "inline": true
    },
    {
      "name": "字段2", 
      "value": "字段值",
      "inline": true
    }
  ],
  "footer": {
    "text": "页脚文本",
    "icon_url": "https://example.com/icon.png"
  },
  "thumbnail": {
    "url": "https://example.com/thumbnail.png"
  },
  "image": {
    "url": "https://example.com/image.png"
  }
}

6. 消息回复与引用

消息回复和引用是对话场景中的常见需求，OpenClaw 提供了统一的支持方式。

6.1 回复消息

回复消息时，目标平台会在原消息下方显示关联，形成对话线索。

参数说明

参数	说明	示例
`replyTo`	回复的消息 ID	`"12345"`
`message`	回复内容	`"这是回复内容"`
`target`	目标渠道（可选）	`"telegram:-1001234567890"`

各平台回复机制

代码示例

# 回复消息的统一调用方式
await message_tool.invoke({
    "action": "send",
    "target": "telegram:-1001234567890",
    "message": "这是回复内容",
    "replyTo": "12345"  # 原消息 ID
})

6.2 引用消息

引用消息与回复消息类似，但不会形成回复线索，而是在消息中显示引用内容。

飞书引用消息

飞书支持在消息中引用其他消息的内容：

{
  "zh_cn": {
    "content": [
      [
        {
          "tag": "quote",
          "text": "这是被引用的消息内容"
        }
      ],
      [
        {
          "tag": "text",
          "text": "这是回复内容"
        }
      ]
    ]
  }
}

6.3 线程消息

Discord 和 Slack 支持线程功能，可以在主消息下创建讨论线程：

# Discord 线程消息
await send_discord_message(
    channel_id="123456789",
    text="这是线程中的消息",
    thread_id="987654321"  # 线程 ID
)

# Slack 线程消息
await send_slack_message(
    channel="C1234567890",
    text="这是线程中的消息",
    thread_ts="1234567890.123456"  # 线程时间戳
)

7. 消息撤回与编辑

消息发送后，有时需要修改或撤回。OpenClaw 支持这些操作，但不同平台的限制不同。

7.1 消息编辑

平台限制对比

平台	时间限制	内容限制	API
Telegram	无限制	仅可编辑自己的消息	`editMessageText`
飞书	24 小时	仅可编辑自己的消息	`im.message.update`
Discord	无限制	仅可编辑自己的消息	`PATCH /messages/{id}`
Slack	无限制	仅可编辑自己的消息	`chat.update`
WhatsApp	不支持	-	-

编辑消息代码示例

async def edit_telegram_message(
    chat_id: str,
    message_id: str,
    new_text: str,
    token: str
):
    """
    编辑 Telegram 消息
    
    Args:
        chat_id: 聊天 ID
        message_id: 要编辑的消息 ID
        new_text: 新的消息内容
        token: Bot Token
    
    Returns:
        dict: 编辑结果
    """
    url = f"https://api.telegram.org/bot{token}/editMessageText"
    
    payload = {
        "chat_id": chat_id,
        "message_id": int(message_id),
        "text": new_text,
        "parse_mode": "Markdown"
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=payload) as resp:
            result = await resp.json()
            if result.get("ok"):
                return {"success": True, "messageId": message_id}
            else:
                raise Exception(f"Edit failed: {result.get('description')}")

上述代码展示了 Telegram 消息编辑的实现。editMessageText API 接收聊天 ID、消息 ID 和新文本内容。Telegram 对消息编辑没有时间限制，这是相比其他平台的优势。编辑成功后，消息会显示"已编辑"标记。

7.2 消息撤回

平台限制对比

平台	时间限制	撤回他人消息	API
Telegram	无限制	需管理员权限	`deleteMessage`
飞书	无限制	需应用权限	`im.message.delete`
Discord	无限制	需管理消息权限	`DELETE /messages/{id}`
Slack	无限制	需管理员权限	`chat.delete`
WhatsApp	约 1 小时	仅自己的消息	-

撤回消息代码示例

async def delete_feishu_message(
    message_id: str,
    client: FeishuClient
):
    """
    撤回飞书消息
    
    Args:
        message_id: 要撤回的消息 ID
        client: 飞书客户端
    
    Returns:
        dict: 撤回结果
    """
    response = await client.im.message.delete(
        path={"message_id": message_id}
    )
    
    if response.code == 0:
        return {"success": True}
    else:
        raise Exception(f"Delete failed: {response.msg}")

8. 跨会话消息发送

OpenClaw 支持跨会话消息发送，即 AI Agent 可以向非当前对话的用户或群组发送消息。

8.1 使用场景

跨会话消息发送适用于以下场景：

主动通知：系统监控到异常时，主动通知管理员
定时提醒：根据日程安排，定时发送提醒消息
广播消息：向多个群组或用户发送相同内容
任务触发：某个任务完成后，通知相关人员

8.2 实现方式

指定目标发送

# 向指定 Telegram 用户发送消息
await message_tool.invoke({
    "action": "send",
    "target": "telegram:123456789",
    "message": "这是一条主动通知消息"
})

# 向指定飞书群组发送消息
await message_tool.invoke({
    "action": "send",
    "channel": "feishu",
    "channelId": "oc_xxxxxxxxxxxxx",
    "message": "任务已完成，请查看结果"
})

广播消息

# 向多个目标广播消息
await message_tool.invoke({
    "action": "broadcast",
    "targets": [
        "telegram:-1001234567890",
        "feishu:oc_xxxxxxxxxxxxx",
        "discord:123456789"
    ],
    "message": "系统维护通知：今晚 22:00-23:00 进行系统升级"
})

8.3 权限控制

跨会话消息发送涉及权限控制，防止滥用：

# 配置示例：限制跨会话发送
channels:
  telegram:
    enabled: true
    botToken: "xxx"
    # 只允许发送给已配对的用户
    dmPolicy: "pairing"
    allowFrom: []
    # 群组需要明确配置
    groupPolicy: "allowlist"
    groupAllowFrom:
      - "-1001234567890"

9. 实战案例

9.1 案例一：智能客服通知系统

场景描述

某电商平台需要构建智能客服通知系统，当订单状态变更时，自动通知用户。要求支持 Telegram 和飞书两种渠道。

架构设计

核心代码实现

import asyncio
from dataclasses import dataclass
from typing import Optional

@dataclass
class NotificationMessage:
    """通知消息数据结构"""
    user_id: str
    channel: str  # telegram / feishu
    template: str  # 模板名称
    data: dict     # 模板数据
    priority: str = "normal"  # normal / high

class NotificationService:
    """通知服务"""
    
    def __init__(self, message_tool):
        self.message_tool = message_tool
        self.templates = {
            "order_created": "🎉 订单创建成功！\n\n订单号：{order_id}\n商品：{product_name}\n金额：¥{amount}\n\n预计 {delivery_time} 送达",
            "order_shipped": "📦 您的订单已发货！\n\n订单号：{order_id}\n快递公司：{carrier}\n运单号：{tracking_number}\n\n点击查看物流详情",
            "order_delivered": "✅ 订单已送达！\n\n订单号：{order_id}\n收货人：{receiver}\n\n感谢您的购买，欢迎再次光临！"
        }
    
    async def send_notification(self, notification: NotificationMessage):
        """
        发送通知消息
        
        Args:
            notification: 通知消息对象
        
        Returns:
            dict: 发送结果
        """
        # 渲染消息模板
        template = self.templates.get(notification.template)
        if not template:
            raise ValueError(f"Unknown template: {notification.template}")
        
        message_text = template.format(**notification.data)
        
        # 构建目标
        target = f"{notification.channel}:{notification.user_id}"
        
        # 发送消息
        result = await self.message_tool.invoke({
            "action": "send",
            "target": target,
            "message": message_text
        })
        
        return result
    
    async def broadcast_notification(
        self,
        user_list: list,
        template: str,
        data: dict
    ):
        """
        批量发送通知
        
        Args:
            user_list: 用户列表 [{user_id, channel}, ...]
            template: 模板名称
            data: 模板数据
        
        Returns:
            list: 发送结果列表
        """
        tasks = []
        for user in user_list:
            notification = NotificationMessage(
                user_id=user["user_id"],
                channel=user["channel"],
                template=template,
                data=data
            )
            tasks.append(self.send_notification(notification))
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

# 使用示例
async def main():
    # 初始化服务
    service = NotificationService(message_tool)
    
    # 发送单个通知
    await service.send_notification(NotificationMessage(
        user_id="123456789",
        channel="telegram",
        template="order_shipped",
        data={
            "order_id": "ORD20240315001",
            "carrier": "顺丰速运",
            "tracking_number": "SF1234567890"
        }
    ))
    
    # 批量发送通知
    await service.broadcast_notification(
        user_list=[
            {"user_id": "123456789", "channel": "telegram"},
            {"user_id": "ou_xxxxx", "channel": "feishu"}
        ],
        template="order_created",
        data={
            "order_id": "ORD20240315002",
            "product_name": "OpenClaw 高级版订阅",
            "amount": "299.00",
            "delivery_time": "即时生效"
        }
    )

上述代码实现了一个完整的通知服务。NotificationMessage 数据类定义了通知消息的结构，包含用户 ID、渠道、模板名称和模板数据。NotificationService 类封装了消息发送逻辑，支持单个发送和批量发送。模板引擎使用 Python 的字符串格式化功能，将数据填充到预定义的模板中。broadcast_notification 方法使用 asyncio.gather 实现并发发送，提高批量发送效率。

9.2 案例二：多平台监控告警

场景描述

运维团队需要在系统出现异常时，通过多个渠道同时发送告警消息，确保告警及时送达。

实现代码

from datetime import datetime
from enum import Enum

class AlertLevel(Enum):
    """告警级别"""
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

class AlertService:
    """告警服务"""
    
    ALERT_EMOJIS = {
        AlertLevel.INFO: "ℹ️",
        AlertLevel.WARNING: "⚠️",
        AlertLevel.ERROR: "❌",
        AlertLevel.CRITICAL: "🚨"
    }
    
    def __init__(self, message_tool, alert_channels: dict):
        """
        初始化告警服务
        
        Args:
            message_tool: message 工具实例
            alert_channels: 告警渠道配置
                {
                    "telegram": ["-1001234567890"],
                    "feishu": ["oc_xxxxx"],
                    "discord": ["123456789"]
                }
        """
        self.message_tool = message_tool
        self.alert_channels = alert_channels
    
    def _format_alert(
        self,
        level: AlertLevel,
        title: str,
        message: str,
        details: dict = None
    ) -> str:
        """格式化告警消息"""
        emoji = self.ALERT_EMOJIS[level]
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        text = f"{emoji} **{title}**\n\n"
        text += f"级别：{level.value.upper()}\n"
        text += f"时间：{timestamp}\n\n"
        text += f"{message}\n"
        
        if details:
            text += "\n**详细信息：**\n"
            for key, value in details.items():
                text += f"- {key}: {value}\n"
        
        return text
    
    async def send_alert(
        self,
        level: AlertLevel,
        title: str,
        message: str,
        details: dict = None,
        channels: list = None
    ):
        """
        发送告警
        
        Args:
            level: 告警级别
            title: 告警标题
            message: 告警内容
            details: 详细信息
            channels: 指定发送渠道（可选，默认发送到所有配置渠道）
        """
        alert_text = self._format_alert(level, title, message, details)
        
        # 确定发送目标
        targets = []
        if channels:
            for channel in channels:
                if channel in self.alert_channels:
                    for channel_id in self.alert_channels[channel]:
                        targets.append(f"{channel}:{channel_id}")
        else:
            for channel, channel_ids in self.alert_channels.items():
                for channel_id in channel_ids:
                    targets.append(f"{channel}:{channel_id}")
        
        # 广播告警
        results = await self.message_tool.invoke({
            "action": "broadcast",
            "targets": targets,
            "message": alert_text
        })
        
        return results

# 使用示例
async def monitor_alert():
    service = AlertService(message_tool, {
        "telegram": ["-1001234567890"],
        "feishu": ["oc_xxxxx"],
        "discord": ["123456789"]
    })
    
    # 发送严重告警
    await service.send_alert(
        level=AlertLevel.CRITICAL,
        title="数据库连接异常",
        message="主数据库连接池已耗尽，当前活跃连接数达到上限",
        details={
            "数据库": "MySQL 主库",
            "活跃连接": "100/100",
            "等待队列": "45",
            "持续时间": "5 分钟"
        }
    )

9.3 案例三：交互式问卷系统

场景描述

HR 部门需要通过飞书发送员工满意度调查问卷，收集员工反馈。

飞书卡片问卷实现

async def send_survey_card(client: FeishuClient, user_id: str):
    """发送飞书问卷卡片"""
    
    card = {
        "schema": "2.0",
        "config": {"wide_screen_mode": True},
        "header": {
            "title": {"tag": "plain_text", "content": "📋 员工满意度调查"},
            "subtitle": {"tag": "plain_text", "content": "感谢您抽出时间参与调查"},
            "template": "turquoise"
        },
        "body": {
            "elements": [
                {
                    "tag": "div",
                    "text": {
                        "tag": "lark_md",
                        "content": "**Q1：您对当前工作环境满意吗？**"
                    }
                },
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "text": {"tag": "plain_text", "content": "非常满意 😊"},
                            "type": "primary",
                            "value": {"q1": "5"}
                        },
                        {
                            "tag": "button",
                            "text": {"tag": "plain_text", "content": "满意 🙂"},
                            "type": "primary",
                            "value": {"q1": "4"}
                        },
                        {
                            "tag": "button",
                            "text": {"tag": "plain_text", "content": "一般 😐"},
                            "type": "default",
                            "value": {"q1": "3"}
                        }
                    ]
                },
                {
                    "tag": "hr"
                },
                {
                    "tag": "div",
                    "text": {
                        "tag": "lark_md",
                        "content": "**Q2：您认为团队协作效率如何？**"
                    }
                },
                {
                    "tag": "input",
                    "placeholder": {"tag": "plain_text", "content": "请输入您的看法..."},
                    "element_id": "q2_input",
                    "required": False
                },
                {
                    "tag": "hr"
                },
                {
                    "tag": "action",
                    "actions": [
                        {
                            "tag": "button",
                            "text": {"tag": "plain_text", "content": "提交问卷"},
                            "type": "primary",
                            "value": {"action": "submit_survey"}
                        }
                    ]
                }
            ]
        }
    }
    
    response = await client.im.message.create(
        params={"receive_id_type": "open_id"},
        data={
            "receive_id": user_id,
            "content": json.dumps(card),
            "msg_type": "interactive"
        }
    )
    
    return response

上述代码展示了飞书交互式问卷卡片的实现。卡片包含标题、副标题和多个交互元素。action 元素中的按钮可以收集用户选择，input 元素允许用户输入文本反馈。用户点击按钮后，飞书会向配置的回调地址发送事件，OpenClaw 可以接收并处理这些事件，实现完整的问卷流程。

10. 最佳实践与注意事项

10.1 性能优化建议

优化点	建议	说明
消息分块	长消息自动分块发送	避免超出平台限制
并发控制	使用异步发送，控制并发数	提高发送效率，避免限流
缓存 Token	缓存访问令牌，避免重复获取	减少网络请求
错误重试	实现指数退避重试机制	应对临时网络故障
消息队列	高频场景使用消息队列	削峰填谷，保证可靠性

10.2 安全注意事项

敏感信息保护

# 配置文件中避免硬编码敏感信息
channels:
  telegram:
    enabled: true
    # 使用环境变量
    botToken: "${TELEGRAM_BOT_TOKEN}"
    # 或使用文件引用
    tokenFile: "/etc/openclaw/telegram-token"

权限最小化原则

# 只开放必要的权限
channels:
  feishu:
    dmPolicy: "allowlist"  # 只允许白名单用户
    allowFrom:
      - "ou_xxxxx"
    groupPolicy: "disabled"  # 禁用群聊

10.3 常见问题与解决方案

问题	原因	解决方案
消息发送失败	Token 过期或无效	检查 Token 有效性，刷新 Token
消息被截断	超出平台长度限制	使用消息分块或卡片消息
群组无响应	未配置群组权限	检查 `groupPolicy` 和 `groups` 配置
卡片显示异常	JSON 格式错误	验证卡片 JSON 结构
媒体发送失败	文件过大或格式不支持	检查文件大小和格式限制

11. 总结

本文全面深入地介绍了 OpenClaw 的 message 消息工具，从核心概念、多渠道支持、消息格式化，到富文本卡片、消息回复编辑、跨会话发送等高级功能，最后通过三个实战案例展示了实际应用场景。

核心要点回顾：

统一抽象，多端适配：message 工具通过统一的接口设计，屏蔽了不同平台的 API 差异，让开发者能够"一次编写，多端运行"。无论是 Telegram、飞书、Discord 还是 Slack，调用方式保持一致，大大降低了开发成本。
功能丰富，灵活可控：支持文本、媒体、卡片、投票等多种消息类型，以及回复、引用、编辑、撤回等操作。通过精细的权限配置，可以有效控制消息发送范围，防止滥用。
平台特性，智能适配：虽然接口统一，但 OpenClaw 也充分利用了各平台的独特能力。飞书的卡片消息、Discord 的 Embed、Slack 的线程消息等，都能在 message 工具中得到体现。
实战导向，即学即用：通过智能客服通知、多平台监控告警、交互式问卷三个案例，展示了 message 工具在实际业务中的应用方法。这些案例可以直接作为项目开发的参考模板。

思考与展望：

随着 AI Agent 技术的发展，消息交互将变得更加智能和自然。未来，message 工具可能会支持更多高级特性，如：

基于上下文的智能消息路由
多模态消息（语音、视频）的统一处理
跨平台会话状态同步
消息效果分析与优化建议

掌握 message 工具，是构建高质量 AI Agent 的重要一步。希望本文能够帮助读者深入理解 OpenClaw 的消息发送机制，在实际项目中发挥更大的价值。

参考资料

龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

个人知识库助手：基于 RAG 的 AI Agent Harness Engineering 搭建

多年积累的笔记、PDF、论文、工作文档分散在Obsidian、硬盘、Zotero、Notion各个地方，需要的时候翻半天找不到；用关键词检索只能匹配到字面相同的内容，语义相关的内容完全搜不到；想让大模型帮你总结文档、提取核心信息，又不敢把私密的工作笔记、个人数据传到公网的ChatGPT、文心一言等服务；开源的RAG项目遍地都是，但大多只能做简单的问答，不能帮你自动打标签、整理笔记、提取日程，没有自

龙虾开发者社区

词嵌入基础：Word2Vec 让文字变成向量

2026年的今天，AI大模型已经彻底渗透到了我们开发工作的方方面面：写代码有AI助手补全，做需求有智能体自动拆解流程，就连写文档都有大模型帮你润色优化。身边很多做了五六年开发的兄弟，都在往大模型、AI Agent方向转型，可我面了不下20个春招的候选人，发现了一个特别扎心的现象：80%的人张口闭口就是Transformer、RAG、多智能体协同，项目经历写的全是大模型应用落地，可当我问一句“大模型

龙虾开发者社区

AI Agent生态：平台型技术供应商的机会

本文将深入探讨AI Agent生态系统的构建逻辑，重点分析平台型技术供应商在这一生态中的核心定位、关键能力和商业机会。我们将从概念解析入手，逐步展开到技术架构、商业模式、实践案例，最后展望未来发展趋势。在深入探讨生态系统之前，我们首先需要明确什么是AI Agent。AI Agent（智能体）是指能够感知环境、做出决策并执行行动的自主实体。fill:#333;important;important;

龙虾开发者社区

所有评论(0)

查看更多评论

七夜zippoe

@sinat_41617212

已为社区贡献26条内容

OpenClaw 消息工具详解：多渠道消息发送实战指南

七夜zippoe

目录

摘要

1. 引言：消息工具的重要性

1.1 为什么需要统一的消息工具

1.2 OpenClaw 消息工具的设计理念

2. message 工具核心概念

2.1 工具定义与参数

2.2 消息发送流程

3. 多渠道消息发送

3.1 支持的渠道概览

3.2 Telegram 消息发送

3.3 飞书消息发送

3.4 Discord 消息发送

3.5 Slack 消息发送

4. 消息格式与样式

4.1 文本格式化

4.2 表格处理

4.3 代码块处理

5. 富文本与卡片消息

5.1 卡片消息概述

5.2 飞书卡片消息

5.3 Discord Embed 消息

6. 消息回复与引用

6.1 回复消息

6.2 引用消息

6.3 线程消息

7. 消息撤回与编辑

7.1 消息编辑

7.2 消息撤回

8. 跨会话消息发送

8.1 使用场景

8.2 实现方式

8.3 权限控制

9. 实战案例

9.1 案例一：智能客服通知系统

9.2 案例二：多平台监控告警

9.3 案例三：交互式问卷系统

10. 最佳实践与注意事项

10.1 性能优化建议

10.2 安全注意事项

10.3 常见问题与解决方案

11. 总结

参考资料

所有评论(0)

温馨提示：您尚未绑定手机号

七夜zippoe