أنماط MCP المتقدمة

التحديثات والإشعارات الفورية

4 دقيقة للقراءة

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

وهذا يترك طريقتين اثنتين فقط يصل بهما الإشعار إلى العميل، واختيار الصحيحة منهما هو مضمون هذا الدرس كله.

أنواع الإشعارات

الطريقتان اللتان يصل بهما الإشعار إلى العميل

يرسلقد يُصدرثم يجيبيرسليبثّالعميلكل تفاعل يبدأ هنا. دائماًطلب قيد التنفيذtools/call ما زال يعملsubscriptions/listenطلب استجابته تدفق طويل الأمدإشعارات مرتبطة بالطلبnotifications/progress و notifi…إشعارات الاشتراكتغيّر القوائم وتحديث الموارد، و…الاستجابةنتيجة واحدة أو خطأ واحد يُغلق ا…

مسار الاشتراك هو ما يقصده الناس ويخطئون فيه. فـ subscriptions/listen طلب عادي، غاية ما في الأمر أن استجابته تبقى مفتوحة. وحالته ملك ذلك الطلب لا الاتصال الذي تحته، فإن انقطعت القناة أعاد العميل إصدار الطلب بدل أن يتوقع من خادمك أن يتذكر شيئاً.

النوعفيمَ يُستخدمما الذي يعطب بدونه
التقدّمأداة تستغرق أطول مما يحتمل المستخدم انتظارهلا يستطيع المضيف التمييز بين «يعمل» و«معلّق»
تحديث موردمحتوى ربما خزّنه المضيف مؤقتاً وقد تغيّريجيب النموذج من نسخة قديمة وبثقة تامة
تنبيهات النظامحالة متدهورة ينبغي للمضيف إظهارهاتبقى الأعطال خفية حتى يفشل نداء أداة صراحةً

الخاصية المهمة: الإشعار لا يحمل رداً. لا شيء ينتظره ولا شيء يعيد إرساله إن ضاع. وهذا يجعل إرساله المتكرر آمناً، ويجعل الاعتماد عليه في صحة النتائج غير آمن — فلا تجعل الإشعار أبداً الطريق الوحيد ليعرف العميل شيئاً يحتاجه.

إرسال الإشعارات

from mcp.server.mcpserver import Context


@mcp.tool()
async def reindex(collection: str, ctx: Context) -> str:
    """إعادة بناء فهرس البحث لمجموعة."""
    chunks = await plan_chunks(collection)

    for i, chunk in enumerate(chunks, start=1):
        await process(chunk)
        await ctx.report_progress(
            progress=i,
            total=len(chunks),
            message=f"فُهرست {i} من {len(chunks)} قطعة",
        )

    return f"أُعيدت فهرسة {collection}: {len(chunks)} قطعة"

لا يوجد مُزيِّن للإشعارات ولا استدعاء عام باسم notify(method, params). أنت ترسل الإشعارات عبر كائن Context الذي يمرره الـ SDK إلى معالِجك، والدوال المتاحة هي تلك التي يعرّفها البروتوكول — وهذا هو المقصود. لا يستطيع خادم أن يخترع طريقة إشعار ثم يتوقع من عميل أن يوجّهها.

إشعارات تغيير الموارد

إعلام العملاء عند تغيير الموارد:

@mcp.tool()
async def update_document(doc_id: str, content: str, ctx: Context) -> str:
    """استبدال محتوى مستند."""
    await db.update(doc_id, content)
    await ctx.notify_resource_updated(f"doc://{doc_id}")
    return f"حُدِّث doc://{doc_id}"

تُخبر notify_resource_updated العملاء بأن عنواناً واحداً قد تغيّر. أما شقيقاتها — notify_resources_changed وnotify_tools_changed وnotify_prompts_changed — فتقول إن القائمة نفسها تغيّرت، وهو ما ترسله عند إنشاء مستند أو حذفه لا عند تحريره. وإرسال الإشعار الخطأ سبب شائع لبقاء إدخالات قديمة عند العميل: فهو يعيد بإخلاص قراءة مورد ما يزال موجوداً دون أن ينتبه إلى ثلاثة ظهرت بجانبه.

الاشتراك في التحديثات

الاشتراك طلب لا إشعار: له id وله استجابة. وتلك الاستجابة ببساطة تدفق يبقى مفتوحاً:

# من العميل إلى الخادم. لاحظ الـ id: هذا طلب، ولذا ينتظر رداً.
{
    "jsonrpc": "2.0",
    "id": 7,
    "method": "subscriptions/listen",
    "params": { ... },
    "_meta": {
        "io.modelcontextprotocol/protocolVersion": "2026-07-28",
        "io.modelcontextprotocol/clientCapabilities": { ... }
    }
}

# الخادم يُقرّ بالاستلام ثم يبثّ الإشعارات على تلك الاستجابة المفتوحة.
# ويحمل كل إشعار معرّف الاشتراك ليتمكن العميل من ربطه:
#   _meta: { "io.modelcontextprotocol/subscriptionId": ... }

ولأن حالة التدفق مرتبطة بالطلب لا بالاتصال، فانقطاع القناة مشكلة العميل يعالجها بإعادة إصدار subscriptions/listen. وخادمك لا يحتفظ بشيء.

التعامل مع الاشتراكات

class SubscriptionManager:
    def __init__(self):
        self.subscriptions = {}

    def subscribe(self, client_id: str, types: list):
        self.subscriptions[client_id] = set(types)

    def should_notify(self, client_id: str, notification_type: str) -> bool:
        if client_id not in self.subscriptions:
            return True  # الافتراضي: استلام الكل
        return notification_type in self.subscriptions[client_id]

subscriptions = SubscriptionManager()

async def notify_clients(notification_type: str, data: dict):
    for client_id in connected_clients:
        if subscriptions.should_notify(client_id, notification_type):
            await send_to_client(client_id, {
                "type": notification_type,
                "data": data
            })

أفضل الممارسات

  • اجعل الإشعارات خفيفة الوزن
  • تضمين سياق كافٍ لتجنب طلبات المتابعة
  • استخدم أنواع الإشعارات المناسبة
  • تعامل مع العملاء المنقطعين بأمان

الآن دعنا نطبق هذه الأنماط في مختبر عملي. :::

اختبار

اختبار الوحدة 4: أنماط MCP المتقدمة

خذ الاختبار
هل كان هذا الدرس مفيدًا؟

سجّل الدخول للتقييم