MCP Streamable Parsing Incomplete response causing the JsonEOFException

[nasonawa]

The issue stems from the QuarkusStreamableHttpMcpTransport.java using the Vert.x HttpClientResponse.handler() method, which processes each response chunk (e.g., TCP packet) individually, passing it to the SseSubscriber.accept() method for parsing. This causes the JsonEOFException because the SSE event (containing the full JSON-RPC "tools/list" response) is split across multiple chunksand the client attempts to parse the first chunk before receiving the complete event.

if (id != null && contentType != null && contentType.contains("text/event-stream")) {
                                        // the server has started a SSE channel
                                        response.result().handler(bodyBuffer -> {
                                            String responseString = bodyBuffer.toString();
                                            SseEvent<String> sseEvent = parseSseEvent(responseString);
                                            sseSubscriber.accept(sseEvent);
                                        });
                                    }

Using bodyHandler() (which buffers the entire response) could work but risks memory issues for large payloads,

String contentType = response.result().getHeader("Content-Type");
                                    if (id != null && contentType != null && contentType.contains("text/event-stream")) {
                                        // the server has started a SSE channel
                                        response.result().bodyHandler(bodyBuffer -> {
                                            if (bodyBuffer.length() > 0) {
                                                String responseString = bodyBuffer.toString();
                                                SseEvent<String> sseEvent = parseSseEvent(responseString);
                                                sseSubscriber.accept(sseEvent);
                                            }
                                        });
                                    }

Before Raising PR i like to discuss this, if there is any alternative way we can solve this.

ERROR:-

2025-07-28 18:48:42,878 INFO  [io.qua.lan.mcp.run.htt.QuarkusStreamableHttpMcpTransport] (executor-thread-2) Request: {"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{"roots":{"listChanged":false}},"clientInfo":{"name":"langchain4j","version":"1.0"}}}
2025-07-28 18:48:42,881 INFO  [io.qua.lan.mcp.run.htt.QuarkusStreamableHttpMcpTransport] (vert.x-eventloop-thread-0) Request: {"jsonrpc":"2.0","method":"notifications/initialized"}
2025-07-28 18:48:42,883 INFO  [io.qua.lan.mcp.run.htt.QuarkusStreamableHttpMcpTransport] (vert.x-eventloop-thread-0) Response: 
2025-07-28 18:48:42,884 INFO  [io.qua.lan.mcp.run.htt.QuarkusStreamableHttpMcpTransport] (executor-thread-2) Request: {"jsonrpc":"2.0","id":1,"method":"tools/list"}
2025-07-28 18:48:42,887 WARN  [io.qua.lan.mcp.run.htt.SseSubscriber] (vert.x-eventloop-thread-0) {"jsonrpc":"2.0","id":1,"result":{"tools":[{"description":"create a configmap in namespace","inputSchema":{"type":"object","required":["Info","data"],"properties":{"Info":{"type":"object","required":["name","namespace"],"properties":{"name":{"type":"string"},"namespace":{"type":"string"}},"additionalProperties":{"not":{}}},"data":{"type":"object","additionalProperties":{"type":"string"}}},"additionalProperties":{"not":{}}},"name":"create_configmap","outputSchema":{"type":"object","additionalProperties":{"not":{}}}},{"description":"create deployment in given namespace","inputSchema":{"type":"object","required":["name","namespace","replicas","containerName","containerImage"],"properties":{"containerImage":{"type":"string"},"containerName":{"type":"string"},"name":{"type":"string"},"namespace":{"type":"string"},"replicas":{"type":"integer"}},"additionalProperties":{"not":{}}},"name":"create_deployment","outputSchema":{"type":"object","additionalProperties":{"not":{}}}},{"description":"Create a pod in the namspace","inputSchema":{"type":"object","required":["name","namespace","containerName","containerImage"],"properties":{"containerImage":{"type":"string"},"containerName":{"type":"string"},"name":{"type":"string"},"namespace":{"type":"string"}},"additionalProperties":{"not":{}}},"name":"create_pod","outputSchema":{"type":"object","additionalProperties":{"not":{}}}},{"description":"create a generic key value secret in the namespace","inputSchema":{"type":"object","required":["Info","key","value"],"properties":{"Info":{"type":"object","required":["name","namespace"],"properties":{"name":{"type":"string"},"namespace":{"type":"string"}},"additionalProperties":{"not":{}}},"key":{"type":"string"},"value":{"type":"string"}},"additionalProperties":{"not":{}}},"name":"cr: com.fasterxml.jackson.core.io.JsonEOFException: Unexpected end-of-input: was expecting closing quote for a string value
 at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 1790]
        at com.fasterxml.jackson.core.base.ParserMinimalBase._reportInvalidEOF(ParserMinimalBase.java:642)
        at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._finishString2(ReaderBasedJsonParser.java:2186)
        at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._finishString(ReaderBasedJsonParser.java:2173)
        at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.getText(ReaderBasedJsonParser.java:295)
        at com.fasterxml.jackson.databind.deser.std.BaseNodeDeserializer._deserializeContainerNoRecursion(JsonNodeDeserializer.java:571)
        at com.fasterxml.jackson.databind.deser.std.JsonNodeDeserializer.deserialize(JsonNodeDeserializer.java:99)
        at com.fasterxml.jackson.databind.deser.std.JsonNodeDeserializer.deserialize(JsonNodeDeserializer.java:24)
        at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:342)
        at com.fasterxml.jackson.databind.ObjectMapper._readTreeAndClose(ObjectMapper.java:5013)
        at com.fasterxml.jackson.databind.ObjectMapper.readTree(ObjectMapper.java:3300)
        at io.quarkiverse.langchain4j.mcp.runtime.http.SseSubscriber.accept(SseSubscriber.java:46)
        at io.quarkiverse.langchain4j.mcp.runtime.http.QuarkusStreamableHttpMcpTransport.lambda$execute$3(QuarkusStreamableHttpMcpTransport.java:152)
        at io.vertx.core.impl.ContextInternal.dispatch(ContextInternal.java:270)
        at io.vertx.core.http.impl.HttpEventHandler.handleChunk(HttpEventHandler.java:51)
        at io.vertx.core.http.impl.HttpClientResponseImpl.handleChunk(HttpClientResponseImpl.java:239)
        at io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.lambda$new$0(Http1xClientConnection.java:432)
        at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:279)
        at io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:157)
        at io.vertx.core.http.impl.Http1xClientConnection$StreamImpl.handleChunk(Http1xClientConnection.java:730)
        at io.vertx.core.impl.ContextImpl.execute(ContextImpl.java:327)
        at io.vertx.core.impl.DuplicatedContext.execute(DuplicatedContext.java:158)
        at io.vertx.core.http.impl.Http1xClientConnection.handleResponseChunk(Http1xClientConnection.java:913)
        at io.vertx.core.http.impl.Http1xClientConnection.handleHttpMessage(Http1xClientConnection.java:834)
        at io.vertx.core.http.impl.Http1xClientConnection.handleMessage(Http1xClientConnection.java:799)
        at io.vertx.core.net.impl.ConnectionBase.read(ConnectionBase.java:159)
        at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:153)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
        at io.netty.channel.CombinedChannelDuplexHandler$DelegatingChannelHandlerContext.fireChannelRead(CombinedChannelDuplexHandler.java:436)
        at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346)
        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:318)
        at io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:251)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
        at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1357)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:868)
        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:796)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:732)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:658)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:998)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:1583)


