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

feat: add Streamable HTTP Server abstractions and implement WebFlux transport provider (#420)

Browse source 
- Add WebFluxStreamableServerTransportProvider for streamable HTTP sessions
- Add WebFluxStatelessServerTransport for stateless HTTP MCP servers
  - Add McpStatelessServerFeatures, McpStatelessServerHandler, McpStatelessRequestHandler, McpStatelessNotificationHandler
- Refactor server architecture to support multiple transport categories:
   streamable, single-session, stateless
- Introduce McpTransportContext for transport-level metadata extraction
- Add session management capabilities for streamable HTTP connections
- Update MCP protocol version to 2025-03-26
- Add test coverage for new transport implementations
- Implement integration tests for both stateless and streamable transports

Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>
Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
Co-authored-by: Christian Tzolov <christian.tzolov@broadcom.com>
This commit is contained in:
Dariusz Jędrzejczyk 2025-07-30 13:41:18 +02:00 • committed by GitHub
commit a8f5a3fb4f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
40 changed files with 6578 additions and 704 deletions
Download patch file Download diff file Expand all files Collapse all files

50
mcp/src/main/java/io/modelcontextprotocol/server/DefaultMcpStatelessServerHandler.java Normal file
View file

@ -0,0 +1,50 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import java.util.Map;
class DefaultMcpStatelessServerHandler implements McpStatelessServerHandler {
private static final Logger logger = LoggerFactory.getLogger(DefaultMcpStatelessServerHandler.class);
Map<String, McpStatelessRequestHandler<?>> requestHandlers;
Map<String, McpStatelessNotificationHandler> notificationHandlers;
public DefaultMcpStatelessServerHandler(Map<String, McpStatelessRequestHandler<?>> requestHandlers,
Map<String, McpStatelessNotificationHandler> notificationHandlers) {
this.requestHandlers = requestHandlers;
this.notificationHandlers = notificationHandlers;
}
@Override
public Mono<McpSchema.JSONRPCResponse> handleRequest(McpTransportContext transportContext,
McpSchema.JSONRPCRequest request) {
McpStatelessRequestHandler<?> requestHandler = this.requestHandlers.get(request.method());
if (requestHandler == null) {
return Mono.error(new McpError("Missing handler for request type: " + request.method()));
}
return requestHandler.handle(transportContext, request.params())
.map(result -> new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), result, null))
.onErrorResume(t -> Mono.just(new McpSchema.JSONRPCResponse(McpSchema.JSONRPC_VERSION, request.id(), null,
new McpSchema.JSONRPCResponse.JSONRPCError(McpSchema.ErrorCodes.INTERNAL_ERROR, t.getMessage(),
null))));
}
@Override
public Mono<Void> handleNotification(McpTransportContext transportContext,
McpSchema.JSONRPCNotification notification) {
McpStatelessNotificationHandler notificationHandler = this.notificationHandlers.get(notification.method());
if (notificationHandler == null) {
logger.warn("Missing handler for notification type: {}", notification.method());
return Mono.empty();
}
return notificationHandler.handle(transportContext, notification.params());
}
}

45
mcp/src/main/java/io/modelcontextprotocol/server/DefaultMcpTransportContext.java Normal file
View file

@ -0,0 +1,45 @@
package io.modelcontextprotocol.server;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Default implementation for {@link McpTransportContext} which uses a Thread-safe map.
* Objects of this kind are mutable.
*
* @author Dariusz Jędrzejczyk
*/
public class DefaultMcpTransportContext implements McpTransportContext {
private final Map<String, Object> storage;
/**
* Create an empty instance.
*/
public DefaultMcpTransportContext() {
this.storage = new ConcurrentHashMap<>();
}
DefaultMcpTransportContext(Map<String, Object> storage) {
this.storage = storage;
}
@Override
public Object get(String key) {
return this.storage.get(key);
}
@Override
public void put(String key, Object value) {
this.storage.put(key, value);
}
/**
* Allows copying the contents.
* @return new instance with the copy of the underlying map
*/
public McpTransportContext copy() {
return new DefaultMcpTransportContext(new ConcurrentHashMap<>(this.storage));
}
}

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

