04 - Telegram 通信实现详解

Telegram 集成架构

OpenClaw 使用 grammY 框架实现 Telegram Bot 功能，这是现代 TypeScript 生态中最受欢迎的 Telegram Bot 框架。该框架基于 Telegram Bot API，提供了类型安全的 API 封装和中间件系统。

架构位置

Telegram Bot API
        ↑
        │ (HTTP/Webhook)
        ↓
┌─────────────────────┐
│  grammY Middleware │
│  - 消息解析       │
│  - 命令路由       │
│  - 会话管理       │
└─────────┬───────────┘
          │
          ▼
┌─────────────────────┐
│  Gateway Channel   │
│  - 消息转换       │
│  - 路由决策       │
│  - 格式化输出     │
└─────────────────────┘

grammY 框架详解

为什么选择 grammY？

grammY 的核心优势：

1. TypeScript 原生支持: 完整的类型定义和智能提示
2. 中间件系统: 类似 Express.js 的请求处理链
3. 上下文管理: 自动会话状态维护
4. 插件生态: 丰富的第三方插件 (rate limiting, i18n 等)
5. 性能优异: 优化的更新处理和并发模型

grammY Bot 初始化

基础配置:

import { Bot } from 'grammy';

const bot = new Bot(process.env.TELEGRAM_BOT_TOKEN || config.channels.telegram.botToken);

// 配置选项
bot.api.config.useWebhookReply = false;  // 使用长轮询而非 Webhook
bot.catch(err => console.error('Telegram error:', err));

中间件栈:

// 错误处理中间件
bot.use((ctx, next) => {
  console.log(`[Telegram] Received: ${ctx.from?.username}: ${ctx.message?.text}`);
  return next();
});

// 命令处理
bot.command('start', handleStart);
bot.command('status', handleStatus);

// 消息处理
bot.on('message:text', handleMessage);

Gateway 通道集成

Telegram Channel 实现

OpenClaw 在 Gateway 中实现了统一的 Channel 接口，Telegram Channel 具体实现：

核心职责:

1. Bot 管理: 初始化和生命周期管理
2. 消息监听: 接收并转换 Telegram 更新
3. 消息发送: 将 Agent 响应发送回 Telegram
4. 权限控制: 实现白名单和群组策略
5. 媒体处理: 图片、文件、音频的转换

消息转换流程:

Telegram Update (Update object)
    ↓
gramY Context (ctx.message)
    ↓
提取: from.id, chat.id, text, caption, media
    ↓
构建内部 ChatMessage:
{
  platform: 'telegram',
  channelId: chat.id,
  userId: from.id,
  username: from.username,
  content: text || caption,
  media: { type, url, caption },
  metadata: { reply_to, forward_from }
}
    ↓
发送到 Gateway 路由器

权限和安全机制

白名单策略

配置示例:

{
  "channels": {
    "telegram": {
      "botToken": "123456:ABCDEF",
      "allowFrom": ["user1", "user2"],        // 用户名白名单
      "dmPolicy": "pairing",                   // 配对策略
      "groups": {                             // 群组配置
        "group-123": {
          "requireMention": true,              // 需要 @提及
          "allowCommands": ["/status", "/new"]
        }
      }
    }
  }
}

配对机制 (Pairing)

未知用户发送消息时：

新用户消息 → Gateway 检测未配对
    ↓
生成配对码: "PAIR-ABC123"
    ↓
回复: "请运行 openclaw pairing approve telegram PAIR-ABC123"
    ↓
用户执行 CLI 命令
    ↓
Gateway 验证配对码并添加到白名单
    ↓
后续消息正常处理

群组保护

策略选项:

1. requireMention: 仅响应 @bot 的消息
2. groupActivation: mention 或 always 模式
3. allowedCommands: 限制可用命令列表

代码示例:

if (ctx.chat.type === 'group' || ctx.chat.type === 'supergroup') {
  const config = config.channels.telegram.groups[ctx.chat.id];
  
  if (config?.requireMention) {
    // 检查是否 @ 提及 bot
    const text = ctx.message?.text || '';
    const isMentioned = text.includes(`@${bot.botInfo.username}`);
    if (!isMentioned) return;  // 忽略非提及消息
  }
}

命令系统

原生命令

OpenClaw 支持的 Telegram 命令：

命令实现:

// /status 命令
bot.command('status', async (ctx) => {
  const session = gateway.getSession(ctx.chat.id);
  const status = await gateway.getSessionStatus(session);
  
  const response = `
🤖 当前状态
━━━━━━━━━━━━
模型: ${status.model}
Token 使用: ${status.tokens}/${status.maxTokens}
成本: ${status.cost || 'N/A'}
思考级别: ${status.thinking}
  `.trim();
  
  await ctx.reply(response);
});

// /think 命令
bot.command('think', async (ctx) => {
  const level = ctx.match;  // 匹配到的级别参数
  const validLevels = ['off', 'minimal', 'low', 'medium', 'high', 'xhigh'];
  
  if (!validLevels.includes(level)) {
    return ctx.reply(`无效级别。有效选项: ${validLevels.join(', ')}`);
  }
  
  await gateway.setSessionOption(ctx.chat.id, 'thinking', level);
  await ctx.reply(`✅ 思考级别已设置为: ${level}`);
});

群组命令

群组中的命令额外验证：

