آموزش و راهنما

MCP Tasks چیست؟ اجرای وظایف طولانی در AI Agentها

با MCP Tasks می‌توان Toolهای طولانی را بدون باز نگه‌داشتن اتصال اجرا کرد، وضعیت آن‌ها را پیگیری کرد، ورودی کاربر را در میانه کار دریافت کرد و نتیجه را پس از اتصال مجدد بازیابی کرد.

همه ابزارهای AI Agent در چند ثانیه پاسخ نمی‌دهند.

برخی عملیات ممکن است چند دقیقه یا حتی چند ساعت طول بکشند:

اجرای CI/CD Pipeline
پردازش هزاران سند
ساخت ویدئو
آموزش یا Fine-Tuning مدل
تحلیل Repository بزرگ
Deployment زیرساخت
مهاجرت داده
تهیه گزارش سازمانی
انتظار برای تأیید انسانی
اجرای Batch Job
فراخوانی APIهای Asynchronous

در معماری معمولی Tool Calling، Client درخواست را ارسال می‌کند و تا زمان آماده‌شدن نتیجه منتظر می‌ماند.

Client
↓
tools/call
↓
MCP Server
↓
عملیات طولانی
↓
نتیجه

این روش برای عملیات کوتاه مناسب است. اما اگر یک Tool سی دقیقه زمان نیاز داشته باشد، باز نگه‌داشتن اتصال مشکلات متعددی ایجاد می‌کند:

Timeout در Client
قطع اتصال شبکه
اشغال Worker یا Connection
از بین رفتن نتیجه پس از Restart
ناتوانی در نمایش وضعیت
دشواری لغو عملیات
نبود مسیر دریافت ورودی کاربر
اجرای تکراری کار پس از Retry

MCP Tasks برای حل این مشکلات طراحی شده است.

به‌جای منتظر نگه‌داشتن Client، سرور یک شناسه پایدار به نام taskId برمی‌گرداند. Client می‌تواند با این شناسه وضعیت کار را بررسی کند و پس از پایان، نتیجه را دریافت کند.

فهرست مطالب

MCP Tasks چیست؟
چرا Tool Calling معمولی کافی نیست؟
معماری MCP Tasks
چرخه عمر Task
Capability Negotiation
ساخت Task
دریافت وضعیت با tasks/get
ارسال ورودی با tasks/update
لغو با tasks/cancel
Polling و Notification
تفاوت Tasks و Progress
تفاوت Tasks و Queue
ارتباط Tasks و Elicitation
ارتباط با Code Mode
طراحی Task Store
Idempotency و Retry
خطا و Partial Failure
امنیت و دسترسی
پیاده‌سازی نمونه
کاربردهای عملی
Observability
معماری Production
چک‌لیست انتشار
پرسش‌های متداول
جمع‌بندی

MCP Tasks چیست؟

MCP Tasks افزونه‌ای برای Model Context Protocol است که اجرای Asynchronous عملیات طولانی را امکان‌پذیر می‌کند.

در این معماری، MCP Server به‌جای بازگرداندن نتیجه نهایی Tool، یک Task Handle پایدار برمی‌گرداند:

{
  "resultType": "task",
  "task": {
    "taskId": "task_7f28c4a1",
    "status": "working",
    "statusMessage": "Processing documents",
    "ttlMs": 86400000,
    "pollIntervalMs": 3000
  }
}

Client اتصال را آزاد می‌کند و هر چند ثانیه وضعیت Task را می‌پرسد:

tasks/get

وقتی Task کامل شد، پاسخ نهایی در همان وضعیت قرار می‌گیرد:

{
  "taskId": "task_7f28c4a1",
  "status": "completed",
  "result": {
    "content": [
      {
        "type": "text",
        "text": "۲۵۰۰ سند با موفقیت پردازش شد."
      }
    ]
  }
}

اگر Client قطع یا Restart شود، می‌تواند با همان taskId ادامه دهد.

طبق مستندات رسمی MCP Tasks، Task یک State Machine پایدار است که وضعیت عملیات، درخواست‌های ورودی و نتیجه نهایی را نگهداری می‌کند.

وضعیت فعلی MCP Tasks

MCP Tasks در نسخه 2025-11-25 ابتدا به‌عنوان قابلیتی آزمایشی در Core Protocol معرفی شد.

در نسخه 2026-07-28 طراحی آن تغییر کرد و Tasks از Core خارج شد و به یک Extension مستقل منتقل شد:

io.modelcontextprotocol/tasks

نسخه جدید از این Methodها استفاده می‌کند:

tasks/get
tasks/update
tasks/cancel

پیاده‌سازی قدیمی و Extension جدید از نظر Wire Protocol کاملاً سازگار نیستند. بنابراین هنگام توسعه باید نسخه Protocol و پشتیبانی SDK بررسی شود.

MCP Tasks در زمان نگارش این مقاله همچنان یک Extension درحال‌توسعه است و میزان پشتیبانی آن در Clientها و SDKها متفاوت است.

چرا Tool Calling معمولی کافی نیست؟

Tool Calling مستقیم معمولاً ساختار هم‌زمان دارد:

Request
↓
Processing
↓
Response

Client تا آماده‌شدن پاسخ منتظر می‌ماند. این معماری برای عملیات چندثانیه‌ای مناسب است، اما برای کارهای طولانی محدودیت دارد.

Timeout

Load Balancer، Proxy، Client یا Server ممکن است اتصال طولانی را ببندد.

قطع اتصال

اگر اینترنت کاربر قطع شود، عملیات Backend ممکن است ادامه پیدا کند، اما Client دیگر راهی برای دریافت نتیجه ندارد.

Retry خطرناک

Client ممکن است تصور کند عملیات شکست خورده و درخواست را دوباره ارسال کند. این رفتار می‌تواند دو Job مشابه بسازد.

نبود وضعیت پایدار

Client نمی‌داند عملیات در چه مرحله‌ای قرار دارد:

در صف
در حال اجرا
منتظر تأیید
کامل‌شده
شکست‌خورده
لغوشده

نبود تعامل در میانه اجرا

بعضی Workflowها پس از چند مرحله به تأیید یا اطلاعات بیشتری نیاز دارند.

برای مثال:

Deployment آماده است.
آیا انتشار روی Production تأیید می‌شود؟

MCP Tasks این وضعیت را با input_required مدیریت می‌کند.

معماری MCP Tasks

معماری معمول شامل چهار جزء است.

MCP Client

Tool را فراخوانی می‌کند، Task Handle را دریافت می‌کند و وضعیت را پیگیری می‌کند.

MCP Server

تصمیم می‌گیرد درخواست به‌صورت مستقیم اجرا شود یا به Task تبدیل شود.

Task Store

وضعیت Task را به‌صورت پایدار نگهداری می‌کند.

Worker

عملیات اصلی را در پس‌زمینه اجرا و وضعیت Task را به‌روزرسانی می‌کند.

MCP Client
↓
MCP Server
↓
Task Store
↓
Queue
↓
Worker
↓
سرویس خارجی

Task Store باید مستقل از اتصال Client و عمر Process باشد. ذخیره Task فقط در Memory برای Production کافی نیست.

چرا Task Store باید پایدار باشد؟

فرض کنید Server این پاسخ را برگرداند:

{
  "taskId": "task_123",
  "status": "working"
}

اما Task فقط در حافظه همان Process ذخیره شده باشد. اگر Server Restart شود، درخواست زیر شکست می‌خورد:

{
  "method": "tasks/get",
  "params": {
    "taskId": "task_123"
  }
}

برای حفظ قابلیت Resume، Task باید در یک Storage پایدار ذخیره شود:

PostgreSQL
Redis همراه Persistence
Durable Object
Managed Job Store
پایگاه داده توزیع‌شده
سیستم Workflow مانند Temporal

طبق Specification، Task باید پیش از بازگرداندن CreateTaskResult واقعاً در Task Store ایجاد و قابل بازیابی شده باشد.

چرخه عمر Task

MCP Tasks پنج وضعیت اصلی دارد:

وضعیت	معنی
`working`	عملیات در حال اجرا است
`input_required`	برای ادامه، ورودی Client لازم است
`completed`	عملیات با نتیجه نهایی کامل شده است
`failed`	عملیات با خطای Protocol یا اجرایی شکست خورده است
`cancelled`	عملیات لغو شده است

وضعیت‌های زیر Terminal هستند:

completed
failed
cancelled

Task پس از رسیدن به وضعیت Terminal نباید دوباره به working بازگردد.

جریان اصلی:

working
├── completed
├── failed
├── cancelled
└── input_required
        └── working

Task ممکن است چند بار بین working و input_required جابه‌جا شود:

working
↓
input_required
↓
working
↓
input_required
↓
working
↓
completed

Capability Negotiation

Client و Server باید پشتیبانی از Tasks Extension را اعلام کنند.

شناسه Extension:

io.modelcontextprotocol/tasks

Client در Metadata درخواست اعلام می‌کند که Tasks را می‌فهمد:

{
  "params": {
    "_meta": {
      "io.modelcontextprotocol/clientCapabilities": {
        "extensions": {
          "io.modelcontextprotocol/tasks": {}
        }
      }
    }
  }
}

Server نیز پشتیبانی خود را در server/discover اعلام می‌کند:

{
  "capabilities": {
    "extensions": {
      "io.modelcontextprotocol/tasks": {}
    }
  }
}

Server نباید به Clientی که این Capability را اعلام نکرده، Task Handle برگرداند.

برای Client فاقد پشتیبانی، Server می‌تواند:

درخواست را به‌شکل عادی و Blocking اجرا کند.
خطای Missing Required Capability برگرداند.
یک مسیر جایگزین ارائه دهد.
اجرای Tool را رد کند.

Task را چه کسی ایجاد می‌کند؟

در طراحی جدید، ایجاد Task به‌صورت Server-directed است.

Client اعلام می‌کند:

من از Tasks پشتیبانی می‌کنم.

اما Server تصمیم می‌گیرد:

این درخواست خاص باید به Task تبدیل شود.

این تصمیم می‌تواند براساس عوامل مختلف گرفته شود:

حجم ورودی
زمان تخمینی اجرا
نوع Tool
شلوغی Queue
محدودیت Provider
نیاز به تأیید انسانی
وضعیت سرویس خارجی
تنظیمات سازمان
سطح دسترسی کاربر

برای مثال، پردازش ۱۰ فایل ممکن است مستقیم انجام شود، اما پردازش ۱۰ هزار فایل به Task تبدیل شود.

پاسخ مستقیم یا Task

Client باید آماده دریافت دو شکل پاسخ باشد.

پاسخ مستقیم

{
  "resultType": "complete",
  "content": [
    {
      "type": "text",
      "text": "عملیات کامل شد."
    }
  ]
}

پاسخ Task

{
  "resultType": "task",
  "task": {
    "taskId": "task_01K3...",
    "status": "working",
    "ttlMs": 86400000,
    "pollIntervalMs": 3000
  }
}

بنابراین نتیجه tools/call در Client باید Polymorphic مدیریت شود.

ساخت Task

فرض کنید کاربر می‌خواهد هزار سند را تحلیل کند.

درخواست:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "analyze_documents",
    "arguments": {
      "folderId": "folder_72",
      "outputFormat": "json"
    },
    "_meta": {
      "io.modelcontextprotocol/clientCapabilities": {
        "extensions": {
          "io.modelcontextprotocol/tasks": {}
        }
      }
    }
  }
}

Server ابتدا Task را در پایگاه داده ایجاد می‌کند:

{
  "task_id": "task_b8701",
  "status": "working",
  "tool_name": "analyze_documents",
  "progress": 0,
  "created_at": "2026-08-17T10:00:00Z"
}

سپس Job را به Queue می‌فرستد و Task Handle را برمی‌گرداند:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "resultType": "task",
    "task": {
      "taskId": "task_b8701",
      "status": "working",
      "statusMessage": "Waiting for a worker",
      "createdAt": "2026-08-17T10:00:00Z",
      "lastUpdatedAt": "2026-08-17T10:00:00Z",
      "ttlMs": 86400000,
      "pollIntervalMs": 5000
    }
  }
}

فیلدهای مهم Task

taskId

شناسه منحصربه‌فرد و غیرقابل‌حدس Task است.

status

وضعیت فعلی Task را نشان می‌دهد.

statusMessage

توضیح کوتاهی درباره مرحله فعلی ارائه می‌کند:

Processing document 420 of 1000

createdAt

زمان ایجاد Task با فرمت ISO 8601.

lastUpdatedAt

آخرین زمان تغییر وضعیت یا اطلاعات Task.

ttlMs

مدتی که Task و نتیجه آن قابل‌بازیابی باقی می‌مانند.

pollIntervalMs

فاصله پیشنهادی میان درخواست‌های tasks/get.

این مقدار ممکن است در طول اجرا تغییر کند.

دریافت وضعیت با tasks/get

Client وضعیت Task را با tasks/get دریافت می‌کند:

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tasks/get",
  "params": {
    "taskId": "task_b8701"
  }
}

پاسخ:

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "taskId": "task_b8701",
    "status": "working",
    "statusMessage": "Processed 420 of 1000 documents",
    "createdAt": "2026-08-17T10:00:00Z",
    "lastUpdatedAt": "2026-08-17T10:04:20Z",
    "ttlMs": 86400000,
    "pollIntervalMs": 5000
  }
}

Client باید حداقل به‌اندازه pollIntervalMs صبر کند و سپس دوباره Poll انجام دهد.

الگوریتم Polling

نمونه مفهومی با Python:

import asyncio
from typing import Any


TERMINAL_STATUSES = {
    "completed",
    "failed",
    "cancelled",
}


async def wait_for_task(
    client,
    task_id: str,
) -> dict[str, Any]:
    while True:
        task = await client.get_task(
            task_id=task_id,
        )

        if task["status"] in TERMINAL_STATUSES:
            return task

        if task["status"] == "input_required":
            return task

        poll_interval_ms = task.get(
            "pollIntervalMs",
            3000,
        )

        await asyncio.sleep(
            poll_interval_ms / 1000
        )

در Production باید موارد زیر نیز مدیریت شوند:

Timeout کلی
Cancellation محلی
Network Error
Backoff
Task Expiration
Authentication Refresh
Rate Limit
توقف Polling هنگام بسته‌شدن UI

از Polling تهاجمی خودداری کنید

درخواست وضعیت هر ۱۰۰ میلی‌ثانیه فشار غیرضروری ایجاد می‌کند.

Client باید مقدار pollIntervalMs را رعایت کند:

Task سریع:
1000 تا 3000 میلی‌ثانیه

Task چنددقیقه‌ای:
3000 تا 10000 میلی‌ثانیه

Task چندساعته:
30000 میلی‌ثانیه یا بیشتر

مقادیر دقیق باید براساس نوع Task و ظرفیت Server تنظیم شوند.

تکمیل Task

وقتی Worker عملیات را تمام می‌کند، نتیجه را در Task Store ذخیره می‌کند:

{
  "taskId": "task_b8701",
  "status": "completed",
  "statusMessage": "All documents processed",
  "result": {
    "content": [
      {
        "type": "text",
        "text": "۱۰۰۰ سند پردازش شد."
      }
    ],
    "structuredContent": {
      "documentsProcessed": 1000,
      "successful": 987,
      "failed": 13,
      "reportId": "report_921"
    }
  }
}

نتیجه باید همان ساختاری را داشته باشد که اگر Tool به‌صورت هم‌زمان اجرا می‌شد، بازمی‌گرداند.

input_required چیست؟

گاهی Task بدون اطلاعات یا تأیید جدید نمی‌تواند ادامه پیدا کند.

برای مثال، Deployment پس از ساخت نسخه و اجرای تست‌ها آماده Production است، اما باید کاربر آن را تأیید کند.

Task به وضعیت زیر منتقل می‌شود:

{
  "taskId": "task_deploy_72",
  "status": "input_required",
  "statusMessage": "Waiting for production approval",
  "inputRequests": {
    "production_approval": {
      "method": "elicitation/create",
      "params": {
        "mode": "form",
        "message": "آیا انتشار نسخه ۲.۴ روی Production تأیید می‌شود؟",
        "requestedSchema": {
          "type": "object",
          "properties": {
            "approved": {
              "type": "boolean",
              "title": "تأیید انتشار"
            }
          },
          "required": [
            "approved"
          ]
        }
      }
    }
  }
}

Client فرم را به کاربر نمایش می‌دهد.

ارسال پاسخ با tasks/update

پس از پاسخ کاربر:

{
  "jsonrpc": "2.0",
  "id": 5,
  "method": "tasks/update",
  "params": {
    "taskId": "task_deploy_72",
    "inputResponses": {
      "production_approval": {
        "action": "accept",
        "content": {
          "approved": true
        }
      }
    }
  }
}

Server پاسخ را اعتبارسنجی می‌کند و Task را دوباره به working می‌برد:

input_required
↓
tasks/update
↓
working
↓
completed

اگر کاربر تأیید نکند، Task می‌تواند:

cancelled شود.
با نتیجه «اجرا نشد» کامل شود.
به مسیر جایگزین برود.
در محیط Staging باقی بماند.

رفتار دقیق باید در منطق Workflow تعریف شود.

پاسخ‌های جزئی به inputRequests

ممکن است یک Task هم‌زمان چند ورودی بخواهد:

{
  "inputRequests": {
    "approval": {},
    "environment": {},
    "notification_preference": {}
  }
}

Client می‌تواند بخشی از پاسخ‌ها را ارسال کند. Task تا زمان دریافت تمام ورودی‌های ضروری در وضعیت input_required باقی می‌ماند.

کلیدهای inputRequests باید در طول عمر Task منحصربه‌فرد باشند.

لغو Task با tasks/cancel

Client می‌تواند درخواست لغو ارسال کند:

{
  "jsonrpc": "2.0",
  "id": 8,
  "method": "tasks/cancel",
  "params": {
    "taskId": "task_b8701"
  }
}

اما Cancellation در MCP Tasks به‌صورت Cooperative است.

یعنی Server درخواست لغو را می‌پذیرد، اما تضمین نمی‌کند عملیات بلافاصله متوقف شود.

ممکن است:

Worker هنوز پیام لغو را ندیده باشد.
سرویس خارجی قابلیت Cancel نداشته باشد.
عملیات وارد مرحله غیرقابل‌بازگشت شده باشد.
Job درست قبل از لغو کامل شده باشد.

بنابراین Task پس از درخواست لغو ممکن است همچنان به یکی از این وضعیت‌ها برسد:

cancelled
completed
failed

Client نباید صرفاً با دریافت Acknowledgement فرض کند عملیات لغو شده است. وضعیت نهایی باید با tasks/get بررسی شود.

طراحی Worker قابل‌لغو

Worker باید در نقاط مناسب Cancellation را بررسی کند:

async def process_documents(
    task_id: str,
    documents: list[str],
    task_store,
) -> None:
    for index, document in enumerate(documents):
        if await task_store.cancel_requested(
            task_id
        ):
            await task_store.mark_cancelled(
                task_id
            )
            return

        await process_document(document)

        await task_store.update_progress(
            task_id=task_id,
            completed=index + 1,
            total=len(documents),
        )

برای عملیات‌های طولانی، بررسی لغو فقط در پایان کافی نیست.

Notification به‌جای Polling

Polling روش پیش‌فرض است. اما Server می‌تواند وضعیت را از طریق Notification ارسال کند.

Client از subscriptions/listen برای دریافت تغییرات استفاده می‌کند و Server اعلان وضعیت Task را Push می‌کند.

مزایا:

درخواست کمتر
نمایش سریع‌تر تغییر وضعیت
کاهش بار Server
تجربه کاربری بهتر

بااین‌حال، Notification جایگزین Task Store نیست. اگر Connection قطع شود، Client باید بتواند با tasks/get وضعیت واقعی را بازیابی کند.

Notification برای اطلاع سریع است؛ Task Store منبع اصلی حقیقت باقی می‌ماند.

تفاوت MCP Tasks و Progress Notification

Progress Notification برای اعلام پیشرفت یک درخواست در حال اجرا استفاده می‌شود:

۲۰ درصد
۵۰ درصد
۸۰ درصد

اما Task قابلیت‌های بیشتری دارد:

ویژگی	Progress	MCP Tasks
نیاز به اتصال باز	معمولاً بله	خیر
بازیابی پس از قطع اتصال	محدود	بله
شناسه پایدار	لزوماً خیر	بله
نتیجه Deferred	خیر	بله
وضعیت Terminal	محدود	بله
ورودی میانه اجرا	محدود	بله
لغو	وابسته به اتصال	با tasks/cancel
TTL نتیجه	ندارد	دارد

Progress برای درخواست‌های کوتاه‌تر با اتصال فعال مناسب است. Tasks برای عملیات Durable و Asynchronous طراحی شده است.

تفاوت MCP Tasks و Queue

MCP Tasks یک Protocol Interface برای Client است. Queue زیرساخت داخلی اجرای Job است.

MCP Tasks
→ نحوه تعامل Client با عملیات طولانی

Queue
→ نحوه اجرای عملیات در Backend

می‌توان MCP Tasks را بدون Queue پیاده‌سازی کرد، اما در Production معمولاً ترکیب آن با Queue مناسب‌تر است:

tools/call
↓
ساخت Task
↓
ارسال Job به Queue
↓
Worker
↓
به‌روزرسانی Task Store
↓
tasks/get

گزینه‌های Queue:

Celery
Redis Queue
RabbitMQ
Kafka
BullMQ
Cloud Queue
Temporal
سیستم Job اختصاصی

MCP Tasks جایگزین Queue، Scheduler یا Workflow Engine نیست؛ بلکه یک رابط استاندارد برای نمایش وضعیت آن‌ها به MCP Client است.

تفاوت Task و Job

در معماری ساده ممکن است هر Task دقیقاً معادل یک Job باشد:

Task → Job

اما در سیستم بزرگ‌تر، یک Task می‌تواند چند Job داشته باشد:

Task تحلیل اسناد
├── Job تقسیم فایل‌ها
├── Job استخراج متن
├── Job تحلیل
├── Job ساخت گزارش
└── Job ذخیره نتیجه

Client فقط Task سطح بالا را می‌بیند. جزئیات Jobها داخل Backend مدیریت می‌شوند.

ارتباط Tasks و Elicitation

MCP Elicitation برای دریافت اطلاعات از کاربر استفاده می‌شود. MCP Tasks می‌تواند اجرای طولانی را در زمان انتظار برای Elicitation متوقف کند.

مثال:

شروع Deployment
↓
Build
↓
Test
↓
Task: input_required
↓
Elicitation برای تأیید
↓
tasks/update
↓
Deploy
↓
Task: completed

بدون Tasks، Server باید اتصال را تا زمان پاسخ کاربر باز نگه دارد یا State را خارج از Protocol مدیریت کند.

ترکیب این دو قابلیت برای Human-in-the-Loop Workflow مناسب است.

ارتباط Tasks و Programmatic Tool Calling

Code Mode برای اجرای چند Tool در یک Script استفاده می‌شود. اگر Script یا Workflow طولانی باشد، کل اجرای Code Mode می‌تواند به Task تبدیل شود:

مدل Script تولید می‌کند
↓
Server یک Task می‌سازد
↓
Sandbox Execution در Worker
↓
Tool Callهای متعدد
↓
ثبت Progress
↓
نتیجه نهایی

در این معماری:

Code Mode نحوه اجرای ابزارها را تعیین می‌کند.
MCP Tasks وضعیت اجرای طولانی را مدیریت می‌کند.
Progressive Discovery ابزارهای موردنیاز را پیدا می‌کند.
Elicitation ورودی کاربر را دریافت می‌کند.

طراحی جدول Task

نمونه Schema در PostgreSQL:

CREATE TABLE mcp_tasks (
    task_id TEXT PRIMARY KEY,
    owner_id TEXT NOT NULL,
    tool_name TEXT NOT NULL,
    status TEXT NOT NULL,
    status_message TEXT,
    arguments JSONB NOT NULL,
    result JSONB,
    error JSONB,
    input_requests JSONB,
    progress JSONB,
    cancel_requested BOOLEAN NOT NULL DEFAULT FALSE,
    created_at TIMESTAMPTZ NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL,
    expires_at TIMESTAMPTZ
);

Constraint وضعیت:

ALTER TABLE mcp_tasks
ADD CONSTRAINT valid_task_status
CHECK (
    status IN (
        'working',
        'input_required',
        'completed',
        'failed',
        'cancelled'
    )
);

در Production ممکن است جداول جداگانه‌ای برای این موارد نیاز باشد:

Task Event
Input Request
Input Response
Worker Lease
Retry Attempt
Side Effect
Audit Log

State Transition اتمیک

دو Worker نباید هم‌زمان یک Task را کامل کنند.

به‌روزرسانی وضعیت باید اتمیک باشد:

UPDATE mcp_tasks
SET
    status = 'completed',
    result = $1,
    updated_at = NOW()
WHERE
    task_id = $2
    AND status = 'working';

اگر تعداد ردیف به‌روزرسانی‌شده صفر باشد، Task احتمالاً قبلاً لغو یا کامل شده است.

Worker Lease

برای جلوگیری از اجرای هم‌زمان یک Job توسط دو Worker می‌توان Lease تعریف کرد:

{
  "workerId": "worker_7",
  "leaseExpiresAt": "2026-08-17T10:05:00Z"
}

Worker باید Lease را تمدید کند. اگر Worker Crash کند، پس از انقضا Worker دیگری می‌تواند کار را ادامه دهد.

