gilles/home_hub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mcp): 6 outils shopping + tests

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
gilles soulier
2026-05-28 06:37:38 +02:00
parent c72ffd0ad6
commit 87efbcb03d
2 changed files with 250 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/mcp_server.py
+175 -1
View File
@@ -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})
backend/tests/test_mcp.py
+75
View File
@@ -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