feat: validate embedded JSON Schema documents against 2020-12 meta-schema (SEP-1613) (#949)
MCP embeds JSON Schema documents in three places — Tool.inputSchema, Tool.outputSchema, and ElicitRequest.requestedSchema — and SEP-1613 mandates that these documents conform to JSON Schema 2020-12 by default. Servers now reject malformed schemas at both build time (McpServer.build()) and runtime (McpAsyncServer/McpStatelessAsyncServer.addTool()), returning an IllegalArgumentException that identifies the offending field and references SEP-1613. Elicitation requests whose requestedSchema violates the meta-schema are rejected before being sent to the client. Schemas that explicitly declare a different dialect via $schema are accepted without meta-schema validation, since 2020-12 is the default, not the only permitted dialect. Refine meta-schema loading and deprecate orphan JsonSchemaValidator Resolves #700 Signed-off-by: Dariusz Jędrzejczyk <2554306+chemicL@users.noreply.github.com>
This commit is contained in:
Dariusz Jędrzejczyk
GitHub
parent
5d4a1cce42
commit
87e2c7d4de
10 changed files with 903 additions and 35 deletions
|
|
@ -13,6 +13,29 @@ import java.util.Map;
|
|||
*/
|
||||
public interface JsonSchemaValidator {
|
||||
|
||||
/**
|
||||
* Asserts that the given schema document is a structurally valid JSON Schema. Schemas
|
||||
* without an explicit {@code $schema} declaration, or those that declare JSON Schema
|
||||
* 2020-12, are validated against the 2020-12 meta-schema. Schemas that explicitly
|
||||
* declare a different dialect are accepted without meta-schema validation. Throws
|
||||
* {@link IllegalArgumentException} if validation fails. Silently returns on null
|
||||
* schema. The default implementation delegates to {@link #validateSchema}.
|
||||
* @param context human-readable description of the schema's location (used in error
|
||||
* messages)
|
||||
* @param schema the schema document to validate, or {@code null} (no-op)
|
||||
* @throws IllegalArgumentException if the schema is structurally invalid
|
||||
*/
|
||||
default void assertConforms(String context, Map<String, Object> schema) {
|
||||
if (schema == null) {
|
||||
return;
|
||||
}
|
||||
var result = validateSchema(schema);
|
||||
if (!result.valid()) {
|
||||
throw new IllegalArgumentException(
|
||||
context + " is not a valid JSON Schema 2020-12 document (SEP-1613): " + result.errorMessage());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Represents the result of a validation operation.
|
||||
*
|
||||
|
|
@ -41,4 +64,15 @@ public interface JsonSchemaValidator {
|
|||
*/
|
||||
ValidationResponse validate(Map<String, Object> schema, Object structuredContent);
|
||||
|
||||
/**
|
||||
* Validates that the given schema document itself conforms to JSON Schema 2020-12
|
||||
* (SEP-1613). Schemas that declare an explicit non-2020-12 {@code $schema} dialect
|
||||
* are skipped and considered valid. The default implementation is a no-op.
|
||||
* @param schema the schema document to check
|
||||
* @return a ValidationResponse indicating conformance
|
||||
*/
|
||||
default ValidationResponse validateSchema(Map<String, Object> schema) {
|
||||
return ValidationResponse.asValid(null);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -156,7 +156,8 @@ public class McpAsyncServer {
|
|||
mcpTransportProvider.setSessionFactory(transport -> {
|
||||
String sessionId = UUID.randomUUID().toString();
|
||||
return new McpServerSession(sessionId, requestTimeout, transport, this::asyncInitializeRequestHandler,
|
||||
requestHandlers, notificationHandlers, () -> this.cleanupForSession(sessionId));
|
||||
requestHandlers, notificationHandlers, () -> this.cleanupForSession(sessionId),
|
||||
this.jsonSchemaValidator);
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -183,9 +184,9 @@ public class McpAsyncServer {
|
|||
|
||||
this.protocolVersions = mcpTransportProvider.protocolVersions();
|
||||
|
||||
mcpTransportProvider.setSessionFactory(
|
||||
new DefaultMcpStreamableServerSessionFactory(requestTimeout, this::asyncInitializeRequestHandler,
|
||||
requestHandlers, notificationHandlers, sessionId -> this.cleanupForSession(sessionId)));
|
||||
mcpTransportProvider.setSessionFactory(new DefaultMcpStreamableServerSessionFactory(requestTimeout,
|
||||
this::asyncInitializeRequestHandler, requestHandlers, notificationHandlers,
|
||||
sessionId -> this.cleanupForSession(sessionId), this.jsonSchemaValidator));
|
||||
}
|
||||
|
||||
private Map<String, McpNotificationHandler> prepareNotificationHandlers(McpServerFeatures.Async features) {
|
||||
|
|
@ -347,6 +348,15 @@ public class McpAsyncServer {
|
|||
return Mono.error(new IllegalStateException("Server must be configured with tool capabilities"));
|
||||
}
|
||||
|
||||
try {
|
||||
var t = toolSpecification.tool();
|
||||
this.jsonSchemaValidator.assertConforms("Tool '" + t.name() + "' inputSchema", t.inputSchema());
|
||||
this.jsonSchemaValidator.assertConforms("Tool '" + t.name() + "' outputSchema", t.outputSchema());
|
||||
}
|
||||
catch (IllegalArgumentException e) {
|
||||
return Mono.error(e);
|
||||
}
|
||||
|
||||
var wrappedToolSpecification = withStructuredOutputHandling(this.jsonSchemaValidator, toolSpecification);
|
||||
|
||||
return Mono.defer(() -> {
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import java.util.ArrayList;
|
|||
import java.util.Collections;
|
||||
|
||||
import io.modelcontextprotocol.json.TypeRef;
|
||||
import io.modelcontextprotocol.json.schema.JsonSchemaValidator;
|
||||
import io.modelcontextprotocol.spec.McpError;
|
||||
import io.modelcontextprotocol.spec.McpLoggableSession;
|
||||
import io.modelcontextprotocol.spec.McpSchema;
|
||||
|
|
@ -37,6 +38,8 @@ public class McpAsyncServerExchange {
|
|||
|
||||
private final McpTransportContext transportContext;
|
||||
|
||||
private final JsonSchemaValidator jsonSchemaValidator;
|
||||
|
||||
private static final TypeRef<McpSchema.CreateMessageResult> CREATE_MESSAGE_RESULT_TYPE_REF = new TypeRef<>() {
|
||||
};
|
||||
|
||||
|
|
@ -51,6 +54,29 @@ public class McpAsyncServerExchange {
|
|||
|
||||
/**
|
||||
* Create a new asynchronous exchange with the client.
|
||||
* @param sessionId the session ID
|
||||
* @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
|
||||
* @param jsonSchemaValidator optional validator used to verify elicitation schemas
|
||||
*/
|
||||
public McpAsyncServerExchange(String sessionId, McpLoggableSession session,
|
||||
McpSchema.ClientCapabilities clientCapabilities, McpSchema.Implementation clientInfo,
|
||||
McpTransportContext transportContext, JsonSchemaValidator jsonSchemaValidator) {
|
||||
this.sessionId = sessionId;
|
||||
this.session = session;
|
||||
this.clientCapabilities = clientCapabilities;
|
||||
this.clientInfo = clientInfo;
|
||||
this.transportContext = transportContext;
|
||||
this.jsonSchemaValidator = jsonSchemaValidator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new asynchronous exchange with the client.
|
||||
* @param sessionId the session ID
|
||||
* @param session The server session representing a 1-1 interaction.
|
||||
* @param clientCapabilities The client capabilities that define the supported
|
||||
* features and functionality.
|
||||
|
|
@ -61,11 +87,7 @@ public class McpAsyncServerExchange {
|
|||
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;
|
||||
this(sessionId, session, clientCapabilities, clientInfo, transportContext, null);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -152,6 +174,15 @@ public class McpAsyncServerExchange {
|
|||
if (this.clientCapabilities.elicitation() == null) {
|
||||
return Mono.error(new IllegalStateException("Client must be configured with elicitation capabilities"));
|
||||
}
|
||||
if (this.jsonSchemaValidator != null) {
|
||||
try {
|
||||
this.jsonSchemaValidator.assertConforms("ElicitRequest requestedSchema",
|
||||
elicitRequest.requestedSchema());
|
||||
}
|
||||
catch (IllegalArgumentException e) {
|
||||
return Mono.error(e);
|
||||
}
|
||||
}
|
||||
return this.session.sendRequest(McpSchema.METHOD_ELICITATION_CREATE, elicitRequest,
|
||||
ELICITATION_RESULT_TYPE_REF);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -243,6 +243,8 @@ public interface McpServer {
|
|||
var jsonSchemaValidator = (this.jsonSchemaValidator != null) ? this.jsonSchemaValidator
|
||||
: McpJsonDefaults.getSchemaValidator();
|
||||
|
||||
validateAsyncToolSchemas(jsonSchemaValidator, this.tools);
|
||||
|
||||
return new McpAsyncServer(transportProvider, jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper,
|
||||
features, requestTimeout, uriTemplateManagerFactory, jsonSchemaValidator, validateToolInputs);
|
||||
}
|
||||
|
|
@ -269,6 +271,9 @@ public interface McpServer {
|
|||
this.instructions);
|
||||
var jsonSchemaValidator = this.jsonSchemaValidator != null ? this.jsonSchemaValidator
|
||||
: McpJsonDefaults.getSchemaValidator();
|
||||
|
||||
validateAsyncToolSchemas(jsonSchemaValidator, this.tools);
|
||||
|
||||
return new McpAsyncServer(transportProvider, jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper,
|
||||
features, requestTimeout, uriTemplateManagerFactory, jsonSchemaValidator, validateToolInputs);
|
||||
}
|
||||
|
|
@ -829,11 +834,14 @@ public interface McpServer {
|
|||
McpServerFeatures.Async asyncFeatures = McpServerFeatures.Async.fromSync(syncFeatures,
|
||||
this.immediateExecution);
|
||||
|
||||
var jsonSchemaValidator = this.jsonSchemaValidator != null ? this.jsonSchemaValidator
|
||||
: McpJsonDefaults.getSchemaValidator();
|
||||
|
||||
validateSyncToolSchemas(jsonSchemaValidator, this.tools);
|
||||
|
||||
var asyncServer = new McpAsyncServer(transportProvider,
|
||||
jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper, asyncFeatures, requestTimeout,
|
||||
uriTemplateManagerFactory,
|
||||
jsonSchemaValidator != null ? jsonSchemaValidator : McpJsonDefaults.getSchemaValidator(),
|
||||
validateToolInputs);
|
||||
uriTemplateManagerFactory, jsonSchemaValidator, validateToolInputs);
|
||||
return new McpSyncServer(asyncServer, this.immediateExecution);
|
||||
}
|
||||
|
||||
|
|
@ -862,6 +870,9 @@ public interface McpServer {
|
|||
this.immediateExecution);
|
||||
var jsonSchemaValidator = this.jsonSchemaValidator != null ? this.jsonSchemaValidator
|
||||
: McpJsonDefaults.getSchemaValidator();
|
||||
|
||||
validateSyncToolSchemas(jsonSchemaValidator, this.tools);
|
||||
|
||||
var asyncServer = new McpAsyncServer(transportProvider,
|
||||
jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper, asyncFeatures, this.requestTimeout,
|
||||
this.uriTemplateManagerFactory, jsonSchemaValidator, validateToolInputs);
|
||||
|
|
@ -1898,10 +1909,13 @@ public interface McpServer {
|
|||
public McpStatelessAsyncServer build() {
|
||||
var features = new McpStatelessServerFeatures.Async(this.serverInfo, this.serverCapabilities, this.tools,
|
||||
this.resources, this.resourceTemplates, this.prompts, this.completions, this.instructions);
|
||||
var jsonSchemaValidator = this.jsonSchemaValidator != null ? this.jsonSchemaValidator
|
||||
: McpJsonDefaults.getSchemaValidator();
|
||||
|
||||
validateStatelessAsyncToolSchemas(jsonSchemaValidator, this.tools);
|
||||
|
||||
return new McpStatelessAsyncServer(transport, jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper,
|
||||
features, requestTimeout, uriTemplateManagerFactory,
|
||||
jsonSchemaValidator != null ? jsonSchemaValidator : McpJsonDefaults.getSchemaValidator(),
|
||||
validateToolInputs);
|
||||
features, requestTimeout, uriTemplateManagerFactory, jsonSchemaValidator, validateToolInputs);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -2412,14 +2426,42 @@ public interface McpServer {
|
|||
var syncFeatures = new McpStatelessServerFeatures.Sync(this.serverInfo, this.serverCapabilities, this.tools,
|
||||
this.resources, this.resourceTemplates, this.prompts, this.completions, this.instructions);
|
||||
var asyncFeatures = McpStatelessServerFeatures.Async.fromSync(syncFeatures, this.immediateExecution);
|
||||
var jsonSchemaValidator = this.jsonSchemaValidator != null ? this.jsonSchemaValidator
|
||||
: McpJsonDefaults.getSchemaValidator();
|
||||
|
||||
validateStatelessSyncToolSchemas(jsonSchemaValidator, this.tools);
|
||||
|
||||
var asyncServer = new McpStatelessAsyncServer(transport,
|
||||
jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper, asyncFeatures, requestTimeout,
|
||||
uriTemplateManagerFactory,
|
||||
this.jsonSchemaValidator != null ? this.jsonSchemaValidator : McpJsonDefaults.getSchemaValidator(),
|
||||
validateToolInputs);
|
||||
uriTemplateManagerFactory, jsonSchemaValidator, validateToolInputs);
|
||||
return new McpStatelessSyncServer(asyncServer, this.immediateExecution);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static void validateAsyncToolSchemas(JsonSchemaValidator validator,
|
||||
List<McpServerFeatures.AsyncToolSpecification> tools) {
|
||||
tools.forEach(spec -> validateToolSchema(validator, spec.tool()));
|
||||
}
|
||||
|
||||
private static void validateSyncToolSchemas(JsonSchemaValidator validator,
|
||||
List<McpServerFeatures.SyncToolSpecification> tools) {
|
||||
tools.forEach(spec -> validateToolSchema(validator, spec.tool()));
|
||||
}
|
||||
|
||||
private static void validateStatelessAsyncToolSchemas(JsonSchemaValidator validator,
|
||||
List<McpStatelessServerFeatures.AsyncToolSpecification> tools) {
|
||||
tools.forEach(spec -> validateToolSchema(validator, spec.tool()));
|
||||
}
|
||||
|
||||
private static void validateStatelessSyncToolSchemas(JsonSchemaValidator validator,
|
||||
List<McpStatelessServerFeatures.SyncToolSpecification> tools) {
|
||||
tools.forEach(spec -> validateToolSchema(validator, spec.tool()));
|
||||
}
|
||||
|
||||
private static void validateToolSchema(JsonSchemaValidator validator, McpSchema.Tool tool) {
|
||||
validator.assertConforms("Tool '" + tool.name() + "' inputSchema", tool.inputSchema());
|
||||
validator.assertConforms("Tool '" + tool.name() + "' outputSchema", tool.outputSchema());
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -339,6 +339,15 @@ public class McpStatelessAsyncServer {
|
|||
return Mono.error(new IllegalStateException("Server must be configured with tool capabilities"));
|
||||
}
|
||||
|
||||
try {
|
||||
var t = toolSpecification.tool();
|
||||
this.jsonSchemaValidator.assertConforms("Tool '" + t.name() + "' inputSchema", t.inputSchema());
|
||||
this.jsonSchemaValidator.assertConforms("Tool '" + t.name() + "' outputSchema", t.outputSchema());
|
||||
}
|
||||
catch (IllegalArgumentException e) {
|
||||
return Mono.error(e);
|
||||
}
|
||||
|
||||
var wrappedToolSpecification = withStructuredOutputHandling(this.jsonSchemaValidator, toolSpecification);
|
||||
|
||||
return Mono.defer(() -> {
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@
|
|||
|
||||
package io.modelcontextprotocol.spec;
|
||||
|
||||
import io.modelcontextprotocol.json.schema.JsonSchemaValidator;
|
||||
import io.modelcontextprotocol.server.McpNotificationHandler;
|
||||
import io.modelcontextprotocol.server.McpRequestHandler;
|
||||
|
||||
|
|
@ -31,6 +32,31 @@ public class DefaultMcpStreamableServerSessionFactory implements McpStreamableSe
|
|||
|
||||
private final Function<String, Mono<Void>> onClose;
|
||||
|
||||
private final JsonSchemaValidator jsonSchemaValidator;
|
||||
|
||||
/**
|
||||
* 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
|
||||
* @param onClose reactive callback invoked with the session ID when a session is
|
||||
* closed
|
||||
* @param jsonSchemaValidator optional validator threaded to sessions user-provided
|
||||
* schema validation
|
||||
*/
|
||||
public DefaultMcpStreamableServerSessionFactory(Duration requestTimeout,
|
||||
McpStreamableServerSession.InitRequestHandler initRequestHandler,
|
||||
Map<String, McpRequestHandler<?>> requestHandlers, Map<String, McpNotificationHandler> notificationHandlers,
|
||||
Function<String, Mono<Void>> onClose, JsonSchemaValidator jsonSchemaValidator) {
|
||||
this.requestTimeout = requestTimeout;
|
||||
this.initRequestHandler = initRequestHandler;
|
||||
this.requestHandlers = requestHandlers;
|
||||
this.notificationHandlers = notificationHandlers;
|
||||
this.onClose = onClose;
|
||||
this.jsonSchemaValidator = jsonSchemaValidator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Constructs an instance.
|
||||
* @param requestTimeout timeout for requests
|
||||
|
|
@ -44,11 +70,7 @@ public class DefaultMcpStreamableServerSessionFactory implements McpStreamableSe
|
|||
McpStreamableServerSession.InitRequestHandler initRequestHandler,
|
||||
Map<String, McpRequestHandler<?>> requestHandlers, Map<String, McpNotificationHandler> notificationHandlers,
|
||||
Function<String, Mono<Void>> onClose) {
|
||||
this.requestTimeout = requestTimeout;
|
||||
this.initRequestHandler = initRequestHandler;
|
||||
this.requestHandlers = requestHandlers;
|
||||
this.notificationHandlers = notificationHandlers;
|
||||
this.onClose = onClose;
|
||||
this(requestTimeout, initRequestHandler, requestHandlers, notificationHandlers, onClose, null);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -73,9 +95,10 @@ public class DefaultMcpStreamableServerSessionFactory implements McpStreamableSe
|
|||
public McpStreamableServerSession.McpStreamableServerSessionInit startSession(
|
||||
McpSchema.InitializeRequest initializeRequest) {
|
||||
String sessionId = UUID.randomUUID().toString();
|
||||
return new McpStreamableServerSession.McpStreamableServerSessionInit(new McpStreamableServerSession(sessionId,
|
||||
initializeRequest.capabilities(), initializeRequest.clientInfo(), requestTimeout, requestHandlers,
|
||||
notificationHandlers, () -> this.onClose.apply(sessionId)),
|
||||
return new McpStreamableServerSession.McpStreamableServerSessionInit(
|
||||
new McpStreamableServerSession(sessionId, initializeRequest.capabilities(),
|
||||
initializeRequest.clientInfo(), requestTimeout, requestHandlers, notificationHandlers,
|
||||
() -> this.onClose.apply(sessionId), this.jsonSchemaValidator),
|
||||
this.initRequestHandler.handle(initializeRequest));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -11,7 +11,9 @@ import java.util.Map;
|
|||
* defines a method to validate structured content based on the provided output schema.
|
||||
*
|
||||
* @author Christian Tzolov
|
||||
* @deprecated Use {@link io.modelcontextprotocol.json.schema.JsonSchemaValidator}
|
||||
*/
|
||||
@Deprecated
|
||||
public interface JsonSchemaValidator {
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -46,6 +46,12 @@ public final class McpSchema {
|
|||
|
||||
public static final String FIRST_PAGE = null;
|
||||
|
||||
/**
|
||||
* The JSON Schema 2020-12 meta-schema URI (SEP-1613). This is the default dialect for
|
||||
* all schema objects in MCP when no explicit {@code $schema} field is present.
|
||||
*/
|
||||
public static final String JSON_SCHEMA_DIALECT_2020_12 = "https://json-schema.org/draft/2020-12/schema";
|
||||
|
||||
// ---------------------------
|
||||
// Method Names
|
||||
// ---------------------------
|
||||
|
|
@ -2666,9 +2672,14 @@ public final class McpSchema {
|
|||
* @param description A human-readable description of what the tool does. This can be
|
||||
* used by clients to improve the LLM's understanding of available tools.
|
||||
* @param inputSchema A JSON Schema object that describes the expected structure of
|
||||
* the arguments when calling this tool. This allows clients to validate tool
|
||||
* the arguments when calling this tool. Per SEP-1613, the dialect defaults to JSON
|
||||
* Schema 2020-12 ({@link #JSON_SCHEMA_DIALECT_2020_12}) when no explicit
|
||||
* {@code $schema} entry is present. To declare a different dialect, include a
|
||||
* {@code "$schema"} key in the map. For tools with no parameters the spec recommends
|
||||
* {@code {"type":"object","additionalProperties":false}}.
|
||||
* @param outputSchema An optional JSON Schema object defining the structure of the
|
||||
* tool's output returned in the structuredContent field of a CallToolResult.
|
||||
* tool's output returned in the structuredContent field of a CallToolResult. Same
|
||||
* dialect rules as {@code inputSchema}.
|
||||
* @param annotations Optional additional tool information.
|
||||
* @param meta See specification for notes on _meta usage
|
||||
*/
|
||||
|
|
@ -3692,7 +3703,10 @@ public final class McpSchema {
|
|||
*
|
||||
* @param message The message to present to the user
|
||||
* @param requestedSchema A restricted subset of JSON Schema. Only top-level
|
||||
* properties are allowed, without nesting
|
||||
* properties are allowed, without nesting. Per SEP-1613, the dialect defaults to JSON
|
||||
* Schema 2020-12 ({@link #JSON_SCHEMA_DIALECT_2020_12}) when no explicit
|
||||
* {@code $schema} entry is present. To declare a different dialect, include a
|
||||
* {@code "$schema"} key in the map.
|
||||
* @param meta See specification for notes on _meta usage
|
||||
* <p>
|
||||
* Note: {@code message} and {@code requestedSchema} are required by the MCP
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ import java.util.concurrent.atomic.AtomicLong;
|
|||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import io.modelcontextprotocol.common.McpTransportContext;
|
||||
import io.modelcontextprotocol.json.schema.JsonSchemaValidator;
|
||||
import io.modelcontextprotocol.server.McpAsyncServerExchange;
|
||||
import io.modelcontextprotocol.server.McpInitRequestHandler;
|
||||
import io.modelcontextprotocol.server.McpNotificationHandler;
|
||||
|
|
@ -68,6 +69,36 @@ public class McpServerSession implements McpLoggableSession {
|
|||
|
||||
private final Supplier<Mono<Void>> onClose;
|
||||
|
||||
private final JsonSchemaValidator jsonSchemaValidator;
|
||||
|
||||
/**
|
||||
* Creates a new server session with the given parameters and the transport to use.
|
||||
* @param id session id
|
||||
* @param requestTimeout duration to wait for request responses before timing out
|
||||
* @param transport the transport to use
|
||||
* @param initHandler called when a
|
||||
* {@link io.modelcontextprotocol.spec.McpSchema.InitializeRequest} is received by the
|
||||
* server
|
||||
* @param requestHandlers map of request handlers to use
|
||||
* @param notificationHandlers map of notification handlers to use
|
||||
* @param onClose supplier of a reactive callback invoked when the session is closed
|
||||
* @param jsonSchemaValidator optional validator threaded to exchanges for elicitation
|
||||
* schema validation
|
||||
*/
|
||||
public McpServerSession(String id, Duration requestTimeout, McpServerTransport transport,
|
||||
McpInitRequestHandler initHandler, Map<String, McpRequestHandler<?>> requestHandlers,
|
||||
Map<String, McpNotificationHandler> notificationHandlers, Supplier<Mono<Void>> onClose,
|
||||
JsonSchemaValidator jsonSchemaValidator) {
|
||||
this.id = id;
|
||||
this.requestTimeout = requestTimeout;
|
||||
this.transport = transport;
|
||||
this.initRequestHandler = initHandler;
|
||||
this.requestHandlers = requestHandlers;
|
||||
this.notificationHandlers = notificationHandlers;
|
||||
this.onClose = onClose;
|
||||
this.jsonSchemaValidator = jsonSchemaValidator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a new server session with the given parameters and the transport to use.
|
||||
* @param id session id
|
||||
|
|
@ -83,13 +114,7 @@ public class McpServerSession implements McpLoggableSession {
|
|||
public McpServerSession(String id, Duration requestTimeout, McpServerTransport transport,
|
||||
McpInitRequestHandler initHandler, Map<String, McpRequestHandler<?>> requestHandlers,
|
||||
Map<String, McpNotificationHandler> notificationHandlers, Supplier<Mono<Void>> onClose) {
|
||||
this.id = id;
|
||||
this.requestTimeout = requestTimeout;
|
||||
this.transport = transport;
|
||||
this.initRequestHandler = initHandler;
|
||||
this.requestHandlers = requestHandlers;
|
||||
this.notificationHandlers = notificationHandlers;
|
||||
this.onClose = onClose;
|
||||
this(id, requestTimeout, transport, initHandler, requestHandlers, notificationHandlers, onClose, null);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -300,7 +325,7 @@ public class McpServerSession implements McpLoggableSession {
|
|||
// FIXME: The session ID passed here is not the same as the one in the
|
||||
// legacy SSE transport.
|
||||
exchangeSink.tryEmitValue(new McpAsyncServerExchange(this.id, this, clientCapabilities.get(),
|
||||
clientInfo.get(), transportContext));
|
||||
clientInfo.get(), transportContext, this.jsonSchemaValidator));
|
||||
}
|
||||
|
||||
var handler = notificationHandlers.get(notification.method());
|
||||
|
|
@ -322,7 +347,7 @@ public class McpServerSession implements McpLoggableSession {
|
|||
*/
|
||||
private McpAsyncServerExchange copyExchange(McpAsyncServerExchange exchange, McpTransportContext transportContext) {
|
||||
return new McpAsyncServerExchange(exchange.sessionId(), this, exchange.getClientCapabilities(),
|
||||
exchange.getClientInfo(), transportContext);
|
||||
exchange.getClientInfo(), transportContext, this.jsonSchemaValidator);
|
||||
}
|
||||
|
||||
record MethodNotFoundError(String method, String message, Object data) {
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import org.slf4j.LoggerFactory;
|
|||
import io.modelcontextprotocol.json.TypeRef;
|
||||
|
||||
import io.modelcontextprotocol.common.McpTransportContext;
|
||||
import io.modelcontextprotocol.json.schema.JsonSchemaValidator;
|
||||
import io.modelcontextprotocol.server.McpAsyncServerExchange;
|
||||
import io.modelcontextprotocol.server.McpNotificationHandler;
|
||||
import io.modelcontextprotocol.server.McpRequestHandler;
|
||||
|
|
@ -64,6 +65,37 @@ public class McpStreamableServerSession implements McpLoggableSession {
|
|||
|
||||
private final Supplier<Mono<Void>> onClose;
|
||||
|
||||
private final JsonSchemaValidator jsonSchemaValidator;
|
||||
|
||||
/**
|
||||
* Create an instance of the streamable session.
|
||||
* @param id session ID
|
||||
* @param clientCapabilities client capabilities
|
||||
* @param clientInfo client info
|
||||
* @param requestTimeout timeout to use for requests
|
||||
* @param requestHandlers the map of MCP request handlers keyed by method name
|
||||
* @param notificationHandlers the map of MCP notification handlers keyed by method
|
||||
* name
|
||||
* @param onClose supplier of a reactive callback invoked when the session is closed
|
||||
* @param jsonSchemaValidator optional validator threaded to exchanges for elicitation
|
||||
* schema validation
|
||||
*/
|
||||
public McpStreamableServerSession(String id, McpSchema.ClientCapabilities clientCapabilities,
|
||||
McpSchema.Implementation clientInfo, Duration requestTimeout,
|
||||
Map<String, McpRequestHandler<?>> requestHandlers, Map<String, McpNotificationHandler> notificationHandlers,
|
||||
Supplier<Mono<Void>> onClose, JsonSchemaValidator jsonSchemaValidator) {
|
||||
this.id = id;
|
||||
this.missingMcpTransportSession = new MissingMcpTransportSession(id);
|
||||
this.listeningStreamRef = new AtomicReference<>(this.missingMcpTransportSession);
|
||||
this.clientCapabilities.lazySet(clientCapabilities);
|
||||
this.clientInfo.lazySet(clientInfo);
|
||||
this.requestTimeout = requestTimeout;
|
||||
this.requestHandlers = requestHandlers;
|
||||
this.notificationHandlers = notificationHandlers;
|
||||
this.onClose = onClose;
|
||||
this.jsonSchemaValidator = jsonSchemaValidator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an instance of the streamable session.
|
||||
* @param id session ID
|
||||
|
|
@ -79,15 +111,7 @@ public class McpStreamableServerSession implements McpLoggableSession {
|
|||
McpSchema.Implementation clientInfo, Duration requestTimeout,
|
||||
Map<String, McpRequestHandler<?>> requestHandlers, Map<String, McpNotificationHandler> notificationHandlers,
|
||||
Supplier<Mono<Void>> onClose) {
|
||||
this.id = id;
|
||||
this.missingMcpTransportSession = new MissingMcpTransportSession(id);
|
||||
this.listeningStreamRef = new AtomicReference<>(this.missingMcpTransportSession);
|
||||
this.clientCapabilities.lazySet(clientCapabilities);
|
||||
this.clientInfo.lazySet(clientInfo);
|
||||
this.requestTimeout = requestTimeout;
|
||||
this.requestHandlers = requestHandlers;
|
||||
this.notificationHandlers = notificationHandlers;
|
||||
this.onClose = onClose;
|
||||
this(id, clientCapabilities, clientInfo, requestTimeout, requestHandlers, notificationHandlers, onClose, null);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -196,7 +220,7 @@ public class McpStreamableServerSession implements McpLoggableSession {
|
|||
}
|
||||
return requestHandler
|
||||
.handle(new McpAsyncServerExchange(this.id, stream, clientCapabilities.get(), clientInfo.get(),
|
||||
transportContext), jsonrpcRequest.params())
|
||||
transportContext, this.jsonSchemaValidator), jsonrpcRequest.params())
|
||||
.map(result -> McpSchema.JSONRPCResponse.result(jsonrpcRequest.id(), result))
|
||||
.onErrorResume(e -> {
|
||||
McpSchema.JSONRPCResponse.JSONRPCError jsonRpcError = (e instanceof McpError mcpError
|
||||
|
|
@ -227,7 +251,8 @@ public class McpStreamableServerSession implements McpLoggableSession {
|
|||
}
|
||||
McpLoggableSession listeningStream = this.listeningStreamRef.get();
|
||||
return notificationHandler.handle(new McpAsyncServerExchange(this.id, listeningStream,
|
||||
this.clientCapabilities.get(), this.clientInfo.get(), transportContext), notification.params());
|
||||
this.clientCapabilities.get(), this.clientInfo.get(), transportContext, this.jsonSchemaValidator),
|
||||
notification.params());
|
||||
});
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue