Release version 1.1.4

Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>

.github/workflows/publish-snapshot.yml

@@ -2,7 +2,7 @@ name: Publish Snapshot
 
 on:
   push:
-    branches: [ "main" ]
+    branches: [ "1.1.x" ]
 
 jobs:
   build:

ROADMAP.md

@@ -27,7 +27,7 @@ Once we catch up on the most recent MCP specification revision we aim to fully s
 
 ### v1.x Development
 
-The Java SDK is currently in active development as v1.x, following a recent stable 1.0.0 release. The SDK provides:
+The Java SDK is currently in active development as v1.x, following a recent stable 1.1.0 release. The SDK provides:
 
 - MCP protocol implementation
 - Synchronous and asynchronous programming models

conformance-tests/VALIDATION_RESULTS.md

@@ -87,7 +87,7 @@ cd conformance-tests/client-jdk-http-client
 # Run all scenarios
 for scenario in initialize tools_call elicitation-sep1034-client-defaults sse-retry; do
   npx @modelcontextprotocol/conformance client \
-    --command "java -jar target/client-jdk-http-client-1.1.0-SNAPSHOT.jar" \
+    --command "java -jar target/client-jdk-http-client-1.1.1-SNAPSHOT.jar" \
     --scenario $scenario
 done
 ```
@@ -104,7 +104,7 @@ cd conformance-tests/client-spring-http-client
 # Run auth suite
 npx @modelcontextprotocol/conformance@0.1.15 client \
   --spec-version 2025-11-25 \
-  --command "java -jar target/client-spring-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar target/client-spring-http-client-1.1.1-SNAPSHOT.jar" \
   --suite auth
 ```
 

conformance-tests/client-jdk-http-client/README.md

@@ -54,7 +54,7 @@ cd conformance-tests/client-jdk-http-client
 
 This creates an executable JAR at:
 ```
-target/client-jdk-http-client-1.1.0-SNAPSHOT.jar
+target/client-jdk-http-client-1.1.1-SNAPSHOT.jar
 ```
 
 ## Running Tests
@@ -65,19 +65,19 @@ Run a single scenario:
 
 ```bash
 npx @modelcontextprotocol/conformance client \
-  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.1-SNAPSHOT.jar" \
   --scenario initialize
 
 npx @modelcontextprotocol/conformance client \
-  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.1-SNAPSHOT.jar" \
   --scenario tools_call
 
 npx @modelcontextprotocol/conformance client \
-  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.1-SNAPSHOT.jar" \
   --scenario elicitation-sep1034-client-defaults
 
 npx @modelcontextprotocol/conformance client \
-  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.1-SNAPSHOT.jar" \
   --scenario sse-retry
 ```
 
@@ -85,7 +85,7 @@ Run with verbose output:
 
 ```bash
 npx @modelcontextprotocol/conformance client \
-  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.1-SNAPSHOT.jar" \
   --scenario initialize \
   --verbose
 ```
@@ -96,7 +96,7 @@ You can also run the client manually if you have a test server:
 
 ```bash
 export MCP_CONFORMANCE_SCENARIO=initialize
-java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.0-SNAPSHOT.jar http://localhost:3000/mcp
+java -jar conformance-tests/client-jdk-http-client/target/client-jdk-http-client-1.1.1-SNAPSHOT.jar http://localhost:3000/mcp
 ```
 
 ## Test Results

conformance-tests/client-jdk-http-client/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>io.modelcontextprotocol.sdk</groupId>
 		<artifactId>conformance-tests</artifactId>
-		<version>1.1.0-SNAPSHOT</version>
+		<version>1.1.4</version>
 	</parent>
 	<artifactId>client-jdk-http-client</artifactId>
 	<packaging>jar</packaging>
@@ -28,7 +28,7 @@
 		<dependency>
 			<groupId>io.modelcontextprotocol.sdk</groupId>
 			<artifactId>mcp</artifactId>
-			<version>1.1.0-SNAPSHOT</version>
+			<version>1.1.4</version>
 		</dependency>
 
 		<!-- Logging -->

conformance-tests/client-spring-http-client/README.md

@@ -67,7 +67,7 @@ cd conformance-tests/client-spring-http-client
 
 This creates an executable JAR at:
 ```
-target/client-spring-http-client-1.1.0-SNAPSHOT.jar
+target/client-spring-http-client-1.1.1-SNAPSHOT.jar
 ```
 
 ## Running Tests
@@ -79,7 +79,7 @@ Run the full auth suite:
 ```bash
 npx @modelcontextprotocol/conformance@0.1.15 client \
   --spec-version 2025-11-25 \
-  --command "java -jar conformance-tests/client-spring-http-client/target/client-spring-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar conformance-tests/client-spring-http-client/target/client-spring-http-client-1.1.1-SNAPSHOT.jar" \
   --suite auth
 ```
 
@@ -88,7 +88,7 @@ Run a single scenario:
 ```bash
 npx @modelcontextprotocol/conformance@0.1.15 client \
   --spec-version 2025-11-25 \
-  --command "java -jar conformance-tests/client-spring-http-client/target/client-spring-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar conformance-tests/client-spring-http-client/target/client-spring-http-client-1.1.1-SNAPSHOT.jar" \
   --scenario auth/metadata-default
 ```
 
@@ -97,7 +97,7 @@ Run with verbose output:
 ```bash
 npx @modelcontextprotocol/conformance@0.1.15 client \
   --spec-version 2025-11-25 \
-  --command "java -jar conformance-tests/client-spring-http-client/target/client-spring-http-client-1.1.0-SNAPSHOT.jar" \
+  --command "java -jar conformance-tests/client-spring-http-client/target/client-spring-http-client-1.1.1-SNAPSHOT.jar" \
   --scenario auth/metadata-default \
   --verbose
 ```
@@ -108,7 +108,7 @@ You can also run the client manually if you have a test server:
 
 ```bash
 export MCP_CONFORMANCE_SCENARIO=auth/metadata-default
-java -jar conformance-tests/client-spring-http-client/target/client-spring-http-client-1.1.0-SNAPSHOT.jar http://localhost:3000/mcp
+java -jar conformance-tests/client-spring-http-client/target/client-spring-http-client-1.1.1-SNAPSHOT.jar http://localhost:3000/mcp
 ```
 
 ## Known Issues

conformance-tests/client-spring-http-client/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>io.modelcontextprotocol.sdk</groupId>
 		<artifactId>conformance-tests</artifactId>
-		<version>1.1.0-SNAPSHOT</version>
+		<version>1.1.4</version>
 	</parent>
 	<artifactId>client-spring-http-client</artifactId>
 	<packaging>jar</packaging>

conformance-tests/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>io.modelcontextprotocol.sdk</groupId>
 		<artifactId>mcp-parent</artifactId>
-		<version>1.1.0-SNAPSHOT</version>
+		<version>1.1.4</version>
 	</parent>
 	<artifactId>conformance-tests</artifactId>
 	<packaging>pom</packaging>

conformance-tests/server-servlet/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>io.modelcontextprotocol.sdk</groupId>
 		<artifactId>conformance-tests</artifactId>
-		<version>1.1.0-SNAPSHOT</version>
+		<version>1.1.4</version>
 	</parent>
 	<artifactId>server-servlet</artifactId>
 	<packaging>jar</packaging>
