feat(sse): add SSEParserConfig, raise default per-line limit to 1 MB

docs/content/dev/client.md

@@ -185,6 +185,35 @@ Client client = Client
         .build();
 ```
 
+### SSE Parser Configuration
+
+The client uses a Server-Sent Events (SSE) parser for streaming responses. You can tune its limits via `SSEParserConfig` to handle agents that return large payloads (e.g., large artifacts or tool results):
+
+| Parameter        | Description                                              | Default   |
+|------------------|----------------------------------------------------------|-----------|
+| `maxLineLength`  | Max bytes per raw SSE line (`0` = disabled)               | 1 MB      |
+| `maxBufferLines` | Max `data:` lines per event block                        | 1 000     |
+| `maxBufferChars` | Max total characters across all `data:` values per event | 1 MB      |
+
+The defaults are suitable for most deployments. To override them, build a custom `SSEParserConfig` and pass it to the HTTP client.
+
+> **Security note:** setting `maxLineLength` to `0` disables the per-line memory protection entirely. Without this limit, a malicious server sending a single unterminated line (no newline) can consume unbounded memory — `maxBufferChars` only limits accumulated `data:` field values _after_ lines have been decoded, not the raw line itself. Only disable the per-line check when you trust the remote agent or have other safeguards (e.g. a reverse proxy with its own line-length limit). Note that `maxLineLength` is enforced as a byte limit on raw UTF-8 data; for ASCII content (the common case) bytes and characters are equivalent.
+
+```java
+SSEParserConfig sseConfig = SSEParserConfig.builder()
+        .maxLineLength(4 * 1024 * 1024)   // 4 MB per line
+        .maxBufferChars(4 * 1024 * 1024)   // 4 MB per event
+        .build();
+
+// Pass to JdkA2AHttpClient, then to your transport config
+JdkA2AHttpClient httpClient = JdkA2AHttpClient.withSseConfig(sseConfig);
+
+Client client = Client
+        .builder(agentCard)
+        .withTransport(JSONRPCTransport.class, new JSONRPCTransportConfig(httpClient))
+        .build();
+```
+
 ## Observability (Optional)
 
 Add distributed tracing and W3C Trace Context propagation to client calls with the [OpenTelemetry extras modules](extra/opentelemetry#client).

extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java

@@ -1,5 +1,7 @@
 package org.a2aproject.sdk.client.http.android;
 
+import static org.a2aproject.sdk.util.Assert.checkNotNullParam;
+
 import static java.net.HttpURLConnection.HTTP_FORBIDDEN;
 import static java.net.HttpURLConnection.HTTP_MULT_CHOICE;
 import static java.net.HttpURLConnection.HTTP_OK;
@@ -27,7 +29,9 @@
 import org.a2aproject.sdk.client.http.A2AHttpClient;
 import org.a2aproject.sdk.client.http.A2AHttpHeaders;
 import org.a2aproject.sdk.client.http.A2AHttpResponse;
+import org.a2aproject.sdk.client.http.BoundedLineAccumulator;
 import org.a2aproject.sdk.client.http.ServerSentEvent;
+import org.a2aproject.sdk.client.http.SSEParserConfig;
 import org.a2aproject.sdk.client.http.ServerSentEventParser;
 import org.a2aproject.sdk.common.A2AErrorMessages;
 import org.a2aproject.sdk.spec.A2AClientHTTPError;
@@ -48,24 +52,44 @@ public class AndroidA2AHttpClient implements A2AHttpClient {
     return t;
   });
 
+  private final SSEParserConfig sseParserConfig;
+
+  public AndroidA2AHttpClient() {
+    this(SSEParserConfig.DEFAULT);
+  }
+
+  /**
+   * Creates a new Android HTTP client with custom SSE parser limits.
+   *
+   * @param sseParserConfig the SSE parser configuration to use for streaming responses
+   */
+  public AndroidA2AHttpClient(SSEParserConfig sseParserConfig) {
+    this.sseParserConfig = checkNotNullParam("sseParserConfig", sseParserConfig);
+  }
+
   @Override
   public GetBuilder createGet() {
-    return new AndroidGetBuilder();
+    return new AndroidGetBuilder(sseParserConfig);
   }
 
   @Override
   public PostBuilder createPost() {
-    return new AndroidPostBuilder();
+    return new AndroidPostBuilder(sseParserConfig);
   }
 
   @Override
   public DeleteBuilder createDelete() {
-    return new AndroidDeleteBuilder();
+    return new AndroidDeleteBuilder(sseParserConfig);
   }
 
   private abstract static class AndroidBuilder<T extends Builder<T>> implements Builder<T> {
     protected String url = "";
     protected Map<String, String> headers = new HashMap<>();
+    protected final SSEParserConfig sseParserConfig;
+
+    AndroidBuilder(SSEParserConfig sseParserConfig) {
+      this.sseParserConfig = sseParserConfig;
+    }
 
     @Override
     public T url(String url) {
@@ -137,6 +161,52 @@ protected static String readStreamWithLimit(InputStream is) throws IOException {
       }
     }
 
+    /**
+     * Reads the next line from {@code is}, treating LF, CRLF, and bare CR as terminators
+     * per the SSE specification, and enforcing the byte limit <em>while accumulating
+     * bytes</em> — not after a full line has been materialised.
+     *
+     * <p>The caller must provide a {@link BoundedLineAccumulator} and call
+     * {@link BoundedLineAccumulator#reset()} between lines. After this method returns
+     * {@code true}, check {@link BoundedLineAccumulator#isTooLong()} to distinguish
+     * between a valid line and one that exceeded the limit.
+     *
+     * <p>The {@code is} parameter must support {@link InputStream#mark(int)} (e.g.
+     * {@link java.io.BufferedInputStream}) so that CR at end-of-buffer can be resolved
+     * without consuming the following byte.
+     *
+     * @param is the input stream to read from (must support mark/reset)
+     * @param accumulator the accumulator to use for byte collection and limit enforcement
+     * @return {@code true} if a line was read, {@code false} at end-of-stream
+     */
+    static boolean readBoundedLine(InputStream is, BoundedLineAccumulator accumulator) throws IOException {
+      boolean hasReadAnyBytes = false;
+
+      while (true) {
+        int b = is.read();
+        if (b == -1) {
+          return hasReadAnyBytes || accumulator.isTooLong();
+        }
+        hasReadAnyBytes = true;
+        if (b == '\n') {
+          return true;
+        }
+        if (b == '\r') {
+          consumeOptionalLF(is);
+          return true;
+        }
+        accumulator.addByte(b);
+      }
+    }
+
+    private static void consumeOptionalLF(InputStream is) throws IOException {
+      is.mark(1);
+      int next = is.read();
+      if (next != '\n' && next != -1) {
+        is.reset();
+      }
+    }
+
     protected A2AHttpResponse execute(HttpURLConnection connection) throws IOException {
       int status = connection.getResponseCode();
       A2AHttpHeaders responseHeaders = fromConnectionHeaders(connection.getHeaderFields());
@@ -199,29 +269,11 @@ protected void processSSEResponse(
         String contentType = connection.getContentType();
         boolean isSse = contentType != null && contentType.contains(EVENT_STREAM);
 
-        try (InputStream is = connection.getInputStream();
-            BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
-          String line;
+        try (InputStream is = connection.getInputStream()) {
           if (isSse) {
-            ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer);
-            while ((line = reader.readLine()) != null) {
-              sseParser.processLine(line);
-            }
-            sseParser.flush();
+            readSSEStream(is, sseParserConfig, messageConsumer, errorConsumer);
           } else {
-            StringBuilder bodyBuffer = new StringBuilder();
-            while ((line = reader.readLine()) != null) {
-              if (!line.isEmpty()) {
-                if (bodyBuffer.length() > 0) {
-                  bodyBuffer.append('\n');
-                }
-                bodyBuffer.append(line);
-              }
-            }
-            String body = bodyBuffer.toString();
-            if (!body.isEmpty()) {
-              messageConsumer.accept(new ServerSentEvent(body));
-            }
+            readNonSSEStream(is, messageConsumer);
           }
           completeRunnable.run();
         }
@@ -232,6 +284,45 @@ protected void processSSEResponse(
       }
     }
 
+    private static void readSSEStream(
+        InputStream is,
+        SSEParserConfig config,
+        Consumer<ServerSentEvent> messageConsumer,
+        Consumer<Throwable> errorConsumer) throws IOException {
+      InputStream buffered = new java.io.BufferedInputStream(is);
+      BoundedLineAccumulator accumulator = new BoundedLineAccumulator(config.maxLineLength());
+      ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, config);
+      while (readBoundedLine(buffered, accumulator)) {
+        if (accumulator.isTooLong()) {
+          sseParser.processLineTooLong();
+        } else {
+          sseParser.processLine(accumulator.toLine());
+        }
+        accumulator.reset();
+      }
+      sseParser.flush();
+    }
+
+    private static void readNonSSEStream(
+        InputStream is,
+        Consumer<ServerSentEvent> messageConsumer) throws IOException {
+      StringBuilder bodyBuffer = new StringBuilder();
+      BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8));
+      String line;
+      while ((line = reader.readLine()) != null) {
+        if (!line.isEmpty()) {
+          if (bodyBuffer.length() > 0) {
+            bodyBuffer.append('\n');
+          }
+          bodyBuffer.append(line);
+        }
+      }
+      String body = bodyBuffer.toString();
+      if (!body.isEmpty()) {
+        messageConsumer.accept(new ServerSentEvent(body));
+      }
+    }
+
     protected CompletableFuture<Void> executeAsyncSSE(
         HttpURLConnection connection,
         Consumer<ServerSentEvent> messageConsumer,
@@ -244,6 +335,10 @@ protected CompletableFuture<Void> executeAsyncSSE(
   }
 
   private static class AndroidGetBuilder extends AndroidBuilder<GetBuilder> implements GetBuilder {
+AndroidGetBuilder(SSEParserConfig sseParserConfig) {
+      super(sseParserConfig);
+    }
+
     @Override
     public A2AHttpResponse get() throws IOException {
       HttpURLConnection connection = createConnection("GET", false);
@@ -271,6 +366,10 @@ private static class AndroidPostBuilder extends AndroidBuilder<PostBuilder>
     private String body = "";
     private boolean followRedirects = false;
 
+AndroidPostBuilder(SSEParserConfig sseParserConfig) {
+      super(sseParserConfig);
+    }
+
     @Override
     public PostBuilder body(String body) {
       this.body = body;
@@ -325,6 +424,10 @@ public CompletableFuture<Void> postAsyncSSE(
 
   private static class AndroidDeleteBuilder extends AndroidBuilder<DeleteBuilder>
       implements DeleteBuilder {
+AndroidDeleteBuilder(SSEParserConfig sseParserConfig) {
+      super(sseParserConfig);
+    }
+
     @Override
     public A2AHttpResponse delete() throws IOException {
       HttpURLConnection connection = createConnection("DELETE", false);

extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java

@@ -2,6 +2,7 @@
 
 import org.a2aproject.sdk.client.http.A2AHttpClient;
 import org.a2aproject.sdk.client.http.A2AHttpClientProvider;
+import org.a2aproject.sdk.client.http.SSEParserConfig;
 
 /**
  * Service provider for {@link AndroidA2AHttpClient}.
@@ -24,6 +25,25 @@ public A2AHttpClient create() {
         return new AndroidA2AHttpClient();
     }
 
+    /**
+     * {@inheritDoc}
+     *
+     * @throws IllegalStateException if the Android runtime is not available
+     */
+    @Override
+    public A2AHttpClient createWithSseConfig(SSEParserConfig sseParserConfig) {
+        if (!ANDROID_AVAILABLE) {
+            throw new IllegalStateException(
+                    "Android classes are not available. This provider is only supported on Android.");
+        }
+        return new AndroidA2AHttpClient(sseParserConfig);
+    }
+
+    @Override
+    public boolean supportsSseConfig() {
+        return true;
+    }
+
     @Override
     public int priority() {
         return ANDROID_AVAILABLE ? 110 : -1; // Higher priority than Vert.x on Android

extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProviderTest.java

@@ -0,0 +1,49 @@
+package org.a2aproject.sdk.client.http.android;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.a2aproject.sdk.client.http.A2AHttpClient;
+import org.a2aproject.sdk.client.http.SSEParserConfig;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests for {@link AndroidA2AHttpClientProvider}.
+ *
+ * <p>The surefire configuration for this module sets {@code java.runtime.name} to
+ * {@code "Android Runtime"}, so the provider treats the environment as Android.
+ */
+public class AndroidA2AHttpClientProviderTest {
+
+    @Test
+    public void testCreateWithSseConfigReturnsAndroidClient() {
+        AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider();
+        SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(4 * 1024 * 1024).build();
+
+        A2AHttpClient client = provider.createWithSseConfig(config);
+        assertNotNull(client);
+        assertInstanceOf(AndroidA2AHttpClient.class, client,
+                "Provider should return AndroidA2AHttpClient when Android runtime is available");
+    }
+
+    @Test
+    public void testSupportsSseConfig() {
+        AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider();
+        assertTrue(provider.supportsSseConfig(), "Android provider must support SSE config");
+    }
+
+    @Test
+    public void testProviderName() {
+        AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider();
+        assertEquals("android", provider.name());
+    }
+
+    @Test
+    public void testPriorityOnAndroid() {
+        AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider();
+        assertEquals(110, provider.priority(),
+                "Android provider should have priority 110 when Android runtime is available");
+    }
+}

extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java

@@ -2,11 +2,17 @@
 
 import org.a2aproject.sdk.client.http.A2AHttpClient;
 import org.a2aproject.sdk.client.http.AbstractA2AHttpClientSSETest;
+import org.a2aproject.sdk.client.http.SSEParserConfig;
 
 public class AndroidA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest {
 
     @Override
     protected A2AHttpClient createClient() {
         return new AndroidA2AHttpClient();
     }
+
+    @Override
+    protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) {
+        return new AndroidA2AHttpClient(sseParserConfig);
+    }
 }