rms_product_ranking
Rank products by sales revenue for a specified date range, returning the top N products.
Instructions
Product sales ranking by revenue (from PackageModelList)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| start_date | No | YYYY-MM-DD | |
| end_date | No | YYYY-MM-DD | |
| top_n | No | Top N |
Implementation Reference
- rms_mcp/server.py:122-150 (handler)Handler function that fetches orders, aggregates product sales from PackageModelList/ItemModelList, ranks by revenue, and returns a formatted markdown table.
async def _product_ranking(args: dict, api: OrderAPI) -> list[TextContent]: now = _now() start = datetime.fromisoformat(args.get("start_date", (now - timedelta(days=30)).strftime("%Y-%m-%d"))) end = datetime.fromisoformat(args.get("end_date", now.strftime("%Y-%m-%d"))) end = end.replace(hour=23, minute=59, second=59) top_n = args.get("top_n", 20) orders = _fetch_all_orders(api, start, end, ACTIVE_PROGRESS) if not orders: return [TextContent(type="text", text="No orders found.")] ps: dict[str, dict] = defaultdict(lambda: {"n": "", "q": 0, "r": 0}) for order in orders: for pkg in order.get("PackageModelList", []): for item in pkg.get("ItemModelList", []): nm = item.get("itemName", "?") qty = item.get("units", 0) pr = item.get("price", 0) key = f"{item.get('itemNumber','')}:{nm}" ps[key]["n"] = nm ps[key]["q"] += qty ps[key]["r"] += qty * pr ranked = sorted(ps.items(), key=lambda x: x[1]["r"], reverse=True)[:top_n] lines = [f"# RMS Product Ranking: {start.date()} ~ {end.date()}\n| # | Product | Qty | Revenue | Avg |\n|---|---|---|---|---|"] for i, (_, s) in enumerate(ranked, 1): avg = s["r"] // s["q"] if s["q"] else 0 lines.append(f"| {i} | {s['n']} | {s['q']} | ¥{s['r']:,} | ¥{avg:,} |") return [TextContent(type="text", text="\n".join(lines))] - rms_mcp/server.py:56-61 (schema)Tool definition and input schema for rms_product_ranking, declaring start_date, end_date (strings) and top_n (integer, default 20).
Tool(name="rms_product_ranking", description="Product sales ranking by revenue (from PackageModelList)", inputSchema={"type": "object", "properties": { "start_date": {"type": "string", "description": "YYYY-MM-DD"}, "end_date": {"type": "string", "description": "YYYY-MM-DD"}, "top_n": {"type": "integer", "description": "Top N", "default": 20}, }}), - rms_mcp/server.py:80-81 (registration)Registration/dispatch in call_tool that routes the 'rms_product_ranking' name to the _product_ranking handler.
elif name == "rms_product_ranking": return await _product_ranking(arguments, api)