این منطق داخلی Backend است، اما برای Durable بودن Task اهمیت زیادی دارد.

Idempotency

اگر Client به‌دلیل Timeout همان tools/call را دوباره ارسال کند، Server نباید دو Task مشابه بسازد.

Client می‌تواند Idempotency Key ارسال کند:

{
  "name": "analyze_documents",
  "arguments": {
    "folderId": "folder_72"
  },
  "_meta": {
    "idempotencyKey": "analyze-folder-72-v1"
  }
}

Server این کلید را با Owner و Tool مرتبط می‌کند:

owner_id
+
tool_name
+
idempotency_key

اگر درخواست تکراری باشد، همان taskId قبلی بازگردانده می‌شود.

Retry در Worker

Retry فقط برای خطاهای موقت مناسب است:

Timeout سرویس خارجی
Rate Limit
خطای شبکه
خطای موقت Provider
Worker Crash

برای این خطاها Retry مناسب نیست:

Permission Denied
ورودی نامعتبر
فایل حذف‌شده
Schema ناسازگار
درخواست ردشده توسط کاربر
عملیات ممنوع
Credential نامعتبر بدون امکان Refresh

Backoff نمونه:

تلاش اول: ۵ ثانیه
تلاش دوم: ۳۰ ثانیه
تلاش سوم: ۲ دقیقه
تلاش چهارم: ۱۰ دقیقه

تعداد Retry باید محدود باشد.

Partial Failure

Batch Job ممکن است بخشی از آیتم‌ها را با موفقیت و بخشی را با خطا پردازش کند.

در این حالت لزوماً نباید کل Task failed شود.

نتیجه مناسب:

{
  "status": "completed",
  "result": {
    "documentsProcessed": 1000,
    "successful": 987,
    "failed": 13,
    "failedItemsReport": "report_failed_72"
  }
}

failed بهتر است برای زمانی استفاده شود که خود عملیات نتوانسته نتیجه تعریف‌شده را تولید کند.

تفاوت مهم:

Task completed with 13 failed items

با:

Task failed before producing a result

Error ساختاریافته

نمونه Task شکست‌خورده:

{
  "taskId": "task_b8701",
  "status": "failed",
  "statusMessage": "Unable to access document storage",
  "error": {
    "code": -32050,
    "message": "External service unavailable",
    "data": {
      "retryable": true,
      "attempts": 4
    }
  }
}

پیام خام Provider، Stack Trace، API Key یا جزئیات داخلی نباید به Client ارسال شوند.

TTL و نگهداری نتیجه

Task نباید الزاماً برای همیشه ذخیره شود.

ttlMs مشخص می‌کند Task چه مدت پس از ایجاد قابل‌بازیابی است:

{
  "ttlMs": 86400000
}

یعنی ۲۴ ساعت.

سیاست نگهداری می‌تواند براساس نوع Task متفاوت باشد:

نوع Task	TTL پیشنهادی نمونه
گزارش کوچک	۲۴ ساعت
پردازش سند	۳ تا ۷ روز
Deployment	۳۰ روز
Audit مهم	نگهداری جداگانه بلندمدت

پس از TTL:

نتیجه Task حذف یا Archive می‌شود.
tasks/get باید خطای Task Not Found یا Expired برگرداند.
Artifactهای مهم می‌توانند جداگانه باقی بمانند.

Task Store نباید جایگزین سیستم نگهداری دائمی گزارش یا فایل شود.

Task ID

Task ID باید:

منحصربه‌فرد باشد.
تصادفی و غیرقابل‌حدس باشد.
Entropy کافی داشته باشد.
اطلاعات حساس را آشکار نکند.
شماره ترتیبی ساده نباشد.

نامناسب:

task_1
task_2
task_3

مناسب‌تر:

task_01K3M8Y2NQ6E7V4W9A1P

Task ID ممکن است نقش Bearer Handle برای State ذخیره‌شده داشته باشد. بااین‌حال Server همچنان باید مالکیت و Authorization را بررسی کند.

چرا tasks/list وجود ندارد؟

در طراحی Extension جدید، Method عمومی tasks/list حذف شده است.

این تصمیم احتمال افشای Taskهای کاربران یا Sessionهای دیگر را کاهش می‌دهد.

Client باید شناسه Taskهای خودش را نگهداری کند. اگر محصول به صفحه «وظایف من» نیاز دارد، این قابلیت باید در Application API جداگانه و همراه با Authorization مناسب ساخته شود.

نبود tasks/list به این معنا نیست که Backend نمی‌تواند فهرست Taskها داشته باشد؛ فقط چنین قابلیتی در Extension عمومی MCP تعریف نشده است.

Authorization

در هر درخواست tasks/get، tasks/update و tasks/cancel باید بررسی شود:

Caller چه کسی است؟
Task متعلق به کدام کاربر است؟
Tenant یکسان است؟
Token هنوز معتبر است؟
Caller اجازه مشاهده نتیجه دارد؟
Caller اجازه لغو دارد؟
ورودی متعلق به همان Task است؟

صرف دانستن taskId نباید همیشه برای دسترسی کافی باشد.

جداسازی Tenant

Task متعلق به سازمان اول نباید برای سازمان دوم قابل‌مشاهده باشد.

Query مناسب:

SELECT *
FROM mcp_tasks
WHERE
    task_id = $1
    AND tenant_id = $2
    AND owner_id = $3;

این کنترل باید برای Result، Artifact، Input Request و Log نیز اعمال شود.

داده حساس در وضعیت Task

statusMessage نباید حاوی اطلاعات حساس باشد.

نامناسب:

Processing passport_ali_ahmadi_123456.pdf

مناسب‌تر:

Processing document 42 of 100

اگر وضعیت در UI، Log یا Notification نمایش داده شود، ممکن است افراد بیشتری آن را ببینند.

Progress چگونه ذخیره شود؟

Specification فیلد وضعیت و پیام را فراهم می‌کند. برنامه می‌تواند جزئیات پیشرفت را نیز در State داخلی ذخیره کند:

{
  "completed": 420,
  "total": 1000,
  "percent": 42,
  "currentStage": "extracting_text"
}

اما درصد باید واقعی باشد. برای Workflowهایی که زمان هر مرحله مشخص نیست، Status مرحله‌ای بهتر از درصد مصنوعی است:

در صف
استخراج متن
تحلیل
ساخت گزارش
ذخیره نتیجه

پیاده‌سازی ساده Task Store با Python

مدل داده:

from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any, Literal
from uuid import uuid4


