dependencies:[.package(url:"https://github.com/modelcontextprotocol/swift-sdk.git", from:"0.10.0")]

.target( name:"YourTarget", dependencies:[.product(name:"MCP",package:"swift-sdk")])

import MCP // Initialize the client letclient=Client(name:"MyApp", version:"1.0.0") // Create a transport and connect lettransport=StdioTransport()letresult=tryawait client.connect(transport: transport) // Check server capabilities if result.capabilities.tools !=nil{ // Server supports tools (implicitly including tool calling if the 'tools' capability object is present) }

// Create a stdio transport (simplest option) lettransport=StdioTransport()tryawait client.connect(transport: transport)

// Create a streaming HTTP transport lettransport=HTTPClientTransport( endpoint:URL(string:"http://localhost:8080")!, streaming:true // Enable Server-Sent Events for real-time updates )tryawait client.connect(transport: transport)

// List available tools let(tools, cursor)=tryawait client.listTools()print("Available tools: \(tools.map{ $0.name }.joined(separator:", "))") // Call a tool with arguments let(content, isError)=tryawait client.callTool( name:"image-generator", arguments:["prompt":"A serene mountain landscape at sunset","style":"photorealistic","width":1024,"height":768]) // Handle tool content foritemin content {switch item {case.text(let text):print("Generated text: \(text)")case.image(let data,let mimeType,let metadata):iflet width =metadata?["width"]as?Int,let height =metadata?["height"]as?Int{print("Generated \(width)x\(height) image of type \(mimeType)") // Save or display the image data }case.audio(let data,let mimeType):print("Received audio data of type \(mimeType)")case.resource(let uri,let mimeType,let text):print("Received resource from \(uri) of type \(mimeType)")iflet text = text {print("Resource text: \(text)")}}}

// List available resources let(resources, nextCursor)=tryawait client.listResources()print("Available resources: \(resources.map{ $0.uri }.joined(separator:", "))") // Read a resource letcontents=tryawait client.readResource(uri:"resource://example")print("Resource content: \(contents)") // Subscribe to resource updates if supported if result.capabilities.resources.subscribe {tryawait client.subscribeToResource(uri:"resource://example") // Register notification handler await client.onNotification(ResourceUpdatedNotification.self){ message inleturi= message.params.uri print("Resource \(uri) updated with new content") // Fetch the updated resource content letupdatedContents=tryawait client.readResource(uri: uri)print("Updated resource content received")}}

// List available prompts let(prompts, nextCursor)=tryawait client.listPrompts()print("Available prompts: \(prompts.map{ $0.name }.joined(separator:", "))") // Get a prompt with arguments let(description, messages)=tryawait client.getPrompt( name:"customer-service", arguments:["customerName":"Alice","orderNumber":"ORD-12345","issue":"delivery delay"]) // Use the prompt messages in your application print("Prompt description: \(description)")formessagein messages {if case .text(text:let text)= message.content {print("\(message.role): \(text)")}}

// Register a sampling handler in the client await client.withSamplingHandler{ parameters in // Review the sampling request (human-in-the-loop step 1) print("Server requests completion for: \(parameters.messages)") // Optionally modify the request based on user input varmessages= parameters.messages iflet systemPrompt = parameters.systemPrompt {print("System prompt: \(systemPrompt)")} // Sample from your LLM (this is where you'd call your AI service) letcompletion=tryawaitcallYourLLMService( messages: messages, maxTokens: parameters.maxTokens, temperature: parameters.temperature ) // Review the completion (human-in-the-loop step 2) print("LLM generated: \(completion)") // User can approve, modify, or reject the completion here // Return the result to the server returnCreateSamplingMessage.Result( model:"your-model-name", stopReason:.endTurn, role:.assistant, content:.text(completion))}