2025-07-28 18:48:42,890 WARN  [io.qua.lan.mcp.run.htt.SseSubscriber] (vert.x-eventloop-thread-0) Received event with null name

[geoand]

Thanks a lot for reporting!

Using bodyHandler is not the right approach here as the stream could be unbounded. I will have a close look later on this week as I think we've fixed something similar recently in the Quarkus REST Client.

[geoand]

Do you know what the delimiter is for each "message"? A new line character? Do you have a good example of captured packets I could see, or a sample application I can run?

Thanks!

[nasonawa]

Hi @geoand , After looking into some resources, I found way to fix this using a 'io.vertx.core.parsetools.RecordParser' , The sse events will be separated by a pair of newline characters ''\n\n". so i added a Record parser with "\n\n" delimiter

                                if (isExpectedStatusCode(response.result().statusCode())) {
                                    // did the server assign a session ID?
                                    String mcpSessionId = response.result().getHeader("Mcp-Session-Id");
                                    if (mcpSessionId != null && !mcpSessionId.isEmpty()) {
                                        log.debug("Assigned MCP session ID: " + mcpSessionId);
                                        this.mcpSessionId.set(mcpSessionId);
                                    }
                                    //added the record parser to parse each event
                                    RecordParser parser = RecordParser.newDelimited("\n\n", bodyBuffer -> {
                                        String responseString = bodyBuffer.toString();
                                        SseEvent<String> sseEvent = parseSseEvent(responseString);
                                        sseSubscriber.accept(sseEvent);
                                    });
                                    String contentType = response.result().getHeader("Content-Type");
                                    if (id != null && contentType != null && contentType.contains("text/event-stream")) {
                                        // the server has started a SSE channel
                                        response.result().handler(parser);
                                    } else {
                                     

let me know your thoughts on this.

[geoand]

RecordParser was exactly what I was planning on using (and what the Quarkus REST Client uses under the hood) :).

So if you have already written the code, feel free to open a PR, which would be amazing to include a test as well

[jmartisk]

Sounds like a good solution, +1 from me too

[nasonawa]

hi @geoand, regarding the Tests, i can see that we have tests for Http Transport. should i create a same sets of tests for the StreamableHttp Transport ?

[geoand]

We really need a scenario which shows the problem you found in action.

[jmartisk]

hi @geoand, regarding the Tests, i can see that we have tests for Http Transport. should i create a same sets of tests for the StreamableHttp Transport ?

I would probably suggest doing in integration-tests: https://github.com/quarkiverse/quarkus-langchain4j/tree/main/integration-tests/mcp/src/test/java/io/quarkiverse/langchain4j/mcp/test there we have the same tests for http and streamable-http

But it depends what exactly you need to return from the server, I'm not sure you'd be able to reliably reproduce the problem (splitting the response into multiple frames) with a regular jbang-based MCP server that the other integration tests use, right?

[jmartisk]

I don't know if that's even possible to reproduce without some extremely low-level server-side API (may have to control the TCP stream directly), and apparently with the normal MCP servers that the tests use, the issue doesn't happen.

[geoand]

Would https://docs.quarkiverse.io/quarkus-mcp-server/dev/index.html#_testing help?

[jmartisk]

McpAssured? That is just a mock MCP client, we need to control the server (on a very low level), I'm thinking we may even have to resort to java.net.ServerSocket to be able to control the exact segments.. I tried chatGPT and it came up with this for a server that offers SSE with control over exact TCP segments:

        ServerSocket serverSocket = new ServerSocket(8080);
        System.out.println("Listening on port 8080...");

        while (true) {
            Socket clientSocket = serverSocket.accept();
            System.out.println("Client connected from " + clientSocket.getInetAddress());

            // Set TCP_NODELAY to avoid coalescing
            clientSocket.setTcpNoDelay(true);

            OutputStream out = clientSocket.getOutputStream();
            BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));

            // Write HTTP headers
            writer.write("HTTP/1.1 200 OK\r\n");
            writer.write("Content-Type: text/event-stream\r\n");
            writer.write("Cache-Control: no-cache\r\n");
            writer.write("Connection: keep-alive\r\n");
            writer.write("\r\n");
            writer.flush(); // Flush headers immediately

            // Start SSE message
            String part1 = "data: Hello";
            String part2 = " world\n\n";

            writer.write(part1);
            writer.flush(); // Send part1 in one TCP segment

            try { Thread.sleep(50); } catch (InterruptedException e) {}

            writer.write(part2);
            writer.flush(); // Send part2 in next TCP segment
        }

