Projet : un serveur MCP sur un logiciel de suivi des clients fictif

id;societe;ville;contact;statut;dernier_contact;prochaine_action;responsable
C-001;Société A;Lyon;compta@societe-a.example;client;2026-09-12;2026-10-12;Claire
C-002;Société B;Nantes;factures@societe-b.example;client;2026-09-30;2026-10-07;Karim
C-003;Société C;Lille;admin@societe-c.example;client;2026-08-21;2026-11-02;Claire
C-004;Société D;Rennes;achats@societe-d.example;prospect;2026-10-01;2026-10-08;Élodie
C-005;Société E;Bordeaux;compta@societe-e.example;client;2026-07-30;2026-10-05;Karim
C-006;Société F;Nantes;contact@societe-f.example;prospect;2026-09-25;2026-10-15;Élodie

"""Un serveur MCP sur un fichier de suivi des clients : deux outils qui lisent, deux qui écrivent,
aucun qui supprime. Chaque écriture est consignée dans journal.csv. Données fabriquées."""
import csv
import logging
import os
from datetime import date, datetime
from pathlib import Path
from typing import TypedDict

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations

DOSSIER = Path(os.environ.get("CRM_DOSSIER", Path(__file__).parent))
CHAMPS = ["id", "societe", "ville", "contact", "statut", "dernier_contact", "prochaine_action", "responsable"]
STATUTS = {"prospect", "client", "perdu"}
logging.basicConfig(level=logging.INFO)          # les journaux techniques vont sur stderr
mcp = MCPServer("Suivi des clients")
LIRE = ToolAnnotations(read_only_hint=True)

class Fiche(TypedDict):
    id: str
    societe: str
    ville: str
    contact: str
    statut: str
    dernier_contact: str
    prochaine_action: str
    responsable: str

def _lire() -> list[dict]:
    with open(DOSSIER / "crm.csv", encoding="utf-8", newline="") as f:
        return list(csv.DictReader(f, delimiter=";"))

def _ecrire(lignes: list[dict]) -> None:
    provisoire = DOSSIER / "crm.csv.tmp"          # écrire à côté, puis remplacer : jamais de fichier à moitié écrit
    with open(provisoire, "w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=CHAMPS, delimiter=";")
        w.writeheader()
        w.writerows(lignes)
    provisoire.replace(DOSSIER / "crm.csv")

def _journal(outil: str, client: str, avant: str, apres: str) -> None:
    neuf = not (DOSSIER / "journal.csv").exists()
    with open(DOSSIER / "journal.csv", "a", encoding="utf-8", newline="") as f:
        w = csv.writer(f, delimiter=";")
        if neuf:
            w.writerow(["horodatage", "outil", "client", "avant", "apres"])
        w.writerow([datetime.now().isoformat(timespec="seconds"), outil, client, avant, apres])

def _fiche(lignes: list[dict], id_client: str) -> dict:
    for l in lignes:
        if l["id"] == id_client:
            return l
    raise ToolError(f"Aucun client {id_client}.")

@mcp.tool(annotations=LIRE)
def chercher_clients(texte: str) -> list[Fiche]:
    """Cherche les clients dont le nom ou la ville contient ce texte.

    Args:
        texte: Un morceau du nom de la société ou de la ville, par exemple Nantes.
    """
    t = texte.strip().lower()
    if len(t) < 2:
        raise ToolError("Donnez au moins deux caractères.")
    return [l for l in _lire() if t in l["societe"].lower() or t in l["ville"].lower()]

@mcp.tool(annotations=LIRE)
def a_relancer(au: str) -> list[Fiche]:
    """Les clients et prospects dont la prochaine action est due à cette date ou avant.

    Args:
        au: La date de référence, au format AAAA-MM-JJ.
    """
    try:
        jour = date.fromisoformat(au)
    except ValueError:
        raise ToolError("Date attendue au format AAAA-MM-JJ.")
    return [l for l in _lire() if l["prochaine_action"] and date.fromisoformat(l["prochaine_action"]) <= jour]

@mcp.tool(annotations=ToolAnnotations(read_only_hint=False, destructive_hint=False))
def noter_contact(id_client: str, date_contact: str, prochaine_action: str) -> Fiche:
    """Enregistre un contact avec un client : la date du contact et celle de la prochaine action.
    N'efface rien d'autre que ces deux dates ; l'ancienne valeur reste dans le journal.

    Args:
        id_client: L'identifiant du client, par exemple C-002.
        date_contact: La date du contact, AAAA-MM-JJ, aujourd'hui ou avant.
        prochaine_action: La date de la prochaine action, AAAA-MM-JJ, après le contact.
    """
    try:
        d, p = date.fromisoformat(date_contact), date.fromisoformat(prochaine_action)
    except ValueError:
        raise ToolError("Dates attendues au format AAAA-MM-JJ.")
    if p <= d:
        raise ToolError("La prochaine action doit venir après le contact.")
    lignes = _lire()
    f = _fiche(lignes, id_client)
    avant = f"{f['dernier_contact']} / {f['prochaine_action']}"
    f["dernier_contact"], f["prochaine_action"] = date_contact, prochaine_action
    _ecrire(lignes)
    _journal("noter_contact", id_client, avant, f"{date_contact} / {prochaine_action}")
    return f

@mcp.tool(annotations=ToolAnnotations(read_only_hint=False, destructive_hint=True, idempotent_hint=True))
def changer_statut(id_client: str, statut: str, motif: str) -> Fiche:
    """Change le statut d'un client : prospect, client ou perdu. Le motif est obligatoire.

    Args:
        id_client: L'identifiant du client, par exemple C-004.
        statut: Le nouveau statut : prospect, client ou perdu.
        motif: Pourquoi, en une phrase ; il est consigné dans le journal.
    """
    if statut not in STATUTS:
        raise ToolError("Statut attendu : prospect, client ou perdu.")
    if len(motif.strip()) < 10:
        raise ToolError("Donnez un motif d'au moins dix caractères.")
    lignes = _lire()
    f = _fiche(lignes, id_client)
    avant = f["statut"]
    f["statut"] = statut
    if statut == "perdu":
        f["prochaine_action"] = ""
    _ecrire(lignes)
    _journal("changer_statut", id_client, avant, f"{statut} : {motif.strip()}")
    return f

if __name__ == "__main__":
    mcp.run(transport="stdio")

"""Lance le serveur sur une copie du fichier, comme le ferait un assistant, et appelle ses outils.
Aucun modèle : chaque appel est celui qu'un modèle pourrait demander."""
import asyncio
import os
import shutil
import sys
import tempfile
from pathlib import Path

from mcp import Client, StdioServerParameters

async def main() -> None:
    copie = Path(tempfile.mkdtemp())
    shutil.copy("crm.csv", copie / "crm.csv")                  # le fichier d'origine ne bouge pas
    serveur = StdioServerParameters(command=sys.executable, args=["serveur.py"],
                                    env={**os.environ, "CRM_DOSSIER": str(copie)})
    async with Client(serveur) as client:
        outils = (await client.list_tools()).tools
        for o in outils:
            a = o.annotations
            nature = "lit" if a and a.read_only_hint else ("écrit, destructif" if a and a.destructive_hint else "écrit")
            print(f"Outil {o.name} : {nature}")
        nantes = await client.call_tool("chercher_clients", {"texte": "Nantes"})
        print("Clients à Nantes :", ", ".join(c["societe"] for c in nantes.structured_content["result"]))
        dus = await client.call_tool("a_relancer", {"au": "2026-10-09"})
        print("À relancer au 9 octobre :", ", ".join(c["id"] for c in dus.structured_content["result"]))
        r = await client.call_tool("noter_contact", {"id_client": "C-002", "date_contact": "2026-10-09", "prochaine_action": "2026-10-23"})
        print("Contact noté :", r.structured_content["id"], r.structured_content["prochaine_action"])
        for args in ({"id_client": "C-004", "statut": "client", "motif": "ok"},
                     {"id_client": "C-004", "statut": "fidèle", "motif": "Premier bon de commande reçu le 9 octobre."},
                     {"id_client": "C-099", "statut": "client", "motif": "Premier bon de commande reçu le 9 octobre."}):
            refus = await client.call_tool("changer_statut", args)
            print("Refusé :", refus.content[0].text if refus.is_error else "accepté")
        r = await client.call_tool("changer_statut", {"id_client": "C-004", "statut": "client", "motif": "Premier bon de commande reçu le 9 octobre."})
        print("Statut changé :", r.structured_content["id"], r.structured_content["statut"])
    print("Journal :")
    for ligne in (copie / "journal.csv").read_text(encoding="utf-8").splitlines()[1:]:
        print("  " + ligne.split(";", 1)[1])                      # sans l'horodatage, qui change à chaque essai
    shutil.rmtree(copie)

asyncio.run(main())