@ -15,6 +15,9 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.BiFunction;
import io.modelcontextprotocol.spec.DefaultMcpStreamableServerSessionFactory;
import io.modelcontextprotocol.spec.McpServerTransportProviderBase;
import io.modelcontextprotocol.spec.McpStreamableServerTransportProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -86,7 +89,7 @@ public class McpAsyncServer {
private static final Logger logger = LoggerFactory.getLogger(McpAsyncServer.class);
private final McpServerTransportProvider mcpTransportProvider;
private final McpServerTransportProviderBase mcpTransportProvider;
private final ObjectMapper objectMapper;
@ -139,7 +142,56 @@ public class McpAsyncServer {
this.uriTemplateManagerFactory = uriTemplateManagerFactory;
this.jsonSchemaValidator = jsonSchemaValidator;
Map<String, McpServerSession.RequestHandler<?>> requestHandlers = new HashMap<>();
Map<String, McpRequestHandler<?>> requestHandlers = prepareRequestHandlers();
Map<String, McpNotificationHandler> notificationHandlers = prepareNotificationHandlers(features);
mcpTransportProvider.setSessionFactory(transport -> new McpServerSession(UUID.randomUUID().toString(),
requestTimeout, transport, this::asyncInitializeRequestHandler, requestHandlers, notificationHandlers));
}
McpAsyncServer(McpStreamableServerTransportProvider mcpTransportProvider, ObjectMapper objectMapper,
McpServerFeatures.Async features, Duration requestTimeout,
McpUriTemplateManagerFactory uriTemplateManagerFactory, JsonSchemaValidator jsonSchemaValidator) {
this.mcpTransportProvider = mcpTransportProvider;
this.objectMapper = objectMapper;
this.serverInfo = features.serverInfo();
this.serverCapabilities = features.serverCapabilities();
this.instructions = features.instructions();
this.tools.addAll(withStructuredOutputHandling(jsonSchemaValidator, features.tools()));
this.resources.putAll(features.resources());
this.resourceTemplates.addAll(features.resourceTemplates());
this.prompts.putAll(features.prompts());
this.completions.putAll(features.completions());
this.uriTemplateManagerFactory = uriTemplateManagerFactory;
this.jsonSchemaValidator = jsonSchemaValidator;
Map<String, McpRequestHandler<?>> requestHandlers = prepareRequestHandlers();
Map<String, McpNotificationHandler> notificationHandlers = prepareNotificationHandlers(features);
mcpTransportProvider.setSessionFactory(new DefaultMcpStreamableServerSessionFactory(requestTimeout,
this::asyncInitializeRequestHandler, requestHandlers, notificationHandlers));
}
private Map<String, McpNotificationHandler> prepareNotificationHandlers(McpServerFeatures.Async features) {
Map<String, McpNotificationHandler> notificationHandlers = new HashMap<>();
notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_INITIALIZED, (exchange, params) -> Mono.empty());
List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers = features
.rootsChangeConsumers();
if (Utils.isEmpty(rootsChangeConsumers)) {
rootsChangeConsumers = List.of((exchange, roots) -> Mono.fromRunnable(() -> logger
.warn("Roots list changed notification, but no consumers provided. Roots list changed: {}", roots)));
}
notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_ROOTS_LIST_CHANGED,
asyncRootsListChangedNotificationHandler(rootsChangeConsumers));
return notificationHandlers;
}
private Map<String, McpRequestHandler<?>> prepareRequestHandlers() {
Map<String, McpRequestHandler<?>> requestHandlers = new HashMap<>();
// Initialize request handlers for standard MCP methods
@ -174,25 +226,7 @@ public class McpAsyncServer {
if (this.serverCapabilities.completions() != null) {
requestHandlers.put(McpSchema.METHOD_COMPLETION_COMPLETE, completionCompleteRequestHandler());
}
Map<String, McpServerSession.NotificationHandler> notificationHandlers = new HashMap<>();
notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_INITIALIZED, (exchange, params) -> Mono.empty());
List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers = features
.rootsChangeConsumers();
if (Utils.isEmpty(rootsChangeConsumers)) {
rootsChangeConsumers = List.of((exchange, roots) -> Mono.fromRunnable(() -> logger
.warn("Roots list changed notification, but no consumers provided. Roots list changed: {}", roots)));
}
notificationHandlers.put(McpSchema.METHOD_NOTIFICATION_ROOTS_LIST_CHANGED,
asyncRootsListChangedNotificationHandler(rootsChangeConsumers));
mcpTransportProvider.setSessionFactory(
transport -> new McpServerSession(UUID.randomUUID().toString(), requestTimeout, transport,
this::asyncInitializeRequestHandler, Mono::empty, requestHandlers, notificationHandlers));
return requestHandlers;
}
// ---------------------------------------
@ -258,7 +292,7 @@ public class McpAsyncServer {
this.mcpTransportProvider.close();
}
private McpServerSession.NotificationHandler asyncRootsListChangedNotificationHandler(
private McpNotificationHandler asyncRootsListChangedNotificationHandler(
List<BiFunction<McpAsyncServerExchange, List<McpSchema.Root>, Mono<Void>>> rootsChangeConsumers) {
return (exchange, params) -> exchange.listRoots()
.flatMap(listRootsResult -> Flux.fromIterable(rootsChangeConsumers)
@ -450,7 +484,7 @@ public class McpAsyncServer {
return this.mcpTransportProvider.notifyClients(McpSchema.METHOD_NOTIFICATION_TOOLS_LIST_CHANGED, null);
}
private McpServerSession.RequestHandler<McpSchema.ListToolsResult> toolsListRequestHandler() {
private McpRequestHandler<McpSchema.ListToolsResult> toolsListRequestHandler() {
return (exchange, params) -> {
List<Tool> tools = this.tools.stream().map(McpServerFeatures.AsyncToolSpecification::tool).toList();
@ -458,7 +492,7 @@ public class McpAsyncServer {
};
}
private McpServerSession.RequestHandler<CallToolResult> toolsCallRequestHandler() {
private McpRequestHandler<CallToolResult> toolsCallRequestHandler() {
return (exchange, params) -> {
McpSchema.CallToolRequest callToolRequest = objectMapper.convertValue(params,
new TypeReference<McpSchema.CallToolRequest>() {
@ -551,7 +585,7 @@ public class McpAsyncServer {
resourcesUpdatedNotification);
}
private McpServerSession.RequestHandler<McpSchema.ListResourcesResult> resourcesListRequestHandler() {
private McpRequestHandler<McpSchema.ListResourcesResult> resourcesListRequestHandler() {
return (exchange, params) -> {
var resourceList = this.resources.values()
.stream()
@ -561,7 +595,7 @@ public class McpAsyncServer {
};
}
private McpServerSession.RequestHandler<McpSchema.ListResourceTemplatesResult> resourceTemplateListRequestHandler() {
private McpRequestHandler<McpSchema.ListResourceTemplatesResult> resourceTemplateListRequestHandler() {
return (exchange, params) -> Mono
.just(new McpSchema.ListResourceTemplatesResult(this.getResourceTemplates(), null));
@ -585,7 +619,7 @@ public class McpAsyncServer {
return list;
}
private McpServerSession.RequestHandler<McpSchema.ReadResourceResult> resourcesReadRequestHandler() {
private McpRequestHandler<McpSchema.ReadResourceResult> resourcesReadRequestHandler() {
return (exchange, params) -> {
McpSchema.ReadResourceRequest resourceRequest = objectMapper.convertValue(params,
new TypeReference<McpSchema.ReadResourceRequest>() {
@ -678,7 +712,7 @@ public class McpAsyncServer {
return this.mcpTransportProvider.notifyClients(McpSchema.METHOD_NOTIFICATION_PROMPTS_LIST_CHANGED, null);
}
private McpServerSession.RequestHandler<McpSchema.ListPromptsResult> promptsListRequestHandler() {
private McpRequestHandler<McpSchema.ListPromptsResult> promptsListRequestHandler() {
return (exchange, params) -> {
// TODO: Implement pagination
// McpSchema.PaginatedRequest request = objectMapper.convertValue(params,
@ -694,7 +728,7 @@ public class McpAsyncServer {
};
}
private McpServerSession.RequestHandler<McpSchema.GetPromptResult> promptsGetRequestHandler() {
private McpRequestHandler<McpSchema.GetPromptResult> promptsGetRequestHandler() {
return (exchange, params) -> {
McpSchema.GetPromptRequest promptRequest = objectMapper.convertValue(params,
new TypeReference<McpSchema.GetPromptRequest>() {
@ -740,7 +774,7 @@ public class McpAsyncServer {
loggingMessageNotification);
}
private McpServerSession.RequestHandler<Object> setLoggerRequestHandler() {
private McpRequestHandler<Object> setLoggerRequestHandler() {
return (exchange, params) -> {
return Mono.defer(() -> {
@ -759,7 +793,7 @@ public class McpAsyncServer {
};
}
private McpServerSession.RequestHandler<McpSchema.CompleteResult> completionCompleteRequestHandler() {
private McpRequestHandler<McpSchema.CompleteResult> completionCompleteRequestHandler() {
return (exchange, params) -> {
McpSchema.CompleteRequest request = parseCompletionParams(params);

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

@ -9,10 +9,11 @@ import java.util.Collections;
import com.fasterxml.jackson.core.type.TypeReference;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpLoggableSession;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.LoggingLevel;
import io.modelcontextprotocol.spec.McpSchema.LoggingMessageNotification;
import io.modelcontextprotocol.spec.McpServerSession;
import io.modelcontextprotocol.spec.McpSession;
import io.modelcontextprotocol.util.Assert;
import reactor.core.publisher.Mono;
@ -25,13 +26,15 @@ import reactor.core.publisher.Mono;
*/
public class McpAsyncServerExchange {
private final McpServerSession session;
private final String sessionId;
private final McpLoggableSession session;
private final McpSchema.ClientCapabilities clientCapabilities;
private final McpSchema.Implementation clientInfo;
private volatile LoggingLevel minLoggingLevel = LoggingLevel.INFO;
private final McpTransportContext transportContext;
private static final TypeReference<McpSchema.CreateMessageResult> CREATE_MESSAGE_RESULT_TYPE_REF = new TypeReference<>() {
};
@ -51,12 +54,39 @@ public class McpAsyncServerExchange {
* @param clientCapabilities The client capabilities that define the supported
* features and functionality.
* @param clientInfo The client implementation information.
* @deprecated Use
* {@link #McpAsyncServerExchange(String, McpLoggableSession, McpSchema.ClientCapabilities, McpSchema.Implementation, McpTransportContext)}
*/
public McpAsyncServerExchange(McpServerSession session, McpSchema.ClientCapabilities clientCapabilities,
@Deprecated
public McpAsyncServerExchange(McpSession session, McpSchema.ClientCapabilities clientCapabilities,
McpSchema.Implementation clientInfo) {
this.sessionId = null;
if (!(session instanceof McpLoggableSession)) {
throw new IllegalArgumentException("Expecting session to be a McpLoggableSession instance");
}
this.session = (McpLoggableSession) session;
this.clientCapabilities = clientCapabilities;
this.clientInfo = clientInfo;
this.transportContext = McpTransportContext.EMPTY;
}
/**
* 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.
* @param transportContext context associated with the client as extracted from the
* transport
*/
public McpAsyncServerExchange(String sessionId, McpLoggableSession session,
McpSchema.ClientCapabilities clientCapabilities, McpSchema.Implementation clientInfo,
McpTransportContext transportContext) {
this.sessionId = sessionId;
this.session = session;
this.clientCapabilities = clientCapabilities;
this.clientInfo = clientInfo;
this.transportContext = transportContext;
}
/**
@ -75,6 +105,24 @@ public class McpAsyncServerExchange {
return this.clientInfo;
}
/**
* Provides the {@link McpTransportContext} associated with the transport layer. For
* HTTP transports it can contain the metadata associated with the HTTP request that
* triggered the processing.
* @return the transport context object
*/
public McpTransportContext transportContext() {
return this.transportContext;
}
/**
* Provides the Session ID.
* @return session ID string
*/
public String sessionId() {
return this.sessionId;
}
/**
* 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
@ -170,7 +218,7 @@ public class McpAsyncServerExchange {
}
return Mono.defer(() -> {
if (this.isNotificationForLevelAllowed(loggingMessageNotification.level())) {
if (this.session.isNotificationForLevelAllowed(loggingMessageNotification.level())) {
return this.session.sendNotification(McpSchema.METHOD_NOTIFICATION_MESSAGE, loggingMessageNotification);
}
return Mono.empty();
@ -205,11 +253,7 @@ public class McpAsyncServerExchange {
*/
void setMinLoggingLevel(LoggingLevel minLoggingLevel) {
Assert.notNull(minLoggingLevel, "minLoggingLevel must not be null");
this.minLoggingLevel = minLoggingLevel;
}
private boolean isNotificationForLevelAllowed(LoggingLevel loggingLevel) {
return loggingLevel.level() >= this.minLoggingLevel.level();
this.session.setMinLoggingLevel(minLoggingLevel);
}
}

18
mcp/src/main/java/io/modelcontextprotocol/server/McpInitRequestHandler.java Normal file
View file

@ -0,0 +1,18 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.spec.McpSchema;
import reactor.core.publisher.Mono;
/**
* Request handler for the initialization request.
*/
public interface McpInitRequestHandler {
/**
* 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);
}

19
mcp/src/main/java/io/modelcontextprotocol/server/McpNotificationHandler.java Normal file
View file

@ -0,0 +1,19 @@
package io.modelcontextprotocol.server;
import reactor.core.publisher.Mono;
/**
* A handler for client-initiated notifications.
*/
public interface McpNotificationHandler {
/**
* 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);
}

22
mcp/src/main/java/io/modelcontextprotocol/server/McpRequestHandler.java Normal file
View file

@ -0,0 +1,22 @@
package io.modelcontextprotocol.server;
import reactor.core.publisher.Mono;
/**
* 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 McpRequestHandler<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);
}

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

File diff suppressed because it is too large Load diff

671
mcp/src/main/java/io/modelcontextprotocol/server/McpStatelessAsyncServer.java Normal file
View file

@ -0,0 +1,671 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.server;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.modelcontextprotocol.spec.JsonSchemaValidator;
import io.modelcontextprotocol.spec.McpError;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.CallToolResult;
import io.modelcontextprotocol.spec.McpSchema.ResourceTemplate;
import io.modelcontextprotocol.spec.McpSchema.Tool;
import io.modelcontextprotocol.spec.McpStatelessServerTransport;
import io.modelcontextprotocol.util.Assert;
import io.modelcontextprotocol.util.DeafaultMcpUriTemplateManagerFactory;
import io.modelcontextprotocol.util.McpUriTemplateManagerFactory;
import io.modelcontextprotocol.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.BiFunction;
/**
* A stateless MCP server implementation for use with Streamable HTTP transport types. It
* allows simple horizontal scalability since it does not maintain a session and does not
* require initialization. Each instance of the server can be reached with no prior
* knowledge and can serve the clients with the capabilities it supports.
*
* @author Dariusz Jędrzejczyk
*/
public class McpStatelessAsyncServer {
private static final Logger logger = LoggerFactory.getLogger(McpStatelessAsyncServer.class);
private final McpStatelessServerTransport mcpTransportProvider;
private final ObjectMapper objectMapper;
private final McpSchema.ServerCapabilities serverCapabilities;
private final McpSchema.Implementation serverInfo;
private final String instructions;
private final CopyOnWriteArrayList<McpStatelessServerFeatures.AsyncToolSpecification> tools = new CopyOnWriteArrayList<>();
private final CopyOnWriteArrayList<ResourceTemplate> resourceTemplates = new CopyOnWriteArrayList<>();
private final ConcurrentHashMap<String, McpStatelessServerFeatures.AsyncResourceSpecification> resources = new ConcurrentHashMap<>();
private final ConcurrentHashMap<String, McpStatelessServerFeatures.AsyncPromptSpecification> prompts = new ConcurrentHashMap<>();
private final ConcurrentHashMap<McpSchema.CompleteReference, McpStatelessServerFeatures.AsyncCompletionSpecification> completions = new ConcurrentHashMap<>();
private List<String> protocolVersions = List.of(McpSchema.LATEST_PROTOCOL_VERSION);
private McpUriTemplateManagerFactory uriTemplateManagerFactory = new DeafaultMcpUriTemplateManagerFactory();
private final JsonSchemaValidator jsonSchemaValidator;
McpStatelessAsyncServer(McpStatelessServerTransport mcpTransport, ObjectMapper objectMapper,
McpStatelessServerFeatures.Async features, Duration requestTimeout,
McpUriTemplateManagerFactory uriTemplateManagerFactory, JsonSchemaValidator jsonSchemaValidator) {
this.mcpTransportProvider = mcpTransport;
this.objectMapper = objectMapper;
this.serverInfo = features.serverInfo();
this.serverCapabilities = features.serverCapabilities();
this.instructions = features.instructions();
this.tools.addAll(withStructuredOutputHandling(jsonSchemaValidator, features.tools()));
this.resources.putAll(features.resources());
this.resourceTemplates.addAll(features.resourceTemplates());
this.prompts.putAll(features.prompts());
this.completions.putAll(features.completions());
this.uriTemplateManagerFactory = uriTemplateManagerFactory;
this.jsonSchemaValidator = jsonSchemaValidator;
Map<String, McpStatelessRequestHandler<?>> requestHandlers = new HashMap<>();
// Initialize request handlers for standard MCP methods
// Ping MUST respond with an empty data, but not NULL response.
requestHandlers.put(McpSchema.METHOD_PING, (ctx, params) -> Mono.just(Map.of()));
requestHandlers.put(McpSchema.METHOD_INITIALIZE, asyncInitializeRequestHandler());
// Add tools API handlers if the tool capability is enabled
if (this.serverCapabilities.tools() != null) {
requestHandlers.put(McpSchema.METHOD_TOOLS_LIST, toolsListRequestHandler());
requestHandlers.put(McpSchema.METHOD_TOOLS_CALL, toolsCallRequestHandler());
}
// Add resources API handlers if provided
if (this.serverCapabilities.resources() != null) {
requestHandlers.put(McpSchema.METHOD_RESOURCES_LIST, resourcesListRequestHandler());
requestHandlers.put(McpSchema.METHOD_RESOURCES_READ, resourcesReadRequestHandler());
requestHandlers.put(McpSchema.METHOD_RESOURCES_TEMPLATES_LIST, resourceTemplateListRequestHandler());
}
// Add prompts API handlers if provider exists
if (this.serverCapabilities.prompts() != null) {
requestHandlers.put(McpSchema.METHOD_PROMPT_LIST, promptsListRequestHandler());
requestHandlers.put(McpSchema.METHOD_PROMPT_GET, promptsGetRequestHandler());
}
// Add completion API handlers if the completion capability is enabled
if (this.serverCapabilities.completions() != null) {
requestHandlers.put(McpSchema.METHOD_COMPLETION_COMPLETE, completionCompleteRequestHandler());
}
McpStatelessServerHandler handler = new DefaultMcpStatelessServerHandler(requestHandlers, Map.of());
mcpTransport.setMcpHandler(handler);
}
// ---------------------------------------
// Lifecycle Management
// ---------------------------------------
private McpStatelessRequestHandler<McpSchema.InitializeResult> asyncInitializeRequestHandler() {
return (ctx, req) -> Mono.defer(() -> {
McpSchema.InitializeRequest initializeRequest = this.objectMapper.convertValue(req,
McpSchema.InitializeRequest.class);
logger.info("Client initialize request - Protocol: {}, Capabilities: {}, Info: {}",
initializeRequest.protocolVersion(), initializeRequest.capabilities(),
initializeRequest.clientInfo());
// The server MUST respond with the highest protocol version it supports
// if
// it does not support the requested (e.g. Client) version.
String serverProtocolVersion = this.protocolVersions.get(this.protocolVersions.size() - 1);
if (this.protocolVersions.contains(initializeRequest.protocolVersion())) {
// If the server supports the requested protocol version, it MUST
// respond
// with the same version.
serverProtocolVersion = initializeRequest.protocolVersion();
}
else {
logger.warn(
"Client requested unsupported protocol version: {}, so the server will suggest the {} version instead",
initializeRequest.protocolVersion(), serverProtocolVersion);
}
return Mono.just(new McpSchema.InitializeResult(serverProtocolVersion, this.serverCapabilities,
this.serverInfo, this.instructions));
});
}
/**
* Get the server capabilities that define the supported features and functionality.
* @return The server capabilities
*/
public McpSchema.ServerCapabilities getServerCapabilities() {
return this.serverCapabilities;
}
/**
* Get the server implementation information.
* @return The server implementation details
*/
public McpSchema.Implementation getServerInfo() {
return this.serverInfo;
}
/**
* Gracefully closes the server, allowing any in-progress operations to complete.
* @return A Mono that completes when the server has been closed
*/
public Mono<Void> closeGracefully() {
return this.mcpTransportProvider.closeGracefully();
}
/**
* Close the server immediately.
*/
public void close() {
this.mcpTransportProvider.close();
}
// ---------------------------------------
// Tool Management
// ---------------------------------------
private static List<McpStatelessServerFeatures.AsyncToolSpecification> withStructuredOutputHandling(
JsonSchemaValidator jsonSchemaValidator, List<McpStatelessServerFeatures.AsyncToolSpecification> tools) {
if (Utils.isEmpty(tools)) {
return tools;
}
return tools.stream().map(tool -> withStructuredOutputHandling(jsonSchemaValidator, tool)).toList();
}
private static McpStatelessServerFeatures.AsyncToolSpecification withStructuredOutputHandling(
JsonSchemaValidator jsonSchemaValidator,
McpStatelessServerFeatures.AsyncToolSpecification toolSpecification) {
if (toolSpecification.callHandler() instanceof StructuredOutputCallToolHandler) {
// If the tool is already wrapped, return it as is
return toolSpecification;
}
if (toolSpecification.tool().outputSchema() == null) {
// If the tool does not have an output schema, return it as is
return toolSpecification;
}
return new McpStatelessServerFeatures.AsyncToolSpecification(toolSpecification.tool(),
new StructuredOutputCallToolHandler(jsonSchemaValidator, toolSpecification.tool().outputSchema(),
toolSpecification.callHandler()));
}
private static class StructuredOutputCallToolHandler
implements BiFunction<McpTransportContext, McpSchema.CallToolRequest, Mono<McpSchema.CallToolResult>> {
private final BiFunction<McpTransportContext, McpSchema.CallToolRequest, Mono<McpSchema.CallToolResult>> delegateHandler;
private final JsonSchemaValidator jsonSchemaValidator;
private final Map<String, Object> outputSchema;
public StructuredOutputCallToolHandler(JsonSchemaValidator jsonSchemaValidator,
Map<String, Object> outputSchema,
BiFunction<McpTransportContext, McpSchema.CallToolRequest, Mono<McpSchema.CallToolResult>> delegateHandler) {
Assert.notNull(jsonSchemaValidator, "JsonSchemaValidator must not be null");
Assert.notNull(delegateHandler, "Delegate call tool result handler must not be null");
this.delegateHandler = delegateHandler;
this.outputSchema = outputSchema;
this.jsonSchemaValidator = jsonSchemaValidator;
}
@Override
public Mono<CallToolResult> apply(McpTransportContext transportContext, McpSchema.CallToolRequest request) {
return this.delegateHandler.apply(transportContext, request).map(result -> {
if (outputSchema == null) {
if (result.structuredContent() != null) {
logger.warn(
"Tool call with no outputSchema is not expected to have a result with structured content, but got: {}",
result.structuredContent());
}
// Pass through. No validation is required if no output schema is
// provided.
return result;
}
// If an output schema is provided, servers MUST provide structured
// results that conform to this schema.
// https://modelcontextprotocol.io/specification/2025-06-18/server/tools#output-schema
if (result.structuredContent() == null) {
logger.warn(
"Response missing structured content which is expected when calling tool with non-empty outputSchema");
return new CallToolResult(
"Response missing structured content which is expected when calling tool with non-empty outputSchema",
true);
}
// Validate the result against the output schema
var validation = this.jsonSchemaValidator.validate(outputSchema, result.structuredContent());
if (!validation.valid()) {
logger.warn("Tool call result validation failed: {}", validation.errorMessage());
return new CallToolResult(validation.errorMessage(), true);
}
if (Utils.isEmpty(result.content())) {
// For backwards compatibility, a tool that returns structured
// content SHOULD also return functionally equivalent unstructured
// content. (For example, serialized JSON can be returned in a
// TextContent block.)
// https://modelcontextprotocol.io/specification/2025-06-18/server/tools#structured-content
return new CallToolResult(List.of(new McpSchema.TextContent(validation.jsonStructuredOutput())),
result.isError(), result.structuredContent());
}
return result;
});
}
}
/**
* Add a new tool specification at runtime.
* @param toolSpecification The tool specification to add
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> addTool(McpStatelessServerFeatures.AsyncToolSpecification toolSpecification) {
if (toolSpecification == null) {
return Mono.error(new McpError("Tool specification must not be null"));
}
if (toolSpecification.tool() == null) {
return Mono.error(new McpError("Tool must not be null"));
}
if (toolSpecification.callHandler() == null) {
return Mono.error(new McpError("Tool call handler must not be null"));
}
if (this.serverCapabilities.tools() == null) {
return Mono.error(new McpError("Server must be configured with tool capabilities"));
}
var wrappedToolSpecification = withStructuredOutputHandling(this.jsonSchemaValidator, toolSpecification);
return Mono.defer(() -> {
// Check for duplicate tool names
if (this.tools.stream().anyMatch(th -> th.tool().name().equals(wrappedToolSpecification.tool().name()))) {
return Mono.error(
new McpError("Tool with name '" + wrappedToolSpecification.tool().name() + "' already exists"));
}
this.tools.add(wrappedToolSpecification);
logger.debug("Added tool handler: {}", wrappedToolSpecification.tool().name());
return Mono.empty();
});
}
/**
* Remove a tool handler at runtime.
* @param toolName The name of the tool handler to remove
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> removeTool(String toolName) {
if (toolName == null) {
return Mono.error(new McpError("Tool name must not be null"));
}
if (this.serverCapabilities.tools() == null) {
return Mono.error(new McpError("Server must be configured with tool capabilities"));
}
return Mono.defer(() -> {
boolean removed = this.tools
.removeIf(toolSpecification -> toolSpecification.tool().name().equals(toolName));
if (removed) {
logger.debug("Removed tool handler: {}", toolName);
return Mono.empty();
}
return Mono.error(new McpError("Tool with name '" + toolName + "' not found"));
});
}
private McpStatelessRequestHandler<McpSchema.ListToolsResult> toolsListRequestHandler() {
return (ctx, params) -> {
List<Tool> tools = this.tools.stream()
.map(McpStatelessServerFeatures.AsyncToolSpecification::tool)
.toList();
return Mono.just(new McpSchema.ListToolsResult(tools, null));
};
}
private McpStatelessRequestHandler<CallToolResult> toolsCallRequestHandler() {
return (ctx, params) -> {
McpSchema.CallToolRequest callToolRequest = objectMapper.convertValue(params,
new TypeReference<McpSchema.CallToolRequest>() {
});
Optional<McpStatelessServerFeatures.AsyncToolSpecification> toolSpecification = this.tools.stream()
.filter(tr -> callToolRequest.name().equals(tr.tool().name()))
.findAny();
if (toolSpecification.isEmpty()) {
return Mono.error(new McpError("Tool not found: " + callToolRequest.name()));
}
return toolSpecification.map(tool -> tool.callHandler().apply(ctx, callToolRequest))
.orElse(Mono.error(new McpError("Tool not found: " + callToolRequest.name())));
};
}
// ---------------------------------------
// Resource Management
// ---------------------------------------
/**
* Add a new resource handler at runtime.
* @param resourceSpecification The resource handler to add
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> addResource(McpStatelessServerFeatures.AsyncResourceSpecification resourceSpecification) {
if (resourceSpecification == null || resourceSpecification.resource() == null) {
return Mono.error(new McpError("Resource must not be null"));
}
if (this.serverCapabilities.resources() == null) {
return Mono.error(new McpError("Server must be configured with resource capabilities"));
}
return Mono.defer(() -> {
if (this.resources.putIfAbsent(resourceSpecification.resource().uri(), resourceSpecification) != null) {
return Mono.error(new McpError(
"Resource with URI '" + resourceSpecification.resource().uri() + "' already exists"));
}
logger.debug("Added resource handler: {}", resourceSpecification.resource().uri());
return Mono.empty();
});
}
/**
* Remove a resource handler at runtime.
* @param resourceUri The URI of the resource handler to remove
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> removeResource(String resourceUri) {
if (resourceUri == null) {
return Mono.error(new McpError("Resource URI must not be null"));
}
if (this.serverCapabilities.resources() == null) {
return Mono.error(new McpError("Server must be configured with resource capabilities"));
}
return Mono.defer(() -> {
McpStatelessServerFeatures.AsyncResourceSpecification removed = this.resources.remove(resourceUri);
if (removed != null) {
logger.debug("Removed resource handler: {}", resourceUri);
return Mono.empty();
}
return Mono.error(new McpError("Resource with URI '" + resourceUri + "' not found"));
});
}
private McpStatelessRequestHandler<McpSchema.ListResourcesResult> resourcesListRequestHandler() {
return (ctx, params) -> {
var resourceList = this.resources.values()
.stream()
.map(McpStatelessServerFeatures.AsyncResourceSpecification::resource)
.toList();
return Mono.just(new McpSchema.ListResourcesResult(resourceList, null));
};
}
private McpStatelessRequestHandler<McpSchema.ListResourceTemplatesResult> resourceTemplateListRequestHandler() {
return (ctx, params) -> Mono.just(new McpSchema.ListResourceTemplatesResult(this.getResourceTemplates(), null));
}
private List<ResourceTemplate> getResourceTemplates() {
var list = new ArrayList<>(this.resourceTemplates);
List<ResourceTemplate> resourceTemplates = this.resources.keySet()
.stream()
.filter(uri -> uri.contains("{"))
.map(uri -> {
var resource = this.resources.get(uri).resource();
var template = new ResourceTemplate(resource.uri(), resource.name(), resource.title(),
resource.description(), resource.mimeType(), resource.annotations());
return template;
})
.toList();
list.addAll(resourceTemplates);
return list;
}
private McpStatelessRequestHandler<McpSchema.ReadResourceResult> resourcesReadRequestHandler() {
return (ctx, params) -> {
McpSchema.ReadResourceRequest resourceRequest = objectMapper.convertValue(params,
new TypeReference<McpSchema.ReadResourceRequest>() {
});
var resourceUri = resourceRequest.uri();
McpStatelessServerFeatures.AsyncResourceSpecification specification = this.resources.values()
.stream()
.filter(resourceSpecification -> this.uriTemplateManagerFactory
.create(resourceSpecification.resource().uri())
.matches(resourceUri))
.findFirst()
.orElseThrow(() -> new McpError("Resource not found: " + resourceUri));
return specification.readHandler().apply(ctx, resourceRequest);
};
}
// ---------------------------------------
// Prompt Management
// ---------------------------------------
/**
* Add a new prompt handler at runtime.
* @param promptSpecification The prompt handler to add
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> addPrompt(McpStatelessServerFeatures.AsyncPromptSpecification promptSpecification) {
if (promptSpecification == null) {
return Mono.error(new McpError("Prompt specification must not be null"));
}
if (this.serverCapabilities.prompts() == null) {
return Mono.error(new McpError("Server must be configured with prompt capabilities"));
}
return Mono.defer(() -> {
McpStatelessServerFeatures.AsyncPromptSpecification specification = this.prompts
.putIfAbsent(promptSpecification.prompt().name(), promptSpecification);
if (specification != null) {
return Mono.error(
new McpError("Prompt with name '" + promptSpecification.prompt().name() + "' already exists"));
}
logger.debug("Added prompt handler: {}", promptSpecification.prompt().name());
return Mono.empty();
});
}
/**
* Remove a prompt handler at runtime.
* @param promptName The name of the prompt handler to remove
* @return Mono that completes when clients have been notified of the change
*/
public Mono<Void> removePrompt(String promptName) {
if (promptName == null) {
return Mono.error(new McpError("Prompt name must not be null"));
}
if (this.serverCapabilities.prompts() == null) {
return Mono.error(new McpError("Server must be configured with prompt capabilities"));
}
return Mono.defer(() -> {
McpStatelessServerFeatures.AsyncPromptSpecification removed = this.prompts.remove(promptName);
if (removed != null) {
logger.debug("Removed prompt handler: {}", promptName);
return Mono.empty();
}
return Mono.error(new McpError("Prompt with name '" + promptName + "' not found"));
});
}
private McpStatelessRequestHandler<McpSchema.ListPromptsResult> promptsListRequestHandler() {
return (ctx, params) -> {
// TODO: Implement pagination
// McpSchema.PaginatedRequest request = objectMapper.convertValue(params,
// new TypeReference<McpSchema.PaginatedRequest>() {
// });
var promptList = this.prompts.values()
.stream()
.map(McpStatelessServerFeatures.AsyncPromptSpecification::prompt)
.toList();
return Mono.just(new McpSchema.ListPromptsResult(promptList, null));
};
}
private McpStatelessRequestHandler<McpSchema.GetPromptResult> promptsGetRequestHandler() {
return (ctx, params) -> {
McpSchema.GetPromptRequest promptRequest = objectMapper.convertValue(params,
new TypeReference<McpSchema.GetPromptRequest>() {
});
// Implement prompt retrieval logic here
McpStatelessServerFeatures.AsyncPromptSpecification specification = this.prompts.get(promptRequest.name());
if (specification == null) {
return Mono.error(new McpError("Prompt not found: " + promptRequest.name()));
}
return specification.promptHandler().apply(ctx, promptRequest);
};
}
private McpStatelessRequestHandler<McpSchema.CompleteResult> completionCompleteRequestHandler() {
return (ctx, params) -> {
McpSchema.CompleteRequest request = parseCompletionParams(params);
if (request.ref() == null) {
return Mono.error(new McpError("ref must not be null"));
}
if (request.ref().type() == null) {
return Mono.error(new McpError("type must not be null"));
}
String type = request.ref().type();
String argumentName = request.argument().name();
// check if the referenced resource exists
if (type.equals("ref/prompt") && request.ref() instanceof McpSchema.PromptReference promptReference) {
McpStatelessServerFeatures.AsyncPromptSpecification promptSpec = this.prompts
.get(promptReference.name());
if (promptSpec == null) {
return Mono.error(new McpError("Prompt not found: " + promptReference.name()));
}
if (promptSpec.prompt().arguments().stream().noneMatch(arg -> arg.name().equals(argumentName))) {
return Mono.error(new McpError("Argument not found: " + argumentName));
}
}
if (type.equals("ref/resource") && request.ref() instanceof McpSchema.ResourceReference resourceReference) {
McpStatelessServerFeatures.AsyncResourceSpecification resourceSpec = this.resources
.get(resourceReference.uri());
if (resourceSpec == null) {
return Mono.error(new McpError("Resource not found: " + resourceReference.uri()));
}
if (!uriTemplateManagerFactory.create(resourceSpec.resource().uri())
.getVariableNames()
.contains(argumentName)) {
return Mono.error(new McpError("Argument not found: " + argumentName));
}
}
McpStatelessServerFeatures.AsyncCompletionSpecification specification = this.completions.get(request.ref());
if (specification == null) {
return Mono.error(new McpError("AsyncCompletionSpecification not found: " + request.ref()));
}
return specification.completionHandler().apply(ctx, request);
};
}
/**
* Parses the raw JSON-RPC request parameters into a {@link McpSchema.CompleteRequest}
* object.
* <p>
* This method manually extracts the `ref` and `argument` fields from the input map,
* determines the correct reference type (either prompt or resource), and constructs a
* fully-typed {@code CompleteRequest} instance.
* @param object the raw request parameters, expected to be a Map containing "ref" and
* "argument" entries.
* @return a {@link McpSchema.CompleteRequest} representing the structured completion
* request.
* @throws IllegalArgumentException if the "ref" type is not recognized.
*/
@SuppressWarnings("unchecked")
private McpSchema.CompleteRequest parseCompletionParams(Object object) {
Map<String, Object> params = (Map<String, Object>) object;
Map<String, Object> refMap = (Map<String, Object>) params.get("ref");
Map<String, Object> argMap = (Map<String, Object>) params.get("argument");
String refType = (String) refMap.get("type");
McpSchema.CompleteReference ref = switch (refType) {
case "ref/prompt" -> new McpSchema.PromptReference(refType, (String) refMap.get("name"),
refMap.get("title") != null ? (String) refMap.get("title") : null);
case "ref/resource" -> new McpSchema.ResourceReference(refType, (String) refMap.get("uri"));
default -> throw new IllegalArgumentException("Invalid ref type: " + refType);
};
String argName = (String) argMap.get("name");
String argValue = (String) argMap.get("value");
McpSchema.CompleteRequest.CompleteArgument argument = new McpSchema.CompleteRequest.CompleteArgument(argName,
argValue);
return new McpSchema.CompleteRequest(ref, argument);
}
/**
* This method is package-private and used for test only. Should not be called by user
* code.
* @param protocolVersions the Client supported protocol versions.
*/
void setProtocolVersions(List<String> protocolVersions) {
this.protocolVersions = protocolVersions;
}
}

20
mcp/src/main/java/io/modelcontextprotocol/server/McpStatelessNotificationHandler.java Normal file
View file

@ -0,0 +1,20 @@
package io.modelcontextprotocol.server;
import reactor.core.publisher.Mono;
/**
* Handler for MCP notifications in a stateless server.
*
* @author Dariusz Jędrzejczyk
*/
public interface McpStatelessNotificationHandler {
/**
* Handle to notification and complete once done.
* @param transportContext {@link McpTransportContext} associated with the transport
* @param params the payload of the MCP notification
* @return Mono which completes once the processing is done
*/
Mono<Void> handle(McpTransportContext transportContext, Object params);
}

21
mcp/src/main/java/io/modelcontextprotocol/server/McpStatelessRequestHandler.java Normal file
View file

@ -0,0 +1,21 @@
package io.modelcontextprotocol.server;
import reactor.core.publisher.Mono;
/**
* Handler for MCP requests in a stateless server.
*
* @param <R> type of the MCP response
* @author Dariusz Jędrzejczyk
*/
public interface McpStatelessRequestHandler<R> {
/**
* Handle the request and complete with a result.
* @param transportContext {@link McpTransportContext} associated with the transport
* @param params the payload of the MCP request
* @return Mono which completes with the response object
*/
Mono<R> handle(McpTransportContext transportContext, Object params);
}

379
mcp/src/main/java/io/modelcontextprotocol/server/McpStatelessServerFeatures.java Normal file
View file

@ -0,0 +1,379 @@
/*
* Copyright 2024-2025 the original author or authors.
*/
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.spec.McpSchema;
import io.modelcontextprotocol.spec.McpSchema.CallToolRequest;
import io.modelcontextprotocol.util.Assert;
import io.modelcontextprotocol.util.Utils;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
/**
* MCP stateless server features specification that a particular server can choose to
* support.
*
* @author Dariusz Jędrzejczyk
*/
public class McpStatelessServerFeatures {
/**
* Asynchronous server features specification.
*
* @param serverInfo The server implementation details
* @param serverCapabilities The server capabilities
* @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 specifications
* @param instructions The server instructions text
*/
record Async(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities serverCapabilities,
List<McpStatelessServerFeatures.AsyncToolSpecification> tools,
Map<String, AsyncResourceSpecification> resources, List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, McpStatelessServerFeatures.AsyncPromptSpecification> prompts,
Map<McpSchema.CompleteReference, McpStatelessServerFeatures.AsyncCompletionSpecification> completions,
String instructions) {
/**
* Create an instance and validate the arguments.
* @param serverInfo The server implementation details
* @param serverCapabilities The server capabilities
* @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 specifications
* @param instructions The server instructions text
*/
Async(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities serverCapabilities,
List<McpStatelessServerFeatures.AsyncToolSpecification> tools,
Map<String, AsyncResourceSpecification> resources, List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, McpStatelessServerFeatures.AsyncPromptSpecification> prompts,
Map<McpSchema.CompleteReference, McpStatelessServerFeatures.AsyncCompletionSpecification> completions,
String instructions) {
Assert.notNull(serverInfo, "Server info must not be null");
this.serverInfo = serverInfo;
this.serverCapabilities = (serverCapabilities != null) ? serverCapabilities
: new McpSchema.ServerCapabilities(null, // completions
null, // experimental
new McpSchema.ServerCapabilities.LoggingCapabilities(), // Enable
// logging
// by
// default
!Utils.isEmpty(prompts) ? new McpSchema.ServerCapabilities.PromptCapabilities(false) : null,
!Utils.isEmpty(resources)
? new McpSchema.ServerCapabilities.ResourceCapabilities(false, false) : null,
!Utils.isEmpty(tools) ? new McpSchema.ServerCapabilities.ToolCapabilities(false) : null);
this.tools = (tools != null) ? tools : List.of();
this.resources = (resources != null) ? resources : Map.of();
this.resourceTemplates = (resourceTemplates != null) ? resourceTemplates : List.of();
this.prompts = (prompts != null) ? prompts : Map.of();
this.completions = (completions != null) ? completions : Map.of();
this.instructions = instructions;
}
/**
* Convert a synchronous specification into an asynchronous one and provide
* blocking code offloading to prevent accidental blocking of the non-blocking
* transport.
* @param syncSpec a potentially blocking, synchronous specification.
* @param immediateExecution when true, do not offload. Do NOT set to true when
* using a non-blocking transport.
* @return a specification which is protected from blocking calls specified by the
* user.
*/
static Async fromSync(Sync syncSpec, boolean immediateExecution) {
List<McpStatelessServerFeatures.AsyncToolSpecification> tools = new ArrayList<>();
for (var tool : syncSpec.tools()) {
tools.add(AsyncToolSpecification.fromSync(tool, immediateExecution));
}
Map<String, AsyncResourceSpecification> resources = new HashMap<>();
syncSpec.resources().forEach((key, resource) -> {
resources.put(key, AsyncResourceSpecification.fromSync(resource, immediateExecution));
});
Map<String, AsyncPromptSpecification> prompts = new HashMap<>();
syncSpec.prompts().forEach((key, prompt) -> {
prompts.put(key, AsyncPromptSpecification.fromSync(prompt, immediateExecution));
});
Map<McpSchema.CompleteReference, McpStatelessServerFeatures.AsyncCompletionSpecification> completions = new HashMap<>();
syncSpec.completions().forEach((key, completion) -> {
completions.put(key, AsyncCompletionSpecification.fromSync(completion, immediateExecution));
});
return new Async(syncSpec.serverInfo(), syncSpec.serverCapabilities(), tools, resources,
syncSpec.resourceTemplates(), prompts, completions, syncSpec.instructions());
}
}
/**
* Synchronous server features specification.
*
* @param serverInfo The server implementation details
* @param serverCapabilities The server capabilities
* @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 specifications
* @param instructions The server instructions text
*/
record Sync(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities serverCapabilities,
List<McpStatelessServerFeatures.SyncToolSpecification> tools,
Map<String, McpStatelessServerFeatures.SyncResourceSpecification> resources,
List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, McpStatelessServerFeatures.SyncPromptSpecification> prompts,
Map<McpSchema.CompleteReference, McpStatelessServerFeatures.SyncCompletionSpecification> completions,
String instructions) {
/**
* Create an instance and validate the arguments.
* @param serverInfo The server implementation details
* @param serverCapabilities The server capabilities
* @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 specifications
* @param instructions The server instructions text
*/
Sync(McpSchema.Implementation serverInfo, McpSchema.ServerCapabilities serverCapabilities,
List<McpStatelessServerFeatures.SyncToolSpecification> tools,
Map<String, McpStatelessServerFeatures.SyncResourceSpecification> resources,
List<McpSchema.ResourceTemplate> resourceTemplates,
Map<String, McpStatelessServerFeatures.SyncPromptSpecification> prompts,
Map<McpSchema.CompleteReference, McpStatelessServerFeatures.SyncCompletionSpecification> completions,
String instructions) {
Assert.notNull(serverInfo, "Server info must not be null");
this.serverInfo = serverInfo;
this.serverCapabilities = (serverCapabilities != null) ? serverCapabilities
: new McpSchema.ServerCapabilities(null, // completions
null, // experimental
new McpSchema.ServerCapabilities.LoggingCapabilities(), // Enable
// logging
// by
// default
!Utils.isEmpty(prompts) ? new McpSchema.ServerCapabilities.PromptCapabilities(false) : null,
!Utils.isEmpty(resources)
? new McpSchema.ServerCapabilities.ResourceCapabilities(false, false) : null,
!Utils.isEmpty(tools) ? new McpSchema.ServerCapabilities.ToolCapabilities(false) : null);
this.tools = (tools != null) ? tools : new ArrayList<>();
this.resources = (resources != null) ? resources : new HashMap<>();
this.resourceTemplates = (resourceTemplates != null) ? resourceTemplates : new ArrayList<>();
this.prompts = (prompts != null) ? prompts : new HashMap<>();
this.completions = (completions != null) ? completions : new HashMap<>();
this.instructions = instructions;
}
}
/**
* 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.
*
* @param tool The tool definition including name, description, and parameter schema
* @param callHandler The function that implements the tool's logic, receiving a
* {@link CallToolRequest} and returning the result.
*/
public record AsyncToolSpecification(McpSchema.Tool tool,
BiFunction<McpTransportContext, CallToolRequest, Mono<McpSchema.CallToolResult>> callHandler) {
static AsyncToolSpecification fromSync(SyncToolSpecification syncToolSpec) {
return fromSync(syncToolSpec, false);
}
static AsyncToolSpecification fromSync(SyncToolSpecification syncToolSpec, boolean immediate) {
// FIXME: This is temporary, proper validation should be implemented
if (syncToolSpec == null) {
return null;
}
BiFunction<McpTransportContext, CallToolRequest, Mono<McpSchema.CallToolResult>> callHandler = (ctx,
req) -> {
var toolResult = Mono.fromCallable(() -> syncToolSpec.callHandler().apply(ctx, req));
return immediate ? toolResult : toolResult.subscribeOn(Schedulers.boundedElastic());
};
return new AsyncToolSpecification(syncToolSpec.tool(), callHandler);
}
}
/**
* 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>
*
* @param resource The resource definition including name, description, and MIME type
* @param readHandler The function that handles resource read requests. The function's
* argument is a {@link McpSchema.ReadResourceRequest}.
*/
public record AsyncResourceSpecification(McpSchema.Resource resource,
BiFunction<McpTransportContext, McpSchema.ReadResourceRequest, Mono<McpSchema.ReadResourceResult>> readHandler) {
static AsyncResourceSpecification fromSync(SyncResourceSpecification resource, boolean immediateExecution) {
// FIXME: This is temporary, proper validation should be implemented
if (resource == null) {
return null;
}
return new AsyncResourceSpecification(resource.resource(), (ctx, req) -> {
var resourceResult = Mono.fromCallable(() -> resource.readHandler().apply(ctx, req));
return immediateExecution ? resourceResult : resourceResult.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>
*
* @param prompt The prompt definition including name and description
* @param promptHandler The function that processes prompt requests and returns
* formatted templates. The function's argument is a
* {@link McpSchema.GetPromptRequest}.
*/
public record AsyncPromptSpecification(McpSchema.Prompt prompt,
BiFunction<McpTransportContext, McpSchema.GetPromptRequest, Mono<McpSchema.GetPromptResult>> promptHandler) {
static AsyncPromptSpecification fromSync(SyncPromptSpecification prompt, boolean immediateExecution) {
// FIXME: This is temporary, proper validation should be implemented
if (prompt == null) {
return null;
}
return new AsyncPromptSpecification(prompt.prompt(), (ctx, req) -> {
var promptResult = Mono.fromCallable(() -> prompt.promptHandler().apply(ctx, req));
return immediateExecution ? promptResult : promptResult.subscribeOn(Schedulers.boundedElastic());
});
}
}
/**
* Specification of a completion handler function with asynchronous execution support.
* Completions generate AI model outputs based on prompt or resource references and
* user-provided arguments. This abstraction enables:
* <ul>
* <li>Customizable response generation logic
* <li>Parameter-driven template expansion
* <li>Dynamic interaction with connected clients
* </ul>
*
* @param referenceKey The unique key representing the completion reference.
* @param completionHandler The asynchronous function that processes completion
* requests and returns results. The function's argument is a
* {@link McpSchema.CompleteRequest}.
*/
public record AsyncCompletionSpecification(McpSchema.CompleteReference referenceKey,
BiFunction<McpTransportContext, McpSchema.CompleteRequest, Mono<McpSchema.CompleteResult>> completionHandler) {
/**
* Converts a synchronous {@link SyncCompletionSpecification} into an
* {@link AsyncCompletionSpecification} by wrapping the handler in a bounded
* elastic scheduler for safe non-blocking execution.
* @param completion the synchronous completion specification
* @return an asynchronous wrapper of the provided sync specification, or
* {@code null} if input is null
*/
static AsyncCompletionSpecification fromSync(SyncCompletionSpecification completion,
boolean immediateExecution) {
if (completion == null) {
return null;
}
return new AsyncCompletionSpecification(completion.referenceKey(), (ctx, req) -> {
var completionResult = Mono.fromCallable(() -> completion.completionHandler().apply(ctx, req));
return immediateExecution ? completionResult
: completionResult.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.
*
* @param tool The tool definition including name, description, and parameter schema
* @param callHandler The function that implements the tool's logic, receiving a
* {@link CallToolRequest} and returning results.
*/
public record SyncToolSpecification(McpSchema.Tool tool,
BiFunction<McpTransportContext, CallToolRequest, McpSchema.CallToolResult> callHandler) {
}
/**
* 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>
*
* @param resource The resource definition including name, description, and MIME type
* @param readHandler The function that handles resource read requests. The function's
* argument is a {@link McpSchema.ReadResourceRequest}.
*/
public record SyncResourceSpecification(McpSchema.Resource resource,
BiFunction<McpTransportContext, 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>
*
* @param prompt The prompt definition including name and description
* @param promptHandler The function that processes prompt requests and returns
* formatted templates. The function's argument is a
* {@link McpSchema.GetPromptRequest}.
*/
public record SyncPromptSpecification(McpSchema.Prompt prompt,
BiFunction<McpTransportContext, McpSchema.GetPromptRequest, McpSchema.GetPromptResult> promptHandler) {
}
/**
* Specification of a completion handler function with synchronous execution support.
*
* @param referenceKey The unique key representing the completion reference.
* @param completionHandler The synchronous function that processes completion
* requests and returns results. The argument is a {@link McpSchema.CompleteRequest}.
*/
public record SyncCompletionSpecification(McpSchema.CompleteReference referenceKey,
BiFunction<McpTransportContext, McpSchema.CompleteRequest, McpSchema.CompleteResult> completionHandler) {
}
}

33
mcp/src/main/java/io/modelcontextprotocol/server/McpStatelessServerHandler.java Normal file
View file

@ -0,0 +1,33 @@
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.spec.McpSchema;
import reactor.core.publisher.Mono;
/**
* Handler for MCP requests and notifications in a Stateless Streamable HTTP Server
* context.
*
* @author Dariusz Jędrzejczyk
*/
public interface McpStatelessServerHandler {
/**
* Handle the request using user-provided feature implementations.
* @param transportContext {@link McpTransportContext} carrying transport layer
* metadata
* @param request the request JSON object
* @return Mono containing the JSON response
*/
Mono<McpSchema.JSONRPCResponse> handleRequest(McpTransportContext transportContext,
McpSchema.JSONRPCRequest request);
/**
* Handle the notification.
* @param transportContext {@link McpTransportContext} carrying transport layer
* metadata
* @param notification the notification JSON object
* @return Mono that completes once handling is finished
*/
Mono<Void> handleNotification(McpTransportContext transportContext, McpSchema.JSONRPCNotification notification);
}

132
mcp/src/main/java/io/modelcontextprotocol/server/McpStatelessSyncServer.java Normal file
View file

@ -0,0 +1,132 @@
/*
* Copyright 2024-2024 the original author or authors.
*/
package io.modelcontextprotocol.server;
import io.modelcontextprotocol.spec.McpSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import java.util.List;
/**
* A stateless MCP server implementation for use with Streamable HTTP transport types. It
* allows simple horizontal scalability since it does not maintain a session and does not
* require initialization. Each instance of the server can be reached with no prior
* knowledge and can serve the clients with the capabilities it supports.
*
* @author Dariusz Jędrzejczyk
*/
public class McpStatelessSyncServer {
private static final Logger logger = LoggerFactory.getLogger(McpStatelessSyncServer.class);
private final McpStatelessAsyncServer asyncServer;
private final boolean immediateExecution;
McpStatelessSyncServer(McpStatelessAsyncServer asyncServer, boolean immediateExecution) {
this.asyncServer = asyncServer;
this.immediateExecution = immediateExecution;
}
/**
* Get the server capabilities that define the supported features and functionality.
* @return The server capabilities
*/
public McpSchema.ServerCapabilities getServerCapabilities() {
return this.asyncServer.getServerCapabilities();
}
/**
* Get the server implementation information.
* @return The server implementation details
*/
public McpSchema.Implementation getServerInfo() {
return this.asyncServer.getServerInfo();
}
/**
* Gracefully closes the server, allowing any in-progress operations to complete.
* @return A Mono that completes when the server has been closed
*/
public Mono<Void> closeGracefully() {
return this.asyncServer.closeGracefully();
}
/**
* Close the server immediately.
*/
public void close() {
this.asyncServer.close();
}
/**
* Add a new tool specification at runtime.
* @param toolSpecification The tool specification to add
*/
public void addTool(McpStatelessServerFeatures.SyncToolSpecification toolSpecification) {
this.asyncServer
.addTool(McpStatelessServerFeatures.AsyncToolSpecification.fromSync(toolSpecification,
this.immediateExecution))
.block();
}
/**
* Remove a tool handler at runtime.
* @param toolName The name of the tool handler to remove
*/
public void removeTool(String toolName) {
this.asyncServer.removeTool(toolName).block();
}
/**
* Add a new resource handler at runtime.
* @param resourceSpecification The resource handler to add
*/
public void addResource(McpStatelessServerFeatures.SyncResourceSpecification resourceSpecification) {
this.asyncServer
.addResource(McpStatelessServerFeatures.AsyncResourceSpecification.fromSync(resourceSpecification,
this.immediateExecution))
.block();
}
/**
* Remove a resource handler at runtime.
* @param resourceUri The URI of the resource handler to remove
*/
public void removeResource(String resourceUri) {
this.asyncServer.removeResource(resourceUri).block();
}
/**
* Add a new prompt handler at runtime.
* @param promptSpecification The prompt handler to add
*/
public void addPrompt(McpStatelessServerFeatures.SyncPromptSpecification promptSpecification) {
this.asyncServer
.addPrompt(McpStatelessServerFeatures.AsyncPromptSpecification.fromSync(promptSpecification,
this.immediateExecution))
.block();
}
/**
* Remove a prompt handler at runtime.
* @param promptName The name of the prompt handler to remove
*/
public void removePrompt(String promptName) {
this.asyncServer.removePrompt(promptName).block();
}
/**
* This method is package-private and used for test only. Should not be called by user
* code.
* @param protocolVersions the Client supported protocol versions.
*/
void setProtocolVersions(List<String> protocolVersions) {
this.asyncServer.setProtocolVersions(protocolVersions);
}
}

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

@ -27,6 +27,14 @@ public class McpSyncServerExchange {
this.exchange = exchange;
}
/**
* Provides the Session ID
* @return session ID
*/
public String sessionId() {
return this.exchange.sessionId();
}
/**
* Get the client capabilities that define the supported features and functionality.
* @return The client capabilities
@ -43,6 +51,16 @@ public class McpSyncServerExchange {
return this.exchange.getClientInfo();
}
/**
* Provides the {@link McpTransportContext} associated with the transport layer. For
* HTTP transports it can contain the metadata associated with the HTTP request that
* triggered the processing.
* @return the transport context object
*/
public McpTransportContext transportContext() {
return this.exchange.transportContext();
}
/**
* 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

46
mcp/src/main/java/io/modelcontextprotocol/server/McpTransportContext.java Normal file
View file

@ -0,0 +1,46 @@
package io.modelcontextprotocol.server;
import java.util.Collections;
/**
* Context associated with the transport layer. It allows to add transport-level metadata
* for use further down the line. Specifically, it can be beneficial to extract HTTP
* request metadata for use in MCP feature implementations.
*
* @author Dariusz Jędrzejczyk
*/
public interface McpTransportContext {
/**
* Key for use in Reactor Context to transport the context to user land.
*/
String KEY = "MCP_TRANSPORT_CONTEXT";
/**
* An empty, unmodifiable context.
*/
@SuppressWarnings("unchecked")
McpTransportContext EMPTY = new DefaultMcpTransportContext(Collections.EMPTY_MAP);
/**
* Extract a value from the context.
* @param key the key under the data is expected
* @return the associated value or {@code null} if missing.
*/
Object get(String key);
/**
* Inserts a value for a given key.
* @param key a String representing the key
* @param value the value to store
*/
void put(String key, Object value);
/**
* Copies the contents of the context to allow further modifications without affecting
* the initial object.
* @return a new instance with the underlying storage copied.
*/
McpTransportContext copy();
}

24
mcp/src/main/java/io/modelcontextprotocol/server/McpTransportContextExtractor.java Normal file
View file

@ -0,0 +1,24 @@
package io.modelcontextprotocol.server;
/**
* The contract for extracting metadata from a generic transport request of type
* {@link T}.
*
* @param <T> transport-specific representation of the request which allows extracting
* metadata for use in the MCP features implementations.
* @author Dariusz Jędrzejczyk
*/
public interface McpTransportContextExtractor<T> {
/**
* Given an empty context, provides the means to fill it with transport-specific
* metadata extracted from the request.
* @param request the generic representation for the request in the context of a
* specific transport implementation
* @param transportContext the mutable context which can be filled in with metadata
* @return the context filled in with metadata. It can be the same instance as
* provided or a new one.
*/
McpTransportContext extract(T request, McpTransportContext transportContext);
}

52
mcp/src/main/java/io/modelcontextprotocol/spec/DefaultMcpStreamableServerSessionFactory.java Normal file
View file

@ -0,0 +1,52 @@
package io.modelcontextprotocol.spec;
import io.modelcontextprotocol.server.McpNotificationHandler;
import io.modelcontextprotocol.server.McpRequestHandler;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.Map;
import java.util.UUID;
/**
* A default implementation of {@link McpStreamableServerSession.Factory}.
*
* @author Dariusz Jędrzejczyk
*/
public class DefaultMcpStreamableServerSessionFactory implements McpStreamableServerSession.Factory {
Duration requestTimeout;
McpStreamableServerSession.InitRequestHandler initRequestHandler;
Map<String, McpRequestHandler<?>> requestHandlers;
Map<String, McpNotificationHandler> notificationHandlers;
/**
* Constructs an instance
* @param requestTimeout timeout for requests
* @param initRequestHandler initialization request handler
* @param requestHandlers map of MCP request handlers keyed by method name
* @param notificationHandlers map of MCP notification handlers keyed by method name
*/
public DefaultMcpStreamableServerSessionFactory(Duration requestTimeout,
McpStreamableServerSession.InitRequestHandler initRequestHandler,
Map<String, McpRequestHandler<?>> requestHandlers,
Map<String, McpNotificationHandler> notificationHandlers) {
this.requestTimeout = requestTimeout;
this.initRequestHandler = initRequestHandler;
this.requestHandlers = requestHandlers;
this.notificationHandlers = notificationHandlers;
}
@Override
public McpStreamableServerSession.McpStreamableServerSessionInit startSession(
McpSchema.InitializeRequest initializeRequest) {
return new McpStreamableServerSession.McpStreamableServerSessionInit(
new McpStreamableServerSession(UUID.randomUUID().toString(), initializeRequest.capabilities(),
initializeRequest.clientInfo(), requestTimeout, requestHandlers, notificationHandlers),
this.initRequestHandler.handle(initializeRequest));
}
}

20
mcp/src/main/java/io/modelcontextprotocol/spec/HttpHeaders.java Normal file
View file

@ -0,0 +1,20 @@
package io.modelcontextprotocol.spec;
/**
* Names of HTTP headers in use by MCP HTTP transports.
*
* @author Dariusz Jędrzejczyk
*/
public interface HttpHeaders {
/**
* Identifies individual MCP sessions.
*/
String MCP_SESSION_ID = "mcp-session-id";
/**
* Identifies events within an SSE Stream.
*/
String LAST_EVENT_ID = "last-event-id";
}

25
mcp/src/main/java/io/modelcontextprotocol/spec/McpLoggableSession.java Normal file
View file

@ -0,0 +1,25 @@
package io.modelcontextprotocol.spec;
/**
* An {@link McpSession} which is capable of processing logging notifications and keeping
* track of a min logging level.
*
* @author Dariusz Jędrzejczyk
*/
public interface McpLoggableSession extends McpSession {
/**
* Set the minimum logging level for the client. Messages below this level will be
* filtered out.
* @param minLoggingLevel The minimum logging level
*/
void setMinLoggingLevel(McpSchema.LoggingLevel minLoggingLevel);
/**
* Allows checking whether a particular logging level is allowed.
* @param loggingLevel the level to check
* @return whether the logging at the specified level is permitted.
*/
boolean isNotificationForLevelAllowed(McpSchema.LoggingLevel loggingLevel);
}

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