第1章では LINE から受け取ったレシートを Gemini で解析し、S3 に JSONL として貯めました。
※第1章へのリンク
https://qiita.com/toshi-arch/items/02ad744b0457fbf60f94
第2章では、その購買履歴を AI エージェントが使えるよう「ツール Lambda」を2つ作ります。
AgentCore のツールとは
Amazon Bedrock AgentCore は AI エージェントを本番環境にデプロイ・管理するマネージドサービスです。エージェントは「ツール」を呼び出しながら推論を進めます。今回は Lambda をツールとして登録します。
エージェント → AgentCore Gateway（MCP） → ツール Lambda
今回作る2つのツール:
| Lambda 関数名 | 役割 |
|---|---|
smart-shopper-tool-s3-search |
指定店舗の購買履歴を S3 から取得 |
smart-shopper-tool-consumption-calc |
消費サイクルを計算して購入提案リストを返す |
共通の作成設定
両 Lambda とも同じ設定で作成します。
- ランタイム: Python 3.13 / アーキテクチャ: arm64
- 実行ロール:
SmartShopperToolLambdaRole（後述） - タイムアウト: 30秒（デフォルト3秒では S3 検索中に終了するので必ず変更）
IAM ロール（ツール Lambda 用）
IAM コンソール →「AWS サービス」→「Lambda」で作成。
アタッチするポリシー:
AWSLambdaBasicExecutionRole- S3 読み取りのインラインポリシー:
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:ListBucket"],
"Resource": [
"arn:aws:s3:::your-s3-bucket-name",
"arn:aws:s3:::your-s3-bucket-name/*"
]
}]
}
ロール名: SmartShopperToolLambdaRole
AgentCore Runtime 実行ロールも先に作っておく
第4章でデプロイするときに必要なので先に作っておきます。
「カスタム信頼ポリシー」で作成:
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": { "Service": "bedrock-agentcore.amazonaws.com" },
"Action": "sts:AssumeRole"
}]
}
アタッチするポリシー:
AmazonBedrockFullAccessAWSLambdaBasicExecutionRole- 以下のインラインポリシー:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:ListBucket"],
"Resource": [
"arn:aws:s3:::your-s3-bucket-name",
"arn:aws:s3:::your-s3-bucket-name/*"
]
},
{
"Effect": "Allow",
"Action": "lambda:InvokeFunction",
"Resource": [
"arn:aws:lambda:ap-northeast-1:[アカウントID]:function:smart-shopper-tool-s3-search",
"arn:aws:lambda:ap-northeast-1:[アカウントID]:function:smart-shopper-tool-consumption-calc"
]
},
{
"Effect": "Allow",
"Action": "bedrock-agentcore:*",
"Resource": "*"
},
{
"Effect": "Allow",
"Action": [
"ecr:GetAuthorizationToken",
"ecr:BatchGetImage",
"ecr:GetDownloadUrlForLayer"
],
"Resource": "*"
}
]
}
lambda:InvokeFunction/bedrock-agentcore:*/ ECR 権限の3つは忘れやすいですが、どれか欠けると第4章で詰まります。
ロール名: SmartShopperAgentCoreRuntimeRole
Step 1. Lambda: S3 検索ツール
関数名: smart-shopper-tool-s3-search
環境変数: S3_BUCKET_NAME = your-s3-bucket-name
import json
import boto3
import os
s3 = boto3.client('s3')
BUCKET = os.environ['S3_BUCKET_NAME']
def lambda_handler(event, context):
tool_name = _get_tool_name(context)
if tool_name == 'search_purchase_history':
return search_purchase_history(event.get('store_name', ''))
return {'error': f'Unknown tool: {tool_name}'}
def search_purchase_history(store_name: str) -> dict:
results = []
paginator = s3.get_paginator('list_objects_v2')
pages = paginator.paginate(Bucket=BUCKET, Prefix='smart-shopper-agent/jsons/')
for page in pages:
for obj in page.get('Contents', []):
try:
body = s3.get_object(Bucket=BUCKET, Key=obj['Key'])['Body'].read().decode('utf-8')
for line in body.strip().split('\n'):
if not line:
continue
record = json.loads(line)
if store_name.lower() in record.get('store_name', '').lower():
results.append(record)
except Exception as e:
print(f"Skipping {obj['Key']}: {e}")
return {'records': results, 'count': len(results)}
def _get_tool_name(context) -> str:
"""
AgentCore Gateway はツール名を「ターゲット名___ツール名」形式で渡す。
プレフィックスを除去して純粋なツール名を返す。
"""
try:
raw = context.client_context.custom.get('bedrockAgentCoreToolName', '')
delimiter = '___'
if delimiter in raw:
return raw[raw.index(delimiter) + len(delimiter):]
return raw
except Exception:
return ''
_get_tool_name()について: AgentCore Gateway は Lambda 呼び出し時にツール名をsmart-shopper-s3-search-target___search_purchase_historyのようにターゲット名 +___+ ツール名の形式で渡します。このプレフィックス除去をしないと、ツール分岐が全て外れてUnknown toolになります。詳しくは第3章で説明します。
Step 2. Lambda: 消費サイクル計算ツール
関数名: smart-shopper-tool-consumption-calc
環境変数: なし
from datetime import datetime, timezone, timedelta
JST = timezone(timedelta(hours=9))
def lambda_handler(event, context):
tool_name = _get_tool_name(context)
if tool_name == 'calculate_consumption':
return calculate_consumption(event.get('records', []), event.get('store_name', ''))
return {'error': f'Unknown tool: {tool_name}'}
def calculate_consumption(records: list, store_name: str) -> dict:
today = datetime.now(JST)
item_history: dict[str, list[datetime]] = {}
for record in records:
try:
purchased_at = datetime.fromisoformat(record['purchased_at']).astimezone(JST)
except (KeyError, ValueError):
continue
for item in record.get('items', []):
name = item.get('name', '')
if name:
item_history.setdefault(name, []).append(purchased_at)
suggestions = []
for name, dates in item_history.items():
dates_sorted = sorted(dates)
last_bought = dates_sorted[-1]
days_since = (today - last_bought).days
if len(dates_sorted) >= 2:
intervals = [(dates_sorted[i+1] - dates_sorted[i]).days for i in range(len(dates_sorted)-1)]
avg_cycle = sum(intervals) / len(intervals)
should_buy = days_since >= avg_cycle * 0.8 # 消費サイクルの 80% 経過で提案
else:
avg_cycle = None
should_buy = days_since >= 14 # 履歴1件のみは 14 日で提案
if should_buy:
suggestions.append({
'name': name,
'last_bought_days_ago': days_since,
'avg_cycle_days': round(avg_cycle) if avg_cycle else None,
})
return {'store_name': store_name, 'suggestions': suggestions, 'suggestion_count': len(suggestions)}
def _get_tool_name(context) -> str:
try:
raw = context.client_context.custom.get('bedrockAgentCoreToolName', '')
delimiter = '___'
if delimiter in raw:
return raw[raw.index(delimiter) + len(delimiter):]
return raw
except Exception:
return ''
次回予告
第3章では、この2つの Lambda を AgentCore Gateway に登録し、AI エージェントが MCP プロトコルで呼び出せるようにします。ここがハマりポイントの多いセクションです。