kb-line

name	kb-line
description	LINE Bot開発のナレッジ。Messaging API、Webhook、署名検証、Push Message、グループチャット対応、SSEストリーミング連携等
user-invocable	true
model	sonnet

LINE Bot 開発ナレッジ

LINE Messaging APIを使ったチャットボット開発のベストプラクティス集。

基本アーキテクチャ

API Gateway + Lambda（非同期呼び出し）パターン

LINE User → API Gateway (REST) → Lambda (非同期起動)
                                    ├── LINE署名検証
                                    ├── ビジネスロジック / AI処理
                                    └── Push Message で返信

LINE Webhookは3秒以内のレスポンスを要求するため、API Gatewayで即座に200を返却し、Lambdaを非同期で起動する。

// CDK: API Gateway → Lambda 非同期呼び出し統合
const lambdaIntegration = new apigateway.AwsIntegration({
  service: "lambda",
  path: `2015-03-31/functions/${webhookFn.functionArn}/invocations`,
  integrationHttpMethod: "POST",
  options: {
    credentialsRole: apiGatewayRole,
    requestParameters: {
      "integration.request.header.X-Amz-Invocation-Type": "'Event'",  // 非同期
    },
    requestTemplates: {
      "application/json": `{
  "body": "$util.escapeJavaScript($input.body)",
  "signature": "$input.params('x-line-signature')"
}`,
    },
    integrationResponses: [
      {
        statusCode: "200",
        responseTemplates: {
          "application/json": '{"message": "accepted"}',
        },
      },
    ],
  },
});

Reply Message vs Push Message

方式	制限	用途
Reply Message	replyToken必須（30秒有効）、無料	同期処理向け
Push Message	user_id/group_id指定、月200通（無料枠）	非同期処理向け

非同期Lambda呼び出しの場合、replyTokenが30秒で失効するため Push Messageのみ使用 する。Push Messageは月の無料枠（コミュニケーションプラン: 200通）があるため、SSEストリーミングで細かく送信すると枠を使い切りやすい。

---

LINE署名検証

VTLテンプレートでのraw body受け渡し

API Gatewayの統合リクエストでLINE Webhookのbodyをそのまま渡す際の注意点。

# NG: パース済みJSONを返すため、署名検証に使えない
$input.json('$')

# OK: raw bodyを文字列として渡す
$util.escapeJavaScript($input.body)

Lambda側ではbodyを文字列として受け取る:

body_str = event.get("body", "")
signature = event.get("signature", "")

# LINE署名検証
events = parser.parse(body_str, signature)

---

LINE Bot SDK v3 (Python)

インストール

# requirements.txt
line-bot-sdk

基本的な使い方

from linebot.v3 import WebhookParser
from linebot.v3.exceptions import InvalidSignatureError
from linebot.v3.messaging import (
    ApiClient,
    Configuration,
    MessagingApi,
    PushMessageRequest,
    TextMessage,
)
from linebot.v3.webhooks import MessageEvent, TextMessageContent

# 初期化
parser = WebhookParser(LINE_CHANNEL_SECRET)
line_config = Configuration(access_token=LINE_CHANNEL_ACCESS_TOKEN)

# Push Message送信
def send_push_message(reply_to: str, text: str) -> None:
    if not text.strip():
        return
    with ApiClient(line_config) as api_client:
        api = MessagingApi(api_client)
        api.push_message(
            PushMessageRequest(
                to=reply_to,
                messages=[TextMessage(text=text.strip())],
            )
        )

テキスト上限

LINE Push Messageのテキスト上限は 5000文字。長いメッセージは切り詰める:

send_push_message(reply_to, text.strip()[:5000])

---

グループチャット対応

メンション検出

グループチャットではBot宛メンション時のみ処理する:

def _is_bot_mentioned(message: TextMessageContent) -> bool:
    if not message.mention:
        return False
    return any(
        getattr(m, "is_self", False) for m in message.mention.mentionees
    )

メンション文字列の除去

@Bot名 をメッセージテキストから除去してからビジネスロジックに渡す:

def _strip_bot_mention(message: TextMessageContent) -> str:
    text = message.text
    if not message.mention:
        return text.strip()
    # index が大きい方から除去（位置ずれ防止）
    mentionees = sorted(
        (m for m in message.mention.mentionees if getattr(m, "is_self", False)),
        key=lambda m: m.index,
        reverse=True,
    )
    for m in mentionees:
        text = text[:m.index] + text[m.index + m.length :]
    return text.strip()

送信先の判定

source = line_event.source
is_group_chat = source.type in ("group", "room")

# 送信先: グループならgroup_id/room_id、1対1ならuser_id
reply_to = (
    getattr(source, "group_id", None)
    or getattr(source, "room_id", None)
    or source.user_id
)

LINE Official Accountの設定

グループチャットで使うには:

