Files

T

gilles bbf264fb61 docs: plan d'implémentation MCP server (16 outils + tutoriel Hermes)

2026-05-25 22:40:41 +02:00

41 KiB

Raw Blame History

MCP Server HomeHub — Plan d'implémentation

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Intégrer un serveur MCP (Model Context Protocol) dans le backend FastAPI de HomeHub, exposant 16 outils aux agents IA (Hermes, Claude Code) via Streamable HTTP authentifié par Bearer token.

Architecture: Le serveur MCP est monté sous /mcp dans l'application FastAPI existante via FastMCP.streamable_http_app(), partageant le pool de connexions PostgreSQL via AsyncSessionLocal. Un middleware ASGI pur intercepte les requêtes /mcp* et vérifie le Bearer token défini en variable d'environnement.

Tech Stack: FastAPI, mcp>=1.9 (SDK officiel Anthropic — FastMCP), SQLAlchemy 2.0 async (AsyncSessionLocal), nginx proxy sans buffer, pytest-asyncio pour les tests.

Structure des fichiers

Fichier	Action	Rôle
`backend/requirements.txt`	Modifier	Ajouter `mcp>=1.9`
`backend/app/core/config.py`	Modifier	Ajouter le champ `mcp_api_key: str`
`backend/app/core/mcp_auth.py`	Créer	Middleware ASGI Bearer token pour `/mcp*`
`backend/app/api/mcp_server.py`	Créer	16 outils MCP + instance `FastMCP`
`backend/app/main.py`	Modifier	Mount MCP ASGI + middleware MCPAuth
`backend/tests/test_mcp.py`	Créer	Tests auth + outils todos/notes/shopping
`docker-compose.yml`	Modifier	Ajouter `MCP_API_KEY` au service backend
`frontend/nginx.conf`	Modifier	Location `/mcp` sans buffer, timeout 86400s

Task 1: Dépendances et configuration

Files:

Modify: backend/requirements.txt
Modify: backend/app/core/config.py
Step 1: Ajouter mcp>=1.9 à requirements.txt

Fichier complet backend/requirements.txt :

fastapi==0.115.5
uvicorn[standard]==0.32.1
sqlalchemy[asyncio]==2.0.36
asyncpg==0.30.0
alembic==1.14.0
pydantic-settings==2.6.1
pillow==11.1.0
python-multipart==0.0.20
httpx==0.28.0
arq==0.26.1
mcp>=1.9
pytest==8.3.4
pytest-asyncio==0.24.0

Step 2: Ajouter mcp_api_key dans Settings

Fichier complet backend/app/core/config.py :

from pydantic_settings import BaseSettings, SettingsConfigDict
from pathlib import Path


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

    database_url: str = "postgresql+asyncpg://homehub:homehub@localhost:5432/homehub"
    upload_dir: str = "/data/uploads"
    data_dir: str = "/data"
    redis_url: str = "redis://redis:6379"
    cors_origins: str = "http://localhost:3000"
    mcp_api_key: str = ""

    @property
    def cors_origins_list(self) -> list[str]:
        return [o.strip() for o in self.cors_origins.split(",")]

    @property
    def upload_path(self) -> Path:
        return Path(self.upload_dir)

    @property
    def notes_md_path(self) -> Path:
        return Path(self.data_dir) / "notes"

    @property
    def backup_path(self) -> Path:
        return Path(self.data_dir) / "backup"


settings = Settings()

Step 3: Commit

git add backend/requirements.txt backend/app/core/config.py
git commit -m "feat(mcp): dépendance mcp>=1.9 + champ mcp_api_key dans Settings"

Task 2: Middleware d'authentification Bearer token

Files:

Create: backend/app/core/mcp_auth.py

Le middleware est un ASGI pur (pas BaseHTTPMiddleware) pour éviter tout problème de streaming avec les réponses SSE du transport MCP.

Step 1: Créer backend/app/core/mcp_auth.py

import json
from starlette.types import ASGIApp, Receive, Scope, Send
from app.core.config import settings


class MCPAuthMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] == "http" and scope.get("path", "").startswith("/mcp"):
            headers = dict(scope.get("headers", []))
            auth = headers.get(b"authorization", b"").decode()
            if auth != f"Bearer {settings.mcp_api_key}":
                body = json.dumps({"detail": "Unauthorized"}).encode()
                await send({
                    "type": "http.response.start",
                    "status": 401,
                    "headers": [
                        (b"content-type", b"application/json"),
                        (b"content-length", str(len(body)).encode()),
                    ],
                })
                await send({"type": "http.response.body", "body": body})
                return
        await self.app(scope, receive, send)

Step 2: Commit

git add backend/app/core/mcp_auth.py
git commit -m "feat(mcp): middleware ASGI Bearer token pour /mcp*"

Task 3: Outils MCP — Todos (5 outils)

Files:

Create: backend/app/api/mcp_server.py
Create: backend/tests/test_mcp.py (début)
Step 1: Écrire les tests failing pour les outils todos

Créer backend/tests/test_mcp.py :

import json
import pytest
from app.api.mcp_server import (
    get_todos, create_todo, update_todo, postpone_todo, delete_todo,
)


async def test_get_todos_retourne_liste_json():
    result = await get_todos(status="pending")
    data = json.loads(result)
    assert isinstance(data, list)


async def test_create_todo_outil_cree_une_tache():
    result = await create_todo(title="TEST_MCP_todo_create")
    data = json.loads(result)
    assert data["title"] == "TEST_MCP_todo_create"
    assert data["status"] == "pending"
    assert data["priority"] == "medium"
    # Cleanup
    await delete_todo(id=str(data["id"]))


async def test_update_todo_outil():
    created = json.loads(await create_todo(title="TEST_MCP_todo_update"))
    result = await update_todo(id=str(created["id"]), status="done")
    data = json.loads(result)
    assert data["status"] == "done"
    await delete_todo(id=str(created["id"]))


async def test_postpone_todo_outil():
    created = json.loads(await create_todo(title="TEST_MCP_todo_postpone"))
    result = await postpone_todo(id=str(created["id"]), days=3)
    data = json.loads(result)
    assert data["postponed_count"] == 1
    await delete_todo(id=str(created["id"]))


async def test_delete_todo_outil():
    created = json.loads(await create_todo(title="TEST_MCP_todo_delete"))
    result = await delete_todo(id=str(created["id"]))
    data = json.loads(result)
    assert "deleted" in data


async def test_update_todo_id_invalide():
    result = await update_todo(id="pas-un-uuid", title="x")
    data = json.loads(result)
    assert "error" in data


async def test_delete_todo_introuvable():
    result = await delete_todo(id="00000000-0000-0000-0000-000000000000")
    data = json.loads(result)
    assert "error" in data

Step 2: Vérifier que les tests échouent (ImportError)

cd backend && pytest tests/test_mcp.py -v 2>&1 | head -20

Résultat attendu : ModuleNotFoundError ou ImportError — mcp_server n'existe pas encore.

Step 3: Créer backend/app/api/mcp_server.py avec les 5 outils todos

import json
import uuid
from datetime import datetime, timedelta, timezone, date as date_type
from decimal import Decimal

from mcp.server.fastmcp import FastMCP
from sqlalchemy import select, text, and_, 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

mcp = FastMCP("HomeHub", stateless_http=True)


def _serialize(obj):
    if isinstance(obj, uuid.UUID):
        return str(obj)
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, date_type):
        return obj.isoformat()
    if isinstance(obj, Decimal):
        return float(obj)
    raise TypeError(f"Type non sérialisable : {type(obj)}")


def _dumps(data) -> str:
    return json.dumps(data, default=_serialize)


def _iso_week_label() -> str:
    now = datetime.now(tz=timezone.utc)
    iso = now.isocalendar()
    return f"S{iso[1]} {iso[0]}"


