[WIP] StreamableHttpServerTransport implementation #449

devcrocod · 2025-12-04T19:21:28Z

[WIP] StreamableHttpServerTransport implementation

Implement StreamableHttpServerTransport for MCP spec
Support both SSE streaming and direct HTTP JSON responses
Add event store interface for resumability

Important

The work is in progress. StreamableHttpServerTransport is not production-ready yet!!!

Based on #235

How Has This Been Tested?

Breaking Changes

None

Types of changes

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

I have read the MCP Documentation
My code follows the repository's style guidelines
New and existing tests pass locally
I have added appropriate error handling
I have added or updated documentation as needed

@zshea

Co-authored-by: @zshea

@zshea

Co-authored-by: @zshea

github-actions · 2025-12-04T19:25:56Z

	Tests	Passed ✅	Skipped ⚠️	Failed	Time ⏱
JUnit Test Report	3417 ran	3328 passed	89 skipped		4m 6s 708ms

Test	Result	Time ⏱
No test annotations available

kotlin-sdk-core/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/types/serializers.kt

kpavlov · 2025-12-05T05:09:33Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+    private suspend fun validateProtocolVersion(call: ApplicationCall): Boolean {
+        val protocolVersions = call.request.headers.getAll(MCP_PROTOCOL_VERSION_HEADER)
+        val version = protocolVersions?.lastOrNull() ?: DEFAULT_NEGOTIATED_PROTOCOL_VERSION
+
+        return when (version) {
+            !in SUPPORTED_PROTOCOL_VERSIONS -> {
+                call.reject(
+                    HttpStatusCode.BadRequest,
+                    RPCError.ErrorCode.CONNECTION_CLOSED,
+                    "Bad Request: Unsupported protocol version (supported versions: ${
+                        SUPPORTED_PROTOCOL_VERSIONS.joinToString(
+                            ", ",
+                        )
+                    })",
+                )
+                false
+            }
+
+            else -> true
+        }
+    }


How about we handle version extraction and validation separately? Also, could we standardize the validation process for all transportation types? Having it in a central location would be great.

not sure
for stdio, for example, a version check is not required

kpavlov · 2025-12-05T05:13:22Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+internal 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"
+private const val MAXIMUM_MESSAGE_SIZE = 4 * 1024 * 1024 // 4 MB


These constants could really help with all the different protocols. How about we make them a standard and reuse them?

For sse, the session-id header is not sent and there is no resumption token. For now, this is needed exclusively for streamable

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

kpavlov · 2025-12-05T05:23:26Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+    public var sessionId: String? = null
+        private set
+
+    private var sessionIdGenerator: (() -> String)? = { Uuid.random().toString() }


Let's create AbstractServerTransport and move it there

kotlin-sdk-core/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/types/jsonRpc.kt

