Default draft protocol support: sessionless + handshake-less (SEP-2575 + SEP-2567) (#1610)

src/ModelContextProtocol.Core/Server/ISseEventStreamReader.cs

@@ -6,6 +6,7 @@ namespace ModelContextProtocol.Server;
 /// <summary>
 /// Provides read access to an SSE event stream, allowing events to be consumed asynchronously.
 /// </summary>
+[Obsolete(Obsoletions.LegacyStatefulHttp_Message, DiagnosticId = Obsoletions.LegacyStatefulHttp_DiagnosticId, UrlFormat = Obsoletions.LegacyStatefulHttp_Url)]
 public interface ISseEventStreamReader
 {
     /// <summary>

src/ModelContextProtocol.Core/Server/ISseEventStreamStore.cs

@@ -3,6 +3,7 @@ namespace ModelContextProtocol.Server;
 /// <summary>
 /// Provides storage and retrieval of SSE event streams, enabling resumability and redelivery of events.
 /// </summary>
+[Obsolete(Obsoletions.LegacyStatefulHttp_Message, DiagnosticId = Obsoletions.LegacyStatefulHttp_DiagnosticId, UrlFormat = Obsoletions.LegacyStatefulHttp_Url)]
 public interface ISseEventStreamStore
 {
     /// <summary>

src/ModelContextProtocol.Core/Server/ISseEventStreamWriter.cs

@@ -6,6 +6,7 @@ namespace ModelContextProtocol.Server;
 /// <summary>
 /// Provides write access to an SSE event stream, allowing events to be written and tracked with unique IDs.
 /// </summary>
+[Obsolete(Obsoletions.LegacyStatefulHttp_Message, DiagnosticId = Obsoletions.LegacyStatefulHttp_DiagnosticId, UrlFormat = Obsoletions.LegacyStatefulHttp_Url)]
 public interface ISseEventStreamWriter : IAsyncDisposable
 {
     /// <summary>

src/ModelContextProtocol.Core/Server/McpServerImpl.cs

@@ -87,11 +87,13 @@ internal sealed partial class McpServerImpl : McpServer
         // Configure all request handlers based on the supplied options.
         ServerCapabilities = new();
         ConfigureInitialize(options);
+        ConfigureDiscover(options);
         ConfigureTools(options);
         ConfigurePrompts(options);
         ConfigureResources(options);
         ConfigureLogging(options);
         ConfigureCompletion(options);
+        ConfigureSubscriptions(options);
         ConfigureExperimentalAndExtensions(options);
         ConfigureTasks(options);
         ConfigureMrtr();
@@ -102,20 +104,10 @@ internal sealed partial class McpServerImpl : McpServer
             _notificationHandlers.RegisterRange(notificationHandlers);
         }
 
-        // In stateless mode, the server cannot send unsolicited notifications,
-        // so listChanged should not be advertised.
-        if (transport is StreamableHttpServerTransport { Stateless: true })
-        {
-            if (ServerCapabilities.Tools is not null)
-                ServerCapabilities.Tools.ListChanged = null;
-            if (ServerCapabilities.Prompts is not null)
-                ServerCapabilities.Prompts.ListChanged = null;
-            if (ServerCapabilities.Resources is not null)
-                ServerCapabilities.Resources.ListChanged = null;
-        }
-
-        // Now that everything has been configured, subscribe to any necessary notifications.
-        if (transport is not StreamableHttpServerTransport streamableHttpTransport || streamableHttpTransport.Stateless is false)
+        // A stateful session can push unsolicited list-changed notifications, so subscribe to the
+        // collection change events. A stateless HTTP server cannot send unsolicited notifications, so
+        // instead suppress the listChanged capability it would otherwise advertise.
+        if (IsStatefulSession())
         {
             Register(ServerOptions.ToolCollection, NotificationMethods.ToolListChangedNotification);
             Register(ServerOptions.PromptCollection, NotificationMethods.PromptListChangedNotification);
@@ -126,16 +118,27 @@ internal sealed partial class McpServerImpl : McpServer
             {
                 if (collection is not null)
                 {
-                    EventHandler changed = (sender, e) => _ = this.SendNotificationAsync(notificationMethod);
+                    EventHandler changed = (sender, e) => _ = SendListChangedNotificationAsync(notificationMethod);
                     collection.Changed += changed;
                     _disposables.Add(() => collection.Changed -= changed);
                 }
             }
         }
+        else
+        {
+            if (ServerCapabilities.Tools is not null)
+                ServerCapabilities.Tools.ListChanged = null;
+            if (ServerCapabilities.Prompts is not null)
+                ServerCapabilities.Prompts.ListChanged = null;
+            if (ServerCapabilities.Resources is not null)
+                ServerCapabilities.Resources.ListChanged = null;
+        }
 
-        // And initialize the session.
-        var incomingMessageFilter = BuildMessageFilterPipeline(options.Filters.Message.IncomingFilters);
+        // And initialize the session. The built-in draft state-sync filter runs ahead of any
+        // user-supplied incoming filters; see PrependDraftStateSyncFilter for what it records and why.
+        var incomingMessageFilter = PrependDraftStateSyncFilter(BuildMessageFilterPipeline(options.Filters.Message.IncomingFilters));
         var outgoingMessageFilter = BuildMessageFilterPipeline(options.Filters.Message.OutgoingFilters);
+
         _sessionHandler = new McpSessionHandler(
             isServer: true,
             _sessionTransport,
@@ -147,12 +150,106 @@ internal sealed partial class McpServerImpl : McpServer
             _logger);
     }
 
+    /// <summary>
+    /// Wraps <paramref name="inner"/> so that, for every JSON-RPC request, a built-in filter first
+    /// synchronizes server-side state (<see cref="_negotiatedProtocolVersion"/>,
+    /// <see cref="_clientCapabilities"/>, <see cref="_clientInfo"/>) from the per-request <c>_meta</c>
+    /// values projected onto <see cref="JsonRpcMessageContext"/> and validates the per-request protocol
+    /// version, before delegating to the user-supplied incoming filters.
+    /// </summary>
+    /// <remarks>
+    /// Under the draft protocol revision (SEP-2575) there is no <c>initialize</c> handshake, so these values
+    /// MUST be populated per-request. For legacy clients the per-request values are absent and the built-in
+    /// filter is a no-op (the values were captured during the initialize handler).
+    /// </remarks>
+    private JsonRpcMessageFilter PrependDraftStateSyncFilter(JsonRpcMessageFilter inner)
+    {
+        JsonRpcMessageFilter draftStateSync = next => async (message, cancellationToken) =>
+        {
+            if (message is JsonRpcRequest { Method: not RequestMethods.Initialize } request && request.Context is { } context)
+            {
+                bool endpointNameNeedsRefresh = false;
+
+                if (context.ProtocolVersion is { } protocolVersion)
+                {
+                    // Per SEP-2575, the server MUST reject any request whose per-request
+                    // _meta/io.modelcontextprotocol/protocolVersion is not one of its supported versions
+                    // with an UnsupportedProtocolVersionError (-32004) carrying the supported list.
+                    if (!McpSessionHandler.SupportedProtocolVersions.Contains(protocolVersion))
+                    {
+                        throw new UnsupportedProtocolVersionException(
+                            requested: protocolVersion,
+                            supported: McpSessionHandler.SupportedProtocolVersions);
+                    }
+
+                    SetNegotiatedProtocolVersion(protocolVersion);
+                }
+
+                if (context.ClientCapabilities is { } clientCapabilities && IsDraftProtocol() && IsStatefulSession())
+                {
+                    // Under the draft revision the per-request _meta envelope carries the client's FULL
+                    // capabilities (SEP-2575), so a plain overwrite is correct. The IsDraftProtocol() gate
+                    // makes any legacy per-request envelope a no-op (legacy capabilities stay as the
+                    // initialize handshake established them); the IsStatefulSession() gate keeps
+                    // _clientCapabilities null under StreamableHttpServerTransport { Stateless = true }
+                    // (where the same server instance handles every request, so persisting per-request
+                    // capability state would both leak across requests and break the StatelessServerTests
+                    // invariant that surfaces the "X is not supported in stateless mode" errors).
+                    _clientCapabilities = clientCapabilities;
+                }
+
+                if (context.ClientInfo is { } clientInfo &&
+                    (_clientInfo is null || !string.Equals(_clientInfo.Name, clientInfo.Name, StringComparison.Ordinal) ||
+                     !string.Equals(_clientInfo.Version, clientInfo.Version, StringComparison.Ordinal)))
+                {
+                    _clientInfo = clientInfo;
+                    endpointNameNeedsRefresh = true;
+                }
+
+                if (endpointNameNeedsRefresh)
+                {
+                    UpdateEndpointNameWithClientInfo();
+                    _sessionHandler.EndpointName = _endpointName;
+                }
+            }
+
+            await next(message, cancellationToken).ConfigureAwait(false);
+        };
+
+        return next => draftStateSync(inner(next));
+    }
+
     /// <inheritdoc/>
     public override string? SessionId => _sessionTransport.SessionId;
 
     /// <inheritdoc/>
     public override string? NegotiatedProtocolVersion => _negotiatedProtocolVersion;
 
+    /// <summary>
+    /// Records the negotiated MCP protocol version for the session. The version is established exactly
+    /// once: the initial <see langword="null"/>-to-value transition is allowed (and racing requests that
+    /// select the same version are idempotent no-ops), but any later attempt to switch to a different
+    /// version throws. A single session MUST NOT change protocol versions, so a conflicting per-request
+    /// <c>_meta</c> protocol version (or <c>Mcp-Protocol-Version</c> header) is a client error rather than
+    /// something we silently overwrite.
+    /// </summary>
+    private void SetNegotiatedProtocolVersion(string protocolVersion)
+    {
+        string? previous = Interlocked.CompareExchange(ref _negotiatedProtocolVersion, protocolVersion, null);
+        if (previous is null)
+        {
+            // We won the initial null-to-value transition; publish it to the session handler for telemetry.
+            _sessionHandler.NegotiatedProtocolVersion = protocolVersion;
+        }
+        else if (!string.Equals(previous, protocolVersion, StringComparison.Ordinal))
+        {
+            throw new McpProtocolException(
+                $"The negotiated protocol version cannot change within a session. " +
+                $"The session negotiated '{previous}', but a request specified '{protocolVersion}'.",
+                McpErrorCode.InvalidRequest);
+        }
+    }
+
     /// <inheritdoc/>
     public ServerCapabilities ServerCapabilities { get; }
 
@@ -273,9 +370,12 @@ internal sealed partial class McpServerImpl : McpServer
                     clientProtocolVersion :
                     McpSessionHandler.LatestProtocolVersion;
 
+                // The legacy initialize handshake is authoritative: it may supersede a protocol version
+                // a prior draft server/discover probe established on the same connection (the dual-era
+                // fallback path a permissive client takes against an unknown server). Unlike the
+                // per-request draft version - which SetNegotiatedProtocolVersion locks once negotiated -
+                // initialize force-sets the version.
                 _negotiatedProtocolVersion = protocolVersion;
-
-                // Update session handler with the negotiated protocol version for telemetry
                 _sessionHandler.NegotiatedProtocolVersion = protocolVersion;
 
                 return new InitializeResult
@@ -290,6 +390,228 @@ internal sealed partial class McpServerImpl : McpServer
             McpJsonUtilities.JsonContext.Default.InitializeResult);
     }
 
+    /// <summary>
+    /// Registers the <c>server/discover</c> request handler introduced by the draft protocol revision (SEP-2575).
+    /// </summary>
+    /// <remarks>
+    /// The handler is registered unconditionally so legacy clients can probe it too. It returns the server's
+    /// supported protocol versions (<see cref="McpSessionHandler.SupportedProtocolVersions"/>), server
+    /// capabilities, server info, and optional instructions.
+    /// </remarks>
+    private void ConfigureDiscover(McpServerOptions options)
+    {
+        _requestHandlers.Set(RequestMethods.ServerDiscover,
+            (request, _, _) =>
+            {
+                return new ValueTask<DiscoverResult>(new DiscoverResult
+                {
+                    SupportedVersions = [.. McpSessionHandler.SupportedProtocolVersions],
+                    Capabilities = ServerCapabilities ?? new(),
+                    ServerInfo = options.ServerInfo ?? DefaultImplementation,
+                    Instructions = options.ServerInstructions,
+                    // Spec PR #2855 makes ttlMs and cacheScope required on DiscoverResult. Default to
+                    // the safest values (immediately stale, not shareable) so existing servers keep
+                    // their "do not cache" behavior while satisfying the wire requirement.
+                    TimeToLive = TimeSpan.Zero,
+                    CacheScope = CacheScope.Private,
+                });
+            },
+            McpJsonUtilities.JsonContext.Default.DiscoverRequestParams,
+            McpJsonUtilities.JsonContext.Default.DiscoverResult);
+    }
+
+    /// <summary>
+    /// Registers the <c>subscriptions/listen</c> request handler introduced by the draft protocol revision (SEP-2575).
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The handler opens a long-lived response stream (over the per-request <see cref="StreamableHttpPostTransport"/>
+    /// for HTTP, or the shared STDIO channel) that first sends
+    /// <see cref="NotificationMethods.SubscriptionsAcknowledgedNotification"/> reporting which subscriptions the
+    /// server agreed to honor, and then streams matching notifications until the request is cancelled.
+    /// </para>
+    /// <para>
+    /// Subscription-bound notifications carry the listen request's id in their
+    /// <c>_meta/io.modelcontextprotocol/subscriptionId</c> field per SEP-2575 so clients can demultiplex.
+    /// </para>
+    /// </remarks>
+    private void ConfigureSubscriptions(McpServerOptions options)
+    {
+        _requestHandlers.Set(RequestMethods.SubscriptionsListen,
+            async (request, jsonRpcRequest, cancellationToken) =>
+            {
+                var requested = request?.Notifications ?? new SubscriptionsListenNotifications();
+
+                // A stateless session (Streamable HTTP with no session) cannot deliver out-of-band
+                // notifications: each request is isolated and nothing outlives it to push later list/resource
+                // changes back to the client (tracked by #1662). Rather than hold the POST open forever only
+                // to deliver nothing - pinning the connection and its request scope - acknowledge the listen
+                // request granting no notifications and complete immediately. This runs after protocol
+                // negotiation, so it is not a legacy-server signal and never triggers a client fallback to the
+                // initialize handshake.
+                if (!IsStatefulSession())
+                {
+                    var statelessSubscription = new ActiveSubscription(
+                        jsonRpcRequest.Id,
+                        new SubscriptionsListenNotifications(),
+                        jsonRpcRequest.Context?.RelatedTransport);
+
+                    await SendSubscriptionAckAsync(statelessSubscription, cancellationToken).ConfigureAwait(false);
+
+                    return new EmptyResult();
+                }
+
+                // Filter the requested notifications against what the server actually supports.
+                var granted = new SubscriptionsListenNotifications
+                {
+                    ToolsListChanged = requested.ToolsListChanged == true && ServerCapabilities?.Tools?.ListChanged == true ? true : null,
+                    PromptsListChanged = requested.PromptsListChanged == true && ServerCapabilities?.Prompts?.ListChanged == true ? true : null,
+                    ResourcesListChanged = requested.ResourcesListChanged == true && ServerCapabilities?.Resources?.ListChanged == true ? true : null,
+                    ResourceSubscriptions = requested.ResourceSubscriptions is { Count: > 0 } subs && ServerCapabilities?.Resources?.Subscribe == true
+                        ? new List<string>(subs)
+                        : null,
+                };
+
+                // Track this subscription so list-changed notifications can be fanned out to it, tagged with
+                // the right subscriptionId, and routed back over the stream this request opened.
+                var subscription = new ActiveSubscription(
+                    jsonRpcRequest.Id,
+                    granted,
+                    jsonRpcRequest.Context?.RelatedTransport);
+                _activeSubscriptions[jsonRpcRequest.Id] = subscription;
+
+                try
+                {
+                    // Send the acknowledgement notification first, as required by SEP-2575. Like every other
+                    // notification delivered on the subscription it is routed back over this request's own
+                    // stream and tagged with the subscription id so shared-channel clients can demultiplex it.
+                    await SendSubscriptionAckAsync(subscription, cancellationToken).ConfigureAwait(false);
+
+                    // Keep the subscription open until the request is cancelled (client disconnect on HTTP,
+                    // or notifications/cancelled on STDIO).
+                    var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                    using var registration = cancellationToken.Register(static state => ((TaskCompletionSource<bool>)state!).TrySetResult(true), tcs);
+                    await tcs.Task.ConfigureAwait(false);
+                }
+                finally
+                {
+                    _activeSubscriptions.TryRemove(jsonRpcRequest.Id, out _);
+                }
+
+                return new EmptyResult();
+            },
+            McpJsonUtilities.JsonContext.Default.SubscriptionsListenRequestParams,
+            McpJsonUtilities.JsonContext.Default.EmptyResult);
+    }
+
+    /// <summary>Tracks an active <c>subscriptions/listen</c> subscription for notification fan-out.</summary>
+    /// <param name="Id">The id of the <c>subscriptions/listen</c> request, reused as the SEP-2575 subscription id.</param>
+    /// <param name="Granted">The notification types the server agreed to deliver on this subscription.</param>
+    /// <param name="RelatedTransport">
+    /// The transport the <c>subscriptions/listen</c> request arrived on. For Streamable HTTP this is the
+    /// per-request response stream the subscription must be delivered on; for stdio it is <see langword="null"/>,
+    /// so notifications fall back to the shared session channel.
+    /// </param>
+    private sealed record ActiveSubscription(RequestId Id, SubscriptionsListenNotifications Granted, ITransport? RelatedTransport);
+
+    private readonly ConcurrentDictionary<RequestId, ActiveSubscription> _activeSubscriptions = new();
+
+    /// <summary>
+    /// Delivers a <c>*/list_changed</c> notification triggered by a server-side collection change.
+    /// </summary>
+    /// <remarks>
+    /// Pre-SEP-2575 clients do not open <c>subscriptions/listen</c> streams, so they keep receiving a single
+    /// session-wide broadcast. Draft clients instead receive only the change notifications they explicitly
+    /// requested, each routed back over the originating subscription stream and tagged with its id; the server
+    /// <b>MUST NOT</b> send a draft client notification types it never subscribed to.
+    /// </remarks>
+    private async Task SendListChangedNotificationAsync(string notificationMethod)
+    {
+        // Legacy clients never open a subscriptions/listen stream, so they keep the session-wide broadcast.
+        // subscriptions/listen is a SEP-2575 draft feature, so draft clients instead get a fan-out limited
+        // to the notification types they explicitly subscribed to.
+        if (!IsDraftProtocol())
+        {
+            await this.SendNotificationAsync(notificationMethod).ConfigureAwait(false);
+            return;
+        }
+
+        foreach (var subscription in _activeSubscriptions.Values)
+        {
+            if (!GrantsListChanged(subscription.Granted, notificationMethod))
+            {
+                continue;
+            }
+
+            try
+            {
+                await SendSubscriptionNotificationAsync(subscription, notificationMethod, paramsNode: null, CancellationToken.None).ConfigureAwait(false);
+            }
+            catch (Exception ex)
+            {
+                // A single closed or faulted subscription stream must not prevent fan-out to the others.
+                SubscriptionNotificationFailed(notificationMethod, subscription.Id.ToString(), ex);
+            }
+        }
+    }
+
+    /// <summary>
+    /// Sends <paramref name="method"/> over <paramref name="subscription"/>'s stream, tagging it with the
+    /// SEP-2575 <c>_meta</c> subscription id so clients sharing a channel (notably stdio) can demultiplex it.
+    /// </summary>
+    private Task SendSubscriptionNotificationAsync(ActiveSubscription subscription, string method, JsonNode? paramsNode, CancellationToken cancellationToken)
+    {
+        var paramsObject = paramsNode as JsonObject ?? new JsonObject();
+        if (paramsObject["_meta"] is not JsonObject meta)
+        {
+            meta = new JsonObject();
+            paramsObject["_meta"] = meta;
+        }
+
+        meta[MetaKeys.SubscriptionId] = subscription.Id.Id switch
+        {
+            string stringId => JsonValue.Create(stringId),
+            long longId => JsonValue.Create(longId),
+            _ => null,
+        };
+
+        var notification = new JsonRpcNotification
+        {
+            Method = method,
+            Params = paramsObject,
+            Context = new JsonRpcMessageContext { RelatedTransport = subscription.RelatedTransport },
+        };
+
+        return SendMessageAsync(notification, cancellationToken);
+    }
+
+    /// <summary>
+    /// Sends the SEP-2575 <c>subscriptions/acknowledged</c> notification for a subscription, carrying the
+    /// notification types the server agreed to deliver. Routed back over the subscription's own stream and
+    /// tagged with its id like every other subscription notification.
+    /// </summary>
+    private Task SendSubscriptionAckAsync(ActiveSubscription subscription, CancellationToken cancellationToken)
+    {
+        var ackParams = JsonSerializer.SerializeToNode(
+            new SubscriptionsAcknowledgedNotificationParams { Notifications = subscription.Granted },
+            McpJsonUtilities.JsonContext.Default.SubscriptionsAcknowledgedNotificationParams);
+
+        return SendSubscriptionNotificationAsync(
+            subscription,
+            NotificationMethods.SubscriptionsAcknowledgedNotification,
+            ackParams,
+            cancellationToken);
+    }
+
+    /// <summary>Maps a <c>*/list_changed</c> method to the subscription filter flag that enables it.</summary>
+    private static bool GrantsListChanged(SubscriptionsListenNotifications granted, string method) => method switch
+    {
+        NotificationMethods.ToolListChangedNotification => granted.ToolsListChanged == true,
+        NotificationMethods.PromptListChangedNotification => granted.PromptsListChanged == true,
+        NotificationMethods.ResourceListChangedNotification => granted.ResourcesListChanged == true,
+        _ => false,
+    };
+
     private void ConfigureCompletion(McpServerOptions options)
     {
         var completeHandler = options.Handlers.CompleteHandler;
@@ -487,6 +809,13 @@ internal sealed partial class McpServerImpl : McpServer
         updateTaskHandler ??= (static async (request, _) => throw new McpProtocolException($"Unknown task: '{request.Params?.TaskId}'", McpErrorCode.InvalidParams));
         cancelTaskHandler ??= (static async (request, _) => throw new McpProtocolException($"Unknown task: '{request.Params?.TaskId}'", McpErrorCode.InvalidParams));
 
+        // The tasks/* methods do not exist before the draft revision (SEP-2663). Reject them with
+        // MethodNotFound when the request was negotiated under a legacy protocol version. The handlers
+        // stay registered so a dual-era server still serves them for draft requests.
+        getTaskHandler = GateTaskMethodToDraft(getTaskHandler, RequestMethods.TasksGet);
+        updateTaskHandler = GateTaskMethodToDraft(updateTaskHandler, RequestMethods.TasksUpdate);
+        cancelTaskHandler = GateTaskMethodToDraft(cancelTaskHandler, RequestMethods.TasksCancel);
+
         // Advertise tasks extension in server capabilities.
         ServerCapabilities.Extensions ??= new Dictionary<string, object>();
         ServerCapabilities.Extensions[McpExtensions.Tasks] = new JsonObject();
@@ -510,6 +839,26 @@ internal sealed partial class McpServerImpl : McpServer
             McpJsonUtilities.JsonContext.Default.CancelTaskResult);
     }
 
+    /// <summary>
+    /// Wraps a tasks/* request handler so it throws <see cref="McpErrorCode.MethodNotFound"/> unless the
+    /// request was negotiated under the draft revision. The tasks extension (SEP-2663) only interoperates
+    /// under draft, and these methods don't exist on legacy peers.
+    /// </summary>
+    private McpRequestHandler<TParams, TResult> GateTaskMethodToDraft<TParams, TResult>(
+        McpRequestHandler<TParams, TResult> inner, string method)
+        => (request, cancellationToken) =>
+        {
+            if (!IsDraftProtocolRequest(request.JsonRpcRequest))
+            {
+                throw new McpProtocolException(
+                    $"The method '{method}' requires the draft protocol revision ('{DraftProtocolVersion}'); " +
+                    $"the negotiated protocol version is '{NegotiatedProtocolVersion ?? "(none)"}'.",
+                    McpErrorCode.MethodNotFound);
+            }
+
+            return inner(request, cancellationToken);
+        };
+
     private void ConfigureExperimentalAndExtensions(McpServerOptions options)
     {
         ServerCapabilities.Experimental = options.Capabilities?.Experimental;
@@ -911,7 +1260,12 @@ internal sealed partial class McpServerImpl : McpServer
             var innerTaskHandler = callToolWithTaskHandler;
             callToolWithTaskHandler = async (request, cancellationToken) =>
             {
-                if (HasTaskExtensionOptIn(request.Params?.Meta))
+                // The SEP-2663 Tasks extension is draft-only: the task wire shapes we ship do not
+                // interoperate with legacy (<= 2025-11-25) peers. Only materialize a task when the
+                // request was negotiated under the draft revision AND the client opted in; otherwise
+                // run the inner handler and return the direct result (best-effort downgrade, which also
+                // defends against a non-conformant legacy client that forges the opt-in envelope).
+                if (IsDraftProtocolRequest(request.JsonRpcRequest) && HasTaskExtensionOptIn(request.Params?.Meta))
                 {
                     var taskInfo = await taskStore.CreateTaskAsync(cancellationToken).ConfigureAwait(false);
                     var taskId = taskInfo.TaskId;
@@ -1353,15 +1707,10 @@ internal sealed partial class McpServerImpl : McpServer
     // Per SEP-2663 §51, the client opts in to the tasks extension on a per-request basis
     // via the SEP-2575 capabilities envelope:
     //   _meta/io.modelcontextprotocol/clientCapabilities/extensions/io.modelcontextprotocol/tasks = {}
-    // TODO: swap the literals for a shared NotificationMethods.ClientCapabilitiesMetaKey once
-    // the SEP-2575 plumbing lands.
-    private const string ClientCapabilitiesMetaKey = "io.modelcontextprotocol/clientCapabilities";
-    private const string ExtensionsKey = "extensions";
-
     private static bool HasTaskExtensionOptIn(JsonObject? meta) =>
         meta is not null &&
-        meta[ClientCapabilitiesMetaKey] is JsonObject caps &&
-        caps[ExtensionsKey] is JsonObject exts &&
+        meta[MetaKeys.ClientCapabilities] is JsonObject caps &&
+        caps["extensions"] is JsonObject exts &&
         exts.ContainsKey(McpExtensions.Tasks);
 
     private JsonRpcMessageFilter BuildMessageFilterPipeline(IList<McpMessageFilter> filters)
@@ -1421,10 +1770,12 @@ internal sealed partial class McpServerImpl : McpServer
         };
 
     /// <summary>
-    /// Checks whether the negotiated protocol version enables MRTR per SEP-2322 (DRAFT-2026-v1).
+    /// Checks whether the negotiated protocol version enables MRTR per SEP-2322 (2026-07-28). MRTR rides on
+    /// the draft revision, so this is the MRTR-meaning alias of <see cref="McpSession.IsDraftProtocol"/> -
+    /// use it at the input-required/handler-suspension sites where the intent is "the client understands
+    /// <see cref="InputRequiredResult"/>" rather than "the peer speaks the draft revision".
     /// </summary>
-    internal bool ClientSupportsMrtr() =>
-        _negotiatedProtocolVersion == McpSessionHandler.DraftProtocolVersion;
+    internal bool ClientSupportsMrtr() => IsDraftProtocol();
 
     /// <summary>
     /// Returns <see langword="true"/> when the session is stateful - the same server instance handles
@@ -1436,6 +1787,18 @@ internal sealed partial class McpServerImpl : McpServer
     internal bool IsStatefulSession() =>
         _sessionTransport is not StreamableHttpServerTransport { Stateless: true };
 
+    /// <summary>
+    /// Returns <see langword="true"/> when the given request was negotiated under the draft protocol
+    /// revision, derived from the per-request <c>_meta</c>/<c>MCP-Protocol-Version</c> value (so it works
+    /// for sessionless draft over stateless HTTP) and falling back to the session-negotiated version.
+    /// Used to gate the SEP-2663 Tasks extension, which only interoperates under the draft revision.
+    /// </summary>
+    private bool IsDraftProtocolRequest(JsonRpcRequest? request) =>
+        string.Equals(
+            request?.Context?.ProtocolVersion ?? NegotiatedProtocolVersion,
+            DraftProtocolVersion,
+            StringComparison.Ordinal);
+
     /// <inheritdoc />
     public override bool IsMrtrSupported => ClientSupportsMrtr() || IsStatefulSession();
 
@@ -1453,14 +1816,6 @@ internal sealed partial class McpServerImpl : McpServer
     {
         const int MaxRetries = 10;
 
-        // In stateless mode, pick up the negotiated draft protocol version from the
-        // transport-provided request context because there is no long-lived initialize handshake state.
-        if (_negotiatedProtocolVersion is null &&
-            request.Context?.ProtocolVersion is { } headerProtocolVersion)
-        {
-            _negotiatedProtocolVersion = headerProtocolVersion;
-        }
-
         for (int retry = 0; ; retry++)
         {
             try
@@ -1479,7 +1834,6 @@ internal sealed partial class McpServerImpl : McpServer
                 // In stateless mode without MRTR, the server can't resolve input requests via
                 // JSON-RPC (no persistent session for server-to-client requests), and the client
                 // won't recognize the InputRequiredResult. This is the one unsupported configuration.
-                // TODO(stateless-draft): When DRAFT-2026-v1 becomes stateless-only, the IsStatefulSession() gate collapses - the stateful path will only matter for legacy clients on the current protocol.
                 if (!IsStatefulSession())
                 {
                     throw new McpException(
@@ -1646,15 +2000,6 @@ internal sealed partial class McpServerImpl : McpServer
 
         _requestHandlers[method] = async (request, cancellationToken) =>
         {
-            // In stateless mode, each request creates a new server instance that never saw the
-            // initialize handshake, so _negotiatedProtocolVersion is null. Pick it up from the
-            // Mcp-Protocol-Version header that the transport layer flowed via JsonRpcMessageContext.
-            if (_negotiatedProtocolVersion is null &&
-                request.Context?.ProtocolVersion is { } headerProtocolVersion)
-            {
-                _negotiatedProtocolVersion = headerProtocolVersion;
-            }
-
             // Check for MRTR retry: if requestState is present, look up the continuation.
             if (request.Params is JsonObject paramsObj &&
                 paramsObj.TryGetPropertyValue("requestState", out var requestStateNode) &&
@@ -1706,7 +2051,7 @@ internal sealed partial class McpServerImpl : McpServer
             }
 
             // Implicit MRTR (handler suspension across ElicitAsync/SampleAsync) emits
-            // InputRequiredResult on the wire, which only DRAFT-2026-v1 clients understand,
+            // InputRequiredResult on the wire, which only 2026-07-28 clients understand,
             // and requires the same server instance to handle the retry (stateful session).
             // For all other cases - legacy clients, stateless sessions - fall through to the
             // exception-based path, which transparently resolves InputRequiredException via
@@ -1873,4 +2218,7 @@ internal sealed partial class McpServerImpl : McpServer
 
     [LoggerMessage(Level = LogLevel.Debug, Message = "An MRTR handler threw an unhandled exception.")]
     private partial void MrtrHandlerError(Exception exception);
+
+    [LoggerMessage(Level = LogLevel.Debug, Message = "Failed to deliver \"{NotificationMethod}\" to subscription \"{SubscriptionId}\".")]
+    private partial void SubscriptionNotificationFailed(string notificationMethod, string subscriptionId, Exception exception);
 }

src/ModelContextProtocol.Core/Server/SseEventStreamMode.cs

@@ -3,6 +3,7 @@ namespace ModelContextProtocol.Server;
 /// <summary>
 /// Represents the mode of an SSE event stream.
 /// </summary>
+[Obsolete(Obsoletions.LegacyStatefulHttp_Message, DiagnosticId = Obsoletions.LegacyStatefulHttp_DiagnosticId, UrlFormat = Obsoletions.LegacyStatefulHttp_Url)]
 public enum SseEventStreamMode
 {
     /// <summary>

src/ModelContextProtocol.Core/Server/SseEventStreamOptions.cs

@@ -3,6 +3,7 @@ namespace ModelContextProtocol.Server;
 /// <summary>
 /// Configuration options for creating an SSE event stream.
 /// </summary>
+[Obsolete(Obsoletions.LegacyStatefulHttp_Message, DiagnosticId = Obsoletions.LegacyStatefulHttp_DiagnosticId, UrlFormat = Obsoletions.LegacyStatefulHttp_Url)]
 public sealed class SseEventStreamOptions
 {
     /// <summary>

src/ModelContextProtocol.Core/Server/StreamableHttpPostTransport.cs

@@ -22,7 +22,9 @@ internal sealed partial class StreamableHttpPostTransport(
     private readonly SseEventWriter _httpSseWriter = new(responseStream);
 
     private TaskCompletionSource<bool>? _storeStreamTcs;
+#pragma warning disable MCP9006 // Stateful Streamable HTTP resumability types are obsolete but still wired up internally.
     private ISseEventStreamWriter? _storeSseWriter;
+#pragma warning restore MCP9006
 
     private RequestId _pendingRequest;
     private bool _finalResponseMessageSent;
@@ -176,7 +178,9 @@ internal sealed partial class StreamableHttpPostTransport(
 
         // Set the mode to 'Polling' so that the replay stream ends as soon as all available messages have been sent.
         // This prevents the client from immediately establishing another long-lived connection.
+#pragma warning disable MCP9006 // Stateful Streamable HTTP resumability types are obsolete but still wired up internally.
         await _storeSseWriter.SetModeAsync(SseEventStreamMode.Polling, cancellationToken).ConfigureAwait(false);
+#pragma warning restore MCP9006
 
         // Signal completion so HandlePostAsync can return.
         _httpResponseTcs.TrySetResult(true);

src/ModelContextProtocol.Core/Server/StreamableHttpServerTransport.cs

@@ -40,7 +40,9 @@ public sealed partial class StreamableHttpServerTransport : ITransport
     private readonly ILogger _logger;
 
     private SseEventWriter? _httpSseWriter;
+#pragma warning disable MCP9006 // Stateful Streamable HTTP resumability types are obsolete but still wired up internally.
     private ISseEventStreamWriter? _storeSseWriter;
+#pragma warning restore MCP9006
     private TaskCompletionSource<bool>? _httpResponseTcs;
     private string? _negotiatedProtocolVersion;
     private bool _getHttpRequestStarted;
@@ -80,6 +82,7 @@ public sealed partial class StreamableHttpServerTransport : ITransport
     /// Gets or sets the event store for resumability support.
     /// When set, events are stored and can be replayed when clients reconnect with a Last-Event-ID header.
     /// </summary>
+    [Obsolete(Obsoletions.LegacyStatefulHttp_Message, DiagnosticId = Obsoletions.LegacyStatefulHttp_DiagnosticId, UrlFormat = Obsoletions.LegacyStatefulHttp_Url)]
     public ISseEventStreamStore? EventStreamStore { get; init; }
 
     /// <summary>
@@ -357,6 +360,7 @@ public sealed partial class StreamableHttpServerTransport : ITransport
         }
     }
 
+#pragma warning disable MCP9006 // Stateful Streamable HTTP resumability types are obsolete but still wired up internally.
     internal async ValueTask<ISseEventStreamWriter?> TryCreateEventStreamAsync(string streamId, CancellationToken cancellationToken)
     {
         if (EventStreamStore is null || !McpSessionHandler.SupportsPrimingEvent(_negotiatedProtocolVersion))
@@ -377,6 +381,7 @@ public sealed partial class StreamableHttpServerTransport : ITransport
 
         return sseEventStreamWriter;
     }
+#pragma warning restore MCP9006
 
     [LoggerMessage(Level = LogLevel.Warning,
         Message = "Sending server-to-client JSON-RPC request '{Method}' over the standalone GET SSE stream (SessionId: '{SessionId}'). " +