diff --git a/docs/content/dev/server.md b/docs/content/dev/server.md
index 9702fd10e..b80227599 100644
--- a/docs/content/dev/server.md
+++ b/docs/content/dev/server.md
@@ -295,4 +295,6 @@ class A2AServerConfiguration {
Register the remaining dependencies, including an `a2aInternalExecutor` bean and a `TaskAuthorizationProvider`, as Spring beans. Keep the cached event consumer executor separate from the agent executor: blocking event polling can exhaust a bounded pool under load.
+Manually built handlers use blocking timeouts of 30 seconds for agent execution, 5 seconds for event consumption, and 1 second for TaskStore reconciliation. Override these with `agentCompletionTimeoutSeconds(int)`, `consumptionCompletionTimeoutSeconds(int)`, and `reconciliationTimeoutSeconds(int)` on the builder. The `a2a.blocking.*` configuration properties apply to CDI-managed handlers; manual wiring uses the builder methods.
+
See [CONTRIBUTING_INTEGRATIONS.md](https://github.com/a2aproject/a2a-java/blob/main/CONTRIBUTING_INTEGRATIONS.md) to submit your own integration.
diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java
index ebda76a1c..aa99eab0a 100644
--- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java
+++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java
@@ -153,9 +153,14 @@
*
Blocking (configuration.blocking=true): Client waits for first event or final task state
* Streaming: Client receives events as they arrive via reactive streams
* Both modes support fire-and-forget (agent continues after client disconnect)
- * Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds},
+ * For CDI-managed handlers, configurable timeouts via {@code a2a.blocking.agent.timeout.seconds},
* {@code a2a.blocking.consumption.timeout.seconds}, and
* {@code a2a.blocking.reconciliation.timeout.seconds}
+ * For manually built handlers, configure timeouts with
+ * {@link Builder#agentCompletionTimeoutSeconds(int)},
+ * {@link Builder#consumptionCompletionTimeoutSeconds(int)}, and
+ * {@link Builder#reconciliationTimeoutSeconds(int)}.
+ * Both paths default to 30, 5, and 1 seconds, respectively.
*
*
* CDI Dependencies
@@ -405,6 +410,9 @@ public static class Builder {
private boolean authorizationRequired = true;
private boolean populateReferredTasks;
private boolean pushNotificationsEnabled = true;
+ private int agentCompletionTimeoutSeconds = 30;
+ private int consumptionCompletionTimeoutSeconds = 5;
+ private int reconciliationTimeoutSeconds = 1;
public Builder agentExecutor(AgentExecutor agentExecutor) {
this.agentExecutor = agentExecutor;
@@ -441,6 +449,51 @@ public Builder eventConsumerExecutor(Executor eventConsumerExecutor) {
return this;
}
+ /**
+ * Sets how long a blocking call waits for agent execution. Defaults to 30 seconds.
+ *
+ * @param seconds timeout in seconds; zero skips waiting
+ * @return this builder for chaining
+ * @throws IllegalArgumentException if the timeout is negative
+ */
+ public Builder agentCompletionTimeoutSeconds(int seconds) {
+ if (seconds < 0) {
+ throw new IllegalArgumentException("Agent completion timeout must not be negative");
+ }
+ this.agentCompletionTimeoutSeconds = seconds;
+ return this;
+ }
+
+ /**
+ * Sets how long a blocking call waits for event consumption. Defaults to 5 seconds.
+ *
+ * @param seconds timeout in seconds; zero skips waiting
+ * @return this builder for chaining
+ * @throws IllegalArgumentException if the timeout is negative
+ */
+ public Builder consumptionCompletionTimeoutSeconds(int seconds) {
+ if (seconds < 0) {
+ throw new IllegalArgumentException("Consumption completion timeout must not be negative");
+ }
+ this.consumptionCompletionTimeoutSeconds = seconds;
+ return this;
+ }
+
+ /**
+ * Sets how long TaskStore reconciliation polls in blocking calls. Defaults to 1 second.
+ *
+ * @param seconds timeout in seconds; zero disables reconciliation polling
+ * @return this builder for chaining
+ * @throws IllegalArgumentException if the timeout is negative
+ */
+ public Builder reconciliationTimeoutSeconds(int seconds) {
+ if (seconds < 0) {
+ throw new IllegalArgumentException("Reconciliation timeout must not be negative");
+ }
+ this.reconciliationTimeoutSeconds = seconds;
+ return this;
+ }
+
public Builder authorizationProvider(@Nullable TaskAuthorizationProvider authorizationProvider) {
this.authorizationProvider = authorizationProvider;
return this;
@@ -490,9 +543,9 @@ public DefaultRequestHandler build() {
DefaultRequestHandler handler =
new DefaultRequestHandler(agentExecutor, taskStore, queueManager, pushConfigStore,
mainEventBusProcessor, executor, eventConsumerExecutor);
- handler.agentCompletionTimeoutSeconds = 5;
- handler.consumptionCompletionTimeoutSeconds = 2;
- handler.reconciliationTimeoutSeconds = 1;
+ handler.agentCompletionTimeoutSeconds = agentCompletionTimeoutSeconds;
+ handler.consumptionCompletionTimeoutSeconds = consumptionCompletionTimeoutSeconds;
+ handler.reconciliationTimeoutSeconds = reconciliationTimeoutSeconds;
handler.authorizationProvider = authorizationProvider;
handler.agentExecutorRouter = agentExecutorRouter;
handler.authorizationRequired = authorizationRequired;
diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java
index 77f4bb9b4..eac9f77a4 100644
--- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java
+++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java
@@ -203,6 +203,44 @@ void testInitConfigReadsBlockingTimeouts() {
assertEquals(7, handler.reconciliationTimeoutSeconds);
}
+ @Test
+ void testBuilderUsesDocumentedBlockingTimeoutDefaults() {
+ DefaultRequestHandler handler = (DefaultRequestHandler) requestHandler;
+
+ assertEquals(30, handler.agentCompletionTimeoutSeconds);
+ assertEquals(5, handler.consumptionCompletionTimeoutSeconds);
+ assertEquals(1, handler.reconciliationTimeoutSeconds);
+ }
+
+ @ParameterizedTest
+ @CsvSource({"41,8,3", "0,0,0"})
+ void testBuilderOverridesBlockingTimeouts(int agentTimeout, int consumptionTimeout, int reconciliationTimeout) {
+ DefaultRequestHandler handler = DefaultRequestHandler.builder()
+ .agentExecutor(executor)
+ .taskStore(taskStore)
+ .queueManager(queueManager)
+ .mainEventBusProcessor(mainEventBusProcessor)
+ .executor(internalExecutor)
+ .eventConsumerExecutor(internalExecutor)
+ .agentCompletionTimeoutSeconds(agentTimeout)
+ .consumptionCompletionTimeoutSeconds(consumptionTimeout)
+ .reconciliationTimeoutSeconds(reconciliationTimeout)
+ .build();
+
+ assertEquals(agentTimeout, handler.agentCompletionTimeoutSeconds);
+ assertEquals(consumptionTimeout, handler.consumptionCompletionTimeoutSeconds);
+ assertEquals(reconciliationTimeout, handler.reconciliationTimeoutSeconds);
+ }
+
+ @Test
+ void testBuilderRejectsNegativeBlockingTimeouts() {
+ DefaultRequestHandler.Builder builder = DefaultRequestHandler.builder();
+
+ assertThrows(IllegalArgumentException.class, () -> builder.agentCompletionTimeoutSeconds(-1));
+ assertThrows(IllegalArgumentException.class, () -> builder.consumptionCompletionTimeoutSeconds(-1));
+ assertThrows(IllegalArgumentException.class, () -> builder.reconciliationTimeoutSeconds(-1));
+ }
+
@Test
void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exception {
InMemoryTaskStore manualTaskStore = new InMemoryTaskStore();
@@ -1758,8 +1796,9 @@ public void cancel(RequestContext context, AgentEmitter emitter) {
* must still reach the queue, which was told to expect it; otherwise the consumer waits out its
* grace period for it and the call fails or returns late.
*/
- @Test
- void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate() throws Exception {
+ @ParameterizedTest
+ @CsvSource({"true", "false"})
+ void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate(boolean omitConfiguration) throws Exception {
AtomicReference callerQueue = new AtomicReference<>();
useQueueManager(new InMemoryQueueManager((TaskStateProvider) taskStore, mainEventBus) {
@Override
@@ -1772,15 +1811,37 @@ public EventQueue createOrTap(String taskId) {
AtomicBoolean callerQueueClosedInTime = new AtomicBoolean();
holdEventsAfterWorkingUntilClosed(callerQueue, callerQueueClosedInTime);
agentExecutorExecute = (context, emitter) -> {
+ emitter.submit();
emitter.startWork();
+ emitter.addArtifact(List.of(new TextPart("answer")), "answer-id", "answer", null);
emitter.complete();
};
- EventKind result = requestHandler.onMessageSend(blockingParams("msg-blocking-final-event"), NULL_CONTEXT);
+ MessageSendParams params = blockingParams("msg-blocking-final-event");
+ if (omitConfiguration) {
+ params = MessageSendParams.builder().message(params.message()).build();
+ }
+ EventKind result = requestHandler.onMessageSend(params, NULL_CONTEXT);
assertTrue(callerQueueClosedInTime.get(), "the caller's queue was not closed while the final event was held");
Task task = assertInstanceOf(Task.class, result);
assertEquals(TaskState.TASK_STATE_COMPLETED, task.status().state());
+ assertEquals(1, task.artifacts().size());
+ assertEquals("answer-id", task.artifacts().get(0).artifactId());
+ assertEquals("answer", assertInstanceOf(TextPart.class, task.artifacts().get(0).parts().get(0)).text());
+ }
+
+ @Test
+ void blockingSendWithoutFinalEventReturnsWorkingTaskAndAllowsResubscription() throws Exception {
+ agentExecutorExecute = (context, emitter) -> emitter.startWork();
+
+ EventKind result = requestHandler.onMessageSend(blockingParams("msg-fire-and-forget"), NULL_CONTEXT);
+
+ Task task = assertInstanceOf(Task.class, result);
+ assertEquals(TaskState.TASK_STATE_WORKING, task.status().state());
+ EventQueue resubscribedQueue = queueManager.tap(task.id());
+ assertNotNull(resubscribedQueue);
+ resubscribedQueue.close();
}
private void useQueueManager(InMemoryQueueManager manager) {