flare-agent Bài 10: Channels — Telegram & Web Chat

Series: Build Your Own AI Agent Framework trên Cloudflare Bài: 10 / 10 — @flare-agent/channels

---

Channel là gì trong framework?

Agent xử lý logic — Channel xử lý I/O với thế giới bên ngoài.

Telegram message
      ↓
  TelegramChannel  ← normalize về IncomingMessage
      ↓
  Agent.run()      ← xử lý
      ↓
  TelegramChannel  ← gửi response về đúng nơi
      ↓
Telegram reply

Mỗi channel implement cùng 1 interface — agent không biết và không cần biết message đến từ Telegram hay Web Chat.

---

Cấu trúc package

packages/channels/
  src/
    types.ts              # IncomingMessage, OutgoingMessage, Channel interface
    telegram/
      TelegramChannel.ts  # Webhook handler + reply
      types.ts            # Telegram-specific types
    webchat/
      WebChatChannel.ts   # SSE + WebSocket handler
      types.ts
    index.ts

---

Channel Interface

// src/types.ts

export interface IncomingMessage {
  channelType: 'telegram' | 'webchat';
  messageId: string;
  sessionId: string;      // dùng làm agent sessionId
  userId: string;         // unique per user per channel
  text: string;
  raw: unknown;           // raw payload từ channel
  metadata?: {
    firstName?: string;
    username?: string;
    chatId?: string | number;
  };
}

export interface OutgoingMessage {
  text: string;
  sessionId: string;
  metadata?: Record<string, unknown>;
}

export interface Channel {
  // Parse raw webhook payload thành IncomingMessage
  parse(request: Request): Promise<IncomingMessage | null>;

  // Gửi response về channel
  send(message: OutgoingMessage, incoming: IncomingMessage): Promise<void>;

  // Verify webhook signature (bảo mật)
  verify(request: Request): Promise<boolean>;
}

---

Telegram Channel

Setup

Trước tiên cần tạo bot qua @BotFather trên Telegram, lấy BOT_TOKEN, rồi set webhook:

curl "https://api.telegram.org/bot<TOKEN>/setWebhook" \
  -d "url=https://your-worker.workers.dev/channels/telegram"

Implementation

// src/telegram/TelegramChannel.ts
import type { Channel, IncomingMessage, OutgoingMessage } from '../types';

interface TelegramConfig {
  botToken: string;
  secretToken?: string; // webhook secret để verify
}

interface TelegramUpdate {
  update_id: number;
  message?: {
    message_id: number;
    from: { id: number; first_name: string; username?: string };
    chat: { id: number };
    text?: string;
  };
}

export class TelegramChannel implements Channel {
  private apiBase: string;

  constructor(private config: TelegramConfig) {
    this.apiBase = `https://api.telegram.org/bot${config.botToken}`;
  }

  async verify(request: Request): Promise<boolean> {
    if (!this.config.secretToken) return true;
    const token = request.headers.get('X-Telegram-Bot-Api-Secret-Token');
    return token === this.config.secretToken;
  }

  async parse(request: Request): Promise<IncomingMessage | null> {
    const update = await request.json<TelegramUpdate>();
    const msg = update.message;

    // Bỏ qua nếu không có text (sticker, photo, ...)
    if (!msg?.text) return null;

    return {
      channelType: 'telegram',
      messageId: `tg_${update.update_id}`,
      // sessionId = chatId — mỗi conversation là 1 session
      sessionId: `tg_${msg.chat.id}`,
      userId: `tg_user_${msg.from.id}`,
      text: msg.text,
      raw: update,
      metadata: {
        firstName: msg.from.first_name,
        username: msg.from.username,
        chatId: msg.chat.id,
      },
    };
  }

  async send(message: OutgoingMessage, incoming: IncomingMessage): Promise<void> {
    const chatId = incoming.metadata?.chatId;
    if (!chatId) return;

    // Telegram giới hạn 4096 chars/message — tự động split
    const chunks = this.splitMessage(message.text);

    for (const chunk of chunks) {
      await fetch(`${this.apiBase}/sendMessage`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          chat_id: chatId,
          text: chunk,
          parse_mode: 'Markdown', // hỗ trợ **bold**, `code`
        }),
      });
    }
  }

  // Typing indicator — hiện "đang nhập..."
  async sendTyping(chatId: number): Promise<void> {
    await fetch(`${this.apiBase}/sendChatAction`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        chat_id: chatId,
        action: 'typing',
      }),
    });
  }

  private splitMessage(text: string, maxLen = 4096): string[] {
    if (text.length <= maxLen) return [text];
    const chunks: string[] = [];
    for (let i = 0; i < text.length; i += maxLen) {
      chunks.push(text.slice(i, i + maxLen));
    }
    return chunks;
  }
}

---

Web Chat Channel

SSE cho streaming, fallback REST cho non-streaming:

// src/webchat/WebChatChannel.ts
import type { Channel, IncomingMessage, OutgoingMessage } from '../types';

interface WebChatConfig {
  // Optional: verify requests bằng API key
  apiKey?: string;
}

interface WebChatPayload {
  sessionId: string;
  userId?: string;
  text: string;
  stream?: boolean;
}

export class WebChatChannel implements Channel {
  // Map sessionId → SSE controller (cho streaming)
  private streams = new Map<string, ReadableStreamDefaultController<string>>();

  constructor(private config: WebChatConfig = {}) {}

  async verify(request: Request): Promise<boolean> {
    if (!this.config.apiKey) return true;
    const key = request.headers.get('X-Api-Key');
    return key === this.config.apiKey;
  }

  async parse(request: Request): Promise<IncomingMessage | null> {
    const body = await request.json<WebChatPayload>();
    if (!body.text) return null;

    return {
      channelType: 'webchat',
      messageId: `wc_${crypto.randomUUID()}`,
      sessionId: body.sessionId,
      userId: body.userId ?? `anon_${body.sessionId}`,
      text: body.text,
      raw: body,
      metadata: { stream: body.stream },
    };
  }

  async send(message: OutgoingMessage, incoming: IncomingMessage): Promise<void> {
    // Web Chat dùng SSE — response được handle ở createSSEHandler
    // Channel này chỉ cần implement interface
  }

  // Tạo SSE response cho 1 agent run
  createSSEResponse(
    agentStream: ReadableStream<string>
  ): Response {
    const encoder = new TextEncoder();

    const sseStream = agentStream.pipeThrough(
      new TransformStream<string, Uint8Array>({
        transform(chunk, controller) {
          controller.enqueue(
            encoder.encode(`data: ${JSON.stringify({ text: chunk })}\n\n`)
          );
        },
        flush(controller) {
          controller.enqueue(encoder.encode('data: [DONE]\n\n'));
        },
      })
    );

    return new Response(sseStream, {
      headers: {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        'Access-Control-Allow-Origin': '*',
      },
    });
  }
}

---

Channel Runner — wire channel + agent

// src/ChannelRunner.ts
import type { Channel, IncomingMessage } from './types';
import type { Agent } from '@flare-agent/core';
import type { AgentContext } from '@flare-agent/types';

export class ChannelRunner {
  constructor(
    private channel: Channel,
    private agent: Agent,
    private buildContext?: (msg: IncomingMessage) => Partial<AgentContext>
  ) {}

  // Handle 1 incoming request — dùng trong Hono route
  async handle(request: Request, env: Record<string, unknown>): Promise<Response> {
    // 1. Verify
    const valid = await this.channel.verify(request);
    if (!valid) return new Response('Unauthorized', { status: 401 });

    // 2. Parse
    const incoming = await this.channel.parse(request.clone());
    if (!incoming) return new Response('OK'); // ignored message type

    // 3. Build context
    const extraCtx = this.buildContext?.(incoming) ?? {};
    const ctx: AgentContext = {
      sessionId: incoming.sessionId,
      userId: incoming.userId,
      env,
      ...extraCtx,
    };

    // 4. Run agent
    const result = await this.agent.run(incoming.text, ctx);

    // 5. Send response
    await this.channel.send(
      { text: result.output, sessionId: incoming.sessionId },
      incoming
    );

    // Telegram cần trả về 200 OK ngay
    return new Response('OK');
  }
}

---

Tích hợp vào Hono Worker

// apps/worker/src/index.ts
import { Hono } from 'hono';
import { TelegramChannel } from '@flare-agent/channels';
import { WebChatChannel } from '@flare-agent/channels';
import { ChannelRunner } from '@flare-agent/channels';
import { vocabularyAgent } from './agents/vocabulary';

const app = new Hono<{ Bindings: Env }>();

