Spaces:
Running
Running
File size: 6,429 Bytes
719b2f9 8dfdd03 ab9ff97 8dfdd03 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 |
"""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()
} |