Telegram

最后更新 2026年7月21日查看 Markdown 源文件 @flue/telegram

快速开始

使用 Telegram 蓝图为已有的 Flue 项目添加已验证的 Telegram Bot API webhook 入口和项目自有的出站 Telegram 访问。在终端或你选择的编码智能体中运行以下命令：

flue add channel telegram

概述

该蓝图安装 @flue/telegram 和 grammY，在源码根目录创建 channels/telegram.ts 模块，其中包含命名的 channel 导出和项目自有的 client 导出，并修改所选智能体以绑定生成的消息工具。

src/channels/telegram.ts（简略版）

import { createTelegramChannel } from '@flue/telegram';
import { dispatch } from '@flue/runtime';
import { Api } from 'grammy';
import { Assistant } from '../agents/assistant.ts';

export const client = new Api(process.env.TELEGRAM_BOT_TOKEN!);

export const channel = createTelegramChannel({
  secretToken: process.env.TELEGRAM_WEBHOOK_SECRET_TOKEN!,
  async webhook({ update }) {
    const incoming = update.message ?? update.channel_post ?? update.business_message;
    if (!incoming) return;
    const conversation = conversationFromMessage(incoming);
    await dispatch(Assistant, {
      id: channel.instanceId(conversation),
      // 此事件创建实例时记录一次；之后忽略。
      initialData: conversationData(conversation, incoming),
      message: {
        kind: 'signal',
        type: 'telegram.message',
        body: messageBody(incoming),
        attributes: { updateId: String(update.update_id) },
      },
    });
  },
});

简略版示例省略了生成的 conversationFromMessage、conversationData 和 messageBody 辅助函数、callback-query 分支和消息工具。配置完成后，传入的消息会继续处理其聊天、商务聊天或主题的智能体实例，绑定的 grammY 工具会回复到相同的目标。grammY 的 Fetch 导出在 Node 和 Cloudflare Workers 上均可运行（使用 Flue 的 nodejs_compat 设置）。

挂载渠道

渠道仅在 app.ts 挂载它的地方提供 HTTP 路由。挂载模块的命名 channel 导出：

src/app.ts

import { channel as telegram } from './channels/telegram.ts';

app.route('/channels/telegram', telegram.route());

channel.route() 是一个纯路由器工厂，相对于挂载路径提供渠道声明的路由。本指南中的 webhook 路径假定使用常规的 /channels/telegram 挂载路径；不同的挂载路径会相应调整。分发目标智能体模块携带 'use agent' 指令——该指令会注册它，因此仅用于分发的智能体不需要自己的 HTTP 挂载。

配置

变量	用途
`TELEGRAM_WEBHOOK_SECRET_TOKEN`	必需 — 验证入站 webhook 请求。
`TELEGRAM_BOT_TOKEN`	必需 — 验证出站 Bot API 调用。

它为已验证的入站安装 @flue/telegram，为项目自有的 Bot API 访问安装 grammY。grammY 发布了一个浏览器/Fetch 构建版本，可在 Node 和 workerd 中运行（使用 Flue 所需的 nodejs_compat 配置）。

将 webhook URL 设置为：

https://example.com/channels/telegram/webhook

使用仅包含字母、数字、下划线和连字符的独立随机 webhook 密钥。使用完整路由配置：

await client.setWebhook('https://example.com/channels/telegram/webhook', {
  secret_token: process.env.TELEGRAM_WEBHOOK_SECRET_TOKEN!,
  allowed_updates: [
    'message',
    'edited_message',
    'channel_post',
    'edited_channel_post',
    'business_message',
    'edited_business_message',
    'guest_message',
    'callback_query',
    'message_reaction',
    'message_reaction_count',
  ],
});

Telegram 在 X-Telegram-Bot-Api-Secret-Token 中发送密钥。@flue/telegram 在解析 Update 之前拒绝缺失或更改的值。Telegram 不签名请求体也不包含签名时间戳，因此不要跨机器人重用同一个密钥。

Webhook 交付和 getUpdates 轮询是互斥的。轮询不在 HTTP 渠道包的范围内。

渠道模块

src/channels/telegram.ts

import { createTelegramChannel, type TelegramConversationRef } from '@flue/telegram';
import { defineTool, dispatch } from '@flue/runtime';
import { Api } from 'grammy';
import type { Message } from 'grammy/types';
import * as v from 'valibot';
import { Assistant } from '../agents/assistant.ts';

export const client = new Api(process.env.TELEGRAM_BOT_TOKEN!);

export const channel = createTelegramChannel({
  secretToken: process.env.TELEGRAM_WEBHOOK_SECRET_TOKEN!,

  // 路径：/channels/telegram/webhook
  async webhook({ update }) {
    const incoming = update.message ?? update.channel_post ?? update.business_message;
    if (incoming) {
      const conversation = conversationFromMessage(incoming);
      await dispatch(Assistant, {
        id: channel.instanceId(conversation),
        // 此事件创建实例时记录一次；之后忽略。
        initialData: conversationData(conversation, incoming),
        message: {
          kind: 'signal',
          type: 'telegram.message',
          body: messageBody(incoming),
          attributes: { updateId: String(update.update_id) },
        },
      });
      return;
    }

    if (update.callback_query) {
      const query = update.callback_query;
      await client.answerCallbackQuery(query.id);
      if (!query.message) return;
      const conversation = conversationFromMessage(query.message);
      await dispatch(Assistant, {
        id: channel.instanceId(conversation),
        // 此事件创建实例时记录一次；之后忽略。
        initialData: conversationData(conversation, query.message),
        message: {
          kind: 'signal',
          type: 'telegram.callback_query',
          body: query.data ?? '',
          attributes: {
            updateId: String(update.update_id),
            fromId: String(query.from.id),
            ...(query.from.username === undefined ? {} : { fromUsername: query.from.username }),
          },
        },
      });
      return;
    }
  },
});

// 消息文本，或描述纯媒体消息的简短占位符。
function messageBody(message: Message): string {
  if (message.text !== undefined) return message.text;
  if (message.caption !== undefined) return message.caption;
  if (message.photo) return '[photo message]';
  if (message.video) return '[video message]';
  if (message.voice) return '[voice message]';
  if (message.document) return '[document message]';
  if (message.sticker) return '[sticker message]';
  return '[non-text message]';
}

// 从原生 Telegram Message 构建规范的目标标识。
function conversationFromMessage(message: Message): TelegramConversationRef {
  const topic = {
    ...(message.message_thread_id === undefined
      ? {}
      : { messageThreadId: message.message_thread_id }),
    ...(message.direct_messages_topic?.topic_id === undefined
      ? {}
      : { directMessagesTopicId: message.direct_messages_topic.topic_id }),
  };
  return message.business_connection_id
    ? {
        type: 'business-chat',
        businessConnectionId: message.business_connection_id,
        chatId: message.chat.id,
        ...topic,
      }
    : { type: 'chat', chatId: message.chat.id, ...topic };
}

// 实例创建数据：目标引用加上小型实例常量上下文。
function conversationData(conversation: TelegramConversationRef, message: Message) {
  return {
    type: conversation.type,
    chatId: conversation.chatId,
    ...(conversation.type === 'business-chat'
      ? { businessConnectionId: conversation.businessConnectionId }
      : {}),
    ...(conversation.messageThreadId === undefined
      ? {}
      : { messageThreadId: conversation.messageThreadId }),
    ...(conversation.directMessagesTopicId === undefined
      ? {}
      : { directMessagesTopicId: conversation.directMessagesTopicId }),
    ...(message.chat.title === undefined ? {} : { chatTitle: message.chat.title }),
  };
}

export function postMessage(ref: TelegramConversationRef) {
  return defineTool({
    name: 'post_telegram_message',
    description: '发送到绑定到此智能体的 Telegram 对话。',
    input: v.object({ text: v.pipe(v.string(), v.minLength(1)) }),
    async run({ data: { text } }) {
      const message = await client.sendMessage(ref.chatId, text, {
        ...(ref.type === 'business-chat'
          ? { business_connection_id: ref.businessConnectionId }
          : {}),
        ...(ref.messageThreadId ? { message_thread_id: ref.messageThreadId } : {}),
        ...(ref.directMessagesTopicId
          ? { direct_messages_topic_id: ref.directMessagesTopicId }
          : {}),
      });
      return { output: { messageId: message.message_id } };
    },
  });
}

绑定工具

initialData 是实例的创建数据：在事件创建实例时记录一次，之后忽略，因此渠道在每次分发时传递它。使用 useInitialData() 从智能体绑定工具，而不是解析实例 ID：

src/agents/assistant.ts

'use agent';
import { useInitialData, useModel, useTool } from '@flue/runtime';
import * as v from 'valibot';
import { postMessage } from '../channels/telegram.ts';

const chatData = v.object({
  type: v.literal('chat'),
  chatId: v.number(),
  messageThreadId: v.optional(v.number()),
  directMessagesTopicId: v.optional(v.number()),
  chatTitle: v.optional(v.string()),
});
const businessChatData = v.object({
  type: v.literal('business-chat'),
  businessConnectionId: v.string(),
  chatId: v.number(),
  messageThreadId: v.optional(v.number()),
  directMessagesTopicId: v.optional(v.number()),
  chatTitle: v.optional(v.string()),
});
const initialData = v.variant('type', [chatData, businessChatData]);

export function Assistant() {
  useModel('anthropic/claude-haiku-4-5');
  const data = useInitialData<v.InferOutput<typeof initialData>>();
  if (!data) throw new Error('This agent is created by the Telegram channel dispatch.');
  useTool(postMessage(data));
  const chatTitle = data.chatTitle ? ` ("${data.chatTitle}")` : '';
  return `Reply concisely in the bound Telegram conversation${chatTitle}.`;
}

Assistant.initialData = initialData;

受信任的代码绑定聊天、商务连接和可选主题。模型仅选择消息文本。

已验证的入站

Flue 在入站侧拥有一项工作：验证 X-Telegram-Bot-Api-Secret-Token 头，强制执行体限制，解析 JSON，并将单个提供商原生的 Bot API Update 转发给你的回调。没有并行的规范化模型——update 保留 Telegram 自己的字段名称、嵌套和判别器。权威类型是规范生成的 @grammyjs/types Update，@flue/telegram 重新导出它（与 grammY 使用的类型相同）。

因为每个交付最多存在一个 Update 的可选字段，所以对这些字段进行分支而不是判别器。上面的示例读取 update.message ?? update.channel_post ?? update.business_message 用于传入消息，update.callback_query 用于回调；将分支扩展到你的机器人在 allowed_updates 中启用的 update 系列。每个原生 Message 携带自己的对话标识，conversationFromMessage 读取它来构建 TelegramConversationRef。

每个交付包含一个 Update 并调用一次回调。update.update_id 是 Telegram 的排序和重复检测键。包不持久化它；当不允许重复入站时，在分发之前在应用存储中声明它。

Telegram 重试不成功的 webhook 请求。不返回任何内容会产生一个空的 200 响应。返回 JSON 以使用 Telegram 的 webhook 回复方法格式，或使用 Hono 上下文进行显式状态控制。

对话标识

conversationFromMessage 从原生 Message 派生规范的实例 ID：普通聊天、商务聊天、论坛线程和通道直接消息主题产生不同的 ID。商务标识包含 businessConnectionId，因为 Telegram 警告商务聊天 ID 可能与普通机器人聊天 ID 相同，且线程 ID（message_thread_id）和直接消息主题 ID（direct_messages_topic.topic_id）是互斥的。

某些原生 update 没有持久的聊天目标，因此不要从它们构建实例 ID。访客消息的 guest_query_id 授权一次短暂的 answerGuestQuery 响应，不得进入模型上下文、日志、持久会话数据或智能体标识。没有 message 的内联 callback_query 同样不提供可访问的聊天。

参见 @flue/telegram README。