fix(server): preserve Streamable HTTP response status and body (#911)
Streamable HTTP responses now keep their status when the client accepts only `text/event-stream`. ## How Has This Been Tested? New `StreamableHttpResponseDeliveryTest` ## Breaking Changes None ## 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) - [ ] 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
This commit is contained in:
parent
a06d699d6c
commit
82fe56e1e8
4 changed files with 191 additions and 21 deletions
|
|
@ -1,6 +1,7 @@
|
||||||
package io.modelcontextprotocol.kotlin.sdk.server
|
package io.modelcontextprotocol.kotlin.sdk.server
|
||||||
|
|
||||||
import io.github.oshai.kotlinlogging.KotlinLogging
|
import io.github.oshai.kotlinlogging.KotlinLogging
|
||||||
|
import io.ktor.http.HttpHeaders
|
||||||
import io.ktor.http.HttpMethod
|
import io.ktor.http.HttpMethod
|
||||||
import io.ktor.http.HttpStatusCode
|
import io.ktor.http.HttpStatusCode
|
||||||
import io.ktor.server.application.Application
|
import io.ktor.server.application.Application
|
||||||
|
|
@ -282,18 +283,10 @@ private fun Application.mcpStatelessStreamableHttp(
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
get {
|
get {
|
||||||
call.reject(
|
call.rejectUnsupportedMethod()
|
||||||
HttpStatusCode.MethodNotAllowed,
|
|
||||||
RPCError.ErrorCode.CONNECTION_CLOSED,
|
|
||||||
"Method not allowed.",
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
delete {
|
delete {
|
||||||
call.reject(
|
call.rejectUnsupportedMethod()
|
||||||
HttpStatusCode.MethodNotAllowed,
|
|
||||||
RPCError.ErrorCode.CONNECTION_CLOSED,
|
|
||||||
"Method not allowed.",
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -419,6 +412,16 @@ private suspend fun RoutingContext.mcpPostEndpoint(transportManager: TransportMa
|
||||||
logger.trace { "Message handled for sessionId: $sessionId" }
|
logger.trace { "Message handled for sessionId: $sessionId" }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** A stateless endpoint serves POST only, offering neither an SSE stream to open nor a session to delete. */
|
||||||
|
private suspend fun ApplicationCall.rejectUnsupportedMethod() {
|
||||||
|
response.header(HttpHeaders.Allow, HttpMethod.Post.value)
|
||||||
|
reject(
|
||||||
|
HttpStatusCode.MethodNotAllowed,
|
||||||
|
RPCError.ErrorCode.CONNECTION_CLOSED,
|
||||||
|
"Method not allowed.",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
private fun ApplicationRequest.sessionId(): String? = header(MCP_SESSION_ID_HEADER)
|
private fun ApplicationRequest.sessionId(): String? = header(MCP_SESSION_ID_HEADER)
|
||||||
|
|
||||||
private suspend fun existingStreamableTransport(
|
private suspend fun existingStreamableTransport(
|
||||||
|
|
|
||||||
|
|
@ -10,7 +10,7 @@ import io.ktor.server.request.header
|
||||||
import io.ktor.server.request.httpMethod
|
import io.ktor.server.request.httpMethod
|
||||||
import io.ktor.server.response.header
|
import io.ktor.server.response.header
|
||||||
import io.ktor.server.response.respond
|
import io.ktor.server.response.respond
|
||||||
import io.ktor.server.response.respondNullable
|
import io.ktor.server.response.respondText
|
||||||
import io.ktor.server.sse.ServerSSESession
|
import io.ktor.server.sse.ServerSSESession
|
||||||
import io.ktor.util.collections.ConcurrentMap
|
import io.ktor.util.collections.ConcurrentMap
|
||||||
import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
|
import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
|
||||||
|
|
@ -474,12 +474,13 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if (!validateProtocolVersion(call)) return
|
if (!validateProtocolVersion(call, initializationRequest.id)) return
|
||||||
if (messages.size > 1) {
|
if (messages.size > 1) {
|
||||||
call.reject(
|
call.reject(
|
||||||
HttpStatusCode.BadRequest,
|
HttpStatusCode.BadRequest,
|
||||||
RPCError.ErrorCode.INVALID_REQUEST,
|
RPCError.ErrorCode.INVALID_REQUEST,
|
||||||
"Invalid Request: Only one initialization request is allowed",
|
"Invalid Request: Only one initialization request is allowed",
|
||||||
|
initializationRequest.id,
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
@ -491,12 +492,13 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
sessionId?.let { onSessionInitialized?.invoke(it) }
|
sessionId?.let { onSessionInitialized?.invoke(it) }
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
if (!validateSession(call) || !validateProtocolVersion(call)) return
|
val requestId = messages.filterIsInstance<JSONRPCRequest>().firstOrNull()?.id
|
||||||
|
if (!validateSession(call, requestId) || !validateProtocolVersion(call, requestId)) return
|
||||||
}
|
}
|
||||||
|
|
||||||
val hasRequest = messages.any { it is JSONRPCRequest }
|
val hasRequest = messages.any { it is JSONRPCRequest }
|
||||||
if (!hasRequest) {
|
if (!hasRequest) {
|
||||||
call.respondNullable(status = HttpStatusCode.Accepted, message = null)
|
call.respond(HttpStatusCode.Accepted)
|
||||||
messages.forEach { message -> _onMessage(message) }
|
messages.forEach { message -> _onMessage(message) }
|
||||||
// A cancellation may target a request still awaiting delivery on a JSON-mode POST;
|
// A cancellation may target a request still awaiting delivery on a JSON-mode POST;
|
||||||
// retire it so that POST completes instead of waiting for a response that never comes.
|
// retire it so that POST completes instead of waiting for a response that never comes.
|
||||||
|
|
@ -552,7 +554,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
if (responses.isEmpty()) {
|
if (responses.isEmpty()) {
|
||||||
// Every request in this POST was cancelled before producing a response; there is
|
// Every request in this POST was cancelled before producing a response; there is
|
||||||
// nothing to return, so acknowledge with 202 rather than an empty JSON body.
|
// nothing to return, so acknowledge with 202 rather than an empty JSON body.
|
||||||
call.respondNullable(status = HttpStatusCode.Accepted, message = null)
|
call.respond(HttpStatusCode.Accepted)
|
||||||
} else {
|
} else {
|
||||||
call.response.header(HttpHeaders.ContentType, ContentType.Application.Json.toString())
|
call.response.header(HttpHeaders.ContentType, ContentType.Application.Json.toString())
|
||||||
val payload = if (responses.size == 1) responses.first() else responses
|
val payload = if (responses.size == 1) responses.first() else responses
|
||||||
|
|
@ -645,7 +647,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
if (!validateSession(call) || !validateProtocolVersion(call)) return
|
if (!validateSession(call) || !validateProtocolVersion(call)) return
|
||||||
sessionId?.let { onSessionClosed?.invoke(it) }
|
sessionId?.let { onSessionClosed?.invoke(it) }
|
||||||
close()
|
close()
|
||||||
call.respondNullable(status = HttpStatusCode.OK, message = null)
|
call.respond(HttpStatusCode.OK)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -734,7 +736,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private suspend fun validateSession(call: ApplicationCall): Boolean {
|
private suspend fun validateSession(call: ApplicationCall, id: RequestId? = null): Boolean {
|
||||||
if (sessionIdGenerator == null) return true
|
if (sessionIdGenerator == null) return true
|
||||||
|
|
||||||
if (!initialized.load()) {
|
if (!initialized.load()) {
|
||||||
|
|
@ -742,6 +744,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
HttpStatusCode.BadRequest,
|
HttpStatusCode.BadRequest,
|
||||||
RPCError.ErrorCode.CONNECTION_CLOSED,
|
RPCError.ErrorCode.CONNECTION_CLOSED,
|
||||||
"Bad Request: Server not initialized",
|
"Bad Request: Server not initialized",
|
||||||
|
id,
|
||||||
)
|
)
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
@ -753,6 +756,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
HttpStatusCode.BadRequest,
|
HttpStatusCode.BadRequest,
|
||||||
RPCError.ErrorCode.CONNECTION_CLOSED,
|
RPCError.ErrorCode.CONNECTION_CLOSED,
|
||||||
"Bad Request: Mcp-Session-Id header is required",
|
"Bad Request: Mcp-Session-Id header is required",
|
||||||
|
id,
|
||||||
)
|
)
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
@ -765,13 +769,14 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
HttpStatusCode.NotFound,
|
HttpStatusCode.NotFound,
|
||||||
REQUEST_TIMEOUT,
|
REQUEST_TIMEOUT,
|
||||||
"Session not found",
|
"Session not found",
|
||||||
|
id,
|
||||||
)
|
)
|
||||||
false
|
false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private suspend fun validateProtocolVersion(call: ApplicationCall): Boolean {
|
private suspend fun validateProtocolVersion(call: ApplicationCall, id: RequestId? = null): Boolean {
|
||||||
val version = call.request.headers[MCP_PROTOCOL_VERSION_HEADER] ?: return true
|
val version = call.request.headers[MCP_PROTOCOL_VERSION_HEADER] ?: return true
|
||||||
|
|
||||||
return when (version) {
|
return when (version) {
|
||||||
|
|
@ -784,6 +789,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
", ",
|
", ",
|
||||||
)
|
)
|
||||||
})",
|
})",
|
||||||
|
id,
|
||||||
)
|
)
|
||||||
false
|
false
|
||||||
}
|
}
|
||||||
|
|
@ -914,12 +920,13 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Writes an already serialized body so a client accepting only `text/event-stream` still receives [status]. */
|
||||||
internal suspend fun ApplicationCall.reject(
|
internal suspend fun ApplicationCall.reject(
|
||||||
status: HttpStatusCode,
|
status: HttpStatusCode,
|
||||||
code: Int,
|
code: Int,
|
||||||
message: String,
|
message: String,
|
||||||
id: RequestId? = null,
|
id: RequestId? = null,
|
||||||
) {
|
) {
|
||||||
this.response.status(status)
|
val error = JSONRPCError(id = id, error = RPCError(code = code, message = message))
|
||||||
this.respond(JSONRPCError(id = id, error = RPCError(code = code, message = message)))
|
respondText(McpJson.encodeToString(error), ContentType.Application.Json, status)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,99 @@
|
||||||
|
package io.modelcontextprotocol.kotlin.sdk.server
|
||||||
|
|
||||||
|
import io.kotest.matchers.shouldBe
|
||||||
|
import io.ktor.client.request.delete
|
||||||
|
import io.ktor.client.request.get
|
||||||
|
import io.ktor.client.request.header
|
||||||
|
import io.ktor.client.request.post
|
||||||
|
import io.ktor.client.request.setBody
|
||||||
|
import io.ktor.client.statement.HttpResponse
|
||||||
|
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.server.testing.testApplication
|
||||||
|
import io.modelcontextprotocol.kotlin.sdk.types.Implementation
|
||||||
|
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCError
|
||||||
|
import io.modelcontextprotocol.kotlin.sdk.types.McpJson
|
||||||
|
import io.modelcontextprotocol.kotlin.sdk.types.RPCError
|
||||||
|
import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities
|
||||||
|
import kotlin.test.Test
|
||||||
|
|
||||||
|
/**
|
||||||
|
* A client opening the standalone stream sends `Accept: text/event-stream`, which matches no JSON
|
||||||
|
* converter. Responses rendered through ContentNegotiation used to reach such a client as an empty
|
||||||
|
* 406 with the intended status discarded.
|
||||||
|
*/
|
||||||
|
class StreamableHttpResponseDeliveryTest {
|
||||||
|
|
||||||
|
private val eventStream = ContentType.Text.EventStream.toString()
|
||||||
|
|
||||||
|
private fun testServer(): Server = Server(
|
||||||
|
Implementation("test-server", "1.0"),
|
||||||
|
ServerOptions(capabilities = ServerCapabilities()),
|
||||||
|
)
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `stateless GET is rejected with 405 and advertises POST`() = testApplication {
|
||||||
|
application { mcpStatelessStreamableHttp { testServer() } }
|
||||||
|
|
||||||
|
val response = client.get("/mcp") {
|
||||||
|
header(HttpHeaders.Host, "localhost")
|
||||||
|
header(HttpHeaders.Accept, eventStream)
|
||||||
|
}
|
||||||
|
|
||||||
|
response.assertMethodNotAllowed()
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `stateless DELETE is rejected with 405 and advertises POST`() = testApplication {
|
||||||
|
application { mcpStatelessStreamableHttp { testServer() } }
|
||||||
|
|
||||||
|
val response = client.delete("/mcp") {
|
||||||
|
header(HttpHeaders.Host, "localhost")
|
||||||
|
header(HttpHeaders.Accept, eventStream)
|
||||||
|
}
|
||||||
|
|
||||||
|
response.assertMethodNotAllowed()
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `stateful DELETE for an unknown session returns 404`() = testApplication {
|
||||||
|
application { mcpStreamableHttp { testServer() } }
|
||||||
|
|
||||||
|
val response = client.delete("/mcp") {
|
||||||
|
header(HttpHeaders.Host, "localhost")
|
||||||
|
header(HttpHeaders.Accept, eventStream)
|
||||||
|
header(MCP_SESSION_ID_HEADER, "unknown-session")
|
||||||
|
}
|
||||||
|
|
||||||
|
response.status shouldBe HttpStatusCode.NotFound
|
||||||
|
response.decodeError().error.message shouldBe "Session not found"
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `POST carrying no request is acknowledged with a bodiless 202`() = testApplication {
|
||||||
|
application { mcpStatelessStreamableHttp { testServer() } }
|
||||||
|
|
||||||
|
val response = client.post("/mcp") {
|
||||||
|
header(HttpHeaders.Host, "localhost")
|
||||||
|
header(HttpHeaders.Accept, "${ContentType.Application.Json}, $eventStream")
|
||||||
|
contentType(ContentType.Application.Json)
|
||||||
|
setBody("""{"jsonrpc":"2.0","method":"notifications/initialized"}""")
|
||||||
|
}
|
||||||
|
|
||||||
|
response.status shouldBe HttpStatusCode.Accepted
|
||||||
|
response.bodyAsText() shouldBe ""
|
||||||
|
}
|
||||||
|
|
||||||
|
private suspend fun HttpResponse.assertMethodNotAllowed() {
|
||||||
|
status shouldBe HttpStatusCode.MethodNotAllowed
|
||||||
|
headers[HttpHeaders.Allow] shouldBe HttpMethod.Post.value
|
||||||
|
contentType()?.withoutParameters() shouldBe ContentType.Application.Json
|
||||||
|
decodeError().error.code shouldBe RPCError.ErrorCode.CONNECTION_CLOSED
|
||||||
|
}
|
||||||
|
|
||||||
|
private suspend fun HttpResponse.decodeError(): JSONRPCError = McpJson.decodeFromString(bodyAsText())
|
||||||
|
}
|
||||||
|
|
@ -216,14 +216,16 @@ class StreamableHttpServerTransportTest {
|
||||||
|
|
||||||
configureTransportEndpoint(transport)
|
configureTransportEndpoint(transport)
|
||||||
|
|
||||||
|
val payload = buildInitializeRequestPayload()
|
||||||
val initResponse = client.post(path) {
|
val initResponse = client.post(path) {
|
||||||
addStreamableHeaders()
|
addStreamableHeaders()
|
||||||
header("mcp-protocol-version", "1900-01-01")
|
header("mcp-protocol-version", "1900-01-01")
|
||||||
setBody(buildInitializeRequestPayload())
|
setBody(payload)
|
||||||
}
|
}
|
||||||
|
|
||||||
initResponse.status shouldBe HttpStatusCode.BadRequest
|
initResponse.status shouldBe HttpStatusCode.BadRequest
|
||||||
initResponse.headers[MCP_SESSION_ID_HEADER] shouldBe null
|
initResponse.headers[MCP_SESSION_ID_HEADER] shouldBe null
|
||||||
|
initResponse.body<JSONRPCError>().id shouldBe payload.id
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
|
|
@ -297,6 +299,65 @@ class StreamableHttpServerTransportTest {
|
||||||
}
|
}
|
||||||
|
|
||||||
response.status shouldBe HttpStatusCode.BadRequest
|
response.status shouldBe HttpStatusCode.BadRequest
|
||||||
|
response.body<JSONRPCError>().id shouldBe RequestId("test-1")
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `batch with an initialization request echoes the initialize id when rejected`() = testApplication {
|
||||||
|
configTestServer()
|
||||||
|
|
||||||
|
val client = createTestClient()
|
||||||
|
|
||||||
|
val transport = StreamableHttpServerTransport(enableJsonResponse = true)
|
||||||
|
transport.onMessage { }
|
||||||
|
|
||||||
|
configureTransportEndpoint(transport)
|
||||||
|
|
||||||
|
val initPayload = buildInitializeRequestPayload()
|
||||||
|
val response = client.post(path) {
|
||||||
|
addStreamableHeaders()
|
||||||
|
setBody(
|
||||||
|
encodeMessages(
|
||||||
|
listOf(
|
||||||
|
initPayload,
|
||||||
|
JSONRPCRequest(id = RequestId("extra"), method = Method.Defined.ToolsList.value),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
response.status shouldBe HttpStatusCode.BadRequest
|
||||||
|
val error = response.body<JSONRPCError>()
|
||||||
|
error.error.message shouldBe "Invalid Request: Only one initialization request is allowed"
|
||||||
|
error.id shouldBe initPayload.id
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `non-init request before initialization echoes the request id`() = testApplication {
|
||||||
|
configTestServer()
|
||||||
|
|
||||||
|
val client = createTestClient()
|
||||||
|
|
||||||
|
val transport = StreamableHttpServerTransport(enableJsonResponse = true)
|
||||||
|
transport.onMessage { }
|
||||||
|
|
||||||
|
configureTransportEndpoint(transport)
|
||||||
|
|
||||||
|
val response = client.post(path) {
|
||||||
|
addStreamableHeaders()
|
||||||
|
setBody(
|
||||||
|
encodeMessages(
|
||||||
|
listOf(
|
||||||
|
JSONRPCRequest(id = RequestId("before-init"), method = Method.Defined.ToolsList.value),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
response.status shouldBe HttpStatusCode.BadRequest
|
||||||
|
val error = response.body<JSONRPCError>()
|
||||||
|
error.error.message shouldBe "Bad Request: Server not initialized"
|
||||||
|
error.id shouldBe RequestId("before-init")
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue