"""FastMCP Resources for FoodWise Inventory Management. This module defines all MCP resources for live contextual data access. Resources provide read-only, contextual information that updates dynamically. """ from typing import Dict, Any, List from datetime import datetime from fastmcp import FastMCP from ..notion.notion_client import FoodWiseNotionClient def register_resources(mcp: FastMCP) -> None: """Register MCP resources for live contextual data.""" @mcp.resource("inventory://items") def get_all_inventory_items() -> Dict[str, Any]: """ Get all inventory items as a resource. This resource provides access to the complete inventory for context. Use this when you need to understand the full state of the inventory. """ try: notion_client = FoodWiseNotionClient() items = notion_client.query_inventory() return { "items": items, "count": len(items), "last_updated": datetime.now().isoformat() } except Exception as e: return {"error": f"Failed to retrieve inventory: {e}", "items": []} @mcp.resource("inventory://expiring-items") def get_expiring_items() -> Dict[str, Any]: """ Get items expiring in the next 7 days for immediate context. This resource provides real-time expiration alerts to help prioritize items that need to be used soon, avoiding repeated tool calls. """ try: notion_client = FoodWiseNotionClient() expiring_items = notion_client.get_expiring_items(7) # Organize by urgency - handle missing or invalid days_until_expiry urgent = [] moderate = [] upcoming = [] for item in expiring_items: days_until = item.get('days_until_expiry') # Handle missing or non-numeric values gracefully if days_until is None or not isinstance(days_until, (int, float)): continue if days_until <= 2: urgent.append(item) elif days_until <= 5: moderate.append(item) elif days_until <= 7: upcoming.append(item) return { "total_expiring": len(expiring_items), "urgent_items": urgent, # 0-2 days "moderate_items": moderate, # 3-5 days "upcoming_items": upcoming, # 6-7 days "last_checked": datetime.now().isoformat() } except Exception as e: return { "error": f"Failed to get expiring items: {e}", "total_expiring": 0, "urgent_items": [], "moderate_items": [], "upcoming_items": [], "last_checked": datetime.now().isoformat() } def register_shopping_resources(mcp: FastMCP) -> None: """Register MCP resources for shopping list data.""" @mcp.resource("shopping://list") def get_shopping_list() -> Dict[str, Any]: """ Get the current shopping list as a resource. This resource provides access to all shopping list items for context. Use this when you need to understand what's on the shopping list. """ try: notion_client = FoodWiseNotionClient() items = notion_client.query_shopping_list() # Organize by status needed = [item for item in items if item.get('status') == 'Needed'] purchased = [item for item in items if item.get('status') == 'Purchased'] return { "items": items, "needed_items": needed, "purchased_items": purchased, "total_count": len(items), "needed_count": len(needed), "last_updated": datetime.now().isoformat() } except Exception as e: return {"error": f"Failed to retrieve shopping list: {e}", "items": []} @mcp.resource("shopping://by-store") def get_shopping_by_store() -> Dict[str, Any]: """ Get shopping list organized by store for efficient shopping trips. This resource groups items by preferred store to help plan shopping routes and optimize trips to multiple stores. """ try: notion_client = FoodWiseNotionClient() # Get only needed items needed_items = notion_client.query_shopping_list({ "property": "Status", "select": {"equals": "Needed"} }) # Group by store stores: Dict[str, List[Dict[str, Any]]] = {} unassigned: List[Dict[str, Any]] = [] total_by_store: Dict[str, Dict[str, Any]] = {} for item in needed_items: store = item.get('store') if not store: unassigned.append(item) continue if store not in stores: stores[store] = [] total_by_store[store] = {"items": 0, "estimated_cost": 0} stores[store].append(item) total_by_store[store]["items"] += 1 # Add to estimated cost if available price = item.get('estimated_price') if price and isinstance(price, (int, float)): total_by_store[store]["estimated_cost"] += price return { "stores": stores, "unassigned_items": unassigned, "store_totals": total_by_store, "total_needed": len(needed_items), "last_updated": datetime.now().isoformat() } except Exception as e: return { "error": f"Failed to get shopping by store: {e}", "stores": {}, "unassigned_items": [], "store_totals": {}, "total_needed": 0, "last_updated": datetime.now().isoformat() }