modelcontextprotocol/java-sdk
Watch
1
0
Fork
You've already forked java-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

break into multiple modules. The mcp module is the MCP implementation with minimal dependecies

Browse source
This commit is contained in:
Christian Tzolov 2024-12-09 18:48:51 +01:00
commit 1982c39bbc
32 changed files with 200 additions and 2858 deletions
Download patch file Download diff file Expand all files Collapse all files

27
mcp/pom.xml
View file

@ -3,13 +3,7 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<!-- <parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>3.4.0</version>
<relativePath />
</parent> -->
<groupId>spring.ai.experiment</groupId>
<groupId>spring.ai.experimental</groupId>
<artifactId>mcp</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>mcp</name>
@ -43,6 +37,25 @@
<artifactId>reactor-core</artifactId>
</dependency>
<dependency>
<groupId>org.assertj</groupId>
<artifactId>assertj-core</artifactId>
<version>3.26.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<version>5.11.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<version>5.11.0</version>
<scope>test</scope>
</dependency>
<!-- <dependency>
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-openai-spring-boot-starter</artifactId>

2
mcp/src/main/java/spring/ai/mcp/client/McpAsyncClient.java
View file

@ -14,7 +14,7 @@ public class McpAsyncClient extends McpAsyncSession {
this(transport, Duration.ofSeconds(10), new ObjectMapper());
}
public McpAsyncClient(McpAsyncTransport transport, Duration requestTimeout, ObjectMapper objectMapper) {
public McpAsyncClient(McpAsyncTransport transport, Duration requestTimeout, ObjectMapper objectMapper) {
super(requestTimeout, objectMapper, transport);
}

37
mcp/src/main/java/spring/ai/mcp/client/McpAsyncSession.java
View file

@ -2,7 +2,6 @@ package spring.ai.mcp.client;
import java.time.Duration;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
@ -10,6 +9,7 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import spring.ai.mcp.client.util.Assert;
import spring.ai.mcp.spec.McpAsyncTransport;
import spring.ai.mcp.spec.McpSchema;
@ -25,9 +25,9 @@ public class McpAsyncSession {
public McpAsyncSession(Duration requestTimeout, ObjectMapper objectMapper, McpAsyncTransport transport) {
Objects.nonNull(requestTimeout);
Objects.nonNull(objectMapper);
Objects.nonNull(transport);
Assert.notNull(objectMapper, "The ObjectMapper can not be null");
Assert.notNull(requestTimeout, "The requstTimeout can not be null");
Assert.notNull(transport, "The transport can not be null");
this.requestTimeout = requestTimeout;
this.objectMapper = objectMapper;
@ -39,8 +39,7 @@ public class McpAsyncSession {
var sink = pendingResponses.remove(response.id());
if (sink == null) {
System.out.println("Unexpected response for unkown id " + response.id());
}
else {
} else {
sink.success(response);
}
}
@ -69,26 +68,23 @@ public class McpAsyncSession {
// TODO: This is non-blocking, but it's actually a synchronous call,
// perhaps there's no need to make it return Mono?
this.transport.sendMessage(jsonrpcRequest)
// TODO: It's most efficient to create a dedicated
// Subscriber here
.subscribe(v -> {
}, e -> {
this.pendingResponses.remove(requestId);
sink.error(e);
});
}
catch (Exception e) {
// TODO: It's most efficient to create a dedicated
// Subscriber here
.subscribe(v -> {
}, e -> {
this.pendingResponses.remove(requestId);
sink.error(e);
});
} catch (Exception e) {
sink.error(e);
}
}).timeout(this.requestTimeout).handle((jsonRpcResponse, s) -> {
if (jsonRpcResponse.error() != null) {
s.error(new McpError(jsonRpcResponse.error()));
}
else {
} else {
if (typeRef.getType().getTypeName().equals("java.lang.Void")) {
s.complete();
}
else {
} else {
s.next(this.objectMapper.convertValue(jsonRpcResponse.result(), typeRef));
}
}
@ -113,8 +109,7 @@ public class McpAsyncSession {
try {
// TODO: make it non-blocking
this.transport.sendMessage(jsonrpcNotification);
}
catch (Exception e) {
} catch (Exception e) {
return Mono.error(new McpError(e));
}
return Mono.empty();

4
mcp/src/main/java/spring/ai/mcp/client/McpSyncClient.java
View file

@ -1,8 +1,8 @@
package spring.ai.mcp.client;
import java.time.Duration;
import java.util.Objects;
import spring.ai.mcp.client.util.Assert;
import spring.ai.mcp.spec.McpSchema;
public class McpSyncClient implements AutoCloseable {
@ -15,7 +15,7 @@ public class McpSyncClient implements AutoCloseable {
McpAsyncClient delegate;
public McpSyncClient(McpAsyncClient delegate) {
Objects.nonNull(delegate);
Assert.notNull(delegate, "The delegate can not be null");
this.delegate = delegate;
}

46
mcp/src/main/java/spring/ai/mcp/client/stdio/StdioServerParameters.java
View file

@ -5,12 +5,10 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.springframework.util.CollectionUtils;
import spring.ai.mcp.client.util.Assert;
/**
* Server parameters for stdio client.
@ -19,11 +17,11 @@ public class StdioServerParameters {
// Environment variables to inherit by default
private static final List<String> DEFAULT_INHERITED_ENV_VARS = System.getProperty("os.name")
.toLowerCase()
.contains("win")
? Arrays.asList("APPDATA", "HOMEDRIVE", "HOMEPATH", "LOCALAPPDATA", "PATH",
"PROCESSOR_ARCHITECTURE", "SYSTEMDRIVE", "SYSTEMROOT", "TEMP", "USERNAME", "USERPROFILE")
: Arrays.asList("HOME", "LOGNAME", "PATH", "SHELL", "TERM", "USER");
.toLowerCase()
.contains("win")
? Arrays.asList("APPDATA", "HOMEDRIVE", "HOMEPATH", "LOCALAPPDATA", "PATH",
"PROCESSOR_ARCHITECTURE", "SYSTEMDRIVE", "SYSTEMROOT", "TEMP", "USERNAME", "USERPROFILE")
: Arrays.asList("HOME", "LOGNAME", "PATH", "SHELL", "TERM", "USER");
@JsonProperty("command")
private String command;
@ -35,13 +33,13 @@ public class StdioServerParameters {
private Map<String, String> env;
private StdioServerParameters(String command, List<String> args, Map<String, String> env) {
Objects.nonNull(command);
Objects.nonNull(args);
Assert.notNull(command, "The command can not be null");
Assert.notNull(args, "The args can not be null");
this.command = command;
this.args = args;
this.env = new HashMap<>(getDefaultEnvironment());
if (!CollectionUtils.isEmpty(env)) {
if (env != null && !env.isEmpty()) {
this.env.putAll(env);
}
}
@ -71,38 +69,38 @@ public class StdioServerParameters {
private Map<String, String> env = new HashMap<>();
public Builder(String command) {
Objects.no.hasText(command, "Command must not be empty");
Assert.notNull(command, "The command can not be null");
this.command = command;
}
public Builder args(String... args) {
Assert.notNull(args, "Arguments must not be null");
Assert.notNull(args, "The args can not be null");
this.args = Arrays.asList(args);
return this;
}
public Builder args(List<String> args) {
Assert.notNull(args, "Arguments must not be null");
Assert.notNull(args, "The args can not be null");
this.args = new ArrayList<>(args);
return this;
}
public Builder arg(String arg) {
Assert.hasText(arg, "Argument must not be empty");
Assert.notNull(arg, "The arg can not be null");
this.args.add(arg);
return this;
}
public Builder env(Map<String, String> env) {
if (!CollectionUtils.isEmpty(env)) {
if (env != null && !env.isEmpty()) {
this.env.putAll(env);
}
return this;
}
public Builder addEnvVar(String key, String value) {
Assert.hasText(key, "Environment variable key must not be empty");
Assert.notNull(value, "Environment variable value must not be null");
Assert.notNull(key, "The key can not be null");
Assert.notNull(value, "The value can not be null");
this.env.put(key, value);
return this;
}
@ -119,12 +117,12 @@ public class StdioServerParameters {
*/
private static Map<String, String> getDefaultEnvironment() {
return System.getenv()
.entrySet()
.stream()
.filter(entry -> DEFAULT_INHERITED_ENV_VARS.contains(entry.getKey()))
.filter(entry -> entry.getValue() != null)
.filter(entry -> !entry.getValue().startsWith("()"))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
.entrySet()
.stream()
.filter(entry -> DEFAULT_INHERITED_ENV_VARS.contains(entry.getKey()))
.filter(entry -> entry.getValue() != null)
.filter(entry -> !entry.getValue().startsWith("()"))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
}

36
mcp/src/main/java/spring/ai/mcp/client/stdio/StdioServerTransport.java
View file

@ -5,20 +5,23 @@ import java.io.BufferedWriter;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.Executors;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import spring.ai.mcp.client.util.Assert;
import spring.ai.mcp.spec.DefaultMcpTransport;
import spring.ai.mcp.spec.McpSchema.JSONRPCMessage;
import spring.ai.mcp.spec.McpSchema.JSONRPCNotification;
import spring.ai.mcp.spec.McpSchema.JSONRPCRequest;
import spring.ai.mcp.spec.McpSchema.JSONRPCResponse;
import org.springframework.ai.model.ModelOptionsUtils;
import org.springframework.util.Assert;
/**
* Stdio client for communicating with a server process.
*/
@ -47,7 +50,7 @@ public class StdioServerTransport extends DefaultMcpTransport {
public StdioServerTransport(StdioServerParameters params, Duration writeTimeout) {
super(writeTimeout);
Assert.notNull(params, "Server parameters must not be null");
Assert.notNull(params, "The params can not be null");
this.params = params;
@ -135,19 +138,34 @@ public class StdioServerTransport extends DefaultMcpTransport {
});
}
private static ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static TypeReference<HashMap<String, Object>> MAP_TYPE_REF = new TypeReference<HashMap<String, Object>>() {
};
public static Map<String, Object> jsonToMap(String json) {
try {
return OBJECT_MAPPER.readValue(json, MAP_TYPE_REF);
}
catch (Exception e) {
throw new RuntimeException(e);
}
}
private JSONRPCMessage deserializeJsonRpcMessage(String jsonText) throws IOException {
var map = ModelOptionsUtils.jsonToMap(jsonText);
var map = jsonToMap(jsonText);
// Determine message type based on specific JSON structure
if (map.containsKey("method") && map.containsKey("id")) {
return ModelOptionsUtils.OBJECT_MAPPER.convertValue(map, JSONRPCRequest.class);
return OBJECT_MAPPER.convertValue(map, JSONRPCRequest.class);
}
else if (map.containsKey("method") && !map.containsKey("id")) {
return ModelOptionsUtils.OBJECT_MAPPER.convertValue(map, JSONRPCNotification.class);
return OBJECT_MAPPER.convertValue(map, JSONRPCNotification.class);
}
else if (map.containsKey("result") || map.containsKey("error")) {
return ModelOptionsUtils.OBJECT_MAPPER.convertValue(map, JSONRPCResponse.class);
return OBJECT_MAPPER.convertValue(map, JSONRPCResponse.class);
}
throw new IllegalArgumentException("Cannot deserialize JSONRPCMessage: " + jsonText);
@ -190,7 +208,7 @@ public class StdioServerTransport extends DefaultMcpTransport {
.handle((message, s) -> {
if (message != null) {
try {
this.processWriter.write(ModelOptionsUtils.toJsonString(message));
this.processWriter.write(OBJECT_MAPPER.writeValueAsString(message));
this.processWriter.newLine();
this.processWriter.flush();
s.next(message);

75
mcp/src/main/java/spring/ai/mcp/client/util/Assert.java Normal file
View file

@ -0,0 +1,75 @@
/*
* 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 spring.ai.mcp.client.util;
import reactor.util.annotation.Nullable;
/**
* @author Christian Tzolov
* @since 1.0.0
*/
public class Assert {
/**
* Assert that an object is not {@code null}.
*
* <pre class="code">
* Assert.notNull(clazz, "The class must not be null");
* </pre>
*
* @param object the object to check
* @param message the exception message to use if the assertion fails
* @throws IllegalArgumentException if the object is {@code null}
*/
public static void notNull(@Nullable Object object, String message) {
if (object == null) {
throw new IllegalArgumentException(message);
}
}
/**
* Assert that the given String contains valid text content; that is, it must not
* be {@code null} and must contain at least one non-whitespace character.
* <pre class="code">Assert.hasText(name, "'name' must not be empty");</pre>
* @param text the String to check
* @param message the exception message to use if the assertion fails
* @throws IllegalArgumentException if the text does not contain valid text content
* @see StringUtils#hasText
*/
public static void hasText(@Nullable String text, String message) {
if (!hasText(text)) {
throw new IllegalArgumentException(message);
}
}
/**
* Check whether the given {@code String} contains actual <em>text</em>.
* <p>More specifically, this method returns {@code true} if the
* {@code String} is not {@code null}, its length is greater than 0,
* and it contains at least one non-whitespace character.
* @param str the {@code String} to check (may be {@code null})
* @return {@code true} if the {@code String} is not {@code null}, its
* length is greater than 0, and it does not contain whitespace only
* @see #hasText(CharSequence)
* @see #hasLength(String)
* @see Character#isWhitespace
*/
public static boolean hasText(@Nullable String str) {
return (str != null && !str.isBlank());
}
}

21
mcp/src/main/java/spring/ai/mcp/spec/DefaultMcpTransport.java
View file

@ -20,6 +20,7 @@ import java.util.function.Consumer;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import spring.ai.mcp.client.util.Assert;
import spring.ai.mcp.spec.McpSchema.JSONRPCMessage;
/**
@ -43,22 +44,22 @@ public class DefaultMcpTransport implements McpAsyncTransport {
public DefaultMcpTransport(Duration readTimeout) {
// TODO: consider the effects of buffering here -> the inter-process pipes are
// independent and the notifications can flood the client/server.
// Potentially, the interest in reading could be communicated from one party
// to the other so the Blocking IO Threads can pause consuming the stream
// buffers when there is no expectation for reading.
// independent and the notifications can flood the client/server.
// Potentially, the interest in reading could be communicated from one party
// to the other so the Blocking IO Threads can pause consuming the stream
// buffers when there is no expectation for reading.
this.errorSink = Sinks.many().unicast().onBackpressureBuffer();
this.inboundSink = Sinks.many().unicast().onBackpressureBuffer();
this.outboundSink = Sinks.many().unicast().onBackpressureBuffer();
this.writeTimeout = readTimeout;
this.writeTimeout = readTimeout;
this.handleIncomingMessages();
}
private void handleIncomingMessages() {
this.inboundSink.asFlux()
.subscribe(message -> this.messageReader.accept(message));
.subscribe(message -> this.messageReader.accept(message));
}
private void handleIncomingErrors() {
@ -104,10 +105,10 @@ public class DefaultMcpTransport implements McpAsyncTransport {
public Mono<Void> sendMessage(JSONRPCMessage message) {
if (this.outboundSink.tryEmitNext(message).isSuccess()) {
// TODO: essentially we could reschedule ourselves in some time and make
// another attempt with the already read data but pause reading until
// success
// In this approach we delegate the retry and the backpressure onto the
// caller. This might be enough for most cases.
// another attempt with the already read data but pause reading until
// success
// In this approach we delegate the retry and the backpressure onto the
// caller. This might be enough for most cases.
return Mono.empty();
} else {
return Mono.error(new RuntimeException("Failed to enqueue message"));

34
mcp/src/main/java/spring/ai/mcp/spec/McpSession.java
View file

@ -17,7 +17,6 @@ package spring.ai.mcp.spec;
import java.time.Duration;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@ -25,6 +24,7 @@ import java.util.concurrent.TimeUnit;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import spring.ai.mcp.client.util.Assert;
import spring.ai.mcp.spec.McpSchema.JSONRPCNotification;
import spring.ai.mcp.spec.McpSchema.JSONRPCRequest;
import spring.ai.mcp.spec.McpSchema.JSONRPCResponse;
@ -41,9 +41,9 @@ public class McpSession implements AutoCloseable {
public McpSession(McpTransport transport, Duration readTimeout, ObjectMapper objectMapper) {
Objects.nonNull(transport);
Objects.nonNull(readTimeout);
Objects.nonNull(objectMapper);
Assert.notNull(transport, "The transport can not be null");
Assert.notNull(readTimeout, "The readTimeout can not be null");
Assert.notNull(objectMapper, "The objectMapper can not be null");
this.waitingForResponseStreams = new ConcurrentHashMap<>();
@ -55,15 +55,13 @@ public class McpSession implements AutoCloseable {
this.transport.setMessageHandler(message -> {
if (message instanceof JSONRPCResponse jsonRpcResponse) {
CompletableFuture<JSONRPCResponse> responseFuture = this.waitingForResponseStreams
.remove(jsonRpcResponse.id());
.remove(jsonRpcResponse.id());
if (responseFuture != null) {
responseFuture.complete(jsonRpcResponse);
}
}
else if (message instanceof JSONRPCRequest jsonRpcRequest) {
} else if (message instanceof JSONRPCRequest jsonRpcRequest) {
handleRequest(new RequestResponder<>(jsonRpcRequest.id(), jsonRpcRequest));
}
else if (message instanceof JSONRPCNotification jsonRpcNotification) {
} else if (message instanceof JSONRPCNotification jsonRpcNotification) {
handleNotification(jsonRpcNotification);
}
});
@ -85,19 +83,18 @@ public class McpSession implements AutoCloseable {
try {
this.transport.sendMessage(jsonrpcRequest);
}
catch (Exception e) {
} catch (Exception e) {
return CompletableFuture.failedFuture(new McpError(e));
}
return responseFuture.orTimeout(this.readTimeout.toMillis(), TimeUnit.MILLISECONDS)
.thenApply(jsonRpcResponse -> {
if (jsonRpcResponse.error() != null) {
throw new McpError(jsonRpcResponse.error());
}
.thenApply(jsonRpcResponse -> {
if (jsonRpcResponse.error() != null) {
throw new McpError(jsonRpcResponse.error());
}
return this.objectMapper.convertValue(jsonRpcResponse.result(), typeRef);
});
return this.objectMapper.convertValue(jsonRpcResponse.result(), typeRef);
});
}
public CompletableFuture<Void> sendNotification(String method) {
@ -108,8 +105,7 @@ public class McpSession implements AutoCloseable {
JSONRPCNotification jsonrpcNotification = new JSONRPCNotification(McpSchema.JSONRPC_VERSION, method, params);
try {
this.transport.sendMessage(jsonrpcNotification);
}
catch (Exception e) {
} catch (Exception e) {
return CompletableFuture.failedFuture(new McpError(e));
}
return CompletableFuture.completedFuture(null);

1
pom.xml
View file

@ -29,6 +29,7 @@
</properties>
<modules>
<module>mcp</module>
<module>spring</module>
</modules>
<repositories>

12
spring/pom.xml
View file

@ -9,10 +9,10 @@
<version>3.4.0</version>
<relativePath /> <!-- lookup parent from repository -->
</parent>
<groupId>spring.ai.experiment</groupId>
<artifactId>mcp</artifactId>
<groupId>spring.ai.experimental</groupId>
<artifactId>mcp-spring-boot-demo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>mcp</name>
<name>mcp-spring-boot-demo</name>
<description>Demo project for Spring Boot</description>
<url />
<licenses>
@ -36,7 +36,11 @@
<groupId>org.springframework.ai</groupId>
<artifactId>spring-ai-openai-spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>spring.ai.experimental</groupId>
<artifactId>mcp</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>

3
spring/src/main/java/spring/ai/mcp/McpApplication.java
View file

@ -31,6 +31,7 @@ public class McpApplication {
.build();
var response = chatClient.prompt("What is the content of the test.txt file?").call().content();
System.out.println(response);
response = chatClient.prompt(
@ -56,7 +57,7 @@ public class McpApplication {
var stdioParams = StdioServerParameters.builder("npx")
.args("-y", "@modelcontextprotocol/server-filesystem",
"dir")
"/Users/christiantzolov/Dev/projects/demo/mcp-server/dir")
.build();
McpSyncClient mcpClient = null;

1
spring/src/main/java/spring/ai/mcp/McpFunctionCallback.java
View file

@ -17,7 +17,6 @@ package spring.ai.mcp;
import java.util.Map;
import spring.ai.mcp.client.McpClient;
import spring.ai.mcp.client.McpSyncClient;
import spring.ai.mcp.spec.McpSchema.CallToolRequest;
import spring.ai.mcp.spec.McpSchema.CallToolResult;

185
spring/src/main/java/spring/ai/mcp/client/McpAsyncClient.java
View file

@ -1,185 +0,0 @@
package spring.ai.mcp.client;
import java.time.Duration;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import reactor.core.publisher.Mono;
import spring.ai.mcp.spec.McpAsyncTransport;
import spring.ai.mcp.spec.McpSchema;
public class McpAsyncClient extends McpAsyncSession {
public McpAsyncClient(McpAsyncTransport transport) {
this(transport, Duration.ofSeconds(10), new ObjectMapper());
}
public McpAsyncClient(McpAsyncTransport transport, Duration requestTimeout, ObjectMapper objectMapper) {
super(requestTimeout, objectMapper, transport);
}
/**
* The initialization phase MUST be the first interaction between client and server.
* During this phase, the client and server:
* <ul>
* <li>Establish protocol version compatibility</li>
* <li>Exchange and negotiate capabilities</li>
* <li>Share implementation details</li>
* </ul>
* <br/>
* The client MUST initiate this phase by sending an initialize request containing:
* <ul>
* <li>The protocol version the client supports</li>
* <li>The client's capabilities</li>
* <li>Client implementation information</li>
* </ul>
*
* The server MUST respond with its own capabilities and information:
* {@link McpSchema.ServerCapabilities}. <br/>
* After successful initialization, the client MUST send an initialized notification
* to indicate it is ready to begin normal operations.
*
* <br/>
*
* <a href=
* "https://github.com/modelcontextprotocol/specification/blob/main/docs/specification/basic/lifecycle.md#initialization">Initialization
* Spec</a>
* @return the initialize result.
*/
public Mono<McpSchema.InitializeResult> initialize() {
McpSchema.InitializeRequest initializeRequest = new McpSchema.InitializeRequest(// @formatter:off
McpSchema.LATEST_PROTOCOL_VERSION,
new McpSchema.ClientCapabilities(null, new McpSchema.ClientCapabilities.RootCapabilities(true), null),
new McpSchema.Implementation("mcp-java-client", "0.0.1")); // @formatter:on
Mono<McpSchema.InitializeResult> result =
this.sendRequest("initialize", initializeRequest, new TypeReference<McpSchema.InitializeResult>() {});
return result.flatMap(initializeResult -> {
if (!McpSchema.LATEST_PROTOCOL_VERSION.equals(initializeResult.protocolVersion())) {
return Mono.error(
new McpError("Unsupported protocol version from the server: "
+ initializeResult.protocolVersion()));
} else {
return this.sendNotification("notifications/initialized", null)
.thenReturn(initializeResult);
}
});
}
/**
* Send a roots/list_changed notification.
*/
public Mono<Void> sendRootsListChanged() {
return this.sendNotification("notifications/roots/list_changed");
}
/**
* Send a synchronous ping request.
*/
public Mono<Void> ping() {
return this.sendRequest("ping", null,
new TypeReference<Void>() {});
}
// --------------------------
// Tools
// --------------------------
/**
* Send a tools/call request.
* @param callToolRequest the call tool request.
* @return the call tool result.
*/
public Mono<McpSchema.CallToolResult> callTool(McpSchema.CallToolRequest callToolRequest) {
return this.sendRequest("tools/call", callToolRequest,
new TypeReference<McpSchema.CallToolResult>() {});
}
/**
* Send a tools/list request.
* @return the list of tools result.
*/
public Mono<McpSchema.ListToolsResult> listTools(String cursor) {
return this.sendRequest("tools/list", new McpSchema.PaginatedRequest(cursor),
new TypeReference<McpSchema.ListToolsResult>() {});
}
// --------------------------
// Resources
// --------------------------
/**
* Send a resources/list request.
* @param cursor the cursor
* @return the list of resources result.
*/
public Mono<McpSchema.ListResourcesResult> listResources(String cursor) {
return this.sendRequest("resources/list", new McpSchema.PaginatedRequest(cursor),
new TypeReference<McpSchema.ListResourcesResult>() {});
}
/**
* Send a resources/read request.
* @param resource the resource to read
* @return the resource content.
*/
public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.Resource resource) {
return this.readResource(new McpSchema.ReadResourceRequest(resource.uri()));
}
/**
* Send a resources/read request.
* @param readResourceRequest the read resource request.
* @return the resource content.
*/
public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.ReadResourceRequest readResourceRequest) {
return this.sendRequest("resources/read", readResourceRequest,
new TypeReference<McpSchema.ReadResourceResult>() {});
}
/**
* Resource templates allow servers to expose parameterized resources using URI
* templates. Arguments may be auto-completed through the completion API.
*
* Request a list of resource templates the server has.
* @param cursor the cursor
* @return the list of resource templates result.
*/
public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates(String cursor) {
return this.sendRequest("resources/templates/list", new McpSchema.PaginatedRequest(cursor),
new TypeReference<McpSchema.ListResourceTemplatesResult>() {});
}
/**
* List Changed Notification. When the list of available resources changes, servers
* that declared the listChanged capability SHOULD send a notification:
*/
public Mono<Void> sendResourcesListChanged() {
return this.sendNotification("notifications/resources/list_changed");
}
/**
* Subscriptions. The protocol supports optional subscriptions to resource changes.
* Clients can subscribe to specific resources and receive notifications when they
* change.
*
* Send a resources/subscribe request.
* @param subscribeRequest the subscribe request contains the uri of the resource to
* subscribe to.
*/
public Mono<Void> subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {
return this.sendRequest("resources/subscribe", subscribeRequest,
new TypeReference<Void>() {});
}
/**
* Send a resources/unsubscribe request.
* @param unsubscribeRequest the unsubscribe request contains the uri of the resource
* to unsubscribe from.
*/
public Mono<Void> unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {
return this.sendRequest("resources/unsubscribe", unsubscribeRequest,
new TypeReference<Void>() {});
}
}

131
spring/src/main/java/spring/ai/mcp/client/McpAsyncSession.java
View file

@ -1,131 +0,0 @@
package spring.ai.mcp.client;
import java.time.Duration;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import spring.ai.mcp.spec.McpAsyncTransport;
import spring.ai.mcp.spec.McpTransport;
import org.springframework.util.Assert;
import spring.ai.mcp.spec.McpSchema;
public class McpAsyncSession {
private final ConcurrentHashMap<Object, MonoSink<McpSchema.JSONRPCResponse>> pendingResponses = new ConcurrentHashMap<>();
private final Duration requestTimeout;
private final ObjectMapper objectMapper;
private final McpAsyncTransport transport;
public McpAsyncSession(Duration requestTimeout, ObjectMapper objectMapper, McpAsyncTransport transport) {
Assert.notNull(requestTimeout, "Request timeout must not be null");
Assert.notNull(objectMapper, "ObjectMapper must not be null");
Assert.notNull(transport, "Transport must not be null");
this.requestTimeout = requestTimeout;
this.objectMapper = objectMapper;
this.transport = transport;
this.transport.setMessageHandler(message -> {
switch (message) {
case McpSchema.JSONRPCResponse response -> {
var sink = pendingResponses.remove(response.id());
if (sink == null) {
System.out.println("Unexpected response for unkown id " + response.id());
}
else {
sink.success(response);
}
}
case McpSchema.JSONRPCRequest request -> {
System.out.println("Client does not yet support server requests");
}
case McpSchema.JSONRPCNotification notification -> {
System.out.println("Notifications not yet supported");
}
}
});
this.transport.setErrorHandler(error -> System.out.println("Error received " + error));
this.transport.start();
}
public <T> Mono<T> sendRequest(String method, Object requestParams, TypeReference<T> typeRef) {
String requestId = UUID.randomUUID().toString();
return Mono.<McpSchema.JSONRPCResponse>create(sink -> {
this.pendingResponses.put(requestId, sink);
McpSchema.JSONRPCRequest jsonrpcRequest = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, method,
requestId, requestParams);
try {
// TODO: This is non-blocking, but it's actually a synchronous call,
// perhaps there's no need to make it return Mono?
this.transport.sendMessage(jsonrpcRequest)
// TODO: It's most efficient to create a dedicated
// Subscriber here
.subscribe(v -> {
}, e -> {
this.pendingResponses.remove(requestId);
sink.error(e);
});
}
catch (Exception e) {
sink.error(e);
}
}).timeout(this.requestTimeout).handle((jsonRpcResponse, s) -> {
if (jsonRpcResponse.error() != null) {
s.error(new McpError(jsonRpcResponse.error()));
}
else {
if (typeRef.getType().getTypeName().equals("java.lang.Void")) {
s.complete();
}
else {
s.next(this.objectMapper.convertValue(jsonRpcResponse.result(), typeRef));
}
}
});
}
public static class McpError extends RuntimeException {
public McpError(Object error) {
super(error.toString());
}
}
public Mono<Void> sendNotification(String method) {
return sendNotification(method, null);
}
public Mono<Void> sendNotification(String method, Map<String, Object> params) {
McpSchema.JSONRPCNotification jsonrpcNotification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION,
method, params);
try {
// TODO: make it non-blocking
this.transport.sendMessage(jsonrpcNotification);
}
catch (Exception e) {
return Mono.error(new McpError(e));
}
return Mono.empty();
}
public Mono<Void> closeGracefully(Duration timeout) {
// TODO handle the timeout in transport
return Mono.fromRunnable(this.transport::close);
}
}

83
spring/src/main/java/spring/ai/mcp/client/McpClient.java
View file

@ -1,83 +0,0 @@
/*
* 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 spring.ai.mcp.client;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import spring.ai.mcp.spec.McpAsyncTransport;
import spring.ai.mcp.spec.McpTransport;
import spring.ai.mcp.spec.McpSchema;
import spring.ai.mcp.spec.McpSession;
import spring.ai.mcp.spec.McpTransport;
import spring.ai.mcp.spec.McpSchema.CallToolRequest;
import spring.ai.mcp.spec.McpSchema.CallToolResult;
import spring.ai.mcp.spec.McpSchema.ClientCapabilities;
import spring.ai.mcp.spec.McpSchema.Implementation;
import spring.ai.mcp.spec.McpSchema.InitializeRequest;
import spring.ai.mcp.spec.McpSchema.InitializeResult;
import spring.ai.mcp.spec.McpSchema.ListResourceTemplatesResult;
import spring.ai.mcp.spec.McpSchema.ListResourcesResult;
import spring.ai.mcp.spec.McpSchema.ListToolsResult;
import spring.ai.mcp.spec.McpSchema.PaginatedRequest;
import spring.ai.mcp.spec.McpSchema.ReadResourceRequest;
import spring.ai.mcp.spec.McpSchema.Resource;
import spring.ai.mcp.spec.McpSchema.SubscribeRequest;
import spring.ai.mcp.spec.McpSchema.UnsubscribeRequest;
import spring.ai.mcp.spec.McpSchema.ClientCapabilities.RootCapabilities;
/**
* The MCP client is the main entry point for interacting with the Model Context Protocol
* (MCP) server.
*
*
* @author Christian Tzolov
* @since 1.0.0
*/
public class McpClient {
private McpClient() {
}
// TODO: introduce a builder like:
// McpClient.using(transport)
// .withRequestTimeout(Duration.ofSeconds(5))
// .withObjectMapper(objectMapper); <- or even a more sophisticated
// JSONtoPOJOCodec type
// .sync();
public static McpAsyncClient async(McpAsyncTransport transport) {
return new McpAsyncClient(transport);
}
public static McpAsyncClient async(McpAsyncTransport transport, Duration requestTimeout,
ObjectMapper objectMapper) {
return new McpAsyncClient(transport, requestTimeout, objectMapper);
}
public static McpSyncClient sync(McpAsyncTransport transport) {
return new McpSyncClient(async(transport));
}
public static McpSyncClient sync(McpAsyncTransport transport, Duration requestTimeout,
ObjectMapper objectMapper) {
return new McpSyncClient(async(transport, requestTimeout, objectMapper));
}
}

180
spring/src/main/java/spring/ai/mcp/client/McpSyncClient.java
View file

@ -1,180 +0,0 @@
package spring.ai.mcp.client;
import java.time.Duration;
import spring.ai.mcp.spec.McpSchema;
import org.springframework.util.Assert;
public class McpSyncClient implements AutoCloseable {
// TODO: Consider providing a client config to set this properly
// this is currently a concern only because AutoCloseable is used - perhaps it is
// not a requirement?
private static final long DEFAULT_CLOSE_TIMEOUT_MS = 10_000L;
McpAsyncClient delegate;
public McpSyncClient(McpAsyncClient delegate) {
Assert.notNull(delegate, "McpAsyncClient must not be null");
this.delegate = delegate;
}
@Override
public void close() {
delegate.closeGracefully(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS)).block();
}
/**
* The initialization phase MUST be the first interaction between client and server.
* During this phase, the client and server:
* <ul>
* <li>Establish protocol version compatibility</li>
* <li>Exchange and negotiate capabilities</li>
* <li>Share implementation details</li>
* </ul>
* <br/>
* The client MUST initiate this phase by sending an initialize request containing:
* <ul>
* <li>The protocol version the client supports</li>
* <li>The client's capabilities</li>
* <li>Client implementation information</li>
* </ul>
*
* The server MUST respond with its own capabilities and information:
* {@link McpSchema.ServerCapabilities}. <br/>
* After successful initialization, the client MUST send an initialized notification
* to indicate it is ready to begin normal operations.
*
* <br/>
*
* <a href=
* "https://github.com/modelcontextprotocol/specification/blob/main/docs/specification/basic/lifecycle.md#initialization">Initialization
* Spec</a>
* @return the initialize result.
*/
public McpSchema.InitializeResult initialize() {
// TODO: block takes no argument here as we assume the async client is
// configured with a requestTimeout at all times
return delegate.initialize()
.block();
}
/**
* Send a roots/list_changed notification.
*/
public void sendRootsListChanged() {
delegate.sendRootsListChanged()
.block();
}
/**
* Send a synchronous ping request.
*/
public void ping() {
delegate.ping()
.block();
}
// --------------------------
// Tools
// --------------------------
/**
* Send a tools/call request.
* @param callToolRequest the call tool request.
* @return the call tool result.
*/
public McpSchema.CallToolResult callTool(McpSchema.CallToolRequest callToolRequest) {
return delegate.callTool(callToolRequest)
.block();
}
/**
* Send a tools/list request.
* @return the list of tools result.
*/
public McpSchema.ListToolsResult listTools(String cursor) {
return delegate.listTools(cursor)
.block();
}
// --------------------------
// Resources
// --------------------------
/**
* Send a resources/list request.
* @param cursor the cursor
* @return the list of resources result.
*/
public McpSchema.ListResourcesResult listResources(String cursor) {
return delegate.listResources(cursor)
.block();
}
/**
* Send a resources/read request.
* @param resource the resource to read
* @return the resource content.
*/
public McpSchema.ReadResourceResult readResource(McpSchema.Resource resource) {
return delegate.readResource(resource)
.block();
}
/**
* Send a resources/read request.
* @param readResourceRequest the read resource request.
* @return the resource content.
*/
public McpSchema.ReadResourceResult readResource(McpSchema.ReadResourceRequest readResourceRequest) {
return delegate.readResource(readResourceRequest)
.block();
}
/**
* Resource templates allow servers to expose parameterized resources using URI
* templates. Arguments may be auto-completed through the completion API.
*
* Request a list of resource templates the server has.
* @param cursor the cursor
* @return the list of resource templates result.
*/
public McpSchema.ListResourceTemplatesResult listResourceTemplates(String cursor) {
return delegate.listResourceTemplates(cursor)
.block();
}
/**
* List Changed Notification. When the list of available resources changes, servers
* that declared the listChanged capability SHOULD send a notification:
*/
public void sendResourcesListChanged() {
delegate.sendResourcesListChanged()
.block();
}
/**
* Subscriptions. The protocol supports optional subscriptions to resource changes.
* Clients can subscribe to specific resources and receive notifications when they
* change.
*
* Send a resources/subscribe request.
* @param subscribeRequest the subscribe request contains the uri of the resource to
* subscribe to.
*/
public void subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {
delegate.subscribeResource(subscribeRequest)
.block();
}
/**
* Send a resources/unsubscribe request.
* @param unsubscribeRequest the unsubscribe request contains the uri of the resource
* to unsubscribe from.
*/
public void unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {
delegate.unsubscribeResource(unsubscribeRequest)
.block();
}
}

130
spring/src/main/java/spring/ai/mcp/client/stdio/StdioServerParameters.java
View file

@ -1,130 +0,0 @@
package spring.ai.mcp.client.stdio;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* Server parameters for stdio client.
*/
public class StdioServerParameters {
// Environment variables to inherit by default
private static final List<String> DEFAULT_INHERITED_ENV_VARS = System.getProperty("os.name")
.toLowerCase()
.contains("win")
? Arrays.asList("APPDATA", "HOMEDRIVE", "HOMEPATH", "LOCALAPPDATA", "PATH",
"PROCESSOR_ARCHITECTURE", "SYSTEMDRIVE", "SYSTEMROOT", "TEMP", "USERNAME", "USERPROFILE")
: Arrays.asList("HOME", "LOGNAME", "PATH", "SHELL", "TERM", "USER");
@JsonProperty("command")
private String command;
@JsonProperty("args")
private List<String> args = new ArrayList<>();
@JsonProperty("env")
private Map<String, String> env;
private StdioServerParameters(String command, List<String> args, Map<String, String> env) {
Assert.hasText(command, "Command must not be empty");
Assert.notNull(args, "Arguments must not be null");
this.command = command;
this.args = args;
this.env = new HashMap<>(getDefaultEnvironment());
if (!CollectionUtils.isEmpty(env)) {
this.env.putAll(env);
}
}
public String getCommand() {
return this.command;
}
public List<String> getArgs() {
return this.args;
}
public Map<String, String> getEnv() {
return this.env;
}
public static Builder builder(String command) {
return new Builder(command);
}
public static class Builder {
private String command;
private List<String> args = new ArrayList<>();
private Map<String, String> env = new HashMap<>();
public Builder(String command) {
Assert.hasText(command, "Command must not be empty");
this.command = command;
}
public Builder args(String... args) {
Assert.notNull(args, "Arguments must not be null");
this.args = Arrays.asList(args);
return this;
}
public Builder args(List<String> args) {
Assert.notNull(args, "Arguments must not be null");
this.args = new ArrayList<>(args);
return this;
}
public Builder arg(String arg) {
Assert.hasText(arg, "Argument must not be empty");
this.args.add(arg);
return this;
}
public Builder env(Map<String, String> env) {
if (!CollectionUtils.isEmpty(env)) {
this.env.putAll(env);
}
return this;
}
public Builder addEnvVar(String key, String value) {
Assert.hasText(key, "Environment variable key must not be empty");
Assert.notNull(value, "Environment variable value must not be null");
this.env.put(key, value);
return this;
}
public StdioServerParameters build() {
return new StdioServerParameters(command, args, env);
}
}
/**
* Returns a default environment object including only environment variables
* deemed safe to inherit.
*/
private static Map<String, String> getDefaultEnvironment() {
return System.getenv()
.entrySet()
.stream()
.filter(entry -> DEFAULT_INHERITED_ENV_VARS.contains(entry.getKey()))
.filter(entry -> entry.getValue() != null)
.filter(entry -> !entry.getValue().startsWith("()"))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
}

243
spring/src/main/java/spring/ai/mcp/client/stdio/StdioServerTransport.java
View file

@ -1,243 +0,0 @@
package spring.ai.mcp.client.stdio;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import spring.ai.mcp.spec.DefaultMcpTransport;
import spring.ai.mcp.spec.McpSchema.JSONRPCMessage;
import spring.ai.mcp.spec.McpSchema.JSONRPCNotification;
import spring.ai.mcp.spec.McpSchema.JSONRPCRequest;
import spring.ai.mcp.spec.McpSchema.JSONRPCResponse;
import org.springframework.ai.model.ModelOptionsUtils;
import org.springframework.util.Assert;
/**
* Stdio client for communicating with a server process.
*/
public class StdioServerTransport extends DefaultMcpTransport {
private Process process;
private BufferedReader processErrorReader;
private BufferedReader processReader;
private BufferedWriter processWriter;
private Scheduler inboundScheduler;
private Scheduler outboundScheduler;
private Scheduler errorScheduler;
private volatile boolean isRunning;
private final StdioServerParameters params;
public StdioServerTransport(StdioServerParameters params) {
this(params, Duration.ofMillis(100));
}
public StdioServerTransport(StdioServerParameters params, Duration writeTimeout) {
super(writeTimeout);
Assert.notNull(params, "Server parameters must not be null");
this.params = params;
// Start threads
this.inboundScheduler =
Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(),
"inbound");
this.outboundScheduler =
Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(),
"outbound");
this.errorScheduler =
Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(),
"error");
}
@Override
public void start() {
// Prepare command and environment
List<String> fullCommand = new ArrayList<>();
fullCommand.add(params.getCommand());
fullCommand.addAll(params.getArgs());
ProcessBuilder processBuilder = new ProcessBuilder(fullCommand);
processBuilder.environment().putAll(params.getEnv());
// Start the process
try {
this.process = processBuilder.start();
}
catch (IOException e) {
throw new RuntimeException("Failed to start process with command: " + fullCommand, e);
}
// Validate process streams
if (this.process.getInputStream() == null || process.getOutputStream() == null) {
this.process.destroy();
throw new RuntimeException("Process input or output stream is null");
}
// Initialize readers and writers
this.processErrorReader = this.process.errorReader();
this.processReader = this.process.inputReader();
this.processWriter = this.process.outputWriter();
// Start threads
this.isRunning = true;
startInboundProcessing();
startOutboundProcessing();
startErrorProcessing();
}
public void awaitForExit() {
try {
this.process.waitFor();
}
catch (InterruptedException e) {
throw new RuntimeException("Process interrupted", e);
}
}
private void startErrorProcessing() {
this.errorScheduler.schedule(() -> {
try {
String line;
while (isRunning && processErrorReader != null && (line = processErrorReader.readLine()) != null) {
try {
System.out.println("Received error line: " + line);
// TODO: handle errors, etc.
this.getErrorSink().tryEmitNext(line);
}
catch (Exception e) {
throw new RuntimeException(e);
}
}
}
catch (IOException e) {
if (this.isRunning) {
throw new RuntimeException(e);
}
}
finally {
this.isRunning = false;
}
});
}
private JSONRPCMessage deserializeJsonRpcMessage(String jsonText) throws IOException {
var map = ModelOptionsUtils.jsonToMap(jsonText);
// Determine message type based on specific JSON structure
if (map.containsKey("method") && map.containsKey("id")) {
return ModelOptionsUtils.OBJECT_MAPPER.convertValue(map, JSONRPCRequest.class);
}
else if (map.containsKey("method") && !map.containsKey("id")) {
return ModelOptionsUtils.OBJECT_MAPPER.convertValue(map, JSONRPCNotification.class);
}
else if (map.containsKey("result") || map.containsKey("error")) {
return ModelOptionsUtils.OBJECT_MAPPER.convertValue(map, JSONRPCResponse.class);
}
throw new IllegalArgumentException("Cannot deserialize JSONRPCMessage: " + jsonText);
}
private void startInboundProcessing() {
this.inboundScheduler.schedule(() -> {
try {
String line;
while (this.isRunning && this.processReader != null && (line = this.processReader.readLine()) != null) {
try {
JSONRPCMessage message = deserializeJsonRpcMessage(line);
if (!this.getInboundSink().tryEmitNext(message).isSuccess()) {
// TODO: Back off, reschedule, give up?
throw new RuntimeException("Failed to enqueue message");
}
}
catch (Exception e) {
throw new RuntimeException(e);
}
}
}
catch (IOException e) {
if (isRunning) {
throw new RuntimeException(e);
}
}
finally {
isRunning = false;
}
});
}
private void startOutboundProcessing() {
this.getOutboundSink()
.asFlux()
// this bit is important since writes come from user threads and we
// want to ensure that the actual writing happens on a dedicated thread
.publishOn(outboundScheduler)
.handle((message, s) -> {
if (message != null) {
try {
this.processWriter.write(ModelOptionsUtils.toJsonString(message));
this.processWriter.newLine();
this.processWriter.flush();
s.next(message);
}
catch (IOException e) {
s.error(new RuntimeException(e));
}
}
})
.subscribe();
}
// TODO: provide a non-blocking variant with graceful option
public void stop() {
this.outboundScheduler.dispose();
this.inboundScheduler.dispose();
this.errorScheduler.dispose();
// Destroy process
if (this.process != null) {
this.process.destroyForcibly();
}
}
@Override
public void close() {
stop();
super.close(); // Do we need this?
// Close resources
if (this.processReader != null) {
try {
this.processReader.close();
}
catch (IOException e) {
e.printStackTrace();
}
}
if (this.processWriter != null) {
try {
this.processWriter.close();
}
catch (IOException e) {
e.printStackTrace();
}
}
}
}

87
spring/src/main/java/spring/ai/mcp/client/util/McpServerParameterParser.java
View file

@ -1,87 +0,0 @@
/*
* 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 spring.ai.mcp.client.util;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import spring.ai.mcp.client.util.McpServerParameterParser.McpServers.McpServer;
/**
* @author Christian Tzolov
* @since 1.0.0
*/
public class McpServerParameterParser {
@JsonInclude(JsonInclude.Include.NON_ABSENT)
public record McpServers(// @formatter:off
@JsonProperty("mcpServers") Map<String, McpServer> mcpServers) {
@JsonInclude(JsonInclude.Include.NON_ABSENT)
public record McpServer(
@JsonProperty("command") String command,
@JsonProperty("args") String[] args,
@JsonProperty("env") Map<String, String> env) {
}
} // @formatter:on
public static void main(String[] args) throws Exception {
ObjectMapper objectMapper = new ObjectMapper();
// Assuming the JSON is in a file named "servers.json"
String jsonInput = """
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed/files"]
},
"git": {
"command": "uvx",
"args": ["mcp-server-git", "--repository", "path/to/git/repo"]
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"env": {
"GITHUB_PERSONAL_ACCESS_TOKEN": "<YOUR_TOKEN>"
}
},
"postgres": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://localhost/mydb"]
}
}
}
""";
// Deserialize JSON into McpServers
McpServers servers = objectMapper.readValue(jsonInput, McpServers.class);
// Access individual servers
for (Map.Entry<String, McpServer> entry : servers.mcpServers().entrySet()) {
System.out.println("Key: " + entry.getKey());
System.out.println("Command: " + entry.getValue().command());
System.out.println("Arguments: " + String.join(", ", entry.getValue().args()));
System.out.println("Environment Variables: " + entry.getValue().env());
System.out.println("-----------");
}
}
}

Some files were not shown because too many files have changed in this diff Show more