do{tryawait client.connect(transport: transport) // Success }catchlet error as MCPError{print("MCP Error: \(error.localizedDescription)")}catch{print("Unexpected error: \(error)")}

// Strict configuration - fail fast if a capability is missing letstrictClient=Client( name:"StrictClient", version:"1.0.0", configuration:.strict ) // With strict configuration, calling a method for an unsupported capability // will throw an error immediately without sending a request do{ // This will throw an error if resources.list capability is not available letresources=tryawait strictClient.listResources()}catchlet error as MCPError{print("Capability not available: \(error.localizedDescription)")} // Default (non-strict) configuration - attempt the request anyway letclient=Client( name:"FlexibleClient", version:"1.0.0", configuration:.default ) // With default configuration, the client will attempt the request // even if the capability wasn't advertised by the server do{letresources=tryawait client.listResources()}catchlet error as MCPError{ // Still handle the error if the server rejects the request print("Server rejected request: \(error.localizedDescription)")}

// Array to hold tool call tasks vartoolTasks:[Task<CallTool.Result,Swift.Error>]=[] // Send a batch of requests tryawait client.withBatch{ batch in // Add multiple tool calls to the batch foriin0..<10{ toolTasks.append(tryawait batch.addRequest(CallTool.request(.init(name:"square", arguments:["n":Value(i)]))))}} // Process results after the batch is sent print("Processing \(toolTasks.count) tool results...")for(index, task)in toolTasks.enumerated(){do{letresult=tryawait task.value print("\(index): \(result.content)")}catch{print("\(index) failed: \(error)")}}

// Declare task variables varpingTask:Task<Ping.Result,Error>?varpromptTask:Task<GetPrompt.Result,Error>? // Send a batch with different request types tryawait client.withBatch{ batch in pingTask =tryawait batch.addRequest(Ping.request()) promptTask =tryawait batch.addRequest(GetPrompt.request(.init(name:"greeting")))} // Process individual results do{iflet pingTask = pingTask {tryawait pingTask.value print("Ping successful")}iflet promptTask = promptTask {letpromptResult=tryawait promptTask.value print("Prompt: \(promptResult.description ??"None")")}}catch{print("Error processing batch results: \(error)")}

import MCP // Create a server with given capabilities letserver=Server( name:"MyModelServer", version:"1.0.0", capabilities:.init( prompts:.init(listChanged:true), resources:.init(subscribe:true, listChanged:true), tools:.init(listChanged:true))) // Create transport and start server lettransport=StdioTransport()tryawait server.start(transport: transport) // Now register handlers for the capabilities you've enabled

// Register a tool list handler await server.withMethodHandler(ListTools.self){ _ inlettools=[Tool( name:"weather", description:"Get current weather for a location", inputSchema:.object(["properties":.object(["location":.string("City name or coordinates"),"units":.string("Units of measurement, e.g., metric, imperial")])])),Tool( name:"calculator", description:"Perform calculations", inputSchema:.object(["properties":.object(["expression":.string("Mathematical expression to evaluate")])]))]return.init(tools: tools)} // Register a tool call handler await server.withMethodHandler(CallTool.self){ params inswitch params.name {case"weather":letlocation= params.arguments?["location"]?.stringValue ??"Unknown"letunits= params.arguments?["units"]?.stringValue ??"metric"letweatherData=getWeatherData(location: location, units: units) // Your implementation return.init( content:[.text("Weather for \(location): \(weatherData.temperature)°, \(weatherData.conditions)")], isError:false)case"calculator":iflet expression = params.arguments?["expression"]?.stringValue {letresult=evaluateExpression(expression) // Your implementation return.init(content:[.text("\(result)")], isError:false)}else{return.init(content:[.text("Missing expression parameter")], isError:true)}default:return.init(content:[.text("Unknown tool")], isError:true)}}

// Register a resource list handler await server.withMethodHandler(ListResources.self){ params inletresources=[Resource( name:"Knowledge Base Articles", uri:"resource://knowledge-base/articles", description:"Collection of support articles and documentation"),Resource( name:"System Status", uri:"resource://system/status", description:"Current system operational status")]return.init(resources: resources, nextCursor:nil)} // Register a resource read handler await server.withMethodHandler(ReadResource.self){ params inswitch params.uri {case"resource://knowledge-base/articles":return.init(contents:[Resource.Content.text("# Knowledge Base\n\nThis is the content of the knowledge base...", uri: params.uri)])case"resource://system/status":letstatus=getCurrentSystemStatus() // Your implementation letstatusJson="""{"status": "\(status.overall)","components":{"database": "\(status.database)","api": "\(status.api)","model": "\(status.model)" },"lastUpdated": "\(status.timestamp)" }"""return.init(contents:[Resource.Content.text(statusJson, uri: params.uri, mimeType:"application/json")])default:throwMCPError.invalidParams("Unknown resource URI: \(params.uri)")}} // Register a resource subscribe handler await server.withMethodHandler(ResourceSubscribe.self){ params in // Store subscription for later notifications. // Client identity for multi-client scenarios needs to be managed by the server application, // potentially using information from the initialize handshake if the server handles one client post-init. // addSubscription(clientID: /* some_client_identifier */, uri: params.uri) print("Client subscribed to \(params.uri). Server needs to implement logic to track this subscription.")return.init()}

// Register a prompt list handler await server.withMethodHandler(ListPrompts.self){ params inletprompts=[Prompt( name:"interview", description:"Job interview conversation starter", arguments:[.init(name:"position", description:"Job position", required:true),.init(name:"company", description:"Company name", required:true),.init(name:"interviewee", description:"Candidate name")]),Prompt( name:"customer-support", description:"Customer support conversation starter", arguments:[.init(name:"issue", description:"Customer issue", required:true),.init(name:"product", description:"Product name", required:true)])]return.init(prompts: prompts, nextCursor:nil)} // Register a prompt get handler await server.withMethodHandler(GetPrompt.self){ params inswitch params.name {case"interview":letposition= params.arguments?["position"]?.stringValue ??"Software Engineer"letcompany= params.arguments?["company"]?.stringValue ??"Acme Corp"letinterviewee= params.arguments?["interviewee"]?.stringValue ??"Candidate"letdescription="Job interview for \(position) position at \(company)"letmessages:[Prompt.Message]=[.user("You are an interviewer for the \(position) position at \(company)."),.user("Hello, I'm \(interviewee) and I'm here for the \(position) interview."),.assistant("Hi \(interviewee), welcome to \(company)! I'd like to start by asking about your background and experience.")]return.init(description: description, messages: messages)case"customer-support": // Similar implementation for customer support prompt default:throwMCPError.invalidParams("Unknown prompt name: \(params.name)")}}

// Enable sampling capability in server letserver=Server( name:"MyModelServer", version:"1.0.0", capabilities:.init( sampling:.init(), // Enable sampling capability tools:.init(listChanged:true))) // Request sampling from the client (conceptual - requires bidirectional transport) do{letresult=tryawait server.requestSampling( messages:[.user("Analyze this data and suggest next steps")], systemPrompt:"You are a helpful data analyst", temperature:0.7, maxTokens:150) // Use the LLM completion in your server logic print("LLM suggested: \(result.content)")}catch{print("Sampling request failed: \(error)")}

// Start the server with an initialize hook tryawait server.start(transport: transport){ clientInfo, clientCapabilities in // Validate client info guard clientInfo.name !="BlockedClient"else{throwMCPError.invalidRequest("This client is not allowed")} // You can also inspect client capabilities if clientCapabilities.sampling ==nil{print("Client does not support sampling")} // Perform any server-side setup based on client info print("Client \(clientInfo.name) v\(clientInfo.version) connected") // If the hook completes without throwing, initialization succeeds }

.package(url:"https://github.com/swift-server/swift-service-lifecycle.git", from:"2.3.0"),

import MCP import ServiceLifecycle import Logging structMCPService:Service{letserver:Serverlettransport:Transportinit(server:Server, transport:Transport){self.server = server self.transport = transport }func run()asyncthrows{ // Start the server tryawait server.start(transport: transport) // Keep running until external cancellation tryawaitTask.sleep(for:.days(365*100)) // Effectively forever }func shutdown()asyncthrows{ // Gracefully shutdown the server await server.stop()}}

import MCP import ServiceLifecycle import Logging letlogger=Logger(label:"com.example.mcp-server") // Create the MCP server letserver=Server( name:"MyModelServer", version:"1.0.0", capabilities:.init( prompts:.init(listChanged:true), resources:.init(subscribe:true, listChanged:true), tools:.init(listChanged:true))) // Add handlers directly to the server await server.withMethodHandler(ListTools.self){ _ in // Your implementation return.init(tools:[Tool(name:"example", description:"An example tool")])}await server.withMethodHandler(CallTool.self){ params in // Your implementation return.init(content:[.text("Tool result")], isError:false)} // Create MCP service and other services lettransport=StdioTransport(logger: logger)letmcpService=MCPService(server: server, transport: transport)letdatabaseService=DatabaseService() // Your other services // Create service group with signal handling letserviceGroup=ServiceGroup( services:[mcpService, databaseService], configuration:.init( gracefulShutdownSignals:[.sigterm,.sigint]), logger: logger ) // Run the service group - this blocks until shutdown tryawait serviceGroup.run()

import MCP import Foundation publicactorMyCustomTransport:Transport{publicnonisolatedletlogger:LoggerprivatevarisConnected=falseprivateletmessageStream:AsyncThrowingStream<Data,anySwift.Error>privateletmessageContinuation:AsyncThrowingStream<Data,anySwift.Error>.Continuationpublicinit(logger:Logger?=nil){self.logger = logger ??Logger(label:"my.custom.transport")varcontinuation:AsyncThrowingStream<Data,anySwift.Error>.Continuation!self.messageStream =AsyncThrowingStream{ continuation = $0 }self.messageContinuation = continuation }publicfunc connect()asyncthrows{ // Implement your connection logic isConnected =true}publicfunc disconnect()async{ // Implement your disconnection logic isConnected =false messageContinuation.finish()}publicfunc send(_ data:Data)asyncthrows{ // Implement your message sending logic }publicfunc receive()->AsyncThrowingStream<Data,anySwift.Error>{return messageStream }}

import Logging import MCP // Configure Logger LoggingSystem.bootstrap{ label invarhandler=StreamLogHandler.standardOutput(label: label) handler.logLevel =.debug return handler } // Create logger letlogger=Logger(label:"com.example.mcp") // Pass to client/server letclient=Client(name:"MyApp", version:"1.0.0") // Pass to transport lettransport=StdioTransport(logger: logger)