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

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

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

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

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

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

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

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

@server.notification()
async def send_notification(method: str, params: dict):
    # MCP يوجه تلقائياً إلى العملاء المتصلين
    pass

# في أداتك أو المهمة الخلفية
async def long_running_task():
    for i, chunk in enumerate(process_data()):
        # إرسال تحديث التقدم
        await server.notify(
            "notifications/progress",
            {"task_id": "abc", "progress": i / 100}
        )

    # إرسال الاكتمال
    await server.notify(
        "notifications/complete",
        {"task_id": "abc", "result": "success"}
    )

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

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

async def update_document(doc_id: str, content: str):
    # تحديث المستند
    await db.update(doc_id, content)

    # إعلام العملاء بالتغيير
    await server.notify(
        "notifications/resources/updated",
        {"uri": f"doc://{doc_id}"}
    )

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

الاشتراك طلب لا إشعار: له 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
            })

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

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

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