diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java b/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java index 75a9879..a79c43c 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/McpAsyncServer.java @@ -603,4 +603,41 @@ public class McpAsyncServer { }; } + // --------------------------------------- + // Sampling + // --------------------------------------- + private static TypeReference CREATE_MESSAGE_RESULT_TYPE_REF = new TypeReference<>() { + }; + + /** + * Create a new message using the sampling capabilities of the client. The Model + * Context Protocol (MCP) provides a standardized way for servers to request LLM + * sampling (“completions” or “generations”) from language models via clients. This + * flow allows clients to maintain control over model access, selection, and + * permissions while enabling servers to leverage AI capabilities—with no server API + * keys necessary. Servers can request text or image-based interactions and optionally + * include context from MCP servers in their prompts. + * @param createMessageRequest The request to create a new message + * @return A Mono that completes when the message has been created + * @throws McpError if the client has not been initialized or does not support + * sampling capabilities + * @throws McpError if the client does not support the createMessage method + * @see McpSchema.CreateMessageRequest + * @see McpSchema.CreateMessageResult + * @see Sampling + * Specification + */ + public Mono createMessage(McpSchema.CreateMessageRequest createMessageRequest) { + + if (this.clientCapabilities == null) { + return Mono.error(new McpError("Client must be initialized. Call the initialize method first!")); + } + if (this.clientCapabilities.sampling() == null) { + return Mono.error(new McpError("Client must be configured with sampling capabilities")); + } + return this.mcpSession.sendRequest(McpSchema.METHOD_SAMPLING_CREATE_MESSAGE, createMessageRequest, + CREATE_MESSAGE_RESULT_TYPE_REF); + } + } diff --git a/mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java b/mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java index 4a11943..6096d9e 100644 --- a/mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java +++ b/mcp/src/main/java/org/springframework/ai/mcp/server/McpSyncServer.java @@ -178,4 +178,27 @@ public class McpSyncServer { return this.asyncServer; } + /** + * Create a new message using the sampling capabilities of the client. The Model + * Context Protocol (MCP) provides a standardized way for servers to request LLM + * sampling (“completions” or “generations”) from language models via clients. This + * flow allows clients to maintain control over model access, selection, and + * permissions while enabling servers to leverage AI capabilities—with no server API + * keys necessary. Servers can request text or image-based interactions and optionally + * include context from MCP servers in their prompts. + * @param createMessageRequest The request to create a new message + * @return A Mono that completes when the message has been created + * @throws McpError if the client has not been initialized or does not support + * sampling capabilities + * @throws McpError if the client does not support the createMessage method + * @see McpSchema.CreateMessageRequest + * @see McpSchema.CreateMessageResult + * @see Sampling + * Specification + */ + public McpSchema.CreateMessageResult createMessage(McpSchema.CreateMessageRequest createMessageRequest) { + return this.asyncServer.createMessage(createMessageRequest).block(); + } + } diff --git a/mcp/src/test/java/org/springframework/ai/mcp/MockMcpTransport.java b/mcp/src/test/java/org/springframework/ai/mcp/MockMcpTransport.java new file mode 100644 index 0000000..a41013b --- /dev/null +++ b/mcp/src/test/java/org/springframework/ai/mcp/MockMcpTransport.java @@ -0,0 +1,103 @@ +/* + * Copyright 2024-2024 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.ai.mcp; + +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.core.scheduler.Schedulers; + +import org.springframework.ai.mcp.spec.McpSchema; +import org.springframework.ai.mcp.spec.McpTransport; +import org.springframework.ai.mcp.spec.McpSchema.JSONRPCMessage; +import org.springframework.ai.mcp.spec.McpSchema.JSONRPCNotification; +import org.springframework.ai.mcp.spec.McpSchema.JSONRPCRequest; + +@SuppressWarnings("unused") +public class MockMcpTransport implements McpTransport { + + private final AtomicInteger inboundMessageCount = new AtomicInteger(0); + + private final Sinks.Many outgoing = Sinks.many().multicast().onBackpressureBuffer(); + + private final Sinks.Many inbound = Sinks.many().unicast().onBackpressureBuffer(); + + private final Flux outboundView = outgoing.asFlux().cache(1); + + public void simulateIncomingMessage(McpSchema.JSONRPCMessage message) { + if (inbound.tryEmitNext(message).isFailure()) { + throw new RuntimeException("Failed to emit message " + message); + } + inboundMessageCount.incrementAndGet(); + } + + @Override + public Mono sendMessage(McpSchema.JSONRPCMessage message) { + if (outgoing.tryEmitNext(message).isFailure()) { + return Mono.error(new RuntimeException("Can't emit outgoing message " + message)); + } + return Mono.empty(); + } + + public McpSchema.JSONRPCRequest getLastSentMessageAsRequest() { + return (JSONRPCRequest) outboundView.blockFirst(); + } + + public McpSchema.JSONRPCNotification getLastSentMessageAsNotifiation() { + return (JSONRPCNotification) outboundView.blockFirst(); + } + + public McpSchema.JSONRPCMessage getLastSentMessage() { + return outboundView.blockFirst(); + } + + private volatile boolean connected = false; + + @Override + public Mono connect(Function, Mono> handler) { + if (connected) { + return Mono.error(new IllegalStateException("Already connected")); + } + connected = true; + return inbound.asFlux() + .publishOn(Schedulers.boundedElastic()) + .flatMap(message -> Mono.just(message).transform(handler)) + .doFinally(signal -> connected = false) + .then(); + } + + @Override + public Mono closeGracefully() { + return Mono.defer(() -> { + connected = false; + outgoing.tryEmitComplete(); + inbound.tryEmitComplete(); + return Mono.empty(); + }); + } + + @Override + public T unmarshalFrom(Object data, TypeReference typeRef) { + return new ObjectMapper().convertValue(data, typeRef); + } + +} \ No newline at end of file diff --git a/mcp/src/test/java/org/springframework/ai/mcp/client/McpAsyncClientResponseHandlerTests.java b/mcp/src/test/java/org/springframework/ai/mcp/client/McpAsyncClientResponseHandlerTests.java index 68bbc32..0b2c374 100644 --- a/mcp/src/test/java/org/springframework/ai/mcp/client/McpAsyncClientResponseHandlerTests.java +++ b/mcp/src/test/java/org/springframework/ai/mcp/client/McpAsyncClientResponseHandlerTests.java @@ -20,100 +20,22 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.function.Function; import com.fasterxml.jackson.core.type.TypeReference; -import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Test; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.publisher.Sinks; -import reactor.core.scheduler.Schedulers; +import org.springframework.ai.mcp.MockMcpTransport; import org.springframework.ai.mcp.spec.McpSchema; import org.springframework.ai.mcp.spec.McpSchema.ClientCapabilities; -import org.springframework.ai.mcp.spec.McpSchema.JSONRPCNotification; -import org.springframework.ai.mcp.spec.McpSchema.JSONRPCRequest; import org.springframework.ai.mcp.spec.McpSchema.Root; -import org.springframework.ai.mcp.spec.McpTransport; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; class McpAsyncClientResponseHandlerTests { - @SuppressWarnings("unused") - private static class MockMcpTransport implements McpTransport { - - private final AtomicInteger inboundMessageCount = new AtomicInteger(0); - - private final Sinks.Many outgoing = Sinks.many().multicast().onBackpressureBuffer(); - - private final Sinks.Many inbound = Sinks.many().unicast().onBackpressureBuffer(); - - private final Flux outboundView = outgoing.asFlux().cache(1); - - public void simulateIncomingMessage(McpSchema.JSONRPCMessage message) { - if (inbound.tryEmitNext(message).isFailure()) { - throw new RuntimeException("Failed to emit message " + message); - } - inboundMessageCount.incrementAndGet(); - } - - @Override - public Mono sendMessage(McpSchema.JSONRPCMessage message) { - if (outgoing.tryEmitNext(message).isFailure()) { - return Mono.error(new RuntimeException("Can't emit outgoing message " + message)); - } - return Mono.empty(); - } - - public McpSchema.JSONRPCRequest getLastSentMessageAsRequest() { - return (JSONRPCRequest) outboundView.blockFirst(); - } - - public McpSchema.JSONRPCNotification getLastSentMessageAsNotifiation() { - return (JSONRPCNotification) outboundView.blockFirst(); - } - - public McpSchema.JSONRPCMessage getLastSentMessage() { - return outboundView.blockFirst(); - } - - private volatile boolean connected = false; - - @Override - public Mono connect(Function, Mono> handler) { - if (connected) { - return Mono.error(new IllegalStateException("Already connected")); - } - connected = true; - return inbound.asFlux() - .publishOn(Schedulers.boundedElastic()) - .flatMap(message -> Mono.just(message).transform(handler)) - .doFinally(signal -> connected = false) - .then(); - } - - @Override - public Mono closeGracefully() { - return Mono.defer(() -> { - connected = false; - outgoing.tryEmitComplete(); - inbound.tryEmitComplete(); - return Mono.empty(); - }); - } - - @Override - public T unmarshalFrom(Object data, TypeReference typeRef) { - return new ObjectMapper().convertValue(data, typeRef); - } - - } - @Test void testToolsChangeNotificationHandling() { MockMcpTransport transport = new MockMcpTransport(); diff --git a/mcp/src/test/java/org/springframework/ai/mcp/server/AbstractMcpAsyncServerTests.java b/mcp/src/test/java/org/springframework/ai/mcp/server/AbstractMcpAsyncServerTests.java index d6e7e26..16a8a73 100644 --- a/mcp/src/test/java/org/springframework/ai/mcp/server/AbstractMcpAsyncServerTests.java +++ b/mcp/src/test/java/org/springframework/ai/mcp/server/AbstractMcpAsyncServerTests.java @@ -25,11 +25,11 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import reactor.test.StepVerifier; -import org.springframework.ai.mcp.spec.McpSchema; import org.springframework.ai.mcp.server.McpServer.PromptRegistration; import org.springframework.ai.mcp.server.McpServer.ResourceRegistration; import org.springframework.ai.mcp.server.McpServer.ToolRegistration; import org.springframework.ai.mcp.spec.McpError; +import org.springframework.ai.mcp.spec.McpSchema; import org.springframework.ai.mcp.spec.McpSchema.CallToolResult; import org.springframework.ai.mcp.spec.McpSchema.GetPromptResult; import org.springframework.ai.mcp.spec.McpSchema.Prompt; diff --git a/mcp/src/test/java/org/springframework/ai/mcp/server/SseAsyncIntegrationTests.java b/mcp/src/test/java/org/springframework/ai/mcp/server/SseAsyncIntegrationTests.java new file mode 100644 index 0000000..65c7661 --- /dev/null +++ b/mcp/src/test/java/org/springframework/ai/mcp/server/SseAsyncIntegrationTests.java @@ -0,0 +1,159 @@ +/* +* Copyright 2024 - 2024 the original author or authors. +* +* Licensed under the Apache License, Version 2.0 (the "License"); +* you may not use this file except in compliance with the License. +* You may obtain a copy of the License at +* +* https://www.apache.org/licenses/LICENSE-2.0 +* +* Unless required by applicable law or agreed to in writing, software +* distributed under the License is distributed on an "AS IS" BASIS, +* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +* See the License for the specific language governing permissions and +* limitations under the License. +*/ +package org.springframework.ai.mcp.server; + +import java.util.List; +import java.util.Map; +import java.util.function.Function; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.netty.DisposableServer; +import reactor.netty.http.server.HttpServer; +import reactor.test.StepVerifier; + +import org.springframework.ai.mcp.client.McpClient; +import org.springframework.ai.mcp.client.transport.SseClientTransport; +import org.springframework.ai.mcp.server.transport.SseServerTransport; +import org.springframework.ai.mcp.spec.McpError; +import org.springframework.ai.mcp.spec.McpSchema; +import org.springframework.ai.mcp.spec.McpSchema.ClientCapabilities; +import org.springframework.ai.mcp.spec.McpSchema.CreateMessageRequest; +import org.springframework.ai.mcp.spec.McpSchema.CreateMessageResult; +import org.springframework.ai.mcp.spec.McpSchema.InitializeResult; +import org.springframework.ai.mcp.spec.McpSchema.Role; +import org.springframework.http.server.reactive.HttpHandler; +import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter; +import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.reactive.function.server.RouterFunctions; + +import static org.assertj.core.api.Assertions.assertThat; + +public class SseAsyncIntegrationTests { + + private static final int PORT = 8181; + + private static final String MESSAGE_ENDPOINT = "/mcp/message"; + + private DisposableServer httpServer; + + private SseServerTransport mcpServerTransport; + + McpClient.Builder clientBuilder; + + @BeforeEach + public void before() { + this.mcpServerTransport = new SseServerTransport(new ObjectMapper(), MESSAGE_ENDPOINT); + + HttpHandler httpHandler = RouterFunctions.toHttpHandler(mcpServerTransport.getRouterFunction()); + ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler); + this.httpServer = HttpServer.create().port(PORT).handle(adapter).bindNow(); + + this.clientBuilder = McpClient + .using(new SseClientTransport(WebClient.builder().baseUrl("http://localhost:" + PORT))); + } + + @AfterEach + public void after() { + if (httpServer != null) { + httpServer.disposeNow(); + } + } + + // --------------------------------------- + // Sampling Tests + // --------------------------------------- + @Test + void testCreateMessageWithoutInitialization() { + var mcpAsyncServer = McpServer.using(mcpServerTransport).serverInfo("test-server", "1.0.0").async(); + + var messages = List + .of(new McpSchema.SamplingMessage(McpSchema.Role.USER, new McpSchema.TextContent("Test message"))); + var modelPrefs = new McpSchema.ModelPreferences(List.of(), 1.0, 1.0, 1.0); + + var request = new McpSchema.CreateMessageRequest(messages, modelPrefs, null, + McpSchema.CreateMessageRequest.ContextInclusionStrategy.NONE, null, 100, List.of(), Map.of()); + + StepVerifier.create(mcpAsyncServer.createMessage(request)).verifyErrorSatisfies(error -> { + assertThat(error).isInstanceOf(McpError.class) + .hasMessage("Client must be initialized. Call the initialize method first!"); + }); + } + + @Test + void testCreateMessageWithoutSamplingCapabilities() { + + var mcpAsyncServer = McpServer.using(mcpServerTransport).serverInfo("test-server", "1.0.0").async(); + + var client = clientBuilder.clientInfo(new McpSchema.Implementation("Sample client", "0.0.0")).sync(); + + InitializeResult initResult = client.initialize(); + assertThat(initResult).isNotNull(); + + var messages = List + .of(new McpSchema.SamplingMessage(McpSchema.Role.USER, new McpSchema.TextContent("Test message"))); + var modelPrefs = new McpSchema.ModelPreferences(List.of(), 1.0, 1.0, 1.0); + + var request = new McpSchema.CreateMessageRequest(messages, modelPrefs, null, + McpSchema.CreateMessageRequest.ContextInclusionStrategy.NONE, null, 100, List.of(), Map.of()); + + StepVerifier.create(mcpAsyncServer.createMessage(request)).verifyErrorSatisfies(error -> { + assertThat(error).isInstanceOf(McpError.class) + .hasMessage("Client must be configured with sampling capabilities"); + }); + } + + @Test + void testCreateMessageSuccess() throws InterruptedException { + + var mcpAsyncServer = McpServer.using(mcpServerTransport).serverInfo("test-server", "1.0.0").async(); + + Function samplingHandler = request -> { + assertThat(request.messages()).hasSize(1); + assertThat(request.messages().get(0).content()).isInstanceOf(McpSchema.TextContent.class); + + return new CreateMessageResult(Role.USER, new McpSchema.TextContent("Test message"), "MockModelName", + CreateMessageResult.StopReason.STOP_SEQUENCE); + }; + + var client = clientBuilder.clientInfo(new McpSchema.Implementation("Sample client", "0.0.0")) + .capabilities(ClientCapabilities.builder().sampling().build()) + .sampling(samplingHandler) + .sync(); + + InitializeResult initResult = client.initialize(); + assertThat(initResult).isNotNull(); + + var messages = List + .of(new McpSchema.SamplingMessage(McpSchema.Role.USER, new McpSchema.TextContent("Test message"))); + var modelPrefs = new McpSchema.ModelPreferences(List.of(), 1.0, 1.0, 1.0); + + var request = new McpSchema.CreateMessageRequest(messages, modelPrefs, null, + McpSchema.CreateMessageRequest.ContextInclusionStrategy.NONE, null, 100, List.of(), Map.of()); + + StepVerifier.create(mcpAsyncServer.createMessage(request)).consumeNextWith(result -> { + assertThat(result).isNotNull(); + assertThat(result.role()).isEqualTo(Role.USER); + assertThat(result.content()).isInstanceOf(McpSchema.TextContent.class); + assertThat(((McpSchema.TextContent) result.content()).text()).isEqualTo("Test message"); + assertThat(result.model()).isEqualTo("MockModelName"); + assertThat(result.stopReason()).isEqualTo(CreateMessageResult.StopReason.STOP_SEQUENCE); + }).verifyComplete(); + } + +} diff --git a/mcp/src/test/java/org/springframework/ai/mcp/server/SseMcpAsyncServerTests.java b/mcp/src/test/java/org/springframework/ai/mcp/server/SseMcpAsyncServerTests.java index 3d8336b..45259a8 100644 --- a/mcp/src/test/java/org/springframework/ai/mcp/server/SseMcpAsyncServerTests.java +++ b/mcp/src/test/java/org/springframework/ai/mcp/server/SseMcpAsyncServerTests.java @@ -18,15 +18,14 @@ package org.springframework.ai.mcp.server; import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Timeout; - -import org.springframework.http.server.reactive.HttpHandler; -import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter; -import org.springframework.web.reactive.function.server.RouterFunctions; import reactor.netty.DisposableServer; import reactor.netty.http.server.HttpServer; import org.springframework.ai.mcp.server.transport.SseServerTransport; import org.springframework.ai.mcp.spec.McpTransport; +import org.springframework.http.server.reactive.HttpHandler; +import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter; +import org.springframework.web.reactive.function.server.RouterFunctions; /** * Tests for {@link McpAsyncServer} using {@link SseServerTransport}. diff --git a/mcp/src/test/java/org/springframework/ai/mcp/server/SseMcpSyncServerTests.java b/mcp/src/test/java/org/springframework/ai/mcp/server/SseMcpSyncServerTests.java index c742969..2426849 100644 --- a/mcp/src/test/java/org/springframework/ai/mcp/server/SseMcpSyncServerTests.java +++ b/mcp/src/test/java/org/springframework/ai/mcp/server/SseMcpSyncServerTests.java @@ -18,15 +18,14 @@ package org.springframework.ai.mcp.server; import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Timeout; - -import org.springframework.http.server.reactive.HttpHandler; -import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter; -import org.springframework.web.reactive.function.server.RouterFunctions; import reactor.netty.DisposableServer; import reactor.netty.http.server.HttpServer; import org.springframework.ai.mcp.server.transport.SseServerTransport; import org.springframework.ai.mcp.spec.McpTransport; +import org.springframework.http.server.reactive.HttpHandler; +import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter; +import org.springframework.web.reactive.function.server.RouterFunctions; /** * Tests for {@link McpSyncServer} using {@link SseServerTransport}. diff --git a/mcp/src/test/java/org/springframework/ai/mcp/spec/DefaultMcpSessionTests.java b/mcp/src/test/java/org/springframework/ai/mcp/spec/DefaultMcpSessionTests.java index a29e0d2..658c49a 100644 --- a/mcp/src/test/java/org/springframework/ai/mcp/spec/DefaultMcpSessionTests.java +++ b/mcp/src/test/java/org/springframework/ai/mcp/spec/DefaultMcpSessionTests.java @@ -18,24 +18,18 @@ package org.springframework.ai.mcp.spec; import java.time.Duration; import java.util.Map; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Function; import com.fasterxml.jackson.core.type.TypeReference; -import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; -import reactor.core.scheduler.Schedulers; import reactor.test.StepVerifier; -import org.springframework.ai.mcp.spec.McpSchema.JSONRPCNotification; -import org.springframework.ai.mcp.spec.McpSchema.JSONRPCRequest; +import org.springframework.ai.mcp.MockMcpTransport; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -62,64 +56,6 @@ class DefaultMcpSessionTests { private MockMcpTransport transport; - @SuppressWarnings("unused") - private static class MockMcpTransport implements McpTransport { - - private final AtomicInteger inboundMessageCount = new AtomicInteger(0); - - private Sinks.Many outgoing = Sinks.many().multicast().onBackpressureBuffer(); - - private Sinks.Many inbound = Sinks.many().unicast().onBackpressureBuffer(); - - private Flux outboundView = outgoing.asFlux().cache(1); - - public void simulateIncomingMessage(McpSchema.JSONRPCMessage message) { - if (inbound.tryEmitNext(message).isFailure()) { - throw new RuntimeException("Failed to emit message " + message); - } - inboundMessageCount.incrementAndGet(); - } - - @Override - public Mono sendMessage(McpSchema.JSONRPCMessage message) { - if (outgoing.tryEmitNext(message).isFailure()) { - return Mono.error(new RuntimeException("Can't emit outgoing message " + message)); - } - return Mono.empty(); - } - - public McpSchema.JSONRPCRequest getLastSentMessageAsRequest() { - return (JSONRPCRequest) outboundView.blockFirst(); - } - - public McpSchema.JSONRPCNotification getLastSentMessageAsNotifiation() { - return (JSONRPCNotification) outboundView.blockFirst(); - } - - public McpSchema.JSONRPCMessage getLastSentMessage() { - return outboundView.blockFirst(); - } - - @Override - public Mono connect(Function, Mono> handler) { - return inbound.asFlux() - .publishOn(Schedulers.boundedElastic()) - .flatMap(message -> Mono.just(message).transform(handler)) - .then(); - } - - @Override - public Mono closeGracefully() { - return Mono.empty(); - } - - @Override - public T unmarshalFrom(Object data, TypeReference typeRef) { - return new ObjectMapper().convertValue(data, typeRef); - } - - } - @BeforeEach void setUp() { transport = new MockMcpTransport();