René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"In Streamable HTTP transport, the sendNotification calls are getting timed out.","articleBody":"**Bug description**\n\nWhen sending a JSON-RPC notification using the MCP Java SDK's sendNotification method (with the Streamable HTTP transport), a timeout occurs if the server does not return any response body or headers. This happens even though notifications should not expect a response from the server. As a result, the returned Mono never completes, causing the client to hang or timeout.\n\n**Environment**\nMCP Java SDK version: Built the JAR on top of July 4 commit.\nJava version: 17\nOS: Windows\n\n**Steps to reproduce**\n1. Use the MCP Java SDK with the Streamable HTTP transport.\n2. Call client.initialize() to send initialization message through which a notifications/initialized notification will be sent to a MCP server that does not return a response for notifications (as per spec).\n3. Observe that a timeout occurs and the Mono returned by sendNotification never completes.\nExpected behavior\n4. The notification should be sent and the returned Mono should complete successfully, even if the server does not return any response. No timeout should occur.\n\n**Minimal Complete Reproducible example**\n```\nMcpClientTransport transport;\n\t\tString url = \"http://localhost:6009\";\n\t\tHttpClientStreamableHttpTransport.Builder httpBuilder = HttpClientStreamableHttpTransport.builder(url);\n\t\thttpBuilder.endpoint(\"/mcp/\");\n\t\ttransport = httpBuilder.build();\n\n\t\tMcpSyncClient client = McpClient.sync(transport)\n\t\t\t\t.requestTimeout(Duration.ofSeconds(20)) // Set a reasonable request timeout\n\t\t\t\t.initializationTimeout(Duration.ofSeconds(21)) // Set a reasonable initialization timeout\n\t\t\t\t.build();\n\n\t\ttry {\n\t\t\t// Initialize the connection\n\t\t\tclient.initialize();\n\n\t\t\tMcpSchema.ListToolsResult tools = client.listTools();\n\t\t\tlogger.log(Level.INFO, \"Available tools: {0}\", tools.tools());\n\n\t\t} catch (Exception e) {\n\t\t\tlogger.log(Level.SEVERE, \"Failed to initialize MCP client from DebugServlet\", e);\n\t\t}\n```\n\n**Workaround I did to fix it (which may not be proper):**\n\n```\npublic Mono\u003cVoid\u003e sendMessage(McpSchema.JSONRPCMessage sendMessage) {\n\t\tlogger.warning(\"DEBUG:: sendMessage called with message: \" + sendMessage);\n\t\treturn Mono.create(messageSink -\u003e {\n\t\t\tlogger.log(Level.INFO, \"Sending message {}\", sendMessage);\n\n\t\t\tfinal AtomicReference\u003cDisposable\u003e disposableRef = new AtomicReference\u003c\u003e();\n\t\t\tfinal McpTransportSession\u003cDisposable\u003e transportSession = this.activeSession.get();\n\n\t\t\tHttpRequest.Builder requestBuilder = this.requestBuilder.copy();\n\n\t\t\tif (transportSession != null \u0026\u0026 transportSession.sessionId().isPresent()) {\n\t\t\t\trequestBuilder = requestBuilder.header(\"mcp-session-id\", transportSession.sessionId().get());\n\t\t\t}\n\n\t\t\tString jsonBody = this.toString(sendMessage);\n\n\t\t\tHttpRequest request = requestBuilder.uri(Utils.resolveUri(this.baseUri, this.endpoint))\n\t\t\t\t.header(\"Accept\", TEXT_EVENT_STREAM + \", \" + APPLICATION_JSON)\n\t\t\t\t.header(\"Content-Type\", APPLICATION_JSON)\n\t\t\t\t.header(\"Cache-Control\", \"no-cache\")\n\t\t\t\t.POST(HttpRequest.BodyPublishers.ofString(jsonBody))\n\t\t\t\t.build();\n\n\t\t\tDisposable connection;\n\t\t\tif (sendMessage instanceof McpSchema.JSONRPCNotification) {\n\t\t\t\tlogger.log(Level.WARNING, \"DEBUG::: Sending notification message, no response expected\");\n\t\t\t\t// For notifications, complete the Mono immediately after sending the request\n\t\t\t\tMono.fromFuture(this.httpClient.sendAsync(request, this.toSendMessageBodySubscriber(null)))\n\t\t\t\t\t\t.doOnSuccess(response -\u003e messageSink.success())\n\t\t\t\t\t\t.doOnError(messageSink::error)\n\t\t\t\t\t\t.subscribe();\n\t\t\t\tconnection = null;\n\t\t\t} else {\n\t\t\t\tconnection = Flux.\u003cResponseEvent\u003ecreate(responseEventSink -\u003e {\n\n\t\t\t\t\t\t\t// Create the async request with proper body subscriber selection\n\t\t\t\t\t\t\tMono.fromFuture(this.httpClient.sendAsync(request, this.toSendMessageBodySubscriber(responseEventSink))\n\t\t\t\t\t\t\t\t\t.whenComplete((response, throwable) -\u003e {\n\t\t\t\t\t\t\t\t\t\tif (throwable != null) {\n\t\t\t\t\t\t\t\t\t\t\tresponseEventSink.error(throwable);\n\t\t\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\t\t\telse {\n\t\t\t\t\t\t\t\t\t\t\tlogger.log(Level.FINE, \"SSE connection established successfully\");\n\t\t\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\t\t})).onErrorMap(CompletionException.class, t -\u003e t.getCause()).onErrorComplete().subscribe();\n\n\t\t\t\t\t\t}).flatMap(responseEvent -\u003e {\n\t\t\t\t\t\t\tif (transportSession.markInitialized(\n\t\t\t\t\t\t\t\t\tresponseEvent.responseInfo().headers().firstValue(\"mcp-session-id\").orElseGet(() -\u003e null))) {\n\t\t\t\t\t\t\t\t// Once we have a session, we try to open an async stream for\n\t\t\t\t\t\t\t\t// the server to send notifications and requests out-of-band.\n\n\t\t\t\t\t\t\t\treconnect(null).contextWrite(messageSink.contextView()).subscribe();\n\t\t\t\t\t\t\t}\n\n\t\t\t\t\t\t\tString sessionRepresentation = sessionIdOrPlaceholder(transportSession);\n\n\t\t\t\t\t\t\tint statusCode = responseEvent.responseInfo().statusCode();\n\n\t\t\t\t\t\t\tif (statusCode \u003e= 200 \u0026\u0026 statusCode \u003c 300) {\n\n\t\t\t\t\t\t\t\tString contentType = responseEvent.responseInfo()\n\t\t\t\t\t\t\t\t\t\t.headers()\n\t\t\t\t\t\t\t\t\t\t.firstValue(\"Content-Type\")\n\t\t\t\t\t\t\t\t\t\t.orElse(\"\")\n\t\t\t\t\t\t\t\t\t\t.toLowerCase();\n\n\t\t\t\t\t\t\t\tif (contentType.isBlank()) {\n\t\t\t\t\t\t\t\t\tlogger.log(Level.INFO, \"No content type returned for POST in session {}\", sessionRepresentation);\n\t\t\t\t\t\t\t\t\treturn Flux.empty();\n\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\telse if (contentType.contains(TEXT_EVENT_STREAM)) {\n\t\t\t\t\t\t\t\t\treturn Flux.just(((ResponseSubscribers.SseResponseEvent) responseEvent).sseEvent())\n\t\t\t\t\t\t\t\t\t\t\t.flatMap(sseEvent -\u003e {\n\t\t\t\t\t\t\t\t\t\t\t\ttry {\n\t\t\t\t\t\t\t\t\t\t\t\t\t// We don't support batching ATM and probably won't\n\t\t\t\t\t\t\t\t\t\t\t\t\t// since the\n\t\t\t\t\t\t\t\t\t\t\t\t\t// next version considers removing it.\n\t\t\t\t\t\t\t\t\t\t\t\t\tMcpSchema.JSONRPCMessage message = McpSchema\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t.deserializeJsonRpcMessage(this.objectMapper, sseEvent.data());\n\n\t\t\t\t\t\t\t\t\t\t\t\t\tTuple2\u003cOptional\u003cString\u003e, Iterable\u003cMcpSchema.JSONRPCMessage\u003e\u003e idWithMessages = Tuples\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t.of(Optional.ofNullable(sseEvent.id()), List.of(message));\n\n\t\t\t\t\t\t\t\t\t\t\t\t\tMcpTransportStream\u003cDisposable\u003e sessionStream = new DefaultMcpTransportStream\u003c\u003e(\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\tthis.resumableStreams, this::reconnect);\n\n\t\t\t\t\t\t\t\t\t\t\t\t\tlogger.log(Level.FINE, \"Connected stream {}\", sessionStream.streamId());\n\n\t\t\t\t\t\t\t\t\t\t\t\t\tmessageSink.success();\n\n\t\t\t\t\t\t\t\t\t\t\t\t\treturn Flux.from(sessionStream.consumeSseStream(Flux.just(idWithMessages)));\n\t\t\t\t\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\t\t\t\t\tcatch (IOException ioException) {\n\t\t\t\t\t\t\t\t\t\t\t\t\treturn Flux.\u003cMcpSchema.JSONRPCMessage\u003eerror(\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\tnew McpError(\"Error parsing JSON-RPC message: \" + sseEvent.data()));\n\t\t\t\t\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\t\t\t\t});\n\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\telse if (contentType.contains(APPLICATION_JSON)) {\n\t\t\t\t\t\t\t\t\tmessageSink.success();\n\t\t\t\t\t\t\t\t\tString data = ((ResponseSubscribers.AggregateResponseEvent) responseEvent).data();\n\t\t\t\t\t\t\t\t\ttry {\n\t\t\t\t\t\t\t\t\t\treturn Mono.just(McpSchema.deserializeJsonRpcMessage(objectMapper, data));\n\t\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\t\tcatch (IOException e) {\n\t\t\t\t\t\t\t\t\t\treturn Mono.error(e);\n\t\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t\tlogger.log(Level.WARNING, \"Unknown media type {0} returned for POST in session {1}\", new Object[]{contentType,\n\t\t\t\t\t\t\t\t\t\tsessionRepresentation});\n\n\t\t\t\t\t\t\t\treturn Flux.\u003cMcpSchema.JSONRPCMessage\u003eerror(\n\t\t\t\t\t\t\t\t\t\tnew RuntimeException(\"Unknown media type returned: \" + contentType));\n\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\telse if (statusCode == NOT_FOUND) {\n\t\t\t\t\t\t\t\tMcpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(\n\t\t\t\t\t\t\t\t\t\t\"Session not found for session ID: \" + sessionRepresentation);\n\t\t\t\t\t\t\t\treturn Flux.\u003cMcpSchema.JSONRPCMessage\u003eerror(exception);\n\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t// Some implementations can return 400 when presented with a\n\t\t\t\t\t\t\t// session id that it doesn't know about, so we will\n\t\t\t\t\t\t\t// invalidate the session\n\t\t\t\t\t\t\t// https://github.com/modelcontextprotocol/typescript-sdk/issues/389\n\t\t\t\t\t\t\telse if (statusCode == BAD_REQUEST) {\n\t\t\t\t\t\t\t\tMcpTransportSessionNotFoundException exception = new McpTransportSessionNotFoundException(\n\t\t\t\t\t\t\t\t\t\t\"Session not found for session ID: \" + sessionRepresentation);\n\t\t\t\t\t\t\t\treturn Flux.\u003cMcpSchema.JSONRPCMessage\u003eerror(exception);\n\t\t\t\t\t\t\t}\n\n\t\t\t\t\t\t\treturn Flux.\u003cMcpSchema.JSONRPCMessage\u003eerror(\n\t\t\t\t\t\t\t\t\tnew RuntimeException(\"Failed to send message: \" + responseEvent));\n\t\t\t\t\t\t})\n\t\t\t\t\t\t.flatMap(jsonRpcMessage -\u003e this.handler.get().apply(Mono.just(jsonRpcMessage)))\n\t\t\t\t\t\t.onErrorMap(CompletionException.class, t -\u003e t.getCause())\n\t\t\t\t\t\t.onErrorComplete(t -\u003e {\n\t\t\t\t\t\t\t// handle the error first\n\t\t\t\t\t\t\tthis.handleException(t);\n\t\t\t\t\t\t\t// inform the caller of sendMessage\n\t\t\t\t\t\t\tmessageSink.error(t);\n\t\t\t\t\t\t\treturn true;\n\t\t\t\t\t\t})\n\t\t\t\t\t\t.doFinally(s -\u003e {\n\t\t\t\t\t\t\tlogger.log(Level.FINE, \"SendMessage finally: {}\", s);\n\t\t\t\t\t\t\tDisposable ref = disposableRef.getAndSet(null);\n\t\t\t\t\t\t\tif (ref != null) {\n\t\t\t\t\t\t\t\ttransportSession.removeConnection(ref);\n\t\t\t\t\t\t\t}\n\t\t\t\t\t\t})\n\t\t\t\t\t\t.contextWrite(messageSink.contextView())\n\t\t\t\t\t\t.subscribe();\n\t\t\t}\n\n\n\n\t\t\tif (connection != null) {\n\t\t\t\tdisposableRef.set(connection);\n\t\t\t\ttransportSession.addConnection(connection);\n\t\t\t}\n\t\t});\n\t}\n```\n\nNote: Posting this here to help fix this bug early. We are eagerly waiting for the release of Streamable Http client / server support from you guys. Appreciate your efforts!","author":{"url":"https://github.com/bharat-avn-14416","@type":"Person","name":"bharat-avn-14416"},"datePublished":"2025-07-14T12:37:48.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":9},"url":"https://github.com/396/java-sdk/issues/396"}