OpenClaw多渠道集成实战：Telegram/Discord/Signal接入指南

# 在Telegram中找 @BotFather
# /newbot → 输入名称 → 获取Token
# Token示例：7812345678:AAExxxxxxxxxxxxxxxxxxxxxxxxxxxx

// openclaw.json
{
  "channels": {
    "telegram": {
      "enabled": true,
      "token": "7812345678:AAExxxxxxxxxxxxxxxxxxxxxxxxxxxx",
      "allowedUsers": ["your_user_id"],
      "commands": {
        "/summary": "总结最近对话",
        "/reset": "重置会话",
        "/status": "查看状态"
      }
    }
  }
}

# 内联按钮
def send_with_buttons(chat_id):
    bot.send_message(
        chat_id=chat_id,
        text="请选择操作：",
        reply_markup={
            "inline_keyboard": [[
                {"text": "📊 查看报告", "callback_data": "report"},
                {"text": "⚙️ 设置", "callback_data": "settings"}
            ]]
        }
    )

# 群组处理
def handle_group_message(message):
    if message.chat.type in ["group", "supergroup"]:
        # 群组中只在@提及或回复时响应
        if is_mentioned(message) or is_reply_to_bot(message):
            return process_message(message)
        return HEARTBEAT_OK  # 否则保持安静

用户: @MyAgent 帮我查一下服务器状态
Agent: 🔍 正在检查...
     
     ✅ 阿里云 ECS (182.92.245.108)
     - CPU: 12% | 内存: 1.2/2GB
     - 磁盘: 21% | 运行时间: 45天
     
     ✅ 腾讯云 (43.165.198.197)
     - CPU: 8% | 内存: 320/957MB
     - 磁盘: 21% | 论坛正常
     
     [📊 详细报告] [🔄 刷新]

# 1. 访问 Discord Developer Portal
# 2. 创建 Application → Bot
# 3. 获取Token
# 4. 开启MESSAGE CONTENT INTENT
# 5. 生成邀请链接，邀请到服务器

{
  "channels": {
    "discord": {
      "enabled": true,
      "token": "MTIxxxxxxxxxxxxxxxxxxx",
      "prefix": "!",
      "allowedChannels": ["1234567890"],
      "reactToMessages": true,
      "threadSupport": true
    }
  }
}

class DiscordBehavior:
    def should_respond(self, message):
        # 回应条件
        if message.mentions_bot:      # @提及
            return True
        if message.is_reply_to_bot:   # 回复bot
            return True
        if self.is_question(message): # 自然语言问题
            return True
        
        # 保持沉默
        return False  # → HEARTBEAT_OK
    
    def should_react(self, message):
        # 有趣的内容
        if "😂" in message.content or "lol" in message.content.lower():
            return "😂"
        if message.is_interesting:
            return "🤔"
        if message.is_helpful:
            return "👍"
        return None

用户: @Agent 分析一下这个错误日志 [附件]
Agent: 正在分析...
     [创建线程"错误日志分析"]
     在线程中回复详细分析结果

# Linux服务器
sudo apt install signal-cli

# 注册号码
signal-cli -u +8613789892969 register
signal-cli -u +8613789892969 verify <验证码>

{
  "channels": {
    "signal": {
      "enabled": true,
      "phoneNumber": "+8613789892969",
      "signalCliPath": "/usr/bin/signal-cli",
      "trustedContacts": ["+8613800138000"]
    }
  }
}

Telegram ─┐
Discord ──┤
Signal ───┼──→ OpenClaw消息总线 ──→ Agent推理 ──→ 响应路由
Slack ────┤                                      ↓
微信 ─────┘                              原渠道回复

class MessageRouter:
    def __init__(self, channels):
        self.channels = channels  # 各渠道的handler
    
    def handle(self, incoming_message):
        # 1. 统一消息格式
        unified = self.normalize(incoming_message)
        
        # 2. 识别渠道
        channel = unified.channel  # telegram/discord/...
        
        # 3. 加载渠道特定配置
        config = self.channels[channel].config
        
        # 4. 检查权限
        if not self.check_permission(unified, config):
            return None
        
        # 5. 处理消息
        response = self.agent.process(unified)
        
        # 6. 格式化响应（适配渠道）
        formatted = self.format_response(response, channel)
        
        # 7. 发送
        self.channels[channel].send(formatted)

def format_response(response, channel):
    if channel == "discord":
        # Discord: 支持Markdown
        return response.markdown
    
    elif channel == "telegram":
        # Telegram: 支持Markdown子集
        return response.markdown_v2
    
    elif channel == "signal":
        # Signal: 纯文本
        return response.plain_text
    
    elif channel == "wechat":
        # 微信：纯文本，限制长度
        return response.plain_text[:2048]
    
    elif channel == "slack":
        # Slack: Block Kit
        return response.slack_blocks

{
  "channels": {
    "telegram": {
      "allowedUsers": ["user_id_1", "user_id_2"],
      "allowedChats": ["chat_id_1"]
    },
    "discord": {
      "allowedChannels": ["channel_id_1"],
      "allowedRoles": ["role_id_1"]
    }
  }
}

# 群聊安全规则
GROUP_RULES = {
    "telegram": {
        "respond_to_mentions": True,
        "respond_to_replies": True,
        "silent_in_banter": True,  # 闲聊时保持沉默
        "max_response_per_hour": 20  # 防刷屏
    },
    "discord": {
        "respond_to_mentions": True,
        "use_threads_for_long_responses": True,
        "react_instead_of_reply": True,  # 优先用emoji反应
        "max_response_per_hour": 30
    }
}

渠道:
  - Telegram（日常使用）
  - Signal（敏感对话）
  - Discord（技术社区）

功能:
  - 服务器监控
  - 日程管理
  - 文件处理
  - 代码审查

安全:
  - 仅响应白名单用户
  - Signal用于密码/密钥等敏感信息
  - Discord仅在特定频道响应

渠道:
  - Discord（主渠道）
  - Telegram（备用）

功能:
  - FAQ自动回答
  - 问题分类和路由
  - 文档搜索
  - 人工转接

行为:
  - 公开频道：自动回答常见问题
  - 帮助频道：响应所有@提及
  - 其他频道：静默

class ReliableDelivery:
    async def send_with_retry(self, channel, message, max_retries=3):
        for attempt in range(max_retries):
            try:
                return await channel.send(message)
            except RateLimitError:
                await asyncio.sleep(2 ** attempt)
            except NetworkError:
                await asyncio.sleep(1)
        
        # 所有重试失败，记录并通知
        self.log_failure(channel, message)
        await self.notify_admin(f"发送失败: {channel}")