# ── TODOS ──────────────────────────────────────────────────────────────────────

@mcp.tool()
async def get_todos(
    status: str = "pending",
    domain: str | None = None,
    priority: str | None = None,
) -> str:
    """Liste filtrée des tâches. status: pending/done/cancelled. priority: low/medium/high."""
    async with AsyncSessionLocal() as session:
        conditions = [TodoItem.status == status]
        if domain:
            conditions.append(TodoItem.domains.contains([domain]))
        if priority:
            conditions.append(TodoItem.priority == priority)
        stmt = (
            select(TodoItem)
            .where(and_(*conditions))
            .order_by(TodoItem.due_date.asc().nulls_last(), TodoItem.created_at.desc())
            .limit(100)
        )
        result = await session.execute(stmt)
        items = result.scalars().all()
        return _dumps([{
            "id": item.id,
            "title": item.title,
            "status": item.status,
            "priority": item.priority,
            "domains": item.domains,
            "tags": item.tags,
            "due_date": item.due_date,
            "postponed_count": item.postponed_count,
        } for item in items])


@mcp.tool()
async def create_todo(
    title: str,
    due_date: str | None = None,
    priority: str | None = None,
    domain: str | None = None,
) -> str:
    """Crée une tâche. due_date en ISO 8601 (ex: 2025-06-01T10:00:00Z). priority: low/medium/high."""
    async with AsyncSessionLocal() as session:
        parsed_due = None
        if due_date:
            try:
                parsed_due = datetime.fromisoformat(due_date.replace("Z", "+00:00"))
            except ValueError:
                return _dumps({"error": f"Format de date invalide : {due_date}"})
        item = TodoItem(
            title=title,
            due_date=parsed_due,
            priority=priority or "medium",
            domains=[domain] if domain else [],
        )
        session.add(item)
        await session.commit()
        await session.refresh(item)
        return _dumps({
            "id": item.id,
            "title": item.title,
            "status": item.status,
            "priority": item.priority,
            "domains": item.domains,
            "due_date": item.due_date,
        })


@mcp.tool()
async def update_todo(
    id: str,
    title: str | None = None,
    status: str | None = None,
    priority: str | None = None,
) -> str:
    """Modifie une tâche. status: pending/done/cancelled. priority: low/medium/high."""
    async with AsyncSessionLocal() as session:
        try:
            todo_id = uuid.UUID(id)
        except ValueError:
            return _dumps({"error": f"UUID invalide : {id}"})
        item = await session.get(TodoItem, todo_id)
        if not item:
            return _dumps({"error": f"Tâche introuvable : {id}"})
        if title is not None:
            item.title = title
        if status is not None:
            item.status = status
        if priority is not None:
            item.priority = priority
        item.updated_at = datetime.now(timezone.utc)
        await session.commit()
        await session.refresh(item)
        return _dumps({
            "id": item.id,
            "title": item.title,
            "status": item.status,
            "priority": item.priority,
        })


@mcp.tool()
async def postpone_todo(id: str, days: int) -> str:
    """Reporte une tâche de N jours à partir de sa date d'échéance (ou maintenant si nulle)."""
    async with AsyncSessionLocal() as session:
        try:
            todo_id = uuid.UUID(id)
        except ValueError:
            return _dumps({"error": f"UUID invalide : {id}"})
        item = await session.get(TodoItem, todo_id)
        if not item:
            return _dumps({"error": f"Tâche introuvable : {id}"})
        now = datetime.now(timezone.utc)
        base = item.due_date if item.due_date else now
        item.due_date = base + timedelta(days=days)
        item.postponed_count += 1
        item.updated_at = now
        await session.commit()
        await session.refresh(item)
        return _dumps({
            "id": item.id,
            "title": item.title,
            "due_date": item.due_date,
            "postponed_count": item.postponed_count,
        })


@mcp.tool()
async def delete_todo(id: str) -> str:
    """Supprime une tâche définitivement."""
    async with AsyncSessionLocal() as session:
        try:
            todo_id = uuid.UUID(id)
        except ValueError:
            return _dumps({"error": f"UUID invalide : {id}"})
        item = await session.get(TodoItem, todo_id)
        if not item:
            return _dumps({"error": f"Tâche introuvable : {id}"})
        await session.delete(item)
        await session.commit()
        return _dumps({"deleted": id})

Step 4: Lancer les tests todos

cd backend && pytest tests/test_mcp.py -v -k "todo"

Résultat attendu : 7 tests PASSED.

Step 5: Commit

git add backend/app/api/mcp_server.py backend/tests/test_mcp.py
git commit -m "feat(mcp): 5 outils todos + tests"

Task 4: Outils MCP — Notes (5 outils)

Files:

Modify: backend/app/api/mcp_server.py (ajout en bas du fichier)
Modify: backend/tests/test_mcp.py (ajout de tests)
Step 1: Ajouter les tests failing pour les outils notes

Ajouter à la fin de backend/tests/test_mcp.py :

from app.api.mcp_server import (
    search_notes, get_note, create_note, update_note, delete_note,
)


async def test_search_notes_retourne_liste():
    result = await search_notes(query="inexistant_xyz_abc_999")
    data = json.loads(result)
    assert isinstance(data, list)
    assert data == []


async def test_create_note_outil():
    result = await create_note(title="TEST_MCP_note_create", content="Contenu de test MCP")
    data = json.loads(result)
    assert data["title"] == "TEST_MCP_note_create"
    assert "id" in data
    await delete_note(id=str(data["id"]))


async def test_get_note_outil():
    created = json.loads(await create_note(title="TEST_MCP_note_get", content="Contenu get"))
    result = await get_note(id=str(created["id"]))
    data = json.loads(result)
    assert data["title"] == "TEST_MCP_note_get"
    assert data["content"] == "Contenu get"
    assert "attachments" in data
    await delete_note(id=str(created["id"]))


async def test_update_note_outil():
    created = json.loads(await create_note(title="TEST_MCP_note_update", content="avant"))
    result = await update_note(id=str(created["id"]), content="après")
    data = json.loads(result)
    assert "id" in data
    await delete_note(id=str(created["id"]))


async def test_delete_note_outil():
    created = json.loads(await create_note(title="TEST_MCP_note_delete", content="x"))
    result = await delete_note(id=str(created["id"]))
    data = json.loads(result)
    assert "deleted" in data


async def test_get_note_introuvable():
    result = await get_note(id="00000000-0000-0000-0000-000000000000")
    data = json.loads(result)
    assert "error" in data

Step 2: Vérifier que les tests notes échouent (ImportError)

cd backend && pytest tests/test_mcp.py -v -k "note" 2>&1 | head -10

Résultat attendu : ImportError — fonctions notes non définies dans mcp_server.py.

Step 3: Ajouter les 5 outils notes dans mcp_server.py

Ajouter après le bloc # ── TODOS ── dans backend/app/api/mcp_server.py :

# ── NOTES ──────────────────────────────────────────────────────────────────────

@mcp.tool()
async def search_notes(
    query: str | None = None,
    category: str | None = None,
    tag: str | None = None,
) -> str:
    """Recherche de notes via FTS PostgreSQL (français). query cherche dans titre+contenu."""
    async with AsyncSessionLocal() as session:
        conditions = []
        if category:
            conditions.append(NoteItem.category == category)
        if tag:
            conditions.append(NoteItem.tags.contains([tag]))
        if query:
            conditions.append(
                text(
                    "to_tsvector('french', coalesce(notes.items.title,'') || ' ' || notes.items.content)"
                    " @@ plainto_tsquery('french', :q)"
                ).bindparams(q=query)
            )
        stmt = (
            select(NoteItem)
            .options(selectinload(NoteItem.attachments))
            .order_by(NoteItem.created_at.desc())
            .limit(20)
        )
        if conditions:
            stmt = stmt.where(and_(*conditions))
        result = await session.execute(stmt)
        notes = result.scalars().all()
        return _dumps([{
            "id": n.id,
            "title": n.title,
            "category": n.category,
            "tags": n.tags,
            "content_preview": n.content[:200] if n.content else "",
            "created_at": n.created_at,
            "attachment_count": len(n.attachments),
        } for n in notes])


@mcp.tool()
async def get_note(id: str) -> str:
    """Retourne une note complète avec ses pièces jointes."""
    async with AsyncSessionLocal() as session:
        try:
            note_id = uuid.UUID(id)
        except ValueError:
            return _dumps({"error": f"UUID invalide : {id}"})
        stmt = (
            select(NoteItem)
            .where(NoteItem.id == note_id)
            .options(selectinload(NoteItem.attachments))
        )
        result = await session.execute(stmt)
        note = result.scalar_one_or_none()
        if not note:
            return _dumps({"error": f"Note introuvable : {id}"})
        return _dumps({
            "id": note.id,
            "title": note.title,
            "content": note.content,
            "category": note.category,
            "tags": note.tags,
            "created_at": note.created_at,
            "attachments": [{
                "id": a.id,
                "file_type": a.file_type,
                "original_name": a.original_name,
            } for a in note.attachments],
        })


@mcp.tool()
async def create_note(
    title: str,
    content: str,
    category: str | None = None,
    tags: list[str] | None = None,
) -> str:
    """Crée une note avec titre, contenu markdown, catégorie optionnelle et tags."""
    async with AsyncSessionLocal() as session:
        note = NoteItem(
            title=title,
            content=content,
            category=category,
            tags=tags or [],
        )
        session.add(note)
        await session.commit()
        await session.refresh(note)
        return _dumps({
            "id": note.id,
            "title": note.title,
            "category": note.category,
            "tags": note.tags,
            "created_at": note.created_at,
        })


@mcp.tool()
async def update_note(
    id: str,
    title: str | None = None,
    content: str | None = None,
    tags: list[str] | None = None,
) -> str:
    """Modifie le titre, le contenu ou les tags d'une note."""
    async with AsyncSessionLocal() as session:
        try:
            note_id = uuid.UUID(id)
        except ValueError:
            return _dumps({"error": f"UUID invalide : {id}"})
        note = await session.get(NoteItem, note_id)
        if not note:
            return _dumps({"error": f"Note introuvable : {id}"})
        if title is not None:
            note.title = title
        if content is not None:
            note.content = content
        if tags is not None:
            note.tags = tags
        await session.commit()
        await session.refresh(note)
        return _dumps({"id": note.id, "title": note.title, "tags": note.tags})


@mcp.tool()
async def delete_note(id: str) -> str:
    """Supprime une note et toutes ses pièces jointes (cascade)."""
    async with AsyncSessionLocal() as session:
        try:
            note_id = uuid.UUID(id)
        except ValueError:
            return _dumps({"error": f"UUID invalide : {id}"})
        note = await session.get(NoteItem, note_id)
        if not note:
            return _dumps({"error": f"Note introuvable : {id}"})
        await session.delete(note)
        await session.commit()
        return _dumps({"deleted": id})

Step 4: Lancer les tests notes

cd backend && pytest tests/test_mcp.py -v -k "note"

Résultat attendu : 6 tests PASSED.

Step 5: Commit

git add backend/app/api/mcp_server.py backend/tests/test_mcp.py
git commit -m "feat(mcp): 5 outils notes + tests"

Task 5: Outils MCP — Shopping (6 outils)

Files:

Modify: backend/app/api/mcp_server.py (ajout en bas du fichier)
Modify: backend/tests/test_mcp.py (ajout de tests)
Step 1: Ajouter les tests failing pour les outils shopping

Ajouter à la fin de backend/tests/test_mcp.py :

from app.api.mcp_server import (
    get_shopping_lists, get_active_shopping_list, search_products,
    create_shopping_list, add_shopping_item, check_shopping_item,
)


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)
    # Le nom auto commence par "S" (format "S{semaine} {année}")
    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_sans_liste():
    # Si aucune liste draft, retourne une erreur structurée (pas une exception)
    result = await get_active_shopping_list()
    data = json.loads(result)
    # Soit une liste avec des infos, soit une erreur — pas une exception Python
    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

Step 2: Vérifier que les tests shopping échouent (ImportError)

cd backend && pytest tests/test_mcp.py -v -k "shopping or liste or article or product" 2>&1 | head -10

Résultat attendu : ImportError — fonctions shopping non définies dans mcp_server.py.

Step 3: Ajouter les 6 outils shopping dans mcp_server.py

Ajouter après le bloc # ── NOTES ── dans backend/app/api/mcp_server.py :

# ── 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 par son ID."""
    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})

Step 4: Lancer les tests shopping

cd backend && pytest tests/test_mcp.py -v -k "shopping or liste or article or product"

Résultat attendu : 8 tests PASSED.

Step 5: Commit

git add backend/app/api/mcp_server.py backend/tests/test_mcp.py
git commit -m "feat(mcp): 6 outils shopping + tests"

Task 6: Câblage complet (main.py + nginx + docker-compose)

Files:

Modify: backend/app/main.py
Modify: frontend/nginx.conf
Modify: docker-compose.yml
Modify: backend/tests/test_mcp.py (ajout tests auth)
Step 1: Modifier backend/app/main.py

Fichier complet backend/app/main.py :

from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from app.api.admin import router as admin_router
from app.api.events import router as events_router
from app.api.health import router as health_router
from app.api.media import router as media_router
from app.api.mcp_server import mcp
from app.api.notes import router as notes_router
from app.api.todos import router as todos_router
from app.api.shopping import router as shopping_router
from app.core.config import settings
from app.core.mcp_auth import MCPAuthMiddleware
from app.core.redis import init_redis, close_redis
from app.data.seed import run_seed


@asynccontextmanager
async def lifespan(app: FastAPI):
    for subdir in ("uploads", "notes", "backup"):
        Path(settings.data_dir, subdir).mkdir(parents=True, exist_ok=True)
    await run_seed()
    await init_redis()
    yield
    await close_redis()


app = FastAPI(title="HomeHub API", version="0.1.0", lifespan=lifespan)

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins_list,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
app.add_middleware(MCPAuthMiddleware)

app.include_router(health_router, prefix="/api")
app.include_router(events_router, prefix="/api/events")
app.include_router(admin_router, prefix="/api/admin")
app.include_router(media_router, prefix="/api/media")
app.include_router(notes_router, prefix="/api/notes")
app.include_router(todos_router, prefix="/api/todos")
app.include_router(shopping_router, prefix="/api/shopping")

app.mount("/mcp", mcp.streamable_http_app())
app.mount("/media", StaticFiles(directory=str(settings.upload_path)), name="media")

