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

Properly propagate errors in NetworkTransport receive stream (#4)

Browse source 
Changes AsyncStream to AsyncThrowingStream to ensure network errors are correctly propagated to callers. Updates receive() to handle errors from the underlying message stream.
This commit is contained in:
Mattt 2025-03-04 06:05:41 -08:00 • committed by GitHub
commit b761d3ea6f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 12 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

19
Sources/MCP/Base/Transports.swift
View file

@ -180,8 +180,8 @@ public actor StdioTransport: Transport {
public nonisolated let logger: Logger
private var isConnected = false
private let messageStream: AsyncStream<String>
private let messageContinuation: AsyncStream<String>.Continuation
private let messageStream: AsyncThrowingStream<String, Swift.Error>
private let messageContinuation: AsyncThrowingStream<String, Swift.Error>.Continuation
// Track connection state for continuations
private var connectionContinuationResumed = false
@ -196,8 +196,8 @@ public actor StdioTransport: Transport {
)
// Create message stream
var continuation: AsyncStream<String>.Continuation!
self.messageStream = AsyncStream { continuation = $0 }
var continuation: AsyncThrowingStream<String, Swift.Error>.Continuation!
self.messageStream = AsyncThrowingStream { continuation = $0 }
self.messageContinuation = continuation
}
@ -330,10 +330,14 @@ public actor StdioTransport: Transport {
public func receive() -> AsyncThrowingStream<String, Swift.Error> {
return AsyncThrowingStream { continuation in
Task {
for await message in messageStream {
continuation.yield(message)
do {
for try await message in messageStream {
continuation.yield(message)
}
continuation.finish()
} catch {
continuation.finish(throwing: error)
}
continuation.finish()
}
}
}
@ -361,6 +365,7 @@ public actor StdioTransport: Transport {
} catch {
if !Task.isCancelled {
logger.error("Receive error: \(error)")
messageContinuation.finish(throwing: error)
}
break
}