Add ktlint (#213)
* Add ktlint * Simplify `initClient` function and clean imports in SseIntegrationTest.
This commit is contained in:
Pavel Gorgulov
GitHub
parent
b0f075900a
commit
91cf1205af
8 changed files with 800 additions and 758 deletions
|
|
@ -80,10 +80,8 @@ public class ClientOptions(
|
|||
* @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) {
|
||||
public open class Client(private val clientInfo: Implementation, options: ClientOptions = ClientOptions()) :
|
||||
Protocol(options) {
|
||||
|
||||
/**
|
||||
* Retrieves the server's reported capabilities after the initialization process completes.
|
||||
|
|
@ -144,13 +142,13 @@ public open class Client(
|
|||
val message = InitializeRequest(
|
||||
protocolVersion = LATEST_PROTOCOL_VERSION,
|
||||
capabilities = capabilities,
|
||||
clientInfo = clientInfo
|
||||
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}"
|
||||
"Server's protocol version is not supported: ${result.protocolVersion}",
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -165,11 +163,9 @@ public open class Client(
|
|||
}
|
||||
|
||||
throw error
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
override fun assertCapabilityForMethod(method: Method) {
|
||||
when (method) {
|
||||
Method.Defined.LoggingSetLevel -> {
|
||||
|
|
@ -181,7 +177,7 @@ public open class Client(
|
|||
Method.Defined.PromptsGet,
|
||||
Method.Defined.PromptsList,
|
||||
Method.Defined.CompletionComplete,
|
||||
-> {
|
||||
-> {
|
||||
if (serverCapabilities?.prompts == null) {
|
||||
throw IllegalStateException("Server does not support prompts (required for $method)")
|
||||
}
|
||||
|
|
@ -192,20 +188,20 @@ public open class Client(
|
|||
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)"
|
||||
"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)")
|
||||
}
|
||||
|
|
@ -213,7 +209,7 @@ public open class Client(
|
|||
|
||||
Method.Defined.Initialize,
|
||||
Method.Defined.Ping,
|
||||
-> {
|
||||
-> {
|
||||
// No specific capability required
|
||||
}
|
||||
|
||||
|
|
@ -228,7 +224,7 @@ public open class Client(
|
|||
Method.Defined.NotificationsRootsListChanged -> {
|
||||
if (capabilities.roots?.listChanged != true) {
|
||||
throw IllegalStateException(
|
||||
"Client does not support roots list changed notifications (required for $method)"
|
||||
"Client does not support roots list changed notifications (required for $method)",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -236,7 +232,7 @@ public open class Client(
|
|||
Method.Defined.NotificationsInitialized,
|
||||
Method.Defined.NotificationsCancelled,
|
||||
Method.Defined.NotificationsProgress,
|
||||
-> {
|
||||
-> {
|
||||
// Always allowed
|
||||
}
|
||||
|
||||
|
|
@ -251,7 +247,7 @@ public open class Client(
|
|||
Method.Defined.SamplingCreateMessage -> {
|
||||
if (capabilities.sampling == null) {
|
||||
throw IllegalStateException(
|
||||
"Client does not support sampling capability (required for $method)"
|
||||
"Client does not support sampling capability (required for $method)",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -259,7 +255,7 @@ public open class Client(
|
|||
Method.Defined.RootsList -> {
|
||||
if (capabilities.roots == null) {
|
||||
throw IllegalStateException(
|
||||
"Client does not support roots capability (required for $method)"
|
||||
"Client does not support roots capability (required for $method)",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -267,7 +263,7 @@ public open class Client(
|
|||
Method.Defined.ElicitationCreate -> {
|
||||
if (capabilities.elicitation == null) {
|
||||
throw IllegalStateException(
|
||||
"Client does not support elicitation capability (required for $method)"
|
||||
"Client does not support elicitation capability (required for $method)",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -280,16 +276,14 @@ public open class Client(
|
|||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 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)
|
||||
}
|
||||
public suspend fun ping(options: RequestOptions? = null): EmptyRequestResult =
|
||||
request<EmptyRequestResult>(PingRequest(), options)
|
||||
|
||||
/**
|
||||
* Sends a completion request to the server, typically to generate or complete some content.
|
||||
|
|
@ -299,9 +293,8 @@ public open class Client(
|
|||
* @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)
|
||||
}
|
||||
public suspend fun complete(params: CompleteRequest, options: RequestOptions? = null): CompleteResult =
|
||||
request(params, options)
|
||||
|
||||
/**
|
||||
* Sets the logging level on the server.
|
||||
|
|
@ -310,9 +303,8 @@ public open class Client(
|
|||
* @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)
|
||||
}
|
||||
public suspend fun setLoggingLevel(level: LoggingLevel, options: RequestOptions? = null): EmptyRequestResult =
|
||||
request<EmptyRequestResult>(SetLevelRequest(level), options)
|
||||
|
||||
/**
|
||||
* Retrieves a prompt by name from the server.
|
||||
|
|
@ -322,9 +314,8 @@ public open class Client(
|
|||
* @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)
|
||||
}
|
||||
public suspend fun getPrompt(request: GetPromptRequest, options: RequestOptions? = null): GetPromptResult =
|
||||
request(request, options)
|
||||
|
||||
/**
|
||||
* Lists all available prompts from the server.
|
||||
|
|
@ -337,9 +328,7 @@ public open class Client(
|
|||
public suspend fun listPrompts(
|
||||
request: ListPromptsRequest = ListPromptsRequest(),
|
||||
options: RequestOptions? = null,
|
||||
): ListPromptsResult {
|
||||
return request(request, options)
|
||||
}
|
||||
): ListPromptsResult = request(request, options)
|
||||
|
||||
/**
|
||||
* Lists all available resources from the server.
|
||||
|
|
@ -352,9 +341,7 @@ public open class Client(
|
|||
public suspend fun listResources(
|
||||
request: ListResourcesRequest = ListResourcesRequest(),
|
||||
options: RequestOptions? = null,
|
||||
): ListResourcesResult {
|
||||
return request(request, options)
|
||||
}
|
||||
): ListResourcesResult = request(request, options)
|
||||
|
||||
/**
|
||||
* Lists resource templates available on the server.
|
||||
|
|
@ -367,9 +354,7 @@ public open class Client(
|
|||
public suspend fun listResourceTemplates(
|
||||
request: ListResourceTemplatesRequest,
|
||||
options: RequestOptions? = null,
|
||||
): ListResourceTemplatesResult {
|
||||
return request(request, options)
|
||||
}
|
||||
): ListResourceTemplatesResult = request(request, options)
|
||||
|
||||
/**
|
||||
* Reads a resource from the server by its URI.
|
||||
|
|
@ -382,9 +367,7 @@ public open class Client(
|
|||
public suspend fun readResource(
|
||||
request: ReadResourceRequest,
|
||||
options: RequestOptions? = null,
|
||||
): ReadResourceResult {
|
||||
return request(request, options)
|
||||
}
|
||||
): ReadResourceResult = request(request, options)
|
||||
|
||||
/**
|
||||
* Subscribes to resource changes on the server.
|
||||
|
|
@ -396,9 +379,7 @@ public open class Client(
|
|||
public suspend fun subscribeResource(
|
||||
request: SubscribeRequest,
|
||||
options: RequestOptions? = null,
|
||||
): EmptyRequestResult {
|
||||
return request(request, options)
|
||||
}
|
||||
): EmptyRequestResult = request(request, options)
|
||||
|
||||
/**
|
||||
* Unsubscribes from resource changes on the server.
|
||||
|
|
@ -410,9 +391,7 @@ public open class Client(
|
|||
public suspend fun unsubscribeResource(
|
||||
request: UnsubscribeRequest,
|
||||
options: RequestOptions? = null,
|
||||
): EmptyRequestResult {
|
||||
return request(request, options)
|
||||
}
|
||||
): EmptyRequestResult = request(request, options)
|
||||
|
||||
/**
|
||||
* Calls a tool on the server by name, passing the specified arguments.
|
||||
|
|
@ -443,7 +422,7 @@ public open class Client(
|
|||
|
||||
val request = CallToolRequest(
|
||||
name = name,
|
||||
arguments = JsonObject(jsonArguments)
|
||||
arguments = JsonObject(jsonArguments),
|
||||
)
|
||||
return callTool(request, compatibility, options)
|
||||
}
|
||||
|
|
@ -461,12 +440,10 @@ public open class Client(
|
|||
request: CallToolRequest,
|
||||
compatibility: Boolean = false,
|
||||
options: RequestOptions? = null,
|
||||
): CallToolResultBase? {
|
||||
return if (compatibility) {
|
||||
request<CompatibilityCallToolResult>(request, options)
|
||||
} else {
|
||||
request<CallToolResult>(request, options)
|
||||
}
|
||||
): CallToolResultBase? = if (compatibility) {
|
||||
request<CompatibilityCallToolResult>(request, options)
|
||||
} else {
|
||||
request<CallToolResult>(request, options)
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -480,9 +457,7 @@ public open class Client(
|
|||
public suspend fun listTools(
|
||||
request: ListToolsRequest = ListToolsRequest(),
|
||||
options: RequestOptions? = null,
|
||||
): ListToolsResult {
|
||||
return request(request, options)
|
||||
}
|
||||
): ListToolsResult = request(request, options)
|
||||
|
||||
/**
|
||||
* Registers a single root.
|
||||
|
|
@ -491,10 +466,7 @@ public open class Client(
|
|||
* @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,
|
||||
) {
|
||||
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.")
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ import kotlin.time.Duration
|
|||
*
|
||||
* @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.
|
||||
* @param requestBuilder Optional lambda to configure the HTTP request.
|
||||
* @return A [SSEClientTransport] configured for MCP communication.
|
||||
*/
|
||||
public fun HttpClient.mcpSseTransport(
|
||||
|
|
@ -38,8 +38,8 @@ public suspend fun HttpClient.mcpSse(
|
|||
val client = Client(
|
||||
Implementation(
|
||||
name = IMPLEMENTATION_NAME,
|
||||
version = LIB_VERSION
|
||||
)
|
||||
version = LIB_VERSION,
|
||||
),
|
||||
)
|
||||
client.connect(transport)
|
||||
return client
|
||||
|
|
|
|||
|
|
@ -139,6 +139,7 @@ public class SseClientTransport(
|
|||
}
|
||||
|
||||
"endpoint" -> handleEndpoint(event.data.orEmpty())
|
||||
|
||||
else -> handleMessage(event.data.orEmpty())
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -35,10 +35,7 @@ import kotlin.coroutines.CoroutineContext
|
|||
* @param output The output stream where messages are sent.
|
||||
*/
|
||||
@OptIn(ExperimentalAtomicApi::class)
|
||||
public class StdioClientTransport(
|
||||
private val input: Source,
|
||||
private val output: Sink
|
||||
) : AbstractTransport() {
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -50,10 +50,8 @@ 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")
|
||||
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.
|
||||
|
|
@ -102,15 +100,16 @@ public class StreamableHttpClientTransport(
|
|||
public suspend fun send(
|
||||
message: JSONRPCMessage,
|
||||
resumptionToken: String?,
|
||||
onResumptionToken: ((String) -> Unit)? = null
|
||||
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
|
||||
resumptionToken = token,
|
||||
onResumptionToken = onResumptionToken,
|
||||
replayMessageId = if (message is JSONRPCRequest) message.id else null,
|
||||
)
|
||||
return
|
||||
}
|
||||
|
|
@ -147,8 +146,9 @@ public class StreamableHttpClientTransport(
|
|||
}
|
||||
|
||||
ContentType.Text.EventStream -> handleInlineSse(
|
||||
response, onResumptionToken = onResumptionToken,
|
||||
replayMessageId = if (message is JSONRPCRequest) message.id else null
|
||||
response,
|
||||
onResumptionToken = onResumptionToken,
|
||||
replayMessageId = if (message is JSONRPCRequest) message.id else null,
|
||||
)
|
||||
|
||||
else -> {
|
||||
|
|
@ -197,7 +197,7 @@ public class StreamableHttpClientTransport(
|
|||
if (!response.status.isSuccess() && response.status != HttpStatusCode.MethodNotAllowed) {
|
||||
val error = StreamableHttpError(
|
||||
response.status.value,
|
||||
"Failed to terminate session: ${response.status.description}"
|
||||
"Failed to terminate session: ${response.status.description}",
|
||||
)
|
||||
logger.error(error) { "Failed to terminate session" }
|
||||
_onError(error)
|
||||
|
|
@ -212,7 +212,7 @@ public class StreamableHttpClientTransport(
|
|||
private suspend fun startSseSession(
|
||||
resumptionToken: String? = null,
|
||||
replayMessageId: RequestId? = null,
|
||||
onResumptionToken: ((String) -> Unit)? = null
|
||||
onResumptionToken: ((String) -> Unit)? = null,
|
||||
) {
|
||||
sseSession?.cancel()
|
||||
sseJob?.cancelAndJoin()
|
||||
|
|
@ -254,7 +254,7 @@ public class StreamableHttpClientTransport(
|
|||
private suspend fun collectSse(
|
||||
session: ClientSSESession,
|
||||
replayMessageId: RequestId?,
|
||||
onResumptionToken: ((String) -> Unit)?
|
||||
onResumptionToken: ((String) -> Unit)?,
|
||||
) {
|
||||
try {
|
||||
session.incoming.collect { event ->
|
||||
|
|
@ -290,7 +290,7 @@ public class StreamableHttpClientTransport(
|
|||
private suspend fun handleInlineSse(
|
||||
response: HttpResponse,
|
||||
replayMessageId: RequestId?,
|
||||
onResumptionToken: ((String) -> Unit)?
|
||||
onResumptionToken: ((String) -> Unit)?,
|
||||
) {
|
||||
logger.trace { "Handling inline SSE from POST response" }
|
||||
val channel = response.bodyAsChannel()
|
||||
|
|
|
|||
|
|
@ -39,4 +39,4 @@ public suspend fun HttpClient.mcpStreamableHttp(
|
|||
val client = Client(Implementation(name = IMPLEMENTATION_NAME, version = LIB_VERSION))
|
||||
client.connect(transport)
|
||||
return client
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -33,8 +33,8 @@ public suspend fun HttpClient.mcpWebSocket(
|
|||
val client = Client(
|
||||
Implementation(
|
||||
name = IMPLEMENTATION_NAME,
|
||||
version = LIB_VERSION
|
||||
)
|
||||
version = LIB_VERSION,
|
||||
),
|
||||
)
|
||||
client.connect(transport)
|
||||
return client
|
||||
|
|
|
|||
|
|
@ -47,7 +47,7 @@ class StreamableHttpClientTransportTest {
|
|||
val message = JSONRPCRequest(
|
||||
id = RequestId.StringId("test-id"),
|
||||
method = "test",
|
||||
params = buildJsonObject { }
|
||||
params = buildJsonObject { },
|
||||
)
|
||||
|
||||
val transport = createTransport { request ->
|
||||
|
|
@ -61,7 +61,7 @@ class StreamableHttpClientTransportTest {
|
|||
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted
|
||||
status = HttpStatusCode.Accepted,
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -76,26 +76,30 @@ class StreamableHttpClientTransportTest {
|
|||
id = RequestId.StringId("test-id"),
|
||||
method = "initialize",
|
||||
params = buildJsonObject {
|
||||
put("clientInfo", buildJsonObject {
|
||||
put("name", JsonPrimitive("test-client"))
|
||||
put("version", JsonPrimitive("1.0"))
|
||||
})
|
||||
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")
|
||||
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
|
||||
status = HttpStatusCode.Accepted,
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -122,7 +126,7 @@ class StreamableHttpClientTransportTest {
|
|||
assertEquals("test-session-id", request.headers["mcp-session-id"])
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.OK
|
||||
status = HttpStatusCode.OK,
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -141,7 +145,7 @@ class StreamableHttpClientTransportTest {
|
|||
assertEquals(HttpMethod.Delete, request.method)
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.MethodNotAllowed
|
||||
status = HttpStatusCode.MethodNotAllowed,
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -160,7 +164,7 @@ class StreamableHttpClientTransportTest {
|
|||
assertEquals("2025-06-18", request.headers["mcp-protocol-version"])
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted
|
||||
status = HttpStatusCode.Accepted,
|
||||
)
|
||||
}
|
||||
transport.protocolVersion = "2025-06-18"
|
||||
|
|
@ -170,7 +174,8 @@ class StreamableHttpClientTransportTest {
|
|||
transport.close()
|
||||
}
|
||||
|
||||
@Ignore //Engine doesn't support SSECapability: https://youtrack.jetbrains.com/issue/KTOR-8177/MockEngine-Add-SSE-support
|
||||
// Engine doesn't support SSECapability: https://youtrack.jetbrains.com/issue/KTOR-8177/MockEngine-Add-SSE-support
|
||||
@Ignore
|
||||
@Test
|
||||
fun testNotificationSchemaE2E() = runTest {
|
||||
val receivedMessages = mutableListOf<JSONRPCMessage>()
|
||||
|
|
@ -182,7 +187,7 @@ class StreamableHttpClientTransportTest {
|
|||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted,
|
||||
headers = headersOf("mcp-session-id", "notification-test-session")
|
||||
headers = headersOf("mcp-session-id", "notification-test-session"),
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -193,7 +198,9 @@ class StreamableHttpClientTransportTest {
|
|||
// 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(
|
||||
"""data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":50,"total":100}}""",
|
||||
)
|
||||
appendLine()
|
||||
|
||||
appendLine("event: message")
|
||||
|
|
@ -210,8 +217,9 @@ class StreamableHttpClientTransportTest {
|
|||
content = ByteReadChannel(sseContent),
|
||||
status = HttpStatusCode.OK,
|
||||
headers = headersOf(
|
||||
HttpHeaders.ContentType, ContentType.Text.EventStream.toString()
|
||||
)
|
||||
HttpHeaders.ContentType,
|
||||
ContentType.Text.EventStream.toString(),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -219,7 +227,7 @@ class StreamableHttpClientTransportTest {
|
|||
HttpMethod.Post -> {
|
||||
respond(
|
||||
content = "",
|
||||
status = HttpStatusCode.Accepted
|
||||
status = HttpStatusCode.Accepted,
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -238,11 +246,14 @@ class StreamableHttpClientTransportTest {
|
|||
method = "notifications/initialized",
|
||||
params = buildJsonObject {
|
||||
put("protocolVersion", JsonPrimitive("1.0"))
|
||||
put("capabilities", buildJsonObject {
|
||||
put("tools", JsonPrimitive(true))
|
||||
put("resources", JsonPrimitive(true))
|
||||
})
|
||||
}
|
||||
put(
|
||||
"capabilities",
|
||||
buildJsonObject {
|
||||
put("tools", JsonPrimitive(true))
|
||||
put("resources", JsonPrimitive(true))
|
||||
},
|
||||
)
|
||||
},
|
||||
)
|
||||
|
||||
transport.send(initializedNotification)
|
||||
|
|
@ -274,25 +285,28 @@ class StreamableHttpClientTransportTest {
|
|||
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))
|
||||
})
|
||||
}
|
||||
)
|
||||
put(
|
||||
"data",
|
||||
buildJsonObject {
|
||||
put("duration", JsonPrimitive(1234))
|
||||
},
|
||||
)
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
// Send all client notifications
|
||||
|
|
@ -305,7 +319,8 @@ class StreamableHttpClientTransportTest {
|
|||
transport.close()
|
||||
}
|
||||
|
||||
@Ignore // Engine doesn't support SSECapability: https://youtrack.jetbrains.com/issue/KTOR-8177/MockEngine-Add-SSE-support
|
||||
// Engine doesn't support SSECapability: https://youtrack.jetbrains.com/issue/KTOR-8177/MockEngine-Add-SSE-support
|
||||
@Ignore
|
||||
@Test
|
||||
fun testNotificationWithResumptionToken() = runTest {
|
||||
var resumptionTokenReceived: String? = null
|
||||
|
|
@ -320,15 +335,18 @@ class StreamableHttpClientTransportTest {
|
|||
val sseContent = buildString {
|
||||
appendLine("event: message")
|
||||
appendLine("id: resume-100")
|
||||
appendLine("""data: {"jsonrpc":"2.0","method":"notifications/resumed","params":{"fromToken":"${lastEventIdSent}"}}""")
|
||||
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()
|
||||
)
|
||||
HttpHeaders.ContentType,
|
||||
ContentType.Text.EventStream.toString(),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -344,12 +362,12 @@ class StreamableHttpClientTransportTest {
|
|||
method = "notifications/test",
|
||||
params = buildJsonObject {
|
||||
put("data", JsonPrimitive("test-data"))
|
||||
}
|
||||
},
|
||||
),
|
||||
resumptionToken = "previous-token-99",
|
||||
onResumptionToken = { token ->
|
||||
resumptionTokenReceived = token
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
// Wait for response
|
||||
|
|
|
|||
Loading…
Reference in a new issue