kpavlov · 2025-12-05T05:34:43Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+     *
+     * Set undefined to disable session management.
+     */
+    public fun setSessionIdGenerator(block: (() -> String)?) {


It would be great if this were standard across all server transport methods. However, I’m not sure it should be publicly available.

kpavlov · 2025-12-05T05:36:10Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+    /**
+     * A callback for session initialization events
+     * This is called when the server initializes a new session.
+     * Useful in cases when you need to register multiple mcp sessions
+     * and need to keep track of them.
+     */
+    public fun setOnSessionInitialized(block: ((String) -> Unit)?) {
+        onSessionInitialized = block
+    }
+
+    /**
+     * A callback for session close events
+     * This is called when the server closes a session due to a DELETE request.
+     * Useful in cases when you need to clean up resources associated with the session.
+     * Note that this is different from the transport closing, if you are handling
+     * HTTP requests from multiple nodes you might want to close each
+     * StreamableHTTPServerTransport after a request is completed while still keeping the
+     * session open/running.
+     */
+    public fun setOnSessionClosed(block: ((String) -> Unit)?) {
+        onSessionClosed = block
+    }


It looks like this kind of functionality is pretty standard across all transport types, or at least for server transports.

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

…nd protocols

Copilot

Pull request overview

This PR implements the StreamableHttpServerTransport for the MCP (Model Context Protocol) specification, adding support for both Server-Sent Events (SSE) streaming and direct HTTP JSON responses with optional resumability support.

Key Changes:

New StreamableHttpServerTransport class supporting SSE and HTTP JSON response modes
EventStore interface for resumability via event storage and replay
JSONRPCEmptyMessage type for priming SSE events and empty message handling

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 10 comments.

Show a summary per file

File	Description
`kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt`	Core implementation of the streamable HTTP transport with session management, SSE streaming, and HTTP JSON response support
`kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/EventStore.kt`	Interface definition for event storage and replay to enable resumability
`kotlin-sdk-core/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/types/jsonRpc.kt`	Added `JSONRPCEmptyMessage` type and made `JSONRPCError.id` nullable
`kotlin-sdk-core/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/types/serializers.kt`	Updated JSON-RPC message serializer to handle empty messages
`kotlin-sdk-core/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/types/common.kt`	Added `DEFAULT_NEGOTIATED_PROTOCOL_VERSION` constant and removed unused imports
`kotlin-sdk-core/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/shared/Protocol.kt`	Added handling for `JSONRPCEmptyMessage` in protocol message processing
`kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt`	Comprehensive test suite covering initialization, batched requests, and notifications
`kotlin-sdk-core/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/types/JsonRpcTest.kt`	Test for `JSONRPCEmptyMessage` deserialization
`kotlin-sdk-server/build.gradle.kts`	Added test dependencies for Ktor and Kotest
`gradle/libs.versions.toml`	Added library definitions for Ktor content negotiation and serialization
`kotlin-sdk-server/api/kotlin-sdk-server.api`	Public API surface additions for `StreamableHttpServerTransport` and `EventStore`
`kotlin-sdk-core/api/kotlin-sdk-core.api`	Public API surface additions for new types and constants

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-12-05T16:04:22Z

kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/EventStore.kt

+     * Returns the stream ID associated with [eventId], or null if the event is unknown.
+     * Default implementation is a no-op which disables extra validation during replay.


Missing KDoc documentation for the getStreamIdForEventId method. The inline comment on line 30 describes it as a "no-op" default implementation, but there's no actual default implementation provided (it's an interface method). Consider either providing a default implementation or clarifying the documentation to explain that implementers may throw NotImplementedError or UnsupportedOperationException to disable this validation.

Suggested change

* Returns the stream ID associated with [eventId], or null if the event is unknown.

* Default implementation is a no-op which disables extra validation during replay.

* Returns the stream ID associated with the given [eventId], or null if the event is unknown.

*

* Implementers may throw [NotImplementedError] or [UnsupportedOperationException] to disable

* extra validation during replay, as there is no default implementation provided.

*

* @param eventId The event ID to look up.

* @return The stream ID associated with the event, or null if unknown.

kotlin-sdk-core/api/kotlin-sdk-core.api

