Spaces:
Running
Running
"""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.""" | |
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)}"}] | |
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)}"} | |
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)}"}] | |
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)}"} | |
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.""" | |
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)}"}] | |
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)}"} | |
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)}"} | |
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)}"} | |
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)}"} | |
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)}"} | |
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)}"} |