moritz/mop-skills
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(skills): add shopping skill with supermarket deals, geizhals, and ebay MCP servers

Browse Source
This commit is contained in:
Moritz Graf
2026-09-10 13:43:29 +02:00
parent 3d78c94a58
commit e8c0046c51
5 changed files with 1352 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/shopping/scripts/geizhals_mcp.py
Executable
+399
View File
@@ -0,0 +1,399 @@
#!/usr/bin/env python3
"""
Geizhals MCP Server & CLI Tool
Searches Geizhals (and Skinflint) price comparison data: best prices across shops,
merchant counts, star ratings, test review counts, and price history trends.
Communicates directly with the official Geizhals mobile app API (api.geizhals.net/gh/v9)
using offline HMAC-SHA256 request signing (bypassing Cloudflare JS challenges).
Requires zero external Python dependencies (pure standard library).
"""
import os
import sys
import json
import time
import hmac
import hashlib
import base64
import urllib.request
import urllib.parse
from typing import Dict, Any, List, Optional
BASE_URL = "https://api.geizhals.net/gh/v9"
API_HOST = "api.geizhals.net"
GH_TOKEN_ID = "mobileapp-2026-android-3.12.7"
GH_AUTH_ROLE = "mobileapp-2026-android-3.12.7"
GH_HMAC_SECRET = "2Uq.p2gD1z5Im,m73u5eSVzeuFZJpcQVtzx.M3OtyW.4LvzO"
GH_CLIENT = "Geizhals/3.13.1 (Android 14; sdk_gphone64_x86_64; emu64xa; x86_64; de)"
DEFAULT_COUNTRY = os.environ.get("GEIZHALS_DEFAULT_LOC", "de")
def _b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def _auth_header(body_bytes: bytes, method: str = "POST", query_string: str = "") -> str:
now = int(time.time())
if method == "POST":
s = f"POST|{API_HOST}||{body_bytes.decode('utf-8')}"
else:
s = f"{method}|{API_HOST}|{query_string}|"
fp = hashlib.sha256(s.encode("utf-8")).hexdigest()
header = {"alg": "HS256", "typ": "JWT"}
payload = {
"token_id": GH_TOKEN_ID,
"timestamp": now,
"request_fingerprint": fp,
"iat": now,
}
header_b64 = _b64url(json.dumps(header, separators=(",", ":")).encode())
payload_b64 = _b64url(json.dumps(payload, separators=(",", ":")).encode())
signing_input = f"{header_b64}.{payload_b64}"
sig = hmac.new(GH_HMAC_SECRET.encode("utf-8"), signing_input.encode("utf-8"), hashlib.sha256).digest()
return f"Bearer {signing_input}.{_b64url(sig)}"
def _post(endpoint: str, payload: dict) -> dict:
body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
req = urllib.request.Request(
f"{BASE_URL}/{endpoint}",
data=body,
headers={
"User-Agent": GH_CLIENT,
"Authorization": _auth_header(body),
"auth-role": GH_AUTH_ROLE,
"Content-Type": "application/json"
}
)
with urllib.request.urlopen(req, timeout=20) as resp:
return json.loads(resp.read().decode("utf-8"))
class GeizhalsClient:
def search(self, query: str, country: Optional[str] = None, limit: int = 8) -> Dict[str, Any]:
loc = (country or DEFAULT_COUNTRY).lower()
# Geizhals strictly requires pagesize to be in [1, 5, 10, 30, 100, 300, 1000]
valid_pagesizes = [1, 5, 10, 30, 100]
pagesize = 10
for ps in valid_pagesizes:
if ps >= limit:
pagesize = ps
break
payload = {
"query": query,
"params": {
"loc": loc,
"hloc": [loc],
"lang": "de",
"pagesize": pagesize,
"n_offers": 1
}
}
try:
raw = _post("search_product", payload)
except Exception as e:
return {"error": f"Geizhals search error: {str(e)}", "query": query, "results": []}
resp = raw.get("response", {})
products = resp.get("products", [])
total = raw.get("total", len(products))
results = []
domain = "geizhals.de" if loc == "de" else "geizhals.at"
for p in products[:limit]:
prices = p.get("prices") or {}
best_price = prices.get("best")
avg_price = prices.get("avg")
offers_list = p.get("offers") or []
offer_count = p.get("offer_count", len(offers_list))
# If best_price is missing in prices dict, check first offer
cheapest_shop = None
if offers_list:
first_offer = offers_list[0]
cheapest_shop = first_offer.get("shop", {}).get("name")
if best_price is None:
best_price = first_offer.get("pricing", {}).get("eur_value") or first_offer.get("price", {}).get("value")
# Determine product link
urls = p.get("urls") or {}
overview_path = urls.get("overview")
web_url = f"https://{domain}{overview_path}" if overview_path else None
# Test review count & metascore
reviews = p.get("test_reviews") or []
metascore = p.get("average_test_reviews_metascore")
# Image handling (already starts with https://gzhls.at or relative)
raw_images = p.get("images") or []
image_url = None
if raw_images:
first_img = raw_images[0]
if isinstance(first_img, str):
image_url = first_img if first_img.startswith("http") else f"https://gzhls.at/{first_img}"
elif isinstance(first_img, dict):
image_url = first_img.get("url")
results.append({
"id": p.get("gzhid"),
"name": p.get("product") or p.get("product_for_sort"),
"manufacturer": p.get("manufacturer_name"),
"best_price": float(best_price) if best_price is not None else None,
"best_price_formatted": f"{float(best_price):.2f} €".replace(".", ",") if best_price is not None else "Kein Live-Angebot",
"average_price": float(avg_price) if avg_price is not None else None,
"cheapest_shop": cheapest_shop,
"offer_count": offer_count,
"available": bool(best_price is not None and offer_count > 0),
"rating_stars": p.get("rating_stars"),
"rating_count": p.get("rating_count"),
"test_reviews_count": len(reviews),
"test_reviews_metascore": metascore,
"url": web_url,
"image_url": image_url
})
return {
"query": query,
"country": loc,
"total_found": total,
"returned_count": len(results),
"results": results
}
def price_history(self, product_id: int, days: int = 31, country: Optional[str] = None) -> Dict[str, Any]:
loc = (country or DEFAULT_COUNTRY).lower()
# Days must be one of 31, 91, 183, 365
valid_days = [31, 91, 183, 365]
closest_days = min(valid_days, key=lambda x: abs(x - days))
payload = {
"id": product_id,
"params": {
"days": closest_days,
"loc": loc
}
}
try:
raw = _post("price_history", payload)
except Exception as e:
return {"error": f"Price history error: {str(e)}", "product_id": product_id}
meta = raw.get("meta") or {}
series = raw.get("response") or []
min_price = meta.get("min")
max_price = meta.get("max")
current_best = meta.get("current_best")
is_all_time_low = False
if current_best and min_price:
is_all_time_low = float(current_best) <= float(min_price) + 0.05
return {
"product_id": product_id,
"country": loc,
"period_days": closest_days,
"current_best_price": float(current_best) if current_best is not None else None,
"period_min_price": float(min_price) if min_price is not None else None,
"period_max_price": float(max_price) if max_price is not None else None,
"is_all_time_low": is_all_time_low,
"data_points_count": len(series)
}
# ==============================================================================
# Model Context Protocol (MCP) stdio Server
# ==============================================================================
MCP_TOOLS = [
{
"name": "geizhals_search",
"description": "Search products on Geizhals (Germany/Austria price comparison). Returns lowest price, offer count across all stores, rating, and product link.",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Product search query, e.g. 'MacBook Air M3', 'Sony WH-1000XM5', 'Mutti Tomaten'"
},
"country": {
"type": "string",
"description": "Country code ('de' for Germany, 'at' for Austria). Defaults to 'de'.",
"default": "de"
},
"limit": {
"type": "integer",
"description": "Max number of items to return (default 8)",
"default": 8
}
},
"required": ["query"]
}
},
{
"name": "geizhals_price_history",
"description": "Get price history trends for a Geizhals product ID over 31, 91, 183, or 365 days to check if the current price is a good deal or all-time low.",
"inputSchema": {
"type": "object",
"properties": {
"product_id": {
"type": "integer",
"description": "Geizhals product ID (gzhid) obtained from geizhals_search"
},
"days": {
"type": "integer",
"description": "History timeframe in days (31, 91, 183, or 365). Defaults to 31.",
"default": 31
},
"country": {
"type": "string",
"description": "Country code ('de' or 'at'). Defaults to 'de'.",
"default": "de"
}
},
"required": ["product_id"]
}
}
]
def run_mcp_server():
client = GeizhalsClient()
while True:
line = sys.stdin.readline()
if not line:
break
line = line.strip()
if not line:
continue
try:
req = json.loads(line)
except Exception as e:
sys.stderr.write(f"Invalid JSON received: {e}\n")
continue
msg_id = req.get("id")
method = req.get("method")
params = req.get("params", {})
if method == "initialize":
resp = {
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": "geizhals-mcp", "version": "1.0.0"}
}
}
sys.stdout.write(json.dumps(resp) + "\n")
sys.stdout.flush()
elif method == "notifications/initialized":
pass
elif method == "ping":
resp = {"jsonrpc": "2.0", "id": msg_id, "result": {}}
sys.stdout.write(json.dumps(resp) + "\n")
sys.stdout.flush()
elif method == "tools/list":
resp = {
"jsonrpc": "2.0",
"id": msg_id,
"result": {"tools": MCP_TOOLS}
}
sys.stdout.write(json.dumps(resp) + "\n")
sys.stdout.flush()
elif method == "tools/call":
tool_name = params.get("name")
args = params.get("arguments", {})
try:
if tool_name == "geizhals_search":
data = client.search(
query=args.get("query", ""),
country=args.get("country"),
limit=args.get("limit", 8)
)
elif tool_name == "geizhals_price_history":
data = client.price_history(
product_id=int(args.get("product_id")),
days=int(args.get("days", 31)),
country=args.get("country")
)
else:
raise ValueError(f"Unknown tool: {tool_name}")
text_result = json.dumps(data, indent=2, ensure_ascii=False)
resp = {
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"content": [{"type": "text", "text": text_result}]
}
}
except Exception as e:
resp = {
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"isError": True,
"content": [{"type": "text", "text": f"Error executing {tool_name}: {str(e)}"}]
}
}
sys.stdout.write(json.dumps(resp) + "\n")
sys.stdout.flush()
else:
if msg_id is not None:
resp = {
"jsonrpc": "2.0",
"id": msg_id,
"error": {"code": -32601, "message": f"Method not found: {method}"}
}
sys.stdout.write(json.dumps(resp) + "\n")
sys.stdout.flush()
def run_cli():
import argparse
parser = argparse.ArgumentParser(description="Geizhals Price Comparison MCP & CLI")
subparsers = parser.add_subparsers(dest="command")
sp_search = subparsers.add_parser("search", help="Search products on Geizhals")
sp_search.add_argument("query", help="Product query")
sp_search.add_argument("--country", default="de", help="Country (de or at)")
sp_search.add_argument("--limit", type=int, default=8, help="Max results")
sp_hist = subparsers.add_parser("history", help="Get price history for product ID")
sp_hist.add_argument("product_id", type=int, help="Geizhals product ID (gzhid)")
sp_hist.add_argument("--days", type=int, default=31, help="History days (31, 91, 183, 365)")
sp_hist.add_argument("--country", default="de", help="Country (de or at)")
args = parser.parse_args()
client = GeizhalsClient()
if args.command == "search":
print(json.dumps(client.search(args.query, country=args.country, limit=args.limit), indent=2, ensure_ascii=False))
elif args.command == "history":
print(json.dumps(client.price_history(args.product_id, days=args.days, country=args.country), indent=2, ensure_ascii=False))
else:
parser.print_help()
if __name__ == "__main__":
if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"):
run_mcp_server()
else:
run_cli()