@@ -48,6 +48,15 @@ public class DefaultMcpSession implements McpSession {
4848
4949 private final McpTransport transport ;
5050
51+ private final ConcurrentHashMap <String , RequestHandler > requestHandlers = new ConcurrentHashMap <>();
52+
53+ @ FunctionalInterface
54+ public interface RequestHandler {
55+
56+ Mono <Object > handle (Object params );
57+
58+ }
59+
5160 public DefaultMcpSession (Duration requestTimeout , ObjectMapper objectMapper , McpTransport transport ) {
5261
5362 Assert .notNull (objectMapper , "The ObjectMapper can not be null" );
@@ -70,7 +79,13 @@ public DefaultMcpSession(Duration requestTimeout, ObjectMapper objectMapper, Mcp
7079 }
7180 }
7281 else if (message instanceof McpSchema .JSONRPCRequest request ) {
73- logger .info ("Client does not yet support server requests" );
82+ handleIncomingRequest (request ).subscribe (response -> transport .sendMessage (response ).subscribe (),
83+ error -> {
84+ var errorResponse = new McpSchema .JSONRPCResponse (McpSchema .JSONRPC_VERSION , request .id (),
85+ null , new McpSchema .JSONRPCResponse .JSONRPCError (
86+ McpSchema .ErrorCodes .INTERNAL_ERROR , error .getMessage (), null ));
87+ transport .sendMessage (errorResponse ).subscribe ();
88+ });
7489 }
7590 else if (message instanceof McpSchema .JSONRPCNotification notification ) {
7691 logger .info ("Notifications not yet supported" );
@@ -82,6 +97,27 @@ else if (message instanceof McpSchema.JSONRPCNotification notification) {
8297 this .transport .start ();
8398 }
8499
100+ private Mono <McpSchema .JSONRPCResponse > handleIncomingRequest (McpSchema .JSONRPCRequest request ) {
101+ return Mono .defer (() -> {
102+ var handler = requestHandlers .get (request .method ());
103+ if (handler == null ) {
104+ return Mono .just (new McpSchema .JSONRPCResponse (McpSchema .JSONRPC_VERSION , request .id (), null ,
105+ new McpSchema .JSONRPCResponse .JSONRPCError (McpSchema .ErrorCodes .METHOD_NOT_FOUND ,
106+ "Method not found: " + request .method (), null )));
107+ }
108+
109+ return handler .handle (request .params ())
110+ .map (result -> new McpSchema .JSONRPCResponse (McpSchema .JSONRPC_VERSION , request .id (), result , null ))
111+ .onErrorResume (error -> Mono .just (new McpSchema .JSONRPCResponse (McpSchema .JSONRPC_VERSION , request .id (),
112+ null , new McpSchema .JSONRPCResponse .JSONRPCError (McpSchema .ErrorCodes .INTERNAL_ERROR ,
113+ error .getMessage (), null ))));
114+ });
115+ }
116+
117+ public void registerRequestHandler (String method , RequestHandler handler ) {
118+ requestHandlers .put (method , handler );
119+ }
120+
85121 @ Override
86122 public <T > Mono <T > sendRequest (String method , Object requestParams , TypeReference <T > typeRef ) {
87123 // TODO: UUID API is blocking. Consider non-blocking alternatives to generate
0 commit comments