feat(shopping): add Bring! shopping lists MCP server and protocol
This commit is contained in:
Moritz Graf
Executable
+545
@@ -0,0 +1,545 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Bring! Shopping Lists MCP Server & CLI Tool
|
||||
Interacts with the unofficial Bring! Shopping List API using bring-api.
|
||||
Provides single-turn MCP tools and CLI utilities to manage multiple shopping lists:
|
||||
- "Einkaufsladen" (Default for groceries / normal shopping)
|
||||
- "Baumarkt" (Hardware store, DIY, tools, repairs)
|
||||
- "Ikea" (Furniture, home decor, Swedish goods)
|
||||
- "Gewuerzeinkaufsliste" (Spices and seasoning shopping)
|
||||
Supports dynamic matching against any other list present in the user's Bring! account.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import asyncio
|
||||
from typing import Dict, Any, List, Optional, Tuple
|
||||
|
||||
import aiohttp
|
||||
from bring_api import Bring
|
||||
|
||||
DEFAULT_LIST_NAME = os.environ.get("BRING_DEFAULT_LIST", "Einkaufsladen")
|
||||
|
||||
# Common aliases mapping user shorthand / natural language to canonical list names
|
||||
LIST_ALIASES: Dict[str, str] = {
|
||||
"normal": "Einkaufsladen",
|
||||
"normal shopping": "Einkaufsladen",
|
||||
"normale shopping": "Einkaufsladen",
|
||||
"einkaufen": "Einkaufsladen",
|
||||
"einkauf": "Einkaufsladen",
|
||||
"einkaufsladen": "Einkaufsladen",
|
||||
"lebensmittel": "Einkaufsladen",
|
||||
"supermarkt": "Einkaufsladen",
|
||||
"groceries": "Einkaufsladen",
|
||||
"shopping": "Einkaufsladen",
|
||||
"baumarkt": "Baumarkt",
|
||||
"diy": "Baumarkt",
|
||||
"hardware": "Baumarkt",
|
||||
"werkzeug": "Baumarkt",
|
||||
"obi": "Baumarkt",
|
||||
"bauhaus": "Baumarkt",
|
||||
"hornbach": "Baumarkt",
|
||||
"ikea": "Ikea",
|
||||
"möbel": "Ikea",
|
||||
"moebel": "Ikea",
|
||||
"gewürz": "Gewuerzeinkaufsliste",
|
||||
"gewuerz": "Gewuerzeinkaufsliste",
|
||||
"gewürze": "Gewuerzeinkaufsliste",
|
||||
"gewuerze": "Gewuerzeinkaufsliste",
|
||||
"gewürzeinkaufsliste": "Gewuerzeinkaufsliste",
|
||||
"gewuerzeinkaufsliste": "Gewuerzeinkaufsliste",
|
||||
"spices": "Gewuerzeinkaufsliste",
|
||||
}
|
||||
|
||||
|
||||
class BringClientManager:
|
||||
"""Manages Bring API session, authentication, and list name resolution."""
|
||||
|
||||
def __init__(self, email: Optional[str] = None, password: Optional[str] = None):
|
||||
self.email = email or os.environ.get("BRING_EMAIL") or os.environ.get("MAIL")
|
||||
self.password = password or os.environ.get("BRING_PASSWORD") or os.environ.get("PW")
|
||||
self.default_list = DEFAULT_LIST_NAME
|
||||
self._lists_cache: List[Dict[str, Any]] = []
|
||||
|
||||
def _check_credentials(self):
|
||||
if not self.email or not self.password:
|
||||
raise ValueError(
|
||||
"Missing Bring credentials. Please set BRING_EMAIL and BRING_PASSWORD environment variables."
|
||||
)
|
||||
|
||||
async def _get_client_and_session(self) -> Tuple[Bring, aiohttp.ClientSession]:
|
||||
self._check_credentials()
|
||||
session = aiohttp.ClientSession()
|
||||
client = Bring(session, self.email, self.password)
|
||||
await client.login()
|
||||
return client, session
|
||||
|
||||
async def load_lists(self) -> List[Dict[str, Any]]:
|
||||
client, session = await self._get_client_and_session()
|
||||
async with session:
|
||||
res = await client.load_lists()
|
||||
lists_data = []
|
||||
for lst in getattr(res, "lists", res):
|
||||
lists_data.append({
|
||||
"listUuid": getattr(lst, "listUuid", lst.get("listUuid") if isinstance(lst, dict) else None),
|
||||
"name": getattr(lst, "name", lst.get("name") if isinstance(lst, dict) else None),
|
||||
"theme": getattr(lst, "theme", lst.get("theme") if isinstance(lst, dict) else None),
|
||||
})
|
||||
self._lists_cache = lists_data
|
||||
return lists_data
|
||||
|
||||
async def resolve_list(self, list_name: Optional[str] = None) -> Tuple[str, str]:
|
||||
"""
|
||||
Resolves a user-provided or alias list name to (listUuid, canonical_name).
|
||||
Falls back to default_list ('Einkaufsladen') if omitted or not matched.
|
||||
"""
|
||||
raw_name = (list_name or "").strip()
|
||||
alias_target = LIST_ALIASES.get(raw_name.lower()) if raw_name else self.default_list
|
||||
target_name = alias_target or raw_name or self.default_list
|
||||
|
||||
if not self._lists_cache:
|
||||
await self.load_lists()
|
||||
|
||||
# 1. Exact match against canonical / loaded names
|
||||
for item in self._lists_cache:
|
||||
if item["name"].lower() == target_name.lower():
|
||||
return item["listUuid"], item["name"]
|
||||
|
||||
# 2. Substring match
|
||||
for item in self._lists_cache:
|
||||
if target_name.lower() in item["name"].lower() or item["name"].lower() in target_name.lower():
|
||||
return item["listUuid"], item["name"]
|
||||
|
||||
# 3. Fallback to default list
|
||||
for item in self._lists_cache:
|
||||
if item["name"].lower() == self.default_list.lower():
|
||||
return item["listUuid"], item["name"]
|
||||
|
||||
# 4. If default list not found, pick first list available
|
||||
if self._lists_cache:
|
||||
first = self._lists_cache[0]
|
||||
return first["listUuid"], first["name"]
|
||||
|
||||
raise ValueError(f"No shopping lists available in Bring account for target '{target_name}'")
|
||||
|
||||
async def get_items(self, list_name: Optional[str] = None) -> Dict[str, Any]:
|
||||
client, session = await self._get_client_and_session()
|
||||
async with session:
|
||||
res = await client.load_lists()
|
||||
self._lists_cache = [
|
||||
{
|
||||
"listUuid": getattr(l, "listUuid", l.get("listUuid") if isinstance(l, dict) else None),
|
||||
"name": getattr(l, "name", l.get("name") if isinstance(l, dict) else None),
|
||||
}
|
||||
for l in getattr(res, "lists", res)
|
||||
]
|
||||
list_uuid, canonical_name = await self.resolve_list(list_name)
|
||||
list_details = await client.get_list(list_uuid)
|
||||
|
||||
# Purchase items (active to-buy)
|
||||
purchase_items = []
|
||||
for item in getattr(list_details, "purchase", []):
|
||||
purchase_items.append({
|
||||
"name": getattr(item, "name", item.get("name") if isinstance(item, dict) else str(item)),
|
||||
"specification": getattr(item, "specification", item.get("specification", "") if isinstance(item, dict) else ""),
|
||||
})
|
||||
|
||||
# Recently bought items
|
||||
recently_items = []
|
||||
for item in getattr(list_details, "recently", []):
|
||||
recently_items.append({
|
||||
"name": getattr(item, "name", item.get("name") if isinstance(item, dict) else str(item)),
|
||||
"specification": getattr(item, "specification", item.get("specification", "") if isinstance(item, dict) else ""),
|
||||
})
|
||||
|
||||
return {
|
||||
"list": canonical_name,
|
||||
"listUuid": list_uuid,
|
||||
"count": len(purchase_items),
|
||||
"items": purchase_items,
|
||||
"recently": recently_items,
|
||||
}
|
||||
|
||||
async def add_item(self, item_name: str, specification: str = "", list_name: Optional[str] = None) -> Dict[str, Any]:
|
||||
client, session = await self._get_client_and_session()
|
||||
async with session:
|
||||
res = await client.load_lists()
|
||||
self._lists_cache = [
|
||||
{
|
||||
"listUuid": getattr(l, "listUuid", l.get("listUuid") if isinstance(l, dict) else None),
|
||||
"name": getattr(l, "name", l.get("name") if isinstance(l, dict) else None),
|
||||
}
|
||||
for l in getattr(res, "lists", res)
|
||||
]
|
||||
list_uuid, canonical_name = await self.resolve_list(list_name)
|
||||
await client.save_item(list_uuid, item_name.strip(), specification.strip())
|
||||
return {
|
||||
"success": True,
|
||||
"action": "added",
|
||||
"item": item_name.strip(),
|
||||
"specification": specification.strip() or None,
|
||||
"list": canonical_name,
|
||||
"listUuid": list_uuid,
|
||||
}
|
||||
|
||||
async def add_batch(self, items: List[Dict[str, str]], list_name: Optional[str] = None) -> Dict[str, Any]:
|
||||
client, session = await self._get_client_and_session()
|
||||
async with session:
|
||||
res = await client.load_lists()
|
||||
self._lists_cache = [
|
||||
{
|
||||
"listUuid": getattr(l, "listUuid", l.get("listUuid") if isinstance(l, dict) else None),
|
||||
"name": getattr(l, "name", l.get("name") if isinstance(l, dict) else None),
|
||||
}
|
||||
for l in getattr(res, "lists", res)
|
||||
]
|
||||
list_uuid, canonical_name = await self.resolve_list(list_name)
|
||||
results = []
|
||||
for entry in items:
|
||||
name = entry.get("name") or entry.get("itemName") or ""
|
||||
spec = entry.get("specification") or entry.get("spec") or ""
|
||||
if name.strip():
|
||||
await client.save_item(list_uuid, name.strip(), spec.strip())
|
||||
results.append({"name": name.strip(), "specification": spec.strip() or None})
|
||||
return {
|
||||
"success": True,
|
||||
"action": "batch_added",
|
||||
"count": len(results),
|
||||
"items": results,
|
||||
"list": canonical_name,
|
||||
"listUuid": list_uuid,
|
||||
}
|
||||
|
||||
async def complete_item(self, item_name: str, list_name: Optional[str] = None) -> Dict[str, Any]:
|
||||
client, session = await self._get_client_and_session()
|
||||
async with session:
|
||||
res = await client.load_lists()
|
||||
self._lists_cache = [
|
||||
{
|
||||
"listUuid": getattr(l, "listUuid", l.get("listUuid") if isinstance(l, dict) else None),
|
||||
"name": getattr(l, "name", l.get("name") if isinstance(l, dict) else None),
|
||||
}
|
||||
for l in getattr(res, "lists", res)
|
||||
]
|
||||
list_uuid, canonical_name = await self.resolve_list(list_name)
|
||||
await client.complete_item(list_uuid, item_name.strip())
|
||||
return {
|
||||
"success": True,
|
||||
"action": "completed",
|
||||
"item": item_name.strip(),
|
||||
"list": canonical_name,
|
||||
"listUuid": list_uuid,
|
||||
}
|
||||
|
||||
async def remove_item(self, item_name: str, list_name: Optional[str] = None) -> Dict[str, Any]:
|
||||
client, session = await self._get_client_and_session()
|
||||
async with session:
|
||||
res = await client.load_lists()
|
||||
self._lists_cache = [
|
||||
{
|
||||
"listUuid": getattr(l, "listUuid", l.get("listUuid") if isinstance(l, dict) else None),
|
||||
"name": getattr(l, "name", l.get("name") if isinstance(l, dict) else None),
|
||||
}
|
||||
for l in getattr(res, "lists", res)
|
||||
]
|
||||
list_uuid, canonical_name = await self.resolve_list(list_name)
|
||||
await client.remove_item(list_uuid, item_name.strip())
|
||||
return {
|
||||
"success": True,
|
||||
"action": "removed",
|
||||
"item": item_name.strip(),
|
||||
"list": canonical_name,
|
||||
"listUuid": list_uuid,
|
||||
}
|
||||
|
||||
|
||||
# MCP Server Implementation
|
||||
MCP_TOOLS = [
|
||||
{
|
||||
"name": "bring_list_lists",
|
||||
"description": "List all shopping lists available in the Bring! account (e.g. Einkaufsladen, Baumarkt, Ikea, Gewuerzeinkaufsliste).",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
"required": [],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "bring_get_items",
|
||||
"description": "Get all active to-buy items on a Bring! shopping list. Defaults to 'Einkaufsladen'.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"list_name": {
|
||||
"type": "string",
|
||||
"description": "Target list name or alias ('Einkaufsladen' [default for normal shopping], 'Baumarkt' [hardware/DIY], 'Ikea', 'Gewuerzeinkaufsliste').",
|
||||
"default": "Einkaufsladen",
|
||||
}
|
||||
},
|
||||
"required": [],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "bring_add_item",
|
||||
"description": "Add an item to a Bring! shopping list. Defaults to 'Einkaufsladen' (normal shopping). Use 'Baumarkt' for hardware/tools, 'Ikea' for furniture/decor, or 'Gewuerzeinkaufsliste' for spices.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"item_name": {
|
||||
"type": "string",
|
||||
"description": "Name of the item to add (e.g. 'Milch', 'Haferflocken', 'Spax Schrauben 4x40', 'Zimt').",
|
||||
},
|
||||
"specification": {
|
||||
"type": "string",
|
||||
"description": "Optional specification or quantity (e.g. '2L', '3.8%', '100 Stück', 'Bio').",
|
||||
"default": "",
|
||||
},
|
||||
"list_name": {
|
||||
"type": "string",
|
||||
"description": "Target list name or alias ('Einkaufsladen' [default], 'Baumarkt', 'Ikea', 'Gewuerzeinkaufsliste').",
|
||||
"default": "Einkaufsladen",
|
||||
},
|
||||
},
|
||||
"required": ["item_name"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "bring_add_batch",
|
||||
"description": "Add multiple items to a Bring! shopping list at once. Useful for recipes or ingredient lists.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"items": {
|
||||
"type": "array",
|
||||
"description": "List of objects with 'name' and optional 'specification' (e.g. [{'name': 'Eier', 'specification': '10er'}, {'name': 'Butter'}])",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string", "description": "Item name"},
|
||||
"specification": {"type": "string", "description": "Quantity or details", "default": ""},
|
||||
},
|
||||
"required": ["name"],
|
||||
},
|
||||
},
|
||||
"list_name": {
|
||||
"type": "string",
|
||||
"description": "Target list name ('Einkaufsladen' [default], 'Baumarkt', 'Ikea', 'Gewuerzeinkaufsliste').",
|
||||
"default": "Einkaufsladen",
|
||||
},
|
||||
},
|
||||
"required": ["items"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "bring_complete_item",
|
||||
"description": "Mark an item as purchased / completed (moves it to recently bought) on a Bring! shopping list.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"item_name": {
|
||||
"type": "string",
|
||||
"description": "Exact name of the item to mark as completed.",
|
||||
},
|
||||
"list_name": {
|
||||
"type": "string",
|
||||
"description": "Target list name ('Einkaufsladen' [default], 'Baumarkt', 'Ikea', 'Gewuerzeinkaufsliste').",
|
||||
"default": "Einkaufsladen",
|
||||
},
|
||||
},
|
||||
"required": ["item_name"],
|
||||
},
|
||||
},
|
||||
{
|
||||
"name": "bring_remove_item",
|
||||
"description": "Completely remove an item from a Bring! shopping list.",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"item_name": {
|
||||
"type": "string",
|
||||
"description": "Exact name of the item to delete completely.",
|
||||
},
|
||||
"list_name": {
|
||||
"type": "string",
|
||||
"description": "Target list name ('Einkaufsladen' [default], 'Baumarkt', 'Ikea', 'Gewuerzeinkaufsliste').",
|
||||
"default": "Einkaufsladen",
|
||||
},
|
||||
},
|
||||
"required": ["item_name"],
|
||||
},
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def run_mcp_server():
|
||||
"""Runs standard MCP stdio JSON-RPC loop."""
|
||||
mgr = BringClientManager()
|
||||
|
||||
for line in sys.stdin:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
try:
|
||||
msg = json.loads(line)
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
msg_id = msg.get("id")
|
||||
method = msg.get("method")
|
||||
params = msg.get("params", {})
|
||||
|
||||
if method == "initialize":
|
||||
resp = {
|
||||
"jsonrpc": "2.0",
|
||||
"id": msg_id,
|
||||
"result": {
|
||||
"protocolVersion": "2024-11-05",
|
||||
"capabilities": {"tools": {}},
|
||||
"serverInfo": {"name": "bring-mcp-server", "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:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
if tool_name == "bring_list_lists":
|
||||
data = loop.run_until_complete(mgr.load_lists())
|
||||
elif tool_name == "bring_get_items":
|
||||
data = loop.run_until_complete(mgr.get_items(list_name=args.get("list_name")))
|
||||
elif tool_name == "bring_add_item":
|
||||
data = loop.run_until_complete(
|
||||
mgr.add_item(
|
||||
item_name=args.get("item_name", ""),
|
||||
specification=args.get("specification", ""),
|
||||
list_name=args.get("list_name"),
|
||||
)
|
||||
)
|
||||
elif tool_name == "bring_add_batch":
|
||||
data = loop.run_until_complete(
|
||||
mgr.add_batch(items=args.get("items", []), list_name=args.get("list_name"))
|
||||
)
|
||||
elif tool_name == "bring_complete_item":
|
||||
data = loop.run_until_complete(
|
||||
mgr.complete_item(item_name=args.get("item_name", ""), list_name=args.get("list_name"))
|
||||
)
|
||||
elif tool_name == "bring_remove_item":
|
||||
data = loop.run_until_complete(
|
||||
mgr.remove_item(item_name=args.get("item_name", ""), list_name=args.get("list_name"))
|
||||
)
|
||||
else:
|
||||
raise ValueError(f"Unknown tool: {tool_name}")
|
||||
|
||||
loop.close()
|
||||
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="Bring! Shopping Lists MCP & CLI Tool")
|
||||
subparsers = parser.add_subparsers(dest="command")
|
||||
|
||||
# lists
|
||||
subparsers.add_parser("lists", help="List all Bring! shopping lists")
|
||||
|
||||
# get
|
||||
sp_get = subparsers.add_parser("get", help="Get items from a shopping list")
|
||||
sp_get.add_argument("--list", default=DEFAULT_LIST_NAME, help="List name or alias")
|
||||
|
||||
# add
|
||||
sp_add = subparsers.add_parser("add", help="Add an item to a list")
|
||||
sp_add.add_argument("item", help="Item name")
|
||||
sp_add.add_argument("--spec", default="", help="Specification (e.g. 2L, 500g)")
|
||||
sp_add.add_argument("--list", default=DEFAULT_LIST_NAME, help="List name or alias")
|
||||
|
||||
# complete
|
||||
sp_comp = subparsers.add_parser("complete", help="Mark an item as completed")
|
||||
sp_comp.add_argument("item", help="Item name")
|
||||
sp_comp.add_argument("--list", default=DEFAULT_LIST_NAME, help="List name or alias")
|
||||
|
||||
# remove
|
||||
sp_rem = subparsers.add_parser("remove", help="Remove an item from a list")
|
||||
sp_rem.add_argument("item", help="Item name")
|
||||
sp_rem.add_argument("--list", default=DEFAULT_LIST_NAME, help="List name or alias")
|
||||
|
||||
args = parser.parse_args()
|
||||
mgr = BringClientManager()
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
|
||||
try:
|
||||
if args.command == "lists":
|
||||
res = loop.run_until_complete(mgr.load_lists())
|
||||
print(json.dumps(res, indent=2, ensure_ascii=False))
|
||||
elif args.command == "get":
|
||||
res = loop.run_until_complete(mgr.get_items(args.list))
|
||||
print(json.dumps(res, indent=2, ensure_ascii=False))
|
||||
elif args.command == "add":
|
||||
res = loop.run_until_complete(mgr.add_item(args.item, specification=args.spec, list_name=args.list))
|
||||
print(json.dumps(res, indent=2, ensure_ascii=False))
|
||||
elif args.command == "complete":
|
||||
res = loop.run_until_complete(mgr.complete_item(args.item, list_name=args.list))
|
||||
print(json.dumps(res, indent=2, ensure_ascii=False))
|
||||
elif args.command == "remove":
|
||||
res = loop.run_until_complete(mgr.remove_item(args.item, list_name=args.list))
|
||||
print(json.dumps(res, indent=2, ensure_ascii=False))
|
||||
else:
|
||||
parser.print_help()
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) == 1 or (len(sys.argv) == 2 and sys.argv[1] == "--stdio"):
|
||||
run_mcp_server()
|
||||
else:
|
||||
run_cli()
|
||||
Reference in New Issue
Block a user