@@ -28,7 +28,7 @@
 		<dependency>
 			<groupId>io.modelcontextprotocol.sdk</groupId>
 			<artifactId>mcp</artifactId>
-			<version>1.1.0-SNAPSHOT</version>
+			<version>1.1.4</version>
 		</dependency>
 
 		<dependency>

docs/quickstart.md

@@ -123,7 +123,7 @@ Add the BOM to your project:
             <dependency>
                 <groupId>io.modelcontextprotocol.sdk</groupId>
                 <artifactId>mcp-bom</artifactId>
-                <version>1.0.0</version>
+                <version>1.1.0</version>
                 <type>pom</type>
                 <scope>import</scope>
             </dependency>
@@ -135,7 +135,7 @@ Add the BOM to your project:
 
     ```groovy
     dependencies {
-        implementation platform("io.modelcontextprotocol.sdk:mcp-bom:1.0.0")
+        implementation platform("io.modelcontextprotocol.sdk:mcp-bom:1.1.0")
         //...
     }
     ```

mcp-bom/pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>io.modelcontextprotocol.sdk</groupId>
         <artifactId>mcp-parent</artifactId>
-        <version>1.1.0-SNAPSHOT</version>
+        <version>1.1.4</version>
     </parent>
 
     <artifactId>mcp-bom</artifactId>

mcp-core/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>io.modelcontextprotocol.sdk</groupId>
 		<artifactId>mcp-parent</artifactId>
-		<version>1.1.0-SNAPSHOT</version>
+		<version>1.1.4</version>
 	</parent>
 	<artifactId>mcp-core</artifactId>
 	<packaging>jar</packaging>

mcp-core/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java

@@ -303,7 +303,7 @@ public class McpAsyncClient {
 				return Mono.empty();
 			}
 
