🤖
【Gemini】画像解析×ブラウザ操作MCPで実現する「ポイ活アドバイザー」
背景
実際の店舗やオンラインのショップなどで、なにかを購入する時、現金以外の決済方法を使うことが多いです（たとえば、QRコード決済やクレジットカードなど）
そして、私の場合は定期的に悔しいことが起きます
「こんなキャンペーンあったのか..」
キャンペーンページを事前にみることもありますが、
キャンペーンが多すぎて見づらいことが多いです
行った先のショッピングモールに入ってる店舗で使えるキャンペーンないのかな、とか
〇〇で買おうと思ってるけどキャンペーンないかな、とか
そこで「AIに探してもらおう！」と思い、本記事の執筆に至っています
前提
私がよく使う楽天ペイとd払いのキャンペーンを対象に検証してみます
楽天ペイ：https://pay.rakuten.co.jp/campaign/
d払い：https://service.smt.docomo.ne.jp/keitai_payment/campaign/
本記事に関する注意事項と免責事項
実装方針
LLMのトークン消費を抑えたいので、
「1回LLMにお願いすればよい処理」と「毎回LLMにお願いしないといけない処理」を
分けることを意識します
- 1回LLMにお願いすればよい処理（時々刻々と変わらない情報の取得）
- キャンペーン情報の取得
- 毎回LLMにお願いしないといけない処理（都度条件が変わる情報の取得）
- 購入先や訪問先情報を伝えて使えそうなキャンペーン情報を教えてもらう
つづいて、LLMによる精度を高めるための処理の分割です
一度に多くの作業を依頼すると精度が悪くなるため、できるだけ作業を分割して依頼します
今回作業の分割の効果が最も大きかったのがキャンペーン情報の取得です
最初は一気に「このキャンペーンサイトにアクセスしてキャンペーン情報を一覧化して」と依頼しました
すると、「キャンペーンサイトに記載のキャンペーンの一覧を取得すること」と
「各キャンペーンの詳細情報をとること」の2つをLLMにまかせるという
複雑な依頼なため、期待する結果がとれませんでした
なので、以下のように作業を分割・単純化することで、精度が向上しました
- キャンペーン情報の取得を以下の２つの作業に分割
- キャンペーンの詳細情報のURL一覧を作る（データ収集バッチ）
- URL一覧にもとづき１つずつキャンペーン詳細情報を取得（対話型アドバイザー）
最後にもう１つ工夫として、マルチモーダルの活用です
キャンペーンサイトには画像が使われることも多く、
重要な情報が実は画像の中に埋め込まれていることも多いです
なので、Playwright MCPを使ってサイトを画像にし、その画像をLLMに渡して
キャンペーン情報の詳細を把握してもらうようにしています
上記のすべての処理を図示すると以下のようなイメージです
サンプルコード（Geminiによるコーディング）
事前準備
環境変数にご自身のGemini APIキー を設定してください
（.env ファイルに記述することを推奨）
GOOGLE_API_KEY={Your API Key}
データ収集バッチ
データ収集バッチ サンプルコード
import asyncio
import json
import os
import re
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset, StdioConnectionParams, StdioServerParameters
from dotenv import load_dotenv
load_dotenv(override=True)
# 保存先ファイル
CAMPAIGN_LIST_FILE = "./data/campaign_links.json"
CAMPAIGN_DATA_FILE = "./data/campaign_data.json"
async def main():
# Playwright MCPの設定
mcp_toolset = McpToolset(
connection_params=StdioConnectionParams(
server_params=StdioServerParameters(
command="npx",
args=["-y", "@playwright/mcp@latest"],
),
timeout=300,
),
)
# 1. 一覧取得用エージェント (VLM/画像解析ベース)
list_agent = Agent(
name="list_collector",
description="各サイトのスクリーンショットからキャンペーンを特定するエージェント",
instruction="""
あなたは視覚情報解析の専門家です。Playwrightのスクリーンショット機能を使用して、キャンペーンを特定してください。
【高速・視覚解析の指示】
1. ページを開き、数秒待機してバナーがすべて表示されたら、ページ全体のスクリーンショットを撮影してください。
2. 画像から「キャンペーン名」と、その「リンク先URL」をすべて抽出してください。
3. URLが相対パスの場合は、必ずベースURLを補完してフルURL(https://...)に変換してください。
4. DOMを解析するよりも、視覚的に認識できる情報を優先してください。
回答は必ず以下のJSONリスト形式のみで出力してください。
[{"title": "...", "url": "..."}]
""",
model="gemini-2.5-flash",
tools=[mcp_toolset],
)
# 2. 詳細取得用エージェント (VLM/画像解析ベース)
detail_agent = Agent(
name="detail_collector",
description="詳細ページのスクリーンショットから条件を読み取るエージェント",
instruction="""
あなたは画像解析の専門家です。提供されたURLのスクリーンショットを撮影して解析してください。
【視覚解析の指示】
1. 指定されたURLのスクリーンショットを撮影してください。
2. 画像内のテキスト（大きな文字、バナー内の数字、注釈）から、以下の情報を抽出してください。
3. 文字が小さすぎて読めない場合は、innerTextを取得するツールを併用しても構いません。
- shop: 対象店舗（特にショッピングモール内の対象外店舗などの例外情報）
- rate: 還元率や付与条件
- end_date: 終了日
- details: その他、ユーザーが店で支払う際に知っておくべき注意点
回答は必ず以下のJSONオブジェクト形式のみで出力してください。
{"shop": "...", "rate": "...", "end_date": "...", "details": "..."}
""",
model="gemini-2.5-flash",
tools=[mcp_toolset],
)
session_service = InMemorySessionService()
list_runner = Runner(app_name="list_app", agent=list_agent, session_service=session_service)
detail_runner = Runner(app_name="detail_app", agent=detail_agent, session_service=session_service)
session_id, user_id = "update_job", "admin"
await session_service.create_session(app_name="list_app", session_id=session_id, user_id=user_id)
await session_service.create_session(app_name="detail_app", session_id=session_id, user_id=user_id)
target_sites = [
{"name": "d払い", "url": "https://service.smt.docomo.ne.jp/keitai_payment/campaign/"},
{"name": "楽天ペイ", "url": "https://pay.rakuten.co.jp/campaign/"}
]
campaign_links_dict = {}
# --- Phase 1: 一覧取得 ---
if os.path.exists(CAMPAIGN_LIST_FILE):
print(f"--- 既存の一覧ファイル {CAMPAIGN_LIST_FILE} を読み込みます ---")
with open(CAMPAIGN_LIST_FILE, "r", encoding="utf-8") as f:
campaign_links_dict = json.load(f)
else:
print("--- キャンペーン一覧の新規取得を開始します (VLM解析) ---")
try:
for site in target_sites:
print(f"\n[{site['name']}] 一覧を取得中...", end="", flush=True)
list_prompt = f"{site['url']} のスクリーンショットを撮り、キャンペーン一覧を抽出してください。"
list_result_text = ""
async for chunk in list_runner.run_async(
new_message=Content(role="user", parts=[Part(text=list_prompt)]),
session_id=session_id, user_id=user_id
):
if chunk.content and chunk.content.parts:
part_text = chunk.content.parts[0].text
if part_text:
list_result_text += part_text
print(".", end="", flush=True)
# JSON抽出
json_match = re.search(r'\[\s*\{.*\}\s*\]', list_result_text, re.DOTALL)
if json_match:
try:
links = json.loads(json_match.group(0))
seen_urls = set()
unique_links = []
for link in links:
if link.get("url") and link["url"] not in seen_urls:
unique_links.append(link)
seen_urls.add(link["url"])
campaign_links_dict[site['name']] = unique_links
print(f" {len(unique_links)}件取得")
except Exception:
print(" (JSONパースエラー)")
else:
print(" 一覧未検出")
with open(CAMPAIGN_LIST_FILE, "w", encoding="utf-8") as f:
json.dump(campaign_links_dict, f, ensure_ascii=False, indent=2)
print(f"\n一覧を {CAMPAIGN_LIST_FILE} に保存しました。")
except Exception as e:
print(f"\n一覧取得中にエラーが発生しました: {e}")
# --- Phase 2: 詳細取得 ---
if not campaign_links_dict:
print("調査対象のキャンペーンリンクがありません。終了します。")
await mcp_toolset.close()
return
# 既存の詳細データを読み込み、取得済みURLのセットを作成
existing_campaigns = []
processed_urls = set()
if os.path.exists(CAMPAIGN_DATA_FILE):
try:
with open(CAMPAIGN_DATA_FILE, "r", encoding="utf-8") as f:
existing_campaigns = json.load(f)
processed_urls = {item["url"] for item in existing_campaigns if "url" in item}
print(f"--- 既存のデータ {CAMPAIGN_DATA_FILE} から {len(processed_urls)}件の既知URLをロードしました ---")
except Exception:
print("--- 既存データのロードに失敗しました。新規に取得を開始します ---")
print("\n--- キャンペーン詳細の調査を開始します (VLM解析) ---")
all_final_campaigns = list(existing_campaigns)
try:
for service_name, links in campaign_links_dict.items():
print(f"\n[{service_name}] の詳細を調査中... (全{len(links)}件)")
for i, link_info in enumerate(links):
title = link_info.get("title", "無題")
url = link_info.get("url")
# 基本チェック
if not url or not url.startswith("http"):
continue
# すでに取得済みならスキップ
if url in processed_urls:
print(f" ({i+1}/{len(links)}) スキップ: {title[:15]}...")
continue
print(f" ({i+1}/{len(links)}) 新規取得中: {title[:15]}...", end="", flush=True)
detail_prompt = f"URL: {url}\nスクリーンショットを撮影し、詳細情報を抽出してください。"
detail_result_text = ""
try:
async for chunk in detail_runner.run_async(
new_message=Content(role="user", parts=[Part(text=detail_prompt)]),
session_id=session_id, user_id=user_id
):
if chunk.content and chunk.content.parts:
part_text = chunk.content.parts[0].text
if part_text:
detail_result_text += part_text
# 詳細JSONのパース
d_match = re.search(r'\{\s*".*"\s*:\s*".*"\s*\}', detail_result_text, re.DOTALL)
if d_match:
detail_data = json.loads(d_match.group(0))
new_entry = {
"service": service_name,
"title": title,
"url": url,
**detail_data
}
all_final_campaigns.append(new_entry)
processed_urls.add(url) # 二重取得防止
# 1件取得ごとに途中保存（クラッシュ対策）
with open(CAMPAIGN_DATA_FILE, "w", encoding="utf-8") as f:
json.dump(all_final_campaigns, f, ensure_ascii=False, indent=2)
print(" OK & 保存")
else:
print(" パース失敗 (JSONが見つかりませんでした)")
except Exception as e:
# エラーメッセージを詳細に表示
print(f" エラー発生: {str(e)}")
# 最終結果の保存（念のため最後にも実行）
if all_final_campaigns:
with open(CAMPAIGN_DATA_FILE, "w", encoding="utf-8") as f:
json.dump(all_final_campaigns, f, ensure_ascii=False, indent=2)
print(f"\n完了: 合計 {len(all_final_campaigns)} 件のデータが {CAMPAIGN_DATA_FILE} に格納されています。")
else:
print("\n有効なデータが取得できませんでした。")
finally:
await mcp_toolset.close()
if __name__ == "__main__":
asyncio.run(main())
対話型アドバイザー
対話型アドバイザー サンプルコード
import asyncio
import json
import os
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part
from google.adk.tools import google_search
from dotenv import load_dotenv
load_dotenv(override=True)
CAMPAIGN_DATA_FILE = "./data/campaign_data.json"
USER_CONFIG_FILE = "./data/user_config.json"
SHOP_CACHE_FILE = "./data/shop_cache.json"
def load_json_file(filepath, default_value):
if os.path.exists(filepath):
with open(filepath, "r", encoding="utf-8") as f:
return json.load(f)
return default_value
def save_json_file(filepath, data):
with open(filepath, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
async def main():
# 各種設定・データの読み込み
my_payments = load_json_file(USER_CONFIG_FILE, {}).get("payments", [])
if not my_payments:
print("利用可能な決済手段が設定されていません。")
p_input = input("利用中の決済をカンマ区切りで入力してください (例: d払い,PayPay): ")
my_payments = [x.strip() for x in p_input.split(",")]
save_json_file(USER_CONFIG_FILE, {"payments": my_payments})
cached_campaigns = load_json_file(CAMPAIGN_DATA_FILE, [])
campaigns_context = json.dumps(cached_campaigns, ensure_ascii=False)
# ショップ情報のキャッシュ読み込み
shop_cache = load_json_file(SHOP_CACHE_FILE, {})
# アドバイザーエージェント
advisor_agent = Agent(
name="shopping_advisor",
description="お出かけ先での最適な決済を提案するエージェント",
instruction=f"""
あなたはショッピングアドバイザーです。
ユーザーの「行き先」に基づいて、以下の手順で回答してください。
1. まず、提供された【既知のショップ情報キャッシュ】に行き先のショップ一覧があるか確認する。
2. キャッシュにない場合のみ、Google検索を使用してその施設に入っている主要なショップ・テナント一覧を取得する。
3. 検索で新しいショップ一覧を取得した場合は、回答の最後に必ず「NEW_SHOPS: [施設名]: [ショップ1, ショップ2...]」という形式で出力して、システムに保存を促してください。
4. 最終的に、【キャンペーン情報】とショップ一覧を照合し、ユーザーの決済手段 {my_payments} の中から最もお得な支払い方法を提案する。
【キャンペーン情報】:
{campaigns_context}
【既知のショップ情報キャッシュ】:
{json.dumps(shop_cache, ensure_ascii=False)}
""",
model="gemini-2.5-flash",
tools=[google_search],
)
runner = Runner(
app_name=advisor_agent.name,
agent=advisor_agent,
session_service=InMemorySessionService(),
)
session_id, user_id = "advisor_sess", "user1"
await runner.session_service.create_session(app_name=advisor_agent.name, session_id=session_id, user_id=user_id)
print(f"--- ショッピングアドバイザー起動 (設定済み決済: {', '.join(my_payments)}) ---")
while True:
destination = input("\nどこにお出かけしますか？ (例: ららぽーと豊洲) / exitで終了: ")
if destination.lower() in ["exit", "quit", "終了"]:
break
print(f"「{destination}」のお得な情報を確認中...\n")
full_response = ""
async for chunk in runner.run_async(
new_message=Content(role="user", parts=[Part(text=f"{destination}に行きます。ここで使えるお得なキャンペーンを教えて。")]),
session_id=session_id,
user_id=user_id
):
if chunk.content and chunk.content.parts:
text = chunk.content.parts[0].text
print(text, end="", flush=True)
full_response += text
# 新しいショップ情報の保存処理
if "NEW_SHOPS:" in full_response:
try:
# 簡易的なパース: "NEW_SHOPS: 施設名: ショップ名1, ショップ名2"
line = [l for l in full_response.split('\n') if "NEW_SHOPS:" in l][-1]
parts = line.replace("NEW_SHOPS:", "").strip().split(":", 1)
if len(parts) == 2:
loc_name = parts[0].strip()
shops = [s.strip() for s in parts[1].split(",")]
shop_cache[loc_name] = shops
save_json_file(SHOP_CACHE_FILE, shop_cache)
print(f"\n(システム: {loc_name} のショップ情報をキャッシュに保存しました)")
except Exception as e:
pass
print()
if __name__ == "__main__":
asyncio.run(main())
対話結果
以下が対話を実行してみた結果です
完璧ではおそらくないですが、少しはキャンペーンの見逃しの防止にはなりそうです！
--- ショッピングアドバイザー起動 (設定済み決済: d払い, 楽天Pay) ---
どこにお出かけしますか？ (例: ららぽーと豊洲) / exitで終了: 新宿 高島屋
「新宿 高島屋」のお得な情報を確認中...
新宿高島屋でご利用いただけるお得なキャンペーンについてご案内します。
まず、新宿高島屋に入っている主要なショップ・テナントは以下の通りです。
百貨店、専門店、ハンズ新宿店、ノジマ 新宿タカシマヤ タイムズスクエア店、ユニクロ 新宿高島屋店、ABCクッキングスタジオ、ボディーズ、エステティック&スパビューティアベニュー ソシエ、ヘアーカッティングガーデンジャック・モアザン、ホワイトエッセンス、レストラン・カフェ。
NEW_SHOPS: 新宿 高島屋: [百貨店, 専門店, ハンズ新宿店, ノジマ 新宿タカシマヤ タイムズスクエア店, ユニクロ 新宿高島屋店, ABCクッキングスタジオ, ボディーズ, エステティック&スパビューティアベニュー ソシエ, ヘアーカッティングガーデンジャック・モアザン, ホワイトエッセンス, レストラン・カフェ]
新宿高島屋では、d払いと楽天ペイのどちらもご利用いただけます。現在開催中のお得なキャンペーンは以下の通りです。
**d払いのおすすめキャンペーン**
* **髙島屋グループ｜抽選で最大2,000pt！**
dポイントカードをご提示の上、期間中合計100ポイント以上ためると、抽選で最大2,000ポイントが当たります。このキャンペーンは2026年2月17日(火)まで開催されています。新宿高島屋も対象店舗です。
* **dポイント最大4％～抽選最大24％還元！ (d曜日)**
毎週金・土曜日にd払いを利用すると、合計最大4.5%のdポイントが還元されます。Webエントリーが必要です。
* **最大全額還元当たる！今すぐエントリー**
d払いのお支払い方法をdカードに設定し、街のお店で合計5,000円（税込）以上d払いをご利用いただくと、抽選で最大全額（上限10万ポイント）が当たるチャンスがあります。毎月のエントリーが必要です。このキャンペーンは2026年3月31日（火）まで開催されています。
* **ノジマで最大10,000ポイントプレゼント**
新宿高島屋タイムズスクエア店にあるノジマで、ノジマアプリとdポイントカードを提示してdポイントを期間中合計5ポイント以上ためると、抽選で最大10,000ポイントが当たります。1等～3等以外の方にも5ポイントがプレゼントされます。このキャンペーンは2026年2月28日（土）までです。
**楽天ペイのおすすめキャンペーン**
* **5と0のつく日は！お支払いが全額**
毎月5と0のつく日に事前エントリーの上、楽天ペイアプリでコード・QR・セルフ払いを行うと、最大3.5%の楽天ポイントが還元されます。
* **ご契約者限定！抽選で総額200万が当たる最強ドリーム！**
楽天モバイルご契約者様限定で、キャンペーン期間中にエントリーし、楽天ペイのコード・QR・セルフ払いで合計1,500円（税込）以上お支払いいただくと、抽選で総額200万ポイントが当たります。楽天ペイをはじめて利用する方は当選確率が10倍になります。キャンペーン期間は2026年1月6日（火）10:00～2026年2月2日（月）9:59です。
**最もお得な支払い方法の提案**
新宿高島屋でのお買い物では、**d払い**が最もお得になる可能性が高いです。
* **特定の曜日でお得に：** 毎週金・土曜日であれば、d払いの「d曜日」キャンペーンで最大4.5%のポイント還元が期待できます。これは楽天ペイの「5と0のつく日」の3.5%よりも高い還元率です。
* **高島屋限定キャンペーン：** d払いでは「髙島屋グループ｜抽選で最大2,000pt！」という、新宿高島屋を含む高島屋グループを対象としたキャンペーンが開催されており、特定の店舗への来店特典があるのは魅力的です。
* **高額利用でチャンス：** dカードをお持ちで5,000円（税込）以上の利用が見込まれる場合、「最大全額還元当たる！」キャンペーンで最大10万ポイントが当たる抽選に参加できるのも大きなメリットです。
* **家電量販店での利用：** もし新宿高島屋タイムズスクエア内のノジマでのお買い物を検討されている場合は、d払いのノジマ限定キャンペーンも活用できます。
ただし、楽天モバイルをご契約されている場合は、「ご契約者限定！抽選で総額200万が当たる最強ドリーム！」で高額ポイントが当たるチャンスがあるため、楽天ペイも検討に値します。
結論として、金・土曜日にお買い物をする場合は「d曜日」キャンペーンがある**d払い**、それ以外の日に高額の買い物をされる場合は「最大全額還元」の抽選チャンスがある**d払い**、または楽天モバイルユーザーであれば「最強ドリーム」の抽選チャンスがある**楽天ペイ**がおすすめです。それぞれのキャンペーンはエントリーが必要な場合が多いので、ご利用前に必ずご確認ください。
(システム: 新宿 高島屋 のショップ情報をキャッシュに保存しました)
まとめ
APIなどで提供されていないキャンペーン情報をPlaywrightMCPを用いて収集し、
その情報をもとにキャンペーンについて提案してくれるエージェントを作ってみました
今後、AIが企業の情報を参照して提案する機会は増えていくと思います
発信側の企業としてもAI向けの情報発信を意識していくことが
徐々に必要になってくるのではと思いました
今すぐできることとして、キャンペーンの画像内にテキストを埋め込む場合などに
同じ情報をテキストとしてもサイトに記載するなどが有効ではないかと思います
Discussion