امتداد Tasks في MCP: المهام غير المتزامنة والمهام الخلفية موضّحة

يستدعي وكيل الذكاء الاصطناعي أداة، فتحتاج الأداة إلى أربعين دقيقة، وفي حوالي الدقيقة الثانية يغلق وسيط الشبكة الاتصال. قد تظل المهمة قيد التشغيل على الخادم، لكن لم يعد بالإمكان الوصول إليها، ويبقى النموذج أمام خطأ بدل إجابة. هذه الفجوة بالضبط هي ما صُمم امتداد Tasks في MCP لسدّها. فبدل إبقاء طلب واحد مفتوحًا حتى ينتهي العمل، يعيد الخادم فورًا معرّفًا دائمًا (taskId)، ويستعلم العميل عنه متى شاء.

يشرح هذا المقال كيف تعمل المهام غير المتزامنة والمهام الخلفية في Model Context Protocol: ما هو الامتداد، وما معنى كل حالة، وكيف تبدو الحمولات، وكيف تبني خادمًا يدعم المهام، وأي عادات للعميل تحافظ على الأعمال الطويلة آمنة. أسماء الحقول أدناه مأخوذة من مواصفات الامتداد المنشورة (io.modelcontextprotocol/tasks، SEP-2663) ومن توثيق FastMCP.

لماذا تفشل الاستدعاءات المحجوبة

يتصرف استدعاء MCP tools/call القياسي كزبون واقف عند المنضدة ينتظر طبقه. يُرسل الطلب، ويبقى الاتصال مفتوحًا، وتعود الإجابة على الخط نفسه. هذا مناسب تمامًا لبحث عن الطقس. أما لخط أنابيب CI أو استيراد دفعات كبيرة أو تدريب نموذج، فهو خيار سيئ.

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

مشكلة المهلة الزمنية

يفرض كثير من العملاء ووسطاء النقل مهلًا زمنية تجعل إبقاء الطلب مفتوحًا غير عملي لأكثر من بضع ثوانٍ. موازنات الأحمال والبوابات المؤسسية والبوابات بلا خادم كلها تقطع الاتصالات الصامتة. وعندما يحدث ذلك، يرى المستدعي فشلًا رغم أن الخادم ما زال مشغولًا، ورد الفعل الطبيعي هو إعادة المحاولة وتشغيل المهمة المكلفة نفسها مرتين.

فقدان العمل بعد انقطاع الاتصال

يربط الاستدعاء المحجوب النتيجة بالاتصال. يُغلق غطاء الحاسوب المحمول، أو يغادر عميل الهاتف شبكة الواي فاي، أو تُعاد تشغيل عملية المضيف، فلا يوجد مكان تصل إليه الإجابة. مع المهمة، يصبح المعرّف مقبضًا دائمًا: يعيد العميل الاتصال، ويستدعي tasks/get بالمعرّف نفسه، ويكمل من حيث توقف.

💡 قاعدة عامة: إذا كانت عملية تستغرق بانتظام أكثر من بضع ثوانٍ، أو تتوقف بانتظار قرار بشري، فيجب أن تُنفَّذ ضمن مهمة.

ما الذي يضيفه امتداد Tasks

من المواصفة الأساسية إلى الامتداد

بدأت المهام كميزة تجريبية في المواصفة الأساسية لبروتوكول MCP. ثم نقلها البروتوكول لاحقًا من النواة إلى امتداد اختياري يُعرَّف بـ io.modelcontextprotocol/tasks، وموثّق في SEP-2663. يبقي خروجها من النواة البروتوكول الأساسي خفيفًا، بينما تفعّل الخوادم والعملاء الذين يحتاجون إلى أعمال طويلة الأمد هذا الدعم عن قصد. تصف الوثائق الرسمية النتيجة بأنها تنفيذ غير متزامن للمهام لعمليات MCP طويلة الأمد، ويوجد نص المواصفة الكامل في مستودع ext-tasks.

ثمة تغيير يستحق الانتباه. تذكر أوصاف أقدم للميزة استدعاءً منفصلًا باسم tasks/result. في الامتداد، تصل المخرجات النهائية داخل استجابة tasks/get، وهذا يُبقي حلقة العميل عند طريقة استطلاع واحدة فقط.

معرّف الامتداد والاشتراك الاختياري

يُتفاوض على الدعم ولا يُفترض أبدًا:

• يُدرج العميل io.modelcontextprotocol/tasks في قدرات كل طلب، داخل _meta تحت io.modelcontextprotocol/clientCapabilities.
• يعلن الخادم الامتداد نفسه في القدرات التي يعيدها للعملاء.
• إذا كان الخادم يشترط دعم المهام والعميل لم يُعلن ذلك، يرد الخادم برمز الخطأ -32003 مع الرسالة Missing required client capability.

قرار متى تُنشأ المهمة يعود إلى الخادم. لا توجد علامة لكل أداة على جانب العميل. يشترك العميل مرة واحدة، ويجب أن يكون مستعدًا لشكلين من النتائج: النتيجة العادية، أو مقبض مهمة. وحتى الآن، tools/call هو نوع الطلب الوحيد القادر على إنتاج مهمة.

دورة حياة المهمة

الحصول على المقبض والاستطلاع

يتكون التدفق من خمس خطوات:

1. يرسل العميل tools/call مع إرفاق قدرة المهام.
2. يقرر الخادم أن العمل طويل، ويعيد CreateTaskResult موسومًا بـ resultType: "task".
3. تُنشأ المهمة بشكل دائم قبل أن يغادر ذلك الرد الخادم.
4. يستدعي العميل tasks/get مع taskId، وينتظر pollIntervalMs على الأقل بين الاستدعاءات.
5. تحمل كل استجابة الحالة الحالية، وعندما تصل المهمة إلى حالة نهائية، فتعرض النتيجة أو الخطأ.

إليك عرض مبسّط لمهمة جديدة. الغلاف الدقيق معرّف في المواصفة، فاعتبر هذا توضيحًا للحقول:

{
  "resultType": "task",
  "taskId": "tsk_8f3a91c2",
  "status": "working",
  "statusMessage": "Rendering 120 pages",
  "createdAt": "2026-10-06T09:00:00Z",
  "lastUpdatedAt": "2026-10-06T09:00:04Z",
  "ttlMs": 3600000,
  "pollIntervalMs": 2000
}

نسخ الأمر النصي

خمس حالات تصف كل مهمة:

بمجرد أن تصل المهمة إلى حالة نهائية، لا تتغير حالتها أبدًا. إن tasks/get عملية idempotent، أي أن الاستطلاع عشر مرات آمن تمامًا كالاستطلاع مرة واحدة.

الإيقاف مؤقتًا لإدخال بشري

تصل بعض الأعمال إلى نقطة قرار في منتصفها: الموافقة على نشر، أو تأكيد عملية شراء، أو اختيار أحد ثلاثة خيارات. تنتقل المهمة إلى input_required، وتتضمن استجابة tasks/get التالية خريطة inputRequests تحمل طلبات الاستيضاح (elicitations) أو طلبات أخرى من الخادم.

يعرض العميل تلك الطلبات على مستخدم أو نموذج، ثم يجيب عنها عبر tasks/update، مرسلًا inputResponses تطابق الطلبات المعلقة. يقرّ الخادم بذلك بنتيجة فارغة، ويتجاهل الردود الخاصة بمدخلات مجهولة أو مُلبّاة سابقًا. وبمجرد أن يحصل كل طلب على إجابة، يواصل الخادم عمله.

💡 لماذا هذا لطيف: لا يوجد اتصال ثانٍ ولا رسالة غير مطلوبة من الخادم إلى العميل. تمرّ الخطوة البشرية عبر حلقة الاستطلاع نفسها التي تمر بها كل الخطوات الأخرى.

الإنهاء والفشل والإلغاء

عندما تنتهي المهمة بنجاح، يحتوي الحقل result على ما كان الطلب الأصلي سيعيده بشكل متزامن. وبالنسبة لاستدعاء أداة، يعني ذلك كتل المحتوى نفسها التي كان سينتجها الاستدعاء المحجوب. وعندما تكون الحالة failed، يحمل الحقل error خطأ JSON-RPC.