// --- Telegram ---
app.post('/channels/telegram', async (c) => {
  const telegram = new TelegramChannel({
    botToken: c.env.TELEGRAM_BOT_TOKEN,
    secretToken: c.env.TELEGRAM_SECRET,
  });

  // Hiện typing indicator trước khi run agent
  const update = await c.req.json<any>();
  if (update.message?.chat?.id) {
    await telegram.sendTyping(update.message.chat.id);
  }

  const runner = new ChannelRunner(
    telegram,
    vocabularyAgent,
    (msg) => ({
      metadata: {
        platform: 'telegram',
        username: msg.metadata?.username,
      },
    })
  );

  return runner.handle(
    new Request(c.req.url, {
      method: 'POST',
      body: JSON.stringify(update),
      headers: c.req.raw.headers,
    }),
    c.env as any
  );
});

// --- Web Chat: REST ---
app.post('/channels/webchat', async (c) => {
  const webchat = new WebChatChannel({
    apiKey: c.env.WEBCHAT_API_KEY,
  });

  const runner = new ChannelRunner(webchat, vocabularyAgent);
  return runner.handle(c.req.raw, c.env as any);
});

// --- Web Chat: Streaming ---
app.post('/channels/webchat/stream', async (c) => {
  const webchat = new WebChatChannel();
  const body = await c.req.json<{ text: string; sessionId: string; userId?: string }>();

  const ctx: AgentContext = {
    sessionId: body.sessionId,
    userId: body.userId,
    env: c.env as any,
  };

  // Lấy provider stream trực tiếp
  const messages = [{ role: 'user' as const, content: body.text }];
  const stream = vocabularyAgent.stream(messages, ctx);

  return webchat.createSSEResponse(stream);
});

export default app;

---

wrangler.toml — thêm secrets

[secrets]
TELEGRAM_BOT_TOKEN = ""
TELEGRAM_SECRET = ""
WEBCHAT_API_KEY = ""

---

React Hook — Web Chat client

// apps/web/src/hooks/useTelegramStyle.ts
// Dùng cho web chat có giao diện giống Telegram

export function useWebChat(sessionId: string) {
  const [messages, setMessages] = useState<Message[]>([]);
  const [streaming, setStreaming] = useState('');

  const send = async (text: string) => {
    // Thêm user message ngay lập tức
    setMessages((prev) => [...prev, { role: 'user', text }]);

    const res = await fetch('/channels/webchat/stream', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({ text, sessionId }),
    });

    // Đọc SSE stream
    const reader = res.body!.getReader();
    const decoder = new TextDecoder();
    let full = '';

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      for (const line of decoder.decode(value).split('\n')) {
        if (!line.startsWith('data: ')) continue;
        const data = line.slice(6);
        if (data === '[DONE]') break;
        try {
          full += JSON.parse(data).text;
          setStreaming(full);
        } catch { /* skip */ }
      }
    }

    setMessages((prev) => [...prev, { role: 'assistant', text: full }]);
    setStreaming('');
  };

  return { messages, streaming, send };
}

---

Thêm channel mới sau này

Chỉ cần implement Channel interface:

// Ví dụ: Slack channel
export class SlackChannel implements Channel {
  async verify(request) { /* verify Slack signature */ }
  async parse(request) { /* parse Slack event */ }
  async send(message, incoming) { /* gửi về Slack */ }
}

// Wire vào ngay
app.post('/channels/slack', async (c) => {
  const runner = new ChannelRunner(
    new SlackChannel({ signingSecret: c.env.SLACK_SIGNING_SECRET }),
    vocabularyAgent
  );
  return runner.handle(c.req.raw, c.env as any);
});

---

Checklist

• Tạo packages/channels/
• Set Telegram webhook: https://your-worker.workers.dev/channels/telegram
• Add secrets vào wrangler: TELEGRAM_BOT_TOKEN, TELEGRAM_SECRET
• Test gửi /start cho bot
• Test web chat stream từ React

---

Series hoàn tất — 10 bài, 10 packages

@flare-agent/types         ← interfaces
@flare-agent/providers     ← Groq, WorkersAI, Ollama
@flare-agent/memory        ← KV, D1, Vectorize
@flare-agent/core          ← Agent Loop, Tool Registry
@flare-agent/workflow      ← Graph-based Workflow
@flare-agent/multi-agent   ← Agent Network, Handoff
@flare-agent/observability ← Tracing, Debugging
@flare-agent/channels      ← Telegram, Web Chat ← mới

Agent của bạn giờ có thể nhận message từ Telegram và trả lời real-time qua SSE — tất cả chạy trên Cloudflare Workers, không cần server.