"""Serveur MCP AntiCoco — accès personnel à HelloFresh, sans noix de coco. Client visé : Hermes (Nous Research), sur le même homelab. Transport streamable-HTTP sur 127.0.0.1:$ANTICOCO_PORT/mcp. Si Hermes n'accepte que le stdio, changer le `transport=` de `mcp.run()` ci-dessous (le reste du code est identique). Le SDK MCP (>=1.x) appelle les outils sync DIRECTEMENT dans la boucle asyncio, ce qui casse l'API *synchrone* de Playwright (auth.py). On déclare donc les outils en `async` et on déporte leur corps bloquant dans un thread worker via `anyio.to_thread.run_sync`, où aucune boucle asyncio ne tourne. Valable pour stdio comme pour streamable-http. """ from __future__ import annotations import os import anyio from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP load_dotenv() from hellofresh import api, auth, filter as hf_filter, webui # noqa: E402 PORT = int(os.environ.get("ANTICOCO_PORT", "9200")) mcp = FastMCP("AntiCoco", host="0.0.0.0", port=PORT) # Interface web d'admin (édition à chaud des excludes/préférences) sur le même port. # MCP reste servi sur /mcp ; l'UI est sur / (cf. hellofresh/webui.py). webui.register(mcp) @mcp.tool() async def hf_auth_status() -> dict: """État de la connexion HelloFresh (utilise le token en cache s'il est frais).""" return await anyio.to_thread.run_sync(auth.auth_status) @mcp.tool() async def hf_login() -> dict: """S'assure d'être connecté et capture un bearer token frais. En headless (homelab), échoue si la session a expiré → refaire le login local. """ def _impl() -> dict: ok = auth.ensure_logged_in() if not ok: return {"logged_in": False, "error": "login non établi (timeout ou session expirée)"} info = auth.capture_token(force=True) return {"logged_in": True, "gateways": info.get("gateways", [])} return await anyio.to_thread.run_sync(_impl) @mcp.tool() async def hf_list_weeks() -> list[dict]: """Liste les semaines de l'abonnement encore modifiables (handle + dates).""" def _impl() -> list[dict]: with api.HelloFreshClient() as client: weeks = client.get_editable_weeks() return [{"week": d.week, "delivery_date": d.delivery_date, "cutoff_date": d.cutoff_date, "status": d.status, "editable": d.editable} for d in weeks] return await anyio.to_thread.run_sync(_impl) @mcp.tool() async def hf_account_info() -> dict: """Infos du compte HelloFresh : client, abonnement, adresse, prochaine livraison. Lecture seule. Aucune donnée de paiement sensible n'est renvoyée (méthode seulement). """ def _impl() -> dict: with api.HelloFreshClient() as client: return client.account_info() return await anyio.to_thread.run_sync(_impl) @mcp.tool() async def hf_next_delivery() -> dict: """Prochaine livraison + recettes RÉELLEMENT sélectionnées (prêt à mettre en forme). Renvoie la semaine, la date de livraison, le cutoff, et les ~4 recettes de la box avec image servable (URL corrigée), temps de prépa, allergènes, tags, `is_favorite`. Erreur stricte si la sélection est introuvable — ne propose JAMAIS de recettes non sélectionnées. """ def _impl() -> dict: with api.HelloFreshClient() as client: acct = client.account_info() nd = acct.get("next_delivery") or {} week = nd.get("week") if not week: raise RuntimeError("Aucune prochaine livraison (abonnement en pause ?).") recipes = client.get_current_selection(week) # lève si vide / KO favs = client.favorite_ids() hf_filter.annotate(recipes) for r in recipes: r.is_favorite = r.id in favs return { "week": week, "delivery_date": nd.get("date"), "cutoff": nd.get("cutoff"), "count": len(recipes), "recipes": [r.summary() for r in recipes], } return await anyio.to_thread.run_sync(_impl) @mcp.tool() async def hf_favorites() -> dict: """Recettes favorites du compte, complètes (image corrigée, allergènes, `is_favorite`). Lecture seule. Liste vide si aucun favori. Indépendant de la semaine (favoris globaux). """ def _impl() -> dict: with api.HelloFreshClient() as client: recipes = client.get_favorites() hf_filter.annotate(recipes) return { "count": len(recipes), "recipes": [r.summary() for r in recipes], } return await anyio.to_thread.run_sync(_impl) @mcp.tool() async def hf_get_menu(week: str = "") -> dict: """Toutes les recettes proposées pour une semaine, chacune annotée. `week` vide = semaine courante (format 'YYYY-Www'). Chaque recette porte `contains_excluded` (true si ingrédient banni, coco en tête) et `matched_excludes`. """ def _impl() -> dict: w = week or api.current_week() with api.HelloFreshClient() as client: recipes = client.get_menu(w) favs = client.favorite_ids() # best-effort (set() si indispo) hf_filter.annotate(recipes) for r in recipes: r.is_favorite = r.id in favs return { "week": w, "count": len(recipes), "recipes": [r.summary() for r in recipes], } return await anyio.to_thread.run_sync(_impl) @mcp.tool() async def hf_propose(week: str = "", count: int = 0) -> dict: """Shortlist de recettes SANS ingrédient exclu, classée par préférences. `week` vide = semaine courante. `count=0` renvoie toutes les recettes sûres. Étape « je propose » : rien n'est écrit ici — utiliser hf_confirm_selection() ensuite. """ def _impl() -> dict: w = week or api.current_week() allow_premium = hf_filter.load_allow_premium() with api.HelloFreshClient() as client: recipes = client.get_menu(w) safe = hf_filter.propose(recipes, count=count or None, allow_premium=allow_premium) excluded = [r.summary() for r in recipes if r.contains_excluded] premium = [r.summary() for r in recipes if r.is_premium and not r.contains_excluded] premium_note = ("Recettes premium (supplément) INCLUSES dans la proposition (réglage UI activé). " if allow_premium else "Les recettes premium (supplément) sont exclues de la proposition. ") return { "week": w, "allow_premium": allow_premium, "proposed": [r.summary() for r in safe], "excluded_for_coco_etc": excluded, "premium_extra_cost": premium, "note": "Aucune écriture effectuée. " + premium_note + "Confirme avec hf_confirm_selection(week, recipe_ids).", } return await anyio.to_thread.run_sync(_impl) @mcp.tool() async def hf_confirm_selection(week: str, recipe_ids: list[str], dry_run: bool = False, allow_premium: bool | None = None) -> dict: """ÉCRIT la sélection de recettes dans la box de la semaine (après confirmation). Garde-fous : refuse toute recette contenant un ingrédient exclu (coco !) ET toute recette payante hors abonnement (premium, supplément) sauf si `allow_premium=True`. `allow_premium=None` (défaut) reprend le réglage de l'UI web (prefs.json) ; passer True/False ici l'emporte ponctuellement sur ce réglage. `dry_run=True` : construit et renvoie la requête sans l'envoyer (vérification). """ def _impl() -> dict: allow = hf_filter.load_allow_premium() if allow_premium is None else allow_premium with api.HelloFreshClient() as client: menu = client.get_menu(week) hf_filter.annotate(menu) by_id = {r.id: r for r in menu} bad = [rid for rid in recipe_ids if rid in by_id and by_id[rid].contains_excluded] if bad: return { "ok": False, "error": "Sélection refusée : recette(s) avec ingrédient exclu (coco ?).", "offending_ids": bad, } premium = [rid for rid in recipe_ids if rid in by_id and by_id[rid].is_premium] if premium and not allow: return { "ok": False, "error": "Sélection refusée : recette(s) payante(s) hors abonnement (supplément). " "Repasse allow_premium=True pour accepter le surcoût.", "premium_recipes": [ {"id": rid, "name": by_id[rid].name, "surcharge_eur": round(by_id[rid].surcharge_cents / 100, 2)} for rid in premium ], } unknown = [rid for rid in recipe_ids if rid not in by_id] if unknown: return {"ok": False, "error": "Recette(s) inconnue(s) pour cette semaine.", "unknown_ids": unknown} result = client.set_selection(week, recipe_ids, dry_run=dry_run) return {"ok": True, "week": week, "selected": recipe_ids, "dry_run": dry_run, "api_response": result} return await anyio.to_thread.run_sync(_impl) @mcp.tool() async def hf_get_excludes() -> list[str]: """Liste actuelle des ingrédients exclus.""" return await anyio.to_thread.run_sync(hf_filter.load_excludes) @mcp.tool() async def hf_add_exclude(term: str) -> list[str]: """Ajoute un ingrédient à exclure. Renvoie la nouvelle liste.""" return await anyio.to_thread.run_sync(hf_filter.add_exclude, term) @mcp.tool() async def hf_remove_exclude(term: str) -> list[str]: """Retire un ingrédient de la liste d'exclusion. Renvoie la nouvelle liste.""" return await anyio.to_thread.run_sync(hf_filter.remove_exclude, term) if __name__ == "__main__": # Défaut homelab/Hermes : streamable-http. Pour Claude Code local : ANTICOCO_TRANSPORT=stdio. transport = os.environ.get("ANTICOCO_TRANSPORT", "streamable-http") mcp.run(transport=transport)