diff --git a/Sources/MCP/Base/Transports/HTTPClientTransport.swift b/Sources/MCP/Base/Transports/HTTPClientTransport.swift index 060cd133..22b24697 100644 --- a/Sources/MCP/Base/Transports/HTTPClientTransport.swift +++ b/Sources/MCP/Base/Transports/HTTPClientTransport.swift @@ -86,8 +86,8 @@ public actor HTTPClientTransport: Transport { private var initialSessionIDSignalTask: Task? private var initialSessionIDContinuation: CheckedContinuation? - /// The last event ID received from the server for SSE stream resumability - private var lastEventID: String? + /// The last event ID from the standalone GET stream. + private var lastGETEventID: String? /// The retry interval (in milliseconds) from the server's SSE `retry:` field private var retryInterval: Int = 3000 // Default 3000ms per SSE spec @@ -96,6 +96,18 @@ public actor HTTPClientTransport: Transport { /// Used to trigger reconnection when a POST SSE stream closes without delivering data. private var activeGETSessionTask: URLSessionDataTask? + private struct RequestInfo: Decodable { + let id: ID? + let method: String? + } + + private struct SSEOutcome { + let hadData: Bool + let lastEventID: String? + let retryInterval: Int? + let completed: Bool + } + /// Creates a new HTTP transport client with the specified endpoint /// /// - Parameters: @@ -271,7 +283,10 @@ public actor HTTPClientTransport: Transport { try await processResponse(response: response, data: responseData) #else let (responseStream, response) = try await session.bytes(for: request) - try await processResponse(response: response, stream: responseStream) + let requestInfo = try? JSONDecoder().decode(RequestInfo.self, from: data) + let requestID = requestInfo?.method == nil ? nil : requestInfo?.id + try await processResponse( + response: response, stream: responseStream, requestID: requestID) #endif return } catch let authError as HTTPAuthenticationChallengeError { @@ -335,7 +350,9 @@ public actor HTTPClientTransport: Transport { } } #else - private func processResponse(response: URLResponse, stream: URLSession.AsyncBytes) + private func processResponse( + response: URLResponse, stream: URLSession.AsyncBytes, requestID: ID? + ) async throws { guard let httpResponse = response as? HTTPURLResponse else { @@ -358,9 +375,19 @@ public actor HTTPClientTransport: Transport { if contentType.contains(ContentType.sse) { logger.trace("Received SSE response, processing in streaming task") - let hadData = try await self.processSSE(stream) + let outcome = try await self.processSSE( + stream, recordsGETEventID: false, expectedResponseID: requestID) - if !hadData { + if let requestID, !outcome.completed { + guard let lastEventID = outcome.lastEventID else { + throw MCPError.connectionClosed + } + try await resumePOSTStream( + after: lastEventID, retryInterval: outcome.retryInterval, for: requestID) + return + } + + if !outcome.hadData { logger.debug("POST SSE stream closed without data, triggering GET reconnection") self.activeGETSessionTask?.cancel() } @@ -599,9 +626,9 @@ public actor HTTPClientTransport: Transport { request.addValue(sessionID, forHTTPHeaderField: HTTPHeaderName.sessionID) } - if let lastEventID = lastEventID { - request.addValue(lastEventID, forHTTPHeaderField: HTTPHeaderName.lastEventID) - logger.info("→ Resuming SSE stream with Last-Event-ID", metadata: ["lastEventID": "\(lastEventID)"]) + if let lastGETEventID = lastGETEventID { + request.addValue(lastGETEventID, forHTTPHeaderField: HTTPHeaderName.lastEventID) + logger.info("→ Resuming SSE stream with Last-Event-ID", metadata: ["lastEventID": "\(lastGETEventID)"]) } else { logger.info("→ Connecting to SSE stream (no last event ID to resume from)") } @@ -638,49 +665,124 @@ public actor HTTPClientTransport: Transport { } defer { self.activeGETSessionTask = nil } - try await self.processSSE(stream) + try await self.processSSE(stream, recordsGETEventID: true) + } + + private func resumePOSTStream( + after eventID: String, retryInterval: Int?, for requestID: ID + ) async throws { + var lastEventID = eventID + var delayMilliseconds = retryInterval ?? 3000 + for attempt in 0..<3 { + guard isConnected, !Task.isCancelled else { throw CancellationError() } + try await Task.sleep(for: .milliseconds(max(0, delayMilliseconds))) + + var request = URLRequest(url: endpoint) + request.httpMethod = "GET" + request.addValue(ContentType.sse, forHTTPHeaderField: HTTPHeaderName.accept) + request.addValue(lastEventID, forHTTPHeaderField: HTTPHeaderName.lastEventID) + if let protocolVersion { + request.addValue(protocolVersion, forHTTPHeaderField: HTTPHeaderName.protocolVersion) + } + if let sessionID { + request.addValue(sessionID, forHTTPHeaderField: HTTPHeaderName.sessionID) + } + if let authValue = authorizer?.authorizationHeader(for: endpoint) { + request.setValue(authValue, forHTTPHeaderField: HTTPHeaderName.authorization) + } + request = requestModifier(request) + + do { + let (stream, response) = try await session.bytes(for: request) + guard let httpResponse = response as? HTTPURLResponse, + httpResponse.statusCode == 200, + httpResponse.value(forHTTPHeaderField: HTTPHeaderName.contentType)? + .contains(ContentType.sse) == true + else { throw MCPError.internalError("Unable to resume POST SSE stream") } + + let outcome = try await processSSE( + stream, recordsGETEventID: false, expectedResponseID: requestID) + if outcome.completed { return } + if let eventID = outcome.lastEventID { lastEventID = eventID } + if let retryInterval = outcome.retryInterval { + delayMilliseconds = retryInterval + } + } catch { + if Task.isCancelled || attempt == 2 { throw error } + logger.debug("POST SSE resume failed; retrying: \(error)") + } + } + throw MCPError.connectionClosed } @discardableResult - private func processSSE(_ stream: URLSession.AsyncBytes) async throws -> Bool { + private func processSSE( + _ stream: URLSession.AsyncBytes, + recordsGETEventID: Bool, + expectedResponseID: ID? = nil + ) async throws -> SSEOutcome { logger.debug("đŸ“Ĩ Starting SSE event processing") var eventCount = 0 var hadDataEvent = false + var lastEventID: String? + var streamRetryInterval: Int? - for try await event in stream.events { - eventCount += 1 + do { + for try await event in stream.events { + eventCount += 1 - if Task.isCancelled { - logger.debug("⏚ī¸ SSE processing cancelled", metadata: ["eventsProcessed": "\(eventCount)"]) - break - } + if Task.isCancelled { + logger.debug("⏚ī¸ SSE processing cancelled", metadata: ["eventsProcessed": "\(eventCount)"]) + break + } - logger.trace( - "SSE event received", - metadata: [ - "type": "\(event.event ?? "message")", - "id": "\(event.id ?? "none")", - ] - ) + logger.trace( + "SSE event received", + metadata: [ + "type": "\(event.event ?? "message")", + "id": "\(event.id ?? "none")", + ] + ) - if let eventID = event.id, !eventID.isEmpty { - self.lastEventID = eventID - logger.debug("Stored event ID for resumability", metadata: ["eventID": "\(eventID)"]) - } + if let eventID = event.id, !eventID.isEmpty { + if recordsGETEventID { self.lastGETEventID = eventID } + lastEventID = eventID + logger.debug("Stored event ID for resumability", metadata: ["eventID": "\(eventID)"]) + } - if let retry = event.retry { - self.retryInterval = retry - logger.debug("SSE retry interval updated", metadata: ["retryMs": "\(retry)"]) - } + if let retry = event.retry { + if recordsGETEventID { self.retryInterval = retry } + streamRetryInterval = retry + logger.debug("SSE retry interval updated", metadata: ["retryMs": "\(retry)"]) + } - if !event.data.isEmpty, let data = event.data.data(using: .utf8) { - hadDataEvent = true - messageContinuation.yield(data) + if !event.data.isEmpty, let data = event.data.data(using: .utf8) { + hadDataEvent = true + messageContinuation.yield(data) + if let expectedResponseID, + let message = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + message["result"] != nil || message["error"] != nil, + let idData = try? JSONSerialization.data( + withJSONObject: message["id"] ?? NSNull(), options: .fragmentsAllowed), + let responseID = try? JSONDecoder().decode(ID.self, from: idData), + responseID == expectedResponseID + { + stream.task.cancel() + return SSEOutcome( + hadData: hadDataEvent, lastEventID: lastEventID, + retryInterval: streamRetryInterval, completed: true) + } + } } + } catch { + if expectedResponseID == nil || lastEventID == nil { throw error } + logger.debug("POST SSE stream closed before its response: \(error)") } logger.debug("✓ SSE event stream completed", metadata: ["eventsProcessed": "\(eventCount)", "hadData": "\(hadDataEvent)"]) - return hadDataEvent + return SSEOutcome( + hadData: hadDataEvent, lastEventID: lastEventID, + retryInterval: streamRetryInterval, completed: false) } #endif } diff --git a/Tests/MCPTests/HTTPClientTransportTests.swift b/Tests/MCPTests/HTTPClientTransportTests.swift index bd8c38f7..ad58abf7 100644 --- a/Tests/MCPTests/HTTPClientTransportTests.swift +++ b/Tests/MCPTests/HTTPClientTransportTests.swift @@ -979,6 +979,182 @@ import Testing // Skip SSE tests on platforms that don't support streaming #if !canImport(FoundationNetworking) + @Test("POST event IDs do not resume the standalone GET stream", .httpClientTransportSetup) + func testPOSTEventIDDoesNotResumeGET() async throws { + actor Requests { + private var getCount = 0 + private var postCount = 0 + private var postCompleted = false + private var getAfterPOST: String?? + + func recordGET(_ eventID: String?) { + getCount += 1 + if postCompleted && getAfterPOST == nil { + getAfterPOST = .some(eventID) + } + } + + func recordPOST() -> Bool { + postCount += 1 + if postCount == 1 { return true } + postCompleted = true + return false + } + func hasReceivedGET() -> Bool { getCount > 0 } + func resumedGETEventID() -> String?? { getAfterPOST } + } + + let requests = Requests() + let configuration = URLSessionConfiguration.ephemeral + configuration.protocolClasses = [MockURLProtocol.self] + + await MockURLProtocol.requestHandlerStorage.setHandler { + [testEndpoint, requests] request in + let headers = ["Content-Type": "text/event-stream"] + let response = HTTPURLResponse( + url: testEndpoint, statusCode: 200, httpVersion: "HTTP/1.1", + headerFields: headers)! + + if request.httpMethod == "GET" { + await requests.recordGET(request.value(forHTTPHeaderField: "Last-Event-ID")) + return (response, Data("id: get-1\nretry: 10\ndata: {}\n\n".utf8)) + } + + if await requests.recordPOST() { + let initialResponse = HTTPURLResponse( + url: testEndpoint, statusCode: 200, httpVersion: "HTTP/1.1", + headerFields: [ + "Content-Type": "text/plain", + "Mcp-Session-Id": "test-session-123", + ])! + return (initialResponse, Data()) + } + return (response, Data( + "id: post-1\ndata: {}\n\nid: post-2\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}\n\n".utf8)) + } + + let transport = HTTPClientTransport( + endpoint: testEndpoint, + configuration: configuration, + streaming: true, + sseInitializationTimeout: 0 + ) + try await transport.connect() + try await transport.send(Data()) + + for _ in 0..<200 where !(await requests.hasReceivedGET()) { + try await Task.sleep(for: .milliseconds(10)) + } + #expect(await requests.hasReceivedGET()) + + try await transport.send(Data(#"{"jsonrpc":"2.0","method":"ping","id":1}"#.utf8)) + + for _ in 0..<200 where (await requests.resumedGETEventID()) == nil { + try await Task.sleep(for: .milliseconds(10)) + } + let resumedID = await requests.resumedGETEventID() + #expect(resumedID != nil) + #expect(resumedID == .some("get-1")) + await transport.disconnect() + } + + @Test("An interrupted POST SSE stream resumes with its own event ID", .httpClientTransportSetup) + func testPOSTSSEResumesWithItsOwnEventID() async throws { + actor Requests { + private(set) var resumedID: String? + func record(_ id: String?) { resumedID = id } + } + let requests = Requests() + let configuration = URLSessionConfiguration.ephemeral + configuration.protocolClasses = [MockURLProtocol.self] + await MockURLProtocol.requestHandlerStorage.setHandler { + [testEndpoint, requests] request in + let response = HTTPURLResponse( + url: testEndpoint, statusCode: 200, httpVersion: "HTTP/1.1", + headerFields: ["Content-Type": "text/event-stream"])! + if request.httpMethod == "GET" { + await requests.record(request.value(forHTTPHeaderField: "Last-Event-ID")) + return (response, Data( + "id: post-2\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}\n\n".utf8)) + } + return (response, Data("id: post-1\nretry: 10\ndata: {}\n\n".utf8)) + } + + let transport = HTTPClientTransport( + endpoint: testEndpoint, configuration: configuration, streaming: false) + try await transport.connect() + try await transport.send(Data(#"{"jsonrpc":"2.0","method":"ping","id":1}"#.utf8)) + #expect(await requests.resumedID == "post-1") + + var iterator = await transport.receive().makeAsyncIterator() + #expect(try await iterator.next() == Data("{}".utf8)) + #expect(try await iterator.next() == Data( + #"{"jsonrpc":"2.0","id":1,"result":{}}"#.utf8)) + await transport.disconnect() + } + + @Test("Concurrent POST streams resume independently", .httpClientTransportSetup) + func testConcurrentPOSTSSEResumption() async throws { + actor Requests { + private(set) var resumedIDs: Set = [] + func record(_ id: String) { resumedIDs.insert(id) } + } + let requests = Requests() + let configuration = URLSessionConfiguration.ephemeral + configuration.protocolClasses = [MockURLProtocol.self] + await MockURLProtocol.requestHandlerStorage.setHandler { + [testEndpoint, requests] request in + let response = HTTPURLResponse( + url: testEndpoint, statusCode: 200, httpVersion: "HTTP/1.1", + headerFields: ["Content-Type": "text/event-stream"])! + if request.httpMethod == "GET" { + let id = try #require(request.value(forHTTPHeaderField: "Last-Event-ID")) + await requests.record(id) + let requestID = id == "post-a" ? 1 : 2 + return (response, Data( + "data: {\"jsonrpc\":\"2.0\",\"id\":\(requestID),\"result\":{}}\n\n".utf8)) + } + let body = try #require(request.readBody()) + let requestID = try #require( + (JSONSerialization.jsonObject(with: body) as? [String: Any])?["id"] as? Int) + let id = requestID == 1 ? "post-a" : "post-b" + return (response, Data("id: \(id)\nretry: 10\ndata: {}\n\n".utf8)) + } + + let transport = HTTPClientTransport( + endpoint: testEndpoint, configuration: configuration, streaming: false) + try await transport.connect() + async let first: Void = transport.send(Data( + #"{"jsonrpc":"2.0","method":"ping","id":1}"#.utf8)) + async let second: Void = transport.send(Data( + #"{"jsonrpc":"2.0","method":"ping","id":2}"#.utf8)) + try await first + try await second + #expect(await requests.resumedIDs == ["post-a", "post-b"]) + await transport.disconnect() + } + + @Test("POST SSE without a response or event ID fails promptly", .httpClientTransportSetup) + func testNonresumablePOSTSSEFails() async throws { + let configuration = URLSessionConfiguration.ephemeral + configuration.protocolClasses = [MockURLProtocol.self] + await MockURLProtocol.requestHandlerStorage.setHandler { [testEndpoint] _ in + let response = HTTPURLResponse( + url: testEndpoint, statusCode: 200, httpVersion: "HTTP/1.1", + headerFields: ["Content-Type": "text/event-stream"])! + return (response, Data("data: {}\n\n".utf8)) + } + + let transport = HTTPClientTransport( + endpoint: testEndpoint, configuration: configuration, streaming: false) + try await transport.connect() + await #expect(throws: MCPError.self) { + try await transport.send(Data( + #"{"jsonrpc":"2.0","method":"ping","id":1}"#.utf8)) + } + await transport.disconnect() + } + @Test("Receive Server-Sent Event (SSE)", .httpClientTransportSetup) func testReceiveSSE() async throws { let configuration = URLSessionConfiguration.ephemeral