1. LINE Official Account Manager → 設定 → アカウント設定
2. 「グループトーク・複数人トークへの参加を許可する」をオン
3. グループにBotを招待

---

Push Message のレート制限と月間上限

レート制限（秒単位）

エンドポイント	レート制限
Push Message	2,000 req/s（チャネル単位）
Loading Animation	100 req/s
Multicast	200 req/s

• レート制限はチャネル単位で適用
• 429レスポンスに Retry-After ヘッダーは含まれない
• LINE公式は429を「リトライすべきでない4xx」に分類 → リトライではなくスロットリング（頻度低減）が正しい対処

月間メッセージ上限

無料プラン（コミュニケーションプラン）は 月200通 が上限。超過すると429エラー:

{"message": "You have reached your monthly limit."}

SSEストリーミングで contentBlockStop ごとにPush Messageを送ると、1回の対話で多数の通数を消費してしまう。通数節約のために最終ブロックのみ送信する方式を推奨。

---

SSE → Push Message 変換パターン

AIエージェント（AgentCore等）のSSEストリーミングをLINE Push Messageに変換するパターン。

方式A: 最終ブロックのみ送信（通数節約・推奨）

ツールステータスだけリアルタイム送信し、テキストは最終ブロックのみ1通で送信する方式。月間メッセージ上限の節約に有効。

text_buffer = ""
last_text_block = ""

# contentBlockDelta → バッファに蓄積
text_buffer += text

# contentBlockStop → 最終ブロック候補として保持（送信しない）
last_text_block = text_buffer.strip()
text_buffer = ""

# ツール開始 → バッファを破棄（途中応答は不要）してステータス送信
text_buffer = ""
throttled_send(status_text)

# SSE完了後 → 最終ブロックのみ送信
send_push_message(reply_to, last_text_block[:5000])

ツールを使わない場合（AIからの質問返し等）も、唯一のテキストブロックが最終ブロックとなるため正常に動作する。

方式B: contentBlockStop ごとに送信（ストリーミング体験重視）

text_buffer = ""

def flush_text_buffer():
    nonlocal text_buffer
    if text_buffer.strip():
        send_push_message(reply_to, text_buffer.strip()[:5000])
        text_buffer = ""

通数を多く消費するため、有料プランでのみ推奨。

SSEイベントとLINEメッセージの対応

SSEイベント                              方式A                    方式B
──────────────────────────────────    ─────────────────────    ─────────────────────
contentBlockDelta(text: "...")      → バッファに蓄積          → バッファに蓄積
contentBlockStop                    → last_text_blockに保持   → flush → Push Message
contentBlockStart(toolUse: name)    → ステータス送信          → flush + ステータス送信
[DONE]                              → last_text_block送信     → 処理完了

送信スロットリング

429エラー防止のため、Push Message送信に最低1秒の間隔を設ける:

import time

last_send_time = 0.0
MIN_SEND_INTERVAL = 1.0

def throttled_send(text: str) -> None:
    nonlocal last_send_time
    elapsed = time.time() - last_send_time
    if elapsed < MIN_SEND_INTERVAL:
        time.sleep(MIN_SEND_INTERVAL - elapsed)
    send_push_message(reply_to, text)
    last_send_time = time.time()

非同期Lambda呼び出しのため、time.sleepでLambda実行時間が伸びてもLINEのレスポンスには影響しない。

UXのコツ

• ユーザーからメッセージを受けたら、エージェント呼び出し前に「考えています...」を即座に返す（体感レスポンス向上）
• ツール使用中は日本語のステータスメッセージを送る（何をしているか可視化）
• Markdownは使わない（LINEではレンダリングされない）
  • NG: **太字**、# 見出し、[リンク](URL)
  • OK: 「・」箇条書き、【】強調、改行区切り

---

LINE ユーザーIDの注意

LINE のユーザーIDは U + 32桁の英数字（例: U56b15391befe93872af9cd62f40fb7aa）。LINE の表示名（ニックネーム）や LINE ID（検索用ID）とは全く別物。

ユーザーIDの確認方法:

• Lambda の CloudWatch Logs に User Uxxxxxxxx の形式で出力される
• 各ユーザーに1回メッセージを送ってもらい、ログからIDを収集する

---

環境変数

変数名	用途
LINE_CHANNEL_SECRET	Webhook署名検証
LINE_CHANNEL_ACCESS_TOKEN	Push Message送信

LINE Channel セットアップ手順

1. LINE Official Account Manager でアカウント作成
2. Messaging API を有効化
3. LINE Developers Console で Channel Secret / Channel Access Token を取得
4. CDKデプロイ後、Webhook URLを設定
5. Auto-reply messages をオフにする

---

参考リンク

• LINE Messaging API ドキュメント
• line-bot-sdk-python GitHub
• LINE Official Account Manager
• LINE Developers Console