#!/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()