Skip to content

Commit 355f39d

Browse files
authored
feat(sse): add SSEParserConfig, raise default per-line limit to 1 MB (a2aproject#1149)
- Introduce SSEParserConfig record with builder to make SSE parser limits (maxLineLength, maxBufferLines, maxBufferChars) configurable - Raise default per-line limit from 64 KB to 1 MB so large JSON-RPC responses are no longer silently dropped - Thread SSEParserConfig through all HTTP client implementations (JDK, Android, OkHttp) via new constructors and provider methods - Prevent corrupt/skipped event blocks from advancing lastEventId, which could cause event loss on reconnect - Add AbstractA2AHttpClientSSETest for shared cross-client SSE tests - Document SSEParserConfig usage and security considerations Fixes a2aproject#1123 Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>
1 parent c89ebfe commit 355f39d

22 files changed

Lines changed: 1950 additions & 48 deletions

‎docs/content/dev/client.md‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,35 @@ Client client = Client
185185
.build();
186186
```
187187

188+
### SSE Parser Configuration
189+
190+
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):
191+
192+
| Parameter | Description | Default |
193+
|------------------|----------------------------------------------------------|-----------|
194+
| `maxLineLength` | Max bytes per raw SSE line (`0` = disabled) | 1 MB |
195+
| `maxBufferLines` | Max `data:` lines per event block | 1 000 |
196+
| `maxBufferChars` | Max total characters across all `data:` values per event | 1 MB |
197+
198+
The defaults are suitable for most deployments. To override them, build a custom `SSEParserConfig` and pass it to the HTTP client.
199+
200+
> **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.
201+
202+
```java
203+
SSEParserConfig sseConfig = SSEParserConfig.builder()
204+
.maxLineLength(4 * 1024 * 1024) // 4 MB per line
205+
.maxBufferChars(4 * 1024 * 1024) // 4 MB per event
206+
.build();
207+
208+
// Pass to JdkA2AHttpClient, then to your transport config
209+
JdkA2AHttpClient httpClient = JdkA2AHttpClient.withSseConfig(sseConfig);
210+
211+
Client client = Client
212+
.builder(agentCard)
213+
.withTransport(JSONRPCTransport.class, new JSONRPCTransportConfig(httpClient))
214+
.build();
215+
```
216+
188217
## Observability (Optional)
189218

190219
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‎

Lines changed: 127 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package org.a2aproject.sdk.client.http.android;
22

3+
import static org.a2aproject.sdk.util.Assert.checkNotNullParam;
4+
35
import static java.net.HttpURLConnection.HTTP_FORBIDDEN;
46
import static java.net.HttpURLConnection.HTTP_MULT_CHOICE;
57
import static java.net.HttpURLConnection.HTTP_OK;
@@ -27,7 +29,9 @@
2729
import org.a2aproject.sdk.client.http.A2AHttpClient;
2830
import org.a2aproject.sdk.client.http.A2AHttpHeaders;
2931
import org.a2aproject.sdk.client.http.A2AHttpResponse;
32+
import org.a2aproject.sdk.client.http.BoundedLineAccumulator;
3033
import org.a2aproject.sdk.client.http.ServerSentEvent;
34+
import org.a2aproject.sdk.client.http.SSEParserConfig;
3135
import org.a2aproject.sdk.client.http.ServerSentEventParser;
3236
import org.a2aproject.sdk.common.A2AErrorMessages;
3337
import org.a2aproject.sdk.spec.A2AClientHTTPError;
@@ -48,24 +52,44 @@ public class AndroidA2AHttpClient implements A2AHttpClient {
4852
return t;
4953
});
5054

55+
private final SSEParserConfig sseParserConfig;
56+
57+
public AndroidA2AHttpClient() {
58+
this(SSEParserConfig.DEFAULT);
59+
}
60+
61+
/**
62+
* Creates a new Android HTTP client with custom SSE parser limits.
63+
*
64+
* @param sseParserConfig the SSE parser configuration to use for streaming responses
65+
*/
66+
public AndroidA2AHttpClient(SSEParserConfig sseParserConfig) {
67+
this.sseParserConfig = checkNotNullParam("sseParserConfig", sseParserConfig);
68+
}
69+
5170
@Override
5271
public GetBuilder createGet() {
53-
return new AndroidGetBuilder();
72+
return new AndroidGetBuilder(sseParserConfig);
5473
}
5574

5675
@Override
5776
public PostBuilder createPost() {
58-
return new AndroidPostBuilder();
77+
return new AndroidPostBuilder(sseParserConfig);
5978
}
6079

6180
@Override
6281
public DeleteBuilder createDelete() {
63-
return new AndroidDeleteBuilder();
82+
return new AndroidDeleteBuilder(sseParserConfig);
6483
}
6584

6685
private abstract static class AndroidBuilder<T extends Builder<T>> implements Builder<T> {
6786
protected String url = "";
6887
protected Map<String, String> headers = new HashMap<>();
88+
protected final SSEParserConfig sseParserConfig;
89+
90+
AndroidBuilder(SSEParserConfig sseParserConfig) {
91+
this.sseParserConfig = sseParserConfig;
92+
}
6993

7094
@Override
7195
public T url(String url) {
@@ -137,6 +161,52 @@ protected static String readStreamWithLimit(InputStream is) throws IOException {
137161
}
138162
}
139163

164+
/**
165+
* Reads the next line from {@code is}, treating LF, CRLF, and bare CR as terminators
166+
* per the SSE specification, and enforcing the byte limit <em>while accumulating
167+
* bytes</em> — not after a full line has been materialised.
168+
*
169+
* <p>The caller must provide a {@link BoundedLineAccumulator} and call
170+
* {@link BoundedLineAccumulator#reset()} between lines. After this method returns
171+
* {@code true}, check {@link BoundedLineAccumulator#isTooLong()} to distinguish
172+
* between a valid line and one that exceeded the limit.
173+
*
174+
* <p>The {@code is} parameter must support {@link InputStream#mark(int)} (e.g.
175+
* {@link java.io.BufferedInputStream}) so that CR at end-of-buffer can be resolved
176+
* without consuming the following byte.
177+
*
178+
* @param is the input stream to read from (must support mark/reset)
179+
* @param accumulator the accumulator to use for byte collection and limit enforcement
180+
* @return {@code true} if a line was read, {@code false} at end-of-stream
181+
*/
182+
static boolean readBoundedLine(InputStream is, BoundedLineAccumulator accumulator) throws IOException {
183+
boolean hasReadAnyBytes = false;
184+
185+
while (true) {
186+
int b = is.read();
187+
if (b == -1) {
188+
return hasReadAnyBytes || accumulator.isTooLong();
189+
}
190+
hasReadAnyBytes = true;
191+
if (b == '\n') {
192+
return true;
193+
}
194+
if (b == '\r') {
195+
consumeOptionalLF(is);
196+
return true;
197+
}
198+
accumulator.addByte(b);
199+
}
200+
}
201+
202+
private static void consumeOptionalLF(InputStream is) throws IOException {
203+
is.mark(1);
204+
int next = is.read();
205+
if (next != '\n' && next != -1) {
206+
is.reset();
207+
}
208+
}
209+
140210
protected A2AHttpResponse execute(HttpURLConnection connection) throws IOException {
141211
int status = connection.getResponseCode();
142212
A2AHttpHeaders responseHeaders = fromConnectionHeaders(connection.getHeaderFields());
@@ -199,29 +269,11 @@ protected void processSSEResponse(
199269
String contentType = connection.getContentType();
200270
boolean isSse = contentType != null && contentType.contains(EVENT_STREAM);
201271

202-
try (InputStream is = connection.getInputStream();
203-
BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
204-
String line;
272+
try (InputStream is = connection.getInputStream()) {
205273
if (isSse) {
206-
ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer);
207-
while ((line = reader.readLine()) != null) {
208-
sseParser.processLine(line);
209-
}
210-
sseParser.flush();
274+
readSSEStream(is, sseParserConfig, messageConsumer, errorConsumer);
211275
} else {
212-
StringBuilder bodyBuffer = new StringBuilder();
213-
while ((line = reader.readLine()) != null) {
214-
if (!line.isEmpty()) {
215-
if (bodyBuffer.length() > 0) {
216-
bodyBuffer.append('\n');
217-
}
218-
bodyBuffer.append(line);
219-
}
220-
}
221-
String body = bodyBuffer.toString();
222-
if (!body.isEmpty()) {
223-
messageConsumer.accept(new ServerSentEvent(body));
224-
}
276+
readNonSSEStream(is, messageConsumer);
225277
}
226278
completeRunnable.run();
227279
}
@@ -232,6 +284,45 @@ protected void processSSEResponse(
232284
}
233285
}
234286

287+
private static void readSSEStream(
288+
InputStream is,
289+
SSEParserConfig config,
290+
Consumer<ServerSentEvent> messageConsumer,
291+
Consumer<Throwable> errorConsumer) throws IOException {
292+
InputStream buffered = new java.io.BufferedInputStream(is);
293+
BoundedLineAccumulator accumulator = new BoundedLineAccumulator(config.maxLineLength());
294+
ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, config);
295+
while (readBoundedLine(buffered, accumulator)) {
296+
if (accumulator.isTooLong()) {
297+
sseParser.processLineTooLong();
298+
} else {
299+
sseParser.processLine(accumulator.toLine());
300+
}
301+
accumulator.reset();
302+
}
303+
sseParser.flush();
304+
}
305+
306+
private static void readNonSSEStream(
307+
InputStream is,
308+
Consumer<ServerSentEvent> messageConsumer) throws IOException {
309+
StringBuilder bodyBuffer = new StringBuilder();
310+
BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8));
311+
String line;
312+
while ((line = reader.readLine()) != null) {
313+
if (!line.isEmpty()) {
314+
if (bodyBuffer.length() > 0) {
315+
bodyBuffer.append('\n');
316+
}
317+
bodyBuffer.append(line);
318+
}
319+
}
320+
String body = bodyBuffer.toString();
321+
if (!body.isEmpty()) {
322+
messageConsumer.accept(new ServerSentEvent(body));
323+
}
324+
}
325+
235326
protected CompletableFuture<Void> executeAsyncSSE(
236327
HttpURLConnection connection,
237328
Consumer<ServerSentEvent> messageConsumer,
@@ -244,6 +335,10 @@ protected CompletableFuture<Void> executeAsyncSSE(
244335
}
245336

246337
private static class AndroidGetBuilder extends AndroidBuilder<GetBuilder> implements GetBuilder {
338+
AndroidGetBuilder(SSEParserConfig sseParserConfig) {
339+
super(sseParserConfig);
340+
}
341+
247342
@Override
248343
public A2AHttpResponse get() throws IOException {
249344
HttpURLConnection connection = createConnection("GET", false);
@@ -271,6 +366,10 @@ private static class AndroidPostBuilder extends AndroidBuilder<PostBuilder>
271366
private String body = "";
272367
private boolean followRedirects = false;
273368

369+
AndroidPostBuilder(SSEParserConfig sseParserConfig) {
370+
super(sseParserConfig);
371+
}
372+
274373
@Override
275374
public PostBuilder body(String body) {
276375
this.body = body;
@@ -325,6 +424,10 @@ public CompletableFuture<Void> postAsyncSSE(
325424

326425
private static class AndroidDeleteBuilder extends AndroidBuilder<DeleteBuilder>
327426
implements DeleteBuilder {
427+
AndroidDeleteBuilder(SSEParserConfig sseParserConfig) {
428+
super(sseParserConfig);
429+
}
430+
328431
@Override
329432
public A2AHttpResponse delete() throws IOException {
330433
HttpURLConnection connection = createConnection("DELETE", false);

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

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import org.a2aproject.sdk.client.http.A2AHttpClient;
44
import org.a2aproject.sdk.client.http.A2AHttpClientProvider;
5+
import org.a2aproject.sdk.client.http.SSEParserConfig;
56

67
/**
78
* Service provider for {@link AndroidA2AHttpClient}.
@@ -24,6 +25,25 @@ public A2AHttpClient create() {
2425
return new AndroidA2AHttpClient();
2526
}
2627

28+
/**
29+
* {@inheritDoc}
30+
*
31+
* @throws IllegalStateException if the Android runtime is not available
32+
*/
33+
@Override
34+
public A2AHttpClient createWithSseConfig(SSEParserConfig sseParserConfig) {
35+
if (!ANDROID_AVAILABLE) {
36+
throw new IllegalStateException(
37+
"Android classes are not available. This provider is only supported on Android.");
38+
}
39+
return new AndroidA2AHttpClient(sseParserConfig);
40+
}
41+
42+
@Override
43+
public boolean supportsSseConfig() {
44+
return true;
45+
}
46+
2747
@Override
2848
public int priority() {
2949
return ANDROID_AVAILABLE ? 110 : -1; // Higher priority than Vert.x on Android
Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package org.a2aproject.sdk.client.http.android;
2+
3+
import static org.junit.jupiter.api.Assertions.assertEquals;
4+
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
5+
import static org.junit.jupiter.api.Assertions.assertNotNull;
6+
import static org.junit.jupiter.api.Assertions.assertTrue;
7+
8+
import org.a2aproject.sdk.client.http.A2AHttpClient;
9+
import org.a2aproject.sdk.client.http.SSEParserConfig;
10+
import org.junit.jupiter.api.Test;
11+
12+
/**
13+
* Tests for {@link AndroidA2AHttpClientProvider}.
14+
*
15+
* <p>The surefire configuration for this module sets {@code java.runtime.name} to
16+
* {@code "Android Runtime"}, so the provider treats the environment as Android.
17+
*/
18+
public class AndroidA2AHttpClientProviderTest {
19+
20+
@Test
21+
public void testCreateWithSseConfigReturnsAndroidClient() {
22+
AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider();
23+
SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(4 * 1024 * 1024).build();
24+
25+
A2AHttpClient client = provider.createWithSseConfig(config);
26+
assertNotNull(client);
27+
assertInstanceOf(AndroidA2AHttpClient.class, client,
28+
"Provider should return AndroidA2AHttpClient when Android runtime is available");
29+
}
30+
31+
@Test
32+
public void testSupportsSseConfig() {
33+
AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider();
34+
assertTrue(provider.supportsSseConfig(), "Android provider must support SSE config");
35+
}
36+
37+
@Test
38+
public void testProviderName() {
39+
AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider();
40+
assertEquals("android", provider.name());
41+
}
42+
43+
@Test
44+
public void testPriorityOnAndroid() {
45+
AndroidA2AHttpClientProvider provider = new AndroidA2AHttpClientProvider();
46+
assertEquals(110, provider.priority(),
47+
"Android provider should have priority 110 when Android runtime is available");
48+
}
49+
}

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,17 @@
22

33
import org.a2aproject.sdk.client.http.A2AHttpClient;
44
import org.a2aproject.sdk.client.http.AbstractA2AHttpClientSSETest;
5+
import org.a2aproject.sdk.client.http.SSEParserConfig;
56

67
public class AndroidA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest {
78

89
@Override
910
protected A2AHttpClient createClient() {
1011
return new AndroidA2AHttpClient();
1112
}
13+
14+
@Override
15+
protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) {
16+
return new AndroidA2AHttpClient(sseParserConfig);
17+
}
1218
}

0 commit comments

Comments
 (0)