ContextQMD
Back to Python Sdk

Tasks

docs/experimental/tasks.md

1.27.06.6 KB
Original Source

Tasks

!!! warning "Experimental"

Tasks are an experimental feature tracking the draft MCP specification.
The API may change without notice.

Tasks enable asynchronous request handling in MCP. Instead of blocking until an operation completes, the receiver creates a task, returns immediately, and the requestor polls for the result.

When to Use Tasks

Tasks are designed for operations that:

  • Take significant time (seconds to minutes)
  • Need progress updates during execution
  • Require user input mid-execution (elicitation, sampling)
  • Should run without blocking the requestor

Common use cases:

  • Long-running data processing
  • Multi-step workflows with user confirmation
  • LLM-powered operations requiring sampling
  • OAuth flows requiring user browser interaction

Task Lifecycle

text
                    ┌─────────────┐
                    │   working   │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
     ┌────────────┐  ┌───────────┐  ┌───────────┐
     │ completed  │  │  failed   │  │ cancelled │
     └────────────┘  └───────────┘  └───────────┘
              ▲
              │
     ┌────────┴────────┐
     │ input_required  │◄──────┐
     └────────┬────────┘       │
              │                │
              └────────────────┘
StatusDescription
workingTask is being processed
input_requiredReceiver needs input from requestor (elicitation/sampling)
completedTask finished successfully
failedTask encountered an error
cancelledTask was cancelled by requestor

Terminal states (completed, failed, cancelled) are final—tasks cannot transition out of them.

Bidirectional Flow

Tasks work in both directions:

Client → Server (most common):

text
Client                              Server
  │                                    │
  │── tools/call (task) ──────────────>│ Creates task
  │<── CreateTaskResult ───────────────│
  │                                    │
  │── tasks/get ──────────────────────>│
  │<── status: working ────────────────│
  │                                    │ ... work continues ...
  │── tasks/get ──────────────────────>│
  │<── status: completed ──────────────│
  │                                    │
  │── tasks/result ───────────────────>│
  │<── CallToolResult ─────────────────│

Server → Client (for elicitation/sampling):

text
Server                              Client
  │                                    │
  │── elicitation/create (task) ──────>│ Creates task
  │<── CreateTaskResult ───────────────│
  │                                    │
  │── tasks/get ──────────────────────>│
  │<── status: working ────────────────│
  │                                    │ ... user interaction ...
  │── tasks/get ──────────────────────>│
  │<── status: completed ──────────────│
  │                                    │
  │── tasks/result ───────────────────>│
  │<── ElicitResult ───────────────────│

Key Concepts

Task Metadata

When augmenting a request with task execution, include TaskMetadata:

python
from mcp.types import TaskMetadata

task = TaskMetadata(ttl=60000)  # TTL in milliseconds

The ttl (time-to-live) specifies how long the task and result are retained after completion.

Task Store

Servers persist task state in a TaskStore. The SDK provides InMemoryTaskStore for development:

python
from mcp.shared.experimental.tasks import InMemoryTaskStore

store = InMemoryTaskStore()

For production, implement TaskStore with a database or distributed cache.

Capabilities

Both servers and clients declare task support through capabilities:

Server capabilities:

  • tasks.requests.tools.call - Server accepts task-augmented tool calls

Client capabilities:

  • tasks.requests.sampling.createMessage - Client accepts task-augmented sampling
  • tasks.requests.elicitation.create - Client accepts task-augmented elicitation

The SDK manages these automatically when you enable task support.

Quick Example

Server (simplified API):

python
from mcp.server import Server
from mcp.server.experimental.task_context import ServerTaskContext
from mcp.types import CallToolResult, TextContent, TASK_REQUIRED

server = Server("my-server")
server.experimental.enable_tasks()  # One-line setup

@server.call_tool()
async def handle_tool(name: str, arguments: dict):
    ctx = server.request_context
    ctx.experimental.validate_task_mode(TASK_REQUIRED)

    async def work(task: ServerTaskContext):
        await task.update_status("Processing...")
        # ... do work ...
        return CallToolResult(content=[TextContent(type="text", text="Done!")])

    return await ctx.experimental.run_task(work)

Client:

python
from mcp.client.session import ClientSession
from mcp.types import CallToolResult

async with ClientSession(read, write) as session:
    await session.initialize()

    # Call tool as task
    result = await session.experimental.call_tool_as_task("my_tool", {"arg": "value"})
    task_id = result.task.taskId

    # Poll until done
    async for status in session.experimental.poll_task(task_id):
        print(f"Status: {status.status}")

    # Get result
    final = await session.experimental.get_task_result(task_id, CallToolResult)

Next Steps