アマゾン ウェブ サービス ジャパン (有志)
👬

LINEからAgentCoreを呼び出して、生活に溶け込ませよう

Shuichi Matsumoto
に公開
AWS
LINE
Amazon Bedrock
tech

はじめに

先日出張に行く機会がありまして、大したことじゃないけど準備で色々調べたんですよね。「天気はどうだろう？」「余裕を持って到着するには何時の電車に乗ればいいかな？」「夕飯何食べよう？」とかとか。
それぞれ異なるスマホアプリで確認したんだけど、これってAIエージェントに任せたいシナリオでありそうですよね。
ところがこう、ふとAIエージェントを使いたい時、AIエージェントを使うためのアプリってまだ日常生活に溶け込んでいないと思うんですよね。
専用のアプリ開いたり、自分で作ったWebアプリにアクセスしたり…実際には全く大したことない労力なんだけど、それが利用の促進を妨げているかもしれません。
AIエージェントに限らず、こういったアクセシビリティに関する課題はちょっと調べてみるだけでもいろんな研究がありましたが、まぁ私レベルの脳みそだと「普段から慣れているメッセージングアプリだったら、使いやすいエージェントアプリになるんじゃない？」と安直な発想に至り、LINEのDeveloper向けの機能を利用して作ってみることにしたのでした。

  • 全体構成イメージ

ここからは図中の番号に従って、実装を解説していきます。

1. LINE Developers

builders.flashの記事を参考に設定を進めようと思いましたが、2024年9月頃に仕様が変わったらしく、そのままではいけなかったので、素直にLINEの公式ページのガイドに従ってセットアップを進めました。

https://developers.line.biz/ja/

特段難しいことはなく、普通にLINEのページをポチポチしていけばLINE Official Account Managerというページに辿り着き、LINEアカウントを開設することができました。
どうやらMessaging APIという機能がユーザとのメッセージ送受信機能を実装しており、webhookを設定することでユーザの発言を通知できる仕組みのようです。
他にも細かい設定はたくさんありますが、今回は最低限のメッセージ送受信だけできれば良いので、スルーします。ちなみに無償版だと月200通までメッセージを送信できるようです。

  • とりあえず一方的に色々送っている様子

私はマーベル映画が大好きなので、エージェントにはJARVISと名付けました。

2. Amazon Bedrock AgentCore

さて、LINEというインターフェースは一旦忘れて、コンシェルジュ的に応対するAIエージェントを作っていきましょう。
エージェントに与えるツールは今後拡張したいですが、まずはWeb検索と現在時刻を取得するツールだけ与えることにしました。
Web検索については同僚の記事が大変わかりやすかったので、そのまま使わせてもらいました。
Grazie, Go-san!

https://zenn.dev/aws_japan/articles/1b29bc6b8de3ca

記事と違うのは最終的にAgentCoreにデプロイした点ですが、これも通常の手順でOKでした。注意点としては、AIエージェントのIAMロールにCognitoのアクセス権限を与えることぐらいですね。
このデプロイ手順やIaCも載せたかったですが、メモもとらず手作業でやっちゃったのでStrandsのコードだけ置いておきます。参考となるコードを公開してくださっている皆様、ありがとうございます！

line-agent.py
import os
import boto3
import requests
from dotenv import load_dotenv
from strands.tools.mcp.mcp_client import MCPClient
from strands import Agent, tool
from mcp.client.streamable_http import streamablehttp_client
from datetime import datetime
import pytz
from bedrock_agentcore.runtime import BedrockAgentCoreApp

load_dotenv()

user_pool_id="xxxxx"
client_id="xxxxx"
gateway_url="https://xxxxx.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp"

system_prompt = """
あなたは私専属のコンシェルジュです。私が普段利用しているメッセージングアプリを通じてメッセージを受信するので、与えられたツールを活用しながら私の要望に可能な限り答えてください。
最適な応対をするため、常に私の個人情報を考慮して発言する必要があります。

# 個人情報・プロファイル
＜知ってて欲しいことを書く。最寄駅とか。本当はAgentCore Memoryを使いたい。＞

# コミュニケーション設定
* 口調: 映画アイアンマンに登場するJ.A.R.V.I.S.のように、丁寧で理想的な執事でジョークも交える
* 返答スタイル: 要点を整理して分かりやすく、必要に応じて詳細も提供
* プライバシー: 家族の個人情報は慎重に取り扱い

# 注意事項
* 医療・法的な専門的判断は提供せず、参考情報にとどめてください。
* 直接的な金銭取引は行わないでください。
* 第三者への個人情報開示は行わないでください。
* あなたの知識だけで作成した回答は不十分です。提供されているツールを積極的に利用し、必ず回答の根拠を示します。
* できるだけ多くの情報を収集し、多角的な情報から判断を下します。
* あなたは日時を誤解する可能性があります。天気や路線情報など、時期性を伴う質問には必ず明確な日時情報も含めてください。
* スマートフォンで表示するため、メッセージは可能な限り短文であることが求められます。丁寧さは維持しながら、聞かれたこと以外の余計な発言は慎む必要があります。
    * 具体的には、名前を呼びかけたり、気の利いた締めくくりの文章は不要です。
* メッセージングアプリはマークダウン記法に対応していません。シンプルテキストのみで発言してください。

**常に「ご主人様にとって最適な選択は何か」を考えて行動してください。**
"""

@tool
def get_current_time_with_timezone(timezone_str="Asia/Tokyo"):
    """指定されたタイムゾーンの現在時刻を返却する関数"""
    tz = pytz.timezone(timezone_str)
    return datetime.now(tz)

client = boto3.client("cognito-idp", region_name='us-east-1')
client_secret = client.describe_user_pool_client(
    UserPoolId=user_pool_id,
    ClientId=client_id
)["UserPoolClient"].get("ClientSecret")
domain = client.describe_user_pool(UserPoolId=user_pool_id)['UserPool'].get('Domain')
token_url = f"https://{domain}.auth.{client._client_config.region_name}.amazoncognito.com/oauth2/token"
response = requests.post(
    token_url,
    data=f"grant_type=client_credentials&client_id={client_id}&client_secret={client_secret}",
    headers={'Content-Type': 'application/x-www-form-urlencoded'}
)
response.raise_for_status()
bearer_token = response.json()["access_token"]

mcp_client = MCPClient(lambda: streamablehttp_client(gateway_url, headers={"Authorization": f"Bearer {bearer_token}"}))

with mcp_client:
    tools = mcp_client.list_tools_sync()

agent = Agent(
    model = "global.anthropic.claude-sonnet-4-5-20250929-v1:0",
    system_prompt=system_prompt,
    tools = [tools, get_current_time_with_timezone]
)

app = BedrockAgentCoreApp()

@app.entrypoint
def invoke_agent(payload, context):
    prompt = payload.get("prompt")
    with mcp_client:
        return {"result": agent(prompt).message}

app.run()

さて、これをデプロイして、サンドボックスで試した結果がこちらです。

  • AWSのマネージメントコンソールから試しているスクショ

伏せている部分には私の居住地域が入っています、いい感じの受け答えですね。初期バージョンのAIエージェントはこんなところでしょう。

3. API Gateway + Lambda

まずはLINEからどんな情報が送られてくるのか見てみたいですよね。
LINEからwebhookイベントが飛んでくる先が必要なので、API Gateway + Lambdaを用意し、LINE Developersの設定画面からwebhook URLにAPI Gatewayのエンドポイントを指定します。

さて、それではお友達になったLINEからJARVISくんにメッセージを送ってみましょう！
webhookイベントを受信して中身を見てみると、bodyの中にメッセージやスタンプの情報が詰まっているようです。テキストチャットとスタンプを送信した場合をそれぞれ以下のような内容でした。

テキストメッセージを送ったとき
{
    "destination": "xxxxx",
    "events": [
        {
            "type": "message",
            "message": {
                "type": "text",
                "id": "xxxxx",
                "quoteToken": "xxxxx",
                "text": "中身を見てみる"
            },
            "webhookEventId": "xxxxx",
            "deliveryContext": {
                "isRedelivery": false
            },
            "timestamp": 1760854663771,
            "source": {
                "type": "user",
                "userId": "xxxxx"
            },
            "replyToken": "xxxxx",
            "mode": "active"
        }
    ]
}
スタンプを送ったとき
{
    "destination": "xxxxx",
    "events": [
        {
            "type": "message",
            "message": {
                "type": "sticker",
                "id": "xxxxx",
                "quoteToken": "xxxxx",
                "stickerId": "164075050",
                "packageId": "6959173",
                "stickerResourceType": "STATIC",
                "keywords": [
                    "Bear",
                    "Celebrate",
                    "Thanks"
                ]
            },
            "webhookEventId": "xxxxx",
            "deliveryContext": {
                "isRedelivery": false
            },
            "timestamp": 1760854728117,
            "source": {
                "type": "user",
                "userId": "xxxxx"
            },
            "replyToken": "xxxxx",
            "mode": "active"
        }
    ]
}

へえ〜スタンプにはキーワードが設定されているんですね、これを使えばスタンプに対してもリアクションすることができそう。

これでwebhookイベントからユーザのインプットを抽出することもできるようになったので、あとはLambdaの中で、AgentCoreへプロンプトを送る→推論結果をユーザへ返す、という2 stepを実装するだけです。

だけ、って言ったんですが、LINEから提供されているpythonのSDKをLambda Layerとして利用できるようにする部分だけ手こずりました。
Macでzipファイルを用意するにはCPUアーキテクチャを考慮する必要があることはわかっていたんですが、なぜか最新のv3をとってこれず、致し方なくCloudShellで以下のコマンドでパッケージを用意しました。

pip install --target ./python --implementation cp --python-version 3.11 --upgrade --only-binary=:all: -r requirements.txt
requirements.txt（あっているかはわかりません、すみません。とりあえず動きます。）
line-bot-sdk >= 3.0.0
pydantic_core

あとはこのあたりの記事を参考にさせていただきながらコードを書いて、最終的にClaudeにキレイにしてもらいました。
https://github.com/line/line-bot-sdk-python/blob/master/linebot/v3/messaging/docs/MessagingApi.md#push_message
https://zenn.dev/yu_yu/articles/402281ccb6a3f2

Lambdaコード
import json
import logging
import boto3
import uuid
import linebot.v3.messaging
from linebot.v3.messaging.models.push_message_request import PushMessageRequest
from linebot.v3.messaging.models.push_message_response import PushMessageResponse
from linebot.v3.messaging.rest import ApiException
from pprint import pprint

# ログ設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 定数定義
BEDROCK_REGION = 'us-east-1'
AGENT_RUNTIME_ARN = '利用するエージェントのARN'
LINE_ACCESS_TOKEN = "LINE Developersの管理画面からチャネルアクセストークンを取得"
LINE_USER_ID = "送り先のID。私はwebhookのテストをしていた時に自分のIDを特定しました。"
LINE_API_HOST = "https://api.line.me"


def lambda_handler(event, context):
    """
    Lambda関数のメインハンドラー
    LINEからのメッセージを受信し、Bedrock AgentCoreで処理後、応答を送信する
    """
    logger.info("受信イベント: %s", json.dumps(event))
    
    # リクエスト情報をログ出力
    _log_request_info(event)
    
    # LINEメッセージ情報を抽出
    user_message, user_id = _extract_line_message_info(event)
    
    # Bedrock AgentCoreで応答を生成
    agent_response = _invoke_bedrock_agent(user_message)
    
    # LINE経由で応答を送信
    _send_line_message(agent_response)
    
    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        },
        'body': json.dumps({'message': 'OK'})
    }


def _log_request_info(event):
    """
    リクエスト情報をログに出力する
    """
    # リクエストボディを取得
    body = event.get('body', '')
    if body:
        try:
            body_json = json.loads(body)
            logger.info("リクエストボディ: %s", json.dumps(body_json))
        except json.JSONDecodeError:
            logger.info("リクエストボディ（生データ）: %s", body)
    
    # ヘッダー情報をログ出力
    headers = event.get('headers', {})
    logger.info("ヘッダー: %s", json.dumps(headers))


