## Summary Fix SSE endpoint resolution so absolute paths resolve against the origin, and add tests covering absolute vs relative endpoint events. ## Motivation and Context The SSE transport currently resolves all endpoint events against the SSE URL’s base path. When the server sends an absolute endpoint (starting with `/`), the client incorrectly prefixes it with the SSE path. This change resolves absolute endpoints against the origin while keeping relative behavior unchanged. addresses: https://github.com/modelcontextprotocol/kotlin-sdk/issues/329 ## How Has This Been Tested? Unit tests added in `SseClientTransportTest` for absolute and relative endpoint events. ## Breaking Changes No. ## Types of changes - [x] Bug fix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [x] Documentation update ## Checklist - [x] I have read the [MCP Documentation](https://modelcontextprotocol.io) - [x] My code follows the repository's style guidelines - [x] New and existing tests pass locally - [x] I have added appropriate error handling - [x] I have added or updated documentation as needed ## Additional context The endpoint resolution logic now treats paths starting with `/` as origin-relative. Relative paths continue to resolve against the SSE base URL.
This commit is contained in:
Jakub Amanowicz
GitHub
parent
27b419c446
commit
43b59c9fc1
3 changed files with 225 additions and 7 deletions
|
|
@ -11,7 +11,6 @@ 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
|
||||
|
|
@ -54,13 +53,17 @@ public class SseClientTransport(
|
|||
private lateinit var scope: CoroutineScope
|
||||
private var job: Job? = null
|
||||
|
||||
private val origin: String by lazy {
|
||||
session.call.request.url.protocolWithAuthority
|
||||
}
|
||||
|
||||
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("/"))
|
||||
path.isEmpty() -> origin
|
||||
path.endsWith("/") -> origin + path.removeSuffix("/")
|
||||
else -> origin + path.take(path.lastIndexOf("/"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -135,11 +138,21 @@ public class SseClientTransport(
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves and completes [endpoint] based on [eventData].
|
||||
* Uses full URLs as-is, treats absolute paths as origin-relative,
|
||||
* and relative paths as relative to [baseUrl].
|
||||
*/
|
||||
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())
|
||||
val endpointUrl = if (eventData.startsWith("http://") || eventData.startsWith("https://")) {
|
||||
eventData
|
||||
} else if (eventData.startsWith("/")) {
|
||||
origin + eventData
|
||||
} else {
|
||||
"$baseUrl/$eventData"
|
||||
}
|
||||
endpoint.complete(endpointUrl)
|
||||
logger.debug { "Client connected to endpoint: $endpointUrl" }
|
||||
} catch (e: Throwable) {
|
||||
_onError(e)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,99 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client.sse
|
||||
|
||||
import io.ktor.client.engine.HttpClientEngine
|
||||
import io.ktor.client.engine.HttpClientEngineConfig
|
||||
import io.ktor.client.plugins.sse.SSECapability
|
||||
import io.ktor.client.request.HttpRequestData
|
||||
import io.ktor.client.request.HttpResponseData
|
||||
import io.ktor.client.request.ResponseAdapterAttributeKey
|
||||
import io.ktor.http.Headers
|
||||
import io.ktor.http.HttpMethod
|
||||
import io.ktor.http.HttpProtocolVersion
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.headersOf
|
||||
import io.ktor.util.date.GMTDate
|
||||
import io.ktor.utils.io.ByteChannel
|
||||
import io.ktor.utils.io.ByteReadChannel
|
||||
import io.ktor.utils.io.InternalAPI
|
||||
import io.ktor.utils.io.writeStringUtf8
|
||||
import kotlinx.coroutines.CoroutineDispatcher
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlin.coroutines.CoroutineContext
|
||||
|
||||
@OptIn(InternalAPI::class)
|
||||
internal open class MockSseClientEngine(
|
||||
private val endpoint: String,
|
||||
private val onPostRequest: (postRequest: HttpRequestData) -> Unit,
|
||||
override val dispatcher: CoroutineDispatcher = Dispatchers.Default,
|
||||
) : HttpClientEngine {
|
||||
|
||||
override val config = HttpClientEngineConfig()
|
||||
override val supportedCapabilities = setOf(SSECapability)
|
||||
|
||||
private val job = SupervisorJob()
|
||||
override val coroutineContext: CoroutineContext = dispatcher + job
|
||||
|
||||
private val channel by lazy { ByteChannel(true) }
|
||||
|
||||
override suspend fun execute(data: HttpRequestData): HttpResponseData = when (data.method) {
|
||||
HttpMethod.Get -> respondWithSseEndpointEvent(data)
|
||||
|
||||
HttpMethod.Post -> {
|
||||
onPostRequest(data)
|
||||
respondWithAccepted()
|
||||
}
|
||||
|
||||
else -> error("Unsupported method: ${data.method}")
|
||||
}
|
||||
|
||||
override fun close() {
|
||||
channel.close()
|
||||
job.cancel()
|
||||
}
|
||||
|
||||
private fun responseContext(): CoroutineContext = dispatcher + Job()
|
||||
|
||||
private suspend fun respondWithSseEndpointEvent(data: HttpRequestData): HttpResponseData {
|
||||
val sseResponseAdapter = requireNotNull(data.attributes.getOrNull(ResponseAdapterAttributeKey)) {
|
||||
"Missing SSE response adapter."
|
||||
}
|
||||
channel.writeStringUtf8(endpointEvent())
|
||||
channel.flush()
|
||||
val responseContext = responseContext()
|
||||
val responseBody = sseResponseAdapter.adapt(
|
||||
data,
|
||||
HttpStatusCode.OK,
|
||||
eventStreamHeader,
|
||||
channel,
|
||||
data.body,
|
||||
responseContext,
|
||||
) ?: error("Failed to adapt SSE response.")
|
||||
return HttpResponseData(
|
||||
statusCode = HttpStatusCode.OK,
|
||||
requestTime = GMTDate(),
|
||||
headers = eventStreamHeader,
|
||||
version = HttpProtocolVersion.HTTP_1_1,
|
||||
body = responseBody,
|
||||
callContext = responseContext,
|
||||
)
|
||||
}
|
||||
|
||||
private val eventStreamHeader by lazy { headersOf("Content-Type", "text/event-stream") }
|
||||
|
||||
private fun endpointEvent(): String = buildString {
|
||||
appendLine("event: endpoint")
|
||||
appendLine("data: $endpoint")
|
||||
appendLine()
|
||||
}
|
||||
|
||||
private fun respondWithAccepted(): HttpResponseData = HttpResponseData(
|
||||
statusCode = HttpStatusCode.Accepted,
|
||||
requestTime = GMTDate(),
|
||||
headers = Headers.Empty,
|
||||
version = HttpProtocolVersion.HTTP_1_1,
|
||||
body = ByteReadChannel.Empty,
|
||||
callContext = responseContext(),
|
||||
)
|
||||
}
|
||||
|
|
@ -0,0 +1,106 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client.sse
|
||||
|
||||
import io.kotest.matchers.collections.shouldHaveSize
|
||||
import io.kotest.matchers.shouldBe
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.plugins.sse.SSE
|
||||
import io.ktor.client.request.HttpRequestData
|
||||
import io.modelcontextprotocol.kotlin.sdk.client.SseClientTransport
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCNotification
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import kotlin.test.Test
|
||||
|
||||
class SseClientTransportTest {
|
||||
|
||||
@Test
|
||||
fun `absolute path endpoint resolves against origin`() = runTest {
|
||||
// Given
|
||||
val sseUrl = "http://example.com/api/mcp/sse"
|
||||
|
||||
// And
|
||||
val endpointEvent = "/messages?sessionId=abc"
|
||||
|
||||
// And
|
||||
val engine = CapturingSseClientEngine(endpoint = endpointEvent)
|
||||
val transport = sseTransport(sseUrl, engine)
|
||||
|
||||
// When
|
||||
transport.start()
|
||||
transport.send(JSONRPCNotification(method = "test"))
|
||||
|
||||
// Then
|
||||
val capturedPosts = engine.capturedPosts
|
||||
capturedPosts shouldHaveSize 1
|
||||
capturedPosts[0].url.toString() shouldBe "http://example.com/messages?sessionId=abc"
|
||||
|
||||
// Cleanup
|
||||
transport.close()
|
||||
engine.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `relative path endpoint resolves against baseUrl`() = runTest {
|
||||
// Given
|
||||
val sseUrl = "http://example.com/api/mcp/sse"
|
||||
|
||||
// And
|
||||
val endpointEvent = "post?sessionId=xyz"
|
||||
|
||||
// And
|
||||
val engine = CapturingSseClientEngine(endpoint = endpointEvent)
|
||||
val transport = sseTransport(sseUrl, engine)
|
||||
|
||||
// When
|
||||
transport.start()
|
||||
transport.send(JSONRPCNotification(method = "test"))
|
||||
|
||||
// Then
|
||||
val capturedPosts = engine.capturedPosts
|
||||
capturedPosts shouldHaveSize 1
|
||||
capturedPosts[0].url.toString() shouldBe "http://example.com/api/mcp/post?sessionId=xyz"
|
||||
|
||||
// Cleanup
|
||||
transport.close()
|
||||
engine.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `full url endpoint is used as-is`() = runTest {
|
||||
// Given
|
||||
val sseUrl = "http://example.com/api/mcp/sse"
|
||||
|
||||
// And
|
||||
val endpointEvent = "https://example.com/messages?sessionId=abc"
|
||||
|
||||
// And
|
||||
val engine = CapturingSseClientEngine(endpoint = endpointEvent)
|
||||
val transport = sseTransport(sseUrl, engine)
|
||||
|
||||
// When
|
||||
transport.start()
|
||||
transport.send(JSONRPCNotification(method = "test"))
|
||||
|
||||
// Then
|
||||
val capturedPosts = engine.capturedPosts
|
||||
capturedPosts shouldHaveSize 1
|
||||
capturedPosts[0].url.toString() shouldBe "https://example.com/messages?sessionId=abc"
|
||||
|
||||
// Cleanup
|
||||
transport.close()
|
||||
engine.close()
|
||||
}
|
||||
|
||||
private fun sseTransport(sseUrl: String, engine: CapturingSseClientEngine) =
|
||||
SseClientTransport(HttpClient(engine) { install(SSE) }, sseUrl)
|
||||
|
||||
private class CapturingSseClientEngine private constructor(
|
||||
endpoint: String,
|
||||
private val capturedPostRequests: MutableList<HttpRequestData> = mutableListOf(),
|
||||
) : MockSseClientEngine(endpoint, capturedPostRequests::add) {
|
||||
|
||||
constructor(endpoint: String) : this(endpoint, mutableListOf())
|
||||
|
||||
val capturedPosts: List<HttpRequestData>
|
||||
get() = capturedPostRequests
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue