modularize sdk (#208)
* Modularize SDK: migrated client and server implementations to separate modules. Removed old configuration files and reorganized the source code structure * add support native targets and update js/wasm configurations * Refactor sample projects: add samples as composite builds, replace hardcoded dependencies with version catalog, remove obsolete gradle wrapper files * Update .gitignore: exclude SWE agent directories (.claude/ and .junie/) * Refine CodeQL workflow: update Kotlin build steps to target individual modules separately instead of building all at once
This commit is contained in:
Pavel Gorgulov
GitHub
parent
641df74f5d
commit
8698b96dd2
9 changed files with 979 additions and 2003 deletions
|
|
@ -0,0 +1,608 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.github.oshai.kotlinlogging.KotlinLogging
|
||||
import io.modelcontextprotocol.kotlin.sdk.CallToolRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.CallToolResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.CallToolResultBase
|
||||
import io.modelcontextprotocol.kotlin.sdk.ClientCapabilities
|
||||
import io.modelcontextprotocol.kotlin.sdk.CompatibilityCallToolResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.CompleteRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.CompleteResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.CreateElicitationRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.CreateElicitationResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.EmptyRequestResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.GetPromptRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.GetPromptResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.Implementation
|
||||
import io.modelcontextprotocol.kotlin.sdk.InitializeRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.InitializeResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.InitializedNotification
|
||||
import io.modelcontextprotocol.kotlin.sdk.LATEST_PROTOCOL_VERSION
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListPromptsRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListPromptsResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListResourceTemplatesRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListResourceTemplatesResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListResourcesRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListResourcesResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListRootsRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListRootsResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListToolsRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.ListToolsResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.LoggingLevel
|
||||
import io.modelcontextprotocol.kotlin.sdk.LoggingMessageNotification.SetLevelRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.Method
|
||||
import io.modelcontextprotocol.kotlin.sdk.PingRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.ReadResourceRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.ReadResourceResult
|
||||
import io.modelcontextprotocol.kotlin.sdk.Root
|
||||
import io.modelcontextprotocol.kotlin.sdk.RootsListChangedNotification
|
||||
import io.modelcontextprotocol.kotlin.sdk.SUPPORTED_PROTOCOL_VERSIONS
|
||||
import io.modelcontextprotocol.kotlin.sdk.ServerCapabilities
|
||||
import io.modelcontextprotocol.kotlin.sdk.SubscribeRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.UnsubscribeRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.Protocol
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.ProtocolOptions
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.RequestOptions
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.Transport
|
||||
import kotlinx.atomicfu.atomic
|
||||
import kotlinx.atomicfu.getAndUpdate
|
||||
import kotlinx.atomicfu.update
|
||||
import kotlinx.collections.immutable.minus
|
||||
import kotlinx.collections.immutable.persistentMapOf
|
||||
import kotlinx.collections.immutable.toPersistentSet
|
||||
import kotlinx.serialization.json.JsonElement
|
||||
import kotlinx.serialization.json.JsonNull
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
import kotlinx.serialization.json.JsonPrimitive
|
||||
import kotlin.coroutines.cancellation.CancellationException
|
||||
|
||||
private val logger = KotlinLogging.logger {}
|
||||
|
||||
/**
|
||||
* Options for configuring the MCP client.
|
||||
*
|
||||
* @property capabilities The capabilities this client supports.
|
||||
* @property enforceStrictCapabilities Whether to strictly enforce capabilities when interacting with the server.
|
||||
*/
|
||||
public class ClientOptions(
|
||||
public val capabilities: ClientCapabilities = ClientCapabilities(),
|
||||
enforceStrictCapabilities: Boolean = true,
|
||||
) : ProtocolOptions(enforceStrictCapabilities = enforceStrictCapabilities)
|
||||
|
||||
/**
|
||||
* An MCP client on top of a pluggable transport.
|
||||
*
|
||||
* The client automatically performs the initialization handshake with the server when [connect] is called.
|
||||
* After initialization, [serverCapabilities] and [serverVersion] provide details about the connected server.
|
||||
*
|
||||
* You can extend this class with custom request/notification/result types if needed.
|
||||
*
|
||||
* @param clientInfo Information about the client implementation (name, version).
|
||||
* @param options Configuration options for this client.
|
||||
*/
|
||||
public open class Client(
|
||||
private val clientInfo: Implementation,
|
||||
options: ClientOptions = ClientOptions(),
|
||||
) : Protocol(options) {
|
||||
|
||||
/**
|
||||
* Retrieves the server's reported capabilities after the initialization process completes.
|
||||
*
|
||||
* @return The server's capabilities, or `null` if initialization is not yet complete.
|
||||
*/
|
||||
public var serverCapabilities: ServerCapabilities? = null
|
||||
private set
|
||||
|
||||
/**
|
||||
* Retrieves the server's reported version information after initialization.
|
||||
*
|
||||
* @return Information about the server's implementation, or `null` if initialization is not yet complete.
|
||||
*/
|
||||
public var serverVersion: Implementation? = null
|
||||
private set
|
||||
|
||||
private val capabilities: ClientCapabilities = options.capabilities
|
||||
|
||||
private val roots = atomic(persistentMapOf<String, Root>())
|
||||
|
||||
init {
|
||||
logger.debug { "Initializing MCP client with capabilities: $capabilities" }
|
||||
|
||||
// Internal handlers for roots
|
||||
if (capabilities.roots != null) {
|
||||
setRequestHandler<ListRootsRequest>(Method.Defined.RootsList) { _, _ ->
|
||||
handleListRoots()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected fun assertCapability(capability: String, method: String) {
|
||||
val caps = serverCapabilities
|
||||
val hasCapability = when (capability) {
|
||||
"logging" -> caps?.logging != null
|
||||
"prompts" -> caps?.prompts != null
|
||||
"resources" -> caps?.resources != null
|
||||
"tools" -> caps?.tools != null
|
||||
else -> true
|
||||
}
|
||||
|
||||
if (!hasCapability) {
|
||||
throw IllegalStateException("Server does not support $capability (required for $method)")
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Connects the client to the given [transport], performing the initialization handshake with the server.
|
||||
*
|
||||
* @param transport The transport to use for communication with the server.
|
||||
* @throws IllegalStateException If the server's protocol version is not supported.
|
||||
*/
|
||||
override suspend fun connect(transport: Transport) {
|
||||
super.connect(transport)
|
||||
|
||||
try {
|
||||
val message = InitializeRequest(
|
||||
protocolVersion = LATEST_PROTOCOL_VERSION,
|
||||
capabilities = capabilities,
|
||||
clientInfo = clientInfo
|
||||
)
|
||||
val result = request<InitializeResult>(message)
|
||||
|
||||
if (!SUPPORTED_PROTOCOL_VERSIONS.contains(result.protocolVersion)) {
|
||||
throw IllegalStateException(
|
||||
"Server's protocol version is not supported: ${result.protocolVersion}"
|
||||
)
|
||||
}
|
||||
|
||||
serverCapabilities = result.capabilities
|
||||
serverVersion = result.serverInfo
|
||||
|
||||
notification(InitializedNotification())
|
||||
} catch (error: Throwable) {
|
||||
close()
|
||||
if (error !is CancellationException) {
|
||||
throw IllegalStateException("Error connecting to transport: ${error.message}")
|
||||
}
|
||||
|
||||
throw error
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
override fun assertCapabilityForMethod(method: Method) {
|
||||
when (method) {
|
||||
Method.Defined.LoggingSetLevel -> {
|
||||
if (serverCapabilities?.logging == null) {
|
||||
throw IllegalStateException("Server does not support logging (required for $method)")
|
||||
}
|
||||
}
|
||||
|
||||
Method.Defined.PromptsGet,
|
||||
Method.Defined.PromptsList,
|
||||
Method.Defined.CompletionComplete,
|
||||
-> {
|
||||
if (serverCapabilities?.prompts == null) {
|
||||
throw IllegalStateException("Server does not support prompts (required for $method)")
|
||||
}
|
||||
}
|
||||
|
||||
Method.Defined.ResourcesList,
|
||||
Method.Defined.ResourcesTemplatesList,
|
||||
Method.Defined.ResourcesRead,
|
||||
Method.Defined.ResourcesSubscribe,
|
||||
Method.Defined.ResourcesUnsubscribe,
|
||||
-> {
|
||||
val resCaps = serverCapabilities?.resources
|
||||
?: error("Server does not support resources (required for $method)")
|
||||
|
||||
if (method == Method.Defined.ResourcesSubscribe && resCaps.subscribe != true) {
|
||||
throw IllegalStateException(
|
||||
"Server does not support resource subscriptions (required for $method)"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
Method.Defined.ToolsCall,
|
||||
Method.Defined.ToolsList,
|
||||
-> {
|
||||
if (serverCapabilities?.tools == null) {
|
||||
throw IllegalStateException("Server does not support tools (required for $method)")
|
||||
}
|
||||
}
|
||||
|
||||
Method.Defined.Initialize,
|
||||
Method.Defined.Ping,
|
||||
-> {
|
||||
// No specific capability required
|
||||
}
|
||||
|
||||
else -> {
|
||||
// For unknown or future methods, no assertion by default
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun assertNotificationCapability(method: Method) {
|
||||
when (method) {
|
||||
Method.Defined.NotificationsRootsListChanged -> {
|
||||
if (capabilities.roots?.listChanged != true) {
|
||||
throw IllegalStateException(
|
||||
"Client does not support roots list changed notifications (required for $method)"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
Method.Defined.NotificationsInitialized,
|
||||
Method.Defined.NotificationsCancelled,
|
||||
Method.Defined.NotificationsProgress,
|
||||
-> {
|
||||
// Always allowed
|
||||
}
|
||||
|
||||
else -> {
|
||||
// For notifications not specifically listed, no assertion by default
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override fun assertRequestHandlerCapability(method: Method) {
|
||||
when (method) {
|
||||
Method.Defined.SamplingCreateMessage -> {
|
||||
if (capabilities.sampling == null) {
|
||||
throw IllegalStateException(
|
||||
"Client does not support sampling capability (required for $method)"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
Method.Defined.RootsList -> {
|
||||
if (capabilities.roots == null) {
|
||||
throw IllegalStateException(
|
||||
"Client does not support roots capability (required for $method)"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
Method.Defined.ElicitationCreate -> {
|
||||
if (capabilities.elicitation == null) {
|
||||
throw IllegalStateException(
|
||||
"Client does not support elicitation capability (required for $method)"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
Method.Defined.Ping -> {
|
||||
// No capability required
|
||||
}
|
||||
|
||||
else -> {}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Sends a ping request to the server to check connectivity.
|
||||
*
|
||||
* @param options Optional request options.
|
||||
* @throws IllegalStateException If the server does not support the ping method (unlikely).
|
||||
*/
|
||||
public suspend fun ping(options: RequestOptions? = null): EmptyRequestResult {
|
||||
return request<EmptyRequestResult>(PingRequest(), options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a completion request to the server, typically to generate or complete some content.
|
||||
*
|
||||
* @param params The completion request parameters.
|
||||
* @param options Optional request options.
|
||||
* @return The completion result returned by the server, or `null` if none.
|
||||
* @throws IllegalStateException If the server does not support prompts or completion.
|
||||
*/
|
||||
public suspend fun complete(params: CompleteRequest, options: RequestOptions? = null): CompleteResult {
|
||||
return request(params, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the logging level on the server.
|
||||
*
|
||||
* @param level The desired logging level.
|
||||
* @param options Optional request options.
|
||||
* @throws IllegalStateException If the server does not support logging.
|
||||
*/
|
||||
public suspend fun setLoggingLevel(level: LoggingLevel, options: RequestOptions? = null): EmptyRequestResult {
|
||||
return request<EmptyRequestResult>(SetLevelRequest(level), options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Retrieves a prompt by name from the server.
|
||||
*
|
||||
* @param request The prompt request containing the prompt name.
|
||||
* @param options Optional request options.
|
||||
* @return The requested prompt details, or `null` if not found.
|
||||
* @throws IllegalStateException If the server does not support prompts.
|
||||
*/
|
||||
public suspend fun getPrompt(request: GetPromptRequest, options: RequestOptions? = null): GetPromptResult {
|
||||
return request(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Lists all available prompts from the server.
|
||||
*
|
||||
* @param request A request object for listing prompts (usually empty).
|
||||
* @param options Optional request options.
|
||||
* @return The list of available prompts, or `null` if none.
|
||||
* @throws IllegalStateException If the server does not support prompts.
|
||||
*/
|
||||
public suspend fun listPrompts(
|
||||
request: ListPromptsRequest = ListPromptsRequest(),
|
||||
options: RequestOptions? = null,
|
||||
): ListPromptsResult {
|
||||
return request(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Lists all available resources from the server.
|
||||
*
|
||||
* @param request A request object for listing resources (usually empty).
|
||||
* @param options Optional request options.
|
||||
* @return The list of resources, or `null` if none.
|
||||
* @throws IllegalStateException If the server does not support resources.
|
||||
*/
|
||||
public suspend fun listResources(
|
||||
request: ListResourcesRequest = ListResourcesRequest(),
|
||||
options: RequestOptions? = null,
|
||||
): ListResourcesResult {
|
||||
return request(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Lists resource templates available on the server.
|
||||
*
|
||||
* @param request The request object for listing resource templates.
|
||||
* @param options Optional request options.
|
||||
* @return The list of resource templates, or `null` if none.
|
||||
* @throws IllegalStateException If the server does not support resources.
|
||||
*/
|
||||
public suspend fun listResourceTemplates(
|
||||
request: ListResourceTemplatesRequest,
|
||||
options: RequestOptions? = null,
|
||||
): ListResourceTemplatesResult {
|
||||
return request(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads a resource from the server by its URI.
|
||||
*
|
||||
* @param request The request object containing the resource URI.
|
||||
* @param options Optional request options.
|
||||
* @return The resource content, or `null` if the resource is not found.
|
||||
* @throws IllegalStateException If the server does not support resources.
|
||||
*/
|
||||
public suspend fun readResource(
|
||||
request: ReadResourceRequest,
|
||||
options: RequestOptions? = null,
|
||||
): ReadResourceResult {
|
||||
return request(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribes to resource changes on the server.
|
||||
*
|
||||
* @param request The subscription request containing resource details.
|
||||
* @param options Optional request options.
|
||||
* @throws IllegalStateException If the server does not support resource subscriptions.
|
||||
*/
|
||||
public suspend fun subscribeResource(
|
||||
request: SubscribeRequest,
|
||||
options: RequestOptions? = null,
|
||||
): EmptyRequestResult {
|
||||
return request(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Unsubscribes from resource changes on the server.
|
||||
*
|
||||
* @param request The unsubscribe request containing resource details.
|
||||
* @param options Optional request options.
|
||||
* @throws IllegalStateException If the server does not support resource subscriptions.
|
||||
*/
|
||||
public suspend fun unsubscribeResource(
|
||||
request: UnsubscribeRequest,
|
||||
options: RequestOptions? = null,
|
||||
): EmptyRequestResult {
|
||||
return request(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Calls a tool on the server by name, passing the specified arguments.
|
||||
*
|
||||
* @param name The name of the tool to call.
|
||||
* @param arguments A map of argument names to values for the tool.
|
||||
* @param compatibility Whether to use compatibility mode for older protocol versions.
|
||||
* @param options Optional request options.
|
||||
* @return The result of the tool call, or `null` if none.
|
||||
* @throws IllegalStateException If the server does not support tools.
|
||||
*/
|
||||
public suspend fun callTool(
|
||||
name: String,
|
||||
arguments: Map<String, Any?>,
|
||||
compatibility: Boolean = false,
|
||||
options: RequestOptions? = null,
|
||||
): CallToolResultBase? {
|
||||
val jsonArguments = arguments.mapValues { (_, value) ->
|
||||
when (value) {
|
||||
is String -> JsonPrimitive(value)
|
||||
is Number -> JsonPrimitive(value)
|
||||
is Boolean -> JsonPrimitive(value)
|
||||
is JsonElement -> value
|
||||
null -> JsonNull
|
||||
else -> JsonPrimitive(value.toString())
|
||||
}
|
||||
}
|
||||
|
||||
val request = CallToolRequest(
|
||||
name = name,
|
||||
arguments = JsonObject(jsonArguments)
|
||||
)
|
||||
return callTool(request, compatibility, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Calls a tool on the server using a [CallToolRequest] object.
|
||||
*
|
||||
* @param request The request object containing the tool name and arguments.
|
||||
* @param compatibility Whether to use compatibility mode for older protocol versions.
|
||||
* @param options Optional request options.
|
||||
* @return The result of the tool call, or `null` if none.
|
||||
* @throws IllegalStateException If the server does not support tools.
|
||||
*/
|
||||
public suspend fun callTool(
|
||||
request: CallToolRequest,
|
||||
compatibility: Boolean = false,
|
||||
options: RequestOptions? = null,
|
||||
): CallToolResultBase? {
|
||||
return if (compatibility) {
|
||||
request<CompatibilityCallToolResult>(request, options)
|
||||
} else {
|
||||
request<CallToolResult>(request, options)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Lists all available tools on the server.
|
||||
*
|
||||
* @param request A request object for listing tools (usually empty).
|
||||
* @param options Optional request options.
|
||||
* @return The list of available tools, or `null` if none.
|
||||
* @throws IllegalStateException If the server does not support tools.
|
||||
*/
|
||||
public suspend fun listTools(
|
||||
request: ListToolsRequest = ListToolsRequest(),
|
||||
options: RequestOptions? = null,
|
||||
): ListToolsResult {
|
||||
return request(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers a single root.
|
||||
*
|
||||
* @param uri The URI of the root.
|
||||
* @param name A human-readable name for the root.
|
||||
* @throws IllegalStateException If the client does not support roots.
|
||||
*/
|
||||
public fun addRoot(
|
||||
uri: String,
|
||||
name: String,
|
||||
) {
|
||||
if (capabilities.roots == null) {
|
||||
logger.error { "Failed to add root '$name': Client does not support roots capability" }
|
||||
throw IllegalStateException("Client does not support roots capability.")
|
||||
}
|
||||
logger.info { "Adding root: $name ($uri)" }
|
||||
roots.update { current -> current.put(uri, Root(uri, name)) }
|
||||
}
|
||||
|
||||
/**
|
||||
* Registers multiple roots at once.
|
||||
*
|
||||
* @param rootsToAdd A list of [Root] objects to register.
|
||||
* @throws IllegalStateException If the client does not support roots.
|
||||
*/
|
||||
public fun addRoots(rootsToAdd: List<Root>) {
|
||||
if (capabilities.roots == null) {
|
||||
logger.error { "Failed to add roots: Client does not support roots capability" }
|
||||
throw IllegalStateException("Client does not support roots capability.")
|
||||
}
|
||||
logger.info { "Adding ${rootsToAdd.size} roots" }
|
||||
roots.update { current -> current.putAll(rootsToAdd.associateBy { it.uri }) }
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes a single root by URI.
|
||||
*
|
||||
* @param uri The URI of the root to remove.
|
||||
* @return True if the root was removed, false if it wasn't found.
|
||||
* @throws IllegalStateException If the client does not support roots.
|
||||
*/
|
||||
public fun removeRoot(uri: String): Boolean {
|
||||
if (capabilities.roots == null) {
|
||||
logger.error { "Failed to remove root '$uri': Client does not support roots capability" }
|
||||
throw IllegalStateException("Client does not support roots capability.")
|
||||
}
|
||||
logger.info { "Removing root: $uri" }
|
||||
val oldMap = roots.getAndUpdate { current -> current.remove(uri) }
|
||||
val removed = uri in oldMap
|
||||
logger.debug {
|
||||
if (removed) {
|
||||
"Root removed: $uri"
|
||||
} else {
|
||||
"Root not found: $uri"
|
||||
}
|
||||
}
|
||||
return removed
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes multiple roots at once.
|
||||
*
|
||||
* @param uris A list of root URIs to remove.
|
||||
* @return The number of roots that were successfully removed.
|
||||
* @throws IllegalStateException If the client does not support roots.
|
||||
*/
|
||||
public fun removeRoots(uris: List<String>): Int {
|
||||
if (capabilities.roots == null) {
|
||||
logger.error { "Failed to remove roots: Client does not support roots capability" }
|
||||
throw IllegalStateException("Client does not support roots capability.")
|
||||
}
|
||||
logger.info { "Removing ${uris.size} roots" }
|
||||
|
||||
val oldMap = roots.getAndUpdate { current -> current - uris.toPersistentSet() }
|
||||
|
||||
val removedCount = uris.count { it in oldMap }
|
||||
|
||||
logger.info {
|
||||
if (removedCount > 0) {
|
||||
"Removed $removedCount roots"
|
||||
} else {
|
||||
"No roots were removed"
|
||||
}
|
||||
}
|
||||
return removedCount
|
||||
}
|
||||
|
||||
/**
|
||||
* Notifies the server that the list of roots has changed.
|
||||
* Typically used if the client is managing some form of hierarchical structure.
|
||||
*
|
||||
* @throws IllegalStateException If the client or server does not support roots.
|
||||
*/
|
||||
public suspend fun sendRootsListChanged() {
|
||||
notification(RootsListChangedNotification())
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the elicitation handler.
|
||||
*
|
||||
* @param handler The elicitation handler.
|
||||
* @throws IllegalStateException if the client does not support elicitation.
|
||||
*/
|
||||
public fun setElicitationHandler(handler: (CreateElicitationRequest) -> CreateElicitationResult) {
|
||||
if (capabilities.elicitation == null) {
|
||||
logger.error { "Failed to set elicitation handler: Client does not support elicitation" }
|
||||
throw IllegalStateException("Client does not support elicitation.")
|
||||
}
|
||||
logger.info { "Setting the elicitation handler" }
|
||||
|
||||
setRequestHandler<CreateElicitationRequest>(Method.Defined.ElicitationCreate) { request, _ ->
|
||||
handler(request)
|
||||
}
|
||||
}
|
||||
|
||||
// --- Internal Handlers ---
|
||||
|
||||
private suspend fun handleListRoots(): ListRootsResult {
|
||||
val rootList = roots.value.values.toList()
|
||||
return ListRootsResult(rootList)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,46 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.request.HttpRequestBuilder
|
||||
import io.modelcontextprotocol.kotlin.sdk.Implementation
|
||||
import io.modelcontextprotocol.kotlin.sdk.LIB_VERSION
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.IMPLEMENTATION_NAME
|
||||
import kotlin.time.Duration
|
||||
|
||||
/**
|
||||
* Returns a new SSE transport for the Model Context Protocol using the provided HttpClient.
|
||||
*
|
||||
* @param urlString Optional URL of the MCP server.
|
||||
* @param reconnectionTime Optional duration to wait before attempting to reconnect.
|
||||
* @param requestBuilder Optional lambda to configure the HTTP request.
|
||||
* @return A [SSEClientTransport] configured for MCP communication.
|
||||
*/
|
||||
public fun HttpClient.mcpSseTransport(
|
||||
urlString: String? = null,
|
||||
reconnectionTime: Duration? = null,
|
||||
requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
): SseClientTransport = SseClientTransport(this, urlString, reconnectionTime, requestBuilder)
|
||||
|
||||
/**
|
||||
* Creates and connects an MCP client over SSE using the provided HttpClient.
|
||||
*
|
||||
* @param urlString Optional URL of the MCP server.
|
||||
* @param reconnectionTime Optional duration to wait before attempting to reconnect.
|
||||
* @param requestBuilder Optional lambda to configure the HTTP request.
|
||||
* @return A connected [Client] ready for MCP communication.
|
||||
*/
|
||||
public suspend fun HttpClient.mcpSse(
|
||||
urlString: String? = null,
|
||||
reconnectionTime: Duration? = null,
|
||||
requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
): Client {
|
||||
val transport = mcpSseTransport(urlString, reconnectionTime, requestBuilder)
|
||||
val client = Client(
|
||||
Implementation(
|
||||
name = IMPLEMENTATION_NAME,
|
||||
version = LIB_VERSION
|
||||
)
|
||||
)
|
||||
client.connect(transport)
|
||||
return client
|
||||
}
|
||||
|
|
@ -0,0 +1,190 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.plugins.sse.ClientSSESession
|
||||
import io.ktor.client.plugins.sse.sseSession
|
||||
import io.ktor.client.request.HttpRequestBuilder
|
||||
import io.ktor.client.request.post
|
||||
import io.ktor.client.request.setBody
|
||||
import io.ktor.client.statement.bodyAsText
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpHeaders
|
||||
import io.ktor.http.Url
|
||||
import io.ktor.http.append
|
||||
import io.ktor.http.isSuccess
|
||||
import io.ktor.http.protocolWithAuthority
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.McpJson
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.CoroutineName
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.ExperimentalCoroutinesApi
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.cancelAndJoin
|
||||
import kotlinx.coroutines.ensureActive
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.serialization.SerializationException
|
||||
import kotlin.concurrent.atomics.AtomicBoolean
|
||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
import kotlin.time.Duration
|
||||
|
||||
@Deprecated("Use SseClientTransport instead", ReplaceWith("SseClientTransport"), DeprecationLevel.WARNING)
|
||||
public typealias SSEClientTransport = SseClientTransport
|
||||
|
||||
/**
|
||||
* Client transport for SSE: this will connect to a server using Server-Sent Events for receiving
|
||||
* messages and make separate POST requests for sending messages.
|
||||
*/
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
public class SseClientTransport(
|
||||
private val client: HttpClient,
|
||||
private val urlString: String?,
|
||||
private val reconnectionTime: Duration? = null,
|
||||
private val requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
) : AbstractTransport() {
|
||||
private val initialized: AtomicBoolean = AtomicBoolean(false)
|
||||
private val endpoint = CompletableDeferred<String>()
|
||||
|
||||
private lateinit var session: ClientSSESession
|
||||
private lateinit var scope: CoroutineScope
|
||||
private var job: Job? = null
|
||||
|
||||
private val baseUrl: String by lazy {
|
||||
session.call.request.url.let { url ->
|
||||
val path = url.encodedPath
|
||||
when {
|
||||
path.isEmpty() -> url.protocolWithAuthority
|
||||
path.endsWith("/") -> url.protocolWithAuthority + path.removeSuffix("/")
|
||||
else -> url.protocolWithAuthority + path.take(path.lastIndexOf("/"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun start() {
|
||||
check(initialized.compareAndSet(expectedValue = false, newValue = true)) {
|
||||
"SSEClientTransport already started! If using Client class, note that connect() calls start() automatically."
|
||||
}
|
||||
|
||||
try {
|
||||
session = urlString?.let {
|
||||
client.sseSession(
|
||||
urlString = it,
|
||||
reconnectionTime = reconnectionTime,
|
||||
block = requestBuilder,
|
||||
)
|
||||
} ?: client.sseSession(
|
||||
reconnectionTime = reconnectionTime,
|
||||
block = requestBuilder,
|
||||
)
|
||||
scope = CoroutineScope(session.coroutineContext + SupervisorJob())
|
||||
|
||||
job = scope.launch(CoroutineName("SseMcpClientTransport.connect#${hashCode()}")) {
|
||||
collectMessages()
|
||||
}
|
||||
|
||||
endpoint.await()
|
||||
} catch (e: Exception) {
|
||||
closeResources()
|
||||
initialized.store(false)
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
@OptIn(ExperimentalCoroutinesApi::class)
|
||||
override suspend fun send(message: JSONRPCMessage) {
|
||||
check(initialized.load()) { "SseClientTransport is not initialized!" }
|
||||
check(job?.isActive == true) { "SseClientTransport is closed!" }
|
||||
check(endpoint.isCompleted) { "Not connected!" }
|
||||
|
||||
try {
|
||||
val response = client.post(endpoint.getCompleted()) {
|
||||
requestBuilder()
|
||||
headers.append(HttpHeaders.ContentType, ContentType.Application.Json)
|
||||
setBody(McpJson.encodeToString(message))
|
||||
}
|
||||
|
||||
if (!response.status.isSuccess()) {
|
||||
val text = response.bodyAsText()
|
||||
error("Error POSTing to endpoint (HTTP ${response.status}): $text")
|
||||
}
|
||||
} catch (e: Throwable) {
|
||||
_onError(e)
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun close() {
|
||||
check(initialized.load()) { "SseClientTransport is not initialized!" }
|
||||
closeResources()
|
||||
}
|
||||
|
||||
private suspend fun CoroutineScope.collectMessages() {
|
||||
try {
|
||||
session.incoming.collect { event ->
|
||||
ensureActive()
|
||||
|
||||
when (event.event) {
|
||||
"error" -> {
|
||||
val error = IllegalStateException("SSE error: ${event.data}")
|
||||
_onError(error)
|
||||
throw error
|
||||
}
|
||||
|
||||
"open" -> {
|
||||
// The connection is open, but we need to wait for the endpoint to be received.
|
||||
}
|
||||
|
||||
"endpoint" -> handleEndpoint(event.data.orEmpty())
|
||||
else -> handleMessage(event.data.orEmpty())
|
||||
}
|
||||
}
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: Throwable) {
|
||||
_onError(e)
|
||||
throw e
|
||||
} finally {
|
||||
closeResources()
|
||||
}
|
||||
}
|
||||
|
||||
private fun handleEndpoint(eventData: String) {
|
||||
try {
|
||||
val path = if (eventData.startsWith("/")) eventData.substring(1) else eventData
|
||||
val endpointUrl = Url("$baseUrl/$path")
|
||||
endpoint.complete(endpointUrl.toString())
|
||||
} catch (e: Throwable) {
|
||||
_onError(e)
|
||||
endpoint.completeExceptionally(e)
|
||||
throw e
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun handleMessage(data: String) {
|
||||
try {
|
||||
val message = McpJson.decodeFromString<JSONRPCMessage>(data)
|
||||
_onMessage(message)
|
||||
} catch (e: SerializationException) {
|
||||
_onError(e)
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun closeResources() {
|
||||
if (!initialized.compareAndSet(expectedValue = true, newValue = false)) return
|
||||
|
||||
job?.cancelAndJoin()
|
||||
try {
|
||||
if (::session.isInitialized) session.cancel()
|
||||
if (::scope.isInitialized) scope.cancel()
|
||||
endpoint.cancel()
|
||||
} catch (e: Throwable) {
|
||||
_onError(e)
|
||||
}
|
||||
|
||||
_onClose()
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,137 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.github.oshai.kotlinlogging.KotlinLogging
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage
|
||||
import io.modelcontextprotocol.kotlin.sdk.internal.IODispatcher
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.ReadBuffer
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.serializeMessage
|
||||
import kotlinx.coroutines.CoroutineName
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancelAndJoin
|
||||
import kotlinx.coroutines.channels.Channel
|
||||
import kotlinx.coroutines.channels.consumeEach
|
||||
import kotlinx.coroutines.isActive
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.io.Buffer
|
||||
import kotlinx.io.Sink
|
||||
import kotlinx.io.Source
|
||||
import kotlinx.io.buffered
|
||||
import kotlinx.io.readByteArray
|
||||
import kotlinx.io.writeString
|
||||
import kotlin.concurrent.atomics.AtomicBoolean
|
||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
import kotlin.coroutines.CoroutineContext
|
||||
|
||||
/**
|
||||
* A transport implementation for JSON-RPC communication that leverages standard input and output streams.
|
||||
*
|
||||
* This class reads from an input stream to process incoming JSON-RPC messages and writes JSON-RPC messages
|
||||
* to an output stream.
|
||||
*
|
||||
* @param input The input stream where messages are received.
|
||||
* @param output The output stream where messages are sent.
|
||||
*/
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
public class StdioClientTransport(
|
||||
private val input: Source,
|
||||
private val output: Sink
|
||||
) : AbstractTransport() {
|
||||
private val logger = KotlinLogging.logger {}
|
||||
private val ioCoroutineContext: CoroutineContext = IODispatcher
|
||||
private val scope by lazy {
|
||||
CoroutineScope(ioCoroutineContext + SupervisorJob())
|
||||
}
|
||||
private var job: Job? = null
|
||||
private val initialized: AtomicBoolean = AtomicBoolean(false)
|
||||
private val sendChannel = Channel<JSONRPCMessage>(Channel.UNLIMITED)
|
||||
private val readBuffer = ReadBuffer()
|
||||
|
||||
override suspend fun start() {
|
||||
if (!initialized.compareAndSet(expectedValue = false, newValue = true)) {
|
||||
error("StdioClientTransport already started!")
|
||||
}
|
||||
|
||||
logger.debug { "Starting StdioClientTransport..." }
|
||||
|
||||
val outputStream = output.buffered()
|
||||
|
||||
job = scope.launch(CoroutineName("StdioClientTransport.IO#${hashCode()}")) {
|
||||
val readJob = launch {
|
||||
logger.debug { "Read coroutine started." }
|
||||
try {
|
||||
input.use {
|
||||
while (isActive) {
|
||||
val buffer = Buffer()
|
||||
val bytesRead = input.readAtMostTo(buffer, 8192)
|
||||
if (bytesRead == -1L) break
|
||||
if (bytesRead > 0L) {
|
||||
readBuffer.append(buffer.readByteArray())
|
||||
processReadBuffer()
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
_onError.invoke(e)
|
||||
logger.error(e) { "Error reading from input stream" }
|
||||
}
|
||||
}
|
||||
|
||||
val writeJob = launch {
|
||||
logger.debug { "Write coroutine started." }
|
||||
try {
|
||||
sendChannel.consumeEach { message ->
|
||||
val json = serializeMessage(message)
|
||||
outputStream.writeString(json)
|
||||
outputStream.flush()
|
||||
}
|
||||
} catch (e: Throwable) {
|
||||
if (isActive) {
|
||||
_onError.invoke(e)
|
||||
logger.error(e) { "Error writing to output stream" }
|
||||
}
|
||||
} finally {
|
||||
output.close()
|
||||
}
|
||||
}
|
||||
|
||||
readJob.join()
|
||||
writeJob.cancelAndJoin()
|
||||
_onClose.invoke()
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun send(message: JSONRPCMessage) {
|
||||
if (!initialized.load()) {
|
||||
error("Transport not started")
|
||||
}
|
||||
|
||||
sendChannel.send(message)
|
||||
}
|
||||
|
||||
override suspend fun close() {
|
||||
if (!initialized.compareAndSet(expectedValue = true, newValue = false)) {
|
||||
error("Transport is already closed")
|
||||
}
|
||||
job?.cancelAndJoin()
|
||||
input.close()
|
||||
output.close()
|
||||
readBuffer.clear()
|
||||
sendChannel.close()
|
||||
_onClose.invoke()
|
||||
}
|
||||
|
||||
private suspend fun processReadBuffer() {
|
||||
while (true) {
|
||||
val msg = readBuffer.readMessage() ?: break
|
||||
try {
|
||||
_onMessage.invoke(msg)
|
||||
} catch (e: Throwable) {
|
||||
_onError.invoke(e)
|
||||
logger.error(e) { "Error processing message." }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,337 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.github.oshai.kotlinlogging.KotlinLogging
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.plugins.ClientRequestException
|
||||
import io.ktor.client.plugins.sse.ClientSSESession
|
||||
import io.ktor.client.plugins.sse.sseSession
|
||||
import io.ktor.client.request.HttpRequestBuilder
|
||||
import io.ktor.client.request.accept
|
||||
import io.ktor.client.request.delete
|
||||
import io.ktor.client.request.headers
|
||||
import io.ktor.client.request.post
|
||||
import io.ktor.client.request.setBody
|
||||
import io.ktor.client.statement.HttpResponse
|
||||
import io.ktor.client.statement.bodyAsChannel
|
||||
import io.ktor.client.statement.bodyAsText
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpHeaders
|
||||
import io.ktor.http.HttpMethod
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.contentType
|
||||
import io.ktor.http.isSuccess
|
||||
import io.ktor.utils.io.readUTF8Line
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCNotification
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCResponse
|
||||
import io.modelcontextprotocol.kotlin.sdk.RequestId
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.McpJson
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CoroutineName
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.cancelAndJoin
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlin.concurrent.atomics.AtomicBoolean
|
||||
import kotlin.concurrent.atomics.ExperimentalAtomicApi
|
||||
import kotlin.time.Duration
|
||||
|
||||
private val logger = KotlinLogging.logger {}
|
||||
|
||||
private const val MCP_SESSION_ID_HEADER = "mcp-session-id"
|
||||
private const val MCP_PROTOCOL_VERSION_HEADER = "mcp-protocol-version"
|
||||
private const val MCP_RESUMPTION_TOKEN_HEADER = "Last-Event-ID"
|
||||
|
||||
/**
|
||||
* Error class for Streamable HTTP transport errors.
|
||||
*/
|
||||
public class StreamableHttpError(
|
||||
public val code: Int? = null,
|
||||
message: String? = null
|
||||
) : Exception("Streamable HTTP error: $message")
|
||||
|
||||
/**
|
||||
* Client transport for Streamable HTTP: this implements the MCP Streamable HTTP transport specification.
|
||||
* It will connect to a server using HTTP POST for sending messages and HTTP GET with Server-Sent Events
|
||||
* for receiving messages.
|
||||
*/
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
public class StreamableHttpClientTransport(
|
||||
private val client: HttpClient,
|
||||
private val url: String,
|
||||
private val reconnectionTime: Duration? = null,
|
||||
private val requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
) : AbstractTransport() {
|
||||
|
||||
public var sessionId: String? = null
|
||||
private set
|
||||
public var protocolVersion: String? = null
|
||||
|
||||
private val initialized: AtomicBoolean = AtomicBoolean(false)
|
||||
|
||||
private var sseSession: ClientSSESession? = null
|
||||
private var sseJob: Job? = null
|
||||
|
||||
private val scope by lazy { CoroutineScope(SupervisorJob() + Dispatchers.Default) }
|
||||
|
||||
private var lastEventId: String? = null
|
||||
|
||||
override suspend fun start() {
|
||||
if (!initialized.compareAndSet(expectedValue = false, newValue = true)) {
|
||||
error("StreamableHttpClientTransport already started!")
|
||||
}
|
||||
logger.debug { "Client transport starting..." }
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a single message with optional resumption support
|
||||
*/
|
||||
override suspend fun send(message: JSONRPCMessage) {
|
||||
send(message, null)
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends one or more messages with optional resumption support.
|
||||
* This is the main send method that matches the TypeScript implementation.
|
||||
*/
|
||||
public suspend fun send(
|
||||
message: JSONRPCMessage,
|
||||
resumptionToken: String?,
|
||||
onResumptionToken: ((String) -> Unit)? = null
|
||||
) {
|
||||
logger.debug { "Client sending message via POST to $url: ${McpJson.encodeToString(message)}" }
|
||||
|
||||
// If we have a resumption token, reconnect the SSE stream with it
|
||||
resumptionToken?.let { token ->
|
||||
startSseSession(
|
||||
resumptionToken = token, onResumptionToken = onResumptionToken,
|
||||
replayMessageId = if (message is JSONRPCRequest) message.id else null
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
val jsonBody = McpJson.encodeToString(message)
|
||||
val response = client.post(url) {
|
||||
applyCommonHeaders(this)
|
||||
headers.append(HttpHeaders.Accept, "${ContentType.Application.Json}, ${ContentType.Text.EventStream}")
|
||||
contentType(ContentType.Application.Json)
|
||||
setBody(jsonBody)
|
||||
requestBuilder()
|
||||
}
|
||||
|
||||
response.headers[MCP_SESSION_ID_HEADER]?.let { sessionId = it }
|
||||
|
||||
if (response.status == HttpStatusCode.Accepted) {
|
||||
if (message is JSONRPCNotification && message.method == "notifications/initialized") {
|
||||
startSseSession(onResumptionToken = onResumptionToken)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if (!response.status.isSuccess()) {
|
||||
val error = StreamableHttpError(response.status.value, response.bodyAsText())
|
||||
_onError(error)
|
||||
throw error
|
||||
}
|
||||
|
||||
when (response.contentType()?.withoutParameters()) {
|
||||
ContentType.Application.Json -> response.bodyAsText().takeIf { it.isNotEmpty() }?.let { json ->
|
||||
runCatching { McpJson.decodeFromString<JSONRPCMessage>(json) }
|
||||
.onSuccess { _onMessage(it) }
|
||||
.onFailure(_onError)
|
||||
}
|
||||
|
||||
ContentType.Text.EventStream -> handleInlineSse(
|
||||
response, onResumptionToken = onResumptionToken,
|
||||
replayMessageId = if (message is JSONRPCRequest) message.id else null
|
||||
)
|
||||
|
||||
else -> {
|
||||
val body = response.bodyAsText()
|
||||
if (response.contentType() == null && body.isBlank()) return
|
||||
|
||||
val ct = response.contentType()?.toString() ?: "<none>"
|
||||
val error = StreamableHttpError(-1, "Unexpected content type: $$ct")
|
||||
_onError(error)
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun close() {
|
||||
if (!initialized.load()) return // Already closed or never started
|
||||
logger.debug { "Client transport closing." }
|
||||
|
||||
try {
|
||||
// Try to terminate session if we have one
|
||||
terminateSession()
|
||||
|
||||
sseSession?.cancel()
|
||||
sseJob?.cancelAndJoin()
|
||||
scope.cancel()
|
||||
} catch (_: Exception) {
|
||||
// Ignore errors during cleanup
|
||||
} finally {
|
||||
initialized.store(false)
|
||||
_onClose()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Terminates the current session by sending a DELETE request to the server.
|
||||
*/
|
||||
public suspend fun terminateSession() {
|
||||
if (sessionId == null) return
|
||||
logger.debug { "Terminating session: $sessionId" }
|
||||
val response = client.delete(url) {
|
||||
applyCommonHeaders(this)
|
||||
requestBuilder()
|
||||
}
|
||||
|
||||
// 405 means server doesn't support explicit session termination
|
||||
if (!response.status.isSuccess() && response.status != HttpStatusCode.MethodNotAllowed) {
|
||||
val error = StreamableHttpError(
|
||||
response.status.value,
|
||||
"Failed to terminate session: ${response.status.description}"
|
||||
)
|
||||
logger.error(error) { "Failed to terminate session" }
|
||||
_onError(error)
|
||||
throw error
|
||||
}
|
||||
|
||||
sessionId = null
|
||||
lastEventId = null
|
||||
logger.debug { "Session terminated successfully" }
|
||||
}
|
||||
|
||||
private suspend fun startSseSession(
|
||||
resumptionToken: String? = null,
|
||||
replayMessageId: RequestId? = null,
|
||||
onResumptionToken: ((String) -> Unit)? = null
|
||||
) {
|
||||
sseSession?.cancel()
|
||||
sseJob?.cancelAndJoin()
|
||||
|
||||
logger.debug { "Client attempting to start SSE session at url: $url" }
|
||||
try {
|
||||
sseSession = client.sseSession(
|
||||
urlString = url,
|
||||
reconnectionTime = reconnectionTime,
|
||||
) {
|
||||
method = HttpMethod.Get
|
||||
applyCommonHeaders(this)
|
||||
accept(ContentType.Text.EventStream)
|
||||
(resumptionToken ?: lastEventId)?.let { headers.append(MCP_RESUMPTION_TOKEN_HEADER, it) }
|
||||
requestBuilder()
|
||||
}
|
||||
logger.debug { "Client SSE session started successfully." }
|
||||
} catch (e: ClientRequestException) {
|
||||
if (e.response.status == HttpStatusCode.MethodNotAllowed) {
|
||||
logger.info { "Server returned 405 for GET/SSE, stream disabled." }
|
||||
return
|
||||
}
|
||||
_onError(e)
|
||||
throw e
|
||||
}
|
||||
|
||||
sseJob = scope.launch(CoroutineName("StreamableHttpTransport.collect#${hashCode()}")) {
|
||||
sseSession?.let { collectSse(it, replayMessageId, onResumptionToken) }
|
||||
}
|
||||
}
|
||||
|
||||
private fun applyCommonHeaders(builder: HttpRequestBuilder) {
|
||||
builder.headers {
|
||||
sessionId?.let { append(MCP_SESSION_ID_HEADER, it) }
|
||||
protocolVersion?.let { append(MCP_PROTOCOL_VERSION_HEADER, it) }
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun collectSse(
|
||||
session: ClientSSESession,
|
||||
replayMessageId: RequestId?,
|
||||
onResumptionToken: ((String) -> Unit)?
|
||||
) {
|
||||
try {
|
||||
session.incoming.collect { event ->
|
||||
event.id?.let {
|
||||
lastEventId = it
|
||||
onResumptionToken?.invoke(it)
|
||||
}
|
||||
logger.trace { "Client received SSE event: event=${event.event}, data=${event.data}, id=${event.id}" }
|
||||
when (event.event) {
|
||||
null, "message" ->
|
||||
event.data?.takeIf { it.isNotEmpty() }?.let { json ->
|
||||
runCatching { McpJson.decodeFromString<JSONRPCMessage>(json) }
|
||||
.onSuccess { msg ->
|
||||
if (replayMessageId != null && msg is JSONRPCResponse) {
|
||||
_onMessage(msg.copy(id = replayMessageId))
|
||||
} else {
|
||||
_onMessage(msg)
|
||||
}
|
||||
}
|
||||
.onFailure(_onError)
|
||||
}
|
||||
|
||||
"error" -> _onError(StreamableHttpError(null, event.data))
|
||||
}
|
||||
}
|
||||
} catch (_: CancellationException) {
|
||||
// ignore
|
||||
} catch (t: Throwable) {
|
||||
_onError(t)
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun handleInlineSse(
|
||||
response: HttpResponse,
|
||||
replayMessageId: RequestId?,
|
||||
onResumptionToken: ((String) -> Unit)?
|
||||
) {
|
||||
logger.trace { "Handling inline SSE from POST response" }
|
||||
val channel = response.bodyAsChannel()
|
||||
|
||||
val sb = StringBuilder()
|
||||
var id: String? = null
|
||||
var eventName: String? = null
|
||||
|
||||
suspend fun dispatch(data: String) {
|
||||
id?.let {
|
||||
lastEventId = it
|
||||
onResumptionToken?.invoke(it)
|
||||
}
|
||||
if (eventName == null || eventName == "message") {
|
||||
runCatching { McpJson.decodeFromString<JSONRPCMessage>(data) }
|
||||
.onSuccess { msg ->
|
||||
if (replayMessageId != null && msg is JSONRPCResponse) {
|
||||
_onMessage(msg.copy(id = replayMessageId))
|
||||
} else {
|
||||
_onMessage(msg)
|
||||
}
|
||||
}
|
||||
.onFailure(_onError)
|
||||
}
|
||||
// reset
|
||||
id = null
|
||||
eventName = null
|
||||
sb.clear()
|
||||
}
|
||||
|
||||
while (!channel.isClosedForRead) {
|
||||
val line = channel.readUTF8Line() ?: break
|
||||
if (line.isEmpty()) {
|
||||
dispatch(sb.toString())
|
||||
continue
|
||||
}
|
||||
when {
|
||||
line.startsWith("id:") -> id = line.substringAfter("id:").trim()
|
||||
line.startsWith("event:") -> eventName = line.substringAfter("event:").trim()
|
||||
line.startsWith("data:") -> sb.append(line.substringAfter("data:").trim())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.request.HttpRequestBuilder
|
||||
import io.modelcontextprotocol.kotlin.sdk.Implementation
|
||||
import io.modelcontextprotocol.kotlin.sdk.LIB_VERSION
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.IMPLEMENTATION_NAME
|
||||
import kotlin.time.Duration
|
||||
|
||||
/**
|
||||
* Returns a new Streamable HTTP transport for the Model Context Protocol using the provided HttpClient.
|
||||
*
|
||||
* @param url URL of the MCP server.
|
||||
* @param reconnectionTime Optional duration to wait before attempting to reconnect.
|
||||
* @param requestBuilder Optional lambda to configure the HTTP request.
|
||||
* @return A [StreamableHttpClientTransport] configured for MCP communication.
|
||||
*/
|
||||
public fun HttpClient.mcpStreamableHttpTransport(
|
||||
url: String,
|
||||
reconnectionTime: Duration? = null,
|
||||
requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
): StreamableHttpClientTransport =
|
||||
StreamableHttpClientTransport(this, url, reconnectionTime, requestBuilder = requestBuilder)
|
||||
|
||||
/**
|
||||
* Creates and connects an MCP client over Streamable HTTP using the provided HttpClient.
|
||||
*
|
||||
* @param url URL of the MCP server.
|
||||
* @param reconnectionTime Optional duration to wait before attempting to reconnect.
|
||||
* @param requestBuilder Optional lambda to configure the HTTP request.
|
||||
* @return A connected [Client] ready for MCP communication.
|
||||
*/
|
||||
public suspend fun HttpClient.mcpStreamableHttp(
|
||||
url: String,
|
||||
reconnectionTime: Duration? = null,
|
||||
requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
): Client {
|
||||
val transport = mcpStreamableHttpTransport(url, reconnectionTime, requestBuilder)
|
||||
val client = Client(Implementation(name = IMPLEMENTATION_NAME, version = LIB_VERSION))
|
||||
client.connect(transport)
|
||||
return client
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.plugins.websocket.webSocketSession
|
||||
import io.ktor.client.request.HttpRequestBuilder
|
||||
import io.ktor.client.request.header
|
||||
import io.ktor.http.HttpHeaders
|
||||
import io.ktor.websocket.WebSocketSession
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.MCP_SUBPROTOCOL
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.WebSocketMcpTransport
|
||||
import kotlin.properties.Delegates
|
||||
|
||||
/**
|
||||
* Client transport for WebSocket: this will connect to a server over the WebSocket protocol.
|
||||
*/
|
||||
public class WebSocketClientTransport(
|
||||
private val client: HttpClient,
|
||||
private val urlString: String?,
|
||||
private val requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
) : WebSocketMcpTransport() {
|
||||
override var session: WebSocketSession by Delegates.notNull()
|
||||
|
||||
override suspend fun initializeSession() {
|
||||
session = urlString?.let {
|
||||
client.webSocketSession(it) {
|
||||
requestBuilder()
|
||||
|
||||
header(HttpHeaders.SecWebSocketProtocol, MCP_SUBPROTOCOL)
|
||||
}
|
||||
} ?: client.webSocketSession {
|
||||
requestBuilder()
|
||||
|
||||
header(HttpHeaders.SecWebSocketProtocol, MCP_SUBPROTOCOL)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.request.HttpRequestBuilder
|
||||
import io.modelcontextprotocol.kotlin.sdk.Implementation
|
||||
import io.modelcontextprotocol.kotlin.sdk.LIB_VERSION
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.IMPLEMENTATION_NAME
|
||||
|
||||
/**
|
||||
* Returns a new WebSocket transport for the Model Context Protocol using the provided HttpClient.
|
||||
*
|
||||
* @param urlString Optional URL of the MCP server.
|
||||
* @param requestBuilder Optional lambda to configure the HTTP request.
|
||||
* @return A [WebSocketClientTransport] configured for MCP communication.
|
||||
*/
|
||||
public fun HttpClient.mcpWebSocketTransport(
|
||||
urlString: String? = null,
|
||||
requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
): WebSocketClientTransport = WebSocketClientTransport(this, urlString, requestBuilder)
|
||||
|
||||
/**
|
||||
* Creates and connects an MCP client over WebSocket using the provided HttpClient.
|
||||
*
|
||||
* @param urlString Optional URL of the MCP server.
|
||||
* @param requestBuilder Optional lambda to configure the HTTP request.
|
||||
* @return A connected [Client] ready for MCP communication.
|
||||
*/
|
||||
public suspend fun HttpClient.mcpWebSocket(
|
||||
urlString: String? = null,
|
||||
requestBuilder: HttpRequestBuilder.() -> Unit = {},
|
||||
): Client {
|
||||
val transport = mcpWebSocketTransport(urlString, requestBuilder)
|
||||
val client = Client(
|
||||
Implementation(
|
||||
name = IMPLEMENTATION_NAME,
|
||||
version = LIB_VERSION
|
||||
)
|
||||
)
|
||||
client.connect(transport)
|
||||
return client
|
||||
}
|
||||
|
|
@ -0,0 +1,365 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.engine.mock.MockEngine
|
||||
import io.ktor.client.engine.mock.MockRequestHandler
|
||||
import io.ktor.client.engine.mock.respond
|
||||
import io.ktor.client.plugins.sse.SSE
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpHeaders
|
||||
import io.ktor.http.HttpMethod
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.content.TextContent
|
||||
import io.ktor.http.headersOf
|
||||
import io.ktor.utils.io.ByteReadChannel
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCNotification
|
||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.RequestId
|
||||
import io.modelcontextprotocol.kotlin.sdk.shared.McpJson
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
import kotlinx.serialization.json.JsonPrimitive
|
||||
import kotlinx.serialization.json.buildJsonObject
|
||||
import kotlin.test.Ignore
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
||||
class StreamableHttpClientTransportTest {
|
||||
|
||||
private fun createTransport(handler: MockRequestHandler): StreamableHttpClientTransport {
|
||||
val mockEngine = MockEngine(handler)
|
||||
val httpClient = HttpClient(mockEngine) {
|
||||
install(SSE) {
|
||||
reconnectionTime = 1.seconds
|
||||
}
|
||||
}
|
||||
|
||||
return StreamableHttpClientTransport(httpClient, url = "http://localhost:8080/mcp")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testSendJsonRpcMessage() = runTest {
|
||||
val message = JSONRPCRequest(
|
||||
id = RequestId.StringId("test-id"),
|
||||
method = "test",
|
||||
params = buildJsonObject { }
|
||||
)
|
||||
|
||||
val transport = createTransport { request ->
|
||||
assertEquals(HttpMethod.Post, request.method)
|
||||
assertEquals("http://localhost:8080/mcp", request.url.toString())
|
||||
assertEquals(ContentType.Application.Json, request.body.contentType)
|
||||
|
||||
val body = (request.body as TextContent).text
|
||||
val decodedMessage = McpJson.decodeFromString<JSONRPCMessage>(body)
|
||||
assertEquals(message, decodedMessage)
|
||||
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted
|
||||
)
|
||||
}
|
||||
|
||||
transport.start()
|
||||
transport.send(message)
|
||||
transport.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testStoreSessionId() = runTest {
|
||||
val initMessage = JSONRPCRequest(
|
||||
id = RequestId.StringId("test-id"),
|
||||
method = "initialize",
|
||||
params = buildJsonObject {
|
||||
put("clientInfo", buildJsonObject {
|
||||
put("name", JsonPrimitive("test-client"))
|
||||
put("version", JsonPrimitive("1.0"))
|
||||
})
|
||||
put("protocolVersion", JsonPrimitive("2025-06-18"))
|
||||
}
|
||||
)
|
||||
|
||||
val transport = createTransport { request ->
|
||||
when (val msg = McpJson.decodeFromString<JSONRPCMessage>((request.body as TextContent).text)) {
|
||||
is JSONRPCRequest if msg.method == "initialize" -> respond(
|
||||
content = "", status = HttpStatusCode.OK,
|
||||
headers = headersOf("mcp-session-id", "test-session-id")
|
||||
)
|
||||
|
||||
is JSONRPCNotification if msg.method == "test" -> {
|
||||
assertEquals("test-session-id", request.headers["mcp-session-id"])
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted
|
||||
)
|
||||
}
|
||||
|
||||
else -> error("Unexpected message: $msg")
|
||||
}
|
||||
}
|
||||
|
||||
transport.start()
|
||||
transport.send(initMessage)
|
||||
|
||||
assertEquals("test-session-id", transport.sessionId)
|
||||
|
||||
transport.send(JSONRPCNotification(method = "test"))
|
||||
|
||||
transport.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testTerminateSession() = runTest {
|
||||
// transport.sessionId = "test-session-id"
|
||||
|
||||
val transport = createTransport { request ->
|
||||
assertEquals(HttpMethod.Delete, request.method)
|
||||
assertEquals("test-session-id", request.headers["mcp-session-id"])
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.OK
|
||||
)
|
||||
}
|
||||
|
||||
transport.start()
|
||||
transport.terminateSession()
|
||||
|
||||
assertNull(transport.sessionId)
|
||||
transport.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testTerminateSessionHandle405() = runTest {
|
||||
// transport.sessionId = "test-session-id"
|
||||
|
||||
val transport = createTransport { request ->
|
||||
assertEquals(HttpMethod.Delete, request.method)
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.MethodNotAllowed
|
||||
)
|
||||
}
|
||||
|
||||
transport.start()
|
||||
// Should not throw for 405
|
||||
transport.terminateSession()
|
||||
|
||||
// Session ID should still be cleared
|
||||
assertNull(transport.sessionId)
|
||||
transport.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testProtocolVersionHeader() = runTest {
|
||||
val transport = createTransport { request ->
|
||||
assertEquals("2025-06-18", request.headers["mcp-protocol-version"])
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted
|
||||
)
|
||||
}
|
||||
transport.protocolVersion = "2025-06-18"
|
||||
|
||||
transport.start()
|
||||
transport.send(JSONRPCNotification(method = "test"))
|
||||
transport.close()
|
||||
}
|
||||
|
||||
@Ignore //Engine doesn't support SSECapability: https://youtrack.jetbrains.com/issue/KTOR-8177/MockEngine-Add-SSE-support
|
||||
@Test
|
||||
fun testNotificationSchemaE2E() = runTest {
|
||||
val receivedMessages = mutableListOf<JSONRPCMessage>()
|
||||
var sseStarted = false
|
||||
|
||||
val transport = createTransport { request ->
|
||||
when (request.method) {
|
||||
HttpMethod.Post if request.body.toString().contains("notifications/initialized") -> {
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted,
|
||||
headers = headersOf("mcp-session-id", "notification-test-session")
|
||||
)
|
||||
}
|
||||
|
||||
// Handle SSE connection
|
||||
HttpMethod.Get -> {
|
||||
sseStarted = true
|
||||
val sseContent = buildString {
|
||||
// Server sends various notifications
|
||||
appendLine("event: message")
|
||||
appendLine("id: 1")
|
||||
appendLine("""data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":50,"total":100}}""")
|
||||
appendLine()
|
||||
|
||||
appendLine("event: message")
|
||||
appendLine("id: 2")
|
||||
appendLine("""data: {"jsonrpc":"2.0","method":"notifications/resources/list_changed"}""")
|
||||
appendLine()
|
||||
|
||||
appendLine("event: message")
|
||||
appendLine("id: 3")
|
||||
appendLine("""data: {"jsonrpc":"2.0","method":"notifications/tools/list_changed"}""")
|
||||
appendLine()
|
||||
}
|
||||
respond(
|
||||
content = ByteReadChannel(sseContent),
|
||||
status = HttpStatusCode.OK,
|
||||
headers = headersOf(
|
||||
HttpHeaders.ContentType, ContentType.Text.EventStream.toString()
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
// Handle regular notifications
|
||||
HttpMethod.Post -> {
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted
|
||||
)
|
||||
}
|
||||
|
||||
else -> respond("", HttpStatusCode.OK)
|
||||
}
|
||||
}
|
||||
|
||||
transport.onMessage { message ->
|
||||
receivedMessages.add(message)
|
||||
}
|
||||
|
||||
transport.start()
|
||||
|
||||
// Test 1: Send initialized notification to trigger SSE
|
||||
val initializedNotification = JSONRPCNotification(
|
||||
method = "notifications/initialized",
|
||||
params = buildJsonObject {
|
||||
put("protocolVersion", JsonPrimitive("1.0"))
|
||||
put("capabilities", buildJsonObject {
|
||||
put("tools", JsonPrimitive(true))
|
||||
put("resources", JsonPrimitive(true))
|
||||
})
|
||||
}
|
||||
)
|
||||
|
||||
transport.send(initializedNotification)
|
||||
|
||||
// Verify SSE was triggered
|
||||
assertTrue(sseStarted, "SSE should start after initialized notification")
|
||||
|
||||
// Test 2: Verify received notifications
|
||||
assertEquals(3, receivedMessages.size)
|
||||
assertTrue(receivedMessages.all { it is JSONRPCNotification })
|
||||
|
||||
val notifications = receivedMessages.filterIsInstance<JSONRPCNotification>()
|
||||
|
||||
// Verify progress notification
|
||||
val progressNotif = notifications[0]
|
||||
assertEquals("notifications/progress", progressNotif.method)
|
||||
val progressParams = progressNotif.params as JsonObject
|
||||
assertEquals("upload-123", (progressParams["progressToken"] as JsonPrimitive).content)
|
||||
assertEquals(50, (progressParams["progress"] as JsonPrimitive).content.toInt())
|
||||
|
||||
// Verify list changed notifications
|
||||
assertEquals("notifications/resources/list_changed", notifications[1].method)
|
||||
assertEquals("notifications/tools/list_changed", notifications[2].method)
|
||||
|
||||
// Test 3: Send various client notifications
|
||||
val clientNotifications = listOf(
|
||||
JSONRPCNotification(
|
||||
method = "notifications/progress",
|
||||
params = buildJsonObject {
|
||||
put("progressToken", JsonPrimitive("download-456"))
|
||||
put("progress", JsonPrimitive(75))
|
||||
}
|
||||
),
|
||||
JSONRPCNotification(
|
||||
method = "notifications/cancelled",
|
||||
params = buildJsonObject {
|
||||
put("requestId", JsonPrimitive("req-789"))
|
||||
put("reason", JsonPrimitive("user_cancelled"))
|
||||
}
|
||||
),
|
||||
JSONRPCNotification(
|
||||
method = "notifications/message",
|
||||
params = buildJsonObject {
|
||||
put("level", JsonPrimitive("info"))
|
||||
put("message", JsonPrimitive("Operation completed"))
|
||||
put("data", buildJsonObject {
|
||||
put("duration", JsonPrimitive(1234))
|
||||
})
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
// Send all client notifications
|
||||
clientNotifications.forEach { notification ->
|
||||
transport.send(notification)
|
||||
}
|
||||
|
||||
// Verify session ID is maintained
|
||||
assertEquals("notification-test-session", transport.sessionId)
|
||||
transport.close()
|
||||
}
|
||||
|
||||
@Ignore // Engine doesn't support SSECapability: https://youtrack.jetbrains.com/issue/KTOR-8177/MockEngine-Add-SSE-support
|
||||
@Test
|
||||
fun testNotificationWithResumptionToken() = runTest {
|
||||
var resumptionTokenReceived: String? = null
|
||||
var lastEventIdSent: String? = null
|
||||
|
||||
val transport = createTransport { request ->
|
||||
// Capture Last-Event-ID header
|
||||
lastEventIdSent = request.headers["Last-Event-ID"]
|
||||
|
||||
when (request.method) {
|
||||
HttpMethod.Get -> {
|
||||
val sseContent = buildString {
|
||||
appendLine("event: message")
|
||||
appendLine("id: resume-100")
|
||||
appendLine("""data: {"jsonrpc":"2.0","method":"notifications/resumed","params":{"fromToken":"${lastEventIdSent}"}}""")
|
||||
appendLine()
|
||||
}
|
||||
respond(
|
||||
content = ByteReadChannel(sseContent),
|
||||
status = HttpStatusCode.OK,
|
||||
headers = headersOf(
|
||||
HttpHeaders.ContentType, ContentType.Text.EventStream.toString()
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
else -> respond("", HttpStatusCode.Accepted)
|
||||
}
|
||||
}
|
||||
|
||||
transport.start()
|
||||
|
||||
// Send notification with resumption token
|
||||
transport.send(
|
||||
message = JSONRPCNotification(
|
||||
method = "notifications/test",
|
||||
params = buildJsonObject {
|
||||
put("data", JsonPrimitive("test-data"))
|
||||
}
|
||||
),
|
||||
resumptionToken = "previous-token-99",
|
||||
onResumptionToken = { token ->
|
||||
resumptionTokenReceived = token
|
||||
}
|
||||
)
|
||||
|
||||
// Wait for response
|
||||
delay(1.seconds)
|
||||
|
||||
// Verify resumption token was sent in header
|
||||
assertEquals("previous-token-99", lastEventIdSent)
|
||||
|
||||
// Verify new resumption token was received
|
||||
assertEquals("resume-100", resumptionTokenReceived)
|
||||
transport.close()
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue