"""FastMCP Tools for FoodWise Inventory Management. This module defines all MCP tools for CRUD operations on inventory items. Tools are the action-oriented primitives that modify data or perform operations. """ from typing import List, Dict, Any, Optional from datetime import datetime from fastmcp import FastMCP from ..notion.notion_client import FoodWiseNotionClient def register_inventory_tools(mcp: FastMCP) -> None: """Register all inventory management tools with the FastMCP server.""" @mcp.tool() def get_inventory( filter_type: Optional[str] = None, category: Optional[str] = None, storage_type: Optional[str] = None, expiring_days: Optional[int] = None ) -> List[Dict[str, Any]]: """ Retrieve inventory items with optional filtering. Use this tool to get food items from the inventory. You can filter by: - filter_type: "expiring_soon", "low_stock", or "all" - category: Specific food category (Dairy, Raw Meat, Vegetables, etc.) - storage_type: Where items are stored (Pantry, Refrigerator, Freezer) - expiring_days: Show items expiring within this many days Returns a list of inventory items with details like name, quantity, dates, storage location, and more. """ try: notion_client = FoodWiseNotionClient() # Handle different filter types if filter_type == "expiring_soon" or expiring_days: days = expiring_days or 7 return notion_client.get_expiring_items(days) # Build filter conditions filter_conditions = None if category or storage_type: conditions: List[Dict[str, Any]] = [] if category: conditions.append({ "property": "Category", "select": {"equals": category} }) if storage_type: conditions.append({ "property": "Storage Type", "select": {"equals": storage_type} }) if len(conditions) == 1: filter_conditions = conditions[0] else: filter_conditions = {"and": conditions} return notion_client.query_inventory(filter_conditions=filter_conditions) except Exception as e: return [{"error": f"Failed to retrieve inventory: {str(e)}"}] @mcp.tool() def add_inventory_item( name: str, category: str, storage_type: str, quantity: float, unit: str, best_by_date: Optional[str] = None, purchase_date: Optional[str] = None, location_shelf: Optional[str] = None, fridge_zone: Optional[str] = None, tags: Optional[List[str]] = None, temperature_sensitive: bool = False, cooked_raw_status: str = "Raw", prep_notes: Optional[str] = None ) -> Dict[str, Any]: """ Add a new food item to the inventory. Use this tool to add food items to the FoodWise inventory database. Required parameters: - name: Name of the food item - category: Food category (Dairy, Raw Meat, Vegetables, Fruits, Canned Goods, Grains, Pasta, Spices, Oils, Baking Supplies, Snacks, Condiments, Leftovers, Beverages, Prepared Meals, Ice Cream) - storage_type: Where to store it (Pantry, Refrigerator, Freezer) - quantity: How much (number) - unit: Unit of measurement (lbs, oz, pieces, cups, etc.) Optional parameters: - best_by_date: Expiration date in YYYY-MM-DD format - purchase_date: When purchased in YYYY-MM-DD format - location_shelf: Specific location like "Top shelf", "Door", etc. - fridge_zone: For refrigerated items (Main, Crisper, Dairy, etc.) - tags: List of tags like ["Organic", "Low Stock"] - temperature_sensitive: True if requires careful temperature control - cooked_raw_status: Raw, Cooked, or Partially Cooked - prep_notes: Any preparation notes Returns success status and item details. """ # Validate inputs valid_categories = [ "Dairy", "Raw Meat", "Cooked Meat", "Vegetables", "Fruits", "Canned Goods", "Grains", "Pasta", "Spices", "Oils", "Baking Supplies", "Snacks", "Condiments", "Leftovers", "Beverages", "Prepared Meals", "Ice Cream" ] if category not in valid_categories: return {"success": False, "error": f"Invalid category. Must be one of: {', '.join(valid_categories)}"} valid_storage_types = ["Pantry", "Refrigerator", "Freezer"] if storage_type not in valid_storage_types: return {"success": False, "error": f"Invalid storage type. Must be one of: {', '.join(valid_storage_types)}"} try: notion_client = FoodWiseNotionClient() item_data = { "name": name, "category": category, "storage_type": storage_type, "quantity": quantity, "unit": unit, "best_by_date": best_by_date, "purchase_date": purchase_date, "location_shelf": location_shelf, "fridge_zone": fridge_zone, "tags": tags, "temperature_sensitive": temperature_sensitive, "cooked_raw_status": cooked_raw_status, "prep_notes": prep_notes } created_item = notion_client.add_inventory_item(item_data) return { "success": True, "message": f"Successfully added {quantity} {unit} of {name} to {storage_type}", "item": created_item } except Exception as e: return {"success": False, "error": f"Failed to add item: {str(e)}"} @mcp.tool() def search_inventory_items( search_term: str, search_type: str = "name" ) -> List[Dict[str, Any]]: """ Search for specific items in the inventory. Use this tool to find items by name, category, or other criteria. Parameters: - search_term: What to search for (e.g., "milk", "vegetables", "chicken") - search_type: Type of search - "name" (default), "category", or "tag" Returns a list of matching inventory items with full details. """ try: notion_client = FoodWiseNotionClient() if search_type == "name": return notion_client.search_items(search_term) elif search_type == "category": return notion_client.query_inventory(filter_conditions={ "property": "Category", "select": {"equals": search_term} }) else: # Fallback to name search return notion_client.search_items(search_term) except Exception as e: return [{"error": f"Failed to search: {str(e)}"}] @mcp.tool() def update_inventory_item( page_id: str, updated_data: Dict[str, Any] ) -> Dict[str, Any]: """ Update an existing inventory item. Use this tool to modify existing items in the inventory. You need the item's page_id (which you can get from search or get_inventory results). Parameters: - page_id: Notion page ID of the item to update - updated_data: Dictionary of fields to update, e.g.: {"quantity": 2.5, "unit": "lbs"} {"best_by_date": "2024-08-15"} {"location_shelf": "Top shelf"} Available fields to update: - quantity, unit, best_by_date, purchase_date, opened_date - location_shelf, fridge_zone, tags, temperature_sensitive - cooked_raw_status, prep_notes Returns success status and updated item details. """ try: notion_client = FoodWiseNotionClient() updated_item = notion_client.update_inventory_item(page_id, updated_data) return { "success": True, "message": f"Item {page_id} updated successfully", "item": updated_item } except Exception as e: return {"success": False, "error": f"Failed to update item: {str(e)}"} @mcp.tool() def remove_inventory_item(page_id: str) -> Dict[str, Any]: """ Remove an inventory item from the database. Use this tool to delete items from the inventory. This archives the item in Notion rather than permanently deleting it. Parameters: - page_id: Notion page ID of the item to remove (get this from search results) Returns success status and confirmation message. """ try: notion_client = FoodWiseNotionClient() result = notion_client.remove_inventory_item(page_id) return result except Exception as e: return {"success": False, "error": f"Failed to remove item: {str(e)}"} def register_shopping_tools(mcp: FastMCP) -> None: """Register all shopping list management tools with the FastMCP server.""" @mcp.tool() def get_shopping_list( store: Optional[str] = None, priority: Optional[str] = None, status: Optional[str] = None ) -> List[Dict[str, Any]]: """ Retrieve shopping list items with optional filtering. Use this tool to get items from your shopping list. You can filter by: - store: Specific store (Costco, Trader Joe's, Safeway, Whole Foods, Target, etc.) - priority: Item priority (Essential, Nice to Have, Stock Up) - status: Purchase status (Needed, Purchased, Skipped) Returns a list of shopping items with details like quantity, store preference, estimated price, and purchase status. """ try: notion_client = FoodWiseNotionClient() # Build filter conditions for shopping list filter_conditions = None if store or priority or status: conditions: List[Dict[str, Any]] = [] if store: conditions.append({ "property": "Store", "select": {"equals": store} }) if priority: conditions.append({ "property": "Priority", "select": {"equals": priority} }) if status: conditions.append({ "property": "Status", "select": {"equals": status} }) if len(conditions) == 1: filter_conditions = conditions[0] else: filter_conditions = {"and": conditions} return notion_client.query_shopping_list(filter_conditions) except Exception as e: return [{"error": f"Failed to retrieve shopping list: {str(e)}"}] @mcp.tool() def add_shopping_item( item_name: str, quantity: float, unit: str, store: Optional[str] = None, priority: str = "Essential", category: Optional[str] = None, estimated_price: Optional[float] = None, brand_preference: Optional[str] = None, size_package: Optional[str] = None, notes: Optional[str] = None, recipe_source: Optional[str] = None ) -> Dict[str, Any]: """ Add a new item to the shopping list. Use this tool to add items you need to buy. Specify: - item_name: Name of the item to buy - quantity: How much you need - unit: Measurement unit (pieces, lbs, gallons, etc.) - store: Preferred store (Costco, Trader Joe's, Safeway, Whole Foods, Target) - priority: Essential, Nice to Have, or Stock Up (default: Essential) - category: Food category (Produce, Dairy, Meat, Pantry, etc.) - estimated_price: Expected cost for budgeting - brand_preference: Specific brand if important - size_package: Package size details (24-pack, organic, family size) - notes: Any additional shopping notes - recipe_source: Which recipe needs this ingredient Returns the created shopping item with confirmation. """ try: notion_client = FoodWiseNotionClient() item_data = { "name": item_name, "quantity": quantity, "unit": unit, "priority": priority, "status": "Needed" # Default status for new items } # Add optional fields if provided if store: item_data["store"] = store if category: item_data["category"] = category if estimated_price: item_data["estimated_price"] = estimated_price if brand_preference: item_data["brand_preference"] = brand_preference if size_package: item_data["size_package"] = size_package if notes: item_data["notes"] = notes if recipe_source: item_data["recipe_source"] = recipe_source new_item = notion_client.add_shopping_item(item_data) return { "success": True, "message": f"Added '{item_name}' to shopping list", "item": new_item } except Exception as e: return {"success": False, "error": f"Failed to add shopping item: {str(e)}"} @mcp.tool() def update_shopping_item( page_id: str, quantity: Optional[float] = None, unit: Optional[str] = None, store: Optional[str] = None, priority: Optional[str] = None, status: Optional[str] = None, estimated_price: Optional[float] = None, brand_preference: Optional[str] = None, size_package: Optional[str] = None, notes: Optional[str] = None, recipe_source: Optional[str] = None ) -> Dict[str, Any]: """ Update an existing shopping list item. Use this tool to modify shopping list items. You can update: - quantity, unit, store, priority, status - estimated_price, brand_preference, size_package - notes, recipe_source Parameters: - page_id: Notion page ID of the shopping item (get from search results) - quantity: New quantity value - unit: New unit of measurement - store: New store preference - priority: New priority (Essential, Nice to Have, Stock Up) - status: New status (Needed, Purchased, Skipped) - estimated_price: New estimated price - brand_preference: New brand preference - size_package: New package size details - notes: New notes - recipe_source: New recipe source Common status updates: "Needed" → "Purchased" → move to inventory """ try: notion_client = FoodWiseNotionClient() # Build update data from provided parameters updated_data = {} if quantity is not None: updated_data["quantity"] = quantity if unit is not None: updated_data["unit"] = unit if store is not None: updated_data["store"] = store if priority is not None: updated_data["priority"] = priority if status is not None: updated_data["status"] = status if estimated_price is not None: updated_data["estimated_price"] = estimated_price if brand_preference is not None: updated_data["brand_preference"] = brand_preference if size_package is not None: updated_data["size_package"] = size_package if notes is not None: updated_data["notes"] = notes if recipe_source is not None: updated_data["recipe_source"] = recipe_source if not updated_data: return {"success": False, "error": "No valid update data provided"} updated_item = notion_client.update_shopping_item(page_id, updated_data) return { "success": True, "message": f"Shopping item {page_id} updated successfully", "item": updated_item } except Exception as e: return {"success": False, "error": f"Failed to update shopping item: {str(e)}"} @mcp.tool() def remove_shopping_item(page_id: str) -> Dict[str, Any]: """ Remove an item from the shopping list. Use this tool to delete items from the shopping list. This archives the item in Notion rather than permanently deleting it. Parameters: - page_id: Notion page ID of the shopping item to remove Returns success status and confirmation message. """ try: notion_client = FoodWiseNotionClient() result = notion_client.remove_shopping_item(page_id) return result except Exception as e: return {"success": False, "error": f"Failed to remove shopping item: {str(e)}"} @mcp.tool() def create_shopping_from_recipe( recipe_ingredients: List[str], servings: int = 4, preferred_store: Optional[str] = None ) -> Dict[str, Any]: """ Generate shopping list from recipe ingredients. Use this tool to create shopping list items from a recipe. It will: - Check current inventory to avoid buying items you already have - Calculate quantities based on desired servings - Set appropriate store preferences Parameters: - recipe_ingredients: List of ingredients needed for recipe - servings: Number of servings to prepare (default: 4) - preferred_store: Store preference for all items Returns list of items added to shopping list. """ try: notion_client = FoodWiseNotionClient() # Get current inventory to check what we already have current_inventory = notion_client.query_inventory() inventory_names = [item.get('name', '').lower() for item in current_inventory] added_items = [] skipped_items = [] for ingredient in recipe_ingredients: ingredient_lower = ingredient.lower() # Check if we already have this ingredient if any(ingredient_lower in inv_name for inv_name in inventory_names): skipped_items.append(ingredient) continue # Add to shopping list item_data = { "name": ingredient, "quantity": 1, # Default quantity, user can adjust "unit": "item", # Default unit "priority": "Essential", "status": "Needed", "notes": f"For recipe (serves {servings})" } if preferred_store: item_data["store"] = preferred_store new_item = notion_client.add_shopping_item(item_data) added_items.append(new_item) return { "success": True, "message": f"Added {len(added_items)} items to shopping list", "added_items": added_items, "skipped_items": skipped_items, "note": f"Skipped {len(skipped_items)} items already in inventory" } except Exception as e: return {"success": False, "error": f"Failed to create shopping list from recipe: {str(e)}"} @mcp.tool() def optimize_shopping_by_store() -> Dict[str, Any]: """ Organize current shopping list by store for efficient shopping. Use this tool to get your shopping list organized by store with: - Items grouped by preferred store - Store-specific shopping tips and strategies - Route optimization suggestions - Total estimated costs per store Returns organized shopping plan with store-specific recommendations. """ try: notion_client = FoodWiseNotionClient() # Get all needed shopping items needed_items = notion_client.query_shopping_list({ "property": "Status", "select": {"equals": "Needed"} }) # Group by store stores: Dict[str, List[Dict[str, Any]]] = {} total_by_store: Dict[str, float] = {} for item in needed_items: store = item.get('store', 'Unassigned') if store not in stores: stores[store] = [] total_by_store[store] = 0 stores[store].append(item) # Add to total if price is available price = item.get('estimated_price') if price and isinstance(price, (int, float)): total_by_store[store] += price # Add store-specific tips store_tips = { "Costco": "Bulk buying - check freezer space. Bring membership card. Best for non-perishables and meat.", "Trader Joe's": "Unique specialty items. Great for prepared foods and seasonal items. Check weekly flyer.", "Safeway": "Full selection grocery. Good for weekly essentials. Check for digital coupons.", "Whole Foods": "Organic and premium items. Best produce selection. Amazon Prime discounts available.", "Target": "Household items + groceries. Good for packaged goods. RedCard savings available." } return { "success": True, "shopping_plan": stores, "estimated_totals": total_by_store, "store_tips": {store: tip for store, tip in store_tips.items() if store in stores}, "total_items": len(needed_items), "recommended_order": ["Costco", "Trader Joe's", "Safeway", "Whole Foods", "Target"] # Bulk → specialty → regular } except Exception as e: return {"success": False, "error": f"Failed to optimize shopping list: {str(e)}"} @mcp.tool() def move_purchased_to_inventory( purchased_item_ids: List[str] ) -> Dict[str, Any]: """ Move purchased items from shopping list to inventory. Use this tool after shopping to: - Mark shopping items as "Purchased" - Add corresponding items to inventory - Clean up the shopping list Parameters: - purchased_item_ids: List of Notion page IDs for purchased shopping items Returns summary of items moved to inventory. """ try: notion_client = FoodWiseNotionClient() moved_items = [] errors = [] for item_id in purchased_item_ids: try: # Get shopping item details shopping_items = notion_client.query_shopping_list({ "property": "ID", "rich_text": {"equals": item_id} }) if not shopping_items: errors.append(f"Shopping item {item_id} not found") continue shopping_item = shopping_items[0] # Create inventory item from shopping item inventory_data = { "name": shopping_item.get('name'), "quantity": shopping_item.get('quantity', 1), "unit": shopping_item.get('unit', 'item'), "category": shopping_item.get('category', 'Pantry'), # Default category "storage_type": "Pantry", # Default storage, user can update "purchase_date": datetime.now().strftime("%Y-%m-%d"), "tags": ["Recently Purchased"] } # Add to inventory new_inventory_item = notion_client.add_inventory_item(inventory_data) # Mark shopping item as purchased notion_client.update_shopping_item(item_id, {"status": "Purchased"}) moved_items.append({ "shopping_item": shopping_item.get('name'), "inventory_item": new_inventory_item }) except Exception as item_error: errors.append(f"Error processing {item_id}: {str(item_error)}") return { "success": True, "message": f"Moved {len(moved_items)} items to inventory", "moved_items": moved_items, "errors": errors if errors else None } except Exception as e: return {"success": False, "error": f"Failed to move items to inventory: {str(e)}"}