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