From 87efbcb03de4316a3ecf1e3dac21a23e7a55b052 Mon Sep 17 00:00:00 2001 From: Gilles Soulier Date: Thu, 28 May 2026 06:37:38 +0200 Subject: [PATCH] feat(mcp): 6 outils shopping + tests Co-Authored-By: Claude Sonnet 4.6 --- backend/app/api/mcp_server.py | 176 +++++++++++++++++++++++++++++++++- backend/tests/test_mcp.py | 75 +++++++++++++++ 2 files changed, 250 insertions(+), 1 deletion(-) diff --git a/backend/app/api/mcp_server.py b/backend/app/api/mcp_server.py index 9796537..f4a54ae 100644 --- a/backend/app/api/mcp_server.py +++ b/backend/app/api/mcp_server.py @@ -4,12 +4,13 @@ from datetime import datetime, timedelta, timezone, date as date_type from decimal import Decimal from mcp.server.fastmcp import FastMCP -from sqlalchemy import select, and_, text +from sqlalchemy import select, and_, text, or_ from sqlalchemy.orm import selectinload from app.core.database import AsyncSessionLocal from app.models.todos import TodoItem from app.models.notes import NoteItem +from app.models.shopping import ShoppingList, ListItem, Product _VALID_STATUSES = {"pending", "done", "cancelled"} _VALID_PRIORITIES = {"low", "medium", "high"} @@ -332,3 +333,176 @@ async def delete_note(id: str) -> str: await session.delete(note) await session.commit() return _dumps({"deleted": id}) + + +# ── SHOPPING ──────────────────────────────────────────────────────────────────── + +@mcp.tool() +async def get_shopping_lists() -> str: + """Retourne toutes les listes de courses avec compteurs d'articles cochés/total.""" + async with AsyncSessionLocal() as session: + stmt = ( + select(ShoppingList) + .options(selectinload(ShoppingList.items)) + .order_by(ShoppingList.created_at.desc()) + ) + result = await session.execute(stmt) + lists = result.scalars().all() + return _dumps([{ + "id": lst.id, + "name": lst.name, + "status": lst.status, + "week_date": lst.week_date, + "created_at": lst.created_at, + "item_count": len(lst.items), + "checked_count": sum(1 for i in lst.items if i.is_checked), + } for lst in lists]) + + +@mcp.tool() +async def get_active_shopping_list() -> str: + """Retourne la première liste en statut 'draft' avec tous ses articles triés.""" + async with AsyncSessionLocal() as session: + stmt = ( + select(ShoppingList) + .where(ShoppingList.status == "draft") + .options(selectinload(ShoppingList.items).selectinload(ListItem.product)) + .order_by(ShoppingList.created_at.desc()) + .limit(1) + ) + result = await session.execute(stmt) + lst = result.scalar_one_or_none() + if not lst: + return _dumps({"error": "Aucune liste active (statut draft)"}) + sorted_items = sorted(lst.items, key=lambda i: (i.sort_order or 999, str(i.id))) + return _dumps({ + "id": lst.id, + "name": lst.name, + "status": lst.status, + "item_count": len(lst.items), + "checked_count": sum(1 for i in lst.items if i.is_checked), + "items": [{ + "id": item.id, + "name": item.custom_name or (item.product.name if item.product else "Article inconnu"), + "quantity": item.quantity, + "unit": item.unit, + "is_checked": item.is_checked, + } for item in sorted_items], + }) + + +@mcp.tool() +async def search_products(q: str) -> str: + """Recherche dans le catalogue produits par nom, description ou catégorie.""" + async with AsyncSessionLocal() as session: + stmt = ( + select(Product) + .where( + or_( + Product.name.ilike(f"%{q}%"), + Product.description.ilike(f"%{q}%"), + Product.category.ilike(f"%{q}%"), + ) + ) + .order_by(Product.frequency_score.desc()) + .limit(20) + ) + result = await session.execute(stmt) + products = result.scalars().all() + return _dumps([{ + "id": p.id, + "name": p.name, + "brand": p.brand, + "category": p.category, + "default_unit": p.default_unit, + "frequency_score": p.frequency_score, + "last_purchased_at": p.last_purchased_at, + } for p in products]) + + +@mcp.tool() +async def create_shopping_list(name: str | None = None) -> str: + """Crée une liste de courses. name auto = semaine ISO courante si absent (ex: 'S22 2026').""" + async with AsyncSessionLocal() as session: + lst = ShoppingList(name=name or _iso_week_label()) + session.add(lst) + await session.commit() + await session.refresh(lst) + return _dumps({ + "id": lst.id, + "name": lst.name, + "status": lst.status, + "created_at": lst.created_at, + }) + + +@mcp.tool() +async def add_shopping_item( + list_id: str, + name: str, + quantity: float | None = 1.0, + unit: str | None = None, +) -> str: + """Ajoute un article à une liste de courses.""" + async with AsyncSessionLocal() as session: + try: + lid = uuid.UUID(list_id) + except ValueError: + return _dumps({"error": f"UUID de liste invalide : {list_id}"}) + lst = await session.get(ShoppingList, lid) + if not lst: + return _dumps({"error": f"Liste introuvable : {list_id}"}) + item = ListItem( + list_id=lid, + custom_name=name, + quantity=Decimal(str(quantity)) if quantity is not None else None, + unit=unit, + ) + session.add(item) + await session.commit() + await session.refresh(item) + return _dumps({ + "id": item.id, + "name": item.custom_name, + "quantity": item.quantity, + "unit": item.unit, + "is_checked": item.is_checked, + }) + + +@mcp.tool() +async def check_shopping_item(list_id: str, item_id: str) -> str: + """Coche un article (marque comme acheté). Met à jour les stats du produit lié si présent.""" + async with AsyncSessionLocal() as session: + try: + lid = uuid.UUID(list_id) + iid = uuid.UUID(item_id) + except ValueError: + return _dumps({"error": "UUID invalide (list_id ou item_id)"}) + stmt = ( + select(ListItem) + .where(ListItem.id == iid, ListItem.list_id == lid) + .options(selectinload(ListItem.product)) + ) + result = await session.execute(stmt) + item = result.scalar_one_or_none() + if not item: + return _dumps({"error": f"Article introuvable : {item_id} dans liste {list_id}"}) + was_checked = item.is_checked + item.is_checked = True + if not was_checked and item.product_id: + product = await session.get(Product, item.product_id) + if product: + today = date_type.today() + if product.last_purchased_at and product.last_purchased_at < today: + days = (today - product.last_purchased_at).days + if product.avg_interval_days is None: + product.avg_interval_days = Decimal(str(days)) + else: + product.avg_interval_days = Decimal(str( + round(float(product.avg_interval_days) * 0.7 + days * 0.3, 1) + )) + product.last_purchased_at = today + product.frequency_score += 1 + await session.commit() + return _dumps({"id": item.id, "is_checked": item.is_checked}) diff --git a/backend/tests/test_mcp.py b/backend/tests/test_mcp.py index 2cfd97a..17a52cf 100644 --- a/backend/tests/test_mcp.py +++ b/backend/tests/test_mcp.py @@ -8,8 +8,13 @@ from app.api.mcp_server import ( from app.api.mcp_server import ( search_notes, get_note, create_note, update_note, delete_note, ) +from app.api.mcp_server import ( + get_shopping_lists, get_active_shopping_list, search_products, + create_shopping_list, add_shopping_item, check_shopping_item, +) from app.models.todos import TodoItem from app.models.notes import NoteItem +from app.models.shopping import ShoppingList, ListItem pytestmark = pytest.mark.usefixtures("mcp_nullpool_session") @@ -21,6 +26,7 @@ async def cleanup_mcp_todos(): async with mcp_server_module.AsyncSessionLocal() as session: await session.execute(delete(TodoItem).where(TodoItem.title.like("TEST_MCP_%"))) await session.execute(delete(NoteItem).where(NoteItem.title.like("TEST_MCP_%"))) + await session.execute(delete(ShoppingList).where(ShoppingList.name.like("TEST_MCP_%"))) await session.commit() @@ -128,3 +134,72 @@ async def test_get_note_introuvable(): result = await get_note(id="00000000-0000-0000-0000-000000000000") data = json.loads(result) assert "error" in data + + +# ── SHOPPING ────────────────────────────────────────────────────────────────── + +async def test_get_shopping_lists_retourne_liste(): + result = await get_shopping_lists() + data = json.loads(result) + assert isinstance(data, list) + + +async def test_create_shopping_list_outil(): + result = await create_shopping_list(name="TEST_MCP_liste") + data = json.loads(result) + assert data["name"] == "TEST_MCP_liste" + assert data["status"] == "draft" + assert "id" in data + + +async def test_create_shopping_list_nom_auto(): + result = await create_shopping_list() + data = json.loads(result) + assert data["name"].startswith("S") + + +async def test_add_shopping_item_outil(): + liste = json.loads(await create_shopping_list(name="TEST_MCP_liste_item")) + result = await add_shopping_item( + list_id=str(liste["id"]), + name="TEST_MCP_article", + quantity=2.0, + unit="kg", + ) + data = json.loads(result) + assert data["name"] == "TEST_MCP_article" + assert data["is_checked"] is False + assert float(data["quantity"]) == 2.0 + + +async def test_check_shopping_item_outil(): + liste = json.loads(await create_shopping_list(name="TEST_MCP_liste_check")) + article = json.loads(await add_shopping_item( + list_id=str(liste["id"]), + name="TEST_MCP_article_check", + )) + result = await check_shopping_item( + list_id=str(liste["id"]), + item_id=str(article["id"]), + ) + data = json.loads(result) + assert data["is_checked"] is True + + +async def test_search_products_retourne_liste(): + result = await search_products(q="inexistant_xyz_abc_999") + data = json.loads(result) + assert isinstance(data, list) + assert data == [] + + +async def test_get_active_shopping_list_structure(): + result = await get_active_shopping_list() + data = json.loads(result) + assert isinstance(data, dict) + + +async def test_add_item_liste_invalide(): + result = await add_shopping_item(list_id="pas-un-uuid", name="article") + data = json.loads(result) + assert "error" in data