Add support for Linux with glibc (#61)
* Conditionally import Glibc on Linux for fcntl calls * Extract transport implementations into separate files * Fix warning in test * Add ubuntu-latest to CI test matrix * Switch to vapor/swiftly-action for setup step * Configure Prettier extension to format GitHub Actions workflows * Setup with different actions depending on platform * Rename step * Rename job * Update README
This commit is contained in:
Mattt
GitHub
parent
20d4fe6f5a
commit
1e61f205a1
8 changed files with 458 additions and 412 deletions
39
.github/workflows/ci.yml
vendored
39
.github/workflows/ci.yml
vendored
|
|
@ -2,9 +2,9 @@ name: CI
|
|||
|
||||
on:
|
||||
push:
|
||||
branches: [ "main" ]
|
||||
branches: ["main"]
|
||||
pull_request:
|
||||
branches: [ "main" ]
|
||||
branches: ["main"]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
|
@ -12,21 +12,32 @@ permissions:
|
|||
|
||||
jobs:
|
||||
test:
|
||||
runs-on: macos-latest
|
||||
|
||||
strategy:
|
||||
matrix:
|
||||
os: [macos-latest, ubuntu-latest]
|
||||
swift-version:
|
||||
- ^6
|
||||
- 6.0.3
|
||||
- 6.1.0
|
||||
|
||||
name: Build and Test (Swift ${{ matrix.swift-version }})
|
||||
runs-on: ${{ matrix.os }}
|
||||
name: Test (${{ matrix.os }}, Swift ${{ matrix.swift-version }})
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: swift-actions/setup-swift@v2
|
||||
with:
|
||||
swift-version: ${{ matrix.swift-version }}
|
||||
- name: Build
|
||||
run: swift build -v
|
||||
- name: Run tests
|
||||
run: swift test -v
|
||||
- uses: actions/checkout@v4
|
||||
|
||||
- name: Setup Swift on Linux
|
||||
if: matrix.os == 'ubuntu-latest'
|
||||
uses: vapor/swiftly-action@v0.2
|
||||
with:
|
||||
toolchain: ${{ matrix.swift-version }}
|
||||
- name: Setup Swift on macOS
|
||||
if: matrix.os == 'macos-latest'
|
||||
uses: swift-actions/setup-swift@v2
|
||||
with:
|
||||
swift-version: ${{ matrix.swift-version }}
|
||||
|
||||
- name: Build
|
||||
run: swift build -v
|
||||
|
||||
- name: Run tests
|
||||
run: swift test -v
|
||||
|
|
|
|||
2
.vscode/extensions.json
vendored
2
.vscode/extensions.json
vendored
|
|
@ -1,3 +1,3 @@
|
|||
{
|
||||
"recommendations": ["swiftlang.swift-vscode"]
|
||||
"recommendations": ["swiftlang.swift-vscode", "esbenp.prettier-vscode"]
|
||||
}
|
||||
|
|
|
|||
3
.vscode/settings.json
vendored
3
.vscode/settings.json
vendored
|
|
@ -4,5 +4,8 @@
|
|||
"editor.codeActionsOnSave": {
|
||||
"source.fixAll": "explicit",
|
||||
"source.organizeImports": "explicit"
|
||||
},
|
||||
"[github-actions-workflow]": {
|
||||
"editor.defaultFormatter": "esbenp.prettier-vscode"
|
||||
}
|
||||
}
|
||||
|
|
|
|||
20
README.md
20
README.md
|
|
@ -4,12 +4,20 @@ Swift implementation of the [Model Context Protocol][mcp] (MCP).
|
|||
|
||||
## Requirements
|
||||
|
||||
- Swift 6.0+ / Xcode 16+
|
||||
- macOS 13.0+
|
||||
- iOS / Mac Catalyst 16.0+
|
||||
- watchOS 9.0+
|
||||
- tvOS 16.0+
|
||||
- visionOS 1.0+
|
||||
- Swift 6.0+ (Xcode 16+)
|
||||
|
||||
### Supported Platforms
|
||||
|
||||
| Platform | Minimum Version |
|
||||
|----------|----------------|
|
||||
| macOS | 13.0+ |
|
||||
| iOS / Mac Catalyst | 16.0+ |
|
||||
| watchOS | 9.0+ |
|
||||
| tvOS | 16.0+ |
|
||||
| visionOS | 1.0+ |
|
||||
| Linux | ✓ [^1] |
|
||||
|
||||
[^1]: Linux support requires glibc-based distributions such as Ubuntu, Debian, Fedora, CentOS, or RHEL. Alpine Linux and other musl-based distributions are not supported.
|
||||
|
||||
## Installation
|
||||
|
||||
|
|
|
|||
|
|
@ -1,14 +1,7 @@
|
|||
import Darwin
|
||||
import Logging
|
||||
|
||||
import struct Foundation.Data
|
||||
|
||||
#if canImport(System)
|
||||
import System
|
||||
#else
|
||||
@preconcurrency import SystemPackage
|
||||
#endif
|
||||
|
||||
/// Protocol defining the transport layer for MCP communication
|
||||
public protocol Transport: Actor {
|
||||
var logger: Logger { get }
|
||||
|
|
@ -25,386 +18,3 @@ public protocol Transport: Actor {
|
|||
/// Receives data in an async sequence
|
||||
func receive() -> AsyncThrowingStream<Data, Swift.Error>
|
||||
}
|
||||
|
||||
/// Standard input/output transport implementation
|
||||
public actor StdioTransport: Transport {
|
||||
private let input: FileDescriptor
|
||||
private let output: FileDescriptor
|
||||
public nonisolated let logger: Logger
|
||||
|
||||
private var isConnected = false
|
||||
private let messageStream: AsyncStream<Data>
|
||||
private let messageContinuation: AsyncStream<Data>.Continuation
|
||||
|
||||
public init(
|
||||
input: FileDescriptor = FileDescriptor.standardInput,
|
||||
output: FileDescriptor = FileDescriptor.standardOutput,
|
||||
logger: Logger? = nil
|
||||
) {
|
||||
self.input = input
|
||||
self.output = output
|
||||
self.logger =
|
||||
logger
|
||||
?? Logger(
|
||||
label: "mcp.transport.stdio",
|
||||
factory: { _ in SwiftLogNoOpLogHandler() })
|
||||
|
||||
// Create message stream
|
||||
var continuation: AsyncStream<Data>.Continuation!
|
||||
self.messageStream = AsyncStream { continuation = $0 }
|
||||
self.messageContinuation = continuation
|
||||
}
|
||||
|
||||
public func connect() async throws {
|
||||
guard !isConnected else { return }
|
||||
|
||||
// Set non-blocking mode
|
||||
try setNonBlocking(fileDescriptor: input)
|
||||
try setNonBlocking(fileDescriptor: output)
|
||||
|
||||
isConnected = true
|
||||
logger.info("Transport connected successfully")
|
||||
|
||||
// Start reading loop in background
|
||||
Task {
|
||||
await readLoop()
|
||||
}
|
||||
}
|
||||
|
||||
private func setNonBlocking(fileDescriptor: FileDescriptor) throws {
|
||||
let flags = fcntl(fileDescriptor.rawValue, F_GETFL)
|
||||
guard flags >= 0 else {
|
||||
throw MCPError.transportError(Errno.badFileDescriptor)
|
||||
}
|
||||
let result = fcntl(fileDescriptor.rawValue, F_SETFL, flags | O_NONBLOCK)
|
||||
guard result >= 0 else {
|
||||
throw MCPError.transportError(Errno.badFileDescriptor)
|
||||
}
|
||||
}
|
||||
|
||||
private func readLoop() async {
|
||||
let bufferSize = 4096
|
||||
var buffer = [UInt8](repeating: 0, count: bufferSize)
|
||||
var pendingData = Data()
|
||||
|
||||
while isConnected && !Task.isCancelled {
|
||||
do {
|
||||
let bytesRead = try buffer.withUnsafeMutableBufferPointer { pointer in
|
||||
try input.read(into: UnsafeMutableRawBufferPointer(pointer))
|
||||
}
|
||||
|
||||
if bytesRead == 0 {
|
||||
logger.notice("EOF received")
|
||||
break
|
||||
}
|
||||
|
||||
pendingData.append(Data(buffer[..<bytesRead]))
|
||||
|
||||
// Process complete messages
|
||||
while let newlineIndex = pendingData.firstIndex(of: UInt8(ascii: "\n")) {
|
||||
let messageData = pendingData[..<newlineIndex]
|
||||
pendingData = pendingData[(newlineIndex + 1)...]
|
||||
|
||||
if !messageData.isEmpty {
|
||||
logger.debug("Message received", metadata: ["size": "\(messageData.count)"])
|
||||
messageContinuation.yield(Data(messageData))
|
||||
}
|
||||
}
|
||||
} catch let error where MCPError.isResourceTemporarilyUnavailable(error) {
|
||||
try? await Task.sleep(for: .milliseconds(10))
|
||||
continue
|
||||
} catch {
|
||||
if !Task.isCancelled {
|
||||
logger.error("Read error occurred", metadata: ["error": "\(error)"])
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
messageContinuation.finish()
|
||||
}
|
||||
|
||||
public func disconnect() async {
|
||||
guard isConnected else { return }
|
||||
isConnected = false
|
||||
messageContinuation.finish()
|
||||
logger.info("Transport disconnected")
|
||||
}
|
||||
|
||||
public func send(_ message: Data) async throws {
|
||||
guard isConnected else {
|
||||
throw MCPError.transportError(Errno.socketNotConnected)
|
||||
}
|
||||
|
||||
// Add newline as delimiter
|
||||
var messageWithNewline = message
|
||||
messageWithNewline.append(UInt8(ascii: "\n"))
|
||||
|
||||
var remaining = messageWithNewline
|
||||
while !remaining.isEmpty {
|
||||
do {
|
||||
let written = try remaining.withUnsafeBytes { buffer in
|
||||
try output.write(UnsafeRawBufferPointer(buffer))
|
||||
}
|
||||
if written > 0 {
|
||||
remaining = remaining.dropFirst(written)
|
||||
}
|
||||
} catch let error where MCPError.isResourceTemporarilyUnavailable(error) {
|
||||
try await Task.sleep(for: .milliseconds(10))
|
||||
continue
|
||||
} catch {
|
||||
throw MCPError.transportError(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public func receive() -> AsyncThrowingStream<Data, Swift.Error> {
|
||||
return AsyncThrowingStream { continuation in
|
||||
Task {
|
||||
for await message in messageStream {
|
||||
continuation.yield(message)
|
||||
}
|
||||
continuation.finish()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#if canImport(Network)
|
||||
import Network
|
||||
|
||||
/// Network connection based transport implementation
|
||||
public actor NetworkTransport: Transport {
|
||||
private let connection: NWConnection
|
||||
public nonisolated let logger: Logger
|
||||
|
||||
private var isConnected = false
|
||||
private let messageStream: AsyncThrowingStream<Data, Swift.Error>
|
||||
private let messageContinuation: AsyncThrowingStream<Data, Swift.Error>.Continuation
|
||||
|
||||
// Track connection state for continuations
|
||||
private var connectionContinuationResumed = false
|
||||
|
||||
public init(connection: NWConnection, logger: Logger? = nil) {
|
||||
self.connection = connection
|
||||
self.logger =
|
||||
logger
|
||||
?? Logger(
|
||||
label: "mcp.transport.network",
|
||||
factory: { _ in SwiftLogNoOpLogHandler() }
|
||||
)
|
||||
|
||||
// Create message stream
|
||||
var continuation: AsyncThrowingStream<Data, Swift.Error>.Continuation!
|
||||
self.messageStream = AsyncThrowingStream { continuation = $0 }
|
||||
self.messageContinuation = continuation
|
||||
}
|
||||
|
||||
/// Connects to the network transport
|
||||
public func connect() async throws {
|
||||
guard !isConnected else { return }
|
||||
|
||||
// Reset continuation state
|
||||
connectionContinuationResumed = false
|
||||
|
||||
// Wait for connection to be ready
|
||||
try await withCheckedThrowingContinuation {
|
||||
[weak self] (continuation: CheckedContinuation<Void, Swift.Error>) in
|
||||
guard let self = self else {
|
||||
continuation.resume(throwing: MCPError.internalError("Transport deallocated"))
|
||||
return
|
||||
}
|
||||
|
||||
connection.stateUpdateHandler = { [weak self] state in
|
||||
guard let self = self else { return }
|
||||
|
||||
Task { @MainActor in
|
||||
switch state {
|
||||
case .ready:
|
||||
await self.handleConnectionReady(continuation: continuation)
|
||||
case .failed(let error):
|
||||
await self.handleConnectionFailed(
|
||||
error: error, continuation: continuation)
|
||||
case .cancelled:
|
||||
await self.handleConnectionCancelled(continuation: continuation)
|
||||
default:
|
||||
// Wait for ready or failed state
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Start the connection if it's not already started
|
||||
if connection.state != .ready {
|
||||
connection.start(queue: .main)
|
||||
} else {
|
||||
Task { @MainActor in
|
||||
await self.handleConnectionReady(continuation: continuation)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func handleConnectionReady(continuation: CheckedContinuation<Void, Swift.Error>)
|
||||
async
|
||||
{
|
||||
if !connectionContinuationResumed {
|
||||
connectionContinuationResumed = true
|
||||
isConnected = true
|
||||
logger.info("Network transport connected successfully")
|
||||
continuation.resume()
|
||||
// Start the receive loop after connection is established
|
||||
Task { await self.receiveLoop() }
|
||||
}
|
||||
}
|
||||
|
||||
private func handleConnectionFailed(
|
||||
error: Swift.Error, continuation: CheckedContinuation<Void, Swift.Error>
|
||||
) async {
|
||||
if !connectionContinuationResumed {
|
||||
connectionContinuationResumed = true
|
||||
logger.error("Connection failed: \(error)")
|
||||
continuation.resume(throwing: error)
|
||||
}
|
||||
}
|
||||
|
||||
private func handleConnectionCancelled(continuation: CheckedContinuation<Void, Swift.Error>)
|
||||
async
|
||||
{
|
||||
if !connectionContinuationResumed {
|
||||
connectionContinuationResumed = true
|
||||
logger.warning("Connection cancelled")
|
||||
continuation.resume(throwing: MCPError.internalError("Connection cancelled"))
|
||||
}
|
||||
}
|
||||
|
||||
public func disconnect() async {
|
||||
guard isConnected else { return }
|
||||
isConnected = false
|
||||
connection.cancel()
|
||||
messageContinuation.finish()
|
||||
logger.info("Network transport disconnected")
|
||||
}
|
||||
|
||||
public func send(_ message: Data) async throws {
|
||||
guard isConnected else {
|
||||
throw MCPError.internalError("Transport not connected")
|
||||
}
|
||||
|
||||
// Add newline as delimiter
|
||||
var messageWithNewline = message
|
||||
messageWithNewline.append(UInt8(ascii: "\n"))
|
||||
|
||||
// Use a local actor-isolated variable to track continuation state
|
||||
var sendContinuationResumed = false
|
||||
|
||||
try await withCheckedThrowingContinuation {
|
||||
[weak self] (continuation: CheckedContinuation<Void, Swift.Error>) in
|
||||
guard let self = self else {
|
||||
continuation.resume(throwing: MCPError.internalError("Transport deallocated"))
|
||||
return
|
||||
}
|
||||
|
||||
connection.send(
|
||||
content: messageWithNewline,
|
||||
completion: .contentProcessed { [weak self] error in
|
||||
guard let self = self else { return }
|
||||
|
||||
Task { @MainActor in
|
||||
if !sendContinuationResumed {
|
||||
sendContinuationResumed = true
|
||||
if let error = error {
|
||||
self.logger.error("Send error: \(error)")
|
||||
continuation.resume(
|
||||
throwing: MCPError.internalError("Send error: \(error)"))
|
||||
} else {
|
||||
continuation.resume()
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
public func receive() -> AsyncThrowingStream<Data, Swift.Error> {
|
||||
return AsyncThrowingStream { continuation in
|
||||
Task {
|
||||
do {
|
||||
for try await message in messageStream {
|
||||
continuation.yield(message)
|
||||
}
|
||||
continuation.finish()
|
||||
} catch {
|
||||
continuation.finish(throwing: error)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func receiveLoop() async {
|
||||
var buffer = Data()
|
||||
|
||||
while isConnected && !Task.isCancelled {
|
||||
do {
|
||||
let newData = try await receiveData()
|
||||
buffer.append(newData)
|
||||
|
||||
// Process complete messages
|
||||
while let newlineIndex = buffer.firstIndex(of: UInt8(ascii: "\n")) {
|
||||
let messageData = buffer[..<newlineIndex]
|
||||
buffer = buffer[(newlineIndex + 1)...]
|
||||
|
||||
if !messageData.isEmpty {
|
||||
logger.debug(
|
||||
"Message received", metadata: ["size": "\(messageData.count)"])
|
||||
messageContinuation.yield(Data(messageData))
|
||||
}
|
||||
}
|
||||
} catch let error as NWError {
|
||||
if !Task.isCancelled {
|
||||
logger.error("Network error occurred", metadata: ["error": "\(error)"])
|
||||
messageContinuation.finish(throwing: MCPError.transportError(error))
|
||||
}
|
||||
break
|
||||
} catch {
|
||||
if !Task.isCancelled {
|
||||
logger.error("Receive error: \(error)")
|
||||
messageContinuation.finish(throwing: error)
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
messageContinuation.finish()
|
||||
}
|
||||
|
||||
private func receiveData() async throws -> Data {
|
||||
var receiveContinuationResumed = false
|
||||
|
||||
return try await withCheckedThrowingContinuation {
|
||||
[weak self] (continuation: CheckedContinuation<Data, Swift.Error>) in
|
||||
guard let self = self else {
|
||||
continuation.resume(throwing: MCPError.internalError("Transport deallocated"))
|
||||
return
|
||||
}
|
||||
|
||||
connection.receive(minimumIncompleteLength: 1, maximumLength: 65536) {
|
||||
content, _, _, error in
|
||||
Task { @MainActor in
|
||||
if !receiveContinuationResumed {
|
||||
receiveContinuationResumed = true
|
||||
if let error = error {
|
||||
continuation.resume(throwing: MCPError.transportError(error))
|
||||
} else if let content = content {
|
||||
continuation.resume(returning: content)
|
||||
} else {
|
||||
continuation.resume(
|
||||
throwing: MCPError.internalError("No data received"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
|
|
|||
242
Sources/MCP/Base/Transports/NetworkTransport.swift
Normal file
242
Sources/MCP/Base/Transports/NetworkTransport.swift
Normal file
|
|
@ -0,0 +1,242 @@
|
|||
import Logging
|
||||
|
||||
import struct Foundation.Data
|
||||
|
||||
#if canImport(Network)
|
||||
import Network
|
||||
|
||||
/// Network connection based transport implementation
|
||||
public actor NetworkTransport: Transport {
|
||||
private let connection: NWConnection
|
||||
public nonisolated let logger: Logger
|
||||
|
||||
private var isConnected = false
|
||||
private let messageStream: AsyncThrowingStream<Data, Swift.Error>
|
||||
private let messageContinuation: AsyncThrowingStream<Data, Swift.Error>.Continuation
|
||||
|
||||
// Track connection state for continuations
|
||||
private var connectionContinuationResumed = false
|
||||
|
||||
public init(connection: NWConnection, logger: Logger? = nil) {
|
||||
self.connection = connection
|
||||
self.logger =
|
||||
logger
|
||||
?? Logger(
|
||||
label: "mcp.transport.network",
|
||||
factory: { _ in SwiftLogNoOpLogHandler() }
|
||||
)
|
||||
|
||||
// Create message stream
|
||||
var continuation: AsyncThrowingStream<Data, Swift.Error>.Continuation!
|
||||
self.messageStream = AsyncThrowingStream { continuation = $0 }
|
||||
self.messageContinuation = continuation
|
||||
}
|
||||
|
||||
/// Connects to the network transport
|
||||
public func connect() async throws {
|
||||
guard !isConnected else { return }
|
||||
|
||||
// Reset continuation state
|
||||
connectionContinuationResumed = false
|
||||
|
||||
// Wait for connection to be ready
|
||||
try await withCheckedThrowingContinuation {
|
||||
[weak self] (continuation: CheckedContinuation<Void, Swift.Error>) in
|
||||
guard let self = self else {
|
||||
continuation.resume(throwing: MCPError.internalError("Transport deallocated"))
|
||||
return
|
||||
}
|
||||
|
||||
connection.stateUpdateHandler = { [weak self] state in
|
||||
guard let self = self else { return }
|
||||
|
||||
Task { @MainActor in
|
||||
switch state {
|
||||
case .ready:
|
||||
await self.handleConnectionReady(continuation: continuation)
|
||||
case .failed(let error):
|
||||
await self.handleConnectionFailed(
|
||||
error: error, continuation: continuation)
|
||||
case .cancelled:
|
||||
await self.handleConnectionCancelled(continuation: continuation)
|
||||
default:
|
||||
// Wait for ready or failed state
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Start the connection if it's not already started
|
||||
if connection.state != .ready {
|
||||
connection.start(queue: .main)
|
||||
} else {
|
||||
Task { @MainActor in
|
||||
await self.handleConnectionReady(continuation: continuation)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func handleConnectionReady(continuation: CheckedContinuation<Void, Swift.Error>)
|
||||
async
|
||||
{
|
||||
if !connectionContinuationResumed {
|
||||
connectionContinuationResumed = true
|
||||
isConnected = true
|
||||
logger.info("Network transport connected successfully")
|
||||
continuation.resume()
|
||||
// Start the receive loop after connection is established
|
||||
Task { await self.receiveLoop() }
|
||||
}
|
||||
}
|
||||
|
||||
private func handleConnectionFailed(
|
||||
error: Swift.Error, continuation: CheckedContinuation<Void, Swift.Error>
|
||||
) async {
|
||||
if !connectionContinuationResumed {
|
||||
connectionContinuationResumed = true
|
||||
logger.error("Connection failed: \(error)")
|
||||
continuation.resume(throwing: error)
|
||||
}
|
||||
}
|
||||
|
||||
private func handleConnectionCancelled(continuation: CheckedContinuation<Void, Swift.Error>)
|
||||
async
|
||||
{
|
||||
if !connectionContinuationResumed {
|
||||
connectionContinuationResumed = true
|
||||
logger.warning("Connection cancelled")
|
||||
continuation.resume(throwing: MCPError.internalError("Connection cancelled"))
|
||||
}
|
||||
}
|
||||
|
||||
public func disconnect() async {
|
||||
guard isConnected else { return }
|
||||
isConnected = false
|
||||
connection.cancel()
|
||||
messageContinuation.finish()
|
||||
logger.info("Network transport disconnected")
|
||||
}
|
||||
|
||||
public func send(_ message: Data) async throws {
|
||||
guard isConnected else {
|
||||
throw MCPError.internalError("Transport not connected")
|
||||
}
|
||||
|
||||
// Add newline as delimiter
|
||||
var messageWithNewline = message
|
||||
messageWithNewline.append(UInt8(ascii: "\n"))
|
||||
|
||||
// Use a local actor-isolated variable to track continuation state
|
||||
var sendContinuationResumed = false
|
||||
|
||||
try await withCheckedThrowingContinuation {
|
||||
[weak self] (continuation: CheckedContinuation<Void, Swift.Error>) in
|
||||
guard let self = self else {
|
||||
continuation.resume(throwing: MCPError.internalError("Transport deallocated"))
|
||||
return
|
||||
}
|
||||
|
||||
connection.send(
|
||||
content: messageWithNewline,
|
||||
completion: .contentProcessed { [weak self] error in
|
||||
guard let self = self else { return }
|
||||
|
||||
Task { @MainActor in
|
||||
if !sendContinuationResumed {
|
||||
sendContinuationResumed = true
|
||||
if let error = error {
|
||||
self.logger.error("Send error: \(error)")
|
||||
continuation.resume(
|
||||
throwing: MCPError.internalError("Send error: \(error)"))
|
||||
} else {
|
||||
continuation.resume()
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
public func receive() -> AsyncThrowingStream<Data, Swift.Error> {
|
||||
return AsyncThrowingStream { continuation in
|
||||
Task {
|
||||
do {
|
||||
for try await message in messageStream {
|
||||
continuation.yield(message)
|
||||
}
|
||||
continuation.finish()
|
||||
} catch {
|
||||
continuation.finish(throwing: error)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func receiveLoop() async {
|
||||
var buffer = Data()
|
||||
|
||||
while isConnected && !Task.isCancelled {
|
||||
do {
|
||||
let newData = try await receiveData()
|
||||
buffer.append(newData)
|
||||
|
||||
// Process complete messages
|
||||
while let newlineIndex = buffer.firstIndex(of: UInt8(ascii: "\n")) {
|
||||
let messageData = buffer[..<newlineIndex]
|
||||
buffer = buffer[(newlineIndex + 1)...]
|
||||
|
||||
if !messageData.isEmpty {
|
||||
logger.debug(
|
||||
"Message received", metadata: ["size": "\(messageData.count)"])
|
||||
messageContinuation.yield(Data(messageData))
|
||||
}
|
||||
}
|
||||
} catch let error as NWError {
|
||||
if !Task.isCancelled {
|
||||
logger.error("Network error occurred", metadata: ["error": "\(error)"])
|
||||
messageContinuation.finish(throwing: MCPError.transportError(error))
|
||||
}
|
||||
break
|
||||
} catch {
|
||||
if !Task.isCancelled {
|
||||
logger.error("Receive error: \(error)")
|
||||
messageContinuation.finish(throwing: error)
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
messageContinuation.finish()
|
||||
}
|
||||
|
||||
private func receiveData() async throws -> Data {
|
||||
var receiveContinuationResumed = false
|
||||
|
||||
return try await withCheckedThrowingContinuation {
|
||||
[weak self] (continuation: CheckedContinuation<Data, Swift.Error>) in
|
||||
guard let self = self else {
|
||||
continuation.resume(throwing: MCPError.internalError("Transport deallocated"))
|
||||
return
|
||||
}
|
||||
|
||||
connection.receive(minimumIncompleteLength: 1, maximumLength: 65536) {
|
||||
content, _, _, error in
|
||||
Task { @MainActor in
|
||||
if !receiveContinuationResumed {
|
||||
receiveContinuationResumed = true
|
||||
if let error = error {
|
||||
continuation.resume(throwing: MCPError.transportError(error))
|
||||
} else if let content = content {
|
||||
continuation.resume(returning: content)
|
||||
} else {
|
||||
continuation.resume(
|
||||
throwing: MCPError.internalError("No data received"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
#endif
|
||||
172
Sources/MCP/Base/Transports/StdioTransport.swift
Normal file
172
Sources/MCP/Base/Transports/StdioTransport.swift
Normal file
|
|
@ -0,0 +1,172 @@
|
|||
import Logging
|
||||
|
||||
import struct Foundation.Data
|
||||
|
||||
#if canImport(System)
|
||||
import System
|
||||
#else
|
||||
@preconcurrency import SystemPackage
|
||||
#endif
|
||||
|
||||
// Import for specific low-level operations not yet in Swift System
|
||||
#if os(macOS) || os(iOS) || os(tvOS) || os(watchOS)
|
||||
import Darwin.POSIX
|
||||
#elseif os(Linux)
|
||||
import Glibc
|
||||
#endif
|
||||
|
||||
/// Standard input/output transport implementation
|
||||
public actor StdioTransport: Transport {
|
||||
private let input: FileDescriptor
|
||||
private let output: FileDescriptor
|
||||
public nonisolated let logger: Logger
|
||||
|
||||
private var isConnected = false
|
||||
private let messageStream: AsyncStream<Data>
|
||||
private let messageContinuation: AsyncStream<Data>.Continuation
|
||||
|
||||
public init(
|
||||
input: FileDescriptor = FileDescriptor.standardInput,
|
||||
output: FileDescriptor = FileDescriptor.standardOutput,
|
||||
logger: Logger? = nil
|
||||
) {
|
||||
self.input = input
|
||||
self.output = output
|
||||
self.logger =
|
||||
logger
|
||||
?? Logger(
|
||||
label: "mcp.transport.stdio",
|
||||
factory: { _ in SwiftLogNoOpLogHandler() })
|
||||
|
||||
// Create message stream
|
||||
var continuation: AsyncStream<Data>.Continuation!
|
||||
self.messageStream = AsyncStream { continuation = $0 }
|
||||
self.messageContinuation = continuation
|
||||
}
|
||||
|
||||
public func connect() async throws {
|
||||
guard !isConnected else { return }
|
||||
|
||||
// Set non-blocking mode
|
||||
try setNonBlocking(fileDescriptor: input)
|
||||
try setNonBlocking(fileDescriptor: output)
|
||||
|
||||
isConnected = true
|
||||
logger.info("Transport connected successfully")
|
||||
|
||||
// Start reading loop in background
|
||||
Task {
|
||||
await readLoop()
|
||||
}
|
||||
}
|
||||
|
||||
private func setNonBlocking(fileDescriptor: FileDescriptor) throws {
|
||||
#if os(macOS) || os(iOS) || os(tvOS) || os(watchOS) || os(Linux)
|
||||
// Get current flags
|
||||
let flags = fcntl(fileDescriptor.rawValue, F_GETFL)
|
||||
guard flags >= 0 else {
|
||||
throw MCPError.transportError(Errno(rawValue: CInt(errno)))
|
||||
}
|
||||
|
||||
// Set non-blocking flag
|
||||
let result = fcntl(fileDescriptor.rawValue, F_SETFL, flags | O_NONBLOCK)
|
||||
guard result >= 0 else {
|
||||
throw MCPError.transportError(Errno(rawValue: CInt(errno)))
|
||||
}
|
||||
#else
|
||||
// For platforms where non-blocking operations aren't supported
|
||||
throw MCPError.internalError("Setting non-blocking mode not supported on this platform")
|
||||
#endif
|
||||
}
|
||||
|
||||
private func readLoop() async {
|
||||
let bufferSize = 4096
|
||||
var buffer = [UInt8](repeating: 0, count: bufferSize)
|
||||
var pendingData = Data()
|
||||
|
||||
while isConnected && !Task.isCancelled {
|
||||
do {
|
||||
let bytesRead = try buffer.withUnsafeMutableBufferPointer { pointer in
|
||||
try input.read(into: UnsafeMutableRawBufferPointer(pointer))
|
||||
}
|
||||
|
||||
if bytesRead == 0 {
|
||||
logger.notice("EOF received")
|
||||
break
|
||||
}
|
||||
|
||||
pendingData.append(Data(buffer[..<bytesRead]))
|
||||
|
||||
// Process complete messages
|
||||
while let newlineIndex = pendingData.firstIndex(of: UInt8(ascii: "\n")) {
|
||||
let messageData = pendingData[..<newlineIndex]
|
||||
pendingData = pendingData[(newlineIndex + 1)...]
|
||||
|
||||
if !messageData.isEmpty {
|
||||
logger.debug("Message received", metadata: ["size": "\(messageData.count)"])
|
||||
messageContinuation.yield(Data(messageData))
|
||||
}
|
||||
}
|
||||
} catch let error where MCPError.isResourceTemporarilyUnavailable(error) {
|
||||
try? await Task.sleep(for: .milliseconds(10))
|
||||
continue
|
||||
} catch {
|
||||
if !Task.isCancelled {
|
||||
logger.error("Read error occurred", metadata: ["error": "\(error)"])
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
messageContinuation.finish()
|
||||
}
|
||||
|
||||
public func disconnect() async {
|
||||
guard isConnected else { return }
|
||||
isConnected = false
|
||||
messageContinuation.finish()
|
||||
logger.info("Transport disconnected")
|
||||
}
|
||||
|
||||
public func send(_ message: Data) async throws {
|
||||
guard isConnected else {
|
||||
#if os(macOS) || os(iOS) || os(tvOS) || os(watchOS) || os(Linux)
|
||||
throw MCPError.transportError(Errno(rawValue: ENOTCONN))
|
||||
#else
|
||||
throw MCPError.internalError("Transport not connected")
|
||||
#endif
|
||||
}
|
||||
|
||||
// Add newline as delimiter
|
||||
var messageWithNewline = message
|
||||
messageWithNewline.append(UInt8(ascii: "\n"))
|
||||
|
||||
var remaining = messageWithNewline
|
||||
while !remaining.isEmpty {
|
||||
do {
|
||||
let written = try remaining.withUnsafeBytes { buffer in
|
||||
try output.write(UnsafeRawBufferPointer(buffer))
|
||||
}
|
||||
if written > 0 {
|
||||
remaining = remaining.dropFirst(written)
|
||||
}
|
||||
} catch let error where MCPError.isResourceTemporarilyUnavailable(error) {
|
||||
try await Task.sleep(for: .milliseconds(10))
|
||||
continue
|
||||
} catch {
|
||||
throw MCPError.transportError(error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public func receive() -> AsyncThrowingStream<Data, Swift.Error> {
|
||||
return AsyncThrowingStream { continuation in
|
||||
Task {
|
||||
for await message in messageStream {
|
||||
continuation.yield(message)
|
||||
}
|
||||
continuation.finish()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -127,7 +127,7 @@ struct RoundtripTests {
|
|||
let pingTask = Task {
|
||||
try await client.ping()
|
||||
// Ping doesn't return anything, so just getting here without throwing is success
|
||||
#expect(true) // Test passed if we reach this point
|
||||
#expect(Bool(true))
|
||||
}
|
||||
|
||||
try await withThrowingTaskGroup(of: Void.self) { group in
|
||||
|
|
|
|||
Loading…
Reference in a new issue