-			return this.listToolsInternal(init, McpSchema.FIRST_PAGE).doOnNext(listToolsResult -> {
+			return this.listToolsInternal(init, McpSchema.FIRST_PAGE, null).doOnNext(listToolsResult -> {
 				listToolsResult.tools()
 					.forEach(tool -> logger.debug("Tool {} schema: {}", tool.name(), tool.outputSchema()));
 				if (enableCallToolSchemaCaching && listToolsResult.tools() != null) {
@@ -645,16 +645,27 @@ public class McpAsyncClient {
 	 * @return A Mono that emits the list of tools result
 	 */
 	public Mono<McpSchema.ListToolsResult> listTools(String cursor) {
-		return this.initializer.withInitialization("listing tools", init -> this.listToolsInternal(init, cursor));
+		return this.initializer.withInitialization("listing tools", init -> this.listToolsInternal(init, cursor, null));
 	}
 
-	private Mono<McpSchema.ListToolsResult> listToolsInternal(Initialization init, String cursor) {
+	/**
+	 * Retrieves a paginated list of tools with optional metadata.
+	 * @param cursor Optional pagination cursor from a previous list request
+	 * @param meta Optional metadata to include in the request (_meta field)
+	 * @return A Mono that emits the list of tools result
+	 */
+	public Mono<McpSchema.ListToolsResult> listTools(String cursor, Map<String, Object> meta) {
+		return this.initializer.withInitialization("listing tools", init -> this.listToolsInternal(init, cursor, meta));
+	}
+
+	private Mono<McpSchema.ListToolsResult> listToolsInternal(Initialization init, String cursor,
+			Map<String, Object> meta) {
 
 		if (init.initializeResult().capabilities().tools() == null) {
 			return Mono.error(new IllegalStateException("Server does not provide tools capability"));
 		}
 		return init.mcpSession()
-			.sendRequest(McpSchema.METHOD_TOOLS_LIST, new McpSchema.PaginatedRequest(cursor),
+			.sendRequest(McpSchema.METHOD_TOOLS_LIST, new McpSchema.PaginatedRequest(cursor, meta),
 					LIST_TOOLS_RESULT_TYPE_REF)
 			.doOnNext(result -> {
 				// Validate tool names (warn only)
@@ -725,12 +736,30 @@ public class McpAsyncClient {
 	 * @see #readResource(McpSchema.Resource)
 	 */
 	public Mono<McpSchema.ListResourcesResult> listResources(String cursor) {
+		return this.listResourcesInternal(cursor, null);
+	}
+
+	/**
+	 * Retrieves a paginated list of resources provided by the server. Resources represent
+	 * any kind of UTF-8 encoded data that an MCP server makes available to clients, such
+	 * as database records, API responses, log files, and more.
+	 * @param cursor Optional pagination cursor from a previous list request
+	 * @param meta Optional metadata to include in the request (_meta field)
+	 * @return A Mono that completes with the list of resources result.
+	 * @see McpSchema.ListResourcesResult
+	 * @see #readResource(McpSchema.Resource)
+	 */
+	public Mono<McpSchema.ListResourcesResult> listResources(String cursor, Map<String, Object> meta) {
+		return this.listResourcesInternal(cursor, meta);
+	}
+
+	private Mono<McpSchema.ListResourcesResult> listResourcesInternal(String cursor, Map<String, Object> meta) {
 		return this.initializer.withInitialization("listing resources", init -> {
 			if (init.initializeResult().capabilities().resources() == null) {
 				return Mono.error(new IllegalStateException("Server does not provide the resources capability"));
 			}
 			return init.mcpSession()
-				.sendRequest(McpSchema.METHOD_RESOURCES_LIST, new McpSchema.PaginatedRequest(cursor),
+				.sendRequest(McpSchema.METHOD_RESOURCES_LIST, new McpSchema.PaginatedRequest(cursor, meta),
 						LIST_RESOURCES_RESULT_TYPE_REF);
 		});
 	}
@@ -795,12 +824,30 @@ public class McpAsyncClient {
 	 * @see McpSchema.ListResourceTemplatesResult
 	 */
 	public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates(String cursor) {
+		return this.listResourceTemplatesInternal(cursor, null);
+	}
+
+	/**
+	 * Retrieves a paginated list of resource templates provided by the server. Resource
+	 * templates allow servers to expose parameterized resources using URI templates,
+	 * enabling dynamic resource access based on variable parameters.
+	 * @param cursor Optional pagination cursor from a previous list request
+	 * @param meta Optional metadata to include in the request (_meta field)
+	 * @return A Mono that completes with the list of resource templates result.
+	 * @see McpSchema.ListResourceTemplatesResult
+	 */
+	public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates(String cursor, Map<String, Object> meta) {
+		return this.listResourceTemplatesInternal(cursor, meta);
+	}
+
+	private Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplatesInternal(String cursor,
+			Map<String, Object> meta) {
 		return this.initializer.withInitialization("listing resource templates", init -> {
 			if (init.initializeResult().capabilities().resources() == null) {
 				return Mono.error(new IllegalStateException("Server does not provide the resources capability"));
 			}
 			return init.mcpSession()
-				.sendRequest(McpSchema.METHOD_RESOURCES_TEMPLATES_LIST, new McpSchema.PaginatedRequest(cursor),
+				.sendRequest(McpSchema.METHOD_RESOURCES_TEMPLATES_LIST, new McpSchema.PaginatedRequest(cursor, meta),
 						LIST_RESOURCE_TEMPLATES_RESULT_TYPE_REF);
 		});
 	}
@@ -895,8 +942,26 @@ public class McpAsyncClient {
 	 * @see #getPrompt(GetPromptRequest)
 	 */
 	public Mono<ListPromptsResult> listPrompts(String cursor) {
-		return this.initializer.withInitialization("listing prompts", init -> init.mcpSession()
-			.sendRequest(McpSchema.METHOD_PROMPT_LIST, new PaginatedRequest(cursor), LIST_PROMPTS_RESULT_TYPE_REF));
+		return this.listPromptsInternal(cursor, null);
+	}
+
+	/**
+	 * Retrieves a paginated list of prompts with optional metadata.
+	 * @param cursor Optional pagination cursor from a previous list request
+	 * @param meta Optional metadata to include in the request (_meta field)
+	 * @return A Mono that completes with the list of prompts result.
+	 * @see McpSchema.ListPromptsResult
+	 * @see #getPrompt(GetPromptRequest)
+	 */
+	public Mono<ListPromptsResult> listPrompts(String cursor, Map<String, Object> meta) {
+		return this.listPromptsInternal(cursor, meta);
+	}
+
+	private Mono<ListPromptsResult> listPromptsInternal(String cursor, Map<String, Object> meta) {
+		return this.initializer.withInitialization("listing prompts",
+				init -> init.mcpSession()
+					.sendRequest(McpSchema.METHOD_PROMPT_LIST, new PaginatedRequest(cursor, meta),
+							LIST_PROMPTS_RESULT_TYPE_REF));
 	}
 
 	/**

mcp-core/src/main/java/io/modelcontextprotocol/client/McpSyncClient.java

@@ -5,6 +5,7 @@
 package io.modelcontextprotocol.client;
 
 import java.time.Duration;
+import java.util.Map;
 import java.util.function.Supplier;
 
 import org.slf4j.Logger;
@@ -259,6 +260,18 @@ public class McpSyncClient implements AutoCloseable {
 
 	}
 
+	/**
+	 * Retrieves a paginated list of tools provided by the server.
+	 * @param cursor Optional pagination cursor from a previous list request
+	 * @param meta Optional metadata to include in the request (_meta field)
+	 * @return The list of tools result containing: - tools: List of available tools, each
+	 * with a name, description, and input schema - nextCursor: Optional cursor for
+	 * pagination if more tools are available
+	 */
+	public McpSchema.ListToolsResult listTools(String cursor, Map<String, Object> meta) {
+		return withProvidedContext(this.delegate.listTools(cursor, meta)).block();
+	}
+
 	// --------------------------
 	// Resources
 	// --------------------------
@@ -282,6 +295,17 @@ public class McpSyncClient implements AutoCloseable {
 
 	}
 
+	/**
+	 * Retrieves a paginated list of resources with optional metadata.
+	 * @param cursor Optional pagination cursor from a previous list request
+	 * @param meta Optional metadata to include in the request (_meta field)
+	 * @return The list of resources result
+	 */
+	public McpSchema.ListResourcesResult listResources(String cursor, Map<String, Object> meta) {
+		return withProvidedContext(this.delegate.listResources(cursor, meta)).block();
+
+	}
+
 	/**
 	 * Send a resources/read request.
 	 * @param resource the resource to read
@@ -324,6 +348,20 @@ public class McpSyncClient implements AutoCloseable {
 
 	}
 
+	/**
+	 * Resource templates allow servers to expose parameterized resources using URI
+	 * templates. Arguments may be auto-completed through the completion API.
+	 *
+	 * Retrieves a paginated list of resource templates provided by the server.
+	 * @param cursor Optional pagination cursor from a previous list request
+	 * @param meta Optional metadata to include in the request (_meta field)
+	 * @return The list of resource templates result.
+	 */
+	public McpSchema.ListResourceTemplatesResult listResourceTemplates(String cursor, Map<String, Object> meta) {
+		return withProvidedContext(this.delegate.listResourceTemplates(cursor, meta)).block();
+
+	}
+
 	/**
 	 * Subscriptions. The protocol supports optional subscriptions to resource changes.
 	 * Clients can subscribe to specific resources and receive notifications when they
@@ -370,6 +408,17 @@ public class McpSyncClient implements AutoCloseable {
 
 	}
 
+	/**
+	 * Retrieves a paginated list of prompts provided by the server.
+	 * @param cursor Optional pagination cursor from a previous list request
+	 * @param meta Optional metadata to include in the request (_meta field)
+	 * @return The list of prompts result.
+	 */
+	public ListPromptsResult listPrompts(String cursor, Map<String, Object> meta) {
+		return withProvidedContext(this.delegate.listPrompts(cursor, meta)).block();
+
+	}
+
 	public GetPromptResult getPrompt(GetPromptRequest getPromptRequest) {
 		return withProvidedContext(this.delegate.getPrompt(getPromptRequest)).block();
 	}

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/DefaultSseMessageEndpointValidator.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2026-2026 the original author or authors.
+ */
+package io.modelcontextprotocol.client.transport;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+import io.modelcontextprotocol.util.Assert;
+
+/**
+ * Default {@link SseMessageEndpointValidator} that validates the {@code message} endpoint
+ * advertised by an SSE server. Message endpoints must either have the same origin as the
+ * SSE uri, or be a relative uri.
+ *
+ * @author Daniel Garnier-Moiroux
+ */
+public final class DefaultSseMessageEndpointValidator implements SseMessageEndpointValidator {
+
+	@Override
+	public void validate(URI sseUri, String messageEndpoint) throws InvalidSseMessageEndpointException {
+		Assert.hasText(messageEndpoint, "messageEndpoint must not be empty");
+
+		URI endpointUri;
+		try {
+			endpointUri = new URI(messageEndpoint);
+		}
+		catch (URISyntaxException ex) {
+			throw new InvalidSseMessageEndpointException("messageEndpoint is not a valid URI: " + ex.getMessage(),
+					messageEndpoint);
+		}
+
+		if (endpointUri.isAbsolute() || endpointUri.getRawAuthority() != null) {
+			String scheme = endpointUri.getScheme();
+			String host = endpointUri.getHost();
+			int port = endpointUri.getPort();
+
+			boolean sameScheme = scheme != null && scheme.equalsIgnoreCase(sseUri.getScheme());
+			boolean sameHost = host != null && host.equalsIgnoreCase(sseUri.getHost());
+			boolean samePort = port == sseUri.getPort();
+
+			if (!sameScheme || !sameHost || !samePort) {
+				throw new InvalidSseMessageEndpointException(
+						"messageEndpoint must be a relative path or a same-origin URI", messageEndpoint);
+			}
+		}
+
+		// Exclude path-traversal
+		String decodedPath = endpointUri.getPath();
+		if (decodedPath != null) {
+			for (String segment : decodedPath.split("/", -1)) {
+				if (".".equals(segment) || "..".equals(segment)) {
+					throw new InvalidSseMessageEndpointException(
+							"messageEndpoint must not contain path-traversal segments", messageEndpoint);
+				}
+			}
+		}
+
+	}
+
+}

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransport.java

@@ -16,8 +16,6 @@ import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent;
 import io.modelcontextprotocol.client.transport.customizer.McpAsyncHttpClientRequestCustomizer;
 import io.modelcontextprotocol.client.transport.customizer.McpSyncHttpClientRequestCustomizer;
@@ -33,6 +31,8 @@ import io.modelcontextprotocol.spec.McpTransportException;
 import io.modelcontextprotocol.spec.ProtocolVersions;
 import io.modelcontextprotocol.util.Assert;
 import io.modelcontextprotocol.util.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -82,6 +82,11 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 	/** Default SSE endpoint path */
 	private static final String DEFAULT_SSE_ENDPOINT = "/sse";
 
+	/**
+	 * Default maximum number of bytes read for a single inbound message.
+	 */
+	private static final int DEFAULT_MAX_RESPONSE_SIZE = 16 * 1024 * 1024; // 16MiB
+
 	/** Base URI for the MCP server */
 	private final URI baseUri;
 
@@ -117,6 +122,17 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 	 */
 	private final McpAsyncHttpClientRequestCustomizer httpRequestCustomizer;
 
+	/**
+	 * Validator for the message endpoint;
+	 */
+	private final SseMessageEndpointValidator messageEndpointValidator;
+
+	/**
+	 * Maximum number of bytes read for a single inbound message, whether it arrives on
+	 * the SSE stream or as the response to a posted message.
+	 */
+	private final int maxResponseSize;
+
 	/**
 	 * Creates a new transport instance with custom HTTP client builder, object mapper,
 	 * and headers.
@@ -127,22 +143,30 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 	 * @param jsonMapper the object mapper for JSON serialization/deserialization
 	 * @param httpRequestCustomizer customizer for the requestBuilder before executing
 	 * requests
+	 * @param messageEndpointValidator validator for the message endpoint
+	 * @param maxResponseSize the maximum number of bytes read for a single inbound
+	 * message
 	 * @throws IllegalArgumentException if objectMapper, clientBuilder, or headers is null
 	 */
 	HttpClientSseClientTransport(HttpClient httpClient, HttpRequest.Builder requestBuilder, String baseUri,
-			String sseEndpoint, McpJsonMapper jsonMapper, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer) {
+			String sseEndpoint, McpJsonMapper jsonMapper, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer,
+			SseMessageEndpointValidator messageEndpointValidator, int maxResponseSize) {
 		Assert.notNull(jsonMapper, "jsonMapper must not be null");
 		Assert.hasText(baseUri, "baseUri must not be empty");
 		Assert.hasText(sseEndpoint, "sseEndpoint must not be empty");
 		Assert.notNull(httpClient, "httpClient must not be null");
 		Assert.notNull(requestBuilder, "requestBuilder must not be null");
 		Assert.notNull(httpRequestCustomizer, "httpRequestCustomizer must not be null");
+		Assert.notNull(messageEndpointValidator, "messageEndpointValidator must not be null");
+		Assert.isTrue(maxResponseSize > 0, "maxResponseSize must be positive");
 		this.baseUri = URI.create(baseUri);
 		this.sseEndpoint = sseEndpoint;
 		this.jsonMapper = jsonMapper;
 		this.httpClient = httpClient;
 		this.requestBuilder = requestBuilder;
 		this.httpRequestCustomizer = httpRequestCustomizer;
+		this.messageEndpointValidator = messageEndpointValidator;
+		this.maxResponseSize = maxResponseSize;
 	}
 
 	@Override
@@ -178,6 +202,10 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 
 		private Duration connectTimeout = Duration.ofSeconds(10);
 
+		private SseMessageEndpointValidator messageEndpointValidator = new DefaultSseMessageEndpointValidator();
+
+		private int maxResponseSize = DEFAULT_MAX_RESPONSE_SIZE;
+
 		/**
 		 * Creates a new builder instance.
 		 */
@@ -241,10 +269,17 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 		}
 
 		/**
-		 * Customizes the HTTP client builder.
+		 * Applies the given consumer to the shared {@link HttpRequest.Builder} <b>once,
+		 * at build time</b>. Any headers set here are frozen into the template and
+		 * <b>cannot be updated</b> after the transport is built.
 		 * @param requestCustomizer the consumer to customize the HTTP request builder
 		 * @return this builder
+		 * @deprecated Use {@link #requestBuilder(HttpRequest.Builder)} for stable
+		 * headers, or {@link #httpRequestCustomizer(McpSyncHttpClientRequestCustomizer)}
+		 * / {@link #asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer)} for
+		 * dynamic per-request customization.
 		 */
+		@Deprecated
 		public Builder customizeRequest(final Consumer<HttpRequest.Builder> requestCustomizer) {
 			Assert.notNull(requestCustomizer, "requestCustomizer must not be null");
 			requestCustomizer.accept(requestBuilder);
@@ -308,6 +343,38 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 			return this;
 		}
 
+		/**
+		 * Sets the validator that ensure the message endpoint returned over the SSE
+		 * connection is valid.
+		 * @param messageEndpointValidator the validator
+		 * @return this builder
+		 */
+		public Builder messageEndpointValidator(SseMessageEndpointValidator messageEndpointValidator) {
+			Assert.notNull(messageEndpointValidator, "messageEndpointValidator must not be null");
+			this.messageEndpointValidator = messageEndpointValidator;
+			return this;
+		}
+
+		/**
+		 * Sets the maximum number of bytes read for a single inbound message, whether it
+		 * arrives on the SSE stream or as the response to a posted message. A peer that
+		 * sends a larger message (or never terminates one) has its stream aborted instead
+		 * of forcing the transport to buffer it in memory. Defaults to 16MiB.
+		 *
+		 * <p>
+		 * The bound applies per message, not to the stream as a whole: a long-lived SSE
+		 * stream may deliver any number of messages, each up to this size. SSE field
+		 * framing is allowed a small amount of headroom on top of this size, so a message
+		 * of exactly this many bytes is still accepted.
+		 * @param maxResponseSize the maximum inbound message size, in bytes
+		 * @return this builder
+		 */
+		public Builder maxResponseSize(int maxResponseSize) {
+			Assert.isTrue(maxResponseSize > 0, "maxResponseSize must be positive");
+			this.maxResponseSize = maxResponseSize;
+			return this;
+		}
+
 		/**
 		 * Builds a new {@link HttpClientSseClientTransport} instance.
 		 * @return a new transport instance
@@ -315,7 +382,8 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 		public HttpClientSseClientTransport build() {
 			HttpClient httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build();
 			return new HttpClientSseClientTransport(httpClient, requestBuilder, baseUri, sseEndpoint,
-					jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper, httpRequestCustomizer);
+					jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper, httpRequestCustomizer,
+					messageEndpointValidator, maxResponseSize);
 		}
 
 	}
@@ -334,13 +402,15 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 			var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
 			return Mono.from(this.httpRequestCustomizer.customize(builder, "GET", uri, null, transportContext));
 		}).flatMap(requestBuilder -> Mono.create(sink -> {
-			Disposable connection = Flux.<ResponseEvent>create(sseSink -> this.httpClient
-				.sendAsync(requestBuilder.build(),
-						responseInfo -> ResponseSubscribers.sseToBodySubscriber(responseInfo, sseSink))
-				.exceptionallyCompose(e -> {
-					sseSink.error(e);
-					return CompletableFuture.failedFuture(e);
-				}))
+			Disposable connection = Flux.<ResponseEvent>create(
+					sseSink -> this.httpClient
+						.sendAsync(requestBuilder.build(),
+								responseInfo -> ResponseSubscribers.sseToBodySubscriber(responseInfo, sseSink,
+										this.maxResponseSize))
+						.exceptionallyCompose(e -> {
+							sseSink.error(e);
+							return CompletableFuture.failedFuture(e);
+						}))
 				.map(responseEvent -> (ResponseSubscribers.SseResponseEvent) responseEvent)
 				.flatMap(responseEvent -> {
 					if (isClosing) {
@@ -353,6 +423,14 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 						try {
 							if (ENDPOINT_EVENT_TYPE.equals(responseEvent.sseEvent().event())) {
 								String messageEndpointUri = responseEvent.sseEvent().data();
+								try {
+									messageEndpointValidator.validate(uri, messageEndpointUri);
+								}
+								catch (InvalidSseMessageEndpointException e) {
+									sink.error(e);
+									this.messageEndpointSink.tryEmitError(e);
+									return Flux.error(e);
+								}
 								if (this.messageEndpointSink.tryEmitValue(messageEndpointUri).isSuccess()) {
 									sink.success();
 									return Flux.empty(); // No further processing needed
@@ -456,14 +534,15 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 		return Mono.deferContextual(ctx -> {
 			var builder = this.requestBuilder.copy()
 				.uri(requestUri)
-				.header(HttpHeaders.CONTENT_TYPE, "application/json")
+				.header(HttpHeaders.CONTENT_TYPE, "application/json; charset=utf-8")
 				.header(MCP_PROTOCOL_VERSION_HEADER_NAME, MCP_PROTOCOL_VERSION)
 				.POST(HttpRequest.BodyPublishers.ofString(body));
 			var transportContext = ctx.getOrDefault(McpTransportContext.KEY, McpTransportContext.EMPTY);
 			return Mono.from(this.httpRequestCustomizer.customize(builder, "POST", requestUri, body, transportContext));
 		}).flatMap(customizedBuilder -> {
 			var request = customizedBuilder.build();
-			return Mono.fromFuture(httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()));
+			return Mono.fromFuture(
+					httpClient.sendAsync(request, ResponseSubscribers.boundedStringBodyHandler(this.maxResponseSize)));
 		});
 	}
 

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java

@@ -8,7 +8,6 @@ import java.io.IOException;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
 import java.net.http.HttpResponse.BodyHandler;
 import java.time.Duration;
 import java.util.Collections;
@@ -85,6 +84,11 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 
 	private static final String DEFAULT_ENDPOINT = "/mcp";
 
+	/**
+	 * Default maximum number of bytes read for a single inbound message.
+	 */
+	private static final int DEFAULT_MAX_RESPONSE_SIZE = 16 * 1024 * 1024; // 16MiB
+
 	/**
 	 * HTTP client for sending messages to the server. Uses HTTP POST over the message
 	 * endpoint
@@ -102,6 +106,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 
 	private static final String APPLICATION_JSON = "application/json";
 
+	private static final String APPLICATION_JSON_UTF8 = "application/json; charset=utf-8";
+
 	private static final String TEXT_EVENT_STREAM = "text/event-stream";
 
 	public static int NOT_FOUND = 404;
@@ -134,10 +140,18 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 
 	private final String latestSupportedProtocolVersion;
 
+	/**
+	 * Maximum number of bytes read for a single inbound message, whether it arrives on an
+	 * SSE stream or as a JSON response body.
+	 */
+	private final int maxResponseSize;
+
 	private HttpClientStreamableHttpTransport(McpJsonMapper jsonMapper, HttpClient httpClient,
 			HttpRequest.Builder requestBuilder, String baseUri, String endpoint, boolean resumableStreams,
 			boolean openConnectionOnStartup, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer,
-			McpHttpClientAuthorizationErrorHandler authorizationErrorHandler, List<String> supportedProtocolVersions) {
+			McpHttpClientAuthorizationErrorHandler authorizationErrorHandler, List<String> supportedProtocolVersions,
+			int maxResponseSize) {
+		Assert.isTrue(maxResponseSize > 0, "maxResponseSize must be positive");
 		this.jsonMapper = jsonMapper;
 		this.httpClient = httpClient;
 		this.requestBuilder = requestBuilder;
@@ -153,6 +167,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 			.sorted(Comparator.reverseOrder())
 			.findFirst()
 			.get();
+		this.maxResponseSize = maxResponseSize;
 	}
 
 	@Override
@@ -209,7 +224,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 			return Mono.from(this.httpRequestCustomizer.customize(builder, "DELETE", uri, null, transportContext));
 		}).flatMap(requestBuilder -> {
 			var request = requestBuilder.build();
-			return Mono.fromFuture(() -> this.httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()));
+			return Mono.fromFuture(() -> this.httpClient.sendAsync(request,
+					ResponseSubscribers.boundedStringBodyHandler(this.maxResponseSize)));
 		}).then();
 	}
 
@@ -298,6 +314,10 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 											"Authorization error connecting to SSE stream",
 											responseEvent.responseInfo()));
 						}
+						else if (statusCode == METHOD_NOT_ALLOWED) {
+							logger.debug("The server does not support SSE streams, using request-response mode.");
+							return Flux.empty();
+						}
 
 						if (!(responseEvent instanceof ResponseSubscribers.SseResponseEvent sseResponseEvent)) {
 							return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
@@ -344,10 +364,6 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 								return Flux.empty();
 							}
 						}
-						else if (statusCode == METHOD_NOT_ALLOWED) { // NotAllowed
-							logger.debug("The server does not support SSE streams, using request-response mode.");
-							return Flux.empty();
-						}
 						else if (statusCode == NOT_FOUND) {
 
 							if (transportSession != null && transportSession.sessionId().isPresent()) {
@@ -432,16 +448,16 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 			if (contentType.contains(TEXT_EVENT_STREAM)) {
 				// For SSE streams, use line subscriber that returns Void
 				logger.debug("Received SSE stream response, using line subscriber");
-				return ResponseSubscribers.sseToBodySubscriber(responseInfo, sink);
+				return ResponseSubscribers.sseToBodySubscriber(responseInfo, sink, this.maxResponseSize);
 			}
 			else if (contentType.contains(APPLICATION_JSON)) {
 				// For JSON responses and others, use string subscriber
 				logger.debug("Received response, using string subscriber");
-				return ResponseSubscribers.aggregateBodySubscriber(responseInfo, sink);
+				return ResponseSubscribers.aggregateBodySubscriber(responseInfo, sink, this.maxResponseSize);
 			}
 
 			logger.debug("Received Bodyless response, using discarding subscriber");
-			return ResponseSubscribers.bodilessBodySubscriber(responseInfo, sink);
+			return ResponseSubscribers.bodilessBodySubscriber(responseInfo, sink, this.maxResponseSize);
 		};
 
 		return responseBodyHandler;
@@ -477,7 +493,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 
 				var builder = requestBuilder.uri(uri)
 					.header(HttpHeaders.ACCEPT, APPLICATION_JSON + ", " + TEXT_EVENT_STREAM)
-					.header(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON)
+					.header(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON_UTF8)
 					.header(HttpHeaders.CACHE_CONTROL, "no-cache")
 					.header(HttpHeaders.PROTOCOL_VERSION,
 							ctx.getOrDefault(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION,
@@ -629,6 +645,10 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 					return Flux.<McpSchema.JSONRPCMessage>error(new McpTransportException(
 							"Bad Request. Status code:" + statusCode + ", response-event:" + responseEvent));
 				}
+				else if (statusCode >= 400 && statusCode < 500) {
+					return Flux.<McpSchema.JSONRPCMessage>error(
+							new McpTransportException("Invalid request. Status code: " + statusCode));
+				}
 
 				return Flux.<McpSchema.JSONRPCMessage>error(
 						new RuntimeException("Failed to send message: " + responseEvent));
@@ -695,6 +715,8 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 
 		private McpHttpClientAuthorizationErrorHandler authorizationErrorHandler = McpHttpClientAuthorizationErrorHandler.NOOP;
 
+		private int maxResponseSize = DEFAULT_MAX_RESPONSE_SIZE;
+
 		/**
 		 * Creates a new builder with the specified base URI.
 		 * @param baseUri the base URI of the MCP server
@@ -738,10 +760,17 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 		}
 
 		/**
-		 * Customizes the HTTP client builder.
+		 * Applies the given consumer to the shared {@link HttpRequest.Builder} <b>once,
+		 * at build time</b>. Any headers set here are frozen into the template and
+		 * <b>cannot be updated</b> after the transport is built.
 		 * @param requestCustomizer the consumer to customize the HTTP request builder
 		 * @return this builder
+		 * @deprecated Use {@link #requestBuilder(HttpRequest.Builder)} for stable
+		 * headers, or {@link #httpRequestCustomizer(McpSyncHttpClientRequestCustomizer)}
+		 * / {@link #asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer)} for
+		 * dynamic per-request customization.
 		 */
+		@Deprecated
 		public Builder customizeRequest(final Consumer<HttpRequest.Builder> requestCustomizer) {
 			Assert.notNull(requestCustomizer, "requestCustomizer must not be null");
 			requestCustomizer.accept(requestBuilder);
@@ -878,6 +907,26 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 			return this;
 		}
 
+		/**
+		 * Sets the maximum number of bytes read for a single inbound message, whether it
+		 * arrives on an SSE stream or as a JSON response body. A peer that sends a larger
+		 * message (or never terminates one) has its stream aborted instead of forcing the
+		 * transport to buffer it in memory. Defaults to 16MiB.
+		 *
+		 * <p>
+		 * The bound applies per message, not to the stream as a whole: a long-lived SSE
+		 * stream may deliver any number of messages, each up to this size. SSE field
+		 * framing is allowed a small amount of headroom on top of this size, so a message
+		 * of exactly this many bytes is still accepted.
+		 * @param maxResponseSize the maximum inbound message size, in bytes
+		 * @return this builder
+		 */
+		public Builder maxResponseSize(int maxResponseSize) {
+			Assert.isTrue(maxResponseSize > 0, "maxResponseSize must be positive");
+			this.maxResponseSize = maxResponseSize;
+			return this;
+		}
+
 		/**
 		 * Construct a fresh instance of {@link HttpClientStreamableHttpTransport} using
 		 * the current builder configuration.
@@ -887,7 +936,7 @@ public class HttpClientStreamableHttpTransport implements McpClientTransport {
 			HttpClient httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build();
 			return new HttpClientStreamableHttpTransport(jsonMapper == null ? McpJsonDefaults.getMapper() : jsonMapper,
 					httpClient, requestBuilder, baseUri, endpoint, resumableStreams, openConnectionOnStartup,
-					httpRequestCustomizer, authorizationErrorHandler, supportedProtocolVersions);
+					httpRequestCustomizer, authorizationErrorHandler, supportedProtocolVersions, maxResponseSize);
 		}
 
 	}

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/InvalidSseMessageEndpointException.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2026-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client.transport;
+
+/**
+ * Exception thrown when the {@code message} endpoint returned from the SSE connection is
+ * not valid.
+ *
+ * @author Daniel Garnier-Moiroux
+ */
+public class InvalidSseMessageEndpointException extends Exception {
+
+	private final String messageEndpoint;
+
+	public InvalidSseMessageEndpointException(String message, String messageEndpoint) {
+		super(message);
+		this.messageEndpoint = messageEndpoint;
+	}
+
+	public String getMessageEndpoint() {
+		return messageEndpoint;
+	}
+
+}

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java

@@ -5,8 +5,13 @@
 package io.modelcontextprotocol.client.transport;
 
 import java.net.http.HttpResponse;
+import java.net.http.HttpResponse.BodyHandler;
 import java.net.http.HttpResponse.BodySubscriber;
 import java.net.http.HttpResponse.ResponseInfo;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.regex.Pattern;
 
@@ -30,11 +35,21 @@ import reactor.core.publisher.FluxSink;
  *
  * @author Christian Tzolov
  * @author Dariusz Jędrzejczyk
+ * @author Daniel Garnier-Moiroux
  */
 class ResponseSubscribers {
 
 	private static final Logger logger = LoggerFactory.getLogger(ResponseSubscribers.class);
 
+	/**
+	 * Bytes of SSE field framing a single line may carry on top of the message payload:
+	 * {@code "event: "} is the longest field prefix this parser recognises. Line
+	 * terminators are not counted, as they reset the running line length. Without this
+	 * allowance, an event carrying exactly the maximum message size would be rejected
+	 * because of the bytes the SSE wire format adds around it.
+	 */
+	private static final int SSE_FRAMING_OVERHEAD = "event: ".length();
+
 	record SseEvent(String id, String event, String data) {
 	}
 
@@ -54,19 +69,80 @@ class ResponseSubscribers {
 	record AggregateResponseEvent(ResponseInfo responseInfo, String data) implements ResponseEvent {
 	}
 
-	static BodySubscriber<Void> sseToBodySubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink) {
-		return HttpResponse.BodySubscribers
-			.fromLineSubscriber(FlowAdapters.toFlowSubscriber(new SseLineSubscriber(responseInfo, sink)));
+	/**
+	 * Creates a {@link BodySubscriber} that parses a Server-Sent Events stream, bounding
+	 * how much memory a single inbound message may occupy. Both the size of an individual
+	 * line (as read off the wire before a terminator is seen) and the accumulated size of
+	 * a multi-line SSE event are capped at {@code maxSize}; a peer exceeding either limit
+	 * has its stream aborted instead of forcing the transport to buffer it in memory. The
+	 * line bound is allowed {@link #SSE_FRAMING_OVERHEAD} extra bytes so that the SSE
+	 * framing around a payload does not count against the payload's own budget.
+	 * @param responseInfo the HTTP response information
+	 * @param sink the sink to emit parsed events to
+	 * @param maxSize the maximum number of bytes read for a single inbound message
+	 */
+	static BodySubscriber<Void> sseToBodySubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink,
+			int maxSize) {
+		BodySubscriber<Void> lineSubscriber = HttpResponse.BodySubscribers
+			.fromLineSubscriber(FlowAdapters.toFlowSubscriber(new SseLineSubscriber(responseInfo, sink, maxSize)));
+		return new BoundedLineBodySubscriber(lineSubscriber, plusFramingOverhead(maxSize));
 	}
 
-	static BodySubscriber<Void> aggregateBodySubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink) {
-		return HttpResponse.BodySubscribers
-			.fromLineSubscriber(FlowAdapters.toFlowSubscriber(new AggregateSubscriber(responseInfo, sink)));
+	/**
+	 * Adds {@link #SSE_FRAMING_OVERHEAD} to {@code maxSize}, saturating at
+	 * {@link Integer#MAX_VALUE} rather than overflowing into a negative bound that would
+	 * reject everything.
+	 */
+	private static int plusFramingOverhead(int maxSize) {
+		return maxSize > Integer.MAX_VALUE - SSE_FRAMING_OVERHEAD ? Integer.MAX_VALUE : maxSize + SSE_FRAMING_OVERHEAD;
 	}
 
-	static BodySubscriber<Void> bodilessBodySubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink) {
-		return HttpResponse.BodySubscribers
+	/**
+	 * Creates a {@link BodySubscriber} that aggregates the whole response body into a
+	 * single event, bounding how much memory it may occupy. Both the size of an
+	 * individual line (as read off the wire before a terminator is seen) and the total
+	 * accumulated body are capped at {@code maxSize}; a peer exceeding either limit has
+	 * its response aborted instead of forcing the transport to buffer it in memory.
+	 * @param responseInfo the HTTP response information
+	 * @param sink the sink to emit the aggregated event to
+	 * @param maxSize the maximum number of bytes read for the response body
+	 */
+	static BodySubscriber<Void> aggregateBodySubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink,
+			int maxSize) {
+		BodySubscriber<Void> lineSubscriber = HttpResponse.BodySubscribers
+			.fromLineSubscriber(FlowAdapters.toFlowSubscriber(new AggregateSubscriber(responseInfo, sink, maxSize)));
+		return new BoundedLineBodySubscriber(lineSubscriber, maxSize);
+	}
+
+	/**
+	 * Creates a {@link BodySubscriber} that discards the response body, bounding how much
+	 * memory reading it may occupy. The body is discarded as it arrives, but the
+	 * underlying line subscriber still buffers each line before handing it over, so a
+	 * peer sending a line longer than {@code maxSize} has its response aborted.
+	 * @param responseInfo the HTTP response information
+	 * @param sink the sink to emit the completion event to
+	 * @param maxSize the maximum number of bytes read for a single line
+	 */
+	static BodySubscriber<Void> bodilessBodySubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink,
+			int maxSize) {
+		BodySubscriber<Void> lineSubscriber = HttpResponse.BodySubscribers
 			.fromLineSubscriber(FlowAdapters.toFlowSubscriber(new BodilessResponseLineSubscriber(responseInfo, sink)));
+		return new BoundedLineBodySubscriber(lineSubscriber, maxSize);
+	}
+
+	/**
+	 * Creates a {@link BodyHandler} that reads the response body into a string, bounding
+	 * how much memory it may occupy. A peer sending more than {@code maxSize} bytes has
+	 * its response aborted instead of forcing the transport to buffer it in memory.
+	 *
+	 * <p>
+	 * Decoding matches {@link HttpResponse.BodyHandlers#ofString()}, including its
+	 * handling of the charset declared in the {@code Content-Type} header.
+	 * @param maxSize the maximum number of bytes read for the response body
+	 */
+	static BodyHandler<String> boundedStringBodyHandler(int maxSize) {
+		BodyHandler<String> delegate = HttpResponse.BodyHandlers.ofString();
+		return responseInfo -> new BoundedTotalBodySubscriber<>(delegate.apply(responseInfo), maxSize);
 	}
 
 	static class SseLineSubscriber extends BaseSubscriber<String> {
@@ -112,18 +188,30 @@ class ResponseSubscribers {
 		 */
 		private ResponseInfo responseInfo;
 
+		/**
+		 * The maximum number of bytes that may accumulate for a single SSE event. A peer
+		 * that never terminates an event (e.g. an endless stream of {@code data:} lines)
+		 * has its stream aborted instead of exhausting memory. The accumulated data is
+		 * measured in characters, which for UTF-8 is never more than the number of bytes
+		 * it was decoded from.
+		 */
+		private final int maxSize;
+
 		/**
 		 * Creates a new LineSubscriber that will emit parsed SSE events to the provided
 		 * sink.
 		 * @param sink the {@link FluxSink} to emit parsed {@link ResponseEvent} objects
 		 * to
+		 * @param maxSize the maximum number of bytes that may accumulate for a single SSE
+		 * event
 		 */
-		public SseLineSubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink) {
+		public SseLineSubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink, int maxSize) {
 			this.sink = sink;
 			this.eventBuilder = new StringBuilder();
 			this.currentEventId = new AtomicReference<>();
 			this.currentEventType = new AtomicReference<>();
 			this.responseInfo = responseInfo;
+			this.maxSize = maxSize;
 		}
 
 		@Override
@@ -155,7 +243,18 @@ class ResponseSubscribers {
 				if (line.startsWith("data:")) {
 					var matcher = EVENT_DATA_PATTERN.matcher(line);
 					if (matcher.find()) {
-						this.eventBuilder.append(matcher.group(1).trim()).append("\n");
+						String data = matcher.group(1).trim();
+						// Measured before appending, so that an event carrying exactly
+						// maxSize of data is accepted: the trailing separator below is
+						// stripped again before the event is emitted.
+						if (this.eventBuilder.length() + data.length() > this.maxSize) {
+							upstream().cancel();
+							this.sink.error(
+									new McpTransportException("Inbound SSE event exceeds the maximum allowed size of "
+											+ this.maxSize + " bytes"));
+							return;
+						}
+						this.eventBuilder.append(data).append("\n");
 					}
 					upstream().request(1);
 				}
@@ -225,15 +324,26 @@ class ResponseSubscribers {
 
 		volatile boolean hasRequestedDemand = false;
 
+		/**
+		 * The maximum number of bytes that may accumulate for the aggregated response
+		 * body. A peer that sends a larger body has its response aborted instead of
+		 * exhausting memory. The accumulated body is measured in characters, which for
+		 * UTF-8 is never more than the number of bytes it was decoded from.
+		 */
+		private final int maxSize;
+
 		/**
 		 * Creates a new JsonLineSubscriber that will emit parsed JSON-RPC messages.
 		 * @param sink the {@link FluxSink} to emit parsed {@link ResponseEvent} objects
 		 * to
+		 * @param maxSize the maximum number of bytes that may accumulate for the
+		 * aggregated response body
 		 */
-		public AggregateSubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink) {
+		public AggregateSubscriber(ResponseInfo responseInfo, FluxSink<ResponseEvent> sink, int maxSize) {
 			this.sink = sink;
 			this.eventBuilder = new StringBuilder();
 			this.responseInfo = responseInfo;
+			this.maxSize = maxSize;
 		}
 
 		@Override
@@ -252,6 +362,15 @@ class ResponseSubscribers {
 
 		@Override
 		protected void hookOnNext(String line) {
+			// Measured before appending, so that a body of exactly maxSize is accepted.
+			// The separator this adds back for each line stands in for the terminator the
+			// peer sent, which the line subscriber has already stripped.
+			if (this.eventBuilder.length() + line.length() > this.maxSize) {
+				upstream().cancel();
+				this.sink.error(new McpTransportException(
+						"Inbound response body exceeds the maximum allowed size of " + this.maxSize + " bytes"));
+				return;
+			}
 			this.eventBuilder.append(line).append("\n");
 		}
 
@@ -324,4 +443,177 @@ class ResponseSubscribers {
 
 	}
 
+	/**
+	 * Base for {@link BodySubscriber} wrappers that transparently forward the response
+	 * body to a delegate, but abort it once the peer exceeds a size bound.
+	 *
+	 * <p>
+	 * Aborting cancels the upstream subscription, which closes the connection, and
+	 * signals a {@link McpTransportException} to the delegate so the failure surfaces
+	 * both through the body's {@link CompletionStage} and through any sink the delegate
+	 * feeds.
+	 */
+	abstract static class BoundedBodySubscriber<T> implements BodySubscriber<T> {
+
+		private final BodySubscriber<T> delegate;
+
+		protected final int maxSize;
+
+		/**
+		 * What the bound applies to, e.g. {@code "Inbound line"}, used to build the
+		 * failure message.
+		 */
+		private final String boundedEntity;
+
+		private Flow.Subscription subscription;
+
+		private volatile boolean done = false;
+
+		BoundedBodySubscriber(BodySubscriber<T> delegate, int maxSize, String boundedEntity) {
+			this.delegate = delegate;
+			this.maxSize = maxSize;
+			this.boundedEntity = boundedEntity;
+		}
+
+		@Override
+		public CompletionStage<T> getBody() {
+			return this.delegate.getBody();
+		}
+
+		@Override
+		public void onSubscribe(Flow.Subscription subscription) {
+			this.subscription = subscription;
+			this.delegate.onSubscribe(subscription);
+		}
+
+		@Override
+		public void onNext(List<ByteBuffer> buffers) {
+			if (this.done) {
+				return;
+			}
+			for (ByteBuffer buffer : buffers) {
+				if (!checkSize(buffer)) {
+					this.done = true;
+					this.subscription.cancel();
+					this.delegate.onError(new McpTransportException(
+							this.boundedEntity + " exceeds the maximum allowed size of " + this.maxSize + " bytes"));
+					return;
+				}
+			}
+			this.delegate.onNext(buffers);
+		}
+
+		/**
+		 * Accounts for the bytes in {@code buffer}, which must be inspected with absolute
+		 * reads only so the delegate still sees the original position.
+		 * @param buffer the buffer about to be handed to the delegate
+		 * @return {@code true} to accept the buffer, or {@code false} to abort the
+		 * response because the bound has been exceeded
+		 */
+		protected abstract boolean checkSize(ByteBuffer buffer);
+
+		@Override
+		public void onError(Throwable throwable) {
+			if (this.done) {
+				return;
+			}
+			this.done = true;
+			this.delegate.onError(throwable);
+		}
+
+		@Override
+		public void onComplete() {
+			if (this.done) {
+				return;
+			}
+			this.done = true;
+			this.delegate.onComplete();
+		}
+
+	}
+
+	/**
+	 * A {@link BoundedBodySubscriber} that aborts the response once a single line (a run
+	 * of bytes with no CR/LF terminator) exceeds {@code maxSize} bytes.
+	 *
+	 * <p>
+	 * {@link HttpResponse.BodySubscribers#fromLineSubscriber} buffers characters until it
+	 * encounters a line terminator, so a peer that never terminates a line (or sends an
+	 * enormous one) would force the transport to buffer it in memory. This wrapper counts
+	 * bytes as they arrive off the wire and cancels the subscription before that buffer
+	 * can grow without bound.
+	 */
+	static final class BoundedLineBodySubscriber extends BoundedBodySubscriber<Void> {
+
+		private long bytesSinceLineTerminator = 0;
+
+		BoundedLineBodySubscriber(BodySubscriber<Void> delegate, int maxSize) {
+			super(delegate, maxSize, "Inbound line");
+		}
+
+		@Override
+		protected boolean checkSize(ByteBuffer buffer) {
+			int position = buffer.position();
+			int limit = buffer.limit();
+			if (position == limit) {
+				return true;
+			}
+			if (this.bytesSinceLineTerminator + (limit - position) <= this.maxSize) {
+				// No line ending in this buffer can exceed the limit, because there are
+				// not enough bytes since the last terminator for one to. Only the
+				// trailing (still unterminated) run matters, so scan back to the last
+				// terminator instead of walking every byte.
+				this.bytesSinceLineTerminator = lengthOfTrailingRun(buffer, position, limit);
+				return true;
+			}
+			// The limit is within reach, so account for every line exactly.
+			for (int i = position; i < limit; i++) {
+				byte b = buffer.get(i);
+				if (b == '\n' || b == '\r') {
+					this.bytesSinceLineTerminator = 0;
+				}
+				else if (++this.bytesSinceLineTerminator > this.maxSize) {
+					return false;
+				}
+			}
+			return true;
+		}
+
+		/**
+		 * Returns the number of bytes after the last line terminator in the buffer, or
+		 * the whole span added to the running count when the buffer holds no terminator.
+		 */
+		private long lengthOfTrailingRun(ByteBuffer buffer, int position, int limit) {
+			for (int i = limit - 1; i >= position; i--) {
+				byte b = buffer.get(i);
+				if (b == '\n' || b == '\r') {
+					return limit - 1 - i;
+				}
+			}
+			return this.bytesSinceLineTerminator + (limit - position);
+		}
+
+	}
+
+	/**
+	 * A {@link BoundedBodySubscriber} that aborts the response once the body as a whole
+	 * exceeds {@code maxSize} bytes. Suitable for delegates that aggregate the entire
+	 * body in memory, such as {@link HttpResponse.BodyHandlers#ofString()}.
+	 */
+	static final class BoundedTotalBodySubscriber<T> extends BoundedBodySubscriber<T> {
+
+		private long totalBytes = 0;
+
+		BoundedTotalBodySubscriber(BodySubscriber<T> delegate, int maxSize) {
+			super(delegate, maxSize, "Inbound response body");
+		}
+
+		@Override
+		protected boolean checkSize(ByteBuffer buffer) {
+			this.totalBytes += buffer.remaining();
+			return this.totalBytes <= this.maxSize;
+		}
+
+	}
+
 }

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/SseMessageEndpointValidator.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2026-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client.transport;
+
+import java.net.URI;
+
+/**
+ * Validate the that message endpoint in the SSE transport is valid. Throws
+ * {@link InvalidSseMessageEndpointException} when then endpoint is not valid.
+ *
+ * @author Daniel Garnier-Moiroux
+ */
+@FunctionalInterface
+public interface SseMessageEndpointValidator {
+
+	/**
+	 * Validate the message endpoint coming from an SSE connection. Throws if not valid.
+	 * @param sseUri the URI used to establish the SSE connection
+	 * @param messageEndpoint the message endpoint from the SSE connection
+	 * @throws InvalidSseMessageEndpointException error thrown if the message endpoint is
+	 * not valid.
+	 */
+	void validate(URI sseUri, String messageEndpoint) throws InvalidSseMessageEndpointException;
+
+}