يُستخدم tasks/cancel للإلغاء. يقرّ الخادم بالطلب بنتيجة فارغة، لكن الإلغاء تعاوني. ربما يكون العمل قد تجاوز نقطة اللاعودة، لذلك قد تنتهي المهمة إلى حالة نهائية مختلفة.

يمكن للخوادم أيضًا دفع التحديثات عبر notifications/tasks. يشترك العملاء عبر subscriptions/listen، وكل إشعار يحمل حالة المهمة الكاملة، بالشكل نفسه الذي تعيده استجابة tasks/get.

بناء خادم مهام

أداة FastMCP بسيطة

أضاف FastMCP 4.0 دعمًا للامتداد. تثبّت fastmcp-tasks، وتسجّل TasksExtension، وتعلّم الأداة بأنها قادرة على تشغيل المهام:

import asyncio
from fastmcp import FastMCP
from fastmcp_tasks import TasksExtension

mcp = FastMCP("ReportServer")
mcp.add_extension(TasksExtension())

@mcp.tool(task=True)
async def slow_computation(duration: int) -> str:
    """A long-running operation."""
    for i in range(duration):
        await asyncio.sleep(1)
    return f"Finished in {duration} seconds"

نسخ الأمر النصي

هناك تفصيلان مهمان هنا. تتطلب المهام الخلفية دوال غير متزامنة، واستخدام task=True على دالة متزامنة يرفع ValueError وقت التسجيل. وtask=True يشير فقط إلى أن الأداة قادرة على العمل في الخلفية. ويتوقف ما إذا كانت تعمل فعلًا على اشتراك العميل وعلى وضع التنفيذ في الخادم. تشير الوثائق إلى أن Docket يشغّل المجدول الموزّع، وهذا ما يجعل الإعداد جاهزًا للإنتاج.

التقدم وأوضاع التنفيذ

تبلغ الأدوات عن التقدم عبر اعتماد Progress مُحقن، ومنه تأتي statusMessage التي يعرضها عملاؤك:

@mcp.tool(task=True)
async def process_files(
    files: list[str],
    progress: Progress = Progress()
) -> str:
    await progress.set_total(len(files))
    for file in files:
        await progress.set_message(f"Processing {file}")
        await progress.increment()
    return f"Processed {len(files)} files"

نسخ الأمر النصي

للتحكم الأدق، استبدل القيمة المنطقية بـ TaskConfig. ثلاثة أوضاع تحدد سلوك الأداة:

الاختصارات تتطابق بوضوح: task=True يعني optional، وtask=False يعني forbidden. ويمكنك أيضًا اقتراح وتيرة استطلاع عبر poll_interval=timedelta(seconds=2).

أنماط عملاء تصمد

استطلع بلباقة، واحفظ كل شيء

يحتاج العميل الذي يتعامل مع خوادم قادرة على المهام إلى خمس عادات:

• أعلن الامتداد في قدرات كل طلب.
• تعامل مع النتائج المتعددة الأشكال، لأن tools/call قد يعيد نتيجة عادية أو مهمة.
• احترم pollIntervalMs، لأن الخادم قد يغيّره بين الاستجابات.
• أجب عن inputRequests عبر tasks/update بدل تجاهلها.
• خزّن معرّفات المهام بشكل دائم حتى يستأنف الاستطلاع بعد عطل أو إعادة تشغيل.

الحلقة أدناه شيفرة وهمية pseudocode، وليست مرتبطة بحزمة SDK محددة:

async def run_tool(session, name, args):
    reply = await session.call_tool(name, args)
    if reply.get("resultType") != "task":
        return reply                              # ordinary synchronous result

    task = reply
    store.save(task["taskId"])                    # survive a crash

    while task["status"] in ("working", "input_required"):
        if task["status"] == "input_required":
            answers = await ask_user(task["inputRequests"])
            await session.request("tasks/update", {
                "taskId": task["taskId"],
                "inputResponses": answers,
            })
        await asyncio.sleep(task["pollIntervalMs"] / 1000)
        task = await session.request("tasks/get", {"taskId": task["taskId"]})

    if task["status"] == "failed":
        raise RuntimeError(task["error"])
    return task.get("result")

نسخ الأمر النصي

الراكب في القطار هو النموذج الذهني هنا. ينقطع الاتصال في كل نفق، ومع ذلك تبقى التذكرة في جيبه صالحة. العميل المبني بهذه الطريقة يعيد الاتصال بعد النفق ويواصل.

الإشعارات بدل الاستطلاع

الاستطلاع هو الافتراضي، وهو يعمل في كل مكان. إذا كان الخادم يدعم notifications/tasks، يمكن للعميل أن يشترك مرة واحدة ويتخطى معظم رحلات tasks/get الذهابية والإيابية، لأن كل إشعار يحمل حالة المهمة الكاملة. أبقِ الاستطلاع احتياطيًا للخوادم التي لا تدفع التحديثات.

أخطاء يجب تجنبها

تتصرف مقابض المهام مثل الطرود في غرفة خلفية لمكتب بريد. إذا تُركت طويلًا، تُفرَّغ. هذه أكثر الفخاخ شيوعًا:

القيمة null في ttlMs تعني غير محدود، وهذا يبدو ودودًا حتى يمتلئ التخزين بمهام منتهية لا يجمعها أحد.

ربط المهام مع PicassoIA

توليد الوسائط هو المثال الكلاسيكي للعمل الطويل، ولهذا تبدو أنماط المهام طبيعية بجانب أدوات الصور والفيديو. يناسب نموذجان من PicassoIA سير عمل المهام مباشرة.

كيفية استخدام Claude Sonnet 5

يتعامل Claude Sonnet 5 مع البرمجة متعددة الخطوات وأعمال استخدام الأدوات، لذلك فهو شريك برمجة جيد لشيفرة المعالج في هذا المقال. من الخيارات الأخرى في الفئة نفسها GPT 5.6 Sol، إذا أردت رأيًا ثانيًا في الشيفرة نفسها.

1. افتح صفحة Claude Sonnet 5 على Picasso IA.
2. الصق توقيع أداتك وحقول المهام من هذا المقال في مربع Prompt، ثم اطلب معالجًا غير متزامن مع رسائل حالة.
3. اضبط Effort على high للآلات الحالة المعقدة، أو اتركه على low للتعديلات السريعة.
4. أضف System Prompt مثل "Write Python, async only, no blocking calls" حتى تحافظ كل الردود على الأسلوب نفسه.
5. ارفع Max Tokens فوق القيمة الافتراضية 8,192 إذا أردت توليد الاختبارات في الإجابة نفسها.
6. شغّله، واقرأ النتيجة، والصق المعالج في مشروعك.

💡 نصيحة: أرفق لقطة شاشة للخطأ عبر حقل Image. يقرأها النموذج كسياق.

قصاصات نظيفة للمخططات

توثيق تدفقات المهام يحتاج عادةً إلى صور نظيفة: صورة جهاز لبطاقة حالة، أو شعار لشريحة معمارية. تعيد Remove Background ملف PNG شفافًا خلال ثوانٍ، وإعداد Preserve Partial Alpha فيها يحافظ على حواف ناعمة طبيعية. أوقفه عندما تريد حوافًا حادة وغير شفافة تمامًا في صور المنتجات.

أما لمشاهد جديدة، فإن Flux 2 Pro وP Image يحوّلان الأمر النصي المكتوب إلى صورة تناسب ترويسة مدونة.

أنشئ صورك الخاصة بعد ذلك

لديك الآن تصوّر كامل للأمر: مقبض بدل اتصال محجوب، وخمس حالات، وثلاث طرق، وقائمة قصيرة من العادات تحفظ الأعمال الطويلة من الضياع. أفضل طريقة لترسيخ ذلك هي بناء شيء صغير. اكتب أداة تستغرق عشر ثوانٍ، وعلّمها بـ task=True، وراقب الحالة تنتقل من working إلى حالة نهائية.

ثم امنح مشروعك وجهًا. افتح Picasso IA، واختر نموذجًا لتحويل النص إلى صورة، وولّد صورة ترويسة لكتابتك. جرّب زوايا الكاميرا والإضاءة وتفاصيل العدسة في أوامرك النصية، وأزل خلفية لشعار نظيف، ولاحظ مدى سرعة تحوّل الفكرة إلى مرئية مكتملة. نتيجة مهمتك التالية تستحق صورة تستحق المشاركة.