[jmartisk]

So writing a (even a relatively stupid stub) MCP server this way would be quite a bunch of work, which brings up the question whether it's worth the effort, unless we find an easier way. @nasonawa I'll leave it up to you, you can try to write a test, but if it turns out too complicated, I guess we can live without it.

[nasonawa]

As you mentioned, creating a server and sending segments is a kind of tedious task

I got this "JsonEOFException: Unexpected end-of-input:" because my MCP server was sending a large response which needs to be divided into 2 TCP packets. I am not sure if this is the correct way to test this, but I followed this approach and created a tool which generates a large response.

@Tool(description = "Converts the string to large event stream response")
    ToolResponse largeResponse() {
        StringBuilder sb = new StringBuilder();
        List<TextContent> lst = new ArrayList<>();
        for (int i = 0; i < 3500; i++) {
              sb.append("This is the long response no"+ i);
        }
        lst.add(new TextContent(sb.toString()));
        ToolResponse t = new ToolResponse(false, lst);
        return t;
    }

When I invoked this tool from my application, I was able to reproduce the actual error.

2025-07-31 09:18:14,699 INFO  [io.qua.lan.mcp.run.htt.QuarkusStreamableHttpMcpTransport] (executor-thread-2) Request: {"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"largeResponse","arguments":{}}}
2025-07-31 09:18:14,723 WARN  [io.qua.lan.run.dev.ChatJsonRPCService] (executor-thread-2) java.lang.RuntimeException: java.util.concurrent.ExecutionException: com.fasterxml.jackson.core.io.JsonEOFException: Unexpected end-of-input: was expecting closing quote for a string value
 at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 1974]

So I added this tool in the "tools_mcp_server.java" JBang script, along with it I added a sample test to check if this tool invocation fails with RuntimeException.

    @Test
    public void largeevent() {
        ToolProviderResult toolProviderResult = toolProvider.provideTools(null);
        ToolExecutor executor = toolProviderResult.toolExecutorByName("largeResponse");
        ToolExecutionRequest toolExecutionRequest = ToolExecutionRequest.builder()
                .name("largeResponse")
                .arguments("{}")
                .build();
        String toolExecutionResultString = executor.execute(toolExecutionRequest, null);
        assertThat(toolExecutionResultString).isNotEmpty();

    }

When I ran the test, the test passed without any error; I am not sure why. Even if I ran "tools_mcp_server.java" JBang script and tested it with my application, the error was generating.

[jmartisk]

Just to be sure, did you run it via McpToolsHttpTransportIT and not McpToolsStdioTransportIT? (stdio obviously won't have this problem). And did you run it without applying your fix for the issue? :)