Add IMcpTaskExecutor for delegating task execution to an external runtime

docs/concepts/tasks/tasks.md

@@ -282,6 +282,108 @@ public sealed class MyTaskStore : IMcpTaskStore
 }
 ```
 
+### Delegating execution to an external runtime
+
+By default, `WithTasks` executes the tool in-process on the .NET thread pool. To delegate
+execution to a durable system such as Temporal, Orleans, Hangfire, or an external queue,
+register an <xref:ModelContextProtocol.Extensions.Tasks.IMcpTaskExecutor>:
+
+```csharp
+builder.WithTasks(
+    myDurableTaskStore,
+    options =>
+    {
+        options.TaskExecutor = new TemporalTaskExecutor(workflowClient);
+    });
+```
+
+An executor can also be resolved from the service provider — register `IMcpTaskExecutor` in DI
+and omit `TaskExecutor`. The executor is resolved from each task's execution scope, so scoped
+registrations get one instance per task; singleton registrations behave as usual. When neither
+is configured, tasks run in-process exactly as before.
+
+The executor is invoked after the task record is durably created in the store.
+<xref:ModelContextProtocol.Extensions.Tasks.IMcpTaskExecutor.StartAsync*> must return only
+after execution has been durably started — for example, after the external runtime has
+accepted the job — mirroring the durability requirement SEP-2663 §306 places on
+<xref:ModelContextProtocol.Extensions.Tasks.IMcpTaskStore.CreateTaskAsync*>. It must not wait
+for the task to complete. If `StartAsync` throws, the exception is not returned as an error
+from the original `tools/call`: that call still succeeds with
+<xref:ModelContextProtocol.Extensions.Tasks.CreateTaskResult>, the task is marked failed via
+`SetFailedAsync`, and the client discovers the failure on its first `tasks/get` poll. By
+contrast, failures before the task record exists — resolving the executor or
+<xref:ModelContextProtocol.Extensions.Tasks.IMcpTaskStore.CreateTaskAsync*> — do fail the
+original `tools/call`. After a successful `StartAsync`, the SDK stops tracking the task and
+the store is the single source of truth for its state.
+
+One boundary to be aware of is the window between `CreateTaskAsync` completing and
+`StartAsync` returning: if the process exits during it, the store is left with a `Working`
+task whose work was never submitted to the external runtime. The SDK performs no
+reconciliation of such tasks, so integrations that must recover from a crash in this window
+need their own strategy — for example TTL cleanup, startup reconciliation, an outbox, or a
+durable execution intent. Using the task ID as an idempotency key makes resubmission safe,
+but it does not by itself retry a submission that was never attempted.
+
+The <xref:ModelContextProtocol.Extensions.Tasks.McpTaskExecutionContext> passed to the
+executor exposes the task identity, the matched tool request bound to a fresh execution
+scope, and a token that fires on `tasks/cancel`. Executors that want the tool to run
+locally call
+<xref:ModelContextProtocol.Extensions.Tasks.McpTaskExecutionContext.RunToolPipelineAsync*>,
+which runs the remaining request filters and the tool, records the outcome in the store,
+and releases the execution scope. Executors that hand execution off to an external system
+should read what they need from
+<xref:ModelContextProtocol.Extensions.Tasks.McpTaskExecutionContext.Request*> and then call
+<xref:ModelContextProtocol.Extensions.Tasks.McpTaskExecutionContext.DisposeAsync*> to release
+the scope-bound services.
+
+Primitive matching and the filters registered before Tasks — including ASP.NET Core
+authorization — have already run by the time `StartAsync` is called. The remaining
+alternate-result filters and the ordinary call-tool filters run only inside
+`RunToolPipelineAsync`, so an executor that performs a pure handoff to an external runtime
+bypasses them. When the tool pipeline will not run locally, validation, auditing,
+transformations, and other cross-cutting policies must be applied by the external runtime —
+or by a filter registered before Tasks — instead.
+
+```csharp
+public sealed class TemporalTaskExecutor(ITemporalClient workflowClient) : IMcpTaskExecutor
+{
+    public async ValueTask StartAsync(
+        McpTaskExecutionContext context, CancellationToken cancellationToken)
+    {
+        // Submit the tool request to the durable runtime. The workflow communicates with
+        // IMcpTaskStore directly to record progress and results.
+        await workflowClient.StartWorkflowAsync(
+            "run-mcp-task",
+            new McpTaskPayload(context.TaskId, context.Request.Params),
+            id: context.TaskId,
+            cancellationToken);
+
+        // The scope-bound services are no longer needed in this process.
+        await context.DisposeAsync();
+    }
+}
+```
+
+`tasks/get`, `tasks/update`, and `tasks/cancel` continue to be served entirely from the
+`IMcpTaskStore`, so a different server instance can serve polling clients after the process
+that started the task exits — the acceptance scenario for durable execution.
+
+Note that <xref:ModelContextProtocol.Extensions.Tasks.McpTaskExecutionContext.CancellationToken*>
+signals cancellation only within the process that created the task. A `tasks/cancel` handled
+by a different server instance can update the shared store, but it cannot signal that
+process's token. External-runtime integrations whose cancellation must survive server
+replacement therefore need to propagate it through their store or another durable mechanism;
+the token remains the cancellation signal for the default in-process executor and other
+same-process execution paths.
+
+Note that elicitation and sampling issued from *outside* the process that owns the client
+session cannot be routed through the task's input-request channel; an external worker that
+needs multi-round-trip input should rely on the store's
+<xref:ModelContextProtocol.Extensions.Tasks.IMcpTaskStore.InputResponseReceived?displayProperty=nameWithType>
+event, or run the pipeline locally via
+<xref:ModelContextProtocol.Extensions.Tasks.McpTaskExecutionContext.RunToolPipelineAsync*>
+from the process that owns the session.
+
 ### Status semantics
 
 <xref:ModelContextProtocol.Extensions.Tasks.McpTaskStatus.Completed> is the terminal status whenever the

src/ModelContextProtocol.Extensions.Tasks/Server/IMcpTaskExecutor.cs

@@ -0,0 +1,51 @@
+using ModelContextProtocol.Protocol;
+using ModelContextProtocol.Server;
+
+namespace ModelContextProtocol.Extensions.Tasks;
+
+/// <summary>
+/// Executes a task created by the Tasks extension after the task record has been
+/// durably created in the <see cref="IMcpTaskStore"/>.
+/// </summary>
+/// <remarks>
+/// <para>
+/// By default, tasks execute in-process via the .NET thread pool. Registering a custom
+/// executor delegates execution to an external system such as Temporal, Orleans, Hangfire,
+/// or a distributed queue. The executor is invoked once per task, after the task record is
+/// created and the execution context is fully wired.
+/// </para>
+/// <para>
+/// <see cref="StartAsync"/> must return only after execution has been durably started
+/// (e.g., the external runtime has accepted the job), mirroring the durability requirement
+/// SEP-2663 §306 places on <see cref="IMcpTaskStore.CreateTaskAsync"/>. It must not wait
+/// for the task to complete; completion is recorded in the store by whichever system
+/// performs the execution.
+/// </para>
+/// <para>
+/// If <see cref="StartAsync"/> throws, the exception is not returned as an error from the
+/// original <c>tools/call</c>: that call still succeeds with <see cref="CreateTaskResult"/>,
+/// the task is marked failed via <see cref="IMcpTaskStore.SetFailedAsync"/>, and the client
+/// discovers the failure on its first poll. After a successful <see cref="StartAsync"/>,
+/// the SDK no longer tracks the task; the store is the single source of truth for its state.
+/// </para>
+/// <para>
+/// See the <see href="https://github.com/modelcontextprotocol/modelcontextprotocol/blob/main/seps/2663-tasks-extension.md">SEP-2663</see>
+/// specification for details on the tasks extension.
+/// </para>
+/// </remarks>
+public interface IMcpTaskExecutor
+{
+    /// <summary>
+    /// Starts execution of a task.
+    /// </summary>
+    /// <param name="context">
+    /// The execution context for the task, providing the task identity, the matched tool
+    /// request bound to a fresh execution scope, and a helper for running the normal tool
+    /// invocation pipeline locally.
+    /// </param>
+    /// <param name="cancellationToken">
+    /// A token that fires when the task is cancelled via <c>tasks/cancel</c>.
+    /// </param>
+    /// <returns>A <see cref="ValueTask"/> that completes when execution has been durably started.</returns>
+    ValueTask StartAsync(McpTaskExecutionContext context, CancellationToken cancellationToken);
+}

src/ModelContextProtocol.Extensions.Tasks/Server/McpTaskExecutionContext.cs

@@ -0,0 +1,118 @@
+using ModelContextProtocol.Protocol;
+using ModelContextProtocol.Server;
+
+namespace ModelContextProtocol.Extensions.Tasks;
+
+/// <summary>
+/// The execution context handed to an <see cref="IMcpTaskExecutor"/> when a task starts.
+/// </summary>
+/// <remarks>
+/// <para>
+/// The context owns the request-scoped services and the cancellation registration for task
+/// execution. Executors that run the tool locally via <see cref="RunToolPipelineAsync"/>
+/// never dispose anything explicitly; the context releases its resources when the pipeline
+/// completes. Executors that hand execution off to an external system should extract what
+/// they need from <see cref="Request"/> and then call <see cref="DisposeAsync"/> once the
+/// scope-bound services are no longer needed.
+/// </para>
+/// <para>
+/// <see cref="Request"/> carries a server whose outgoing requests (elicitation, sampling)
+/// are intercepted and routed through the task's pending input requests, so responses
+/// submitted via <c>tasks/update</c> are delivered even when a different server instance
+/// serves the polling client.
+/// </para>
+/// <para>
+/// <see cref="RunToolPipelineAsync"/> and <see cref="DisposeAsync"/> coordinate through an
+/// atomic state transition: concurrent callers cannot both win, so the pipeline runs at most
+/// once and disposal cannot race with it. Every caller but the winner observes the context
+/// as disposed.
+/// </para>
+/// </remarks>
+public sealed class McpTaskExecutionContext : IAsyncDisposable
+{
+    private readonly Func<RequestContext<CallToolRequestParams>, CancellationToken, Task> _pipelineRunner;
+    private readonly Func<Task> _disposer;
+    private int _disposed;
+
+    internal McpTaskExecutionContext(
+        McpTaskInfo taskInfo,
+        RequestContext<CallToolRequestParams> request,
+        CancellationToken cancellation,
+        Func<RequestContext<CallToolRequestParams>, CancellationToken, Task> pipelineRunner,
+        Func<Task> disposer)
+    {
+        TaskInfo = taskInfo;
+        Request = request;
+        CancellationToken = cancellation;
+        _pipelineRunner = pipelineRunner;
+        _disposer = disposer;
+    }
+
+    /// <summary>
+    /// Gets the unique identifier of the created task.
+    /// </summary>
+    public string TaskId => TaskInfo.TaskId;
+
+    /// <summary>
+    /// Gets the store record for the created task, with an initial status of
+    /// <see cref="McpTaskStatus.Working"/>.
+    /// </summary>
+    public McpTaskInfo TaskInfo { get; }
+
+    /// <summary>
+    /// Gets the matched tool request, bound to the task's execution scope, with the task
+    /// outgoing-request interceptor already attached.
+    /// </summary>
+    public RequestContext<CallToolRequestParams> Request { get; }
+
+    /// <summary>
+    /// Gets a token that fires when the task is cancelled via <c>tasks/cancel</c>.
+    /// </summary>
+    public CancellationToken CancellationToken { get; }
+
+    /// <summary>
+    /// Runs the normal tool invocation pipeline (the remaining request filters and the tool
+    /// itself), records the outcome in the task store, and releases the context's resources.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// Use this when the executor wants the tool to run locally, preserving the behavior of
+    /// <c>WithTasks</c> without a custom executor. Outcomes — including cancellation, protocol
+    /// errors, and unhandled exceptions — are recorded in the store by this call; it does not
+    /// rethrow them.
+    /// </para>
+    /// <para>
+    /// After calling this method, the executor must not use <see cref="Request"/> again, and
+    /// <see cref="DisposeAsync"/> becomes a no-op.
+    /// </para>
+    /// </remarks>
+    /// <param name="cancellationToken">A token to cancel pipeline execution.</param>
+    public async ValueTask RunToolPipelineAsync(CancellationToken cancellationToken)
+    {
+        if (Interlocked.CompareExchange(ref _disposed, 1, 0) != 0)
+        {
+            throw new ObjectDisposedException(nameof(McpTaskExecutionContext));
+        }
+
+        await _pipelineRunner(Request, cancellationToken).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Releases the context's resources: the request-scoped services of the execution scope
+    /// and the cancellation registration for the task.
+    /// </summary>
+    /// <remarks>
+    /// Executors that hand execution off to an external system call this once they no longer
+    /// need <see cref="Request"/>. It is called automatically when
+    /// <see cref="RunToolPipelineAsync"/> completes; calling it afterwards is a no-op.
+    /// </remarks>
+    public async ValueTask DisposeAsync()
+    {
+        if (Interlocked.CompareExchange(ref _disposed, 1, 0) != 0)
+        {
+            return;
+        }
+
+        await _disposer().ConfigureAwait(false);
+    }
+}

src/ModelContextProtocol.Extensions.Tasks/Server/McpTasksBuilderExtensions.cs

@@ -171,8 +171,17 @@ private async ValueTask<ResultOrAlternate<CallToolResult>> RunAsTaskAsync(
             };
 
             McpTaskInfo taskInfo;
+            IMcpTaskExecutor executor;
             try
             {
+                // Resolve the executor from the execution scope (falling back to the process-local
+                // default) rather than the root provider so scoped and transient registrations get
+                // correct lifetimes; singleton registrations still yield the same instance. Resolving
+                // before the task record is created keeps a DI misconfiguration from leaving a
+                // durably-created task stuck at Working: the resolution error fails tools/call instead.
+                executor = _taskOptions.TaskExecutor
+                    ?? executionScope.ServiceProvider.GetService<IMcpTaskExecutor>()
+                    ?? ProcessLocalMcpTaskExecutor.Instance;
                 taskInfo = await _store.CreateTaskAsync(cancellationToken).ConfigureAwait(false);
             }
             catch
@@ -189,9 +198,24 @@ private async ValueTask<ResultOrAlternate<CallToolResult>> RunAsTaskAsync(
             // Capture the token before dispatching. Cancellation can remove and dispose the source
             // before the background delegate starts.
             var taskCancellationToken = cts.Token;
-            _ = Task.Run(
-                () => ExecuteTaskAsync(next, executionRequest, taskId, taskCancellationToken, executionScope),
-                CancellationToken.None);
+            var context = new McpTaskExecutionContext(
+                taskInfo,
+                executionRequest,
+                taskCancellationToken,
+                (req, ct) => ExecuteTaskAsync(next, req, taskId, ct, executionScope),
+                () => ReleaseExecutionResourcesAsync(executionScope, taskId));
+
+            try
+            {
+                await executor.StartAsync(context, taskCancellationToken).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                // The task record exists, so the client will poll it. Record the start failure as
+                // the task's failure before returning the task alternate, so the client's first
+                // poll observes the terminal state rather than racing it.
+                await RecordStartFailureAsync(context, ex).ConfigureAwait(false);
+            }
 
             return ResultOrAlternate<CallToolResult>.FromAlternate(
                 ToCreateTaskResult(taskInfo),
@@ -235,10 +259,52 @@ private async Task ExecuteTaskAsync(
             }
             finally
             {
-                if (_cancellationSources.TryRemove(taskId, out var registeredCts))
-                {
-                    registeredCts.Dispose();
-                }
+                RemoveCancellationSource(taskId);
+            }
+        }
+
+        private async Task RecordStartFailureAsync(McpTaskExecutionContext context, Exception exception)
+        {
+            _logger.LogError(exception, "Starting execution of task '{TaskId}' failed.", context.TaskId);
+
+            try
+            {
+                await context.DisposeAsync().ConfigureAwait(false);
+            }
+            catch (Exception disposeEx)
+            {
+                _logger.LogError(disposeEx, "Failed to release resources of task '{TaskId}' after a failed start.", context.TaskId);
+            }
+
+            try
+            {
+                var error = new JsonRpcErrorDetail { Code = (int)McpErrorCode.InternalError, Message = exception.Message };
+                var errorJson = JsonSerializer.SerializeToElement(error, McpJsonUtilities.DefaultOptions.GetTypeInfo<JsonRpcErrorDetail>());
+                await _store.SetFailedAsync(context.TaskId, errorJson).ConfigureAwait(false);
+            }
+            catch (Exception storeEx)
+            {
+                _logger.LogError(storeEx, "Failed to record the failure of task '{TaskId}'.", context.TaskId);
+            }
+        }
+
+        private async Task ReleaseExecutionResourcesAsync(AsyncServiceScope executionScope, string taskId)
+        {
+            try
+            {
+                await executionScope.DisposeAsync().ConfigureAwait(false);
+            }
+            finally
+            {
+                RemoveCancellationSource(taskId);
+            }
+        }
+
+        private void RemoveCancellationSource(string taskId)
+        {
+            if (_cancellationSources.TryRemove(taskId, out var registeredCts))
+            {
+                registeredCts.Dispose();
             }
         }
 

src/ModelContextProtocol.Extensions.Tasks/Server/McpTasksOptions.cs

@@ -19,4 +19,15 @@ public sealed class McpTasksOptions
     /// </remarks>
     public Func<RequestContext<CallToolRequestParams>, McpTaskExecutionMode> ExecutionModeSelector { get; set; } =
         static _ => McpTaskExecutionMode.Optional;
+
+    /// <summary>
+    /// Gets or sets the executor that starts task execution.
+    /// </summary>
+    /// <remarks>
+    /// When <see langword="null"/> (the default), the extension resolves a single registered
+    /// <see cref="IMcpTaskExecutor"/> from the service provider, if one exists. If neither is
+    /// present, tasks execute in-process on the .NET thread pool, preserving the behavior of
+    /// <c>WithTasks</c> without a custom executor.
+    /// </remarks>
+    public IMcpTaskExecutor? TaskExecutor { get; set; }
 }