TaskStatus = Literal[
    "working",
    "input_required",
    "completed",
    "failed",
    "cancelled",
]


@dataclass
class Task:
    task_id: str
    owner_id: str
    status: TaskStatus
    status_message: str | None
    created_at: datetime
    updated_at: datetime
    poll_interval_ms: int
    result: dict[str, Any] | None = None
    error: dict[str, Any] | None = None
    input_requests: dict[str, Any] = field(
        default_factory=dict
    )
    cancel_requested: bool = False


def create_task(
    owner_id: str,
) -> Task:
    now = datetime.now(UTC)

    return Task(
        task_id=f"task_{uuid4().hex}",
        owner_id=owner_id,
        status="working",
        status_message="Waiting for execution",
        created_at=now,
        updated_at=now,
        poll_interval_ms=3000,
    )

این نمونه فقط برای توضیح ساختار است. In-memory Store برای Production Durable نیست.

Worker نمونه

import asyncio
from datetime import UTC, datetime


async def run_document_task(
    task: Task,
    documents: list[str],
    task_store,
) -> None:
    try:
        total = len(documents)
        successful = 0
        failed = 0

        for index, document in enumerate(
            documents,
            start=1,
        ):
            latest = await task_store.get(
                task.task_id
            )

            if latest.cancel_requested:
                latest.status = "cancelled"
                latest.status_message = (
                    "Cancelled by the client"
                )
                latest.updated_at = datetime.now(UTC)

                await task_store.save(latest)
                return

            try:
                await process_document(document)
                successful += 1
            except DocumentProcessingError:
                failed += 1

            latest.status_message = (
                f"Processed {index} of {total} documents"
            )
            latest.updated_at = datetime.now(UTC)

            await task_store.save(latest)

        latest = await task_store.get(
            task.task_id
        )

        latest.status = "completed"
        latest.status_message = "Processing completed"
        latest.result = {
            "documentsProcessed": total,
            "successful": successful,
            "failed": failed,
        }
        latest.updated_at = datetime.now(UTC)

        await task_store.save(latest)

    except Exception:
        latest = await task_store.get(
            task.task_id
        )

        latest.status = "failed"
        latest.status_message = (
            "Unable to complete document processing"
        )
        latest.error = {
            "code": "PROCESSING_FAILED",
            "message": (
                "The task could not be completed"
            ),
        }
        latest.updated_at = datetime.now(UTC)

        await task_store.save(latest)

در Production خطای واقعی باید در Log داخلی ثبت شود، اما Stack Trace نباید داخل پاسخ عمومی قرار گیرد.

اتصال Task به API مدل

ممکن است Worker برای تحلیل هر سند از یک مدل زبانی استفاده کند.

ساخت Client درواره:

import os

from openai import AsyncOpenAI


client = AsyncOpenAI(
    api_key=os.environ["DARVAREH_API_KEY"],
    base_url="https://api.darvareh.ir/v1",
)

پردازش نمونه:

async def analyze_document(
    text: str,
    model_id: str,
) -> str:
    response = await client.chat.completions.create(
        model=model_id,
        messages=[
            {
                "role": "system",
                "content": (
                    "سند را دقیق و کوتاه خلاصه کن. "
                    "اطلاعاتی را که در متن وجود ندارد "
                    "حدس نزن."
                ),
            },
            {
                "role": "user",
                "content": text,
            },
        ],
    )

    return (
        response.choices[0]
        .message
        .content
        or ""
    )

برای Batch بزرگ باید:

Concurrency محدود شود.
Rate Limit رعایت شود.
هزینه ثبت شود.
Timeout وجود داشته باشد.
Retry چندلایه نشود.
مدل و Prompt نسخه‌بندی شوند.
خروجی Validate شود.
Task Budget تعریف شود.

Budget برای Task

عملیات طولانی ممکن است هزینه زیادی ایجاد کند. برای هر Task سقف تعیین کنید:

{
  "maxModelRequests": 1000,
  "maxInputTokens": 5000000,
  "maxOutputTokens": 500000,
  "maxRuntimeMinutes": 60,
  "maxCost": 25
}

اگر Budget به پایان رسید، Task می‌تواند:

با وضعیت failed متوقف شود.
به input_required برود و افزایش Budget بخواهد.
نتیجه جزئی تولید کند.
ادامه آیتم‌ها را متوقف کند.

انتخاب رفتار باید از ابتدا تعریف شود.

کاربردهای عملی MCP Tasks

CI/CD Pipeline

شروع Pipeline
↓
Build
↓
Test
↓
input_required برای Production
↓
Deploy
↓
completed

پردازش انبوه اسناد

استخراج متن
Chunking
Embedding
Indexing
ساخت گزارش

تولید ویدئو

API تولید ویدئو معمولاً Job ID برمی‌گرداند. MCP Server می‌تواند Job خارجی را به MCP Task نگاشت کند.

External Job ID
↔
MCP taskId

Fine-Tuning

آپلود Dataset
Validation
شروع Job
نمایش وضعیت
دریافت Model ID
ثبت نتیجه نهایی

مهاجرت داده

خواندن Batch
تبدیل Schema
Validation
ذخیره در مقصد
گزارش موارد شکست‌خورده

Human-in-the-Loop

تولید پیش‌نویس
انتظار برای بازبینی
دریافت اصلاحات
ادامه عملیات
انتشار نتیجه

گزارش سازمانی

دریافت اطلاعات از چند منبع
اجرای Query
ترکیب داده
ساخت فایل
بازگرداندن لینک گزارش

نگاشت APIهای Async خارجی

بسیاری از APIها ساختاری مانند زیر دارند:

POST /jobs
→ job_id

GET /jobs/{job_id}
→ status

MCP Tasks به‌صورت طبیعی با این APIها سازگار است.

Task Store:

{
  "taskId": "task_72",
  "externalJobId": "provider_job_921",
  "status": "working"
}

Worker یا Poller وضعیت External Job را بررسی و Task را به‌روزرسانی می‌کند.

Webhook یا Polling خارجی

برای بررسی Job خارجی دو روش وجود دارد.

Polling Provider

Worker هر چند ثانیه وضعیت Provider را می‌پرسد.

Webhook

Provider پس از تغییر وضعیت، Webhook ارسال می‌کند.

Webhook معمولاً کارآمدتر است، اما باید:

Signature بررسی شود.
Replay جلوگیری شود.
Event Idempotent باشد.
ترتیب Eventها مدیریت شود.
Task صحیح پیدا شود.

Observability

برای هر Task این اطلاعات را ثبت کنید:

Task ID
Owner و Tenant
Tool Name
وضعیت
مرحله فعلی
زمان ایجاد
زمان شروع Worker
زمان پایان
Queue Wait
تعداد Retry
Worker ID
Model ID
Token Usage
هزینه
Tool Call Count
External Job ID
Cancellation
Input Requestها
Error Category
Artifactهای خروجی

شاخص‌های مهم:

Task Completion Rate
Failure Rate
Cancellation Rate
Average Queue Time
Average Runtime
P95 Runtime
Input Required Rate
Expired Result Rate
Duplicate Task Rate
Retry Count
Cost per Task

Alertهای مهم

Taskهای طولانی‌تر از حد انتظار
Taskهای گیرکرده در working
Taskهای قدیمی در input_required
Queue Backlog
Worker Crash
افزایش Failure Rate
افزایش هزینه
نتیجه‌های منقضی‌نشده
Taskهای بدون Owner
Retryهای بیش از حد
اختلاف Task Store و External Job

تشخیص Task گیرکرده

هر Worker باید Heartbeat ثبت کند:

{
  "lastHeartbeatAt": "2026-08-17T10:05:00Z"
}

اگر Heartbeat مدت زیادی به‌روزرسانی نشود:

Lease آزاد شود.
Task به Worker دیگری منتقل شود.
Retry انجام شود.
وضعیت failed شود.
Alert ایجاد شود.

نباید Task برای همیشه در working باقی بماند.

معماری پیشنهادی Production

MCP Client
↓
API Gateway
↓
Authentication و Rate Limit
↓
MCP Server
↓
Capability Check
↓
Idempotency Check
↓
ایجاد Durable Task
↓
Queue
↓
Worker Pool
↓
مدل یا سرویس خارجی
↓
Task Store
↓
tasks/get یا Notification
↓
MCP Client

برای ورودی میانه اجرا:

Worker
↓
input_required
↓
Task Store
↓
Client
↓
Elicitation
↓
tasks/update
↓
Worker

ساختار پیشنهادی پروژه

app/
├── mcp/
│   ├── server.py
│   ├── capabilities.py
│   ├── task_methods.py
│   └── tools.py
├── tasks/
│   ├── models.py
│   ├── repository.py
│   ├── service.py
│   ├── transitions.py
│   └── cleanup.py
├── workers/
│   ├── document_worker.py
│   ├── deployment_worker.py
│   └── report_worker.py
├── queue/
│   ├── producer.py
│   └── consumer.py
├── integrations/
│   ├── darvareh.py
│   └── external_jobs.py
├── observability/
│   ├── metrics.py
│   ├── logging.py
│   └── audit.py
├── core/
│   ├── config.py
│   ├── security.py
│   └── exceptions.py
└── main.py

خطاهای رایج

ذخیره Task فقط در Memory

با Restart شدن Server تمام Taskها از بین می‌روند.

بازگرداندن taskId پیش از ذخیره پایدار

Client ممکن است بلافاصله tasks/get را فراخوانی کند و Task پیدا نشود.

نادیده‌گرفتن pollIntervalMs

Polling بیش از حد به Server فشار وارد می‌کند.

Task ID قابل‌حدس

شناسه ترتیبی می‌تواند باعث Enumeration شود.

نبود Authorization روی tasks/get

هر درخواست وضعیت باید مالکیت و دسترسی Caller را بررسی کند.

فرض قطعی‌بودن Cancellation

tasks/cancel فقط درخواست لغو است. وضعیت نهایی باید بررسی شود.

Retry بدون Idempotency

ممکن است Job، Ticket، پرداخت یا Deployment تکراری ایجاد شود.

استفاده از failed برای خطای یک آیتم

در Batch Job، شکست چند آیتم می‌تواند بخشی از نتیجه کامل‌شده باشد.

باقی‌ماندن Task در working

Heartbeat، Lease و Timeout برای شناسایی Taskهای گیرکرده لازم است.

نگهداری دائمی نتایج

Task Store باید TTL و Cleanup داشته باشد.

ارسال جزئیات داخلی خطا

Stack Trace و پیام خام Provider نباید به Client نمایش داده شوند.

ترکیب Task Store و Artifact Store

گزارش یا فایل نهایی بهتر است در Storage مناسب ذخیره شود و Task فقط Reference آن را نگه دارد.

استفاده از Tasks برای عملیات بسیار کوتاه

برای Tool چندمیلی‌ثانیه‌ای، Task فقط پیچیدگی اضافه ایجاد می‌کند.

چک‌لیست انتشار

نسخه جدید Tasks Extension استفاده می‌شود.
شناسه io.modelcontextprotocol/tasks درست اعلام شده است.
Client Capability پیش از ساخت Task بررسی می‌شود.
Client پاسخ مستقیم و Task را مدیریت می‌کند.
Task پیش از ارسال Handle به‌صورت پایدار ذخیره می‌شود.
Task Store با Restart از بین نمی‌رود.
Task ID تصادفی و غیرقابل‌حدس است.
Owner و Tenant برای هر Task ثبت می‌شوند.
Authorization روی تمام Methodهای Task اجرا می‌شود.
State Transitionها اعتبارسنجی می‌شوند.
وضعیت Terminal دوباره تغییر نمی‌کند.
tasks/get Idempotent است.
pollIntervalMs مشخص و رعایت می‌شود.
TTL و Cleanup Policy وجود دارد.
Queue و Worker از Task Store جدا هستند.
Worker Lease و Heartbeat وجود دارد.
Task گیرکرده شناسایی می‌شود.
Idempotency Key برای ساخت Task وجود دارد.
عملیات Write قابل Retry و Idempotent هستند.
Retry فقط برای خطاهای موقت انجام می‌شود.
سقف Retry تعریف شده است.
Cancellation در نقاط مناسب بررسی می‌شود.
Acknowledgement لغو با وضعیت نهایی اشتباه گرفته نمی‌شود.
Partial Failure به‌درستی گزارش می‌شود.
ورودی‌های tasks/update Validate می‌شوند.
inputRequests شناسه منحصربه‌فرد دارند.
اطلاعات حساس در statusMessage نمایش داده نمی‌شوند.
Budget زمانی، Token و هزینه تعریف شده است.
نتیجه Tool با Schema معتبر ذخیره می‌شود.
Artifactهای بزرگ خارج از Task Store نگهداری می‌شوند.
Notification جایگزین منبع اصلی State نشده است.
Metrics و Alert وجود دارند.
نسخه مدل و Prompt ثبت می‌شود.
Taskهای واقعی با قطع اتصال و Restart آزمایش شده‌اند.

پرسش‌های متداول

MCP Tasks چیست؟

افزونه‌ای برای Model Context Protocol است که به Server اجازه می‌دهد عملیات طولانی را به‌صورت Asynchronous اجرا و به‌جای نتیجه فوری، یک taskId پایدار برگرداند.

MCP Tasks برای چه کارهایی مناسب است؟

برای CI/CD، پردازش انبوه، ساخت ویدئو، Fine-Tuning، مهاجرت داده، گزارش‌های طولانی، Jobهای خارجی و Workflowهای نیازمند تأیید انسانی مناسب است.

آیا MCP Tasks بخشی از Core Protocol است؟

در نسخه جدید، Tasks به Extension مستقلی با شناسه io.modelcontextprotocol/tasks منتقل شده است. این Extension درحال‌توسعه است و پشتیبانی Clientها متفاوت است.

تفاوت MCP Tasks و Tool Calling چیست؟

Tool Calling عملیات را آغاز می‌کند. MCP Tasks اجازه می‌دهد نتیجه همان عملیات به‌صورت Deferred و از طریق یک Task Handle دریافت شود.

تفاوت MCP Tasks و Queue چیست؟

Tasks رابط Protocol برای Client است. Queue زیرساخت داخلی اجرای Job در Backend است. این دو معمولاً با یکدیگر استفاده می‌شوند.

تفاوت MCP Tasks و Progress چیست؟

Progress معمولاً به اتصال فعال وابسته است. Task شناسه و State پایدار دارد و پس از قطع اتصال نیز قابل‌بازیابی است.

taskId چیست؟

شناسه منحصربه‌فرد و غیرقابل‌حدس یک Task است که Client با استفاده از آن وضعیت، ورودی موردنیاز یا نتیجه را دریافت می‌کند.

tasks/get چه کاری انجام می‌دهد؟

وضعیت فعلی Task را برمی‌گرداند. اگر Task کامل یا شکست‌خورده باشد، نتیجه یا خطا نیز در پاسخ قرار می‌گیرد.

tasks/update چیست؟

برای ارسال پاسخ به درخواست‌های میانه Workflow استفاده می‌شود؛ برای مثال تأیید کاربر در یک Elicitation.

tasks/cancel چیست؟

درخواست لغو Task را ارسال می‌کند. لغو Cooperative است و ممکن است عملیات پیش از توقف کامل یا شکست بخورد.

آیا Task پس از قطع اتصال ادامه پیدا می‌کند؟

اگر Server و Worker به‌درستی پیاده‌سازی شده باشند، بله. وضعیت Task در Storage پایدار باقی می‌ماند و Client بعداً با همان شناسه ادامه می‌دهد.

آیا Taskها برای همیشه باقی می‌مانند؟

خیر. Server می‌تواند با ttlMs مدت نگهداری State و نتیجه را تعیین کند.

چرا tasks/list وجود ندارد؟

برای کاهش خطر افشای Taskهای Callerهای مختلف، Extension جدید Method عمومی tasks/list ارائه نمی‌کند.

آیا MCP Tasks از Human-in-the-Loop پشتیبانی می‌کند؟

بله. Task می‌تواند به وضعیت input_required برود، درخواست Elicitation ارائه دهد و پس از دریافت پاسخ با tasks/update ادامه پیدا کند.

آیا Task می‌تواند چند Worker داشته باشد؟

در Backend بله، اما باید Lease، Lock و State Transition اتمیک وجود داشته باشد تا یک مرحله به‌صورت تکراری اجرا نشود.

اگر بعضی آیتم‌های Batch شکست بخورند چه می‌شود؟

Task می‌تواند با وضعیت completed یک نتیجه جزئی شامل تعداد موفق و ناموفق برگرداند. وضعیت failed معمولاً برای شکست کل عملیات مناسب‌تر است.

آیا می‌توان MCP Tasks را با API درواره استفاده کرد؟

بله. Worker می‌تواند از API درواره برای فراخوانی مدل‌ها استفاده کند و MCP Tasks وضعیت اجرای طولانی را برای Client مدیریت کند.

جمع‌بندی

Tool Calling معمولی برای عملیات کوتاه مناسب است، اما اجرای وظایف چنددقیقه‌ای یا چندساعته با یک اتصال باز قابل‌اعتماد نیست.

MCP Tasks این مشکل را با یک مدل Call-now, Fetch-later حل می‌کند:

درخواست را آغاز کن
↓
taskId دریافت کن
↓
اتصال را آزاد کن
↓
وضعیت را پیگیری کن
↓
در صورت نیاز ورودی بده
↓
نتیجه را بعداً دریافت کن

اصول اصلی یک پیاده‌سازی مناسب عبارت‌اند از:

Task پیش از بازگرداندن Handle به‌صورت پایدار ایجاد شود.
Client و Server پشتیبانی Extension را اعلام کنند.
Task ID غیرقابل‌حدس باشد.
وضعیت‌ها به‌صورت State Machine مدیریت شوند.
tasks/get برای مشاهده وضعیت استفاده شود.
tasks/update ورودی میانه اجرا را دریافت کند.
tasks/cancel به‌عنوان درخواست Cooperative مدیریت شود.
Polling مقدار پیشنهادی Server را رعایت کند.
Worker از Queue، Lease و Heartbeat استفاده کند.
Retry با Idempotency ترکیب شود.
Taskهای گیرکرده شناسایی شوند.
Result و Error ساختاریافته باشند.
TTL و Cleanup Policy وجود داشته باشد.
Budget زمان، Token و هزینه کنترل شود.
Authorization برای تمام عملیات Task اجرا شود.
Workflow با قطع اتصال و Restart آزمایش شود.

با ترکیب MCP Tasks، Elicitation، Progressive Tool Discovery و Programmatic Tool Calling می‌توان Agentهایی ساخت که علاوه بر دسترسی به ابزارهای متعدد، عملیات طولانی، تعاملی و قابل‌ادامه را نیز به‌شکل قابل‌اعتماد اجرا کنند.

برای استفاده از مدل‌های مختلف در Workerها و AI Agentها می‌توانید از API درواره استفاده کنید. فهرست مدل‌ها، قابلیت‌ها و قیمت جاری آن‌ها در صفحه مدل‌های درواره در دسترس است.

منابع

مقالات مرتبط

برای مطالعه شرایط استفاده و محدودیت‌های مسئولیت، صفحه «سلب مسئولیت» را مشاهده کنید.