Add notifications test

Signed-off-by: Sergey Karpov <[email protected]>

Author: skarpovdev

kotlin-sdk-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TypeScriptClientKotlinServerTest.kt

@@ -57,6 +57,24 @@ class TypeScriptClientKotlinServerTest : TypeScriptTestBase() {
         )
     }
 
+    @Test
+    @Timeout(30, unit = TimeUnit.SECONDS)
+    fun testNotifications() {
+        val name = "NotifUser"
+        val command = "npx tsx myClient.ts $serverUrl multi-greet $name"
+        val output = executeCommand(command, tsClientDir)
+
+        assertTrue(
+            output.contains("Multiple greetings") || output.contains("greeting"),
+            "Tool response should contain greeting message",
+        )
+        // verify that the server sent 3 notifications
+        assertTrue(
+            output.contains("\"notificationCount\": 3") || output.contains("notificationCount: 3"),
+            "Structured content should indicate that 3 notifications were emitted by the server.\nOutput:\n$output",
+        )
+    }
+
     @Test
     @Timeout(30, unit = TimeUnit.SECONDS)
     fun testToolCallWithSessionManagement() {

kotlin-sdk-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/utils/KotlinServerForTypeScriptClient.kt

@@ -12,7 +12,9 @@ import io.ktor.server.request.receiveText
 import io.ktor.server.response.header
 import io.ktor.server.response.respond
 import io.ktor.server.response.respondText
+import io.ktor.server.response.respondTextWriter
 import io.ktor.server.routing.delete
+import io.ktor.server.routing.get
 import io.ktor.server.routing.post
 import io.ktor.server.routing.routing
 import io.modelcontextprotocol.kotlin.sdk.CallToolResult
@@ -66,6 +68,20 @@ class KotlinServerForTypeScriptClient {
 
         server = embeddedServer(CIO, port = port) {
             routing {
+                get("/mcp") {
+                    val sessionId = call.request.header("mcp-session-id")
+                    if (sessionId == null) {
+                        call.respond(HttpStatusCode.BadRequest, "Missing mcp-session-id header")
+                        return@get
+                    }
+                    val transport = serverTransports[sessionId]
+                    if (transport == null) {
+                        call.respond(HttpStatusCode.BadRequest, "Invalid mcp-session-id")
+                        return@get
+                    }
+                    transport.stream(call)
+                }
+
                 post("/mcp") {
                     val sessionId = call.request.header("mcp-session-id")
                     val requestBody = call.receiveText()
@@ -235,6 +251,32 @@ class KotlinServerForTypeScriptClient {
         ) { request ->
             val name = (request.arguments["name"] as? JsonPrimitive)?.content ?: "World"
 
+            server.sendToolListChanged()
+            server.sendLoggingMessage(
+                io.modelcontextprotocol.kotlin.sdk.LoggingMessageNotification(
+                    io.modelcontextprotocol.kotlin.sdk.LoggingMessageNotification.Params(
+                        level = io.modelcontextprotocol.kotlin.sdk.LoggingLevel.info,
+                        data = JsonPrimitive("Preparing greeting for $name")
+                    )
+                )
+            )
+            server.sendLoggingMessage(
+                io.modelcontextprotocol.kotlin.sdk.LoggingMessageNotification(
+                    io.modelcontextprotocol.kotlin.sdk.LoggingMessageNotification.Params(
+                        level = io.modelcontextprotocol.kotlin.sdk.LoggingLevel.info,
+                        data = JsonPrimitive("Halfway there for $name")
+                    )
+                )
+            )
+            server.sendLoggingMessage(
+                io.modelcontextprotocol.kotlin.sdk.LoggingMessageNotification(
+                    io.modelcontextprotocol.kotlin.sdk.LoggingMessageNotification.Params(
+                        level = io.modelcontextprotocol.kotlin.sdk.LoggingLevel.info,
+                        data = JsonPrimitive("Done sending greetings to $name")
+                    )
+                )
+            )
+
             CallToolResult(
                 content = listOf(TextContent("Multiple greetings sent to $name!")),
                 structuredContent = buildJsonObject {
@@ -297,6 +339,30 @@ class HttpServerTransport(private val sessionId: String) : AbstractTransport() {
     private val pendingResponses = ConcurrentHashMap<String, CompletableDeferred<JSONRPCMessage>>()
     private val messageQueue = Channel<JSONRPCMessage>(Channel.UNLIMITED)
 
+    suspend fun stream(call: ApplicationCall) {
+        logger.debug { "Starting SSE stream for session: $sessionId" }
+        call.response.header("Cache-Control", "no-cache")
+        call.response.header("Connection", "keep-alive")
+        call.respondTextWriter(ContentType.Text.EventStream) {
+            try {
+                while (true) {
+                    val result = messageQueue.receiveCatching()
+                    val msg = result.getOrNull() ?: break
+                    val json = McpJson.encodeToString(msg)
+                    write("event: message\n")
+                    write("data: ")
+                    write(json)
+                    write("\n\n")
+                    flush()
+                }
+            } catch (e: Exception) {
+                logger.warn(e) { "SSE stream terminated for session: $sessionId" }
+            } finally {
+                logger.debug { "SSE stream closed for session: $sessionId" }
+            }
+        }
+    }
+
     suspend fun handleRequest(call: ApplicationCall, requestBody: JsonElement) {
         try {
             logger.info { "Handling request body: $requestBody" }

kotlin-sdk-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/utils/myClient.ts

@@ -44,6 +44,20 @@ async function main() {
         await client.connect(transport, {protocolVersion: PROTOCOL_VERSION});
         console.log('Connected to server');
 
+        try {
+            if (typeof (client as any).on === 'function') {
+                (client as any).on('notification', (n: any) => {
+                    try {
+                        const method = (n && (n.method || (n.params && n.params.method))) || 'unknown';
+                        console.log('Notification:', method, JSON.stringify(n));
+                    } catch {
+                        console.log('Notification: <unparsable>');
+                    }
+                });
+            }
+        } catch {
+        }
+
         const toolsResult = await client.listTools();
         const tools = toolsResult.tools;
         console.log('Available utils:', tools.map((t: { name: any; }) => t.name).join(', '));
@@ -105,4 +119,4 @@ main().catch(error => {
     console.error('Unhandled error:', error);
     // @ts-ignore
     process.exit(1);
-});
+});