feat: add Streamable HTTP transport for MCP server - #1921
Conversation
Replace legacy SSE transport with Streamable HTTP transport for the MCP server. This upgrades the MCP SDK from 0.10.0 to 1.1.0 and exposes a single /mcp endpoint instead of separate /mcp/sse + /mcp/message paths. The WebFluxStreamableServerTransportProvider is vendored from Spring AI 2.0.0-M4 with a compatibility fix for Spring Framework 6.2 (replaces HttpHeaders.asMultiValueMap() which requires Spring 7.0). Closes kafbat#1137
kapybro
Bot
added
status/triage/manual
|
AI Summary The issue proposes replacing the legacy Server-Sent Events (SSE) transport in the MCP server with a Streamable HTTP transport, upgrading the MCP SDK from 0.10.0 to 1.1.0. This change consolidates endpoints into a single |
kapybro
Bot
changed the title
kapybro
Bot
added
area/internal
📝 WalkthroughWalkthroughThe MCP server migrates from the legacy SSE transport to Streamable HTTP, adds session-aware WebFlux GET/POST/DELETE handling, updates MCP SDK APIs and dependencies, and configures analysis exclusions for the new transport provider. ChangesMCP Streamable HTTP migration
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related issues
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
sappusaketh
changed the title
There was a problem hiding this comment.
Pull request overview
This PR migrates the Kafka UI MCP server from the legacy SSE transport to the MCP “Streamable HTTP” transport, upgrading the MCP SDK to 1.1.0 and consolidating transport routing behind a single /mcp endpoint. It also vendors a Spring AI WebFlux transport provider (with a Spring Framework 6.2 compatibility tweak) and adjusts code style/static analysis settings to accommodate the vendored source.
Changes:
- Replace the previous MCP Spring WebFlux SSE transport dependency with MCP SDK 1.1.0 core + Jackson JSON modules and update MCP tool invocation/signatures accordingly.
- Introduce a vendored
WebFluxStreamableServerTransportProviderand wire it via Spring configuration to expose/mcpwith Streamable HTTP semantics. - Update Sonar + Checkstyle configuration to exclude/relax rules for the vendored transport provider file.
Reviewed changes
Copilot reviewed 8 out of 9 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| sonar-project.properties | Adds Sonar project settings and CPD exclusion for the vendored transport provider. |
| gradle/libs.versions.toml | Upgrades MCP dependencies by adding MCP SDK 1.1.0 core + Jackson modules. |
| etc/checkstyle/checkstyle-suppressions.xml | Suppresses Checkstyle checks for the vendored transport provider file. |
| build.gradle | Adds Sonar CPD exclusion for the vendored transport provider. |
| api/build.gradle | Switches dependencies to MCP SDK 1.1.0 artifacts and wires Checkstyle suppression config. |
| api/src/test/java/io/kafbat/ui/service/mcp/McpSpecificationGeneratorTest.java | Updates tests for MCP SDK API changes (tool builder + call handler signature). |
| api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java | Adds vendored Streamable HTTP WebFlux transport provider implementation. |
| api/src/main/java/io/kafbat/ui/service/mcp/McpSpecificationGenerator.java | Adapts tool schema generation + call handling to MCP SDK 1.1.0 APIs. |
| api/src/main/java/io/kafbat/ui/config/McpConfig.java | Rewires MCP server beans to use the new Streamable HTTP transport provider and unified /mcp endpoint. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| List<MediaType> acceptHeaders = request.headers().asHttpHeaders().getAccept(); | ||
| if (!(acceptHeaders.contains(MediaType.APPLICATION_JSON) | ||
| && acceptHeaders.contains(MediaType.TEXT_EVENT_STREAM))) { | ||
| return ServerResponse.badRequest().build(); | ||
| } |
| String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID); | ||
|
|
||
| McpStreamableServerSession session = this.sessions.get(sessionId); | ||
|
|
||
| if (session == null) { | ||
| return ServerResponse.notFound().build(); | ||
| } | ||
|
|
||
| return session.delete().then(ServerResponse.ok().build()); |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
🧹 Nitpick comments (1)
api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java (1)
192-192: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winExtract the duplicated shutdown response text.
Sonar reports this duplication as a blocking failure. Introduce one
SERVER_SHUTTING_DOWNconstant and use it in all three handlers.Also applies to: 248-248, 361-361
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java` at line 192, Extract the repeated "Server is shutting down" text into a single SERVER_SHUTTING_DOWN constant in WebFluxStreamableServerTransportProvider, then replace the literal in all three shutdown response handlers, including the locations corresponding to the referenced lines.Source: Linters/SAST tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@api/src/main/java/io/kafbat/ui/config/McpConfig.java`:
- Around line 28-31: Update McpConfig.streamableServerTransport() to configure a
non-NOOP Origin security validator on
WebFluxStreamableServerTransportProvider.builder(), using the transport’s
supported security-validator configuration so the /mcp endpoint retains
DNS-rebinding protection.
In
`@api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java`:
- Around line 283-300: Update the initialization flow around
sessionFactory.startSession, sessions.put, and init.initResult() to remove
init.session() from sessions whenever initialization, mapping, serialization, or
response creation fails. Preserve the existing successful response behavior and
ensure cleanup occurs for the session inserted before initResult completes.
- Around line 246-356: The handlePost method exceeds the cognitive-complexity
threshold; split its message-specific logic into focused private handlers.
Extract initialization handling, session ID validation and lookup, and JSON-RPC
request/notification/response dispatch from handlePost while preserving all
existing responses, session behavior, streaming, and error handling.
- Around line 384-392: Update the session deletion flow in the request handler
using the sessions registry and McpStreamableServerSession lookup: atomically
remove the session associated with sessionId from this.sessions before invoking
session.delete(). Preserve the existing not-found response when no session is
registered and return the successful response after deletion completes.
- Around line 170-183: Update closeGracefully() to perform session clearing and
keepAliveScheduler shutdown in a doFinally callback instead of doOnSuccess,
ensuring cleanup runs after successful completion, errors, and cancellation
while preserving the existing shutdown flow.
- Around line 216-218: The shared session-request handling around sessionId
lookup must also validate HttpHeaders.MCP_PROTOCOL_VERSION before retrieving or
using the session. Accept only the requested, supported, or negotiated protocol
value; return HTTP 400 for missing, invalid, or unsupported values, and apply
this consistently to GET, non-initialize POST, and DELETE paths.
---
Nitpick comments:
In
`@api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java`:
- Line 192: Extract the repeated "Server is shutting down" text into a single
SERVER_SHUTTING_DOWN constant in WebFluxStreamableServerTransportProvider, then
replace the literal in all three shutdown response handlers, including the
locations corresponding to the referenced lines.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: c4bd728f-797a-4648-b133-56d89c92ae0d
📒 Files selected for processing (9)
api/build.gradleapi/src/main/java/io/kafbat/ui/config/McpConfig.javaapi/src/main/java/io/kafbat/ui/service/mcp/McpSpecificationGenerator.javaapi/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.javaapi/src/test/java/io/kafbat/ui/service/mcp/McpSpecificationGeneratorTest.javabuild.gradleetc/checkstyle/checkstyle-suppressions.xmlgradle/libs.versions.tomlsonar-project.properties
| public WebFluxStreamableServerTransportProvider streamableServerTransport() { | ||
| return WebFluxStreamableServerTransportProvider.builder() | ||
| .messageEndpoint("/mcp") | ||
| .build(); |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== Locate McpConfig.java and MCP deps ="
git ls-files | rg 'McpConfig.java|pom.xml|build.gradle|gradle.properties' || true
echo
echo "== McpConfig contents =="
cat -n api/src/main/java/io/kafbat/ui/config/McpConfig.java
echo
echo "== Search WebFluxStreamableServerTransportProvider usages/imports/settings =="
rg -n "WebFluxStreamableServerTransportProvider|ServerTransportSecurityValidator|securityValidator|/mcp|ServerTransportSecurityValidator" .Repository: kafbat/kafka-ui
Length of output: 7849
🌐 Web query:
Model Context Protocol streamable HTTP transport origin validation modelcontextprotocol.io
💡 Result:
In the Model Context Protocol (MCP) Streamable HTTP transport, origin validation is a mandatory security requirement designed to prevent DNS rebinding attacks [1][2][3]. The specification mandates the following regarding Origin header validation: 1. Mandatory Validation: Servers must validate the Origin header on all incoming connections if it is present [1][2][3]. 2. Rejection Criteria: If the Origin header is present and deemed invalid, the server must reject the request with an HTTP 403 Forbidden status [1][2][3]. The response body may optionally contain a JSON-RPC error response [1][2]. 3. Handling Missing Headers: Requests that do not include an Origin header are generally permitted to pass, as non-browser MCP clients do not typically send this header, and they are not considered relevant to the specific DNS rebinding threats that this validation defends against [4][5]. Implementation details for developers: - Official SDKs provide middleware and transport options (e.g., allowedOrigins in the TypeScript SDK) to facilitate this validation [6][7][4]. - Validation is strictly "deny-on-failure": if a header is present but cannot be parsed (or matches no entry in an allowlist), the request must be rejected [4]. - For local development, servers are also strongly encouraged to bind only to localhost (127.0.0.1) and implement authentication as additional layers of security [1][3]. These requirements ensure that local MCP servers are protected from unauthorized cross-site requests while remaining compatible with non-browser-based clients [4][5].
Citations:
- 1: https://modelcontextprotocol.io/specification/2026-07-28/basic/transports/streamable-http
- 2: https://modelcontextprotocol.io/specification/draft/basic/transports/streamable-http
- 3: https://modelcontextprotocol.io/specification/2025-11-25/basic/transports
- 4: https://ts.sdk.modelcontextprotocol.io/v2/api/@modelcontextprotocol/server/server/middleware/originValidation.html
- 5: Modify Origin header validation in validateRequestHeaders (streamableHttp.ts and sse.ts) to allow requests without an Origin, as they are not relevant to server DNS rebinding protection. modelcontextprotocol/typescript-sdk#1205
- 6: https://ts.sdk.modelcontextprotocol.io/v2/api/@modelcontextprotocol/server/server/streamableHttp.html
- 7: https://github.com/modelcontextprotocol/typescript-sdk/blob/2c0c481cb9dbfd15c8613f765c940a5f5bace94d/packages/server/src/server/streamableHttp.ts
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== MCP dependency declarations =="
rg -n "modelcontextprotocol|io.modelcontextprotocol|mcp-enabled|mcp" build.gradle api/build.gradle gradle.properties README.md || true
echo
echo "== Builder and validator implementation slice =="
sed -n '430,515p' api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java
echo
echo "== Security validator implementation declaration and calls =="
rg -n "class ServerTransportSecurityValidator|interface ServerTransportSecurityValidator|impl|validateHeaders|Origin|Origin" api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.javaRepository: kafbat/kafka-ui
Length of output: 3357
Configure an Origin security validator for the MCP transport.
WebFluxStreamableServerTransportProvider.builder() defaults to ServerTransportSecurityValidator.NOOP, so the /mcp route performs no Origin validation. Set a non-noop validator (for example via .securityValidator(...)) to preserve the Streamable HTTP transport’s DNS-rebinding protection.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@api/src/main/java/io/kafbat/ui/config/McpConfig.java` around lines 28 - 31,
Update McpConfig.streamableServerTransport() to configure a non-NOOP Origin
security validator on WebFluxStreamableServerTransportProvider.builder(), using
the transport’s supported security-validator configuration so the /mcp endpoint
retains DNS-rebinding protection.
| public Mono<Void> closeGracefully() { | ||
| return Mono.defer(() -> { | ||
| this.isClosing = true; | ||
| return Flux.fromIterable(this.sessions.values()) | ||
| .doFirst(() -> logger.debug("Initiating graceful shutdown with {} active sessions", | ||
| this.sessions.size())) | ||
| .flatMap(McpStreamableServerSession::closeGracefully) | ||
| .then(); | ||
| }).then().doOnSuccess(v -> { | ||
| this.sessions.clear(); | ||
| if (this.keepAliveScheduler != null) { | ||
| this.keepAliveScheduler.shutdown(); | ||
| } | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Run shutdown cleanup on errors and cancellation too.
doOnSuccess is skipped if any session fails to close, leaving the keep-alive scheduler running and sessions retained. Move unconditional cleanup to doFinally.
Proposed fix
- }).then().doOnSuccess(v -> {
+ }).then().doFinally(signalType -> {
this.sessions.clear();
if (this.keepAliveScheduler != null) {
this.keepAliveScheduler.shutdown();
}
});📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| public Mono<Void> closeGracefully() { | |
| return Mono.defer(() -> { | |
| this.isClosing = true; | |
| return Flux.fromIterable(this.sessions.values()) | |
| .doFirst(() -> logger.debug("Initiating graceful shutdown with {} active sessions", | |
| this.sessions.size())) | |
| .flatMap(McpStreamableServerSession::closeGracefully) | |
| .then(); | |
| }).then().doOnSuccess(v -> { | |
| this.sessions.clear(); | |
| if (this.keepAliveScheduler != null) { | |
| this.keepAliveScheduler.shutdown(); | |
| } | |
| }); | |
| public Mono<Void> closeGracefully() { | |
| return Mono.defer(() -> { | |
| this.isClosing = true; | |
| return Flux.fromIterable(this.sessions.values()) | |
| .doFirst(() -> logger.debug("Initiating graceful shutdown with {} active sessions", | |
| this.sessions.size())) | |
| .flatMap(McpStreamableServerSession::closeGracefully) | |
| .then(); | |
| }).then().doFinally(signalType -> { | |
| this.sessions.clear(); | |
| if (this.keepAliveScheduler != null) { | |
| this.keepAliveScheduler.shutdown(); | |
| } | |
| }); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java`
around lines 170 - 183, Update closeGracefully() to perform session clearing and
keepAliveScheduler shutdown in a doFinally callback instead of doOnSuccess,
ensuring cleanup runs after successful completion, errors, and cancellation
while preserving the existing shutdown flow.
| String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID); | ||
|
|
||
| McpStreamableServerSession session = this.sessions.get(sessionId); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and inspect the referenced file and related symbol usages.
file="api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java"
if [ ! -f "$file" ]; then
fd -i 'WebFluxStreamableServerTransportProvider.java'
exit 0
fi
wc -l "$file"
echo '--- outline ---'
ast-grep outline "$file" --view compact || true
echo '--- relevant sections 180-405 ---'
sed -n '180,405p' "$file" | nl -ba -v180
echo '--- protocol header references ---'
rg -n "MCP-Protocol-Version|MCP_SESSION_ID|StreamableServerSession|GET|DELETE|post" "$file"Repository: kafbat/kafka-ui
Length of output: 486
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java"
echo '--- relevant sections 1-430 ---'
awk '{printf "%d\t%s\n", NR, $0}' "$file" | sed -n '1,430p'
echo '--- protocol/header references ---'
grep -nE "MCP-Protocol-Version|PROTOCOL|SESSION_ID|McpStreamableServerSession|GET|DELETE|Post|post" "$file" || trueRepository: kafbat/kafka-ui
Length of output: 19941
🌐 Web query:
MCP specification 2025-06-18 basic transports MCP-Protocol-Version HTTP requests subsequent requests required unsupported values 400
💡 Result:
Under the Model Context Protocol (MCP) specification, particularly when using the Streamable HTTP transport, clients are required to include the MCP-Protocol-Version HTTP header on all requests [1][2][3]. This header allows the MCP server to respond according to the specific protocol version requested [1][2]. If a server receives a request with an invalid or unsupported MCP-Protocol-Version header, it is required to respond with a 400 Bad Request status code [2][3]. In modern versions of the specification, the server should also provide an UnsupportedProtocolVersionError in the response body, which lists the protocol versions that the server does support [3][4]. Key aspects of this requirement include: 1. Consistency: The value in the MCP-Protocol-Version header must match the version declared in the io.modelcontextprotocol/protocolVersion field within the request body's _meta object [3][5]. If these values mismatch, the server must reject the request with a 400 Bad Request and a HeaderMismatch error [3]. 2. Negotiation: The protocol version used by the client should typically align with the version negotiated during the initialization phase [2][3]. 3. Backwards Compatibility: While older versions of the specification (pre-2025-06-18) did not always define this header, servers that do not support legacy behavior must reject requests missing this header [3]. Servers that do support legacy clients may optionally assume a default version (such as 2025-03-26) if the header is omitted, depending on the specific implementation requirements [2][3]. For detailed implementation, consult the specific version of the MCP specification you are targeting, as newer drafts have formalized the UnsupportedProtocolVersionError and header-body validation requirements [3][4].
Citations:
- 1: https://modelcontextprotocol.io/specification/2025-06-18/basic/transports
- 2: https://modelcontextprotocol.io/specification/2025-11-25/basic/transports
- 3: https://modelcontextprotocol.io/specification/draft/basic/transports/streamable-http
- 4: https://modelcontextprotocol.io/specification/draft/basic/versioning
- 5: https://modelcontextprotocol.io/specification/2026-07-28/basic
Validate MCP-Protocol-Version on session requests.
GET, non-initialize POST, and DELETE retrieve a session only after checking MCP_SESSION_ID; they should also check HttpHeaders.MCP_PROTOCOL_VERSION for the requested/unsupported/negotiated value and return HTTP 400 when it is invalid or unsupported. Add this validation in shared handling so all session-dependent request paths enforce it.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java`
around lines 216 - 218, The shared session-request handling around sessionId
lookup must also validate HttpHeaders.MCP_PROTOCOL_VERSION before retrieving or
using the session. Accept only the requested, supported, or negotiated protocol
value; return HTTP 400 for missing, invalid, or unsupported values, and apply
this consistently to GET, non-initialize POST, and DELETE paths.
| private Mono<ServerResponse> handlePost(ServerRequest request) { | ||
| if (this.isClosing) { | ||
| return ServerResponse.status(HttpStatus.SERVICE_UNAVAILABLE).bodyValue("Server is shutting down"); | ||
| } | ||
|
|
||
| try { | ||
| Map<String, List<String>> headers = toHeaderMap(request.headers().asHttpHeaders()); | ||
| this.securityValidator.validateHeaders(headers); | ||
| } | ||
| catch (ServerTransportSecurityException e) { | ||
| String errorMessage = e.getMessage(); | ||
| return ServerResponse.status(e.getStatusCode()).bodyValue(errorMessage != null ? errorMessage : ""); | ||
| } | ||
|
|
||
| McpTransportContext transportContext = this.contextExtractor.extract(request); | ||
|
|
||
| List<MediaType> acceptHeaders = request.headers().asHttpHeaders().getAccept(); | ||
| if (!(acceptHeaders.contains(MediaType.APPLICATION_JSON) | ||
| && acceptHeaders.contains(MediaType.TEXT_EVENT_STREAM))) { | ||
| return ServerResponse.badRequest().build(); | ||
| } | ||
|
|
||
| return request.bodyToMono(String.class).<ServerResponse>flatMap(body -> { | ||
| try { | ||
| McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(this.jsonMapper, body); | ||
| if (message instanceof McpSchema.JSONRPCRequest jsonrpcRequest | ||
| && jsonrpcRequest.method().equals(McpSchema.METHOD_INITIALIZE)) { | ||
| if (this.sessionFactory == null) { | ||
| return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR) | ||
| .bodyValue(McpError.builder(McpSchema.ErrorCodes.INTERNAL_ERROR) | ||
| .message("Session factory not initialized") | ||
| .build()); | ||
| } | ||
| var typeReference = new TypeRef<McpSchema.InitializeRequest>() { | ||
| }; | ||
| McpSchema.InitializeRequest initializeRequest = this.jsonMapper | ||
| .convertValue(jsonrpcRequest.params(), typeReference); | ||
| McpStreamableServerSession.McpStreamableServerSessionInit init = this.sessionFactory | ||
| .startSession(initializeRequest); | ||
| this.sessions.put(init.session().getId(), init.session()); | ||
| return init.initResult().map(initializeResult -> { | ||
| McpSchema.JSONRPCResponse jsonrpcResponse = new McpSchema.JSONRPCResponse( | ||
| McpSchema.JSONRPC_VERSION, jsonrpcRequest.id(), initializeResult, null); | ||
| try { | ||
| return this.jsonMapper.writeValueAsString(jsonrpcResponse); | ||
| } | ||
| catch (IOException e) { | ||
| logger.warn("Failed to serialize initResponse", e); | ||
| throw Exceptions.propagate(e); | ||
| } | ||
| }) | ||
| .flatMap(initResult -> ServerResponse.ok() | ||
| .contentType(MediaType.APPLICATION_JSON) | ||
| .header(HttpHeaders.MCP_SESSION_ID, init.session().getId()) | ||
| .bodyValue(initResult)); | ||
| } | ||
|
|
||
| if (request.headers().header(HttpHeaders.MCP_SESSION_ID).isEmpty()) { | ||
| return ServerResponse.badRequest() | ||
| .bodyValue(McpError.builder(McpSchema.ErrorCodes.METHOD_NOT_FOUND) | ||
| .message("Session ID missing") | ||
| .build()); | ||
| } | ||
|
|
||
| String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID); | ||
| McpStreamableServerSession session = this.sessions.get(sessionId); | ||
|
|
||
| if (session == null) { | ||
| return ServerResponse.status(HttpStatus.NOT_FOUND) | ||
| .bodyValue(McpError.builder(McpSchema.ErrorCodes.INTERNAL_ERROR) | ||
| .message("Session not found: " + sessionId) | ||
| .build()); | ||
| } | ||
|
|
||
| if (message instanceof McpSchema.JSONRPCResponse jsonrpcResponse) { | ||
| return session.accept(jsonrpcResponse).then(ServerResponse.accepted().build()); | ||
| } | ||
| else if (message instanceof McpSchema.JSONRPCNotification jsonrpcNotification) { | ||
| return session.accept(jsonrpcNotification).then(ServerResponse.accepted().build()); | ||
| } | ||
| else if (message instanceof McpSchema.JSONRPCRequest jsonrpcRequest) { | ||
| return ServerResponse.ok() | ||
| .contentType(MediaType.TEXT_EVENT_STREAM) | ||
| .body(Flux.<ServerSentEvent<?>>create(sink -> { | ||
| WebFluxStreamableMcpSessionTransport st = new WebFluxStreamableMcpSessionTransport(sink); | ||
| Mono<Void> stream = session.responseStream(jsonrpcRequest, st); | ||
| Disposable streamSubscription = stream.onErrorComplete(err -> { | ||
| sink.error(err); | ||
| return true; | ||
| }).contextWrite(sink.contextView()).subscribe(); | ||
| sink.onCancel(streamSubscription); | ||
| }).contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext)), | ||
| ServerSentEvent.class); | ||
| } | ||
| else { | ||
| return ServerResponse.badRequest() | ||
| .bodyValue(McpError.builder(McpSchema.ErrorCodes.INVALID_REQUEST) | ||
| .message("Unknown message type") | ||
| .build()); | ||
| } | ||
| } | ||
| catch (IllegalArgumentException | IOException e) { | ||
| logger.error("Failed to deserialize message: {}", e.getMessage()); | ||
| return ServerResponse.badRequest() | ||
| .bodyValue(McpError.builder(McpSchema.ErrorCodes.INVALID_REQUEST) | ||
| .message("Invalid message format") | ||
| .build()); | ||
| } | ||
| }) | ||
| .switchIfEmpty(ServerResponse.badRequest().build()) | ||
| .contextWrite(ctx -> ctx.put(McpTransportContext.KEY, transportContext)); |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Split handlePost into message-specific handlers.
Sonar reports cognitive complexity 27 versus the allowed 15. Extract initialization, session lookup, and request/notification/response dispatch into focused methods to unblock the quality gate.
🧰 Tools
🪛 GitHub Check: SonarCloud Code Analysis
[warning] 347-347: Replace "e" with an unnamed pattern.
[failure] 246-246: Refactor this method to reduce its Cognitive Complexity from 27 to the 15 allowed.
[warning] 320-320: Replace the chain of if/else with a switch expression.
[warning] 292-292: Replace "e" with an unnamed pattern.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java`
around lines 246 - 356, The handlePost method exceeds the cognitive-complexity
threshold; split its message-specific logic into focused private handlers.
Extract initialization handling, session ID validation and lookup, and JSON-RPC
request/notification/response dispatch from handlePost while preserving all
existing responses, session behavior, streaming, and error handling.
Source: Linters/SAST tools
| McpStreamableServerSession.McpStreamableServerSessionInit init = this.sessionFactory | ||
| .startSession(initializeRequest); | ||
| this.sessions.put(init.session().getId(), init.session()); | ||
| return init.initResult().map(initializeResult -> { | ||
| McpSchema.JSONRPCResponse jsonrpcResponse = new McpSchema.JSONRPCResponse( | ||
| McpSchema.JSONRPC_VERSION, jsonrpcRequest.id(), initializeResult, null); | ||
| try { | ||
| return this.jsonMapper.writeValueAsString(jsonrpcResponse); | ||
| } | ||
| catch (IOException e) { | ||
| logger.warn("Failed to serialize initResponse", e); | ||
| throw Exceptions.propagate(e); | ||
| } | ||
| }) | ||
| .flatMap(initResult -> ServerResponse.ok() | ||
| .contentType(MediaType.APPLICATION_JSON) | ||
| .header(HttpHeaders.MCP_SESSION_ID, init.session().getId()) | ||
| .bodyValue(initResult)); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Remove the session if initialization fails.
The session enters sessions before initResult() completes. Serialization or initialization errors therefore leave an unreachable session retained indefinitely. Remove it in the error path.
🧰 Tools
🪛 GitHub Check: SonarCloud Code Analysis
[warning] 292-292: Replace "e" with an unnamed pattern.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java`
around lines 283 - 300, Update the initialization flow around
sessionFactory.startSession, sessions.put, and init.initResult() to remove
init.session() from sessions whenever initialization, mapping, serialization, or
response creation fails. Preserve the existing successful response behavior and
ensure cleanup occurs for the session inserted before initResult completes.
| String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID); | ||
|
|
||
| McpStreamableServerSession session = this.sessions.get(sessionId); | ||
|
|
||
| if (session == null) { | ||
| return ServerResponse.notFound().build(); | ||
| } | ||
|
|
||
| return session.delete().then(ServerResponse.ok().build()); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Remove deleted sessions from the registry.
session.delete() terminates the session object, but the map retains it. Future requests still find the deleted session, while notifications and keep-alives continue targeting it.
Atomically remove it before deletion:
Proposed fix
- McpStreamableServerSession session = this.sessions.get(sessionId);
+ McpStreamableServerSession session = this.sessions.remove(sessionId);
if (session == null) {
return ServerResponse.notFound().build();
}
return session.delete().then(ServerResponse.ok().build());📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID); | |
| McpStreamableServerSession session = this.sessions.get(sessionId); | |
| if (session == null) { | |
| return ServerResponse.notFound().build(); | |
| } | |
| return session.delete().then(ServerResponse.ok().build()); | |
| String sessionId = request.headers().asHttpHeaders().getFirst(HttpHeaders.MCP_SESSION_ID); | |
| McpStreamableServerSession session = this.sessions.remove(sessionId); | |
| if (session == null) { | |
| return ServerResponse.notFound().build(); | |
| } | |
| return session.delete().then(ServerResponse.ok().build()); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@api/src/main/java/io/kafbat/ui/transport/WebFluxStreamableServerTransportProvider.java`
around lines 384 - 392, Update the session deletion flow in the request handler
using the sessions registry and McpStreamableServerSession lookup: atomically
remove the session associated with sessionId from this.sessions before invoking
session.delete(). Preserve the existing not-found response when no session is
registered and return the successful response after deletion completes.
Replace legacy SSE transport with Streamable HTTP transport for the MCP server. This upgrades the MCP SDK from 0.10.0 to 1.1.0 and exposes a single
/mcpendpoint instead of separate/mcp/sse+/mcp/messagepaths.The
WebFluxStreamableServerTransportProvideris vendored from Spring AI 2.0.0-M4 with a compatibility fix for Spring Framework 6.2 (replacesHttpHeaders.asMultiValueMap()which requires Spring 7.0).Closes #1137
Summary by CodeRabbit
New Features
Bug Fixes