bot.command('restart', async (ctx) => {
  if (ctx.chat.type === 'private') return;  // 私聊不允许
  
  // 验证是否为群组所有者
  const groupConfig = config.channels.telegram.groups[ctx.chat.id];
  const isOwner = ctx.from?.id === groupConfig?.ownerId;
  
  if (!isOwner) {
    return ctx.reply('❌ 仅群组所有者可执行此命令');
  }
  
  await gateway.restart();
  await ctx.reply('🔄 Gateway 正在重启...');
});

消息格式化与响应

Markdown 支持

Telegram 支持两种 Markdown 模式：

MarkdownV2 (默认):

const message = `
*粗体文本*
_斜体文本*
`代码片段`
[链接](https://example.com)
`.trim();

await ctx.reply(message, { parse_mode: 'MarkdownV2' });

HTML 格式:

await ctx.reply('<b>粗体</b> <i>斜体</i>', { 
  parse_mode: 'HTML' 
});

流式输出

Agent 长响应的流式发送：

async function streamResponse(ctx, runId) {
  let chunks = [];
  
  // 订阅 Agent 事件
  gateway.on(`agent:${runId}`, async (event) => {
    chunks.push(event.content);
    
    // 每 1000ms 或 500 字符发送一次更新
    if (shouldSendUpdate(chunks)) {
      const message = chunks.join('');
      
      if (ctx.chat.type === 'private') {
        // 私聊：编辑消息
        await ctx.api.editMessageText(
          ctx.chat.id,
          ctx.message?.message_id,
          message,
          { parse_mode: 'MarkdownV2' }
        );
      } else {
        // 群聊：发送新消息
        await ctx.reply(message, { parse_mode: 'MarkdownV2' });
      }
    }
  });
}

媒体文件处理

接收媒体:

bot.on('message:photo', async (ctx) => {
  const photo = ctx.message.photo[ctx.message.photo.length - 1];  // 最大尺寸
  const file = await ctx.api.getFile(photo.file_id);
  const url = `https://api.telegram.org/file/bot${bot.token}/${file.file_path}`;
  
  // 下载到本地
  const localPath = await downloadMedia(url);
  
  // 转换为内部消息格式
  const chatMessage = {
    platform: 'telegram',
    media: { type: 'image', path: localPath }
  };
  
  await gateway.routeMessage(chatMessage);
});

发送媒体:

async function sendMedia(chatId, media) {
  if (media.type === 'image') {
    await bot.api.sendPhoto(chatId, media.path, {
      caption: media.caption,
      parse_mode: 'MarkdownV2'
    });
  } else if (media.type === 'document') {
    await bot.api.sendDocument(chatId, media.path);
  }
}

Webhook 集成

Webhook 配置

OpenClaw 支持通过 Webhook 接收 Telegram 更新，比长轮询更高效：

配置:

{
  "channels": {
    "telegram": {
      "webhookUrl": "https://mydomain.com/webhook/telegram",
      "webhookSecret": "my-secret-key"
    }
  }
}

Webhook 端点:

// Gateway HTTP 服务器
app.post('/webhook/telegram', async (req, res) => {
  const signature = req.headers['x-telegram-bot-api-secret-token'];
  const secret = config.channels.telegram.webhookSecret;
  
  // 验证签名
  if (signature !== secret) {
    return res.status(401).json({ error: 'Unauthorized' });
  }
  
  const update = req.body;
  
  // 转发给 grammY
  await bot.handleUpdate(update);
  
  res.status(200).send('OK');
});

启动 Webhook:

async function setupWebhook() {
  const url = config.channels.telegram.webhookUrl;
  const secret = config.channels.telegram.webhookSecret;
  
  await bot.api.setWebhook(url, {
    secret_token: secret,
    drop_pending_updates: true,
    allowed_updates: ['message', 'callback_query']
  });
  
  console.log(`✅ Telegram webhook set: ${url}`);
}

故障排查

常见问题

1. Bot Token 无效

Error: 401 Unauthorized
解决: 检查 TELEGRAM_BOT_TOKEN 环境变量或配置文件

2. Webhook 更新失败

Error: Conflict: can't use getUpdates method while webhook is active
解决: telegramBot.webhookUrl 设为 null 删除现有 webhook

3. 消息格式错误

Error: Bad Request: can't parse entities
解决: 检查 MarkdownV2 特殊字符转义

4. 群组消息无响应

解决: 检查 requireMention 设置和 @bot 拼写

调试技巧

启用详细日志:

grammY logger 设置:
bot.use((ctx, next) => {
  console.log(JSON.stringify(ctx.update, null, 2));
  return next();
});

测试 Webhook:

curl -X POST https://api.telegram.org/bot<TOKEN>/getWebhookInfo

手动删除 Webhook:

curl -X POST https://api.telegram.org/bot<TOKEN>/deleteWebhook

结论

OpenClaw 的 Telegram 集成通过 grammY 框架和 Gateway Channel 抽象层，实现了功能丰富且安全的通信机制。其核心优势在于：

1. 统一的消息转换: Telegram 特定格式转换为内部协议，便于多平台统一处理
2. 细粒度权限控制: 白名单、配对机制、群组保护等多层安全策略
3. 丰富的命令系统: 支持会话管理、调试、监控等完整功能集
4. 流式响应支持: 长文本响应的分块发送，提升用户体验
5. Webhook 集成: 高效的更新接收机制，支持生产环境部署

这种设计使得 OpenClaw 在 Telegram 上提供了接近原生应用的体验，同时保持了与 WhatsApp、Slack 等其他平台的一致性。