はじめに
以前以下の記事にて、LINEのチャットボットを作成していました。
今回は、AgentCore Runtimeにデプロイしたエージェントに対して、対話したいなあと思い作ってみました。
今回作成したものは以下のリポジトリにあるので、よければ参考にしてください！
構成
構成は前回作成したものに、AgentCore Runtimeに接続してあげるようにしてあげるだけです。
せっかくなので今回は新規でチャンネル作成してみました。
構築手順
LINE Developersにログインして、新規チャンネル開設
以下のサイトにログインして、新規のチャンネルを開設する必要があります。
以下で「Messaging API」を選択します。
ですが、LINE Developersコンソールからチャンネルは作れなくなったようです。
現在では、以下のサイトから作成できるようです。
Developersコンソールで各種設定をする
チャンネルの開設ができたら、Developersコンソールから各種設定を行います。
今回私は以下の「Bedrock AgenrCore」という名前で作成しましたので、こちらを選択します。
Messaging APIの設定
作成したチャンネルの「Messaging API」設定より、
Webhookの利用を有効化します。また、Webhook URLには後ほど作成するAPI Gatewayのリンクを入れてあげます。
環境変数の取得
そのまま下まで行き、「応答メッセージ」を無効にします。
これが有効のままだと、エージェントの応答とは別に、デフォルトの応答メッセージが返信され、邪魔になるので不要です。
次に「チャンネル基本設定」を選択します。
チャンネルシークレットを控えておきます。これらの環境変数はLambdaにて使用するので使えるように控えておきます。
ここまで完了すれば、LINE側の設定は概ね終了となります。
リソースの作成
エージェントの作成
まずはメインとなるエージェントの作成をしていきます。
今回はツールとかは使用せずに、シンプルにStrandsのエージェントをRuntimeにデプロイしています。
Runtimeへのデプロイ手順は過去に書いた以下のブログを見ていただけると嬉しいです。
エージェントコードは以下になります。
from strands import Agent
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCoreApp
app = BedrockAgentCoreApp()
bedrock_model = BedrockModel(
model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0",
)
agent = Agent(
model=bedrock_model
)
@app.entrypoint
def invoke(payload):
"""いい感じに返信して"""
user_message = payload.get("prompt","")
result = agent(user_message)
return {"result": result.message}
if __name__ == "__main__":
app.run()
デプロイ・動作確認
以下のコマンドでデプロイから動作確認まで実施します。
uv run agentcore configure --entrypoint line_bot.py
uv run agentcore launch
uv run agentcore invoke '{"prompt": "こんにちは"}'
デプロイ後、以下のようにレスポンスが来てくれたら、ひとまずはオッケーです🙆
Lambda関数の作成
import json
import os
import uuid
import urllib.request
import urllib.error
from typing import Any, Dict
import boto3
def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
try:
if "body" not in event:
return _response(400, {"error": "Missing request body"})
body = event["body"]
if isinstance(body, str):
body = json.loads(body)
if "events" in body and isinstance(body["events"], list) and len(body["events"]) > 0:
return handle_line_webhook(body)
prompt = body.get("prompt", "Hello")
result_text = call_bedrock_agentcore(prompt)
return _response(200, {"result": result_text})
except json.JSONDecodeError:
return _response(400, {"error": "Invalid JSON in request body"})
except Exception as e:
print(f"ERROR: {str(e)}")
return _response(500, {"error": str(e)})
def handle_line_webhook(body: Dict[str, Any]) -> Dict[str, Any]:
try:
event = body["events"][0]
if event.get("type") != "message":
return _response(200, {"message": "Ignored non-message event"})
message = event.get("message", {})
if message.get("type") != "text":
return _response(200, {"message": "Ignored non-text message"})
user_text = message.get("text", "")
reply_token = event.get("replyToken")
result_text = call_bedrock_agentcore(user_text)
if reply_token:
send_line_reply(reply_token, result_text)
except Exception as e:
print(f"ERROR in handle_line_webhook: {str(e)}")
return {
"statusCode": 200,
"body": json.dumps("OK")
}
def call_bedrock_agentcore(prompt: str) -> str:
agent_arn = os.environ.get("BEDROCK_AGENT_RUNTIME_ARN")
if not agent_arn:
return "Agent runtime ARN not configured"
client = boto3.client("bedrock-agentcore")
payload = json.dumps({"prompt": prompt}, ensure_ascii=False).encode("utf-8")
response = client.invoke_agent_runtime(
agentRuntimeArn=agent_arn,
contentType="application/json",
payload=payload,
traceId=str(uuid.uuid4()).replace("-", "")
)
try:
if response.get("contentType") == "application/json":
buf = bytearray()
for chunk in response.get("response", []):
if isinstance(chunk, (bytes, bytearray)):
buf.extend(chunk)
else:
buf.extend(str(chunk).encode("utf-8", errors="ignore"))
full_content = buf.decode("utf-8", errors="ignore")
data = json.loads(full_content)
inner = data.get("result")
if isinstance(inner, dict):
content = inner.get("content")
if isinstance(content, list) and len(content) > 0:
first = content[0]
if isinstance(first, dict) and "text" in first:
return first["text"]
return full_content
else:
return str(response)
except Exception as e:
print(f"ERROR processing AgentCore response: {str(e)}")
return f"Error processing response: {str(e)}"
def send_line_reply(reply_token: str, message_text: str) -> None:
access_token = os.environ.get("LINE_CHANNEL_ACCESS_TOKEN")
if not access_token:
print("LINE_CHANNEL_ACCESS_TOKEN is not set")
return
url = "https://api.line.me/v2/bot/message/reply"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}",
}
body = {
"replyToken": reply_token,
"messages": [
{"type": "text", "text": message_text}
]
}
req = urllib.request.Request(
url,
data=json.dumps(body).encode("utf-8"),
headers=headers,
method="POST"
)
try:
with urllib.request.urlopen(req, timeout=5) as res:
print(f"LINE reply status: {res.status}")
except urllib.error.HTTPError as e:
print(f"HTTPError sending LINE reply: {e.code}, {e.read()}")
except Exception as e:
print(f"Error sending LINE reply: {str(e)}")
def _response(status: int, body: Dict[str, Any]) -> Dict[str, Any]:
return {
"statusCode": status,
"headers": {
"Content-Type": "application/json",
"Access-Control-Allow-Origin": "*",
},
"body": json.dumps(body, ensure_ascii=False),
}
環境変数の設定
作成したLambda関数の環境変数に、あらかじめ控えていた値を設定しておきます。
-
BEDROCK_AGENT_RUNTIME_ARN -
LINE_CHANNEL_ACCESS_TOKEN -
LINE_CHANNEL_SECRET
権限設定
LambdaからRuntimeを起動するため、Lambdanの実行ロールにAgentCoreのポリシーを付与する必要があります。
以下のようなAWS管理ポリシーがあるので、Lambdaのロールに付与しておきます。
BedrockAgentCoreFullAccess
Lambda関数のテスト
以下のような形でテストを実行して確認します。
Lambdaのタイムアウト時間はデフォルトだと、3秒に設定されており、そのままだとタイムアウトになってしまうことがあります。
私の場合は10秒くらいに伸ばしてありますし、ある程度伸ばしておくといいかもしれないです。
API Gatewayの設定
次に、上で作成したLambda関数のトリガーになるようなAPI Gatewayを作成します。
設定手順は以前実施したものと全く同じになるので割愛させてください。
Webhookの設定や動作確認も併せて行うことができるので、やっておきましょう。
動作確認
試しにトーク画面から話しかけてみましたが、正常に返信してくれました。
ツールとかは何も付けてないので、ただお話しできるだけですが、ここまで動くのが確認できれば最低限完成になります。
前回はただの基盤モデルとお話しするだけでしたが、今回のAgentCoreはGatewayをはじめ機能の拡張性が豊富なので色々試せそうです。
最後に
ということで、今回はAgentCoreのエージェントとLINEでお話ししてみました。
最終的な感想になりますが、LINEbotの場合、streamlitのようなストリーミング出力ができないようです。
そのため、メッセージを送ってから返信が来るまで数秒じっと待っている必要がありました。
それはかなりユーザー体験が悪くなってしまうので、AIエージェントはチャットに全振りするなら、やはりストリーム出力はあると嬉しいなと思いました。