Bedrock AgentCore で作った居酒屋検索AIエージェント「シンバシくん」をLINE公式アカウントに登録し Messaging API でチャットボットにしてみた

"""
LINE Bot Webhook Handler for シンバシくん
AWS Lambda + Bedrock AgentCore
"""
import json
import os
import hashlib
import hmac
import base64
import urllib.request
import boto3

# 環境変数
LINE_CHANNEL_SECRET = os.environ.get('LINE_CHANNEL_SECRET')
LINE_CHANNEL_ACCESS_TOKEN = os.environ.get('LINE_CHANNEL_ACCESS_TOKEN')
AGENTCORE_RUNTIME_ARN = os.environ.get('AGENTCORE_RUNTIME_ARN')

# AgentCore クライアント（Lambda起動時に1回だけ初期化）
agentcore_client = boto3.client('bedrock-agentcore', region_name='ap-northeast-1')

def lambda_handler(event, context):
    """
    LINE Webhook エントリーポイント

    LINEプラットフォームからのWebhookリクエストを受け取り、
    メッセージイベントを処理してAgentCoreに問い合わせる
    """
    body = event.get('body', '')
    signature = event.get('headers', {}).get('x-line-signature', '')

    # LINE署名検証（なりすまし防止）
    if not _verify_signature(body, signature):
        return {'statusCode': 403, 'body': 'Invalid signature'}

    # Webhookイベントを処理
    for evt in json.loads(body).get('events', []):
        if evt.get('type') == 'message' and evt['message'].get('type') == 'text':
            _handle_text_message(evt)

    return {'statusCode': 200, 'body': 'OK'}

def _verify_signature(body: str, signature: str) -> bool:
    """
    LINE署名を検証

    LINEプラットフォームからのリクエストであることを確認する
    https://developers.line.biz/ja/docs/messaging-api/receiving-messages/#verifying-signatures
    """
    if not LINE_CHANNEL_SECRET or not signature:
        return False

    expected = base64.b64encode(
        hmac.new(
            LINE_CHANNEL_SECRET.encode('utf-8'),
            body.encode('utf-8'),
            hashlib.sha256
        ).digest()
    ).decode('utf-8')

    return hmac.compare_digest(signature, expected)

def _handle_text_message(event: dict) -> None:
    """テキストメッセージを処理してAgentCoreに問い合わせ"""
    reply_token = event['replyToken']
    user_message = event['message']['text']
    user_id = event['source'].get('userId', 'anonymous')

    # セッションID（AgentCoreは33文字以上必要）
    session_id = f"line-{user_id}-0000000000000000"

    try:
        reply_text = _invoke_agentcore(user_message, session_id)
    except Exception as e:
        print(f"AgentCore error: {e}")
        reply_text = "すみません、エラーが発生しました。もう一度お試しください🙇"

    _reply_to_line(reply_token, reply_text)

def _invoke_agentcore(message: str, session_id: str) -> str:
    """
    AgentCore Runtime を呼び出し

    シンバシくん（居酒屋検索エージェント）にメッセージを送信し、
    検索結果を取得する
    """
    response = agentcore_client.invoke_agent_runtime(
        agentRuntimeArn=AGENTCORE_RUNTIME_ARN,
        runtimeSessionId=session_id,
        payload=json.dumps({"prompt": message}).encode('utf-8'),
        contentType='application/json',
        accept='application/json'
    )

    # StreamingBodyからレスポンスを読み取り
    streaming_body = response.get('response')
    if not streaming_body:
        return "検索結果が見つかりませんでした。"

    result = json.loads(streaming_body.read().decode('utf-8'))
    return result.get('result', '検索結果が見つかりませんでした。')

def _reply_to_line(reply_token: str, text: str) -> None:
    """
    LINEに返信

    Reply APIを使用してユーザーにメッセージを送信する
    https://developers.line.biz/ja/docs/messaging-api/sending-messages/#reply-messages
    """
    # LINEの文字数制限（5000文字）
    if len(text) > 5000:
        text = text[:4997] + "..."

    req = urllib.request.Request(
        'https://api.line.me/v2/bot/message/reply',
        data=json.dumps({
            'replyToken': reply_token,
            'messages': [{'type': 'text', 'text': text}]
        }).encode('utf-8'),
        headers={
            'Content-Type': 'application/json',
            'Authorization': f'Bearer {LINE_CHANNEL_ACCESS_TOKEN}'
        },
        method='POST'
    )

    try:
        with urllib.request.urlopen(req) as res:
            print(f"LINE reply: {res.status}")
    except urllib.error.HTTPError as e:
        print(f"LINE reply failed: {e.code} {e.reason}")

lambda_handler()        # エントリーポイント
  ├── _verify_signature()   # LINE署名検証
  ├── _handle_text_message() # メッセージ処理
  │     ├── _invoke_agentcore()  # AgentCore呼び出し
  │     └── _reply_to_line()     # LINE返信

agentcore_client = boto3.client('bedrock-agentcore', region_name='ap-northeast-1')

pythondef lambda_handler(event, context):

def _verify_signature(body: str, signature: str) -> bool: