はじめに
こんにちは。
今年からAIエージェントを本格的に学ぼうと勉強を始めた浅野です。
本日はBedrock AgentCoreとStrands Agentを使って、LINEからAWSアカウントを操作できるエージェントを作成してみました。
私は12月まで全くAI系に詳しくなかったのですが、Re:Inventに行き、AIを学ばないと僕の人生が終わる（言いすぎ？）と思いました。
そんな中で下記のようなAIエージェント参考書を読み、年末年始寝る間を惜しむことなく寝ましたが、ちゃんと勉強しました。
僕はハンズオンを通じて構築したことが重要ではなく、構築の過程で出てきた「なぜ？」から、自分で調べて得ていくことが重要だと考えています。
ぜひ皆さんもエージェントを触りながら、ハンズオンライフを楽しんでいただければと思います。
構成
下記の構成で行います。
※一部料金が発生しますので、使い過ぎに注意してください。
ユーザー → LINE → Lambda（Webhook） → AgentCore → AWSアカウント操作
前提
AIエージェント初心者のため、スクリプトやエラーハンドリング、プロンプトエンジニアリング等まだまだ不完全な箇所はあると思います。
スクリプトを見て「ん？」と思われた方は、適宜修正してデプロイしていただければと思います。
作ってみる
LINEのMessaging APIを有効化する
下記からLINE Developersコンソールへ飛びます。
トップからスクロールをすると、プロバイダーという項目があります。
プロバイダー横の「作成」ボタンを押します。
プロバイダーを登録すると、チャネル設定画面に飛びます。
ここでは「Messaging API」を押してください。
「公式アカウントを作成する」ボタンを押下し、アカウントの作成を行ってください。
アカウント作成後、プロバイダへ飛ぶとMessaging API設定というタブがあります。
そちらのタブから、APIキーの発行を行ってください。
Bedrock AgentCoreをデプロイする
AgentCoreをデプロイします。
下記の二つのファイルを配置後、CodeSpaceからagentcoreコマンドを実行します。
私は下記の記事を参考にデプロイしています。
今回は環境変数に「LINE_CHANNEL_ACCESS_TOKEN」を含める必要があります。
エージェントコアをデプロイする際、下記のように指定してもらえればと思います。
agentcore launch --env LINE_CHANNEL_ACCESS_TOKEN=xxxxxxxxxx
entrypoint.py
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent, tool
from mcp.client.stdio import stdio_client, StdioServerParameters
import os
import urllib.request
import json
LINE_CHANNEL_ACCESS_TOKEN = os.environ.get("LINE_CHANNEL_ACCESS_TOKEN")
# AgentCoreランタイム用のAPIサーバーを作成
app = BedrockAgentCoreApp()
def _mcp_env() -> dict:
env = os.environ.copy()
env["OTEL_PROPAGATORS"] = "tracecontext,baggage"
env.setdefault("OTEL_TRACES_EXPORTER", "none")
env.setdefault("OTEL_METRICS_EXPORTER", "none")
env.setdefault("OTEL_LOGS_EXPORTER", "none")
return env
#####################################
# AWS Controle Agent Tool
#####################################
@tool
def aws_control_agent(command: str) -> str:
"""
引数で受け取ったコマンドをMCPサーバ経由で実行し、その結果を返す
"""
try:
client = MCPClient(
lambda: stdio_client(
StdioServerParameters(
command="uvx",
args=["awslabs.aws-api-mcp-server"],
env=_mcp_env(),
)
)
)
with client:
agent = Agent(
model="jp.anthropic.claude-sonnet-4-5-20250929-v1:0",
tools=client.list_tools_sync(),
)
result = agent(command)
return str(result)
except Exception as e:
return f"aws_control_agent実行時にエラーが発生しました: {str(e)}"
#####################################
# AWS Knowledge Agent Tool
#####################################
@tool
def aws_knowledge_agent(prompt: str) -> str:
"""
AWSの知識ベースを活用して質問に答えるエージェントツール
"""
try:
client = MCPClient(lambda: streamablehttp_client("https://knowledge-mcp.global.api.aws"))
with client:
agent = Agent(
model="jp.anthropic.claude-sonnet-4-5-20250929-v1:0",
tools=client.list_tools_sync(),
system_prompt="""AWS操作に関する情報の取得を行ってください。
問い合わせは2回までとしてください。""",
)
result = agent(prompt)
return str(result)
except Exception as e:
return f"aws_knowledge_agent実行時にエラーが発生しました: {str(e)}"
async def send_line_message(user_id, message):
"""
LINEにメッセージをpush APIで送信する非同期関数
"""
if not LINE_CHANNEL_ACCESS_TOKEN or not user_id or not message:
return
line_payload = {
"to": user_id,
"messages": [{"type": "text", "text": message}],
}
req = urllib.request.Request(
"https://api.line.me/v2/bot/message/push",
data=json.dumps(line_payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req) as res:
res.read()
except Exception:
pass
@app.entrypoint
async def invoke_agent(payload, context):
prompt = payload.get("prompt")
user_id = payload.get("userId")
orchestrator = Agent(
model="jp.anthropic.claude-sonnet-4-5-20250929-v1:0",
system_prompt="""AWS開発チームのメンバーです。
AWSアカウント内にあるリソースの管理や操作を支援します。
aws_knowledge_agent: AWSの知識ベースを使って、技術的な質問に答えます。
aws_control_agent: AWSリソースの管理や操作を行います。
以下のツールを使って、ユーザーからのリクエストに対応してください。""",
tools=[aws_knowledge_agent, aws_control_agent],
)
# 最初の1回だけreplyTokenで即時応答（Lambda側で実施想定）
# 以降はpush APIでuserId宛に送信
if hasattr(orchestrator, "stream"):
async for chunk in orchestrator.stream(prompt):
text = str(chunk)
if text.strip():
await send_line_message(user_id, text)
return "OK"
else:
result = orchestrator(prompt)
await send_line_message(user_id, str(result))
return "OK"
# APIサーバーを起動
app.run()
requirements.txt
boto3>=1.34.0
botocore>=1.34.0
strands-agents
strands-agents-tools
strands-agents[otel]
bedrock-agentcore
bedrock-agentcore[strands-agents]
Lambda関数を作成する
lambda_function.py
import json
import os
import urllib.request
import boto3
import uuid
import threading
LINE_CHANNEL_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
AGENTCORE_RUNTIME_ARN = os.environ["AGENTCORE_RUNTIME_ARN"]
agentcore = boto3.client("bedrock-agentcore")
def _read_agentcore_response(resp: dict) -> str:
"""
AgentCore の戻りは resp["response"] (StreamingBody)。
contentType により JSON or text/event-stream を判定して取り出す。
"""
content_type = resp.get("contentType", "") or ""
# Streaming (SSE)
if "text/event-stream" in content_type:
chunks = []
for line in resp["response"].iter_lines(chunk_size=1024):
if not line:
continue
s = line.decode("utf-8", errors="replace")
if s.startswith("data: "):
chunks.append(s[6:])
return "\n".join(chunks).strip() or "AgentCoreからの応答がありません。"
# Non-stream JSON
if "application/json" in content_type:
raw = resp["response"].read().decode("utf-8", errors="replace")
try:
obj = json.loads(raw)
if isinstance(obj, str):
return obj
return json.dumps(obj, ensure_ascii=False)
except Exception:
return raw.strip() or "AgentCoreからの応答がありません。"
# Fallback
raw = resp["response"].read().decode("utf-8", errors="replace")
return raw.strip() or "AgentCoreからの応答がありません。"
def _invoke_agentcore_and_reply(prompt, reply_token, user_id, group_id, room_id, session_id):
"""
AgentCoreを呼び出し、結果をLINEに返信する（バックグラウンド用）
"""
try:
payload = json.dumps({
"prompt": prompt,
"replyToken": reply_token,
"userId": user_id
}).encode("utf-8")
print(user_id, reply_token, prompt)
resp = agentcore.invoke_agent_runtime(
agentRuntimeArn=AGENTCORE_RUNTIME_ARN,
runtimeSessionId=session_id,
contentType="application/json",
accept="text/event-stream",
payload=payload,
)
agent_reply = _read_agentcore_response(resp)
except Exception as e:
agent_reply = f"AgentCore呼び出しでエラー: {str(e)}"
# LINEに返信
line_payload = {
"replyToken": reply_token,
"messages": [{"type": "text", "text": agent_reply}],
}
req = urllib.request.Request(
"https://api.line.me/v2/bot/message/reply",
data=json.dumps(line_payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req) as res:
res.read()
except Exception:
pass
def lambda_handler(event, context):
body = json.loads(event["body"])
for ev in body.get("events", []):
if ev.get("type") != "message":
continue
if ev.get("message", {}).get("type") != "text":
continue
prompt = ev["message"]["text"]
reply_token = ev["replyToken"]
user_id = ev.get("source", {}).get("userId")
group_id = ev.get("source", {}).get("groupId")
room_id = ev.get("source", {}).get("roomId")
session_id = user_id or str(uuid.uuid4())
# AgentCore呼び出しをバックグラウンドで実行
threading.Thread(
target=_invoke_agentcore_and_reply,
args=(prompt, reply_token, user_id, group_id, room_id, session_id),
daemon=True
).start()
# すぐにLINEへ応答
line_payload = {
"replyToken": reply_token,
"messages": [{"type": "text", "text": "AgentCoreへ送信しました。結果が来るまでお待ちください。"}],
}
req = urllib.request.Request(
"https://api.line.me/v2/bot/message/reply",
data=json.dumps(line_payload).encode("utf-8"),
headers={
"Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}",
"Content-Type": "application/json",
},
method="POST",
)
with urllib.request.urlopen(req) as res:
res.read()
return {"statusCode": 200, "body": "OK"}
実際に動かしてみる
今回はEC2インスタンスの作成指示を行いました。
指示を送ると、すぐに返答が返ってきます。
Cloudwatch AgentにもAgentが起動されたログが配信されています...!
返答がきました。
なんとVPCまで作ってくれています。
まとめ
今回はBedrock AgentCoreを使って、LINEからAIエージェントを起動する方法を試して見ました。
コンソールログインがめんどくさいとき、今PC開けない！と言う時。LINEで指示できたらめちゃくちゃ楽だろうなーと思い、作ったことがきっかけです。
このエージェントは逐次報告してくれなかったり、メモリによる記憶ができてないのでまだまだ改善の余地はあります。
またアップデートできたら、随時更新していこうと思います！
最後まで見ていただき、ありがとうございました。