modelcontextprotocol/swift-sdk
Watch
1
0
Fork
You've already forked swift-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

V0.12.x (#215)

Browse source 
* fix #203 NetworkTransport race, with main-isolated variables (#206)

Co-authored-by: wti <wti@users.noreply.github.com>

* feat/auth: Authentication implementation (#205)

* feat: added authentication

* feat(auth): improvement to error handling

* feat(auth): updated errors and tests in OAuthClientRegistrar

* feat(auth): updated ci conformance package version

* feat(auth): bumped node version

* feat(auth): updated ci conformance step

* feat(auth): improvement to Linux support

* chore: added docc plugin

* chore: added spi yml

* chore: added documentation step to ci

* chore: increased timeout for test from 5 to 10m

---------

Co-authored-by: Paal Øye-Strømme <paal.o.eye@gmail.com>
Co-authored-by: wti <wti@users.noreply.github.com>
Co-authored-by: yehorsobko <yehorsobko@macpaw.com>
This commit is contained in:
Maksym Mova 2026-03-24 11:05:22 +02:00 • committed by GitHub
commit 6132fd4b5b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 10925 additions and 199 deletions
Download patch file Download diff file Expand all files Collapse all files

302
Sources/MCP/Base/Transports/HTTPClientTransport.swift
View file

@ -76,6 +76,9 @@ public actor HTTPClientTransport: Transport {
/// Closure to modify requests before they are sent
private let requestModifier: (URLRequest) -> URLRequest
/// Optional OAuth 2.1 authorizer.
private let authorizer: (any HTTPClientAuthorizer)?
private var isConnected = false
private let messageStream: AsyncThrowingStream<Data, Swift.Error>
private let messageContinuation: AsyncThrowingStream<Data, Swift.Error>.Continuation
@ -101,6 +104,7 @@ public actor HTTPClientTransport: Transport {
/// - streaming: Whether to enable SSE streaming mode (default: true)
/// - sseInitializationTimeout: Maximum time to wait for session ID before proceeding with SSE (default: 10 seconds)
/// - protocolVersion: The MCP protocol version to use (default: "2025-11-25")
/// - authorizer: Optional ``HTTPClientAuthorizer`` for automatic Bearer token acquisition and retries.
/// - requestModifier: Optional closure to customize requests before they are sent (default: no modification)
/// - logger: Optional logger instance for transport events
public init(
@ -109,6 +113,7 @@ public actor HTTPClientTransport: Transport {
streaming: Bool = true,
sseInitializationTimeout: TimeInterval = 10,
protocolVersion: String = Version.latest,
authorizer: (any HTTPClientAuthorizer)? = nil,
requestModifier: @escaping (URLRequest) -> URLRequest = { $0 },
logger: Logger? = nil
) {
@ -118,6 +123,7 @@ public actor HTTPClientTransport: Transport {
streaming: streaming,
sseInitializationTimeout: sseInitializationTimeout,
protocolVersion: protocolVersion,
authorizer: authorizer,
requestModifier: requestModifier,
logger: logger
)
@ -129,6 +135,7 @@ public actor HTTPClientTransport: Transport {
streaming: Bool = false,
sseInitializationTimeout: TimeInterval = 10,
protocolVersion: String = Version.latest,
authorizer: (any HTTPClientAuthorizer)? = nil,
requestModifier: @escaping (URLRequest) -> URLRequest = { $0 },
logger: Logger? = nil
) {
@ -138,6 +145,7 @@ public actor HTTPClientTransport: Transport {
self.sseInitializationTimeout = sseInitializationTimeout
self.protocolVersion = protocolVersion
self.requestModifier = requestModifier
self.authorizer = authorizer
// Create message stream
var continuation: AsyncThrowingStream<Data, Swift.Error>.Continuation!
@ -157,7 +165,6 @@ public actor HTTPClientTransport: Transport {
self.initialSessionIDSignalTask = Task {
await withCheckedContinuation { continuation in
self.initialSessionIDContinuation = continuation
// This task will suspend here until continuation.resume() is called
}
}
}
@ -166,7 +173,7 @@ public actor HTTPClientTransport: Transport {
private func triggerInitialSessionIDSignal() {
if let continuation = self.initialSessionIDContinuation {
continuation.resume()
self.initialSessionIDContinuation = nil // Consume the continuation
self.initialSessionIDContinuation = nil
logger.debug("✓ Initial session ID signal triggered for SSE task")
} else {
logger.debug("✗ No continuation to trigger - signal already consumed or SSE task not waiting")
@ -174,19 +181,13 @@ public actor HTTPClientTransport: Transport {
}
/// Establishes connection with the transport
///
/// This prepares the transport for communication and sets up SSE streaming
/// if streaming mode is enabled. The actual HTTP connection happens with the
/// first message sent.
public func connect() async throws {
guard !isConnected else { return }
isConnected = true
// Setup initial session ID signal
setupInitialSessionIDSignal()
if streaming {
// Start listening to server events
streamingTask = Task { await startListeningForServerEvents() }
}
@ -194,27 +195,18 @@ public actor HTTPClientTransport: Transport {
}
/// Disconnects from the transport
///
/// This terminates any active connections, cancels the streaming task,
/// and releases any resources being used by the transport.
public func disconnect() async {
guard isConnected else { return }
isConnected = false
// Cancel streaming task if active
streamingTask?.cancel()
streamingTask = nil
// Cancel any in-progress requests
session.invalidateAndCancel()
// Clean up message stream
messageContinuation.finish()
// Cancel the initial session ID signal task if active
initialSessionIDSignalTask?.cancel()
initialSessionIDSignalTask = nil
// Resume the continuation if it's still pending to avoid leaks
initialSessionIDContinuation?.resume()
initialSessionIDContinuation = nil
@ -222,77 +214,108 @@ public actor HTTPClientTransport: Transport {
}
/// Updates the protocol version used for `MCP-Protocol-Version` headers on subsequent requests.
///
/// Call this after lifecycle negotiation if the server responds with a different version.
/// - Parameter version: The negotiated protocol version.
public func updateNegotiatedProtocolVersion(_ version: String) {
self.protocolVersion = version
}
/// Sends data through an HTTP POST request
///
/// This sends a JSON-RPC message to the server via HTTP POST and processes
/// the response according to the MCP Streamable HTTP specification. It handles:
///
/// - Adding appropriate Accept headers for both JSON and SSE
/// - Including the MCP-Protocol-Version header as required by the specification
/// - Including the session ID in requests if one has been established
/// - Processing different response types (JSON vs SSE)
/// - Handling HTTP error codes according to the specification
///
/// - Parameter data: The JSON-RPC message to send
/// - Throws: MCPError for transport failures or server errors
public func send(_ data: Data) async throws {
guard isConnected else {
throw MCPError.internalError("Transport not connected")
}
var request = URLRequest(url: endpoint)
request.httpMethod = "POST"
request.addValue("application/json, text/event-stream", forHTTPHeaderField: "Accept")
request.addValue("application/json", forHTTPHeaderField: "Content-Type")
request.httpBody = data
// Add protocol version header (required by MCP specification 2025-11-25)
if let protocolVersion = protocolVersion {
request.addValue(protocolVersion, forHTTPHeaderField: "MCP-Protocol-Version")
if let authorizer {
do {
try authorizer.validateEndpointSecurity(for: endpoint)
} catch {
throw MCPError.internalError(
"Authorization flow failed: \(error.localizedDescription)"
)
}
}
// Add session ID if available
if let sessionID = sessionID {
request.addValue(sessionID, forHTTPHeaderField: "MCP-Session-Id")
if let authorizer {
try? await authorizer.prepareAuthorization(for: endpoint, session: session)
}
// Apply request modifier
request = requestModifier(request)
var attempts = 0
let operationKey = jsonRPCOperationKey(from: data)
#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
while true {
var request = URLRequest(url: endpoint)
request.httpMethod = "POST"
request.addValue(
"\(ContentType.json), \(ContentType.sse)",
forHTTPHeaderField: HTTPHeaderName.accept
)
request.addValue(ContentType.json, forHTTPHeaderField: HTTPHeaderName.contentType)
request.httpBody = data
if let protocolVersion = protocolVersion {
request.addValue(protocolVersion, forHTTPHeaderField: HTTPHeaderName.protocolVersion)
}
if let sessionID = sessionID {
request.addValue(sessionID, forHTTPHeaderField: HTTPHeaderName.sessionID)
}
if let authValue = authorizer?.authorizationHeader(for: endpoint) {
request.setValue(authValue, forHTTPHeaderField: HTTPHeaderName.authorization)
}
request = requestModifier(request)
do {
#if os(Linux)
let (responseData, response) = try await session.data(for: request)
try await processResponse(response: response, data: responseData)
#else
let (responseStream, response) = try await session.bytes(for: request)
try await processResponse(response: response, stream: responseStream)
#endif
return
} catch let authError as HTTPAuthenticationChallengeError {
guard let authorizer else {
throw mapAuthenticationChallengeError(authError)
}
let handled: Bool
do {
handled = try await authorizer.handleChallenge(
statusCode: authError.statusCode,
headers: authError.headers,
endpoint: endpoint,
operationKey: operationKey,
session: session
)
} catch {
throw MCPError.internalError(
"Authorization flow failed: \(error.localizedDescription)")
}
attempts += 1
if handled, attempts < authorizer.maxAuthorizationAttempts {
continue
}
throw mapAuthenticationChallengeError(authError)
}
}
}
#if os(Linux)
// 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
let contentType = httpResponse.value(forHTTPHeaderField: "Content-Type") ?? ""
let contentType = httpResponse.value(forHTTPHeaderField: HTTPHeaderName.contentType) ?? ""
// Extract session ID if present
if let newSessionID = httpResponse.value(forHTTPHeaderField: "MCP-Session-Id") {
if let newSessionID = httpResponse.value(forHTTPHeaderField: HTTPHeaderName.sessionID) {
let wasSessionIDNil = (self.sessionID == nil)
self.sessionID = newSessionID
if wasSessionIDNil {
// Trigger signal on first session ID
triggerInitialSessionIDSignal()
}
logger.debug("Session ID received", metadata: ["sessionID": "\(newSessionID)"])
@ -301,11 +324,10 @@ public actor HTTPClientTransport: Transport {
try processHTTPResponse(httpResponse, contentType: contentType)
guard case 200..<300 = httpResponse.statusCode else { return }
// For JSON responses, yield the data
if contentType.contains("text/event-stream") {
if contentType.contains(ContentType.sse) {
logger.warning("SSE responses aren't fully supported on Linux")
messageContinuation.yield(data)
} else if contentType.contains("application/json") {
} else if contentType.contains(ContentType.json) {
logger.trace("Received JSON response", metadata: ["size": "\(data.count)"])
messageContinuation.yield(data)
} else {
@ -313,7 +335,6 @@ public actor HTTPClientTransport: Transport {
}
}
#else
// Process response with byte stream (macOS, iOS, etc.)
private func processResponse(response: URLResponse, stream: URLSession.AsyncBytes)
async throws
{
@ -321,15 +342,12 @@ public actor HTTPClientTransport: Transport {
throw MCPError.internalError("Invalid HTTP response")
}
// Process the response based on content type and status code
let contentType = httpResponse.value(forHTTPHeaderField: "Content-Type") ?? ""
let contentType = httpResponse.value(forHTTPHeaderField: HTTPHeaderName.contentType) ?? ""
// Extract session ID if present
if let newSessionID = httpResponse.value(forHTTPHeaderField: "MCP-Session-Id") {
if let newSessionID = httpResponse.value(forHTTPHeaderField: HTTPHeaderName.sessionID) {
let wasSessionIDNil = (self.sessionID == nil)
self.sessionID = newSessionID
if wasSessionIDNil {
// Trigger signal on first session ID
triggerInitialSessionIDSignal()
}
logger.debug("Session ID received", metadata: ["sessionID": "\(newSessionID)"])
@ -338,20 +356,15 @@ public actor HTTPClientTransport: Transport {
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
if contentType.contains(ContentType.sse) {
logger.trace("Received SSE response, processing in streaming task")
let hadData = try await self.processSSE(stream)
// If the POST SSE stream closed without delivering a JSON-RPC response,
// trigger GET reconnection so the server can deliver it there.
// This implements standard SSE reconnection behavior per the spec.
if !hadData {
logger.debug("POST SSE stream closed without data, triggering GET reconnection")
self.activeGETSessionTask?.cancel()
}
} else if contentType.contains("application/json") {
// For JSON responses, collect and deliver the data
} else if contentType.contains(ContentType.json) {
var buffer = Data()
for try await byte in stream {
buffer.append(byte)
@ -364,25 +377,27 @@ public actor HTTPClientTransport: Transport {
}
#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")
throw HTTPAuthenticationChallengeError(
statusCode: response.statusCode,
headers: responseHeaders(from: response)
)
case 403:
throw MCPError.internalError("Access forbidden")
throw HTTPAuthenticationChallengeError(
statusCode: response.statusCode,
headers: responseHeaders(from: response)
)
case 404:
// If we get a 404 with a session ID, it means our session is invalid
if sessionID != nil {
logger.warning("Session has expired")
sessionID = nil
@ -391,8 +406,6 @@ public actor HTTPClientTransport: Transport {
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")
@ -406,7 +419,6 @@ public actor HTTPClientTransport: Transport {
throw MCPError.internalError("Too many requests")
case 500..<600:
// Server error range
throw MCPError.internalError("Server error: \(response.statusCode)")
default:
@ -415,43 +427,55 @@ public actor HTTPClientTransport: Transport {
}
}
private func mapAuthenticationChallengeError(_ error: HTTPAuthenticationChallengeError) -> MCPError {
switch error.statusCode {
case 401:
return MCPError.internalError("Authentication required")
case 403:
return MCPError.internalError("Access forbidden")
default:
return MCPError.internalError("HTTP authorization error: \(error.statusCode)")
}
}
private func responseHeaders(from response: HTTPURLResponse) -> [String: String] {
var headers: [String: String] = [:]
for (key, value) in response.allHeaderFields {
guard let key = key as? String, let value = value as? String else { continue }
headers[key] = value
}
return headers
}
private func jsonRPCOperationKey(from data: Data) -> String? {
guard
let jsonObject = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
let method = jsonObject["method"] as? String
else {
return nil
}
let normalized = method.trimmingCharacters(in: .whitespacesAndNewlines)
return normalized.isEmpty ? nil : normalized
}
/// Receives data in an async sequence
///
/// This returns an AsyncThrowingStream that emits Data objects representing
/// each JSON-RPC message received from the server. This includes:
///
/// - Direct responses to client requests
/// - Server-initiated messages delivered via SSE streams
///
/// - Returns: An AsyncThrowingStream of Data objects
public func receive() -> AsyncThrowingStream<Data, Swift.Error> {
return messageStream
}
// MARK: - SSE
/// Starts listening for server events using SSE
///
/// This establishes a long-lived HTTP connection using Server-Sent Events (SSE)
/// to enable server-to-client push messaging. It handles:
///
/// - Waiting for session ID if needed
/// - Opening the SSE connection
/// - Automatic reconnection on connection drops
/// - Processing received events
private func startListeningForServerEvents() async {
#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 }
// Wait for session ID to be available before opening SSE stream
if self.sessionID == nil, let signalTask = self.initialSessionIDSignalTask {
logger.debug("⏳ Waiting for session ID to be set (timeout: \(self.sseInitializationTimeout)s)...")
@ -484,7 +508,6 @@ public actor HTTPClientTransport: Transport {
logger.debug("✓ Session ID already available, proceeding with SSE connection immediately")
}
// Retry loop for connection drops
var isFirstAttempt = true
var attemptCount = 0
@ -501,7 +524,6 @@ public actor HTTPClientTransport: Transport {
])
do {
// Wait for retry interval before reconnecting (except first attempt)
if !isFirstAttempt {
let delayMs = self.retryInterval
logger.debug("⏳ Waiting before SSE reconnection", metadata: ["retryMs": "\(delayMs)"])
@ -514,8 +536,6 @@ public actor HTTPClientTransport: Transport {
try await self.connectToEventStream()
// If connectToEventStream() returns without error,
// it means the stream closed gracefully - reconnect with retry interval
logger.info("🔌 SSE stream closed gracefully, will reconnect", metadata: [
"attempt": "\(attemptCount)",
"willRetryAfter": "\(self.retryInterval)ms"
@ -523,7 +543,6 @@ public actor HTTPClientTransport: Transport {
} catch {
if !Task.isCancelled {
logger.error("❌ SSE connection error (attempt #\(attemptCount)): \(error)")
// Error case - will also use retry interval on next iteration
} else {
logger.debug("⏹️ SSE task cancelled")
}
@ -545,50 +564,56 @@ public actor HTTPClientTransport: Transport {
}
#if !os(Linux)
/// Establishes an SSE connection to the server
///
/// This initiates a GET request to the server endpoint with appropriate
/// headers to establish an SSE stream according to the MCP specification.
/// Supports stream resumability via Last-Event-ID header.
///
/// - Throws: MCPError for connection failures or server errors
private func connectToEventStream() async throws {
guard isConnected else {
logger.debug("⚠️ Skipping connectToEventStream - transport not connected")
return
}
if let authorizer {
do {
try authorizer.validateEndpointSecurity(for: endpoint)
} catch {
throw MCPError.internalError(
"Authorization flow failed: \(error.localizedDescription)"
)
}
}
if let authorizer {
try? await authorizer.prepareAuthorization(for: endpoint, session: session)
}
logger.debug("🔌 Preparing SSE connection request")
var request = URLRequest(url: endpoint)
request.httpMethod = "GET"
request.addValue("text/event-stream", forHTTPHeaderField: "Accept")
request.addValue("no-cache", forHTTPHeaderField: "Cache-Control")
request.addValue(ContentType.sse, forHTTPHeaderField: HTTPHeaderName.accept)
request.addValue("no-cache", forHTTPHeaderField: HTTPHeaderName.cacheControl)
// Add protocol version header (required by MCP specification 2025-11-25)
if let protocolVersion = protocolVersion {
request.addValue(protocolVersion, forHTTPHeaderField: "MCP-Protocol-Version")
request.addValue(protocolVersion, forHTTPHeaderField: HTTPHeaderName.protocolVersion)
}
// Add session ID if available
if let sessionID = sessionID {
request.addValue(sessionID, forHTTPHeaderField: "MCP-Session-Id")
request.addValue(sessionID, forHTTPHeaderField: HTTPHeaderName.sessionID)
}
// Add last event ID for resumability (if available)
if let lastEventID = lastEventID {
request.addValue(lastEventID, forHTTPHeaderField: "Last-Event-ID")
request.addValue(lastEventID, forHTTPHeaderField: HTTPHeaderName.lastEventID)
logger.info("→ Resuming SSE stream with Last-Event-ID", metadata: ["lastEventID": "\(lastEventID)"])
} else {
logger.info("→ Connecting to SSE stream (no last event ID to resume from)")
}
// Apply request modifier
if let authValue = authorizer?.authorizationHeader(for: endpoint) {
request.setValue(authValue, forHTTPHeaderField: HTTPHeaderName.authorization)
}
request = requestModifier(request)
logger.debug("Starting SSE connection")
// Create URLSession task for SSE
let (stream, response) = try await session.bytes(for: request)
self.activeGETSessionTask = stream.task
@ -596,24 +621,17 @@ public actor HTTPClientTransport: Transport {
throw MCPError.internalError("Invalid HTTP response")
}
// Check response status
guard httpResponse.statusCode == 200 else {
// If the server returns 405 Method Not Allowed,
// it indicates that the server doesn't support SSE streaming.
// We should cancel the task instead of retrying the connection.
if httpResponse.statusCode == 405 {
self.streamingTask?.cancel()
}
throw MCPError.internalError("HTTP error: \(httpResponse.statusCode)")
}
// Extract session ID if present
if let newSessionID = httpResponse.value(forHTTPHeaderField: "MCP-Session-Id") {
if let newSessionID = httpResponse.value(forHTTPHeaderField: HTTPHeaderName.sessionID) {
let wasSessionIDNil = (self.sessionID == nil)
self.sessionID = newSessionID
if wasSessionIDNil {
// Trigger signal on first session ID, though this is unlikely to happen here
// as GET usually follows a POST that would have already set the session ID
triggerInitialSessionIDSignal()
}
logger.debug("Session ID received", metadata: ["sessionID": "\(newSessionID)"])
@ -623,14 +641,6 @@ public actor HTTPClientTransport: Transport {
try await self.processSSE(stream)
}
/// Processes an SSE byte stream, extracting events and delivering them.
///
/// This method processes Server-Sent Events according to the MCP specification,
/// including support for event IDs for resumability.
///
/// - Parameter stream: The URLSession.AsyncBytes stream to process
/// - Returns: `true` if any data events were received, `false` otherwise.
/// - Throws: Error for stream processing failures
@discardableResult
private func processSSE(_ stream: URLSession.AsyncBytes) async throws -> Bool {
logger.debug("📥 Starting SSE event processing")
@ -640,7 +650,6 @@ public actor HTTPClientTransport: Transport {
for try await event in stream.events {
eventCount += 1
// Check if task has been cancelled
if Task.isCancelled {
logger.debug("⏹️ SSE processing cancelled", metadata: ["eventsProcessed": "\(eventCount)"])
break
@ -654,19 +663,16 @@ public actor HTTPClientTransport: Transport {
]
)
// Store event ID for resumability support
if let eventID = event.id, !eventID.isEmpty {
self.lastEventID = eventID
logger.debug("Stored event ID for resumability", metadata: ["eventID": "\(eventID)"])
}
// Store retry interval if provided by server
if let retry = event.retry {
self.retryInterval = retry
logger.debug("SSE retry interval updated", metadata: ["retryMs": "\(retry)"])
}
// Convert the event data to Data and yield it to the message stream
if !event.data.isEmpty, let data = event.data.data(using: .utf8) {
hadDataEvent = true
messageContinuation.yield(data)