Skip to content

feat(sse): make parser limits configurable via SSEParserConfig #1149

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ehsavoie wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(sse): make parser limits configurable via SSEParserConfig #1149

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions docs/content/dev/client.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 characters 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 length check, relying on `maxBufferChars` alone to bound memory. Only disable it when you trust the remote agent or have other safeguards (e.g. a reverse proxy with its own line-length limit).

```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 = new JdkA2AHttpClient(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).
Expand Down
43 changes: 39 additions & 4 deletions ...nt-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -28,6 +30,7 @@
import org.a2aproject.sdk.client.http.A2AHttpHeaders;
import org.a2aproject.sdk.client.http.A2AHttpResponse;
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;
Expand All @@ -48,24 +51,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) {
Expand Down Expand Up @@ -203,7 +226,7 @@ protected void processSSEResponse(
BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
String line;
if (isSse) {
ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer);
ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, sseParserConfig);
while ((line = reader.readLine()) != null) {
sseParser.processLine(line);
}
Expand Down Expand Up @@ -244,6 +267,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);
Expand Down Expand Up @@ -271,6 +298,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;
Expand Down Expand Up @@ -325,6 +356,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);
Expand Down
15 changes: 15 additions & 0 deletions ...id/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}.
Expand All @@ -24,6 +25,20 @@ public A2AHttpClient create() {
return new AndroidA2AHttpClient();
}

/**
* {@inheritDoc}
*
* @throws IllegalStateException if the Android runtime is not available
*/
@Override
public A2AHttpClient create(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 int priority() {
return ANDROID_AVAILABLE ? 110 : -1; // Higher priority than Vert.x on Android
Expand Down
6 changes: 6 additions & 0 deletions ...oid/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
36 changes: 32 additions & 4 deletions ...ent-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import org.a2aproject.sdk.client.http.A2AHttpHeaders;
import org.a2aproject.sdk.client.http.A2AHttpResponse;
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;
Expand All @@ -48,24 +49,39 @@ class OkHttpA2AHttpClient implements A2AHttpClient {
return t;
});

private final SSEParserConfig sseParserConfig;

OkHttpA2AHttpClient() {
this(SSEParserConfig.DEFAULT);
}

OkHttpA2AHttpClient(SSEParserConfig sseParserConfig) {
this.sseParserConfig = sseParserConfig;
}

@Override
public GetBuilder createGet() {
return new OkHttpGetBuilder();
return new OkHttpGetBuilder(sseParserConfig);
}

@Override
public PostBuilder createPost() {
return new OkHttpPostBuilder();
return new OkHttpPostBuilder(sseParserConfig);
}

@Override
public DeleteBuilder createDelete() {
return new OkHttpDeleteBuilder();
return new OkHttpDeleteBuilder(sseParserConfig);
}

private abstract static class OkHttpBuilder<T extends Builder<T>> implements Builder<T> {
protected String url = "";
protected final Map<String, String> headers = new HashMap<>();
protected final SSEParserConfig sseParserConfig;

OkHttpBuilder(SSEParserConfig sseParserConfig) {
this.sseParserConfig = sseParserConfig;
}

@Override
public T url(String url) {
Expand Down Expand Up @@ -189,7 +205,7 @@ private void parseResponseBody(
new InputStreamReader(body.byteStream(), StandardCharsets.UTF_8))) {
String line;
if (isSse) {
ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer);
ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, sseParserConfig);
while ((line = reader.readLine()) != null) {
sseParser.processLine(line);
}
Expand All @@ -214,6 +230,10 @@ private void parseResponseBody(
}

private static class OkHttpGetBuilder extends OkHttpBuilder<GetBuilder> implements GetBuilder {
OkHttpGetBuilder(SSEParserConfig sseParserConfig) {
super(sseParserConfig);
}

@Override
public A2AHttpResponse get() throws IOException {
OkHttpClient client = buildClient(false);
Expand Down Expand Up @@ -248,6 +268,10 @@ private static class OkHttpPostBuilder extends OkHttpBuilder<PostBuilder> implem
private String body = "";
private boolean followRedirects = false;

OkHttpPostBuilder(SSEParserConfig sseParserConfig) {
super(sseParserConfig);
}

@Override
public PostBuilder body(String body) {
this.body = body;
Expand Down Expand Up @@ -295,6 +319,10 @@ public CompletableFuture<Void> postAsyncSSE(
}

private static class OkHttpDeleteBuilder extends OkHttpBuilder<DeleteBuilder> implements DeleteBuilder {
OkHttpDeleteBuilder(SSEParserConfig sseParserConfig) {
super(sseParserConfig);
}

@Override
public A2AHttpResponse delete() throws IOException {
OkHttpClient client = buildClient(false);
Expand Down
6 changes: 6 additions & 0 deletions ...roid/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 OkHttpA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest {

@Override
protected A2AHttpClient createClient() {
return new OkHttpA2AHttpClient();
}

@Override
protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) {
return new OkHttpA2AHttpClient(sseParserConfig);
}
}
29 changes: 29 additions & 0 deletions http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,35 @@ public static A2AHttpClient create() {
.orElseThrow(() -> new IllegalStateException("No A2AHttpClientProvider could be instantiated"));
}

/**
* Creates a new A2AHttpClient instance with the given {@link SSEParserConfig} using the
* highest available priority provider.
*
* <p>
* Providers that support SSE parser configuration will use it; others fall back to their
* default configuration. Providers are tried in descending priority order.
*
* @param sseParserConfig the SSE parser configuration to apply
* @return a new A2AHttpClient instance
* @throws IllegalStateException if no provider found or all providers failed to instantiate
*/
public static A2AHttpClient create(SSEParserConfig sseParserConfig) {
if (sseParserConfig == null) {
return create();
}
return PROVIDERS.stream()
.flatMap(p -> {
try {
return Stream.of(p.create(sseParserConfig));
} catch (Exception e) {
LOGGER.log(Level.WARNING, e, () -> "Provider " + p.name() + " skipped");
return Stream.empty();
}
})
.findFirst()
.orElseThrow(() -> new IllegalStateException("No A2AHttpClientProvider could be instantiated"));
}

/**
* Creates a new A2AHttpClient instance using a specific provider by name.
*
Expand Down
13 changes: 13 additions & 0 deletions http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,19 @@ public interface A2AHttpClientProvider {
*/
A2AHttpClient create();

/**
* Creates a new instance of an A2AHttpClient with the given {@link SSEParserConfig}.
*
* <p>Providers that support SSE parser configuration should override this method.
* The default implementation ignores {@code sseParserConfig} and delegates to {@link #create()}.
*
* @param sseParserConfig the SSE parser configuration to apply
* @return a new A2AHttpClient instance
*/
default A2AHttpClient create(SSEParserConfig sseParserConfig) {
return create();
}

/**
* Returns the priority of this provider. Higher priority providers are
* tried first; the first one whose {@link #create()} succeeds is used.
Expand Down
Loading
Loading