LINE Bot 開發教學 2026：Node.js 與 Python 實作

使用者 ── 訊息 ──▶ LINE 平台 ── Webhook（POST + x-line-signature）──▶ 你的伺服器
                                                                       │
使用者 ◀── 回覆 ── LINE 平台 ◀──── Messaging API（回覆或推播）────────┘

mkdir line-bot && cd line-bot
npm init -y && npm pkg set type=module
npm install @line/bot-sdk express
npm install -D typescript tsx @types/express @types/node

{
  "scripts": {
    "dev": "tsx watch --env-file=.env src/index.ts",
    "build": "tsc",
    "start": "node dist/index.js"
  }
}

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true,
    "outDir": "dist",
    "skipLibCheck": true
  },
  "include": ["src"]
}

mkdir line-bot && cd line-bot
python -m venv .venv
source .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install line-bot-sdk flask

LINE_CHANNEL_SECRET=your_channel_secret
LINE_CHANNEL_ACCESS_TOKEN=your_channel_access_token
PORT=3000

// src/index.ts
import express from "express"
import {
  JSONParseError,
  LineBotClient,
  SignatureValidationFailed,
  middleware,
  webhook,
} from "@line/bot-sdk"

const channelSecret = process.env.LINE_CHANNEL_SECRET
const channelAccessToken = process.env.LINE_CHANNEL_ACCESS_TOKEN
if (!channelSecret || !channelAccessToken) {
  throw new Error("請設定 LINE_CHANNEL_SECRET 與 LINE_CHANNEL_ACCESS_TOKEN")
}

const client = LineBotClient.fromChannelAccessToken({ channelAccessToken })
const app = express()

// LINE 中介軟體要用原始請求內容驗證 x-line-signature，
// 所以必須放在任何 JSON 解析器之前。
app.post("/webhook", middleware({ channelSecret }), (req, res) => {
  res.sendStatus(200) // 先回應 LINE，再處理事件
  const { events } = req.body as webhook.CallbackRequest
  for (const event of events) {
    handleEvent(event).catch((err) => console.error("事件處理失敗", err))
  }
})

app.use(express.json()) // 其他路由照常解析 JSON

async function handleEvent(event: webhook.Event) {
  if (event.type !== "message" || event.message.type !== "text" || !event.replyToken) return

  await client.replyMessage({
    replyToken: event.replyToken,
    messages: [{ type: "text", text: `你說：${event.message.text}` }],
  })
}

// 中介軟體只會把錯誤交給 next()；沒有這個處理器，Express 會回 500。
app.use((err: unknown, _req: express.Request, res: express.Response, next: express.NextFunction) => {
  if (err instanceof SignatureValidationFailed) {
    res.status(401).send("Invalid signature")
    return
  }
  if (err instanceof JSONParseError) {
    res.status(400).send("Invalid JSON")
    return
  }
  next(err)
})

const port = Number(process.env.PORT ?? 3000)
app.listen(port, () => console.log(`LINE Bot 在連接埠 ${port} 等待中`))

# app.py
import os

from flask import Flask, abort, request
from linebot.v3 import WebhookHandler
from linebot.v3.exceptions import InvalidSignatureError
from linebot.v3.messaging import (
    ApiClient,
    Configuration,
    MessagingApi,
    ReplyMessageRequest,
    TextMessage,
)
from linebot.v3.webhooks import MessageEvent, TextMessageContent

app = Flask(__name__)
configuration = Configuration(access_token=os.environ["LINE_CHANNEL_ACCESS_TOKEN"])
handler = WebhookHandler(os.environ["LINE_CHANNEL_SECRET"])


@app.post("/webhook")
def webhook():
    signature = request.headers.get("X-Line-Signature", "")
    body = request.get_data(as_text=True)  # 解析 JSON 之前的原始內容
    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        abort(400)
    return "OK"


@handler.add(MessageEvent, message=TextMessageContent)
def handle_text(event):
    with ApiClient(configuration) as api_client:
        MessagingApi(api_client).reply_message(
            ReplyMessageRequest(
                reply_token=event.reply_token,
                messages=[TextMessage(text=f"你說：{event.message.text}")],
            )
        )


if __name__ == "__main__":
    app.run(port=int(os.environ.get("PORT", 3000)))

import { messagingApi } from "@line/bot-sdk"

const text: messagingApi.TextMessage = {
  type: "text",
  text: "您好！請問今天需要什麼服務？", // 最多 5,000 個字元
}

const image: messagingApi.ImageMessage = {
  type: "image",
  originalContentUrl: "https://example.com/price-list.jpg", // HTTPS，JPEG 或 PNG，最大 10 MB
  previewImageUrl: "https://example.com/price-list-preview.jpg", // 最大 1 MB
}

const card: messagingApi.FlexMessage = {
  type: "flex",
  altText: "剪髮 NT$600", // 顯示在通知與聊天列表
  contents: {
    type: "bubble",
    hero: {
      type: "image",
      url: "https://example.com/haircut.jpg",
      size: "full",
      aspectRatio: "20:13",
      aspectMode: "cover",
    },
    body: {
      type: "box",
      layout: "vertical",
      contents: [
        { type: "text", text: "剪髮", weight: "bold", size: "xl" },
        { type: "text", text: "NT$600・45 分鐘", color: "#06C755" },
      ],
    },
    footer: {
      type: "box",
      layout: "vertical",
      contents: [
        {
          type: "button",
          style: "primary",
          color: "#06C755",
          action: { type: "postback", label: "立即預約", data: "action=service&id=cut", displayText: "我想預約剪髮" },
        },
      ],
    },
  },
}

const menu: messagingApi.TextMessage = {
  type: "text",
  text: "請選擇您需要的服務：",
  quickReply: {
    items: [
      // 最多 13 個按鈕
      { type: "action", action: { type: "postback", label: "預約", data: "action=start_booking", displayText: "預約" } },
      { type: "action", action: { type: "message", label: "營業時間", text: "營業時間" } },
      { type: "action", action: { type: "location", label: "傳送我的位置" } },
    ],
  },
}

type Session =
  | { step: "idle" }
  | { step: "pick_date"; service: string }
  | { step: "confirm"; service: string; date: string }
  | { step: "human" } // 店員已接手這段對話

// 教學用的記憶體儲存：重新啟動就會消失，也無法在多個執行個體間共用。
// 正式環境請改用 Redis 或資料庫。
const sessions = new Map<string, Session>()

const SERVICES: Record<string, string> = { cut: "剪髮", color: "染髮" }

async function handleEvent(event: webhook.Event) {
  // 本教學只處理一對一聊天
  if (event.source?.type !== "user" || !event.source.userId) return
  const userId = event.source.userId

  if (event.type === "follow") {
    await reply(event.replyToken, [menuMessage("歡迎！謝謝您加入好友。")])
    return
  }

  const session = sessions.get(userId) ?? { step: "idle" }
  if (session.step === "human") return // 由店員在 LINE Official Account Manager 回覆

  if (event.type === "postback" && event.replyToken) {
    await onPostback(userId, session, event.postback, event.replyToken)
    return
  }

  if (event.type === "message" && event.replyToken) {
    if (event.message.type !== "text") {
      await reply(event.replyToken, [menuMessage("謝謝！目前只能讀取文字訊息。")])
      return
    }
    const text = event.message.text.trim()
    if (["staff", "真人客服"].includes(text.toLowerCase())) {
      sessions.set(userId, { step: "human" })
      await reply(event.replyToken, [
        { type: "text", text: "好的！店員會在營業時間（10:00–20:00）於這個聊天室回覆您。" },
      ])
      return
    }
    await reply(event.replyToken, [menuMessage("請從下方選擇服務。")]) // AI 章節會取代這一行
  }
}

async function onPostback(userId: string, session: Session, postback: webhook.PostbackContent, replyToken: string) {
  const data = new URLSearchParams(postback.data)
  const action = data.get("action")

  if (action === "start_booking") {
    await reply(replyToken, [{
      type: "text",
      text: "請問要預約哪一項服務？",
      quickReply: {
        items: Object.entries(SERVICES).map(([id, label]) => ({
          type: "action" as const,
          action: { type: "postback" as const, label, data: `action=service&id=${id}`, displayText: label },
        })),
      },
    }])
    return
  }

  if (action === "service") {
    const service = SERVICES[data.get("id") ?? ""]
    if (!service) return
    sessions.set(userId, { step: "pick_date", service })
    await reply(replyToken, [{
      type: "text",
      text: `${service}：請選擇方便的日期。`,
      quickReply: {
        items: [{ type: "action", action: { type: "datetimepicker", label: "選擇日期", data: "action=date", mode: "date" } }],
      },
    }])
    return
  }

  if (action === "date" && session.step === "pick_date" && postback.params?.date) {
    const next: Session = { step: "confirm", service: session.service, date: postback.params.date }
    sessions.set(userId, next)
    await reply(replyToken, [{
      type: "text",
      text: `${next.date} ${next.service}，要送出預約申請嗎？`,
      quickReply: {
        items: [
          { type: "action", action: { type: "postback", label: "確認送出", data: "action=confirm", displayText: "確認送出" } },
          { type: "action", action: { type: "postback", label: "取消", data: "action=cancel", displayText: "取消" } },
        ],
      },
    }])
    return
  }

  if (action === "confirm" && session.step === "confirm") {
    sessions.delete(userId)
    // 在這裡寫入預約系統或通知店員
    await reply(replyToken, [
      { type: "text", text: `已收到申請：${session.date} ${session.service}。確切時間會在這個聊天室跟您確認。` },
    ])
    return
  }

  if (action === "cancel") {
    sessions.delete(userId)
    await reply(replyToken, [menuMessage("已取消。還有什麼需要幫忙的嗎？")])
  }
}

function menuMessage(text: string): messagingApi.TextMessage {
  return {
    type: "text",
    text,
    quickReply: {
      items: [
        { type: "action", action: { type: "postback", label: "預約", data: "action=start_booking", displayText: "預約" } },
        { type: "action", action: { type: "message", label: "找真人客服", text: "真人客服" } },
      ],
    },
  }
}

async function reply(replyToken: string, messages: messagingApi.Message[]) {
  await client.replyMessage({ replyToken, messages })
}

from urllib.parse import parse_qsl

from linebot.v3.webhooks import PostbackEvent


@handler.add(PostbackEvent)
def handle_postback(event):
    data = dict(parse_qsl(event.postback.data))  # "action=service&id=cut"
    if data.get("action") == "date" and event.postback.params:
        text = f"選擇的日期：{event.postback.params['date']}"
    else:
        text = f"動作：{data.get('action')}"
    with ApiClient(configuration) as api_client:
        MessagingApi(api_client).reply_message(
            ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text=text)])
        )

// src/ai.ts
import { GoogleGenAI } from "@google/genai"

const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY })
const model = process.env.GEMINI_MODEL // Google 文件中目前提供的模型名稱
if (!model) throw new Error("請設定 GEMINI_MODEL")

// 只放你願意讓 Bot 原封不動說出的資訊
const FAQ = `
營業時間：10:00–20:00，每週三公休。
地址：台北市大安區忠孝東路四段 100 號（捷運忠孝敦化站步行 3 分鐘）。
剪髮：NT$600，約 45 分鐘。染髮：NT$2,000 起，約 2 小時。
預約：點選單中的「預約」。最晚可在 3 小時前更改。
`

export async function answerFromFaq(question: string): Promise<string | null> {
  const response = await ai.models.generateContent({
    model: model!,
    contents: question,
    config: {
      systemInstruction:
        "你是美髮店的客服助理，只能根據下方 FAQ 回答顧客問題。" +
        "如果 FAQ 裡沒有答案，只回覆 NO_ANSWER。" +
        "用顧客使用的語言回答，最多 3 句，簡短清楚。不可以自行編造價格、時間或優惠。\n\n" +
        `FAQ:\n${FAQ}`,
      temperature: 0.2,
      maxOutputTokens: 300,
      abortSignal: AbortSignal.timeout(15_000), // 確保在回覆權杖時限內完成
    },
  })
  const text = response.text?.trim()
  if (!text || text.includes("NO_ANSWER")) return null
  return text.slice(0, 5000) // LINE 文字訊息最多 5,000 個字元
}

// src/index.ts 開頭（ES 模組需要加上 .js 副檔名）
import { answerFromFaq } from "./ai.js"

// 在 handleEvent 中，不再對每則自由輸入都回選單：
await client.showLoadingAnimation({ chatId: userId, loadingSeconds: 20 })
const answer = await answerFromFaq(text).catch(() => null)
await reply(event.replyToken, [
  answer
    ? { type: "text", text: answer }
    : menuMessage("這個問題我無法確定，請點「找真人客服」由店員為您說明。"),
])

FROM node:22-slim AS build
WORKDIR /app
COPY package*.json tsconfig.json ./
RUN npm ci
COPY src ./src
RUN npm run build

FROM node:22-slim
WORKDIR /app
ENV NODE_ENV=production
COPY package*.json ./
RUN npm ci --omit=dev
COPY --from=build /app/dist ./dist
EXPOSE 3000
CMD ["node", "dist/index.js"]

import { HTTPFetchError } from "@line/bot-sdk"

const seen = new Map<string, number>() // webhookEventId -> 收到時間；多個執行個體請改用 Redis（SET NX + TTL）

function firstDelivery(event: webhook.Event): boolean {
  const now = Date.now()
  for (const [id, at] of seen) if (now - at > 30 * 60_000) seen.delete(id)
  if (seen.has(event.webhookEventId)) return false
  seen.set(event.webhookEventId, now)
  return true
}

function log(level: "info" | "error", msg: string, data: Record<string, unknown> = {}) {
  console.log(JSON.stringify({ time: new Date().toISOString(), level, msg, ...data }))
}

app.post("/webhook", middleware({ channelSecret }), (req, res) => {
  res.sendStatus(200)
  const { events } = req.body as webhook.CallbackRequest
  for (const event of events) {
    if (!firstDelivery(event)) continue // 同一事件收到兩次
    handleEvent(event).catch((err) => {
      if (err instanceof HTTPFetchError) {
        // 429：速率限制或每月上限；400「Invalid reply token」：已使用或逾時
        log("error", "LINE API error", { status: err.status, body: err.body, eventId: event.webhookEventId })
      } else {
        log("error", "event failed", { error: String(err), eventId: event.webhookEventId })
      }
    })
  }
})