Note ordre middleware : add_middleware en FastAPI — le dernier ajouté est le plus externe (s'exécute en premier). MCPAuthMiddleware ajouté en dernier intercepte donc les requêtes /mcp avant tout.

Step 2: Modifier frontend/nginx.conf

Fichier complet frontend/nginx.conf :

server {
    listen 80;
    server_name _;
    root /usr/share/nginx/html;
    index index.html;

    client_max_body_size 200m;

    gzip on;
    gzip_types text/plain text/css application/json application/javascript text/xml application/xml image/svg+xml;

    location ~* \.(js|css|png|jpg|jpeg|gif|ico|webp|woff2|svg)$ {
        expires 1y;
        add_header Cache-Control "public, immutable";
    }

    location /api/events/stream {
        proxy_pass http://backend:8000/api/events/stream;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 86400s;
    }

    location /mcp {
        proxy_pass http://backend:8000/mcp;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 86400s;
    }

    location /api/ {
        proxy_pass http://backend:8000/api/;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }

    location ^~ /media/ {
        proxy_pass http://backend:8000/media/;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }

    location / {
        try_files $uri $uri/ /index.html;
    }
}

Step 3: Modifier docker-compose.yml — ajouter MCP_API_KEY

Dans la section backend.environment, ajouter la ligne MCP_API_KEY :

  backend:
    build: ./backend
    user: "1000:1000"
    environment:
      DATABASE_URL: postgresql+asyncpg://homehub:homehub@db:5432/homehub
      UPLOAD_DIR: /data/uploads
      DATA_DIR: /data
      REDIS_URL: redis://redis:6379
      CORS_ORIGINS: http://localhost:3001,http://localhost:3000
      MCP_API_KEY: "4zfCmiC3Z_28F3xPOxBSDi0DQx6aRzAQrpplyywo_VI"

Faire de même pour backend-worker.environment si ce service accède jamais au MCP (facultatif, par cohérence).

Step 4: Ajouter les tests d'authentification dans test_mcp.py

Ajouter en haut de backend/tests/test_mcp.py (après les imports existants) :

import pytest
from app.core.config import settings

Ajouter à la fin de backend/tests/test_mcp.py :

async def test_mcp_auth_rejet_sans_token(client):
    """Le middleware renvoie 401 si aucun header Authorization."""
    resp = await client.get("/mcp")
    assert resp.status_code == 401


async def test_mcp_auth_rejet_mauvais_token(client):
    """Le middleware renvoie 401 si le token est incorrect."""
    resp = await client.get("/mcp", headers={"Authorization": "Bearer mauvais-token"})
    assert resp.status_code == 401


async def test_mcp_auth_accepte_bon_token(client, monkeypatch):
    """Le middleware laisse passer avec le token correct."""
    monkeypatch.setattr(settings, "mcp_api_key", "test-mcp-key-xyz")
    resp = await client.get(
        "/mcp",
        headers={"Authorization": "Bearer test-mcp-key-xyz"},
    )
    assert resp.status_code != 401

Step 5: Lancer la suite de tests complète

cd backend && pytest tests/test_mcp.py -v

Résultat attendu : tous les tests PASSED (auth + todos + notes + shopping).

Step 6: Commit

git add backend/app/main.py frontend/nginx.conf docker-compose.yml backend/tests/test_mcp.py
git commit -m "feat(mcp): câblage FastAPI + nginx proxy + docker-compose MCP_API_KEY"

Task 7: Validation et suite de tests finale

Files:

No new files — validation de l'existant
Step 1: Lancer la suite de tests complète du backend

cd backend && pytest -v

Résultat attendu : tous les tests PASSED (todos, shopping, health, media, mcp).

Step 2: Vérifier que l'import MCP fonctionne

cd backend && python -c "from app.api.mcp_server import mcp; print('OK — outils:', len(mcp._tool_manager._tools))"

Résultat attendu : OK — outils: 16

Step 3: Vérifier la description des outils exposés

cd backend && python -c "
from app.api.mcp_server import mcp
tools = mcp._tool_manager._tools
for name in sorted(tools.keys()):
    print(f'  {name}')
"

Résultat attendu (16 lignes) :

  add_shopping_item
  check_shopping_item
  create_note
  create_shopping_list
  create_todo
  delete_note
  delete_todo
  get_active_shopping_list
  get_note
  get_shopping_lists
  get_todos
  postpone_todo
  search_notes
  search_products
  update_note
  update_todo

Step 4: Commit de version

cd frontend && npm version patch
git add frontend/package.json
git commit -m "chore: bump version — MCP server v0.5.10"

Post-déploiement : Intégration avec Hermes

Cette section est à exécuter sur la machine Hermes (10.0.0.80) après que le backend HomeHub soit redéployé avec les nouvelles images Docker.

Prérequis

HomeHub doit être accessible depuis Hermes à l'adresse http://10.0.0.50:3001. Vérifier :

curl -s http://10.0.0.50:3001/api/health

Réponse attendue : {"status": "ok"} (ou similaire).

Configuration Hermes

Dans le fichier de configuration YAML de Hermes (chemin typique : ~/.config/hermes/config.yml ou ~/hermes/config.yml), ajouter le bloc suivant dans la section mcp_servers :

mcp_servers:
  homehub:
    url: http://10.0.0.50:3001/mcp
    headers:
      Authorization: "Bearer 4zfCmiC3Z_28F3xPOxBSDi0DQx6aRzAQrpplyywo_VI"

Champs :

url — point d'entrée Streamable HTTP du serveur MCP (proxifié par nginx)
headers.Authorization — Bearer token statique défini dans MCP_API_KEY

Filtrage optionnel des outils (si Hermes doit n'exposer qu'un sous-ensemble) :

mcp_servers:
  homehub:
    url: http://10.0.0.50:3001/mcp
    headers:
      Authorization: "Bearer 4zfCmiC3Z_28F3xPOxBSDi0DQx6aRzAQrpplyywo_VI"
    tools:
      include:
        - get_todos
        - create_todo
        - update_todo
        - postpone_todo
        - get_active_shopping_list
        - add_shopping_item
        - search_notes

Rechargement des outils

Après avoir sauvegardé la configuration, dans une session Hermes active, taper :

/reload-mcp

Hermes affichera la liste des outils chargés depuis HomeHub. Les 16 outils doivent apparaître avec leurs descriptions.

Test rapide depuis Hermes

Demander à Hermes :

"Quelles sont mes tâches en attente ?"

Hermes devrait appeler get_todos(status="pending") et afficher la liste des todos.

Utilisation avec Claude Code

Pour utiliser le serveur MCP depuis Claude Code (dans .mcp.json ou claude_desktop_config.json) :

{
  "mcpServers": {
    "homehub": {
      "url": "http://10.0.0.50:3001/mcp",
      "headers": {
        "Authorization": "Bearer 4zfCmiC3Z_28F3xPOxBSDi0DQx6aRzAQrpplyywo_VI"
      }
    }
  }
}

Auto-review du plan

Couverture spec :

✅ 5 outils Todos (get, create, update, postpone, delete)
✅ 5 outils Notes (search, get, create, update, delete)
✅ 6 outils Shopping (get_lists, get_active, search_products, create_list, add_item, check_item)
✅ Middleware auth Bearer token
✅ Mount /mcp FastAPI
✅ nginx proxy sans buffer + timeout 86400s
✅ docker-compose.yml — MCP_API_KEY
✅ Retour JSON structuré avec {"error": "..."} sans exception MCP
✅ Clé API 4zfCmiC3Z_28F3xPOxBSDi0DQx6aRzAQrpplyywo_VI dans docker-compose
✅ Tutoriel Hermes (YAML, /reload-mcp, filtrage outils)
✅ Config Claude Code (mcpServers JSON)

Cohérence des types :

_serialize gère UUID, datetime, date, Decimal — utilisé dans _dumps partout
AsyncSessionLocal importé depuis app.core.database — même pool que les routes REST
Les noms d'outils dans les tests (ex: get_todos) correspondent exactement aux fonctions décorées @mcp.tool()

Aucun placeholder détecté.

41 KiB Raw Blame History

MCP Server HomeHub — Plan d'implémentation

Structure des fichiers

Task 1: Dépendances et configuration

Task 2: Middleware d'authentification Bearer token

Task 3: Outils MCP — Todos (5 outils)

Task 4: Outils MCP — Notes (5 outils)

Task 5: Outils MCP — Shopping (6 outils)

Task 6: Câblage complet (main.py + nginx + docker-compose)

Task 7: Validation et suite de tests finale

Post-déploiement : Intégration avec Hermes

Prérequis

Configuration Hermes

Rechargement des outils

Test rapide depuis Hermes

Utilisation avec Claude Code

Auto-review du plan

41 KiB

Raw Blame History