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

refactor: introduce session-based architecture for MCP server (#31)

Browse source 
This commit introduces a major refactoring of the MCP Java SDK to implement a session-based architecture
for server-side implementations. The changes improve the SDK's ability to handle multiple concurrent
client connections and provide an API better aligned with the MCP specification.

Key changes:

- Introduce McpServerTransportProvider interface to manage client connections
- Rename ClientMcpTransport to McpClientTransport and ServerMcpTransport to McpServerTransport
- Add exchange objects (McpAsyncServerExchange, McpSyncServerExchange) for client interaction
- Update handler signatures to include exchange parameter: (args) -> result to (exchange, args) -> result
- Rename Registration classes to Specification classes
- Update method names (e.g., rootsChangeConsumers to rootsChangeHandlers)
- Deprecate old interfaces and classes for removal in 0.9.0
- Add migration guide (migration-0.8.0.md)

Resolves #9, #15

Co-authored-by: Christian Tzolov <christian.tzolov@broadcom.com>
Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
This commit is contained in:
Dariusz Jędrzejczyk 2025-03-20 18:14:57 +01:00 • committed by Christian Tzolov
commit 34a733509d
25 changed files with 9969 additions and 1127 deletions
Download patch file Download diff file Expand all files Collapse all files

12
mcp/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java
View file

@ -15,9 +15,9 @@ import java.util.function.Function;
import com.fasterxml.jackson.core.type.TypeReference;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.DefaultMcpSession;
import io.modelcontextprotocol.spec.DefaultMcpSession.NotificationHandler;
import io.modelcontextprotocol.spec.DefaultMcpSession.RequestHandler;
import io.modelcontextprotocol.spec.McpClientSession;
import io.modelcontextprotocol.spec.McpClientSession.NotificationHandler;
import io.modelcontextprotocol.spec.McpClientSession.RequestHandler;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
@ -73,7 +73,7 @@ import reactor.core.publisher.Sinks;
* @author Christian Tzolov
* @see McpClient
* @see McpSchema
* @see DefaultMcpSession
* @see McpClientSession
*/
public class McpAsyncClient {
@ -95,7 +95,7 @@ public class McpAsyncClient {
* The MCP session implementation that manages bidirectional JSON-RPC communication
* between clients and servers.
*/
private final DefaultMcpSession mcpSession;
private final McpClientSession mcpSession;
/**
* Client capabilities.
@ -228,7 +228,7 @@ public class McpAsyncClient {
notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_MESSAGE,
asyncLoggingNotificationHandler(loggingConsumersFinal));
this.mcpSession = new DefaultMcpSession(requestTimeout, transport, requestHandlers, notificationHandlers);
this.mcpSession = new McpClientSession(requestTimeout, transport, requestHandlers, notificationHandlers);
}

41
mcp/src/main/java/io/modelcontextprotocol/client/McpClient.java
View file

@ -13,6 +13,7 @@ import java.util.function.Consumer;
import java.util.function.Function;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpTransport;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
@ -113,11 +114,31 @@ public interface McpClient {
* and {@code SseClientTransport} for SSE-based communication.
* @return A new builder instance for configuring the client
* @throws IllegalArgumentException if transport is null
* @deprecated This method will be removed in 0.9.0. Use
* {@link #sync(McpClientTransport)}
*/
@Deprecated
static SyncSpec sync(ClientMcpTransport transport) {
return new SyncSpec(transport);
}
/**
* Start building a synchronous MCP client with the specified transport layer. The
* synchronous MCP client provides blocking operations. Synchronous clients wait for
* each operation to complete before returning, making them simpler to use but
* potentially less performant for concurrent operations. The transport layer handles
* the low-level communication between client and server using protocols like stdio or
* Server-Sent Events (SSE).
* @param transport The transport layer implementation for MCP communication. Common
* implementations include {@code StdioClientTransport} for stdio-based communication
* and {@code SseClientTransport} for SSE-based communication.
* @return A new builder instance for configuring the client
* @throws IllegalArgumentException if transport is null
*/
static SyncSpec sync(McpClientTransport transport) {
return new SyncSpec(transport);
}
/**
* Start building an asynchronous MCP client with the specified transport layer. The
* asynchronous MCP client provides non-blocking operations. Asynchronous clients
@ -130,11 +151,31 @@ public interface McpClient {
* and {@code SseClientTransport} for SSE-based communication.
* @return A new builder instance for configuring the client
* @throws IllegalArgumentException if transport is null
* @deprecated This method will be removed in 0.9.0. Use
* {@link #async(McpClientTransport)}
*/
@Deprecated
static AsyncSpec async(ClientMcpTransport transport) {
return new AsyncSpec(transport);
}
/**
* Start building an asynchronous MCP client with the specified transport layer. The
* asynchronous MCP client provides non-blocking operations. Asynchronous clients
* return reactive primitives (Mono/Flux) immediately, allowing for concurrent
* operations and reactive programming patterns. The transport layer handles the
* low-level communication between client and server using protocols like stdio or
* Server-Sent Events (SSE).
* @param transport The transport layer implementation for MCP communication. Common
* implementations include {@code StdioClientTransport} for stdio-based communication
* and {@code SseClientTransport} for SSE-based communication.
* @return A new builder instance for configuring the client
* @throws IllegalArgumentException if transport is null
*/
static AsyncSpec async(McpClientTransport transport) {
return new AsyncSpec(transport);
}
/**
* Synchronous client specification. This class follows the builder pattern to provide
* a fluent API for setting up clients with custom configurations.

5
mcp/src/main/java/io/modelcontextprotocol/client/McpSyncClient.java
View file

@ -6,7 +6,7 @@ package io.modelcontextprotocol.client;
import java.time.Duration;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.ClientCapabilities;
import io.modelcontextprotocol.spec.McpSchema.GetPromptRequest;
@ -66,7 +66,8 @@ public class McpSyncClient implements AutoCloseable {
* Create a new McpSyncClient with the given delegate.
* @param delegate the asynchronous kernel on top of which this synchronous client
* provides a blocking API.
* @deprecated Use {@link McpClient#sync(ClientMcpTransport)} to obtain an instance.
* @deprecated This method will be removed in 0.9.0. Use
* {@link McpClient#sync(McpClientTransport)} to obtain an instance.
*/
@Deprecated
// TODO make the constructor package private post-deprecation

28
mcp/src/main/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransport.java
View file

@ -3,18 +3,6 @@
*/
package io.modelcontextprotocol.client.transport;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.client.transport.FlowSseClient.SseEvent;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage;
import io.modelcontextprotocol.util.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
@ -27,6 +15,18 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.client.transport.FlowSseClient.SseEvent;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage;
import io.modelcontextprotocol.util.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
/**
* Server-Sent Events (SSE) implementation of the
* {@link io.modelcontextprotocol.spec.McpTransport} that follows the MCP HTTP with SSE
@ -52,9 +52,9 @@ import java.util.function.Function;
*
* @author Christian Tzolov
* @see io.modelcontextprotocol.spec.McpTransport
* @see io.modelcontextprotocol.spec.ClientMcpTransport
* @see io.modelcontextprotocol.spec.McpClientTransport
*/
public class HttpClientSseClientTransport implements ClientMcpTransport {
public class HttpClientSseClientTransport implements McpClientTransport {
private static final Logger logger = LoggerFactory.getLogger(HttpClientSseClientTransport.class);

5
mcp/src/main/java/io/modelcontextprotocol/client/transport/StdioClientTransport.java
View file

@ -11,14 +11,13 @@ import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.function.Consumer;
import java.util.function.Function;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.ClientMcpTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage;
import io.modelcontextprotocol.util.Assert;
@ -38,7 +37,7 @@ import reactor.core.scheduler.Schedulers;
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
public class StdioClientTransport implements ClientMcpTransport {
public class StdioClientTransport implements McpClientTransport {
private static final Logger logger = LoggerFactory.getLogger(StdioClientTransport.class);

1641
mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServer.java
View file

File diff suppressed because it is too large Load diff

104
mcp/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java Normal file
View file

@ -0,0 +1,104 @@
package io.modelcontextprotocol.server;
import com.fasterxml.jackson.core.type.TypeReference;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpServerSession;
import reactor.core.publisher.Mono;
/**
* Represents an asynchronous exchange with a Model Context Protocol (MCP) client. The
* exchange provides methods to interact with the client and query its capabilities.
*
* @author Dariusz Jędrzejczyk
*/
public class McpAsyncServerExchange {
private final McpServerSession session;
private final McpSchema.ClientCapabilities clientCapabilities;
private final McpSchema.Implementation clientInfo;
private static final TypeReference<McpSchema.CreateMessageResult> CREATE_MESSAGE_RESULT_TYPE_REF = new TypeReference<>() {
};
private static final TypeReference<McpSchema.ListRootsResult> LIST_ROOTS_RESULT_TYPE_REF = new TypeReference<>() {
};
/**
* Create a new asynchronous exchange with the client.
* @param session The server session representing a 1-1 interaction.
* @param clientCapabilities The client capabilities that define the supported
* features and functionality.
* @param clientInfo The client implementation information.
*/
public McpAsyncServerExchange(McpServerSession session, McpSchema.ClientCapabilities clientCapabilities,
McpSchema.Implementation clientInfo) {
this.session = session;
this.clientCapabilities = clientCapabilities;
this.clientInfo = clientInfo;
}
/**
* Get the client capabilities that define the supported features and functionality.
* @return The client capabilities
*/
public McpSchema.ClientCapabilities getClientCapabilities() {
return this.clientCapabilities;
}
/**
* Get the client implementation information.
* @return The client implementation details
*/
public McpSchema.Implementation getClientInfo() {
return this.clientInfo;
}
/**
* 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
* @see McpSchema.CreateMessageRequest
* @see McpSchema.CreateMessageResult
* @see <a href=
* "https://spec.modelcontextprotocol.io/specification/client/sampling/">Sampling
* Specification</a>
*/
public Mono<McpSchema.CreateMessageResult> 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.session.sendRequest(McpSchema.METHOD_SAMPLING_CREATE_MESSAGE, createMessageRequest,
CREATE_MESSAGE_RESULT_TYPE_REF);
}
/**
* Retrieves the list of all roots provided by the client.
* @return A Mono that emits the list of roots result.
*/
public Mono<McpSchema.ListRootsResult> listRoots() {
return this.listRoots(null);
}
/**
* Retrieves a paginated list of roots provided by the client.
* @param cursor Optional pagination cursor from a previous list request
* @return A Mono that emits the list of roots result containing
*/
public Mono<McpSchema.ListRootsResult> listRoots(String cursor) {
return this.session.sendRequest(McpSchema.METHOD_ROOTS_LIST, new McpSchema.PaginatedRequest(cursor),
LIST_ROOTS_RESULT_TYPE_REF);
}
}

995
mcp/src/main/java/io/modelcontextprotocol/server/McpServer.java
View file

File diff suppressed because it is too large Load diff

360
mcp/src/main/java/io/modelcontextprotocol/server/McpServerFeatures.java
View file

@ -8,7 +8,8 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import io.modelcontextprotocol.spec.McpSchema;
@ -29,35 +30,35 @@ public class McpServerFeatures {
*
* @param serverInfo The server implementation details
* @param serverCapabilities The server capabilities
* @param tools The list of tool registrations
* @param resources The map of resource registrations
* @param tools The list of tool specifications
* @param resources The map of resource specifications
* @param resourceTemplates The list of resource templates
* @param prompts The map of prompt registrations
* @param prompts The map of prompt specifications
* @param rootsChangeConsumers The list of consumers that will be notified when the
* roots list changes
*/
record Async(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities serverCapabilities,
List<McpServerFeatures.AsyncToolRegistration> tools, Map<String, AsyncResourceRegistration> resources,
List<McpServerFeatures.AsyncToolSpecification> tools, Map<String, AsyncResourceSpecification> resources,
List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, McpServerFeatures.AsyncPromptRegistration> prompts,
List<Function<List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers) {
Map<String, McpServerFeatures.AsyncPromptSpecification> prompts,
List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers) {
/**
* Create an instance and validate the arguments.
* @param serverInfo The server implementation details
* @param serverCapabilities The server capabilities
* @param tools The list of tool registrations
* @param resources The map of resource registrations
* @param tools The list of tool specifications
* @param resources The map of resource specifications
* @param resourceTemplates The list of resource templates
* @param prompts The map of prompt registrations
* @param prompts The map of prompt specifications
* @param rootsChangeConsumers The list of consumers that will be notified when
* the roots list changes
*/
Async(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities serverCapabilities,
List<McpServerFeatures.AsyncToolRegistration> tools, Map<String, AsyncResourceRegistration> resources,
List<McpServerFeatures.AsyncToolSpecification> tools, Map<String, AsyncResourceSpecification> resources,
List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, McpServerFeatures.AsyncPromptRegistration> prompts,
List<Function<List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers) {
Map<String, McpServerFeatures.AsyncPromptSpecification> prompts,
List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers) {
Assert.notNull(serverInfo, "Server info must not be null");
@ -89,25 +90,26 @@ public class McpServerFeatures {
* user.
*/
static Async fromSync(Sync syncSpec) {
List<McpServerFeatures.AsyncToolRegistration> tools = new ArrayList<>();
List<McpServerFeatures.AsyncToolSpecification> tools = new ArrayList<>();
for (var tool : syncSpec.tools()) {
tools.add(AsyncToolRegistration.fromSync(tool));
tools.add(AsyncToolSpecification.fromSync(tool));
}
Map<String, AsyncResourceRegistration> resources = new HashMap<>();
Map<String, AsyncResourceSpecification> resources = new HashMap<>();
syncSpec.resources().forEach((key, resource) -> {
resources.put(key, AsyncResourceRegistration.fromSync(resource));
resources.put(key, AsyncResourceSpecification.fromSync(resource));
});
Map<String, AsyncPromptRegistration> prompts = new HashMap<>();
Map<String, AsyncPromptSpecification> prompts = new HashMap<>();
syncSpec.prompts().forEach((key, prompt) -> {
prompts.put(key, AsyncPromptRegistration.fromSync(prompt));
prompts.put(key, AsyncPromptSpecification.fromSync(prompt));
});
List<Function<List<McpSchema.Root>, Mono<Void>>> rootChangeConsumers = new ArrayList<>();
List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootChangeConsumers = new ArrayList<>();
for (var rootChangeConsumer : syncSpec.rootsChangeConsumers()) {
rootChangeConsumers.add(list -> Mono.<Void>fromRunnable(() -> rootChangeConsumer.accept(list))
rootChangeConsumers.add((exchange, list) -> Mono
.<Void>fromRunnable(() -> rootChangeConsumer.accept(new McpSyncServerExchange(exchange), list))
.subscribeOn(Schedulers.boundedElastic()));
}
@ -121,37 +123,37 @@ public class McpServerFeatures {
*
* @param serverInfo The server implementation details
* @param serverCapabilities The server capabilities
* @param tools The list of tool registrations
* @param resources The map of resource registrations
* @param tools The list of tool specifications
* @param resources The map of resource specifications
* @param resourceTemplates The list of resource templates
* @param prompts The map of prompt registrations
* @param prompts The map of prompt specifications
* @param rootsChangeConsumers The list of consumers that will be notified when the
* roots list changes
*/
record Sync(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities serverCapabilities,
List<McpServerFeatures.SyncToolRegistration> tools,
Map<String, McpServerFeatures.SyncResourceRegistration> resources,
List<McpServerFeatures.SyncToolSpecification> tools,
Map<String, McpServerFeatures.SyncResourceSpecification> resources,
List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, McpServerFeatures.SyncPromptRegistration> prompts,
List<Consumer<List<McpSchema.Root>>> rootsChangeConsumers) {
Map<String, McpServerFeatures.SyncPromptSpecification> prompts,
List<BiConsumer<McpSyncServerExchange, List<McpSchema.Root>>> rootsChangeConsumers) {
/**
* Create an instance and validate the arguments.
* @param serverInfo The server implementation details
* @param serverCapabilities The server capabilities
* @param tools The list of tool registrations
* @param resources The map of resource registrations
* @param tools The list of tool specifications
* @param resources The map of resource specifications
* @param resourceTemplates The list of resource templates
* @param prompts The map of prompt registrations
* @param prompts The map of prompt specifications
* @param rootsChangeConsumers The list of consumers that will be notified when
* the roots list changes
*/
Sync(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities serverCapabilities,
List<McpServerFeatures.SyncToolRegistration> tools,
Map<String, McpServerFeatures.SyncResourceRegistration> resources,
List<McpServerFeatures.SyncToolSpecification> tools,
Map<String, McpServerFeatures.SyncResourceSpecification> resources,
List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, McpServerFeatures.SyncPromptRegistration> prompts,
List<Consumer<List<McpSchema.Root>>> rootsChangeConsumers) {
Map<String, McpServerFeatures.SyncPromptSpecification> prompts,
List<BiConsumer<McpSyncServerExchange, List<McpSchema.Root>>> rootsChangeConsumers) {
Assert.notNull(serverInfo, "Server info must not be null");
@ -176,6 +178,255 @@ public class McpServerFeatures {
}
/**
* Specification of a tool with its asynchronous handler function. Tools are the
* primary way for MCP servers to expose functionality to AI models. Each tool
* represents a specific capability, such as:
* <ul>
* <li>Performing calculations
* <li>Accessing external APIs
* <li>Querying databases
* <li>Manipulating files
* <li>Executing system commands
* </ul>
*
* <p>
* Example tool specification: <pre>{@code
* new McpServerFeatures.AsyncToolSpecification(
* new Tool(
* "calculator",
* "Performs mathematical calculations",
* new JsonSchemaObject()
* .required("expression")
* .property("expression", JsonSchemaType.STRING)
* ),
* (exchange, args) -> {
* String expr = (String) args.get("expression");
* return Mono.fromSupplier(() -> evaluate(expr))
* .map(result -> new CallToolResult("Result: " + result));
* }
* )
* }</pre>
*
* @param tool The tool definition including name, description, and parameter schema
* @param call The function that implements the tool's logic, receiving arguments and
* returning results. The function's first argument is an
* {@link McpAsyncServerExchange} upon which the server can interact with the
* connected client. The second arguments is a map of tool arguments.
*/
public record AsyncToolSpecification(McpSchema.Tool tool,
BiFunction<McpAsyncServerExchange, Map<String, Object>, Mono<McpSchema.CallToolResult>> call) {
static AsyncToolSpecification fromSync(SyncToolSpecification tool) {
// FIXME: This is temporary, proper validation should be implemented
if (tool == null) {
return null;
}
return new AsyncToolSpecification(tool.tool(),
(exchange, map) -> Mono
.fromCallable(() -> tool.call().apply(new McpSyncServerExchange(exchange), map))
.subscribeOn(Schedulers.boundedElastic()));
}
}
/**
* Specification of a resource with its asynchronous handler function. Resources
* provide context to AI models by exposing data such as:
* <ul>
* <li>File contents
* <li>Database records
* <li>API responses
* <li>System information
* <li>Application state
* </ul>
*
* <p>
* Example resource specification: <pre>{@code
* new McpServerFeatures.AsyncResourceSpecification(
* new Resource("docs", "Documentation files", "text/markdown"),
* (exchange, request) ->
* Mono.fromSupplier(() -> readFile(request.getPath()))
* .map(ReadResourceResult::new)
* )
* }</pre>
*
* @param resource The resource definition including name, description, and MIME type
* @param readHandler The function that handles resource read requests. The function's
* first argument is an {@link McpAsyncServerExchange} upon which the server can
* interact with the connected client. The second arguments is a
* {@link io.modelcontextprotocol.spec.McpSchema.ReadResourceRequest}.
*/
public record AsyncResourceSpecification(McpSchema.Resource resource,
BiFunction<McpAsyncServerExchange, McpSchema.ReadResourceRequest, Mono<McpSchema.ReadResourceResult>> readHandler) {
static AsyncResourceSpecification fromSync(SyncResourceSpecification resource) {
// FIXME: This is temporary, proper validation should be implemented
if (resource == null) {
return null;
}
return new AsyncResourceSpecification(resource.resource(),
(exchange, req) -> Mono
.fromCallable(() -> resource.readHandler().apply(new McpSyncServerExchange(exchange), req))
.subscribeOn(Schedulers.boundedElastic()));
}
}
/**
* Specification of a prompt template with its asynchronous handler function. Prompts
* provide structured templates for AI model interactions, supporting:
* <ul>
* <li>Consistent message formatting
* <li>Parameter substitution
* <li>Context injection
* <li>Response formatting
* <li>Instruction templating
* </ul>
*
* <p>
* Example prompt specification: <pre>{@code
* new McpServerFeatures.AsyncPromptSpecification(
* new Prompt("analyze", "Code analysis template"),
* (exchange, request) -> {
* String code = request.getArguments().get("code");
* return Mono.just(new GetPromptResult(
* "Analyze this code:\n\n" + code + "\n\nProvide feedback on:"
* ));
* }
* )
* }</pre>
*
* @param prompt The prompt definition including name and description
* @param promptHandler The function that processes prompt requests and returns
* formatted templates. The function's first argument is an
* {@link McpAsyncServerExchange} upon which the server can interact with the
* connected client. The second arguments is a
* {@link io.modelcontextprotocol.spec.McpSchema.GetPromptRequest}.
*/
public record AsyncPromptSpecification(McpSchema.Prompt prompt,
BiFunction<McpAsyncServerExchange, McpSchema.GetPromptRequest, Mono<McpSchema.GetPromptResult>> promptHandler) {
static AsyncPromptSpecification fromSync(SyncPromptSpecification prompt) {
// FIXME: This is temporary, proper validation should be implemented
if (prompt == null) {
return null;
}
return new AsyncPromptSpecification(prompt.prompt(),
(exchange, req) -> Mono
.fromCallable(() -> prompt.promptHandler().apply(new McpSyncServerExchange(exchange), req))
.subscribeOn(Schedulers.boundedElastic()));
}
}
/**
* Specification of a tool with its synchronous handler function. Tools are the
* primary way for MCP servers to expose functionality to AI models. Each tool
* represents a specific capability, such as:
* <ul>
* <li>Performing calculations
* <li>Accessing external APIs
* <li>Querying databases
* <li>Manipulating files
* <li>Executing system commands
* </ul>
*
* <p>
* Example tool specification: <pre>{@code
* new McpServerFeatures.SyncToolSpecification(
* new Tool(
* "calculator",
* "Performs mathematical calculations",
* new JsonSchemaObject()
* .required("expression")
* .property("expression", JsonSchemaType.STRING)
* ),
* (exchange, args) -> {
* String expr = (String) args.get("expression");
* return new CallToolResult("Result: " + evaluate(expr));
* }
* )
* }</pre>
*
* @param tool The tool definition including name, description, and parameter schema
* @param call The function that implements the tool's logic, receiving arguments and
* returning results. The function's first argument is an
* {@link McpSyncServerExchange} upon which the server can interact with the connected
* client. The second arguments is a map of arguments passed to the tool.
*/
public record SyncToolSpecification(McpSchema.Tool tool,
BiFunction<McpSyncServerExchange, Map<String, Object>, McpSchema.CallToolResult> call) {
}
/**
* Specification of a resource with its synchronous handler function. Resources
* provide context to AI models by exposing data such as:
* <ul>
* <li>File contents
* <li>Database records
* <li>API responses
* <li>System information
* <li>Application state
* </ul>
*
* <p>
* Example resource specification: <pre>{@code
* new McpServerFeatures.SyncResourceSpecification(
* new Resource("docs", "Documentation files", "text/markdown"),
* (exchange, request) -> {
* String content = readFile(request.getPath());
* return new ReadResourceResult(content);
* }
* )
* }</pre>
*
* @param resource The resource definition including name, description, and MIME type
* @param readHandler The function that handles resource read requests. The function's
* first argument is an {@link McpSyncServerExchange} upon which the server can
* interact with the connected client. The second arguments is a
* {@link io.modelcontextprotocol.spec.McpSchema.ReadResourceRequest}.
*/
public record SyncResourceSpecification(McpSchema.Resource resource,
BiFunction<McpSyncServerExchange, McpSchema.ReadResourceRequest, McpSchema.ReadResourceResult> readHandler) {
}
/**
* Specification of a prompt template with its synchronous handler function. Prompts
* provide structured templates for AI model interactions, supporting:
* <ul>
* <li>Consistent message formatting
* <li>Parameter substitution
* <li>Context injection
* <li>Response formatting
* <li>Instruction templating
* </ul>
*
* <p>
* Example prompt specification: <pre>{@code
* new McpServerFeatures.SyncPromptSpecification(
* new Prompt("analyze", "Code analysis template"),
* (exchange, request) -> {
* String code = request.getArguments().get("code");
* return new GetPromptResult(
* "Analyze this code:\n\n" + code + "\n\nProvide feedback on:"
* );
* }
* )
* }</pre>
*
* @param prompt The prompt definition including name and description
* @param promptHandler The function that processes prompt requests and returns
* formatted templates. The function's first argument is an
* {@link McpSyncServerExchange} upon which the server can interact with the connected
* client. The second arguments is a
* {@link io.modelcontextprotocol.spec.McpSchema.GetPromptRequest}.
*/
public record SyncPromptSpecification(McpSchema.Prompt prompt,
BiFunction<McpSyncServerExchange, McpSchema.GetPromptRequest, McpSchema.GetPromptResult> promptHandler) {
}
// ---------------------------------------
// Deprecated registrations
// ---------------------------------------
/**
* Registration of a tool with its asynchronous handler function. Tools are the
* primary way for MCP servers to expose functionality to AI models. Each tool
@ -208,7 +459,10 @@ public class McpServerFeatures {
* @param tool The tool definition including name, description, and parameter schema
* @param call The function that implements the tool's logic, receiving arguments and
* returning results
* @deprecated This class is deprecated and will be removed in 0.9.0. Use
* {@link AsyncToolSpecification}.
*/
@Deprecated
public record AsyncToolRegistration(McpSchema.Tool tool,
Function<Map<String, Object>, Mono<McpSchema.CallToolResult>> call) {
@ -220,6 +474,10 @@ public class McpServerFeatures {
return new AsyncToolRegistration(tool.tool(),
map -> Mono.fromCallable(() -> tool.call().apply(map)).subscribeOn(Schedulers.boundedElastic()));
}
public AsyncToolSpecification toSpecification() {
return new AsyncToolSpecification(tool(), (exchange, map) -> call.apply(map));
}
}
/**
@ -246,7 +504,10 @@ public class McpServerFeatures {
*
* @param resource The resource definition including name, description, and MIME type
* @param readHandler The function that handles resource read requests
* @deprecated This class is deprecated and will be removed in 0.9.0. Use
* {@link AsyncResourceSpecification}.
*/
@Deprecated
public record AsyncResourceRegistration(McpSchema.Resource resource,
Function<McpSchema.ReadResourceRequest, Mono<McpSchema.ReadResourceResult>> readHandler) {
@ -259,6 +520,10 @@ public class McpServerFeatures {
req -> Mono.fromCallable(() -> resource.readHandler().apply(req))
.subscribeOn(Schedulers.boundedElastic()));
}
public AsyncResourceSpecification toSpecification() {
return new AsyncResourceSpecification(resource(), (exchange, request) -> readHandler.apply(request));
}
}
/**
@ -288,7 +553,10 @@ public class McpServerFeatures {
* @param prompt The prompt definition including name and description
* @param promptHandler The function that processes prompt requests and returns
* formatted templates
* @deprecated This class is deprecated and will be removed in 0.9.0. Use
* {@link AsyncPromptSpecification}.
*/
@Deprecated
public record AsyncPromptRegistration(McpSchema.Prompt prompt,
Function<McpSchema.GetPromptRequest, Mono<McpSchema.GetPromptResult>> promptHandler) {
@ -301,6 +569,10 @@ public class McpServerFeatures {
req -> Mono.fromCallable(() -> prompt.promptHandler().apply(req))
.subscribeOn(Schedulers.boundedElastic()));
}
public AsyncPromptSpecification toSpecification() {
return new AsyncPromptSpecification(prompt(), (exchange, request) -> promptHandler.apply(request));
}
}
/**
@ -335,9 +607,15 @@ public class McpServerFeatures {
* @param tool The tool definition including name, description, and parameter schema
* @param call The function that implements the tool's logic, receiving arguments and
* returning results
* @deprecated This class is deprecated and will be removed in 0.9.0. Use
* {@link SyncToolSpecification}.
*/
@Deprecated
public record SyncToolRegistration(McpSchema.Tool tool,
Function<Map<String, Object>, McpSchema.CallToolResult> call) {
public SyncToolSpecification toSpecification() {
return new SyncToolSpecification(tool, (exchange, map) -> call.apply(map));
}
}
/**
@ -364,9 +642,15 @@ public class McpServerFeatures {
*
* @param resource The resource definition including name, description, and MIME type
* @param readHandler The function that handles resource read requests
* @deprecated This class is deprecated and will be removed in 0.9.0. Use
* {@link SyncResourceSpecification}.
*/
@Deprecated
public record SyncResourceRegistration(McpSchema.Resource resource,
Function<McpSchema.ReadResourceRequest, McpSchema.ReadResourceResult> readHandler) {
public SyncResourceSpecification toSpecification() {
return new SyncResourceSpecification(resource, (exchange, request) -> readHandler.apply(request));
}
}
/**
@ -396,9 +680,15 @@ public class McpServerFeatures {
* @param prompt The prompt definition including name and description
* @param promptHandler The function that processes prompt requests and returns
* formatted templates
* @deprecated This class is deprecated and will be removed in 0.9.0. Use
* {@link SyncPromptSpecification}.
*/
@Deprecated
public record SyncPromptRegistration(McpSchema.Prompt prompt,
Function<McpSchema.GetPromptRequest, McpSchema.GetPromptResult> promptHandler) {
public SyncPromptSpecification toSpecification() {
return new SyncPromptSpecification(prompt, (exchange, request) -> promptHandler.apply(request));
}
}
}

48
mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServer.java
View file

@ -68,7 +68,10 @@ public class McpSyncServer {
/**
* Retrieves the list of all roots provided by the client.
* @return The list of roots
* @deprecated This method will be removed in 0.9.0. Use
* {@link McpSyncServerExchange#listRoots()}.
*/
@Deprecated
public McpSchema.ListRootsResult listRoots() {
return this.listRoots(null);
}
@ -77,7 +80,10 @@ public class McpSyncServer {
* Retrieves a paginated list of roots provided by the server.
* @param cursor Optional pagination cursor from a previous list request
* @return The list of roots
* @deprecated This method will be removed in 0.9.0. Use
* {@link McpSyncServerExchange#listRoots(String)}.
*/
@Deprecated
public McpSchema.ListRootsResult listRoots(String cursor) {
return this.asyncServer.listRoots(cursor).block();
}
@ -85,11 +91,22 @@ public class McpSyncServer {
/**
* Add a new tool handler.
* @param toolHandler The tool handler to add
* @deprecated This method will be removed in 0.9.0. Use
* {@link #addTool(McpServerFeatures.SyncToolSpecification)}.
*/
@Deprecated
public void addTool(McpServerFeatures.SyncToolRegistration toolHandler) {
this.asyncServer.addTool(McpServerFeatures.AsyncToolRegistration.fromSync(toolHandler)).block();
}
/**
* Add a new tool handler.
* @param toolHandler The tool handler to add
*/
public void addTool(McpServerFeatures.SyncToolSpecification toolHandler) {
this.asyncServer.addTool(McpServerFeatures.AsyncToolSpecification.fromSync(toolHandler)).block();
}
/**
* Remove a tool handler.
* @param toolName The name of the tool handler to remove
@ -101,11 +118,22 @@ public class McpSyncServer {
/**
* Add a new resource handler.
* @param resourceHandler The resource handler to add
* @deprecated This method will be removed in 0.9.0. Use
* {@link #addResource(McpServerFeatures.SyncResourceSpecification)}.
*/
@Deprecated
public void addResource(McpServerFeatures.SyncResourceRegistration resourceHandler) {
this.asyncServer.addResource(McpServerFeatures.AsyncResourceRegistration.fromSync(resourceHandler)).block();
}
/**
* Add a new resource handler.
* @param resourceHandler The resource handler to add
*/
public void addResource(McpServerFeatures.SyncResourceSpecification resourceHandler) {
this.asyncServer.addResource(McpServerFeatures.AsyncResourceSpecification.fromSync(resourceHandler)).block();
}
/**
* Remove a resource handler.
* @param resourceUri The URI of the resource handler to remove
@ -117,11 +145,22 @@ public class McpSyncServer {
/**
* Add a new prompt handler.
* @param promptRegistration The prompt registration to add
* @deprecated This method will be removed in 0.9.0. Use
* {@link #addPrompt(McpServerFeatures.SyncPromptSpecification)}.
*/
@Deprecated
public void addPrompt(McpServerFeatures.SyncPromptRegistration promptRegistration) {
this.asyncServer.addPrompt(McpServerFeatures.AsyncPromptRegistration.fromSync(promptRegistration)).block();
}
/**
* Add a new prompt handler.
* @param promptSpecification The prompt specification to add
*/
public void addPrompt(McpServerFeatures.SyncPromptSpecification promptSpecification) {
this.asyncServer.addPrompt(McpServerFeatures.AsyncPromptSpecification.fromSync(promptSpecification)).block();
}
/**
* Remove a prompt handler.
* @param promptName The name of the prompt handler to remove
@ -156,7 +195,10 @@ public class McpSyncServer {
/**
* Get the client capabilities that define the supported features and functionality.
* @return The client capabilities
* @deprecated This method will be removed in 0.9.0. Use
* {@link McpSyncServerExchange#getClientCapabilities()}.
*/
@Deprecated
public ClientCapabilities getClientCapabilities() {
return this.asyncServer.getClientCapabilities();
}
@ -164,7 +206,10 @@ public class McpSyncServer {
/**
* Get the client implementation information.
* @return The client implementation details
* @deprecated This method will be removed in 0.9.0. Use
* {@link McpSyncServerExchange#getClientInfo()}.
*/
@Deprecated
public McpSchema.Implementation getClientInfo() {
return this.asyncServer.getClientInfo();
}
@ -237,7 +282,10 @@ public class McpSyncServer {
* @see <a href=
* "https://spec.modelcontextprotocol.io/specification/client/sampling/">Sampling
* Specification</a>
* @deprecated This method will be removed in 0.9.0. Use
* {@link McpSyncServerExchange#createMessage(McpSchema.CreateMessageRequest)}.
*/
@Deprecated
public McpSchema.CreateMessageResult createMessage(McpSchema.CreateMessageRequest createMessageRequest) {
return this.asyncServer.createMessage(createMessageRequest).block();
}

78
mcp/src/main/java/io/modelcontextprotocol/server/McpSyncServerExchange.java Normal file
View file

@ -0,0 +1,78 @@
package io.modelcontextprotocol.server;
import com.fasterxml.jackson.core.type.TypeReference;
import io.modelcontextprotocol.spec.McpSchema;
/**
* Represents a synchronous exchange with a Model Context Protocol (MCP) client. The
* exchange provides methods to interact with the client and query its capabilities.
*
* @author Dariusz Jędrzejczyk
*/
public class McpSyncServerExchange {
private final McpAsyncServerExchange exchange;
/**
* Create a new synchronous exchange with the client using the provided asynchronous
* implementation as a delegate.
* @param exchange The asynchronous exchange to delegate to.
*/
public McpSyncServerExchange(McpAsyncServerExchange exchange) {
this.exchange = exchange;
}
/**
* Get the client capabilities that define the supported features and functionality.
* @return The client capabilities
*/
public McpSchema.ClientCapabilities getClientCapabilities() {
return this.exchange.getClientCapabilities();
}
/**
* Get the client implementation information.
* @return The client implementation details
*/
public McpSchema.Implementation getClientInfo() {
return this.exchange.getClientInfo();
}
/**
* 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 result containing the details of the sampling response
* @see McpSchema.CreateMessageRequest
* @see McpSchema.CreateMessageResult
* @see <a href=
* "https://spec.modelcontextprotocol.io/specification/client/sampling/">Sampling
* Specification</a>
*/
public McpSchema.CreateMessageResult createMessage(McpSchema.CreateMessageRequest createMessageRequest) {
return this.exchange.createMessage(createMessageRequest).block();
}
/**
* Retrieves the list of all roots provided by the client.
* @return The list of roots result.
*/
public McpSchema.ListRootsResult listRoots() {
return this.exchange.listRoots().block();
}
/**
* Retrieves a paginated list of roots provided by the client.
* @param cursor Optional pagination cursor from a previous list request
* @return The list of roots result
*/
public McpSchema.ListRootsResult listRoots(String cursor) {
return this.exchange.listRoots(cursor).block();
}
}

5
mcp/src/main/java/io/modelcontextprotocol/server/transport/HttpServletSseServerTransport.java
View file

@ -32,6 +32,9 @@ import reactor.core.publisher.Mono;
* specification. This implementation provides similar functionality to
* WebFluxSseServerTransport but uses the traditional Servlet API instead of WebFlux.
*
* @deprecated This class will be removed in 0.9.0. Use
* {@link HttpServletSseServerTransportProvider}.
*
* <p>
* The transport handles two types of endpoints:
* <ul>
@ -48,7 +51,6 @@ import reactor.core.publisher.Mono;
* <li>Graceful shutdown support</li>
* <li>Error handling and response formatting</li>
* </ul>
*
* @author Christian Tzolov
* @author Alexandros Pappas
* @see ServerMcpTransport
@ -56,6 +58,7 @@ import reactor.core.publisher.Mono;
*/
@WebServlet(asyncSupported = true)
@Deprecated
public class HttpServletSseServerTransport extends HttpServlet implements ServerMcpTransport {
/** Logger for this class */

432
mcp/src/main/java/io/modelcontextprotocol/server/transport/HttpServletSseServerTransportProvider.java Normal file
View file

@ -0,0 +1,432 @@
/*
* Copyright 2024 - 2024 the original author or authors.
*/
package io.modelcontextprotocol.server.transport;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpServerSession;
import io.modelcontextprotocol.spec.McpServerTransport;
import io.modelcontextprotocol.spec.McpServerTransportProvider;
import jakarta.servlet.AsyncContext;
import jakarta.servlet.ServletException;
import jakarta.servlet.annotation.WebServlet;
import jakarta.servlet.http.HttpServlet;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* A Servlet-based implementation of the MCP HTTP with Server-Sent Events (SSE) transport
* specification. This implementation provides similar functionality to
* WebFluxSseServerTransportProvider but uses the traditional Servlet API instead of
* WebFlux.
*
* <p>
* The transport handles two types of endpoints:
* <ul>
* <li>SSE endpoint (/sse) - Establishes a long-lived connection for server-to-client
* events</li>
* <li>Message endpoint (configurable) - Handles client-to-server message requests</li>
* </ul>
*
* <p>
* Features:
* <ul>
* <li>Asynchronous message handling using Servlet 6.0 async support</li>
* <li>Session management for multiple client connections</li>
* <li>Graceful shutdown support</li>
* <li>Error handling and response formatting</li>
* </ul>
*
* @author Christian Tzolov
* @author Alexandros Pappas
* @see McpServerTransportProvider
* @see HttpServlet
*/
@WebServlet(asyncSupported = true)
public class HttpServletSseServerTransportProvider extends HttpServlet implements McpServerTransportProvider {
/** Logger for this class */
private static final Logger logger = LoggerFactory.getLogger(HttpServletSseServerTransportProvider.class);
public static final String UTF_8 = "UTF-8";
public static final String APPLICATION_JSON = "application/json";
public static final String FAILED_TO_SEND_ERROR_RESPONSE = "Failed to send error response: {}";
/** Default endpoint path for SSE connections */
public static final String DEFAULT_SSE_ENDPOINT = "/sse";
/** Event type for regular messages */
public static final String MESSAGE_EVENT_TYPE = "message";
/** Event type for endpoint information */
public static final String ENDPOINT_EVENT_TYPE = "endpoint";
/** JSON object mapper for serialization/deserialization */
private final ObjectMapper objectMapper;
/** The endpoint path for handling client messages */
private final String messageEndpoint;
/** The endpoint path for handling SSE connections */
private final String sseEndpoint;
/** Map of active client sessions, keyed by session ID */
private final Map<String, McpServerSession> sessions = new ConcurrentHashMap<>();
/** Flag indicating if the transport is in the process of shutting down */
private final AtomicBoolean isClosing = new AtomicBoolean(false);
/** Session factory for creating new sessions */
private McpServerSession.Factory sessionFactory;
/**
* Creates a new HttpServletSseServerTransportProvider instance with a custom SSE
* endpoint.
* @param objectMapper The JSON object mapper to use for message
* serialization/deserialization
* @param messageEndpoint The endpoint path where clients will send their messages
* @param sseEndpoint The endpoint path where clients will establish SSE connections
*/
public HttpServletSseServerTransportProvider(ObjectMapper objectMapper, String messageEndpoint,
String sseEndpoint) {
this.objectMapper = objectMapper;
this.messageEndpoint = messageEndpoint;
this.sseEndpoint = sseEndpoint;
}
/**
* Creates a new HttpServletSseServerTransportProvider instance with the default SSE
* endpoint.
* @param objectMapper The JSON object mapper to use for message
* serialization/deserialization
* @param messageEndpoint The endpoint path where clients will send their messages
*/
public HttpServletSseServerTransportProvider(ObjectMapper objectMapper, String messageEndpoint) {
this(objectMapper, messageEndpoint, DEFAULT_SSE_ENDPOINT);
}
/**
* Sets the session factory for creating new sessions.
* @param sessionFactory The session factory to use
*/
@Override
public void setSessionFactory(McpServerSession.Factory sessionFactory) {
this.sessionFactory = sessionFactory;
}
/**
* Broadcasts a notification to all connected clients.
* @param method The method name for the notification
* @param params The parameters for the notification
* @return A Mono that completes when the broadcast attempt is finished
*/
@Override
public Mono<Void> notifyClients(String method, Map<String, Object> params) {
if (sessions.isEmpty()) {
logger.debug("No active sessions to broadcast message to");
return Mono.empty();
}
logger.debug("Attempting to broadcast message to {} active sessions", sessions.size());
return Flux.fromIterable(sessions.values())
.flatMap(session -> session.sendNotification(method, params)
.doOnError(
e -> logger.error("Failed to send message to session {}: {}", session.getId(), e.getMessage()))
.onErrorComplete())
.then();
}
/**
* Handles GET requests to establish SSE connections.
* <p>
* This method sets up a new SSE connection when a client connects to the SSE
* endpoint. It configures the response headers for SSE, creates a new session, and
* sends the initial endpoint information to the client.
* @param request The HTTP servlet request
* @param response The HTTP servlet response
* @throws ServletException If a servlet-specific error occurs
* @throws IOException If an I/O error occurs
*/
@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException {
String pathInfo = request.getPathInfo();
if (!sseEndpoint.equals(pathInfo)) {
response.sendError(HttpServletResponse.SC_NOT_FOUND);
return;
}
if (isClosing.get()) {
response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, "Server is shutting down");
return;
}
response.setContentType("text/event-stream");
response.setCharacterEncoding(UTF_8);
response.setHeader("Cache-Control", "no-cache");
response.setHeader("Connection", "keep-alive");
response.setHeader("Access-Control-Allow-Origin", "*");
String sessionId = UUID.randomUUID().toString();
AsyncContext asyncContext = request.startAsync();
asyncContext.setTimeout(0);
PrintWriter writer = response.getWriter();
// Create a new session transport
HttpServletMcpSessionTransport sessionTransport = new HttpServletMcpSessionTransport(sessionId, asyncContext,
writer);
// Create a new session using the session factory
McpServerSession session = sessionFactory.create(sessionTransport);
this.sessions.put(sessionId, session);
// Send initial endpoint event
this.sendEvent(writer, ENDPOINT_EVENT_TYPE, messageEndpoint + "?sessionId=" + sessionId);
}
/**
* Handles POST requests for client messages.
* <p>
* This method processes incoming messages from clients, routes them through the
* session handler, and sends back the appropriate response. It handles error cases
* and formats error responses according to the MCP specification.
* @param request The HTTP servlet request
* @param response The HTTP servlet response
* @throws ServletException If a servlet-specific error occurs
* @throws IOException If an I/O error occurs
*/
@Override
protected void doPost(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException {
if (isClosing.get()) {
response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, "Server is shutting down");
return;
}
String pathInfo = request.getPathInfo();
if (!messageEndpoint.equals(pathInfo)) {
response.sendError(HttpServletResponse.SC_NOT_FOUND);
return;
}
// Get the session ID from the request parameter
String sessionId = request.getParameter("sessionId");
if (sessionId == null) {
response.setContentType(APPLICATION_JSON);
response.setCharacterEncoding(UTF_8);
response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
String jsonError = objectMapper.writeValueAsString(new McpError("Session ID missing in message endpoint"));
PrintWriter writer = response.getWriter();
writer.write(jsonError);
writer.flush();
return;
}
// Get the session from the sessions map
McpServerSession session = sessions.get(sessionId);
if (session == null) {
response.setContentType(APPLICATION_JSON);
response.setCharacterEncoding(UTF_8);
response.setStatus(HttpServletResponse.SC_NOT_FOUND);
String jsonError = objectMapper.writeValueAsString(new McpError("Session not found: " + sessionId));
PrintWriter writer = response.getWriter();
writer.write(jsonError);
writer.flush();
return;
}
try {
BufferedReader reader = request.getReader();
StringBuilder body = new StringBuilder();
String line;
while ((line = reader.readLine()) != null) {
body.append(line);
}
McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(objectMapper, body.toString());
// Process the message through the session's handle method
session.handle(message).block(); // Block for Servlet compatibility
response.setStatus(HttpServletResponse.SC_OK);
}
catch (Exception e) {
logger.error("Error processing message: {}", e.getMessage());
try {
McpError mcpError = new McpError(e.getMessage());
response.setContentType(APPLICATION_JSON);
response.setCharacterEncoding(UTF_8);
response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
String jsonError = objectMapper.writeValueAsString(mcpError);
PrintWriter writer = response.getWriter();
writer.write(jsonError);
writer.flush();
}
catch (IOException ex) {
logger.error(FAILED_TO_SEND_ERROR_RESPONSE, ex.getMessage());
response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, "Error processing message");
}
}
}
/**
* Initiates a graceful shutdown of the transport.
* <p>
* This method marks the transport as closing and closes all active client sessions.
* New connection attempts will be rejected during shutdown.
* @return A Mono that completes when all sessions have been closed
*/
@Override
public Mono<Void> closeGracefully() {
isClosing.set(true);
logger.debug("Initiating graceful shutdown with {} active sessions", sessions.size());
return Flux.fromIterable(sessions.values()).flatMap(McpServerSession::closeGracefully).then();
}
/**
* Sends an SSE event to a client.
* @param writer The writer to send the event through
* @param eventType The type of event (message or endpoint)
* @param data The event data
* @throws IOException If an error occurs while writing the event
*/
private void sendEvent(PrintWriter writer, String eventType, String data) throws IOException {
writer.write("event: " + eventType + "\n");
writer.write("data: " + data + "\n\n");
writer.flush();
if (writer.checkError()) {
throw new IOException("Client disconnected");
}
}
/**
* Cleans up resources when the servlet is being destroyed.
* <p>
* This method ensures a graceful shutdown by closing all client connections before
* calling the parent's destroy method.
*/
@Override
public void destroy() {
closeGracefully().block();
super.destroy();
}
/**
* Implementation of McpServerTransport for HttpServlet SSE sessions. This class
* handles the transport-level communication for a specific client session.
*/
private class HttpServletMcpSessionTransport implements McpServerTransport {
private final String sessionId;
private final AsyncContext asyncContext;
private final PrintWriter writer;
/**
* Creates a new session transport with the specified ID and SSE writer.
* @param sessionId The unique identifier for this session
* @param asyncContext The async context for the session
* @param writer The writer for sending server events to the client
*/
HttpServletMcpSessionTransport(String sessionId, AsyncContext asyncContext, PrintWriter writer) {
this.sessionId = sessionId;
this.asyncContext = asyncContext;
this.writer = writer;
logger.debug("Session transport {} initialized with SSE writer", sessionId);
}
/**
* Sends a JSON-RPC message to the client through the SSE connection.
* @param message The JSON-RPC message to send
* @return A Mono that completes when the message has been sent
*/
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
return Mono.fromRunnable(() -> {
try {
String jsonText = objectMapper.writeValueAsString(message);
sendEvent(writer, MESSAGE_EVENT_TYPE, jsonText);
logger.debug("Message sent to session {}", sessionId);
}
catch (Exception e) {
logger.error("Failed to send message to session {}: {}", sessionId, e.getMessage());
sessions.remove(sessionId);
asyncContext.complete();
}
});
}
/**
* Converts data from one type to another using the configured ObjectMapper.
* @param data The source data object to convert
* @param typeRef The target type reference
* @return The converted object of type T
* @param <T> The target type
*/
@Override
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
return objectMapper.convertValue(data, typeRef);
}
/**
* Initiates a graceful shutdown of the transport.
* @return A Mono that completes when the shutdown is complete
*/
@Override
public Mono<Void> closeGracefully() {
return Mono.fromRunnable(() -> {
logger.debug("Closing session transport: {}", sessionId);
try {
sessions.remove(sessionId);
asyncContext.complete();
logger.debug("Successfully completed async context for session {}", sessionId);
}
catch (Exception e) {
logger.warn("Failed to complete async context for session {}: {}", sessionId, e.getMessage());
}
});
}
/**
* Closes the transport immediately.
*/
@Override
public void close() {
try {
sessions.remove(sessionId);
asyncContext.complete();
logger.debug("Successfully completed async context for session {}", sessionId);
}
catch (Exception e) {
logger.warn("Failed to complete async context for session {}: {}", sessionId, e.getMessage());
}
}
}
}

3
mcp/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransport.java
View file

@ -33,7 +33,10 @@ import reactor.core.scheduler.Schedulers;
* over stdin/stdout, with errors and debug information sent to stderr.
*
* @author Christian Tzolov
* @deprecated This method will be removed in 0.9.0. Use
* {@link io.modelcontextprotocol.server.transport.StdioServerTransportProvider} instead.
*/
@Deprecated
public class StdioServerTransport implements ServerMcpTransport {
private static final Logger logger = LoggerFactory.getLogger(StdioServerTransport.class);

306
mcp/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java Normal file
View file

@ -0,0 +1,306 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.server.transport;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.Reader;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.JSONRPCMessage;
import io.modelcontextprotocol.spec.McpServerSession;
import io.modelcontextprotocol.spec.McpServerTransport;
import io.modelcontextprotocol.spec.McpServerTransportProvider;
import io.modelcontextprotocol.util.Assert;
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.Scheduler;
import reactor.core.scheduler.Schedulers;
/**
* Implementation of the MCP Stdio transport provider for servers that communicates using
* standard input/output streams. Messages are exchanged as newline-delimited JSON-RPC
* messages over stdin/stdout, with errors and debug information sent to stderr.
*
* @author Christian Tzolov
*/
public class StdioServerTransportProvider implements McpServerTransportProvider {
private static final Logger logger = LoggerFactory.getLogger(StdioServerTransportProvider.class);
private final ObjectMapper objectMapper;
private final InputStream inputStream;
private final OutputStream outputStream;
private McpServerSession session;
private final AtomicBoolean isClosing = new AtomicBoolean(false);
private final Sinks.One<Void> inboundReady = Sinks.one();
/**
* Creates a new StdioServerTransportProvider with a default ObjectMapper and System
* streams.
*/
public StdioServerTransportProvider() {
this(new ObjectMapper());
}
/**
* Creates a new StdioServerTransportProvider with the specified ObjectMapper and
* System streams.
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
*/
public StdioServerTransportProvider(ObjectMapper objectMapper) {
this(objectMapper, System.in, System.out);
}
/**
* Creates a new StdioServerTransportProvider with the specified ObjectMapper and
* streams.
* @param objectMapper The ObjectMapper to use for JSON serialization/deserialization
* @param inputStream The input stream to read from
* @param outputStream The output stream to write to
*/
public StdioServerTransportProvider(ObjectMapper objectMapper, InputStream inputStream, OutputStream outputStream) {
Assert.notNull(objectMapper, "The ObjectMapper can not be null");
Assert.notNull(inputStream, "The InputStream can not be null");
Assert.notNull(outputStream, "The OutputStream can not be null");
this.objectMapper = objectMapper;
this.inputStream = inputStream;
this.outputStream = outputStream;
}
@Override
public void setSessionFactory(McpServerSession.Factory sessionFactory) {
// Create a single session for the stdio connection
this.session = sessionFactory.create(new StdioMcpSessionTransport());
}
@Override
public Mono<Void> notifyClients(String method, Map<String, Object> params) {
if (this.session == null) {
return Mono.error(new McpError("No session to close"));
}
return this.session.sendNotification(method, params)
.doOnError(e -> logger.error("Failed to send notification: {}", e.getMessage()));
}
@Override
public Mono<Void> closeGracefully() {
if (this.session == null) {
return Mono.empty();
}
return this.session.closeGracefully();
}
/**
* Implementation of McpServerTransport for the stdio session.
*/
private class StdioMcpSessionTransport implements McpServerTransport {
private final Sinks.Many<JSONRPCMessage> inboundSink;
private final Sinks.Many<JSONRPCMessage> outboundSink;
private final AtomicBoolean isStarted = new AtomicBoolean(false);
/** Scheduler for handling inbound messages */
private Scheduler inboundScheduler;
/** Scheduler for handling outbound messages */
private Scheduler outboundScheduler;
private final Sinks.One<Void> outboundReady = Sinks.one();
public StdioMcpSessionTransport() {
this.inboundSink = Sinks.many().unicast().onBackpressureBuffer();
this.outboundSink = Sinks.many().unicast().onBackpressureBuffer();
// Use bounded schedulers for better resource management
this.inboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(),
"stdio-inbound");
this.outboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(),
"stdio-outbound");
handleIncomingMessages();
startInboundProcessing();
startOutboundProcessing();
}
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {
return Mono.zip(inboundReady.asMono(), outboundReady.asMono()).then(Mono.defer(() -> {
if (outboundSink.tryEmitNext(message).isSuccess()) {
return Mono.empty();
}
else {
return Mono.error(new RuntimeException("Failed to enqueue message"));
}
}));
}
@Override
public <T> T unmarshalFrom(Object data, TypeReference<T> typeRef) {
return objectMapper.convertValue(data, typeRef);
}
@Override
public Mono<Void> closeGracefully() {
return Mono.fromRunnable(() -> {
isClosing.set(true);
logger.debug("Session transport closing gracefully");
inboundSink.tryEmitComplete();
});
}
@Override
public void close() {
isClosing.set(true);
logger.debug("Session transport closed");
}
private void handleIncomingMessages() {
this.inboundSink.asFlux().flatMap(message -> session.handle(message)).doOnTerminate(() -> {
// The outbound processing will dispose its scheduler upon completion
this.outboundSink.tryEmitComplete();
this.inboundScheduler.dispose();
}).subscribe();
}
/**
* Starts the inbound processing thread that reads JSON-RPC messages from stdin.
* Messages are deserialized and passed to the session for handling.
*/
private void startInboundProcessing() {
if (isStarted.compareAndSet(false, true)) {
this.inboundScheduler.schedule(() -> {
inboundReady.tryEmitValue(null);
BufferedReader reader = null;
try {
reader = new BufferedReader(new InputStreamReader(inputStream));
while (!isClosing.get()) {
try {
String line = reader.readLine();
if (line == null || isClosing.get()) {
break;
}
logger.debug("Received JSON message: {}", line);
try {
McpSchema.JSONRPCMessage message = McpSchema.deserializeJsonRpcMessage(objectMapper,
line);
if (!this.inboundSink.tryEmitNext(message).isSuccess()) {
// logIfNotClosing("Failed to enqueue message");
break;
}
}
catch (Exception e) {
logIfNotClosing("Error processing inbound message", e);
break;
}
}
catch (IOException e) {
logIfNotClosing("Error reading from stdin", e);
break;
}
}
}
catch (Exception e) {
logIfNotClosing("Error in inbound processing", e);
}
finally {
isClosing.set(true);
if (session != null) {
session.close();
}
inboundSink.tryEmitComplete();
}
});
}
}
/**
* Starts the outbound processing thread that writes JSON-RPC messages to stdout.
* Messages are serialized to JSON and written with a newline delimiter.
*/
private void startOutboundProcessing() {
Function<Flux<JSONRPCMessage>, Flux<JSONRPCMessage>> outboundConsumer = messages -> messages // @formatter:off
.doOnSubscribe(subscription -> outboundReady.tryEmitValue(null))
.publishOn(outboundScheduler)
.handle((message, sink) -> {
if (message != null && !isClosing.get()) {
try {
String jsonMessage = objectMapper.writeValueAsString(message);
// Escape any embedded newlines in the JSON message as per spec
jsonMessage = jsonMessage.replace("\r\n", "\\n").replace("\n", "\\n").replace("\r", "\\n");
synchronized (outputStream) {
outputStream.write(jsonMessage.getBytes(StandardCharsets.UTF_8));
outputStream.write("\n".getBytes(StandardCharsets.UTF_8));
outputStream.flush();
}
sink.next(message);
}
catch (IOException e) {
if (!isClosing.get()) {
logger.error("Error writing message", e);
sink.error(new RuntimeException(e));
}
else {
logger.debug("Stream closed during shutdown", e);
}
}
}
else if (isClosing.get()) {
sink.complete();
}
})
.doOnComplete(() -> {
isClosing.set(true);
outboundScheduler.dispose();
})
.doOnError(e -> {
if (!isClosing.get()) {
logger.error("Error in outbound processing", e);
isClosing.set(true);
outboundScheduler.dispose();
}
})
.map(msg -> (JSONRPCMessage) msg);
outboundConsumer.apply(outboundSink.asFlux()).subscribe();
} // @formatter:on
private void logIfNotClosing(String message, Exception e) {
if (!isClosing.get()) {
logger.error(message, e);
}
}
}
}

2
mcp/src/main/java/io/modelcontextprotocol/spec/ClientMcpTransport.java
View file

@ -7,7 +7,9 @@ package io.modelcontextprotocol.spec;
* Marker interface for the client-side MCP transport.
*
* @author Christian Tzolov
* @deprecated This class will be removed in 0.9.0. Use {@link McpClientTransport}.
*/
@Deprecated
public interface ClientMcpTransport extends McpTransport {
}

3
mcp/src/main/java/io/modelcontextprotocol/spec/DefaultMcpSession.java
View file

@ -34,7 +34,10 @@ import reactor.core.publisher.MonoSink;
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
* @deprecated This method will be removed in 0.9.0. Use {@link McpClientSession} instead
*/
@Deprecated
public class DefaultMcpSession implements McpSession {
/** Logger for this class */

288
mcp/src/main/java/io/modelcontextprotocol/spec/McpClientSession.java Normal file
View file

@ -0,0 +1,288 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.spec;
import java.time.Duration;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import com.fasterxml.jackson.core.type.TypeReference;
import io.modelcontextprotocol.util.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
/**
* Default implementation of the MCP (Model Context Protocol) session that manages
* bidirectional JSON-RPC communication between clients and servers. This implementation
* follows the MCP specification for message exchange and transport handling.
*
* <p>
* The session manages:
* <ul>
* <li>Request/response handling with unique message IDs</li>
* <li>Notification processing</li>
* <li>Message timeout management</li>
* <li>Transport layer abstraction</li>
* </ul>
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
public class McpClientSession implements McpSession {
/** Logger for this class */
private static final Logger logger = LoggerFactory.getLogger(McpClientSession.class);
/** Duration to wait for request responses before timing out */
private final Duration requestTimeout;
/** Transport layer implementation for message exchange */
private final McpTransport transport;
/** Map of pending responses keyed by request ID */
private final ConcurrentHashMap<Object, MonoSink<McpSchema.JSONRPCResponse>> pendingResponses = new ConcurrentHashMap<>();
/** Map of request handlers keyed by method name */
private final ConcurrentHashMap<String, RequestHandler<?>> requestHandlers = new ConcurrentHashMap<>();
/** Map of notification handlers keyed by method name */
private final ConcurrentHashMap<String, NotificationHandler> notificationHandlers = new ConcurrentHashMap<>();
/** Session-specific prefix for request IDs */
private final String sessionPrefix = UUID.randomUUID().toString().substring(0, 8);
/** Atomic counter for generating unique request IDs */
private final AtomicLong requestCounter = new AtomicLong(0);
private final Disposable connection;
/**
* Functional interface for handling incoming JSON-RPC requests. Implementations
* should process the request parameters and return a response.
*
* @param <T> Response type
*/
@FunctionalInterface
public interface RequestHandler<T> {
/**
* Handles an incoming request with the given parameters.
* @param params The request parameters
* @return A Mono containing the response object
*/
Mono<T> handle(Object params);
}
/**
* Functional interface for handling incoming JSON-RPC notifications. Implementations
* should process the notification parameters without returning a response.
*/
@FunctionalInterface
public interface NotificationHandler {
/**
* Handles an incoming notification with the given parameters.
* @param params The notification parameters
* @return A Mono that completes when the notification is processed
*/
Mono<Void> handle(Object params);
}
/**
* Creates a new McpClientSession with the specified configuration and handlers.
* @param requestTimeout Duration to wait for responses
* @param transport Transport implementation for message exchange
* @param requestHandlers Map of method names to request handlers
* @param notificationHandlers Map of method names to notification handlers
*/
public McpClientSession(Duration requestTimeout, McpTransport transport,
Map<String, RequestHandler<?>> requestHandlers, Map<String, NotificationHandler> notificationHandlers) {
Assert.notNull(requestTimeout, "The requstTimeout can not be null");
Assert.notNull(transport, "The transport can not be null");
Assert.notNull(requestHandlers, "The requestHandlers can not be null");
Assert.notNull(notificationHandlers, "The notificationHandlers can not be null");
this.requestTimeout = requestTimeout;
this.transport = transport;
this.requestHandlers.putAll(requestHandlers);
this.notificationHandlers.putAll(notificationHandlers);
// TODO: consider mono.transformDeferredContextual where the Context contains
// the
// Observation associated with the individual message - it can be used to
// create child Observation and emit it together with the message to the
// consumer
this.connection = this.transport.connect(mono -> mono.doOnNext(message -> {
if (message instanceof McpSchema.JSONRPCResponse response) {
logger.debug("Received Response: {}", response);
var sink = pendingResponses.remove(response.id());
if (sink == null) {
logger.warn("Unexpected response for unkown id {}", response.id());
}
else {
sink.success(response);
}
}
else if (message instanceof McpSchema.JSONRPCRequest request) {
logger.debug("Received request: {}", request);
handleIncomingRequest(request).subscribe(response -> transport.sendMessage(response).subscribe(),
error -> {
var errorResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
null, new McpSchema.JSONRPCResponse.JSONRPCError(
McpSchema.ErrorCodes.INTERNAL_ERROR, error.getMessage(), null));
transport.sendMessage(errorResponse).subscribe();
});
}
else if (message instanceof McpSchema.JSONRPCNotification notification) {
logger.debug("Received notification: {}", notification);
handleIncomingNotification(notification).subscribe(null,
error -> logger.error("Error handling notification: {}", error.getMessage()));
}
})).subscribe();
}
/**
* Handles an incoming JSON-RPC request by routing it to the appropriate handler.
* @param request The incoming JSON-RPC request
* @return A Mono containing the JSON-RPC response
*/
private Mono<McpSchema.JSONRPCResponse> handleIncomingRequest(McpSchema.JSONRPCRequest request) {
return Mono.defer(() -> {
var handler = this.requestHandlers.get(request.method());
if (handler == null) {
MethodNotFoundError error = getMethodNotFoundError(request.method());
return Mono.just(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), null,
new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.METHOD_NOT_FOUND,
error.message(), error.data())));
}
return handler.handle(request.params())
.map(result -> new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), result, null))
.onErrorResume(error -> Mono.just(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
null, new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.INTERNAL_ERROR,
error.getMessage(), null)))); // TODO: add error message
// through the data field
});
}
record MethodNotFoundError(String method, String message, Object data) {
}
public static MethodNotFoundError getMethodNotFoundError(String method) {
switch (method) {
case McpSchema.METHOD_ROOTS_LIST:
return new MethodNotFoundError(method, "Roots not supported",
Map.of("reason", "Client does not have roots capability"));
default:
return new MethodNotFoundError(method, "Method not found: " + method, null);
}
}
/**
* Handles an incoming JSON-RPC notification by routing it to the appropriate handler.
* @param notification The incoming JSON-RPC notification
* @return A Mono that completes when the notification is processed
*/
private Mono<Void> handleIncomingNotification(McpSchema.JSONRPCNotification notification) {
return Mono.defer(() -> {
var handler = notificationHandlers.get(notification.method());
if (handler == null) {
logger.error("No handler registered for notification method: {}", notification.method());
return Mono.empty();
}
return handler.handle(notification.params());
});
}
/**
* Generates a unique request ID in a non-blocking way. Combines a session-specific
* prefix with an atomic counter to ensure uniqueness.
* @return A unique request ID string
*/
private String generateRequestId() {
return this.sessionPrefix + "-" + this.requestCounter.getAndIncrement();
}
/**
* Sends a JSON-RPC request and returns the response.
* @param <T> The expected response type
* @param method The method name to call
* @param requestParams The request parameters
* @param typeRef Type reference for response deserialization
* @return A Mono containing the response
*/
@Override
public <T> Mono<T> sendRequest(String method, Object requestParams, TypeReference<T> typeRef) {
String requestId = this.generateRequestId();
return Mono.<McpSchema.JSONRPCResponse>create(sink -> {
this.pendingResponses.put(requestId, sink);
McpSchema.JSONRPCRequest jsonrpcRequest = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, method,
requestId, requestParams);
this.transport.sendMessage(jsonrpcRequest)
// TODO: It's most efficient to create a dedicated Subscriber here
.subscribe(v -> {
}, error -> {
this.pendingResponses.remove(requestId);
sink.error(error);
});
}).timeout(this.requestTimeout).handle((jsonRpcResponse, sink) -> {
if (jsonRpcResponse.error() != null) {
sink.error(new McpError(jsonRpcResponse.error()));
}
else {
if (typeRef.getType().equals(Void.class)) {
sink.complete();
}
else {
sink.next(this.transport.unmarshalFrom(jsonRpcResponse.result(), typeRef));
}
}
});
}
/**
* Sends a JSON-RPC notification.
* @param method The method name for the notification
* @param params The notification parameters
* @return A Mono that completes when the notification is sent
*/
@Override
public Mono<Void> sendNotification(String method, Map<String, Object> params) {
McpSchema.JSONRPCNotification jsonrpcNotification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION,
method, params);
return this.transport.sendMessage(jsonrpcNotification);
}
/**
* Closes the session gracefully, allowing pending operations to complete.
* @return A Mono that completes when the session is closed
*/
@Override
public Mono<Void> closeGracefully() {
return Mono.defer(() -> {
this.connection.dispose();
return transport.closeGracefully();
});
}
/**
* Closes the session immediately, potentially interrupting pending operations.
*/
@Override
public void close() {
this.connection.dispose();
transport.close();
}
}

21
mcp/src/main/java/io/modelcontextprotocol/spec/McpClientTransport.java Normal file
View file

@ -0,0 +1,21 @@
/*
* Copyright 2024 - 2024 the original author or authors.
*/
package io.modelcontextprotocol.spec;
import java.util.function.Function;
import reactor.core.publisher.Mono;
/**
* Marker interface for the client-side MCP transport.
*
* @author Christian Tzolov
* @author Dariusz Jędrzejczyk
*/
public interface McpClientTransport extends ClientMcpTransport {
@Override
Mono<Void> connect(Function<Mono<McpSchema.JSONRPCMessage>, Mono<McpSchema.JSONRPCMessage>> handler);
}

354
mcp/src/main/java/io/modelcontextprotocol/spec/McpServerSession.java Normal file
View file

@ -0,0 +1,354 @@
package io.modelcontextprotocol.spec;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import com.fasterxml.jackson.core.type.TypeReference;
import io.modelcontextprotocol.server.McpAsyncServerExchange;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.publisher.Sinks;
/**
* Represents a Model Control Protocol (MCP) session on the server side. It manages
* bidirectional JSON-RPC communication with the client.
*/
public class McpServerSession implements McpSession {
private static final Logger logger = LoggerFactory.getLogger(McpServerSession.class);
private final ConcurrentHashMap<Object, MonoSink<McpSchema.JSONRPCResponse>> pendingResponses = new ConcurrentHashMap<>();
private final String id;
private final AtomicLong requestCounter = new AtomicLong(0);
private final InitRequestHandler initRequestHandler;
private final InitNotificationHandler initNotificationHandler;
private final Map<String, RequestHandler<?>> requestHandlers;
private final Map<String, NotificationHandler> notificationHandlers;
private final McpServerTransport transport;
private final Sinks.One<McpAsyncServerExchange> exchangeSink = Sinks.one();
private final AtomicReference<McpSchema.ClientCapabilities> clientCapabilities = new AtomicReference<>();
private final AtomicReference<McpSchema.Implementation> clientInfo = new AtomicReference<>();
private static final int STATE_UNINITIALIZED = 0;
private static final int STATE_INITIALIZING = 1;
private static final int STATE_INITIALIZED = 2;
private final AtomicInteger state = new AtomicInteger(STATE_UNINITIALIZED);
/**
* Creates a new server session with the given parameters and the transport to use.
* @param id session id
* @param transport the transport to use
* @param initHandler called when a
* {@link io.modelcontextprotocol.spec.McpSchema.InitializeRequest} is received by the
* server
* @param initNotificationHandler called when a
* {@link McpSchema.METHOD_NOTIFICATION_INITIALIZED} is received.
* @param requestHandlers map of request handlers to use
* @param notificationHandlers map of notification handlers to use
*/
public McpServerSession(String id, McpServerTransport transport, InitRequestHandler initHandler,
InitNotificationHandler initNotificationHandler, Map<String, RequestHandler<?>> requestHandlers,
Map<String, NotificationHandler> notificationHandlers) {
this.id = id;
this.transport = transport;
this.initRequestHandler = initHandler;
this.initNotificationHandler = initNotificationHandler;
this.requestHandlers = requestHandlers;
this.notificationHandlers = notificationHandlers;
}
/**
* Retrieve the session id.
* @return session id
*/
public String getId() {
return this.id;
}
/**
* Called upon successful initialization sequence between the client and the server
* with the client capabilities and information.
*
* <a href=
* "https://github.com/modelcontextprotocol/specification/blob/main/docs/specification/basic/lifecycle.md#initialization">Initialization
* Spec</a>
* @param clientCapabilities the capabilities the connected client provides
* @param clientInfo the information about the connected client
*/
public void init(McpSchema.ClientCapabilities clientCapabilities, McpSchema.Implementation clientInfo) {
this.clientCapabilities.lazySet(clientCapabilities);
this.clientInfo.lazySet(clientInfo);
}
private String generateRequestId() {
return this.id + "-" + this.requestCounter.getAndIncrement();
}
@Override
public <T> Mono<T> sendRequest(String method, Object requestParams, TypeReference<T> typeRef) {
String requestId = this.generateRequestId();
return Mono.<McpSchema.JSONRPCResponse>create(sink -> {
this.pendingResponses.put(requestId, sink);
McpSchema.JSONRPCRequest jsonrpcRequest = new McpSchema.JSONRPCRequest(McpSchema.JSONRPC_VERSION, method,
requestId, requestParams);
this.transport.sendMessage(jsonrpcRequest).subscribe(v -> {
}, error -> {
this.pendingResponses.remove(requestId);
sink.error(error);
});
}).timeout(Duration.ofSeconds(10)).handle((jsonRpcResponse, sink) -> {
if (jsonRpcResponse.error() != null) {
sink.error(new McpError(jsonRpcResponse.error()));
}
else {
if (typeRef.getType().equals(Void.class)) {
sink.complete();
}
else {
sink.next(this.transport.unmarshalFrom(jsonRpcResponse.result(), typeRef));
}
}
});
}
@Override
public Mono<Void> sendNotification(String method, Map<String, Object> params) {
McpSchema.JSONRPCNotification jsonrpcNotification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION,
method, params);
return this.transport.sendMessage(jsonrpcNotification);
}
/**
* Called by the {@link McpServerTransportProvider} once the session is determined.
* The purpose of this method is to dispatch the message to an appropriate handler as
* specified by the MCP server implementation
* ({@link io.modelcontextprotocol.server.McpAsyncServer} or
* {@link io.modelcontextprotocol.server.McpSyncServer}) via
* {@link McpServerSession.Factory} that the server creates.
* @param message the incoming JSON-RPC message
* @return a Mono that completes when the message is processed
*/
public Mono<Void> handle(McpSchema.JSONRPCMessage message) {
return Mono.defer(() -> {
// TODO handle errors for communication to without initialization happening
// first
if (message instanceof McpSchema.JSONRPCResponse response) {
logger.debug("Received Response: {}", response);
var sink = pendingResponses.remove(response.id());
if (sink == null) {
logger.warn("Unexpected response for unknown id {}", response.id());
}
else {
sink.success(response);
}
return Mono.empty();
}
else if (message instanceof McpSchema.JSONRPCRequest request) {
logger.debug("Received request: {}", request);
return handleIncomingRequest(request).onErrorResume(error -> {
var errorResponse = new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), null,
new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.INTERNAL_ERROR,
error.getMessage(), null));
// TODO: Should the error go to SSE or back as POST return?
return this.transport.sendMessage(errorResponse).then(Mono.empty());
}).flatMap(this.transport::sendMessage);
}
else if (message instanceof McpSchema.JSONRPCNotification notification) {
// TODO handle errors for communication to without initialization
// happening first
logger.debug("Received notification: {}", notification);
// TODO: in case of error, should the POST request be signalled?
return handleIncomingNotification(notification)
.doOnError(error -> logger.error("Error handling notification: {}", error.getMessage()));
}
else {
logger.warn("Received unknown message type: {}", message);
return Mono.empty();
}
});
}
/**
* Handles an incoming JSON-RPC request by routing it to the appropriate handler.
* @param request The incoming JSON-RPC request
* @return A Mono containing the JSON-RPC response
*/
private Mono<McpSchema.JSONRPCResponse> handleIncomingRequest(McpSchema.JSONRPCRequest request) {
return Mono.defer(() -> {
Mono<?> resultMono;
if (McpSchema.METHOD_INITIALIZE.equals(request.method())) {
// TODO handle situation where already initialized!
McpSchema.InitializeRequest initializeRequest = transport.unmarshalFrom(request.params(),
new TypeReference<McpSchema.InitializeRequest>() {
});
this.state.lazySet(STATE_INITIALIZING);
this.init(initializeRequest.capabilities(), initializeRequest.clientInfo());
resultMono = this.initRequestHandler.handle(initializeRequest);
}
else {
// TODO handle errors for communication to this session without
// initialization happening first
var handler = this.requestHandlers.get(request.method());
if (handler == null) {
MethodNotFoundError error = getMethodNotFoundError(request.method());
return Mono.just(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), null,
new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.METHOD_NOT_FOUND,
error.message(), error.data())));
}
resultMono = this.exchangeSink.asMono().flatMap(exchange -> handler.handle(exchange, request.params()));
}
return resultMono
.map(result -> new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), result, null))
.onErrorResume(error -> Mono.just(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(),
null, new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.INTERNAL_ERROR,
error.getMessage(), null)))); // TODO: add error message
// through the data field
});
}
/**
* Handles an incoming JSON-RPC notification by routing it to the appropriate handler.
* @param notification The incoming JSON-RPC notification
* @return A Mono that completes when the notification is processed
*/
private Mono<Void> handleIncomingNotification(McpSchema.JSONRPCNotification notification) {
return Mono.defer(() -> {
if (McpSchema.METHOD_NOTIFICATION_INITIALIZED.equals(notification.method())) {
this.state.lazySet(STATE_INITIALIZED);
exchangeSink.tryEmitValue(new McpAsyncServerExchange(this, clientCapabilities.get(), clientInfo.get()));
return this.initNotificationHandler.handle();
}
var handler = notificationHandlers.get(notification.method());
if (handler == null) {
logger.error("No handler registered for notification method: {}", notification.method());
return Mono.empty();
}
return this.exchangeSink.asMono().flatMap(exchange -> handler.handle(exchange, notification.params()));
});
}
record MethodNotFoundError(String method, String message, Object data) {
}
static MethodNotFoundError getMethodNotFoundError(String method) {
switch (method) {
case McpSchema.METHOD_ROOTS_LIST:
return new MethodNotFoundError(method, "Roots not supported",
Map.of("reason", "Client does not have roots capability"));
default:
return new MethodNotFoundError(method, "Method not found: " + method, null);
}
}
@Override
public Mono<Void> closeGracefully() {
return this.transport.closeGracefully();
}
@Override
public void close() {
this.transport.close();
}
/**
* Request handler for the initialization request.
*/
public interface InitRequestHandler {
/**
* Handles the initialization request.
* @param initializeRequest the initialization request by the client
* @return a Mono that will emit the result of the initialization
*/
Mono<McpSchema.InitializeResult> handle(McpSchema.InitializeRequest initializeRequest);
}
/**
* Notification handler for the initialization notification from the client.
*/
public interface InitNotificationHandler {
/**
* Specifies an action to take upon successful initialization.
* @return a Mono that will complete when the initialization is acted upon.
*/
Mono<Void> handle();
}
/**
* A handler for client-initiated notifications.
*/
public interface NotificationHandler {
/**
* Handles a notification from the client.
* @param exchange the exchange associated with the client that allows calling
* back to the connected client or inspecting its capabilities.
* @param params the parameters of the notification.
* @return a Mono that completes once the notification is handled.
*/
Mono<Void> handle(McpAsyncServerExchange exchange, Object params);
}
/**
* A handler for client-initiated requests.
*
* @param <T> the type of the response that is expected as a result of handling the
* request.
*/
public interface RequestHandler<T> {
/**
* Handles a request from the client.
* @param exchange the exchange associated with the client that allows calling
* back to the connected client or inspecting its capabilities.
* @param params the parameters of the request.
* @return a Mono that will emit the response to the request.
*/
Mono<T> handle(McpAsyncServerExchange exchange, Object params);
}
/**
* Factory for creating server sessions which delegate to a provided 1:1 transport
* with a connected client.
*/
@FunctionalInterface
public interface Factory {
/**
* Creates a new 1:1 representation of the client-server interaction.
* @param sessionTransport the transport to use for communication with the client.
* @return a new server session.
*/
McpServerSession create(McpServerTransport sessionTransport);
}
}

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