feat: implement support for SEP-1686 Tasks

[LucaButBoring]

Implements SEP-1686 (Tasks), from the latest specification release.

Motivation and Context

Tasks address several protocol gaps:

1. Avoiding tool splitting anti-patterns - No need to create separate start_tool, get_status, get_result tools; a single task-aware tool handles the full lifecycle
2. Decoupling execution from result retrieval - Clients can dispatch operations immediately and collect results later, without blocking or re-executing if responses are dropped
3. Moving orchestration to host applications - The polling logic lives in the host application/SDK, not in agents that might hallucinate or fail to follow polling contracts

Usage

Server: Defining a Task-Aware Tool

TaskAwareAsyncToolSpecification.builder()
    .name("long-operation")
    .description("A long-running operation")
    .inputSchema(schema)
    // TaskSupportMode.REQUIRED is the default for task-aware tools
    .createTaskHandler((args, extra) -> {
        return extra.createTask()
            .flatMap(task -> {
                startBackgroundWork(task.taskId(), args);
                return Mono.just(CreateTaskResult.builder().task(task).build());
            });
    })
    .build()

Server: Using TaskContext for Lifecycle Management

TaskContext ctx = extra.createTaskContext(task);
ctx.updateStatus("Processing batch 1 of 10");
ctx.isCancelled().flatMap(cancelled -> {
    if (cancelled) return ctx.fail("Cancelled by user");
    return doWork().then(ctx.complete(result));
});

TaskSupportMode options:

• REQUIRED (default): Must have task metadata; returns error otherwise
• OPTIONAL: Works with or without task metadata; auto-polling shim provides backward compatibility
• FORBIDDEN: No task support (regular tools)

Client: Streaming API (Recommended)

Drop-in replacement for callTool that handles polling automatically:

client.callToolStream(request)
    .subscribe(msg -> switch(msg) {
        case TaskCreatedMessage<CallToolResult> tc -> log.info("Task: {}", tc.task().taskId());
        case TaskStatusMessage<CallToolResult> ts -> log.info("Status: {}", ts.task().status());
        case ResultMessage<CallToolResult> r -> handleResult(r.result());
        case ErrorMessage<CallToolResult> e -> handleError(e.error());
    });

Client: Task-Based API (For Explicit Control)

For consumers who need custom polling behavior, cancellation logic, or batched task management:

// Create task and get immediate response with task ID
CreateTaskResult created = client.callToolTask(request);
String taskId = created.task().taskId();

// Poll for status updates (with custom logic)
GetTaskResult status = client.getTask(taskId);
while (!status.isTerminal()) {
    Thread.sleep(status.pollInterval());
    status = client.getTask(taskId);
}

// Retrieve final result
CallToolResult result = client.getTaskResult(taskId, new TypeRef<>() {});

// Or cancel if needed
client.cancelTask(taskId);

Similar patterns exist for sampling (createMessageStream / createMessageTask) and elicitation (createElicitationStream / createElicitationTask).

Server: Bidirectional Task Flows

Servers can send task-augmented requests to clients, assuming the client has configured its own TaskStore:

// Server requesting sampling from client with task tracking
exchange.createMessageStream(new CreateMessageRequest(..., taskMetadata, null))
    .subscribe(msg -> ...);

// Server requesting elicitation from client with task tracking
exchange.createElicitationStream(new ElicitRequest(..., taskMetadata, null))
    .subscribe(msg -> ...);

Key Design Decisions

1. Experimental namespace - All task APIs are in io.modelcontextprotocol.experimental.tasks, signaling that the API may change (matches TypeScript/Python SDKs)

2. TaskStore abstraction - Interface for pluggable storage; InMemoryTaskStore provided for development and testing. The originating request (e.g., CallToolRequest) is stored alongside the task, so tool routing can be derived from stored context rather than maintained as separate mapping state.

3. Auto-polling shim - OPTIONAL mode tools work transparently for non-task-aware clients

4. Defense-in-depth session isolation - Session ID required on all TaskStore operations; enforced at both server and storage layers to prevent cross-session task access

   • Passing null for sessionId bypasses validation (single-tenant mode). This is used by McpAsyncClient since clients are inherently single-tenant - there's only one session, so cross-session isolation doesn't apply.

How Has This Been Tested?

• Unit tests
• Client-server integration tests for the typical flows
• Spot-checking compatibility with the TS SDK

Breaking Changes

None

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the MCP Documentation
• My code follows the repository's style guidelines
• New and existing tests pass locally
• I have added appropriate error handling
• I have added or updated documentation as needed

Additional context

Closes #668

This PR also includes a tweak to how 202 Accepted is handled by the client implementation, which was done to handle how the TypeScript server SDK configures its response headers when accepting JSON-RPC responses and notifications from the client - in particular, sending InitializeNotification produced an exception in the Java SDK client before this, which made testing compatibility with the existing TS SDK's Tasks implementation rather difficult.

[He-Pin]

Nice to have this in, the api seems quite the same as typescript sdk