def _extract_line_message_info(event):
    """
    LINEイベントからメッセージとユーザーIDを抽出する
    """
    body = event.get('body', '')
    body_json = json.loads(body)
    
    user_message = body_json['events'][0]['message']['text']
    user_id = body_json['events'][0]['source']['userId']
    
    return user_message, user_id


def _invoke_bedrock_agent(user_message):
    """
    Bedrock AgentCoreを呼び出してユーザーメッセージに対する応答を生成する
    """
    # セッションIDを生成
    session_id = str(uuid.uuid4())
    
    # Bedrock AgentCoreクライアントを初期化
    agent_core_client = boto3.client('bedrock-agentcore', region_name=BEDROCK_REGION)
    
    # ペイロードを準備
    payload = json.dumps({
        "prompt": user_message
    })
    
    # エージェントランタイムを呼び出し
    response = agent_core_client.invoke_agent_runtime(
        agentRuntimeArn=AGENT_RUNTIME_ARN,
        runtimeSessionId=session_id,
        payload=payload,
        qualifier="DEFAULT"
    )
    
    # レスポンスを解析
    response_body = response['response'].read()
    response_data = json.loads(response_body)
    
    logger.info("エージェント応答: %s", response_data)
    
    response_text = response_data['result']['content'][0]['text']
    logger.info("応答テキスト: %s", response_text)
    
    return response_text


def _send_line_message(message_text):
    """
    LINE Messaging APIを使用してメッセージを送信する
    """
    # LINE API設定を構成
    configuration = linebot.v3.messaging.Configuration(
        host=LINE_API_HOST,
        access_token=LINE_ACCESS_TOKEN
    )
    
    # メッセージ辞書を作成
    message_dict = {
        "to": LINE_USER_ID,
        "messages": [
            {"type": "text", "text": message_text}
        ]
    }
    
    # リトライキーを生成
    x_line_retry_key = str(uuid.uuid4())
    logger.info("リトライキー: %s", x_line_retry_key)
    
    # APIクライアントを使用してメッセージを送信
    with linebot.v3.messaging.ApiClient(configuration) as api_client:
        api_instance = linebot.v3.messaging.MessagingApi(api_client)
        push_message_request = linebot.v3.messaging.PushMessageRequest.from_dict(message_dict)
        
        try:
            api_response = api_instance.push_message(
                push_message_request, 
                x_line_retry_key=x_line_retry_key
            )
            logger.info("MessagingApi->push_messageの応答:")
            pprint(api_response)
        except Exception as e:
            logger.error("MessagingApi->push_message呼び出し時の例外: %s", e)

本当は受信したメッセージのparseも公式SDKを使えばよくて、ちゃんとsignatureのチェックもやるべきなんですが、最初にゴリ押しで書いちゃったのでそのままで…手抜きだけど！まずは体験優先で！

4. 使ってみた

全ての環境準備が整ったので、さっそくスマホから色々聞いてみましょう〜！

このように、期待通りの回答をweb検索などを駆使しながら生成してくれました。普通にweb検索する代わりにパッと使えて、思ったよりも便利かもしれない。

おわりに

お試しに作ってみてわかったことは、「ストリーミングがないと待ち時間を長く感じるなあ」ということです。いや、当たり前のことなんで頭では分かってたんですが、改めて感じたというか。特に今回はWeb検索もしているので余計待ち時間を長く感じてしまいますね。
あと、今回は利用しませんでしたが、AgentCore Memoryを利用してもう少しユーザコンテキストが育っていくようにしたいなぁ。

まだまだ工夫の余地という名の課題はあるわけですが、当初の目的だった日常生活に溶け込んだAIエージェントインターフェースの実装は完了したので、しばらく使ってみたいと思います。

実業務でLINEを利用するにはハードルがあると思いますが、個人用途での開発や生成AIの生活への統合を体感する方法としては悪くないなじゃないかな〜と思います、おためしあれ〜。

アマゾン ウェブ サービス ジャパン (有志)
アマゾン ウェブ サービス ジャパン (有志)

この Publication に投稿している記事は、アマゾン ウェブ サービス ジャパン合同会社または Amazon Web Services, Inc. 所属社員による個人の見解であり、所属する組織の公式見解ではありません。参加したい従業員の方は、Sugiyama Suguru までお知らせください。

Discussion