Skip to content

fix: reject batch requests in streamable HTTP transport #957

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ez-lbz wants to merge 1 commit into
base: main
Choose a base branch
from
+20 −106
Open

fix: reject batch requests in streamable HTTP transport #957

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 8 additions & 1 deletion ...monMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -853,7 +853,14 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
return when (val element = McpJson.parseToJsonElement(body)) {
is JsonObject -> listOf(McpJson.decodeFromJsonElement(element))

is JsonArray -> McpJson.decodeFromJsonElement<List<JSONRPCMessage>>(element)
is JsonArray -> {
call.reject(
HttpStatusCode.BadRequest,
RPCError.ErrorCode.INVALID_REQUEST,
"Invalid Request: JSON-RPC batch requests are not supported",
)
null
}

else -> {
call.reject(
Expand Down
117 changes: 12 additions & 105 deletions ...est/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
package io.modelcontextprotocol.kotlin.sdk.server

import io.kotest.matchers.collections.shouldContainAll
import io.kotest.matchers.equals.shouldBeEqual
import io.kotest.matchers.nulls.shouldNotBeNull
import io.kotest.matchers.shouldBe
Expand Down Expand Up @@ -45,15 +44,11 @@ import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCMessage
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCRequest
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCResponse
import io.modelcontextprotocol.kotlin.sdk.types.LATEST_PROTOCOL_VERSION
import io.modelcontextprotocol.kotlin.sdk.types.ListResourcesResult
import io.modelcontextprotocol.kotlin.sdk.types.ListToolsResult
import io.modelcontextprotocol.kotlin.sdk.types.McpJson
import io.modelcontextprotocol.kotlin.sdk.types.Method
import io.modelcontextprotocol.kotlin.sdk.types.RPCError
import io.modelcontextprotocol.kotlin.sdk.types.RequestId
import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities
import io.modelcontextprotocol.kotlin.sdk.types.Tool
import io.modelcontextprotocol.kotlin.sdk.types.ToolSchema
import io.modelcontextprotocol.kotlin.sdk.types.toJSON
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
Expand All @@ -66,8 +61,6 @@ import kotlinx.coroutines.launch
import kotlinx.coroutines.test.runTest
import kotlinx.coroutines.withTimeout
import kotlinx.serialization.builtins.ListSerializer
import kotlinx.serialization.json.buildJsonObject
import kotlinx.serialization.json.put
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.Arguments
import org.junit.jupiter.params.provider.MethodSource
Expand Down Expand Up @@ -506,68 +499,35 @@ class StreamableHttpServerTransportTest {
}

@Test
fun `batched requests wait for all responses before replying`() = testApplication {
fun `batch request body is rejected with Bad Request`() = testApplication {
configTestServer()

val client = createTestClient(logging = true)
val client = createTestClient()

val transport = StreamableHttpServerTransport(enableJsonResponse = true)
val firstRequest = JSONRPCRequest(id = RequestId("first"), method = Method.Defined.ToolsList.value)
val secondRequest = JSONRPCRequest(id = RequestId("second"), method = Method.Defined.ResourcesList.value)

val firstResult = ListToolsResult(
tools = listOf(
Tool(name = "tool-1", inputSchema = ToolSchema()),
),
meta = buildJsonObject { put("label", "first") },
)
val secondResult = ListResourcesResult(
resources = emptyList(),
meta = buildJsonObject { put("label", "second") },
)

transport.onMessage { message ->
if (message is JSONRPCRequest) {
val result = when (message.id) {
firstRequest.id -> firstResult
secondRequest.id -> secondResult
else -> EmptyResult()
}
transport.send(JSONRPCResponse(message.id, result), null)
transport.send(JSONRPCResponse(message.id, EmptyResult()))
}
}

configureTransportEndpoint(transport)

val initRequest = buildInitializeRequestPayload()

val responseInit = client.post(path) {
addStreamableHeaders()
setBody(initRequest)
}

val payload = encodeMessages(listOf(firstRequest, secondRequest))
val payload = encodeMessages(
listOf(
JSONRPCRequest(id = RequestId("first"), method = Method.Defined.ToolsList.value),
JSONRPCRequest(id = RequestId("second"), method = Method.Defined.ResourcesList.value),
),
)

val response = client.post(path) {
addStreamableHeaders()
header("mcp-session-id", responseInit.headers[MCP_SESSION_ID_HEADER])
setBody(payload)
}

assertEquals(HttpStatusCode.OK, response.status)

val responses = response.body<List<JSONRPCResponse>>()
val results = responses.map { it.result }
results.shouldContainAll(firstResult, secondResult)

// Check responses' order

// TODO Uncomment when fixed https://github.com/modelcontextprotocol/kotlin-sdk/issues/548
/*assertEquals(listOf(firstRequest.id, secondRequest.id), responses.map { it.id })
val firstMeta = (responses[0] as ListToolsResult).meta
val secondMeta = (responses[1] as ListResourcesResult).meta
assertEquals("first", firstMeta?.get("label")?.jsonPrimitive?.content)
assertEquals("second", secondMeta?.get("label")?.jsonPrimitive?.content)*/
response.status shouldBe HttpStatusCode.BadRequest
val error = response.body<JSONRPCError>()
error.error.message shouldBe "Invalid Request: JSON-RPC batch requests are not supported"
}

@Test
Expand Down Expand Up @@ -619,59 +579,6 @@ class StreamableHttpServerTransportTest {
}
}

@Test
fun `json response for a batch waits until every asynchronously produced response is ready`() = testApplication {
configTestServer()

val client = createTestClient()

val transport = StreamableHttpServerTransport(
StreamableHttpServerTransport.Configuration(enableJsonResponse = true),
)
val firstRequest = JSONRPCRequest(id = RequestId("slow"), method = Method.Defined.ToolsList.value)
val secondRequest = JSONRPCRequest(id = RequestId("fast"), method = Method.Defined.ResourcesList.value)

// Simulates concurrent dispatch with out-of-order completion: the second request responds
// first; the POST must stay open until the slower first response settles too.
val handlerScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
try {
transport.onMessage { message ->
if (message is JSONRPCRequest) {
if (message.method == Method.Defined.Initialize.value) {
transport.send(JSONRPCResponse(message.id, EmptyResult()))
} else {
handlerScope.launch {
delay(if (message.id == firstRequest.id) 200 else 50)
transport.send(JSONRPCResponse(message.id, EmptyResult()))
}
}
}
}

configureTransportEndpoint(transport)

val initResponse = client.post(path) {
addStreamableHeaders()
setBody(buildInitializeRequestPayload())
}
initResponse.status shouldBe HttpStatusCode.OK

val response = withTimeout(10.seconds) {
client.post(path) {
addStreamableHeaders()
header(MCP_SESSION_ID_HEADER, initResponse.headers[MCP_SESSION_ID_HEADER])
setBody(encodeMessages(listOf(firstRequest, secondRequest)))
}
}

response.status shouldBe HttpStatusCode.OK
val responses = response.body<List<JSONRPCResponse>>()
responses.map { it.id }.toSet() shouldBe setOf(firstRequest.id, secondRequest.id)
} finally {
handlerScope.cancel()
}
}

@Test
fun `json response POST completes with 202 when its only request is cancelled`() = testApplication {
configTestServer()
Expand Down