callAsync(ToolCallParam param);
+
+ /**
+ * Executes the tool through the framework's execution infrastructure.
+ *
+ * Called by {@link ToolExecutor} when running a tool call through the scheduling, timeout,
+ * retry and shutdown layers; application code should keep using
+ * {@link #callAsync(ToolCallParam)}. Unlike {@link #callAsync(ToolCallParam)}, failures stay
+ * as reactive error signals so they can be retried; the executor converts the final error
+ * into an error result once retries are exhausted.
+ *
+ *
Each subscription must be a fresh attempt, business failures must stay completed
+ * {@link ToolResultBlock} error results, and {@link ToolSuspendException} must propagate as
+ * an error signal. The default implementation delegates to {@link #callAsync(ToolCallParam)};
+ * built-in tools override it to keep exceptions as error signals.
+ *
+ * @param param The tool call parameters
+ * @return Mono containing ToolResultBlock, or signalling an error on failure
+ */
+ default Mono callAsyncForExecution(ToolCallParam param) {
+ return callAsync(param);
+ }
}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java
index c92ef950df..0962445f3b 100644
--- a/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ReflectiveFunctionTool.java
@@ -165,6 +165,11 @@ public Mono callAsync(ToolCallParam param) {
return methodInvoker.invokeAsync(toolObject, method, param, customConverter);
}
+ @Override
+ public Mono callAsyncForExecution(ToolCallParam param) {
+ return methodInvoker.invokeRawAsync(toolObject, method, param, customConverter);
+ }
+
Method getMethod() {
return method;
}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java
index 7d9c13a681..5b87c90dbd 100644
--- a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolExecutor.java
@@ -28,6 +28,7 @@
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import java.util.function.Predicate;
import org.slf4j.Logger;
@@ -159,13 +160,43 @@ private void invokeChunkCallback(
// ==================== Single Tool Execution ====================
/**
- * Execute a single tool call with full infrastructure support.
+ * Execute a single tool call without the infrastructure layers.
+ *
+ * Compatibility entry point used by {@link Toolkit#callTool(ToolCallParam)}: failures
+ * surface as {@link ToolResultBlock} error results, not reactive error signals.
*
* @param param Tool call parameters
* @return Mono containing execution result
*/
Mono execute(ToolCallParam param) {
- return TracerRegistry.get().callTool(this.toolkit, param, () -> executeCore(param));
+ return executeWithTracing(param, false)
+ .onErrorResume(e -> Mono.just(buildToolErrorResult(e)));
+ }
+
+ /**
+ * Execute a single tool call through the raw execution channel, keeping failures as reactive
+ * error signals. Used only by {@link #executeWithInfrastructure}.
+ *
+ * @param param Tool call parameters
+ * @return Mono containing execution result, or signalling an error on failure
+ */
+ private Mono executeRaw(ToolCallParam param) {
+ return executeWithTracing(param, true);
+ }
+
+ /**
+ * Run the traced core execution for the given tool call.
+ *
+ * @param param Tool call parameters
+ * @param useExecutionPath whether to use the raw execution channel
+ * ({@link AgentTool#callAsyncForExecution(ToolCallParam)}) instead of the compatibility
+ * channel ({@link AgentTool#callAsync(ToolCallParam)})
+ * @return Mono containing execution result, or signalling an error on failure
+ */
+ private Mono executeWithTracing(
+ ToolCallParam param, boolean useExecutionPath) {
+ return TracerRegistry.get()
+ .callTool(this.toolkit, param, () -> executeCore(param, useExecutionPath));
}
/**
@@ -181,7 +212,7 @@ Mono execute(ToolCallParam param) {
* Actual tool invocation
*
*/
- private Mono executeCore(ToolCallParam param) {
+ private Mono executeCore(ToolCallParam param, boolean useExecutionPath) {
ToolUseBlock toolCall = param.getToolUseBlock();
AgentTool tool = toolRegistry.getTool(toolCall.getName());
@@ -263,31 +294,52 @@ private Mono executeCore(ToolCallParam param) {
.emitter(toolEmitter)
.build();
- return tool.callAsync(executionParam)
- .onErrorResume(
- ToolSuspendException.class,
- e -> {
- // Convert ToolSuspendException to suspended result
- logger.debug(
- "Tool '{}' suspended: {}",
- toolCall.getName(),
- e.getReason() != null ? e.getReason() : "no reason");
- return Mono.just(ToolResultBlock.suspended(toolCall, e));
- })
- .onErrorResume(
+ // Compatibility channel converts failures to error results here; the execution channel
+ // keeps them as error signals for the timeout/retry layers. Mono.defer re-invokes the
+ // tool per subscription so each retry runs a fresh attempt.
+ Mono invocation;
+ if (useExecutionPath) {
+ invocation =
+ Mono.defer(() -> tool.callAsyncForExecution(executionParam))
+ // Unwrap reflection/future wrappers for the retry predicate
+ .onErrorMap(ExceptionUtils::unwrapExecutionWrapper);
+ } else {
+ invocation = tool.callAsync(executionParam);
+ }
+
+ Mono chain =
+ invocation.onErrorResume(
e -> {
- String errorMsg =
- e.getMessage() != null
- ? e.getMessage()
- : e.getClass().getSimpleName();
- return Mono.just(
- ToolResultBlock.error("Tool execution failed: " + errorMsg));
- })
- .switchIfEmpty(
- Mono.just(
- ToolResultBlock.error(
- "Tool execution failed: Tool completed without returning a"
- + " result")));
+ // Convert any wrapped ToolSuspendException to a suspended result
+ ToolSuspendException suspended =
+ ExceptionUtils.findToolSuspendException(e);
+ if (suspended != null) {
+ logger.debug(
+ "Tool '{}' suspended: {}",
+ toolCall.getName(),
+ suspended.getReason() != null
+ ? suspended.getReason()
+ : "no reason");
+ return Mono.just(ToolResultBlock.suspended(toolCall, suspended));
+ }
+ return Mono.error(e);
+ });
+ if (!useExecutionPath) {
+ chain = chain.onErrorResume(e -> Mono.just(buildToolErrorResult(e)));
+ }
+ return chain.switchIfEmpty(
+ Mono.just(
+ ToolResultBlock.error(
+ "Tool execution failed: Tool completed without returning a"
+ + " result")));
+ }
+
+ /**
+ * Build the standard {@link ToolResultBlock} error result for a failed tool call.
+ */
+ private ToolResultBlock buildToolErrorResult(Throwable e) {
+ String errorMsg = e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName();
+ return ToolResultBlock.error("Tool execution failed: " + errorMsg);
}
// ==================== Batch Tool Execution ====================
@@ -384,8 +436,9 @@ private Mono executeWithInfrastructure(
.runtimeContext(agentRuntimeContext)
.build();
- // Get core execution
- Mono execution = execute(param);
+ // Keep failures as error signals for the layers below; convert what remains after
+ // retries are exhausted into an error result.
+ Mono execution = executeRaw(param);
// Apply infrastructure layers
execution = applyScheduling(execution);
@@ -424,9 +477,10 @@ private Mono applyTimeout(
Duration timeout = config.getTimeout();
logger.debug("Applied timeout: {} for tool: {}", timeout, toolCall.getName());
+ // TimeoutException so ExecutionConfig.RETRYABLE_ERRORS recognizes timeouts as retryable
return execution.timeout(
timeout,
- Mono.error(new RuntimeException("Tool execution timeout after " + timeout)));
+ Mono.error(new TimeoutException("Tool execution timeout after " + timeout)));
}
private Mono applyRetry(
@@ -450,6 +504,8 @@ private Mono applyRetry(
.maxBackoff(maxBackoff)
.jitter(0.5)
.filter(retryOn)
+ // Propagate the last failure instead of RetryExhaustedException
+ .onRetryExhaustedThrow((spec, signal) -> signal.failure())
.doBeforeRetry(
signal ->
logger.warn(
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java
index 73bb16c3fe..f2fef4381d 100644
--- a/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/ToolMethodInvoker.java
@@ -46,6 +46,8 @@ class ToolMethodInvoker {
/**
* Invoke tool method asynchronously with custom converter support.
*
+ * Failures are converted into {@link ToolResultBlock} error results.
+ *
* @param toolObject the object containing the method
* @param method the method to invoke
* @param param the tool call parameters containing input, toolUseBlock, agent, and context
@@ -57,6 +59,28 @@ Mono invokeAsync(
Method method,
ToolCallParam param,
ToolResultConverter customConverter) {
+ return invokeRawAsync(toolObject, method, param, customConverter)
+ .onErrorResume(this::handleError);
+ }
+
+ /**
+ * Invoke tool method asynchronously while keeping failures as reactive error signals.
+ *
+ * Raw channel used by {@link ToolExecutor}'s execution infrastructure: only the
+ * error-to-result conversion of {@link #invokeAsync} is omitted, so the timeout and retry
+ * layers can act on failures.
+ *
+ * @param toolObject the object containing the method
+ * @param method the method to invoke
+ * @param param the tool call parameters containing input, toolUseBlock, agent, and context
+ * @param customConverter custom converter for this invocation (null to use default)
+ * @return Mono containing ToolResultBlock, or signalling an error on failure
+ */
+ Mono invokeRawAsync(
+ Object toolObject,
+ Method method,
+ ToolCallParam param,
+ ToolResultConverter customConverter) {
// Use custom converter if provided, otherwise use default
final ToolResultConverter converter =
customConverter != null ? customConverter : defaultConverter;
@@ -87,9 +111,8 @@ Mono invokeAsync(
.map(
r ->
converter.convert(
- r, extractGenericType(method)))
- .onErrorResume(this::handleError))
- .onErrorResume(this::handleError);
+ r,
+ extractGenericType(method))));
} else if (returnType == Mono.class) {
// Async method returning Mono: invoke and flatMap
@@ -105,22 +128,19 @@ r, extractGenericType(method)))
})
.flatMap(
mono ->
- mono.map(r -> converter.convert(r, extractGenericType(method)))
- .onErrorResume(this::handleError))
- .onErrorResume(this::handleError);
+ mono.map(
+ r -> converter.convert(r, extractGenericType(method))));
} else {
// Sync method: wrap in Mono.fromCallable
return Mono.fromCallable(
- () -> {
- method.setAccessible(true);
- Object[] args =
- convertParameters(
- method, input, agent, runtimeContext, emitter);
- Object result = method.invoke(toolObject, args);
- return converter.convert(result, method.getGenericReturnType());
- })
- .onErrorResume(this::handleError);
+ () -> {
+ method.setAccessible(true);
+ Object[] args =
+ convertParameters(method, input, agent, runtimeContext, emitter);
+ Object result = method.invoke(toolObject, args);
+ return converter.convert(result, method.getGenericReturnType());
+ });
}
}
diff --git a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java
index b7b3a9ec49..dcb6e19110 100644
--- a/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java
+++ b/agentscope-core/src/main/java/io/agentscope/core/tool/mcp/McpTool.java
@@ -178,6 +178,32 @@ public Mono checkPermissions(
*/
@Override
public Mono callAsync(ToolCallParam param) {
+ return callAsyncForExecution(param)
+ .onErrorResume(
+ e -> {
+ logger.error(
+ "Error calling MCP tool '{}': {}", getName(), e.getMessage());
+ String errorMsg =
+ e.getMessage() != null
+ ? e.getMessage()
+ : e.getClass().getSimpleName();
+ return Mono.just(ToolResultBlock.error("MCP tool error: " + errorMsg));
+ });
+ }
+
+ /**
+ * Executes this MCP tool through the framework's execution infrastructure.
+ *
+ * Transport-level failures stay as reactive error signals so the executor can apply its
+ * retry policy. Protocol-level business errors ({@code isError=true}) remain completed error
+ * results and are never retried.
+ *
+ * @param param The tool call parameters containing toolUseBlock, input, and agent
+ * @return a Mono that emits the tool result, or signals an error on failure
+ */
+ @Override
+ public Mono callAsyncForExecution(ToolCallParam param) {
+ Objects.requireNonNull(param, "param must not be null");
logger.debug("Calling MCP tool '{}' with input: {}", getName(), param.getInput());
// Merge preset arguments with input arguments
@@ -190,17 +216,7 @@ public Mono callAsync(ToolCallParam param) {
.callTool(getName(), mergedArgs, metaMap)
.map(McpContentConverter::convertCallToolResult)
.doOnSuccess(
- result -> logger.debug("MCP tool '{}' completed successfully", getName()))
- .onErrorResume(
- e -> {
- logger.error(
- "Error calling MCP tool '{}': {}", getName(), e.getMessage());
- String errorMsg =
- e.getMessage() != null
- ? e.getMessage()
- : e.getClass().getSimpleName();
- return Mono.just(ToolResultBlock.error("MCP tool error: " + errorMsg));
- });
+ result -> logger.debug("MCP tool '{}' completed successfully", getName()));
}
/**
diff --git a/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java b/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java
index 681793e6a0..272f995f9a 100644
--- a/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java
+++ b/agentscope-core/src/main/java/io/agentscope/core/util/ExceptionUtils.java
@@ -15,7 +15,11 @@
*/
package io.agentscope.core.util;
+import io.agentscope.core.tool.ToolSuspendException;
+import java.lang.reflect.InvocationTargetException;
import java.util.IdentityHashMap;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
/**
* Utility methods for exception handling.
@@ -79,4 +83,46 @@ public static boolean containsInterruptedException(Throwable error) {
}
return false;
}
+
+ /**
+ * Unwraps {@link InvocationTargetException}, {@link ExecutionException} and
+ * {@link CompletionException} wrappers to expose the original failure.
+ *
+ * @param error the throwable to unwrap (may be {@code null})
+ * @return the innermost non-wrapper throwable, or {@code null} if the input is {@code null}
+ */
+ public static Throwable unwrapExecutionWrapper(Throwable error) {
+ Throwable current = error;
+ while (current instanceof InvocationTargetException
+ || current instanceof ExecutionException
+ || current instanceof CompletionException) {
+ Throwable cause = current.getCause();
+ if (cause == null) {
+ break;
+ }
+ current = cause;
+ }
+ return current;
+ }
+
+ /**
+ * Finds a {@link ToolSuspendException} anywhere in the given throwable's cause chain.
+ *
+ * The cause chain is walked with an identity set guard so circular causes cannot cause an
+ * infinite loop.
+ *
+ * @param error the throwable to inspect (may be {@code null})
+ * @return the first {@link ToolSuspendException} found, or {@code null} if none is present
+ */
+ public static ToolSuspendException findToolSuspendException(Throwable error) {
+ IdentityHashMap visited = new IdentityHashMap<>();
+ Throwable current = error;
+ while (current != null && visited.put(current, Boolean.TRUE) == null) {
+ if (current instanceof ToolSuspendException suspendException) {
+ return suspendException;
+ }
+ current = current.getCause();
+ }
+ return null;
+ }
}
diff --git a/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java b/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java
index 55ae0cb786..da8c048885 100644
--- a/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java
+++ b/agentscope-core/src/test/java/io/agentscope/core/tool/ToolExecutorTest.java
@@ -18,19 +18,32 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import io.agentscope.core.message.ContentBlock;
import io.agentscope.core.message.TextBlock;
import io.agentscope.core.message.ToolResultBlock;
+import io.agentscope.core.message.ToolResultState;
import io.agentscope.core.message.ToolUseBlock;
+import io.agentscope.core.model.ExecutionConfig;
import io.agentscope.core.model.ToolSchema;
+import io.agentscope.core.tool.mcp.McpClientWrapper;
+import io.agentscope.core.tool.mcp.McpTool;
import io.agentscope.core.tool.test.SampleTools;
import io.agentscope.core.tool.test.ToolTestUtils;
import io.agentscope.core.util.JsonUtils;
+import io.modelcontextprotocol.spec.McpSchema;
+import java.io.IOException;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -669,6 +682,698 @@ public Mono callAsync(ToolCallParam param) {
}
}
+ // ==================== Retry Behavior Tests ====================
+
+ @Test
+ @DisplayName("Should retry annotation sync tools that throw exceptions")
+ void shouldRetryAnnotationSyncToolExceptions() {
+ FlakyTools flaky = new FlakyTools();
+ toolkit.registerTool(flaky);
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-flaky-sync", "flaky_sync")),
+ retryConfig(2),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals("\"recovered\"", extractFirstText(responses.get(0)));
+ assertEquals(2, flaky.calls.get());
+ }
+
+ @Test
+ @DisplayName("Should retry annotation Mono tools that fail with exceptions")
+ void shouldRetryAnnotationMonoToolExceptions() {
+ FlakyTools flaky = new FlakyTools();
+ toolkit.registerTool(flaky);
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-flaky-mono", "flaky_mono")),
+ retryConfig(2),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals("\"recovered\"", extractFirstText(responses.get(0)));
+ assertEquals(2, flaky.calls.get());
+ }
+
+ @Test
+ @DisplayName("Should retry annotation CompletableFuture tools that fail with exceptions")
+ void shouldRetryAnnotationFutureToolExceptions() {
+ FlakyTools flaky = new FlakyTools();
+ toolkit.registerTool(flaky);
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-flaky-future", "flaky_future")),
+ retryConfig(2),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals("\"recovered\"", extractFirstText(responses.get(0)));
+ assertEquals(2, flaky.calls.get());
+ }
+
+ @Test
+ @DisplayName("Should retry MCP tool calls on transport failures")
+ void shouldRetryMcpToolTransportFailures() {
+ McpClientWrapper wrapper = mock(McpClientWrapper.class);
+ when(wrapper.getName()).thenReturn("test-client");
+ McpTool mcpTool = new McpTool("flaky_mcp", "Flaky MCP tool", emptySchema(), wrapper);
+ when(wrapper.callTool(eq("flaky_mcp"), any(), any()))
+ .thenReturn(Mono.error(new IOException("Network down")))
+ .thenReturn(
+ Mono.just(
+ new McpSchema.CallToolResult(
+ List.of(new McpSchema.TextContent("mcp recovered")),
+ false)));
+ toolkit.registerTool(mcpTool);
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-flaky-mcp", "flaky_mcp")),
+ retryConfig(2),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertTrue(extractFirstText(responses.get(0)).contains("mcp recovered"));
+ verify(wrapper, times(2)).callTool(eq("flaky_mcp"), any(), any());
+ }
+
+ @Test
+ @DisplayName("Should not retry MCP protocol-level business errors")
+ void shouldNotRetryMcpProtocolErrors() {
+ McpClientWrapper wrapper = mock(McpClientWrapper.class);
+ when(wrapper.getName()).thenReturn("test-client");
+ McpTool mcpTool = new McpTool("mcp_error", "Failing MCP tool", emptySchema(), wrapper);
+ when(wrapper.callTool(eq("mcp_error"), any(), any()))
+ .thenReturn(
+ Mono.just(
+ new McpSchema.CallToolResult(
+ List.of(new McpSchema.TextContent("boom")), true)));
+ toolkit.registerTool(mcpTool);
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-mcp-error", "mcp_error")),
+ retryConfig(3),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(ToolResultState.ERROR, responses.get(0).getState());
+ verify(wrapper, times(1)).callTool(eq("mcp_error"), any(), any());
+ }
+
+ @Test
+ @DisplayName("Should retry custom AgentTool failures that surface as errors")
+ void shouldRetryCustomAgentToolErrors() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "custom_flaky";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Custom tool that fails once";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ int attempt = calls.incrementAndGet();
+ if (attempt == 1) {
+ return Mono.error(new IOException("Transient custom failure"));
+ }
+ return Mono.just(ToolResultBlock.text("custom recovered"));
+ }
+ });
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-custom-flaky", "custom_flaky")),
+ retryConfig(2),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals("custom recovered", extractFirstText(responses.get(0)));
+ assertEquals(2, calls.get());
+ }
+
+ @Test
+ @DisplayName("Should not retry tool-returned error results")
+ void shouldNotRetrySemanticErrorResults() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "semantic_error";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Tool that reports a business error result";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ calls.incrementAndGet();
+ return Mono.just(ToolResultBlock.error("Business failure"));
+ }
+ });
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-semantic-error", "semantic_error")),
+ retryConfig(3),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(1, calls.get());
+ assertTrue(extractFirstText(responses.get(0)).contains("Business failure"));
+ }
+
+ @Test
+ @DisplayName("Should respect the retryOn predicate")
+ void shouldRespectRetryOnPredicate() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "never_retry";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Tool whose failures are filtered out by retryOn";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ calls.incrementAndGet();
+ return Mono.error(new IOException("Always failing"));
+ }
+ });
+
+ ExecutionConfig config =
+ ExecutionConfig.builder()
+ .maxAttempts(3)
+ .initialBackoff(Duration.ofMillis(1))
+ .maxBackoff(Duration.ofMillis(10))
+ .retryOn(error -> false)
+ .build();
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-never-retry", "never_retry")),
+ config,
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(1, calls.get());
+ assertEquals(ToolResultState.ERROR, responses.get(0).getState());
+ }
+
+ @Test
+ @DisplayName("Should exhaust retries and return an error result with id and name")
+ void shouldExhaustRetriesWithErrorResult() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "always_fail";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Tool that always fails";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ calls.incrementAndGet();
+ return Mono.error(new IOException("Always failing"));
+ }
+ });
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-always-fail", "always_fail")),
+ retryConfig(3),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(3, calls.get());
+ ToolResultBlock result = responses.get(0);
+ assertEquals(ToolResultState.ERROR, result.getState());
+ assertEquals("call-always-fail", result.getId());
+ assertEquals("always_fail", result.getName());
+ assertTrue(extractFirstText(result).contains("Always failing"));
+ }
+
+ @Test
+ @DisplayName("Should retry tool execution timeouts when retryOn matches")
+ void shouldRetryTimeouts() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "never_tool";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Tool that never completes";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ calls.incrementAndGet();
+ return Mono.never();
+ }
+ });
+
+ ExecutionConfig config =
+ ExecutionConfig.builder()
+ .timeout(Duration.ofMillis(100))
+ .maxAttempts(2)
+ .initialBackoff(Duration.ofMillis(1))
+ .maxBackoff(Duration.ofMillis(10))
+ .retryOn(ExecutionConfig.RETRYABLE_ERRORS)
+ .build();
+
+ List responses =
+ toolkit.callTools(List.of(toolCall("call-never", "never_tool")), config, null, null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(2, calls.get());
+ assertTrue(extractFirstText(responses.get(0)).contains("timeout"));
+ }
+
+ @Test
+ @DisplayName("Should isolate retry exhaustion in parallel batches")
+ void shouldIsolateRetryExhaustionInParallelBatches() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "parallel_fail";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Tool that always fails in parallel batches";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ calls.incrementAndGet();
+ return Mono.error(new IOException("Always failing"));
+ }
+ });
+
+ Map addInput = Map.of("a", 1, "b", 2);
+ List responses =
+ toolkit.callTools(
+ List.of(
+ toolCall("call-parallel-fail", "parallel_fail"),
+ toolCall("call-parallel-add", "add", addInput)),
+ retryConfig(2),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(2, responses.size());
+ Map byId =
+ responses.stream()
+ .collect(Collectors.toMap(ToolResultBlock::getId, Function.identity()));
+ assertEquals(2, calls.get());
+ assertEquals(ToolResultState.ERROR, byId.get("call-parallel-fail").getState());
+ assertEquals("3", extractFirstText(byId.get("call-parallel-add")));
+ }
+
+ @Test
+ @DisplayName("Should not retry external tool suspensions")
+ void shouldNotRetryExternalToolSuspensions() {
+ toolkit.registerSchema(
+ ToolSchema.builder()
+ .name("external_retry")
+ .description("External tool with retry config")
+ .parameters(
+ Map.of(
+ "type",
+ "object",
+ "properties",
+ Map.of("endpoint", Map.of("type", "string"))))
+ .build());
+
+ List responses =
+ toolkit.callTools(
+ List.of(
+ toolCall(
+ "call-external-retry",
+ "external_retry",
+ Map.of("endpoint", "/users"))),
+ retryConfig(3),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertTrue(responses.get(0).isSuspended());
+ }
+
+ @Test
+ @DisplayName("Should not retry annotation tools that suspend via ToolSuspendException")
+ void shouldNotRetryToolSuspensionExceptions() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new Object() {
+ @Tool(name = "suspend_tool", description = "Tool that suspends")
+ public String suspend() {
+ calls.incrementAndGet();
+ throw new ToolSuspendException("awaiting external execution");
+ }
+ });
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-suspend", "suspend_tool")),
+ retryConfig(3),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(1, calls.get());
+ assertTrue(responses.get(0).isSuspended());
+ }
+
+ @Test
+ @DisplayName("Should not retry Mono tools that suspend via ToolSuspendException")
+ void shouldNotRetryMonoToolSuspension() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new Object() {
+ @Tool(name = "suspend_mono", description = "Mono tool that suspends")
+ public Mono suspend() {
+ calls.incrementAndGet();
+ return Mono.error(new ToolSuspendException("awaiting external execution"));
+ }
+ });
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-suspend-mono", "suspend_mono")),
+ retryConfig(3),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(1, calls.get());
+ assertTrue(responses.get(0).isSuspended());
+ }
+
+ @Test
+ @DisplayName("Should not retry CompletableFuture tools that suspend via ToolSuspendException")
+ void shouldNotRetryFutureToolSuspension() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new Object() {
+ @Tool(name = "suspend_future", description = "Future tool that suspends")
+ public CompletableFuture suspend() {
+ calls.incrementAndGet();
+ return CompletableFuture.failedFuture(
+ new ToolSuspendException("awaiting external execution"));
+ }
+ });
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-suspend-future", "suspend_future")),
+ retryConfig(3),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(1, calls.get());
+ assertTrue(responses.get(0).isSuspended());
+ }
+
+ @Test
+ @DisplayName("Should not retry ToolSuspendException wrapped in a non-standard exception")
+ void shouldNotRetryWrappedToolSuspension() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "wrapped_suspend";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Tool that suspends behind a custom wrapper";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ return Mono.just(ToolResultBlock.error("unused"));
+ }
+
+ @Override
+ public Mono callAsyncForExecution(ToolCallParam param) {
+ calls.incrementAndGet();
+ return Mono.error(
+ new IllegalStateException(
+ "wrapped",
+ new ToolSuspendException("awaiting external execution")));
+ }
+ });
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-wrapped-suspend", "wrapped_suspend")),
+ retryConfig(3),
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(1, calls.get());
+ assertTrue(responses.get(0).isSuspended());
+ }
+
+ @Test
+ @DisplayName("Should not retry deterministic failures with the retryable-errors predicate")
+ void shouldNotRetryDeterministicFailureWithRetryableErrors() {
+ AtomicInteger calls = new AtomicInteger(0);
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "misconfigured_tool";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Tool that fails deterministically";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ calls.incrementAndGet();
+ return Mono.error(new IllegalStateException("Not initialized"));
+ }
+ });
+
+ ExecutionConfig config =
+ ExecutionConfig.builder()
+ .maxAttempts(3)
+ .initialBackoff(Duration.ofMillis(1))
+ .maxBackoff(Duration.ofMillis(10))
+ .retryOn(ExecutionConfig.RETRYABLE_ERRORS)
+ .build();
+
+ List responses =
+ toolkit.callTools(
+ List.of(toolCall("call-misconfigured", "misconfigured_tool")),
+ config,
+ null,
+ null)
+ .block(TIMEOUT);
+
+ assertEquals(1, responses.size());
+ assertEquals(1, calls.get());
+ assertEquals(ToolResultState.ERROR, responses.get(0).getState());
+ }
+
+ @Test
+ @DisplayName("Should keep the callAsync error-result contract on direct calls")
+ void shouldKeepCallAsyncContractOnDirectCalls() {
+ // Custom tool failing with an error signal: direct calls still receive an error result
+ toolkit.registerTool(
+ new AgentTool() {
+ @Override
+ public String getName() {
+ return "direct_error";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Tool that fails with an error signal";
+ }
+
+ @Override
+ public Map getParameters() {
+ return emptySchema();
+ }
+
+ @Override
+ public Mono callAsync(ToolCallParam param) {
+ return Mono.error(new IllegalStateException("Direct failure"));
+ }
+ });
+ ToolResultBlock result =
+ toolkit.callTool(
+ ToolCallParam.builder()
+ .toolUseBlock(toolCall("call-direct", "direct_error"))
+ .build())
+ .block(TIMEOUT);
+ assertEquals("Error: Tool execution failed: Direct failure", extractFirstText(result));
+
+ // Annotation tools keep converting exceptions to error results on the direct path
+ ToolResultBlock annotationResult =
+ toolkit.callTool(
+ ToolCallParam.builder()
+ .toolUseBlock(
+ toolCall(
+ "call-direct-annotation",
+ "error_tool",
+ Map.of("message", "direct")))
+ .input(Map.of("message", "direct"))
+ .build())
+ .block(TIMEOUT);
+ assertEquals(
+ "Error: Tool execution failed: Tool error: direct",
+ extractFirstText(annotationResult));
+
+ // MCP tools keep their own error formatting on the direct path
+ McpClientWrapper wrapper = mock(McpClientWrapper.class);
+ when(wrapper.getName()).thenReturn("test-client");
+ McpTool mcpTool = new McpTool("direct_mcp", "Direct MCP tool", emptySchema(), wrapper);
+ when(wrapper.callTool(eq("direct_mcp"), any(), any()))
+ .thenReturn(Mono.error(new RuntimeException("Network error")));
+ toolkit.registerTool(mcpTool);
+ ToolResultBlock mcpResult =
+ toolkit.callTool(
+ ToolCallParam.builder()
+ .toolUseBlock(toolCall("call-direct-mcp", "direct_mcp"))
+ .build())
+ .block(TIMEOUT);
+ String mcpText = extractFirstText(mcpResult);
+ assertTrue(mcpText.contains("MCP tool error"));
+ assertTrue(mcpText.contains("Network error"));
+ }
+
+ // ==================== Test Helpers ====================
+
+ private ToolUseBlock toolCall(String id, String name) {
+ return toolCall(id, name, Map.of());
+ }
+
+ private ToolUseBlock toolCall(String id, String name, Map input) {
+ return ToolUseBlock.builder()
+ .id(id)
+ .name(name)
+ .input(input)
+ .content(JsonUtils.getJsonCodec().toJson(input))
+ .build();
+ }
+
+ private ExecutionConfig retryConfig(int maxAttempts) {
+ return ExecutionConfig.builder()
+ .maxAttempts(maxAttempts)
+ .initialBackoff(Duration.ofMillis(1))
+ .maxBackoff(Duration.ofMillis(10))
+ .build();
+ }
+
+ private Map emptySchema() {
+ Map schema = new HashMap<>();
+ schema.put("type", "object");
+ schema.put("properties", new HashMap<>());
+ return schema;
+ }
+
private String extractFirstText(ToolResultBlock response) {
assertTrue(
ToolTestUtils.isValidToolResultBlock(response),
@@ -677,4 +1382,36 @@ private String extractFirstText(ToolResultBlock response) {
if (outputs.isEmpty()) return "";
return ((TextBlock) outputs.get(0)).getText();
}
+
+ /** Annotation-based tools that fail on the first call and recover afterwards. */
+ public static class FlakyTools {
+
+ final AtomicInteger calls = new AtomicInteger(0);
+
+ @Tool(name = "flaky_sync", description = "Throws IOException on the first call")
+ public String flakySync() throws IOException {
+ if (calls.incrementAndGet() == 1) {
+ throw new IOException("Transient sync failure");
+ }
+ return "recovered";
+ }
+
+ @Tool(name = "flaky_mono", description = "Fails with IOException on the first call")
+ public Mono flakyMono() {
+ int attempt = calls.incrementAndGet();
+ if (attempt == 1) {
+ return Mono.error(new IOException("Transient mono failure"));
+ }
+ return Mono.just("recovered");
+ }
+
+ @Tool(name = "flaky_future", description = "Fails with IOException on the first call")
+ public CompletableFuture flakyFuture() {
+ int attempt = calls.incrementAndGet();
+ if (attempt == 1) {
+ return CompletableFuture.failedFuture(new IOException("Transient future failure"));
+ }
+ return CompletableFuture.completedFuture("recovered");
+ }
+ }
}