Allow stateful HTTP endpoints to recover after DELETE

.github/workflows/ci.yml

@@ -25,7 +25,7 @@ jobs:
           # Current leg: catches breakage from new Xcode and SDK releases.
           - os: macos-latest
             swift-version: Xcode default
-          - os: ubuntu-latest
+          - os: ubuntu-24.04
             swift-version: 6.1.0
 
     runs-on: ${{ matrix.os }}
@@ -35,13 +35,13 @@ jobs:
       - uses: actions/checkout@v4
 
       - name: Prepare Linux apt dependencies
-        if: matrix.os == 'ubuntu-latest'
+        if: matrix.os == 'ubuntu-24.04'
         run: |
           sudo -n apt-get update
           sudo -n apt-get install -y --no-install-recommends libcurl4-openssl-dev
 
       - name: Setup Swift on Linux
-        if: matrix.os == 'ubuntu-latest'
+        if: matrix.os == 'ubuntu-24.04'
         uses: vapor/swiftly-action@v0.2
         with:
           toolchain: ${{ matrix.swift-version }}

README.md

@@ -1343,10 +1343,13 @@ The Swift SDK provides multiple built-in transports:
 | [`StdioTransport`](/Sources/MCP/Base/Transports/StdioTransport.swift) | Implements [stdio transport](https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#stdio) using standard input/output streams | Apple platforms, Linux with glibc | Local subprocesses, CLI tools |
 | [`HTTPClientTransport`](/Sources/MCP/Base/Transports/HTTPClientTransport.swift) | Implements [Streamable HTTP transport](https://modelcontextprotocol.io/specification/2025-06-18/basic/transports#streamable-http) using Foundation's URL Loading System | All platforms with Foundation | Remote servers, web applications |
 | [`StatelessHTTPServerTransport`](/Sources/MCP/Base/Transports/HTTPServer/StatelessHTTPServerTransport.swift) | HTTP server transport with simple request-response semantics; no session management or SSE streaming | All platforms with Foundation | Simple HTTP servers, serverless/edge functions |
-| [`StatefulHTTPServerTransport`](/Sources/MCP/Base/Transports/HTTPServer/StatefulHTTPServerTransport.swift) | HTTP server transport with full session management and SSE streaming for server-initiated messages | All platforms with Foundation | Full-featured HTTP servers, streaming notifications |
+| [`StatefulHTTPServerTransport`](/Sources/MCP/Base/Transports/HTTPServer/StatefulHTTPServerTransport.swift) | HTTP server transport for one session, with SSE streaming for server-initiated messages | All platforms with Foundation | One client session |
+| [`StatefulHTTPServerSessionManager`](/Sources/MCP/Base/Transports/HTTPServer/StatefulHTTPServerSessionManager.swift) | Routes an HTTP endpoint to a separate server and transport per session, including after DELETE | All platforms with Foundation | Stateful HTTP endpoints serving multiple sessions |
 | [`InMemoryTransport`](/Sources/MCP/Base/Transports/InMemoryTransport.swift) | Custom in-memory transport for direct communication within the same process | All platforms | Testing, debugging, same-process client-server communication |
 | [`NetworkTransport`](/Sources/MCP/Base/Transports/NetworkTransport.swift) | Custom transport using Apple's Network framework for TCP/UDP connections | Apple platforms only | Low-level networking, custom protocols |
 
+For exposed stateful HTTP endpoints, set `maxSessions` on `StatefulHTTPServerSessionManager` to a limit that fits your server's capacity.
+
 
 ### Custom Transport Implementation
 

Sources/MCP/Base/Transports/HTTPServer/StatefulHTTPServerSessionManager.swift

@@ -0,0 +1,197 @@
+import Foundation
+import Logging
+
+/// Routes Streamable HTTP requests to a separate server and transport per session.
+/// Use one manager for the HTTP endpoint so a client can initialize a new session
+/// after its previous session ends.
+public actor StatefulHTTPServerSessionManager {
+    public typealias ServerFactory = @Sendable (String, StatefulHTTPServerTransport) async throws -> Server
+
+    private struct Session {
+        let server: Server
+        let transport: StatefulHTTPServerTransport
+        var lastAccessedAt: Date
+    }
+
+    private struct FixedSessionIDGenerator: SessionIDGenerator {
+        let sessionID: String
+        func generateSessionID() -> String { sessionID }
+    }
+
+    private let serverFactory: ServerFactory
+    private let validationPipeline: any HTTPRequestValidationPipeline
+    private let retryInterval: Int?
+    private let sessionTimeout: TimeInterval
+    private let maxSessions: Int?
+    private let logger: Logger
+    private var sessions: [String: Session] = [:]
+    private var creatingSessions = 0
+    private var cleanupTask: Task<Void, Never>?
+    private var stoppingSessions: [UUID: Task<Void, Never>] = [:]
+    private var stopped = false
+
+    /// `serverFactory` must return a new `Server` for each session.
+    /// Set `maxSessions` to limit active sessions, including sessions still being created.
+    /// The default is `nil`, so deployments can choose a limit for their capacity.
+    public init(
+        sessionTimeout: TimeInterval = 3600,
+        maxSessions: Int? = nil,
+        validationPipeline: (any HTTPRequestValidationPipeline)? = nil,
+        retryInterval: Int? = nil,
+        logger: Logger? = nil,
+        serverFactory: @escaping ServerFactory
+    ) {
+        self.sessionTimeout = sessionTimeout
+        self.maxSessions = maxSessions.map { max(0, $0) }
+        self.validationPipeline = validationPipeline ?? StandardValidationPipeline(validators: [
+            OriginValidator.localhost(),
+            AcceptHeaderValidator(mode: .sseRequired),
+            ContentTypeValidator(),
+            ProtocolVersionValidator(),
+            SessionValidator(),
+        ])
+        self.retryInterval = retryInterval
+        self.logger = logger ?? Logger(
+            label: "mcp.transport.http.server.sessions",
+            factory: { _ in SwiftLogNoOpLogHandler() }
+        )
+        self.serverFactory = serverFactory
+    }
+
+    public func handleRequest(_ request: HTTPRequest) async -> HTTPResponse {
+        guard !stopped else {
+            return .error(statusCode: 503, .internalError("Server has stopped"))
+        }
+
+        if cleanupTask == nil {
+            cleanupTask = Task { [weak self] in
+                while !Task.isCancelled {
+                    try? await Task.sleep(for: .seconds(60))
+                    guard !Task.isCancelled, let self else { return }
+                    await self.removeExpiredSessions()
+                }
+            }
+        }
+
+        await removeExpiredSessions()
+        guard !stopped else {
+            return .error(statusCode: 503, .internalError("Server has stopped"))
+        }
+
+        if let sessionID = request.header(HTTPHeaderName.sessionID) {
+            guard var session = sessions[sessionID] else {
+                return .error(statusCode: 404, .invalidRequest("Session not found"))
+            }
+
+            if request.method.uppercased() == "DELETE" {
+                let response = await session.transport.handleRequest(request)
+                if response.statusCode == 200 {
+                    sessions.removeValue(forKey: sessionID)
+                    await session.server.stop()
+                }
+                return response
+            }
+
+            session.lastAccessedAt = Date()
+            sessions[sessionID] = session
+            return await session.transport.handleRequest(request)
+        }
+
+        guard request.method.uppercased() == "POST",
+            let body = request.body,
+            let kind = JSONRPCMessageKind(data: body),
+            kind.isInitializeRequest
+        else {
+            return .error(statusCode: 400, .invalidRequest("Missing session ID"))
+        }
+
+        let context = HTTPValidationContext(
+            httpMethod: "POST",
+            isInitializationRequest: true,
+            supportedProtocolVersions: Version.supported
+        )
+        if let errorResponse = validationPipeline.validate(request, context: context) {
+            return errorResponse
+        }
+
+        if let maxSessions, sessions.count + creatingSessions >= maxSessions {
+            return .error(statusCode: 503, .internalError("Session limit reached"))
+        }
+        creatingSessions += 1
+        defer { creatingSessions -= 1 }
+
+        let sessionID = UUID().uuidString
+        let transport = StatefulHTTPServerTransport(
+            sessionIDGenerator: FixedSessionIDGenerator(sessionID: sessionID),
+            validationPipeline: validationPipeline,
+            retryInterval: retryInterval,
+            logger: logger
+        )
+
+        var server: Server?
+        do {
+            let createdServer = try await serverFactory(sessionID, transport)
+            server = createdServer
+            try await createdServer.start(transport: transport)
+            guard !stopped else {
+                await createdServer.stop()
+                return .error(statusCode: 503, .internalError("Server has stopped"))
+            }
+
+            sessions[sessionID] = Session(
+                server: createdServer, transport: transport, lastAccessedAt: Date()
+            )
+            let response = await transport.handleValidatedInitializationRequest(request)
+            if case .error = response {
+                sessions.removeValue(forKey: sessionID)
+                await createdServer.stop()
+                return response
+            }
+            return response
+        } catch {
+            if let server {
+                await server.stop()
+            } else {
+                await transport.disconnect()
+            }
+            logger.error("Failed to create session: \(error)")
+            return .error(statusCode: 500, .internalError("Failed to create session"))
+        }
+    }
+
+    public func stop() async {
+        stopped = true
+        cleanupTask?.cancel()
+        cleanupTask = nil
+        let activeSessions = sessions
+        sessions.removeAll()
+        for (_, session) in activeSessions {
+            await session.server.stop()
+        }
+        for task in Array(stoppingSessions.values) {
+            await task.value
+        }
+    }
+
+    private func removeExpiredSessions() async {
+        let now = Date()
+        let expiredIDs = sessions.compactMap { sessionID, session in
+            now.timeIntervalSince(session.lastAccessedAt) > sessionTimeout ? sessionID : nil
+        }
+        for sessionID in expiredIDs {
+            guard let session = sessions[sessionID],
+                now.timeIntervalSince(session.lastAccessedAt) > sessionTimeout
+            else { continue }
+            guard !(await session.transport.hasActiveStandaloneStream) else { continue }
+            guard let current = sessions[sessionID],
+                now.timeIntervalSince(current.lastAccessedAt) > sessionTimeout
+            else { continue }
+            sessions.removeValue(forKey: sessionID)
+            let taskID = UUID()
+            stoppingSessions[taskID] = Task {
+                await session.server.stop()
+                self.stoppingSessions.removeValue(forKey: taskID)
+            }
+        }
+    }
+}

Sources/MCP/Base/Transports/HTTPServer/StatefulHTTPServerTransport.swift

@@ -1,7 +1,7 @@
 import Foundation
 import Logging
 
-/// A stateful HTTP server transport that manages sessions and uses SSE for streaming responses.
+/// A stateful HTTP server transport for one session, using SSE for responses.
 ///
 /// This transport implements the MCP Streamable HTTP specification with full session management:
 /// - Assigns a session ID during initialization (via `Mcp-Session-Id` header)
@@ -13,15 +13,17 @@ import Logging
 /// ## Usage
 ///
 /// ```swift
-/// let transport = StatefulHTTPServerTransport()  // Uses UUID by default
+/// let transport = StatefulHTTPServerTransport()  // One transport per session
 ///
 /// // Start the MCP server with this transport
 /// try await server.start(transport: transport)
 ///
-/// // In your HTTP framework handler:
+/// // In your HTTP framework handler for this session:
 /// let response = await transport.handleRequest(httpRequest)
 /// // Convert response to your framework's response type and return it
 /// ```
+/// Use ``StatefulHTTPServerSessionManager`` for an endpoint that creates new
+/// sessions after DELETE. A terminated transport cannot be reused.
 ///
 /// ## Framework Integration
 ///
@@ -64,6 +66,7 @@ public actor StatefulHTTPServerTransport: Transport, HTTPContextProviding {
     /// The standalone SSE stream continuation for server-initiated messages.
     /// Only one GET stream is allowed per session.
     private var standaloneSSEContinuation: AsyncThrowingStream<Data, Swift.Error>.Continuation?
+    private var standaloneStreamGeneration = 0
 
     /// Internal identifier for the standalone GET stream in the event store.
     private let standaloneStreamID = "_GET_stream"
@@ -158,6 +161,8 @@ public actor StatefulHTTPServerTransport: Transport, HTTPContextProviding {
         incomingStream
     }
 
+    var hasActiveStandaloneStream: Bool { standaloneSSEContinuation != nil }
+
     // MARK: - HTTP Request Handler
 
     /// Handles an incoming HTTP request from the framework adapter.
@@ -193,9 +198,20 @@ public actor StatefulHTTPServerTransport: Transport, HTTPContextProviding {
         }
     }
 
+    func handleValidatedInitializationRequest(_ request: HTTPRequest) -> HTTPResponse {
+        if terminated {
+            return .error(
+                statusCode: 404,
+                .invalidRequest("Not Found: Session has been terminated"),
+                sessionID: sessionID
+            )
+        }
+        return handlePost(request, skipValidation: true)
+    }
+
     // MARK: - POST Handler
 
-    private func handlePost(_ request: HTTPRequest) -> HTTPResponse {
+    private func handlePost(_ request: HTTPRequest, skipValidation: Bool = false) -> HTTPResponse {
         // Parse body first so we can determine if it's an initialization request
         guard let body = request.body, !body.isEmpty else {
             return .error(
@@ -222,7 +238,9 @@ public actor StatefulHTTPServerTransport: Transport, HTTPContextProviding {
         )
 
         // Run validation pipeline
-        if let errorResponse = validationPipeline.validate(request, context: context) {
+        if !skipValidation,
+            let errorResponse = validationPipeline.validate(request, context: context)
+        {
             return errorResponse
         }
 
@@ -341,7 +359,7 @@ public actor StatefulHTTPServerTransport: Transport, HTTPContextProviding {
 
         // Create standalone SSE stream
         let (sseStream, sseContinuation) = AsyncThrowingStream<Data, Swift.Error>.makeStream()
-        standaloneSSEContinuation = sseContinuation
+        registerStandaloneStream(sseContinuation)
 
         // Extract protocol version for priming event
         let protocolVersion = request.header(HTTPHeaderName.protocolVersion) ?? Version.latest
@@ -448,7 +466,7 @@ public actor StatefulHTTPServerTransport: Transport, HTTPContextProviding {
 
         // Re-register the stream for future messages
         if replay.streamID == standaloneStreamID {
-            standaloneSSEContinuation = sseContinuation
+            registerStandaloneStream(sseContinuation)
         } else {
             requestSSEContinuations[replay.streamID] = sseContinuation
         }
@@ -519,6 +537,23 @@ public actor StatefulHTTPServerTransport: Transport, HTTPContextProviding {
 
     // MARK: - Session Helpers
 
+    private func registerStandaloneStream(
+        _ continuation: AsyncThrowingStream<Data, Swift.Error>.Continuation
+    ) {
+        standaloneStreamGeneration += 1
+        let generation = standaloneStreamGeneration
+        standaloneSSEContinuation = continuation
+        continuation.onTermination = { [weak self] _ in
+            Task { await self?.clearStandaloneStream(generation: generation) }
+        }
+    }
+
+    private func clearStandaloneStream(generation: Int) {
+        if standaloneStreamGeneration == generation {
+            standaloneSSEContinuation = nil
+        }
+    }
+
     private func sessionHeaders() -> [String: String] {
         var headers: [String: String] = [:]
         if let sessionID {

Sources/MCP/Server/Server.swift

@@ -295,6 +295,11 @@ public actor Server {
         task?.cancel()
         task = nil
 
+        for (_, requestTask) in pendingRequestTasks {
+            requestTask.cancel()
+        }
+        pendingRequestTasks.removeAll()
+
         // Clear pending requests with errors
         let pendingRequestsToCancel = self.pendingRequests
         self.pendingRequests = [:]