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

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

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

يدعم MCP الإشعارات المبدوءة من الخادم، مما يمكّن التحديثات الفورية للعملاء المتصلين.

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

النوع حالة الاستخدام
تحديثات الموارد تغيرت البيانات
تحديثات التقدم المهام طويلة المدى
تنبيهات النظام الأخطاء، التحذيرات

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

@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}"}
    )

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

يمكن لمضيفي MCP الاشتراك في إشعارات محددة:

# اشتراك العميل (جانب المضيف)
{
    "jsonrpc": "2.0",
    "method": "notifications/subscribe",
    "params": {
        "types": ["resources/updated", "progress"]
    }
}

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

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 المتقدمة

خذ الاختبار