Use EventSource for SSE implementation (#96)

* Add package dependency for github.com/loopwork-ai/EventSource

* Use EventSource for SSE implementation

* Refactor response stream and content type handling

Co-authored-by: Stephen Tallent <stallent@mercury.io>

* Conditionalize use of streaming for supported platforms

* Refactor conditional platform handling

* Conditionalize code and package dependency on Linux

---------

Co-authored-by: Stephen Tallent <stallent@mercury.io>

Package.resolved

@@ -1,6 +1,15 @@
 {
-  "originHash" : "cce36cb33302c2f1c28458e19b8439f736fc28106e4c6ea95d7992c74594c242",
+  "originHash" : "08de61941b7919a65e36c0e34f8c1c41995469b86a39122158b75b4a68c4527d",
   "pins" : [
+    {
+      "identity" : "eventsource",
+      "kind" : "remoteSourceControl",
+      "location" : "https://github.com/loopwork-ai/eventsource.git",
+      "state" : {
+        "revision" : "e83f076811f32757305b8bf69ac92d05626ffdd7",
+        "version" : "1.1.0"
+      }
+    },
     {
       "identity" : "swift-log",
       "kind" : "remoteSourceControl",

Package.swift

@@ -3,6 +3,25 @@
 
 import PackageDescription
 
+// Base dependencies needed on all platforms
+var dependencies: [Package.Dependency] = [
+    .package(url: "https://github.com/apple/swift-system.git", from: "1.0.0"),
+    .package(url: "https://github.com/apple/swift-log.git", from: "1.5.0"),
+]
+
+// Target dependencies needed on all platforms
+var targetDependencies: [Target.Dependency] = [
+    .product(name: "SystemPackage", package: "swift-system"),
+    .product(name: "Logging", package: "swift-log"),
+]
+
+// Add EventSource only on Apple platforms (non-Linux)
+#if !os(Linux)
+    dependencies.append(
+        .package(url: "https://github.com/loopwork-ai/eventsource.git", from: "1.1.0"))
+    targetDependencies.append(.product(name: "EventSource", package: "eventsource"))
+#endif
+
 let package = Package(
     name: "mcp-swift-sdk",
     platforms: [
@@ -19,25 +38,15 @@ let package = Package(
             name: "MCP",
             targets: ["MCP"])
     ],
-    dependencies: [
+    dependencies: dependencies,
-        .package(url: "https://github.com/apple/swift-system.git", from: "1.0.0"),
-        .package(url: "https://github.com/apple/swift-log.git", from: "1.5.0"),
-    ],
     targets: [
         // Targets are the basic building blocks of a package, defining a module or a test suite.
         // Targets can depend on other targets in this package and products from dependencies.
         .target(
             name: "MCP",
-            dependencies: [
+            dependencies: targetDependencies),
-                .product(name: "SystemPackage", package: "swift-system"),
-                .product(name: "Logging", package: "swift-log"),
-            ]),
         .testTarget(
             name: "MCPTests",
-            dependencies: [
+            dependencies: ["MCP"] + targetDependencies),
-                "MCP",
-                .product(name: "SystemPackage", package: "swift-system"),
-                .product(name: "Logging", package: "swift-log"),
-            ]),
     ]
 )

Sources/MCP/Base/Transports/HTTPClientTransport.swift

@@ -1,6 +1,10 @@
 import Foundation
 import Logging
 
+#if !os(Linux)
+    import EventSource
+#endif
+
 #if canImport(FoundationNetworking)
     import FoundationNetworking
 #endif
@@ -11,7 +15,6 @@ public actor HTTPClientTransport: Actor, Transport {
     public private(set) var sessionID: String?
     private let streaming: Bool
     private var streamingTask: Task<Void, Never>?
-    private var lastEventID: String?
     public nonisolated let logger: Logger
 
     private var isConnected = false
@@ -103,36 +106,103 @@ public actor HTTPClientTransport: Actor, Transport {
             request.addValue(sessionID, forHTTPHeaderField: "Mcp-Session-Id")
         }
 
-        let (responseData, response) = try await session.data(for: request)
+        #if os(Linux)
+            // Linux implementation using data(for:) instead of bytes(for:)
+            let (responseData, response) = try await session.data(for: request)
+            try await processResponse(response: response, data: responseData)
+        #else
+            // macOS and other platforms with bytes(for:) support
+            let (responseStream, response) = try await session.bytes(for: request)
+            try await processResponse(response: response, stream: responseStream)
+        #endif
+    }
 
-        guard let httpResponse = response as? HTTPURLResponse else {
+    #if os(Linux)
-            throw MCPError.internalError("Invalid HTTP response")
+        // Process response with data payload (Linux)
-        }
+        private func processResponse(response: URLResponse, data: Data) async throws {
+            guard let httpResponse = response as? HTTPURLResponse else {
+                throw MCPError.internalError("Invalid HTTP response")
+            }
 
-        // Process the response based on content type and status code
+            // Process the response based on content type and status code
-        let contentType = httpResponse.value(forHTTPHeaderField: "Content-Type") ?? ""
+            let contentType = httpResponse.value(forHTTPHeaderField: "Content-Type") ?? ""
 
-        // Extract session ID if present
+            // Extract session ID if present
-        if let newSessionID = httpResponse.value(forHTTPHeaderField: "Mcp-Session-Id") {
+            if let newSessionID = httpResponse.value(forHTTPHeaderField: "Mcp-Session-Id") {
-            self.sessionID = newSessionID
+                self.sessionID = newSessionID
-            logger.debug("Session ID received", metadata: ["sessionID": "\(newSessionID)"])
+                logger.debug("Session ID received", metadata: ["sessionID": "\(newSessionID)"])
-        }
+            }
 
-        // Handle different response types
+            try processHTTPResponse(httpResponse, contentType: contentType)
-        switch httpResponse.statusCode {
+            guard case 200..<300 = httpResponse.statusCode else { return }
-        case 200, 201, 202:
+
-            // For SSE, the processing happens in the streaming task
+            // For JSON responses, yield the data
             if contentType.contains("text/event-stream") {
-                logger.debug("Received SSE response, processing in streaming task")
+                logger.warning("SSE responses aren't fully supported on Linux")
-                // The streaming is handled by the SSE task if active
+                messageContinuation.yield(data)
-                return
+            } else if contentType.contains("application/json") {
+                logger.debug("Received JSON response", metadata: ["size": "\(data.count)"])
+                messageContinuation.yield(data)
+            } else {
+                logger.warning("Unexpected content type: \(contentType)")
+            }
+        }
+    #else
+        // Process response with byte stream (macOS, iOS, etc.)
+        private func processResponse(response: URLResponse, stream: URLSession.AsyncBytes)
+            async throws
+        {
+            guard let httpResponse = response as? HTTPURLResponse else {
+                throw MCPError.internalError("Invalid HTTP response")
             }
 
-            // For JSON responses, deliver the data directly
+            // Process the response based on content type and status code
-            if contentType.contains("application/json") && !responseData.isEmpty {
+            let contentType = httpResponse.value(forHTTPHeaderField: "Content-Type") ?? ""
-                logger.debug("Received JSON response", metadata: ["size": "\(responseData.count)"])
+
-                messageContinuation.yield(responseData)
+            // Extract session ID if present
+            if let newSessionID = httpResponse.value(forHTTPHeaderField: "Mcp-Session-Id") {
+                self.sessionID = newSessionID
+                logger.debug("Session ID received", metadata: ["sessionID": "\(newSessionID)"])
             }
+
+            try processHTTPResponse(httpResponse, contentType: contentType)
+            guard case 200..<300 = httpResponse.statusCode else { return }
+
+            if contentType.contains("text/event-stream") {
+                // For SSE, processing happens via the stream
+                logger.debug("Received SSE response, processing in streaming task")
+                try await self.processSSE(stream)
+            } else if contentType.contains("application/json") {
+                // For JSON responses, collect and deliver the data
+                var buffer = Data()
+                for try await byte in stream {
+                    buffer.append(byte)
+                }
+                logger.debug("Received JSON response", metadata: ["size": "\(buffer.count)"])
+                messageContinuation.yield(buffer)
+            } else {
+                logger.warning("Unexpected content type: \(contentType)")
+            }
+        }
+    #endif
+
+    // Common HTTP response handling for all platforms
+    private func processHTTPResponse(_ response: HTTPURLResponse, contentType: String) throws {
+        // Handle status codes according to HTTP semantics
+        switch response.statusCode {
+        case 200..<300:
+            // Success range - these are handled by the platform-specific code
+            return
+
+        case 400:
+            throw MCPError.internalError("Bad request")
+
+        case 401:
+            throw MCPError.internalError("Authentication required")
+
+        case 403:
+            throw MCPError.internalError("Access forbidden")
+
         case 404:
             // If we get a 404 with a session ID, it means our session is invalid
             if sessionID != nil {
@@ -141,8 +211,29 @@ public actor HTTPClientTransport: Actor, Transport {
                 throw MCPError.internalError("Session expired")
             }
             throw MCPError.internalError("Endpoint not found")
+
+        case 405:
+            // If we get a 405, it means the server does not support the requested method
+            // If streaming was requested, we should cancel the streaming task
+            if streaming {
+                self.streamingTask?.cancel()
+                throw MCPError.internalError("Server does not support streaming")
+            }
+            throw MCPError.internalError("Method not allowed")
+
+        case 408:
+            throw MCPError.internalError("Request timeout")
+
+        case 429:
+            throw MCPError.internalError("Too many requests")
+
+        case 500..<600:
+            // Server error range
+            throw MCPError.internalError("Server error: \(response.statusCode)")
+
         default:
-            throw MCPError.internalError("HTTP error: \(httpResponse.statusCode)")
+            throw MCPError.internalError(
+                "Unexpected HTTP response: \(response.statusCode) (\(contentType))")
         }
     }
 
@@ -155,27 +246,33 @@ public actor HTTPClientTransport: Actor, Transport {
 
     /// Starts listening for server events using SSE
     private func startListeningForServerEvents() async {
-        guard isConnected else { return }
+        #if os(Linux)
+            // SSE is not fully supported on Linux
+            if streaming {
+                logger.warning(
+                    "SSE streaming was requested but is not fully supported on Linux. SSE connection will not be attempted."
+                )
+            }
+        #else
+            // This is the original code for platforms that support SSE
+            guard isConnected else { return }
 
-        // Retry loop for connection drops
+            // Retry loop for connection drops
-        while isConnected && !Task.isCancelled {
+            while isConnected && !Task.isCancelled {
-            do {
+                do {
-                try await connectToEventStream()
+                    try await connectToEventStream()
-            } catch {
+                } catch {
-                if !Task.isCancelled {
+                    if !Task.isCancelled {
-                    logger.error("SSE connection error: \(error)")
+                        logger.error("SSE connection error: \(error)")
-                    // Wait before retrying
+                        // Wait before retrying
-                    try? await Task.sleep(nanoseconds: 1_000_000_000)  // 1 second
+                        try? await Task.sleep(for: .seconds(1))
+                    }
                 }
             }
-        }
+        #endif
     }
 
-    #if canImport(FoundationNetworking)
+    #if !os(Linux)
-        private func connectToEventStream() async throws {
-            logger.warning("SSE is not supported on this platform")
-        }
-    #else
         /// Establishes an SSE connection to the server
         private func connectToEventStream() async throws {
             guard isConnected else { return }
@@ -183,17 +280,13 @@ public actor HTTPClientTransport: Actor, Transport {
             var request = URLRequest(url: endpoint)
             request.httpMethod = "GET"
             request.addValue("text/event-stream", forHTTPHeaderField: "Accept")
+            request.addValue("no-cache", forHTTPHeaderField: "Cache-Control")
 
             // Add session ID if available
             if let sessionID = sessionID {
                 request.addValue(sessionID, forHTTPHeaderField: "Mcp-Session-Id")
             }
 
-            // Add Last-Event-ID header for resumability if available
-            if let lastEventID = lastEventID {
-                request.addValue(lastEventID, forHTTPHeaderField: "Last-Event-ID")
-            }
-
             logger.debug("Starting SSE connection")
 
             // Create URLSession task for SSE
@@ -217,95 +310,34 @@ public actor HTTPClientTransport: Actor, Transport {
             // Extract session ID if present
             if let newSessionID = httpResponse.value(forHTTPHeaderField: "Mcp-Session-Id") {
                 self.sessionID = newSessionID
+                logger.debug("Session ID received", metadata: ["sessionID": "\(newSessionID)"])
             }
 
-            // Process the SSE stream
+            try await self.processSSE(stream)
-            var buffer = ""
+        }
-            var eventType = ""
-            var eventID: String?
-            var eventData = ""
 
-            for try await byte in stream {
+        private func processSSE(_ stream: URLSession.AsyncBytes) async throws {
-                if Task.isCancelled { break }
+            do {
+                for try await event in stream.events {
+                    // Check if task has been cancelled
+                    if Task.isCancelled { break }
 
-                guard let char = String(bytes: [byte], encoding: .utf8) else { continue }
+                    logger.debug(
-                buffer.append(char)
+                        "SSE event received",
+                        metadata: [
+                            "type": "\(event.event ?? "message")",
+                            "id": "\(event.id ?? "none")",
+                        ]
+                    )
 
-                // Process complete lines
+                    // Convert the event data to Data and yield it to the message stream
-                while let newlineIndex = buffer.utf8.firstIndex(where: { $0 == 10 }) {
+                    if !event.data.isEmpty, let data = event.data.data(using: .utf8) {
-                    var line = buffer[..<newlineIndex]
+                        messageContinuation.yield(data)
-                    if line.hasSuffix("\r") {
-                        line = line.dropLast()
-                    }
-
-                    buffer = String(buffer[buffer.index(after: newlineIndex)...])
-
-                    // Empty line marks the end of an event
-                    if line.isEmpty {
-                        if !eventData.isEmpty {
-                            // Process the event
-                            if eventType == "id" {
-                                lastEventID = eventID
-                            } else {
-                                // Default event type is "message" if not specified
-                                if let data = eventData.data(using: .utf8) {
-                                    logger.debug(
-                                        "SSE event received",
-                                        metadata: [
-                                            "type": "\(eventType.isEmpty ? "message" : eventType)",
-                                            "id": "\(eventID ?? "none")",
-                                        ])
-                                    messageContinuation.yield(data)
-                                }
-                            }
-
-                            // Reset for next event
-                            eventType = ""
-                            eventData = ""
-                        }
-                        continue
-                    }
-
-                    if line.hasSuffix("\r") {
-                        line = line.dropLast()
-                    }
-
-                    // Lines starting with ":" are comments
-                    if line.hasPrefix(":") { continue }
-
-                    // Parse field: value format
-                    if let colonIndex = line.firstIndex(of: ":") {
-                        let field = String(line[..<colonIndex])
-                        var value = String(line[line.index(after: colonIndex)...])
-
-                        // Trim leading space
-                        if value.hasPrefix(" ") {
-                            value = String(value.dropFirst())
-                        }
-
-                        // Process based on field
-                        switch field {
-                        case "event":
-                            eventType = value
-                        case "data":
-                            if !eventData.isEmpty {
-                                eventData.append("\n")
-                            }
-                            eventData.append(value)
-                        case "id":
-                            if !value.contains("\0") {  // ID must not contain NULL
-                                eventID = value
-                                lastEventID = value
-                            }
-                        case "retry":
-                            // Retry timing not implemented
-                            break
-                        default:
-                            // Unknown fields are ignored per SSE spec
-                            break
-                        }
                     }
                 }
+            } catch {
+                logger.error("Error processing SSE events: \(error)")
+                throw error
             }
         }
     #endif

Tests/MCPTests/HTTPClientTransportTests.swift

@@ -341,7 +341,7 @@ import Testing
                     Issue.record("Expected MCPError.internalError, got \(error)")
                     throw error
                 }
-                #expect(message?.contains("HTTP error: 500") ?? false)
+                #expect(message?.contains("Server error: 500") ?? false)
             } catch {
                 Issue.record("Expected MCPError, got \(error)")
                 throw error