Add server Streamable HTTP transport

- Changes IJsonRpcMessage to an abstract base class so RelatedTransport will always be available
- Streamable HTTP supports multiple concurrent HTTP request with their own indpendent SSE response streams
- RelatedTransport indicates the source or destination of the JsonRpcMessage

- Changes the default RequestId to a JSON number for better compatibility with MCP servers in the wild

Author: halter73

src/ModelContextProtocol.AspNetCore/HttpMcpServerBuilderExtensions.cs

@@ -11,18 +11,20 @@ public static class HttpMcpServerBuilderExtensions
 {
     /// <summary>
     /// Adds the services necessary for <see cref="M:McpEndpointRouteBuilderExtensions.MapMcp"/>
-    /// to handle MCP requests and sessions using the MCP HTTP Streaming transport. For more information on configuring the underlying HTTP server
+    /// to handle MCP requests and sessions using the MCP Streamable HTTP transport. For more information on configuring the underlying HTTP server
     /// to control things like port binding custom TLS certificates, see the <see href="https://learn.microsoft.com/aspnet/core/fundamentals/minimal-apis">Minimal APIs quick reference</see>.
     /// </summary>
     /// <param name="builder">The builder instance.</param>
-    /// <param name="configureOptions">Configures options for the HTTP Streaming transport. This allows configuring per-session
+    /// <param name="configureOptions">Configures options for the Streamable HTTP transport. This allows configuring per-session
     /// <see cref="McpServerOptions"/> and running logic before and after a session.</param>
     /// <returns>The builder provided in <paramref name="builder"/>.</returns>
     /// <exception cref="ArgumentNullException"><paramref name="builder"/> is <see langword="null"/>.</exception>
     public static IMcpServerBuilder WithHttpTransport(this IMcpServerBuilder builder, Action<HttpServerTransportOptions>? configureOptions = null)
     {
         ArgumentNullException.ThrowIfNull(builder);
         builder.Services.TryAddSingleton<StreamableHttpHandler>();
+        builder.Services.TryAddSingleton<SseHandler>();
+        builder.Services.AddHostedService<IdleSessionBackgroundService>();
 
         if (configureOptions is not null)
         {

src/ModelContextProtocol.AspNetCore/HttpMcpSession.cs

@@ -1,19 +1,31 @@
-using ModelContextProtocol.Protocol.Transport;
+using ModelContextProtocol.Server;
 using System.Security.Claims;
 
 namespace ModelContextProtocol.AspNetCore;
 
-internal class HttpMcpSession
+internal sealed class HttpMcpSession<TTransport>(string sessionId, TTransport transport, ClaimsPrincipal user, TimeProvider timeProvider)
 {
-    public HttpMcpSession(SseResponseStreamTransport transport, ClaimsPrincipal user)
+    private int _getStarted;
+    private int _referenceCount;
+
+    public string Id { get; } = sessionId;
+    public TTransport Transport { get; } = transport;
+    public (string Type, string Value, string Issuer)? UserIdClaim { get; } = GetUserIdClaim(user);
+
+    public bool IsActive => _referenceCount > 0;
+    public long LastActivityTicks { get; private set; } = timeProvider.GetUtcNow().UtcTicks;
+
+    public bool TryStartGet() => Interlocked.Exchange(ref _getStarted, 1) == 0;
+
+    public IMcpServer? Server { get; init; }
+    public Task? ServerRunTask { get; init; }
+
+    public IDisposable AcquireReference()
     {
-        Transport = transport;
-        UserIdClaim = GetUserIdClaim(user);
+        Interlocked.Increment(ref _referenceCount);
+        return new UnreferenceDisposable(this, timeProvider);
     }
 
-    public SseResponseStreamTransport Transport { get; }
-    public (string Type, string Value, string Issuer)? UserIdClaim { get; }
-
     public bool HasSameUserId(ClaimsPrincipal user)
         => UserIdClaim == GetUserIdClaim(user);
 
@@ -36,4 +48,15 @@ private static (string Type, string Value, string Issuer)? GetUserIdClaim(Claims
 
         return null;
     }
+
+    private sealed class UnreferenceDisposable(HttpMcpSession<TTransport> session, TimeProvider timeProvider) : IDisposable
+    {
+        public void Dispose()
+        {
+            if (Interlocked.Decrement(ref session._referenceCount) == 0)
+            {
+                session.LastActivityTicks = timeProvider.GetUtcNow().UtcTicks;
+            }
+        }
+    }
 }

src/ModelContextProtocol.AspNetCore/HttpServerTransportOptions.cs

@@ -21,4 +21,18 @@ public class HttpServerTransportOptions
     /// This is useful for running logic before a sessions starts and after it completes.
     /// </summary>
     public Func<HttpContext, IMcpServer, CancellationToken, Task>? RunSessionHandler { get; set; }
+
+    /// <summary>
+    /// Represents the duration of time the server will wait between any active requests before timing out an
+    /// MCP session. This is checked in background every 5 seconds. A client trying to resume a session will
+    /// receive a 404 status code and should restart their session. A client can keep their session open by
+    /// keeping a GET request open. The default value is set to 2 minutes.
+    /// minutes.
+    /// </summary>
+    public TimeSpan IdleTimeout { get; set; } = TimeSpan.FromMinutes(2);
+
+    /// <summary>
+    /// Used for testing the <see cref="IdleTimeout"/>.
+    /// </summary>
+    public TimeProvider TimeProvider { get; set; } = TimeProvider.System;
 }

src/ModelContextProtocol.AspNetCore/IdleSessionBackgroundService.cs

@@ -0,0 +1,49 @@
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Options;
+
+namespace ModelContextProtocol.AspNetCore;
+
+internal sealed class IdleSessionBackgroundService(StreamableHttpHandler handler, IOptions<HttpServerTransportOptions> options) : BackgroundService
+{
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        var timeProvider = options.Value.TimeProvider;
+        var timer = new PeriodicTimer(TimeSpan.FromSeconds(5), timeProvider);
+
+        while (!stoppingToken.IsCancellationRequested && await timer.WaitForNextTickAsync(stoppingToken))
+        {
+            var idleActivityCutoff = timeProvider.GetUtcNow().Ticks - options.Value.IdleTimeout.Ticks;
+
+            foreach (var (_, session) in handler.Sessions)
+            {
+                if (session.IsActive || session.LastActivityTicks > idleActivityCutoff)
+                {
+                    continue;
+                }
+
+                if (handler.Sessions.TryRemove(session.Id, out var removedSession))
+                {
+                    await removedSession.Transport.DisposeAsync();
+                }
+            }
+        }
+    }
+
+    public override async Task StopAsync(CancellationToken cancellationToken)
+    {
+        try
+        {
+            foreach (var (sessionKey, _) in handler.Sessions)
+            {
+                if (handler.Sessions.TryRemove(sessionKey, out var session))
+                {
+                    await session.Transport.DisposeAsync();
+                }
+            }
+        }
+        finally
+        {
+            await base.StopAsync(cancellationToken);
+        }
+    }
+}

src/ModelContextProtocol.AspNetCore/McpEndpointRouteBuilderExtensions.cs

@@ -1,6 +1,9 @@
-using Microsoft.AspNetCore.Routing;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Http.Metadata;
+using Microsoft.AspNetCore.Routing;
 using Microsoft.Extensions.DependencyInjection;
 using ModelContextProtocol.AspNetCore;
+using ModelContextProtocol.Protocol.Messages;
 using System.Diagnostics.CodeAnalysis;
 
 namespace Microsoft.AspNetCore.Builder;
@@ -11,21 +14,42 @@ namespace Microsoft.AspNetCore.Builder;
 public static class McpEndpointRouteBuilderExtensions
 {
     /// <summary>
-    /// Sets up endpoints for handling MCP HTTP Streaming transport.
-    /// See <see href="https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http">the protocol specification</see> for details about the Streamable HTTP transport.
+    /// Sets up endpoints for handling MCP Streamable HTTP transport.
+    /// See <see href="https://modelcontextprotocol.io/specification/2025-03-26/basic/transports#streamable-http">the 2025-03-26 protocol specification</see> for details about the Streamable HTTP transport.
+    /// Also maps legacy SSE endpoints for backward compatibility at the path "/sse" and "/message". <see href="https://modelcontextprotocol.io/specification/2024-11-05/basic/transports#http-with-sse">the 2024-11-05 protocol specification</see> for details about the HTTP with SSE transport.
     /// </summary>
     /// <param name="endpoints">The web application to attach MCP HTTP endpoints.</param>
     /// <param name="pattern">The route pattern prefix to map to.</param>
     /// <returns>Returns a builder for configuring additional endpoint conventions like authorization policies.</returns>
     public static IEndpointConventionBuilder MapMcp(this IEndpointRouteBuilder endpoints, [StringSyntax("Route")] string pattern = "")
     {
-        var handler = endpoints.ServiceProvider.GetService<StreamableHttpHandler>() ??
+        var streamableHttpHandler = endpoints.ServiceProvider.GetService<StreamableHttpHandler>() ??
             throw new InvalidOperationException("You must call WithHttpTransport(). Unable to find required services. Call builder.Services.AddMcpServer().WithHttpTransport() in application startup code.");
 
-        var routeGroup = endpoints.MapGroup(pattern);
-        routeGroup.MapGet("", handler.HandleRequestAsync);
-        routeGroup.MapGet("/sse", handler.HandleRequestAsync);
-        routeGroup.MapPost("/message", handler.HandleRequestAsync);
-        return routeGroup;
+        var mcpGroup = endpoints.MapGroup(pattern);
+        var streamableHttpGroup = mcpGroup.MapGroup("")
+            .WithDisplayName(b => $"MCP Streamable HTTP | {b.DisplayName}")
+            .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status404NotFound, typeof(JsonRpcError), contentTypes: ["application/json"]));
+
+        streamableHttpGroup.MapPost("", streamableHttpHandler.HandlePostRequestAsync)
+            .WithMetadata(new AcceptsMetadata(["application/json"]))
+            .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status200OK, contentTypes: ["text/event-stream"]))
+            .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status202Accepted));
+        streamableHttpGroup.MapGet("", streamableHttpHandler.HandleGetRequestAsync)
+            .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status200OK, contentTypes: ["text/event-stream"]));
+        streamableHttpGroup.MapDelete("", streamableHttpHandler.HandleDeleteRequestAsync);
+
+        // Map legacy HTTP with SSE endpoints.
+        var sseHandler = endpoints.ServiceProvider.GetRequiredService<SseHandler>();
+        var sseGroup = mcpGroup.MapGroup("")
+            .WithDisplayName(b => $"MCP HTTP with SSE | {b.DisplayName}");
+
+        sseGroup.MapGet("/sse", sseHandler.HandleSseRequestAsync)
+            .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status200OK, contentTypes: ["text/event-stream"]));
+        sseGroup.MapPost("/message", sseHandler.HandleMessageRequestAsync)
+            .WithMetadata(new AcceptsMetadata(["application/json"]))
+            .WithMetadata(new ProducesResponseTypeMetadata(StatusCodes.Status202Accepted));
+
+        return mcpGroup;
     }
 }

src/ModelContextProtocol.AspNetCore/SseHandler.cs

@@ -0,0 +1,108 @@
+using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using ModelContextProtocol.Protocol.Messages;
+using ModelContextProtocol.Protocol.Transport;
+using ModelContextProtocol.Server;
+using ModelContextProtocol.Utils.Json;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+
+namespace ModelContextProtocol.AspNetCore;
+
+internal sealed class SseHandler(
+    IOptions<McpServerOptions> mcpServerOptionsSnapshot,
+    IOptionsFactory<McpServerOptions> mcpServerOptionsFactory,
+    IOptions<HttpServerTransportOptions> httpMcpServerOptions,
+    IHostApplicationLifetime hostApplicationLifetime,
+    ILoggerFactory loggerFactory)
+{
+    private readonly ConcurrentDictionary<string, HttpMcpSession<SseResponseStreamTransport>> _sessions = new(StringComparer.Ordinal);
+
+    public async Task HandleSseRequestAsync(HttpContext context)
+    {
+        var sessionId = StreamableHttpHandler.MakeNewSessionId();
+
+        // If the server is shutting down, we need to cancel all SSE connections immediately without waiting for HostOptions.ShutdownTimeout
+        // which defaults to 30 seconds.
+        using var sseCts = CancellationTokenSource.CreateLinkedTokenSource(context.RequestAborted, hostApplicationLifetime.ApplicationStopping);
+        var cancellationToken = sseCts.Token;
+
+        StreamableHttpHandler.InitializeSseResponse(context);
+
+        await using var transport = new SseResponseStreamTransport(context.Response.Body, $"message?sessionId={sessionId}");
+        var httpMcpSession = new HttpMcpSession<SseResponseStreamTransport>(sessionId, transport, context.User, httpMcpServerOptions.Value.TimeProvider);
+        if (!_sessions.TryAdd(sessionId, httpMcpSession))
+        {
+            throw new UnreachableException($"Unreachable given good entropy! Session with ID '{sessionId}' has already been created.");
+        }
+
+        try
+        {
+            var mcpServerOptions = mcpServerOptionsSnapshot.Value;
+            if (httpMcpServerOptions.Value.ConfigureSessionOptions is { } configureSessionOptions)
+            {
+                mcpServerOptions = mcpServerOptionsFactory.Create(Options.DefaultName);
+                await configureSessionOptions(context, mcpServerOptions, cancellationToken);
+            }
+
+            var transportTask = transport.RunAsync(cancellationToken);
+
+            try
+            {
+                await using var mcpServer = McpServerFactory.Create(transport, mcpServerOptions, loggerFactory, context.RequestServices);
+                context.Features.Set(mcpServer);
+
+                var runSessionAsync = httpMcpServerOptions.Value.RunSessionHandler ?? StreamableHttpHandler.RunSessionAsync;
+                await runSessionAsync(context, mcpServer, cancellationToken);
+            }
+            finally
+            {
+                await transport.DisposeAsync();
+                await transportTask;
+            }
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            // RequestAborted always triggers when the client disconnects before a complete response body is written,
+            // but this is how SSE connections are typically closed.
+        }
+        finally
+        {
+            _sessions.TryRemove(sessionId, out _);
+        }
+    }
+
+    public async Task HandleMessageRequestAsync(HttpContext context)
+    {
+        if (!context.Request.Query.TryGetValue("sessionId", out var sessionId))
+        {
+            await Results.BadRequest("Missing sessionId query parameter.").ExecuteAsync(context);
+            return;
+        }
+
+        if (!_sessions.TryGetValue(sessionId.ToString(), out var httpMcpSession))
+        {
+            await Results.BadRequest($"Session ID not found.").ExecuteAsync(context);
+            return;
+        }
+
+        if (!httpMcpSession.HasSameUserId(context.User))
+        {
+            await Results.Forbid().ExecuteAsync(context);
+            return;
+        }
+
+        var message = (JsonRpcMessage?)await context.Request.ReadFromJsonAsync(McpJsonUtilities.DefaultOptions.GetTypeInfo(typeof(JsonRpcMessage)), context.RequestAborted);
+        if (message is null)
+        {
+            await Results.BadRequest("No message in request body.").ExecuteAsync(context);
+            return;
+        }
+
+        await httpMcpSession.Transport.OnMessageReceivedAsync(message, context.RequestAborted);
+        context.Response.StatusCode = StatusCodes.Status202Accepted;
+        await context.Response.WriteAsync("Accepted");
+    }
+}