MCP Tasks API

Durable task management for long-running operations per MCP 2025-11-25

MCP Tasks API

The Tasks API provides durable state machines for tracking long-running operations. Tasks allow clients to poll for status updates and retrieve results when operations complete.

Protocol Version

This feature implements MCP Protocol Version 2025-11-25.

Overview

Tasks are durable state machines that:

• Track the execution state of long-running requests
• Support polling for status updates
• Enable deferred result retrieval
• Provide status notifications

Task Status Values

Status	Description
pending	Task created but not yet started
working	Task is actively being processed
input_required	Task needs user input (elicitation)
completed	Task finished successfully
failed	Task failed with error
cancelled	Task was cancelled

Python API

Basic Usage

import asyncio
from praisonai.mcp_server.tasks import TaskManager, TaskStatus

async def execute_operation(method: str, params: dict) -> str:
    """Your async operation handler."""
    await asyncio.sleep(1)  # Simulate work
    return f"Result for {method}"

async def main():
    # Create task manager with executor
    manager = TaskManager(executor=execute_operation)
    
    # Create a task
    task = await manager.create_task(
        method="tools/call",
        params={"name": "search", "arguments": {"query": "AI"}},
        metadata={"user": "demo"},
        execute=True,  # Start execution immediately
    )
    
    print(f"Task ID: {task.id}")
    print(f"Status: {task.status.value}")
    
    # Poll for completion
    while True:
        current = manager.get_task(task.id)
        if current.status in (TaskStatus.COMPLETED, TaskStatus.FAILED):
            break
        await asyncio.sleep(1)
    
    # Get result
    final = manager.get_task(task.id)
    print(f"Result: {final.result}")

asyncio.run(main())

Task Response Format

Tasks return MCP-compliant responses:

task = await manager.create_task(method="tools/call", params={})
response = task.to_dict()

# Response format:
{
    "taskId": "task-abc123",
    "status": "working",
    "statusMessage": "The operation is now in progress.",
    "createdAt": "2025-11-25T10:30:00Z",
    "lastUpdatedAt": "2025-11-25T10:40:00Z",
    "ttl": 60000,
    "pollInterval": 5000,
    "_meta": {"user": "demo"}
}

Cancel a Task

cancelled = await manager.cancel_task(task.id)
print(f"Status: {cancelled.status.value}")  # "cancelled"
print(f"Message: {cancelled.status_message}")  # "The task was cancelled by request."

List Tasks

# List all tasks
tasks = manager.list_tasks()

# Filter by status
working_tasks = manager.list_tasks(status=TaskStatus.WORKING)

# Filter by session
session_tasks = manager.list_tasks(session_id="session-123")

CLI Usage

List Tasks

praisonai mcp tasks list
praisonai mcp tasks list --status working
praisonai mcp tasks list --json

Get Task Status

praisonai mcp tasks get <task-id>
praisonai mcp tasks get <task-id> --json

Cancel Task

praisonai mcp tasks cancel <task-id>

MCP Protocol Messages

Creating Tasks

Request with task field:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "search",
    "arguments": {"query": "AI"},
    "task": {"ttl": 60000}
  }
}

Response (CreateTaskResult):

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "task": {
      "taskId": "786512e2-9e0d-44bd-8f29-789f320fe840",
      "status": "working",
      "statusMessage": "The operation is now in progress.",
      "createdAt": "2025-11-25T10:30:00Z",
      "lastUpdatedAt": "2025-11-25T10:40:00Z",
      "ttl": 60000,
      "pollInterval": 5000
    }
  }
}

Getting Task Status

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tasks/get",
  "params": {"taskId": "786512e2-9e0d-44bd-8f29-789f320fe840"}
}

Getting Task Result

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "tasks/result",
  "params": {"taskId": "786512e2-9e0d-44bd-8f29-789f320fe840"}
}

Cancelling Tasks

{
  "jsonrpc": "2.0",
  "id": 4,
  "method": "tasks/cancel",
  "params": {"taskId": "786512e2-9e0d-44bd-8f29-789f320fe840"}
}

Best Practices

1. Respect pollInterval - Use the server-provided poll interval
2. Handle all statuses - Check for completed, failed, and cancelled
3. Use TTL wisely - Set appropriate TTL for your use case
4. Monitor input_required - Handle elicitation when needed

Related

• MCP Elicitation - Request user input during tasks
• MCP Sampling - LLM completions during tasks
• PraisonAI MCP Server - Full server documentation