[WIP] Streamable HTTP Server abstractions and WebFlux transport provider #420
Conversation
Signed-off-by: Dariusz Jędrzejczyk <[email protected]>
…raction Signed-off-by: Dariusz Jędrzejczyk <[email protected]>
Signed-off-by: Dariusz Jędrzejczyk <[email protected]>
Signed-off-by: Dariusz Jędrzejczyk <[email protected]>
Signed-off-by: Dariusz Jędrzejczyk <[email protected]>
Signed-off-by: Dariusz Jędrzejczyk <[email protected]>
Signed-off-by: Dariusz Jędrzejczyk <[email protected]>
|
can we support sse and streamble at the same time ？ |
We could with spec support for something like this, amongst other routes. |
|
Will WebMVC based implementations be provided in the future? My project is based on WebMVC and has been running for some time now. I don't want to modify it to WebFlux anymore. |
|
I agree with @gemo12123 . Plain Java without reactive version should be supported, especially with Project Loom. |
| @@ -70,8 +73,9 @@ public class McpServerSession implements McpSession { | |||
| * @param notificationHandlers map of notification handlers to use | |||
| */ | |||
| public McpServerSession(String id, Duration requestTimeout, McpServerTransport transport, | |||
| InitRequestHandler initHandler, InitNotificationHandler initNotificationHandler, | |||
| Map<String, RequestHandler<?>> requestHandlers, Map<String, NotificationHandler> notificationHandlers) { | |||
| McpInitRequestHandler initHandler, InitNotificationHandler initNotificationHandler, | |||
There was a problem hiding this comment.
InitNotificationHandler -> McpInitNotificationHandler
| @@ -9,6 +9,9 @@ | |||
|
|
|||
| import com.fasterxml.jackson.core.type.TypeReference; | |||
| import io.modelcontextprotocol.server.McpAsyncServerExchange; | |||
| import io.modelcontextprotocol.server.McpInitRequestHandler; | |||
There was a problem hiding this comment.
I believe the McpInitRequestHandler, McpNotificationHandler, McpRequestHandler should be under the spec package?
| /** | ||
| * Request handler for the initialization request. | ||
| */ | ||
| public interface InitRequestHandler { |
There was a problem hiding this comment.
Don't think this is used and should be removed?
| /** | ||
| * Notification handler for the initialization notification from the client. | ||
| */ | ||
| public interface InitNotificationHandler { |
There was a problem hiding this comment.
Don't think this is used and should be removed?
| } | ||
|
|
||
| @FunctionalInterface | ||
| interface RequestHandler<T> extends BiFunction<McpTransportContext, Object, Mono<T>> { |
There was a problem hiding this comment.
- This should be called
StatlessRequestHandler. - Also all request and notification handlers should be grouped in common (parent) package inside the spec.
| this.uriTemplateManagerFactory = uriTemplateManagerFactory; | ||
| this.jsonSchemaValidator = jsonSchemaValidator; | ||
|
|
||
| Map<String, RequestHandler<?>> requestHandlers = new HashMap<>(); |
There was a problem hiding this comment.
I believe we will need notification handling as well. I guess we will need stateless notification handler as well.
| }); | ||
| } | ||
|
|
||
| public Mono<Void> accept(McpSchema.JSONRPCNotification notification) { |
There was a problem hiding this comment.
The accept(McpSchema.JSONRPCNotification notification) and accept(McpSchema.JSONRPCResponse response) methods appear to be a split version of McpServerSession#handle(McpSchema.JSONRPCMessage message). For consistency, I’d prefer we stick with a single handler method.
Additionally, I believe a handle(Message) abstraction should be part of the McpSession interface. Since McpSession represents bidirectional JSON-RPC communication—already exposing sendXyz(Message) for outgoing messages—it makes sense to also include a handle(Message) method for incoming communication.
In fact all McpSession implement some form of handle(Msg). This applies for the clients as well
| requestHandlers.put(McpSchema.METHOD_COMPLETION_COMPLETE, completionCompleteRequestHandler()); | ||
| } | ||
|
|
||
| mcpTransport.setRequestHandler((context, request) -> requestHandlers.get(request.method()) |
There was a problem hiding this comment.
It is a good point that we don't use McpSession as the stateless server is unidirectional and should support only the handle(Message) for request and notifications.
Related to my other comment about the McpSession not having handle(msg). Perhaps we can introduce McpHandlerSession that has handle(JSONRPCMessage) used by the stateless server). Then the McpSession can extend it.
| public interface McpStatelessServerTransport { | ||
|
|
||
| void setRequestHandler( | ||
| BiFunction<McpTransportContext, McpSchema.JSONRPCRequest, Mono<McpSchema.JSONRPCResponse>> message); |
There was a problem hiding this comment.
The signature of the setRequestHandler needs to change from JSONRPCRequest to JSONRPCMessage to allow handling both requests and notifications?
| Mono<McpSchema.InitializeResult> initResult) { | ||
| } | ||
|
|
||
| public final class McpStreamableServerSessionStream implements McpSession { |
There was a problem hiding this comment.
The McpStreamableServerSessionStream isn't bidirectional like McpSession.
We can split this into two interfaces:
McpRpcSender- handles outgoing messages (copies current McpSession functionality)McpRpcReceiver- handles incoming messages withhandle(JSONRPCMessage)
(names be refined)
McpSession would implement both interfaces, McpStreamableServerSessionStream would only implement McpRpcSender, and the Stateless Server would use McpRpcReceiver.
Still WIP, but this outlines the direction for the Streamable HTTP server side. It is based on the research started with cooperation with @ZachGerman in #290. Eventually, Zach's PR should be able to be rebased on top of this work and we will have two transport implementations (WebFlux and JDK Servlet-based).