كتابة خادم MCP

نادرًا ما يعمل أول خادم MCP تكتبه من المحاولة الأولى، وغالبًا لا يكون السبب ما تتخيله. فالبروتوكول ليس المشكلة، إذ تتكفل به المكتبات. المشكلة تكون في print للتنقيح يصل إلى مخرج المعيار فيُفسد تدفق JSON-RPC، أو في أداة ترفع استثناءً بدل أن تُرجع خطأً مقروءًا، أو في خادم يبدأ بشكل سليم عند تشغيله يدويًا بينما يعجز المضيف عن تشغيله لأنه لا يجد مفسّر البيئة الافتراضية الصحيحة.

يبني هذا الدرس خادمًا كاملًا، مكوّنًا من أداتين ومورد واحد في مجال الفوترة، ثم يختبره دون وكيل ولا نموذج ولا مضيف. الترتيب هو الأهم: ما لم تشاهد ردود JSON-RPC بعينيك، فإن ربط وكيل بالخادم يعني تنقيح نظامين غير حتميين في آن واحد. خادم MCP برنامج عادي، ويُختبر كما يُختبر أي برنامج عادي، وهذا بالضبط ما يجعل العمل بالبروتوكول مريحًا.

الإعداد

python -m venv .venv
.venv/bin/pip install "mcp[cli]>=1.9"

توفّر الحزمة الرسمية مستويين من واجهة البرمجة. يولّد FastMCP مخططات JSON انطلاقًا من تعليقات الأنواع التوضيحية (type annotations) والـ docstrings، بينما يتيح لك الخادم منخفض المستوى كتابة معالجات tools/list وtools/call يدويًا. استخدم الأول ما لم تكن بحاجة صريحة إلى تحكم كامل في المخططات.

الخادم

"""Serveur MCP de facturation : consultation et annulation de commandes."""

import logging
import sys
from datetime import date
from typing import Literal

from mcp.server.fastmcp import FastMCP

# تُوجَّه السجلات إلى مجرى الأخطاء المعياري. في نقل stdio، ينقل مخرج المعيار
# رسائل JSON-RPC: يكفي استدعاء print واحد لتصبح الجلسة غير قابلة للإصلاح
# بالنسبة إلى المضيف، دون أي رسالة خطأ يمكن الاستفادة منها.
logging.basicConfig(level=logging.INFO, stream=sys.stderr)
journal = logging.getLogger("facturation")

mcp = FastMCP("facturation")

COMMANDES = {
    "CMD-4417": {"courriel": "a@exemple.fr", "statut": "expediee",
                 "total_cents": 8990, "date": "2026-08-14"},
    "CMD-4418": {"courriel": "a@exemple.fr", "statut": "confirmee",
                 "total_cents": 2400, "date": "2026-08-18"},
    "CMD-4501": {"courriel": "b@exemple.fr", "statut": "brouillon",
                 "total_cents": 15000, "date": "2026-08-19"},
}

POLITIQUE = """# Politique de remboursement

- Une commande en brouillon ou confirmee est annulable sans frais.
- Une commande expediee ne s'annule pas : elle suit la procédure de retour.
- Passé 30 jours après expédition, aucun remboursement n'est accordé.
"""

@mcp.tool(annotations={"readOnlyHint": True, "idempotentHint": True})
def rechercher_commandes(
    courriel_client: str,
    statut: Literal["brouillon", "confirmee", "expediee", "annulee"] | None = None,
    apres: str | None = None,
) -> dict:
    """Recherche les commandes d'un client dans le système de facturation.

    Retourne au plus 50 commandes, de la plus récente à la plus ancienne, avec
    leurs métadonnées seulement. Pour le détail des lignes, appelez
    lire_commande avec un identifiant obtenu ici.

    Args:
        courriel_client: adresse exacte, sans recherche approchante.
        statut: omettre pour tous les statuts.
        apres: date ISO 8601 incluse, par exemple 2026-01-15.
    """
    if "@" not in courriel_client:
        return {
            "erreur": "courriel_invalide",
            "message": f"{courriel_client} n'est pas une adresse de courriel.",
            "correction": "Fournissez une adresse complète, pas un nom de client.",
        }

    if apres is not None:
        try:
            date.fromisoformat(apres)
        except ValueError:
            return {
                "erreur": "date_invalide",
                "message": f"{apres} n'est pas une date ISO 8601.",
                "correction": "Utilisez le format AAAA-MM-JJ.",
            }

    trouvees = [
        {"id": identifiant, **donnees}
        for identifiant, donnees in COMMANDES.items()
        if donnees["courriel"] == courriel_client
        and (statut is None or donnees["statut"] == statut)
        and (apres is None or donnees["date"] >= apres)
    ]
    trouvees.sort(key=lambda c: c["date"], reverse=True)
    journal.info("recherche courriel=%s statut=%s -> %d", courriel_client, statut, len(trouvees))

    if not trouvees:
        return {
            "commandes": [],
            "total_correspondances": 0,
            "explication": (
                "Aucune commande pour ce client avec ces critères. Vérifiez "
                "l'adresse, ou relancez sans filtre de statut."
            ),
        }

    return {
        "commandes": trouvees[:50],
        "total_correspondances": len(trouvees),
        "retournees": min(50, len(trouvees)),
        "tronque": len(trouvees) > 50,
    }

@mcp.tool(annotations={"readOnlyHint": False, "destructiveHint": True, "idempotentHint": True})
def annuler_commande(identifiant: str, motif: str) -> dict:
    """Annule une commande non expédiée. Opération irréversible.

    N'annule jamais une commande expédiée : consultez la ressource
    facturation://politiques/remboursement pour la procédure de retour.

    Args:
        identifiant: identifiant de la forme CMD-0000.
        motif: raison de l'annulation, conservée dans l'historique.
    """
    commande = COMMANDES.get(identifiant)
    if commande is None:
        return {
            "erreur": "commande_introuvable",
            "message": f"{identifiant} n'existe pas.",
            "correction": "Obtenez un identifiant valide via rechercher_commandes.",
        }

    if commande["statut"] == "annulee":
        # خاصية عدم تغيّر النتيجة بالتكرار (idempotence): إعادة استدعاء الأداة على
        # طلب ملغى سلفًا تُعدّ نجاحًا لا خطأً. من دون ذلك، يحوّل الاستئناف بعد
        # انقطاع إجراءً ناجحًا إلى فشل ظاهري.
        return {"identifiant": identifiant, "statut": "annulee", "deja_annulee": True}

    if commande["statut"] == "expediee":
        return {
            "erreur": "annulation_impossible",
            "message": "Cette commande est déjà expédiée.",
            "correction": "Lisez facturation://politiques/remboursement et proposez un retour.",
        }

    commande["statut"] = "annulee"
    commande["motif_annulation"] = motif
    journal.info("annulation %s motif=%s", identifiant, motif)
    return {"identifiant": identifiant, "statut": "annulee", "deja_annulee": False}

@mcp.resource(
    "facturation://politiques/remboursement",
    name="Politique de remboursement",
    mime_type="text/markdown",
)
def politique_remboursement() -> str:
    """Règles applicables aux annulations et remboursements."""
    return POLITIQUE

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