Copilot · 2025-12-05T16:04:23Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+    private val streamMutex = Mutex()
+
+    private companion object {
+        const val STANDALONE_SSE_STREAM_ID = "_GET_stream"


The constant STANDALONE_SSE_STREAM_ID is marked as part of the companion object but is being exported in the public API (line 161 of the API file). If this is intended as an internal implementation detail, it should not be exposed in the public API. Consider making this internal or private if it's not meant for external consumption.

Suggested change

const val STANDALONE_SSE_STREAM_ID = "_GET_stream"

private const val STANDALONE_SSE_STREAM_ID = "_GET_stream"

Copilot · 2025-12-05T16:04:23Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+            call.reject(
+                HttpStatusCode.BadRequest,
+                RPCError.ErrorCode.CONNECTION_CLOSED,
+                "Bad Request: Mcp-Session-Id header is required",


The error message references "Mcp-Session-Id" with inconsistent capitalization. The constant MCP_SESSION_ID_HEADER is defined as "mcp-session-id" (lowercase), but the error message uses "Mcp-Session-Id". For consistency and clarity, use the actual header name as defined in the constant.

Copilot · 2025-12-05T16:04:23Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+            call.reject(
+                HttpStatusCode.BadRequest,
+                RPCError.ErrorCode.CONNECTION_CLOSED,
+                "Bad Request: Mcp-Session-Id header must be a single value",


The error message references "Mcp-Session-Id" with inconsistent capitalization. The constant MCP_SESSION_ID_HEADER is defined as "mcp-session-id" (lowercase), but the error message uses "Mcp-Session-Id". For consistency and clarity, use the actual header name as defined in the constant.

Copilot · 2025-12-05T16:04:24Z

kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/EventStore.kt

+     * Stores an event for later retrieval
+     * @param streamId ID of the stream the event belongs to
+     * @param message The JSON-RPC message to store
+     * @returns The generated event ID for the stored event


The comment style uses @returns instead of the Kotlin standard @return. KDoc convention uses @return (singular) for documenting return values.

Suggested change

* @returns The generated event ID for the stored event

* @return The generated event ID for the stored event

Copilot · 2025-12-05T16:04:25Z

...vmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt

+        // TODO(check order)
+//        assertEquals(listOf(firstRequest.id, secondRequest.id), responses.map { it.id })
+//        val firstMeta = (responses[0].result as EmptyResult).meta
+//        val secondMeta = (responses[1].result as EmptyResult).meta
+//        assertEquals("first", firstMeta?.get("label")?.jsonPrimitive?.content)
+//        assertEquals("second", secondMeta?.get("label")?.jsonPrimitive?.content)


Commented-out test code should be removed before merging. If order verification is intended for future implementation, consider creating a separate TODO issue or ticket instead of leaving commented code in the codebase.

Suggested change

// TODO(check order)

// assertEquals(listOf(firstRequest.id, secondRequest.id), responses.map { it.id })

// val firstMeta = (responses[0].result as EmptyResult).meta

// val secondMeta = (responses[1].result as EmptyResult).meta

// assertEquals("first", firstMeta?.get("label")?.jsonPrimitive?.content)

// assertEquals("second", secondMeta?.get("label")?.jsonPrimitive?.content)

Copilot · 2025-12-05T16:04:25Z

kotlin-sdk-core/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/types/jsonRpc.kt

 */
 @Serializable
-public data class JSONRPCError(val id: RequestId, val error: RPCError) : JSONRPCMessage {
+public data class JSONRPCError(val id: RequestId?, val error: RPCError) : JSONRPCMessage {


The id field of JSONRPCError is now nullable, but this could be a breaking API change. According to the JSON-RPC 2.0 specification, the id member is required in error responses except when there was an error in detecting the id in the request object (e.g., parse error). Consider documenting when null is appropriate, or ensure this aligns with how the codebase handles error scenarios where the request ID cannot be determined.

Copilot · 2025-12-05T16:04:26Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+            streamsMapping.values.forEach {
+                try {
+                    it.session?.close()
+                } catch (_: Exception) {


Empty catch block silently swallows exceptions. While this may be intentional during cleanup, consider at least logging the exception or adding a comment explaining why it's safe to ignore.

Suggested change

} catch (_: Exception) {

} catch (e: Exception) {

println("Exception while closing session: ${e.message}")

logger should be used

Copilot · 2025-12-05T16:04:27Z

...commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

+        val eventId = eventStore?.storeEvent(streamId, message)
+        try {
+            session?.send(event = "message", id = eventId, data = McpJson.encodeToString(message))
+        } catch (_: Exception) {


Empty catch block silently swallows exceptions. While this may be intentional when a stream is already closed, consider at least logging the exception or adding a comment explaining why it's safe to ignore, especially since this is in the message emission path.

Suggested change

} catch (_: Exception) {

} catch (e: Exception) {

println("emitOnStream: Exception sending message on stream $streamId: ${e.message}")

kpavlov

@devcrocod 👍🏻
Let's merge this PR and continue in the follow-ups

devcrocod added 2 commits December 4, 2025 20:14

Add PrimingEventMessage type and update Protocol to handle it

d189d4f

Co-authored-by: @zshea

Add StreamableHttpServerTransport implementation

4667bf0

Co-authored-by: @zshea

update api

a40db1d

kpavlov reviewed Dec 5, 2025

View reviewed changes

devcrocod added 3 commits December 5, 2025 16:00

Add tests for StreamableHttpServerTransport with necessary dependencies

e593c2e

Replace PrimingEventMessage with JSONRPCEmptyMessage across SDK a…

02c3c15

…nd protocols

Add JSONRPCEmptyMessage type to the API

04956fb

devcrocod marked this pull request as ready for review December 5, 2025 15:55

Copilot AI review requested due to automatic review settings December 5, 2025 15:55

Copilot started reviewing on behalf of devcrocod December 5, 2025 15:55 View session

Copilot finished reviewing on behalf of devcrocod December 5, 2025 15:57

Copilot AI reviewed Dec 5, 2025

View reviewed changes

Remove PrimingEventMessage type from API

600850e

kpavlov changed the title ~~StreamableHttpServerTransport implementation~~ [WIP] StreamableHttpServerTransport implementation Dec 5, 2025

kpavlov approved these changes Dec 5, 2025

View reviewed changes

kpavlov merged commit 10a0b67 into main Dec 5, 2025
5 of 6 checks passed

kpavlov deleted the devcrocod/streamable-server-transport branch December 5, 2025 17:51

kpavlov mentioned this pull request Dec 6, 2025

Add PrimingEventMessage #448

Closed

9 tasks

		* Returns the stream ID associated with [eventId], or null if the event is unknown.
		* Default implementation is a no-op which disables extra validation during replay.

-     * Returns the stream ID associated with [eventId], or null if the event is unknown.
-     * Default implementation is a no-op which disables extra validation during replay.
+     * Returns the stream ID associated with the given [eventId], or null if the event is unknown.
+     *
+     * Implementers may throw [NotImplementedError] or [UnsupportedOperationException] to disable
+     * extra validation during replay, as there is no default implementation provided.
+     *
+     * @param eventId The event ID to look up.
+     * @return The stream ID associated with the event, or null if unknown.

	const val STANDALONE_SSE_STREAM_ID = "_GET_stream"
	private const val STANDALONE_SSE_STREAM_ID = "_GET_stream"

	* @returns The generated event ID for the stored event
	* @return The generated event ID for the stored event

	} catch (_: Exception) {
	} catch (e: Exception) {
	println("Exception while closing session: ${e.message}")

	} catch (_: Exception) {
	} catch (e: Exception) {
	println("emitOnStream: Exception sending message on stream $streamId: ${e.message}")

[WIP] StreamableHttpServerTransport implementation #449

[WIP] StreamableHttpServerTransport implementation #449

Conversation

devcrocod commented Dec 4, 2025 • edited by kpavlov Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

[WIP] StreamableHttpServerTransport implementation

How Has This Been Tested?

Breaking Changes

Types of changes

Checklist

Uh oh!

github-actions bot commented Dec 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

kpavlov Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

devcrocod Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

kpavlov Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

devcrocod Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

kpavlov Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

kpavlov Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

kpavlov Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

kpavlov Dec 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI Dec 5, 2025

Choose a reason for hiding this comment

Uh oh!

kpavlov left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

devcrocod commented Dec 4, 2025 •

edited by kpavlov

Loading

github-actions bot commented Dec 4, 2025 •

edited

Loading

kpavlov Dec 5, 2025 •

edited

Loading

kpavlov left a comment •

edited

Loading