This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 660de9efd [API][Tool] Align Java and Python Tool outcome contracts
(#1074)
660de9efd is described below
commit 660de9efdeb998935187b63b81541694c41319eb
Author: Yu Tang <[email protected]>
AuthorDate: Mon Sep 14 14:34:36 2026 +0800
[API][Tool] Align Java and Python Tool outcome contracts (#1074)
Co-authored-by: Claude Code <[email protected]>
---
.../api/resource/python/PythonResourceAdapter.java | 3 +-
docs/content/docs/development/tool_use.md | 2 +
.../resource/test/MCPCrossLanguageAgent.java | 7 ++
.../src/test/resources/mcp_server.py | 6 ++
.../flink/agents/integrations/mcp/MCPServer.java | 45 +++++++-----
.../agents/integrations/mcp/MCPServerTest.java | 25 +++++++
.../agents/plan/resource/python/PythonMCPTool.java | 6 +-
.../resource/python/PythonToolResultConverter.java | 63 +++++++++++++++++
.../flink/agents/plan/tools/FunctionTool.java | 4 +-
.../plan/resource/python/PythonMCPToolTest.java | 71 +++++++++++++++++++
.../tools/FunctionToolSetPythonAdapterTest.java | 79 ++++++++++++++++++++++
.../__init__.py => tests/test_tool_response.py} | 30 ++++----
python/flink_agents/api/tools/__init__.py | 2 +
python/flink_agents/api/tools/tool.py | 3 +
python/flink_agents/api/tools/tool_response.py | 76 +++++++++++++++++++++
python/flink_agents/integrations/mcp/mcp.py | 31 ++++++---
.../integrations/mcp/tests/test_mcp.py | 79 ++++++++++++++++++++++
.../flink_agents/plan/actions/tool_call_action.py | 44 ++++++++----
python/flink_agents/plan/function.py | 30 ++++++--
.../plan/tests/actions/test_tool_call_action.py | 56 ++++++++++++++-
.../plan/tests/tools/test_function_tool.py | 42 +++++++++++-
python/flink_agents/runtime/python_java_utils.py | 31 ++++++++-
python/flink_agents/runtime/skill/skill_tools.py | 16 +++--
.../runtime/skill/tests/test_load_skill.py | 23 +++++--
.../runtime/tests/test_python_java_utils.py | 69 ++++++++++++++++++-
.../runtime/python/utils/JavaResourceAdapter.java | 21 ++++--
.../flink/agents/runtime/skill/LoadSkillTool.java | 8 +--
.../python/utils/JavaResourceAdapterTest.java | 53 ++++++++++++++-
.../agents/runtime/skill/LoadSkillToolTest.java | 9 ++-
29 files changed, 842 insertions(+), 92 deletions(-)
diff --git
a/api/src/main/java/org/apache/flink/agents/api/resource/python/PythonResourceAdapter.java
b/api/src/main/java/org/apache/flink/agents/api/resource/python/PythonResourceAdapter.java
index 9c4ad0bb6..272e1eeb7 100644
---
a/api/src/main/java/org/apache/flink/agents/api/resource/python/PythonResourceAdapter.java
+++
b/api/src/main/java/org/apache/flink/agents/api/resource/python/PythonResourceAdapter.java
@@ -175,7 +175,8 @@ public interface PythonResourceAdapter {
* @param qualName the qualified name of the callable inside the module
* @param kwargs keyword arguments to pass to the callable; LLM tool calls
always arrive as
* keyword arguments
- * @return the raw return value from the Python callable
+ * @return the bridge-encoded tool result; implementations may return a
raw value for backward
+ * compatibility
*/
Object invokePythonTool(String module, String qualName, Map<String,
Object> kwargs);
}
diff --git a/docs/content/docs/development/tool_use.md
b/docs/content/docs/development/tool_use.md
index 12a35810c..c468ba833 100644
--- a/docs/content/docs/development/tool_use.md
+++ b/docs/content/docs/development/tool_use.md
@@ -336,6 +336,8 @@ See [MCP]({{< ref "docs/development/mcp" >}}) for details.
The built-in `tool_call_action` listens to `ToolRequestEvent`. For each tool
call, it looks up the tool resource by function name, executes it through
durable execution, and records whether it succeeded. After all tool calls in
the batch have been processed, it sends a `ToolResponseEvent`.
+Python tools can continue returning raw values, which are treated as
successful results. A tool that completes normally but cannot perform the
requested operation can return `ToolResponse.error("reason")`; an exception
still represents an invocation failure. Both failure forms are recorded as
failed tool calls in `ToolResponseEvent`.
+
When the tool request comes from `chat_model_action`, the emitted
`ToolResponseEvent` is automatically consumed by `chat_model_action` to
continue the chat. See [Built-in Events and Actions in Chat Models]({{< ref
"docs/development/chat_models#built-in-events-and-actions" >}}) for details on
how `chat_model_action` handles tool responses.
Users can also send `ToolRequestEvent` directly when they want to invoke tools
programmatically.
diff --git
a/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/MCPCrossLanguageAgent.java
b/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/MCPCrossLanguageAgent.java
index 30b921e8c..74ecfecbc 100644
---
a/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/MCPCrossLanguageAgent.java
+++
b/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/MCPCrossLanguageAgent.java
@@ -59,6 +59,13 @@ public class MCPCrossLanguageAgent extends Agent {
ToolResponse response = add.call(new ToolParameters(Map.of("a", 1,
"b", 2)));
Assertions.assertTrue(response.getResult().toString().contains("3"));
+
+ Tool failingTool = (Tool)
ctx.getResource("fail_with_recovery_hint", ResourceType.TOOL);
+ ToolResponse failedResponse =
+ failingTool.call(new ToolParameters(Map.of("query", "all
records")));
+ Assertions.assertTrue(failedResponse.isError());
+ Assertions.assertTrue(
+ failedResponse.getError().contains("retry with a narrower
query"));
System.out.println("[TEST] MCP Tools PASSED");
Prompt askSum = (Prompt) ctx.getResource("ask_sum",
ResourceType.PROMPT);
diff --git
a/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/resources/mcp_server.py
b/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/resources/mcp_server.py
index 5c1d90a81..1f8bb7c62 100644
---
a/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/resources/mcp_server.py
+++
b/e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/resources/mcp_server.py
@@ -50,4 +50,10 @@ async def add(a: int, b: int) -> int:
return a + b
[email protected]()
+async def fail_with_recovery_hint(query: str) -> str:
+ """Return a protocol-level tool error with a model-facing recovery hint."""
+ raise ValueError(f"retry with a narrower query than '{query}'")
+
+
mcp.run("streamable-http")
diff --git
a/integrations/mcp/src/main/java/org/apache/flink/agents/integrations/mcp/MCPServer.java
b/integrations/mcp/src/main/java/org/apache/flink/agents/integrations/mcp/MCPServer.java
index 5692de7a0..217bdecfc 100644
---
a/integrations/mcp/src/main/java/org/apache/flink/agents/integrations/mcp/MCPServer.java
+++
b/integrations/mcp/src/main/java/org/apache/flink/agents/integrations/mcp/MCPServer.java
@@ -443,26 +443,37 @@ public class MCPServer extends Resource {
* @param toolName The name of the tool to call
* @param arguments The arguments to pass to the tool
* @return The result as a list of content items
+ * @throws IllegalStateException if the MCP response reports a
protocol-level tool error
*/
public List<Object> callTool(String toolName, Map<String, Object>
arguments) {
- return getRetryExecutor()
- .execute(
- () -> {
- McpSyncClient mcpClient = getClient();
- McpSchema.CallToolRequest request =
- new McpSchema.CallToolRequest(
- toolName,
- arguments != null ? arguments :
new HashMap<>());
- McpSchema.CallToolResult result =
mcpClient.callTool(request);
-
- List<Object> content = new ArrayList<>();
- for (var item : result.content()) {
-
content.add(MCPContentExtractor.extractContentItem(item));
- }
+ McpSchema.CallToolResult result =
+ getRetryExecutor()
+ .execute(
+ () -> {
+ McpSyncClient mcpClient = getClient();
+ McpSchema.CallToolRequest request =
+ new McpSchema.CallToolRequest(
+ toolName,
+ arguments != null
+ ? arguments
+ : new HashMap<>());
+ return mcpClient.callTool(request);
+ },
+ "callTool:" + toolName);
+ return extractToolContent(toolName, result);
+ }
- return content;
- },
- "callTool:" + toolName);
+ static List<Object> extractToolContent(
+ String toolName, McpSchema.CallToolResult callToolResult) {
+ List<Object> content = new ArrayList<>();
+ for (var item : callToolResult.content()) {
+ content.add(MCPContentExtractor.extractContentItem(item));
+ }
+ if (Boolean.TRUE.equals(callToolResult.isError())) {
+ throw new IllegalStateException(
+ "MCP tool '" + toolName + "' returned an error: " +
content);
+ }
+ return content;
}
/**
diff --git
a/integrations/mcp/src/test/java/org/apache/flink/agents/integrations/mcp/MCPServerTest.java
b/integrations/mcp/src/test/java/org/apache/flink/agents/integrations/mcp/MCPServerTest.java
index c63bbd187..847fd2248 100644
---
a/integrations/mcp/src/test/java/org/apache/flink/agents/integrations/mcp/MCPServerTest.java
+++
b/integrations/mcp/src/test/java/org/apache/flink/agents/integrations/mcp/MCPServerTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.agents.integrations.mcp;
import com.fasterxml.jackson.databind.ObjectMapper;
+import io.modelcontextprotocol.spec.McpSchema;
import org.apache.flink.agents.api.resource.ResourceContext;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
import org.apache.flink.agents.api.resource.ResourceName;
@@ -306,4 +307,28 @@ class MCPServerTest {
List<MCPPrompt> prompts = server.listPrompts();
assertThat(prompts).isEmpty();
}
+
+ @Test
+ @DisabledOnJre(JRE.JAVA_11)
+ void protocolToolErrorIsNotReturnedAsSuccessfulContent() {
+ McpSchema.CallToolResult result =
+ McpSchema.CallToolResult.builder()
+ .addTextContent("business failure")
+ .isError(true)
+ .build();
+
+ assertThatThrownBy(() -> MCPServer.extractToolContent("lookup",
result))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("lookup")
+ .hasMessageContaining("business failure");
+ }
+
+ @Test
+ @DisabledOnJre(JRE.JAVA_11)
+ void successfulProtocolToolResultReturnsContent() {
+ McpSchema.CallToolResult result =
+
McpSchema.CallToolResult.builder().addTextContent("result").build();
+
+ assertThat(MCPServer.extractToolContent("lookup",
result)).containsExactly("result");
+ }
}
diff --git
a/plan/src/main/java/org/apache/flink/agents/plan/resource/python/PythonMCPTool.java
b/plan/src/main/java/org/apache/flink/agents/plan/resource/python/PythonMCPTool.java
index 8a831db5a..ecd16286a 100644
---
a/plan/src/main/java/org/apache/flink/agents/plan/resource/python/PythonMCPTool.java
+++
b/plan/src/main/java/org/apache/flink/agents/plan/resource/python/PythonMCPTool.java
@@ -39,6 +39,8 @@ public class PythonMCPTool extends Tool
implements PythonResourceWrapper, ToolExecutionMetadataProvider {
private static final String GET_JAVA_TOOL_META =
"python_java_utils.get_java_tool_metadata_from_tool";
+ private static final String INVOKE_PYTHON_TOOL =
+ "python_java_utils.invoke_python_tool_instance";
private final PyObject tool;
private final PythonResourceAdapter adapter;
@Nullable private final String mcpServerName;
@@ -84,8 +86,8 @@ public class PythonMCPTool extends Tool
kwargs.put(paramName, parameters.getParameter(paramName));
}
try {
- Object result = adapter.callMethod(tool, "call", kwargs);
- return ToolResponse.success(result);
+ Object result = adapter.invoke(INVOKE_PYTHON_TOOL, tool, kwargs);
+ return PythonToolResultConverter.fromBridgeResult(result);
} catch (Exception e) {
return ToolResponse.error(e);
}
diff --git
a/plan/src/main/java/org/apache/flink/agents/plan/resource/python/PythonToolResultConverter.java
b/plan/src/main/java/org/apache/flink/agents/plan/resource/python/PythonToolResultConverter.java
new file mode 100644
index 000000000..0e1218782
--- /dev/null
+++
b/plan/src/main/java/org/apache/flink/agents/plan/resource/python/PythonToolResultConverter.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.plan.resource.python;
+
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.annotation.Internal;
+
+import java.util.Map;
+
+/** Converts the internal Python bridge representation into a Java {@link
ToolResponse}. */
+@Internal
+public final class PythonToolResultConverter {
+
+ private static final String RESULT_MARKER = "__flink_agents_tool_result__";
+
+ public static ToolResponse fromBridgeResult(Object result) {
+ if (!(result instanceof Map)) {
+ return ToolResponse.success(result);
+ }
+
+ Map<?, ?> response = (Map<?, ?>) result;
+ Object resultKind = response.get(RESULT_MARKER);
+ if ("raw".equals(resultKind)) {
+ return ToolResponse.success(response.get("result"));
+ }
+ if (!"response".equals(resultKind)) {
+ return ToolResponse.success(result);
+ }
+
+ long executionTimeMs = numberValue(response.get("execution_time_ms"));
+ String toolName = stringValue(response.get("tool_name"));
+ if (Boolean.TRUE.equals(response.get("success"))) {
+ return ToolResponse.success(response.get("result"),
executionTimeMs, toolName);
+ }
+ return ToolResponse.error(stringValue(response.get("error")),
executionTimeMs, toolName);
+ }
+
+ private static long numberValue(Object value) {
+ return value instanceof Number ? ((Number) value).longValue() : 0L;
+ }
+
+ private static String stringValue(Object value) {
+ return value == null ? null : String.valueOf(value);
+ }
+
+ private PythonToolResultConverter() {}
+}
diff --git
a/plan/src/main/java/org/apache/flink/agents/plan/tools/FunctionTool.java
b/plan/src/main/java/org/apache/flink/agents/plan/tools/FunctionTool.java
index 83534d31b..4c0178af2 100644
--- a/plan/src/main/java/org/apache/flink/agents/plan/tools/FunctionTool.java
+++ b/plan/src/main/java/org/apache/flink/agents/plan/tools/FunctionTool.java
@@ -36,6 +36,7 @@ import org.apache.flink.agents.api.tools.ToolType;
import org.apache.flink.agents.plan.Function;
import org.apache.flink.agents.plan.JavaFunction;
import org.apache.flink.agents.plan.PythonFunction;
+import org.apache.flink.agents.plan.resource.python.PythonToolResultConverter;
import
org.apache.flink.agents.plan.tools.serializer.FunctionToolJsonDeserializer;
import
org.apache.flink.agents.plan.tools.serializer.FunctionToolJsonSerializer;
@@ -57,7 +58,6 @@ import java.util.Map;
public class FunctionTool extends Tool {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
private final Function function;
private Map<String, ToolParameterInjection> injectedArgs;
@@ -185,7 +185,7 @@ public class FunctionTool extends Tool {
}
Object result =
pythonResourceAdapter.invokePythonTool(pf.getModule(),
pf.getQualName(), kwargs);
- return ToolResponse.success(result);
+ return PythonToolResultConverter.fromBridgeResult(result);
}
public Function getFunction() {
diff --git
a/plan/src/test/java/org/apache/flink/agents/plan/resource/python/PythonMCPToolTest.java
b/plan/src/test/java/org/apache/flink/agents/plan/resource/python/PythonMCPToolTest.java
new file mode 100644
index 000000000..c6ce28b84
--- /dev/null
+++
b/plan/src/test/java/org/apache/flink/agents/plan/resource/python/PythonMCPToolTest.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.plan.resource.python;
+
+import org.apache.flink.agents.api.resource.python.PythonResourceAdapter;
+import org.apache.flink.agents.api.tools.ToolParameters;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.junit.jupiter.api.Test;
+import pemja.core.object.PyObject;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class PythonMCPToolTest {
+
+ @Test
+ void preservesExplicitPythonMcpToolFailure() {
+ PythonResourceAdapter adapter = mock(PythonResourceAdapter.class);
+ PyObject pythonTool = mock(PyObject.class);
+
when(adapter.invoke("python_java_utils.get_java_tool_metadata_from_tool",
pythonTool))
+ .thenReturn(
+ Map.of(
+ "name", "lookup",
+ "description", "Lookup a value.",
+ "inputSchema", "{\"type\":\"object\"}"));
+ when(adapter.invoke(
+ "python_java_utils.invoke_python_tool_instance",
+ pythonTool,
+ Map.of("query", "flink")))
+ .thenReturn(
+ Map.of(
+ "__flink_agents_tool_result__", "response",
+ "success", false,
+ "error", "retry with a narrower query",
+ "execution_time_ms", 7L,
+ "tool_name", "lookup"));
+ PythonMCPTool tool = new PythonMCPTool(adapter, pythonTool,
"search-server");
+
+ ToolResponse response = tool.call(new ToolParameters(Map.of("query",
"flink")));
+
+ assertThat(response.isError()).isTrue();
+ assertThat(response.getError()).isEqualTo("retry with a narrower
query");
+ assertThat(response.getExecutionTimeMs()).isEqualTo(7L);
+ assertThat(response.getToolName()).isEqualTo("lookup");
+ verify(adapter)
+ .invoke(
+ "python_java_utils.invoke_python_tool_instance",
+ pythonTool,
+ Map.of("query", "flink"));
+ }
+}
diff --git
a/plan/src/test/java/org/apache/flink/agents/plan/tools/FunctionToolSetPythonAdapterTest.java
b/plan/src/test/java/org/apache/flink/agents/plan/tools/FunctionToolSetPythonAdapterTest.java
index 88ab75375..876095c5f 100644
---
a/plan/src/test/java/org/apache/flink/agents/plan/tools/FunctionToolSetPythonAdapterTest.java
+++
b/plan/src/test/java/org/apache/flink/agents/plan/tools/FunctionToolSetPythonAdapterTest.java
@@ -20,6 +20,8 @@ package org.apache.flink.agents.plan.tools;
import org.apache.flink.agents.api.resource.python.PythonResourceAdapter;
import org.apache.flink.agents.api.tools.ToolMetadata;
import org.apache.flink.agents.api.tools.ToolParameterInjection;
+import org.apache.flink.agents.api.tools.ToolParameters;
+import org.apache.flink.agents.api.tools.ToolResponse;
import org.apache.flink.agents.plan.JavaFunction;
import org.apache.flink.agents.plan.PythonFunction;
import org.junit.jupiter.api.Test;
@@ -37,6 +39,9 @@ import static org.mockito.Mockito.when;
class FunctionToolSetPythonAdapterTest {
+ private static final ToolMetadata PYTHON_TOOL_METADATA =
+ new ToolMetadata("notify", "Send a notification.",
"{\"properties\":{}}");
+
@Test
void replacesPlaceholderMetadataForPythonFunction() {
ToolMetadata placeholder = new ToolMetadata("notify", "", "{}");
@@ -111,6 +116,80 @@ class FunctionToolSetPythonAdapterTest {
.getPythonToolMetadata(Mockito.anyString(),
Mockito.anyString(), anyList());
}
+ @Test
+ void preservesExplicitPythonToolFailure() {
+ PythonFunction function = new PythonFunction("pkg.mod", "notify");
+ FunctionTool tool = new FunctionTool(PYTHON_TOOL_METADATA, function);
+ PythonResourceAdapter adapter = pythonAdapter();
+ when(adapter.invokePythonTool(eq("pkg.mod"), eq("notify"),
eq(Map.of("id", "1"))))
+ .thenReturn(
+ Map.of(
+ "__flink_agents_tool_result__", "response",
+ "success", false,
+ "error", "recipient not found",
+ "execution_time_ms", 7L,
+ "tool_name", "notify"));
+ tool.setPythonResourceAdapter(adapter);
+
+ ToolResponse response = tool.call(new ToolParameters(Map.of("id",
"1")));
+
+ assertThat(response.isError()).isTrue();
+ assertThat(response.getError()).isEqualTo("recipient not found");
+ assertThat(response.getExecutionTimeMs()).isEqualTo(7L);
+ assertThat(response.getToolName()).isEqualTo("notify");
+ }
+
+ @Test
+ void preservesExplicitPythonToolSuccess() {
+ PythonFunction function = new PythonFunction("pkg.mod", "notify");
+ FunctionTool tool = new FunctionTool(PYTHON_TOOL_METADATA, function);
+ PythonResourceAdapter adapter = pythonAdapter();
+ when(adapter.invokePythonTool(eq("pkg.mod"), eq("notify"),
eq(Map.of("id", "1"))))
+ .thenReturn(
+ Map.of(
+ "__flink_agents_tool_result__", "response",
+ "result", "sent",
+ "success", true,
+ "execution_time_ms", 5L,
+ "tool_name", "notify"));
+ tool.setPythonResourceAdapter(adapter);
+
+ ToolResponse response = tool.call(new ToolParameters(Map.of("id",
"1")));
+
+ assertThat(response.isSuccess()).isTrue();
+ assertThat(response.getResult()).isEqualTo("sent");
+ assertThat(response.getExecutionTimeMs()).isEqualTo(5L);
+ assertThat(response.getToolName()).isEqualTo("notify");
+ }
+
+ @Test
+ void unwrapsRawPythonToolResultEnvelope() {
+ PythonFunction function = new PythonFunction("pkg.mod", "notify");
+ FunctionTool tool = new FunctionTool(PYTHON_TOOL_METADATA, function);
+ PythonResourceAdapter adapter = pythonAdapter();
+ Map<String, Object> rawResult =
+ Map.of("__flink_agents_tool_result__", "response", "value",
"raw");
+ when(adapter.invokePythonTool(eq("pkg.mod"), eq("notify"),
eq(Map.of("id", "1"))))
+ .thenReturn(Map.of("__flink_agents_tool_result__", "raw",
"result", rawResult));
+ tool.setPythonResourceAdapter(adapter);
+
+ ToolResponse response = tool.call(new ToolParameters(Map.of("id",
"1")));
+
+ assertThat(response.isSuccess()).isTrue();
+ assertThat(response.getResult()).isEqualTo(rawResult);
+ }
+
+ private static PythonResourceAdapter pythonAdapter() {
+ PythonResourceAdapter adapter =
Mockito.mock(PythonResourceAdapter.class);
+ when(adapter.getPythonToolMetadata(eq("pkg.mod"), eq("notify"),
anyList()))
+ .thenReturn(
+ Map.of(
+ "name", "notify",
+ "description", "Send a notification.",
+ "inputSchema", "{\"properties\":{}}"));
+ return adapter;
+ }
+
/** Helper static method to back JavaFunction in the no-op test. */
public static int stubMethod(int x) {
return x;
diff --git a/python/flink_agents/api/tools/__init__.py
b/python/flink_agents/api/tests/test_tool_response.py
similarity index 60%
copy from python/flink_agents/api/tools/__init__.py
copy to python/flink_agents/api/tests/test_tool_response.py
index 78d07b8ba..73b9228fa 100644
--- a/python/flink_agents/api/tools/__init__.py
+++ b/python/flink_agents/api/tests/test_tool_response.py
@@ -15,16 +15,22 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
-from flink_agents.api.tools.tool_execution_metadata_provider import (
- ToolExecutionMetadataProvider,
-)
-from flink_agents.api.tools.tool_parameter_injection import (
- InjectedArg,
- ToolParameterSource,
-)
+from flink_agents.api.tools import ToolResponse
-__all__ = [
- "InjectedArg",
- "ToolExecutionMetadataProvider",
- "ToolParameterSource",
-]
+
+def test_tool_response_represents_success() -> None:
+ response = ToolResponse.success({"answer": 42}, tool_name="calculator")
+
+ assert response.is_success()
+ assert not response.is_error()
+ assert response.result == {"answer": 42}
+ assert str(response) == "{'answer': 42}"
+
+
+def test_tool_response_represents_failure() -> None:
+ response = ToolResponse.error("not found", tool_name="lookup")
+
+ assert response.is_error()
+ assert not response.is_success()
+ assert response.error_message == "not found"
+ assert str(response) == "not found"
diff --git a/python/flink_agents/api/tools/__init__.py
b/python/flink_agents/api/tools/__init__.py
index 78d07b8ba..90f8b9c8f 100644
--- a/python/flink_agents/api/tools/__init__.py
+++ b/python/flink_agents/api/tools/__init__.py
@@ -22,9 +22,11 @@ from flink_agents.api.tools.tool_parameter_injection import (
InjectedArg,
ToolParameterSource,
)
+from flink_agents.api.tools.tool_response import ToolResponse
__all__ = [
"InjectedArg",
"ToolExecutionMetadataProvider",
"ToolParameterSource",
+ "ToolResponse",
]
diff --git a/python/flink_agents/api/tools/tool.py
b/python/flink_agents/api/tools/tool.py
index 5e399dd13..9fcd28de4 100644
--- a/python/flink_agents/api/tools/tool.py
+++ b/python/flink_agents/api/tools/tool.py
@@ -157,4 +157,7 @@ class Tool(SerializableResource, ABC):
"""Call the tools with arguments.
This is the method that should be implemented by the tools' developer.
+ A raw return value represents success. Return
+ :class:`flink_agents.api.tools.ToolResponse` to report an explicit
+ tool-level failure without raising an exception.
"""
diff --git a/python/flink_agents/api/tools/tool_response.py
b/python/flink_agents/api/tools/tool_response.py
new file mode 100644
index 000000000..9d403d573
--- /dev/null
+++ b/python/flink_agents/api/tools/tool_response.py
@@ -0,0 +1,76 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True)
+class ToolResponse:
+ """Represents the result and status of one Python tool execution.
+
+ Python tools may continue returning raw values, which the runtime treats as
+ successful results. Return ``ToolResponse.error(...)`` when a tool call
+ completed normally but the tool operation itself failed.
+ """
+
+ result: Any = None
+ error_message: str | None = None
+ execution_time_ms: int = 0
+ tool_name: str | None = None
+
+ @classmethod
+ def success(
+ cls,
+ result: Any,
+ execution_time_ms: int = 0,
+ tool_name: str | None = None,
+ ) -> "ToolResponse":
+ """Create a successful tool response."""
+ return cls(
+ result=result,
+ execution_time_ms=execution_time_ms,
+ tool_name=tool_name,
+ )
+
+ @classmethod
+ def error(
+ cls,
+ error: str,
+ execution_time_ms: int = 0,
+ tool_name: str | None = None,
+ ) -> "ToolResponse":
+ """Create a failed tool response."""
+ if error is None:
+ msg = "error cannot be None"
+ raise ValueError(msg)
+ return cls(
+ error_message=error,
+ execution_time_ms=execution_time_ms,
+ tool_name=tool_name,
+ )
+
+ def is_success(self) -> bool:
+ """Return whether the tool operation succeeded."""
+ return self.error_message is None
+
+ def is_error(self) -> bool:
+ """Return whether the tool operation failed."""
+ return not self.is_success()
+
+ def __str__(self) -> str:
+ return str(self.result) if self.is_success() else
str(self.error_message)
diff --git a/python/flink_agents/integrations/mcp/mcp.py
b/python/flink_agents/integrations/mcp/mcp.py
index aa80073f2..f6753760d 100644
--- a/python/flink_agents/integrations/mcp/mcp.py
+++ b/python/flink_agents/integrations/mcp/mcp.py
@@ -39,7 +39,7 @@ from typing_extensions import override
from flink_agents.api.chat_message import ChatMessage, MessageRole
from flink_agents.api.prompts.prompt import Prompt
from flink_agents.api.resource import Resource, ResourceType
-from flink_agents.api.tools import ToolExecutionMetadataProvider
+from flink_agents.api.tools import ToolExecutionMetadataProvider, ToolResponse
from flink_agents.api.tools.tool import Tool, ToolMetadata, ToolType
from flink_agents.api.trace import (
ToolExecutionMetadataKeys,
@@ -65,12 +65,20 @@ class MCPTool(Tool, ToolExecutionMetadataProvider):
def call(self, *args: Any, **kwargs: Any) -> Any:
"""Call the MCP tool with the given arguments."""
if self.mcp_server is None:
- msg = "MCP tool call requires a reference to the MCP server"
- raise ValueError(msg)
+ return ToolResponse.error(
+ "MCP tool call requires a reference to the MCP server",
+ tool_name=self.metadata.name,
+ )
- return asyncio.run(
- self.mcp_server.call_tool_async(self.metadata.name, *args,
**kwargs)
- )
+ try:
+ return asyncio.run(
+ self.mcp_server.call_tool_async(self.metadata.name, *args,
**kwargs)
+ )
+ except Exception as e:
+ return ToolResponse.error(
+ f"Error calling MCP tool '{self.metadata.name}': {e}",
+ tool_name=self.metadata.name,
+ )
@override
def get_tool_execution_metadata(
@@ -225,7 +233,11 @@ class MCPServer(Resource, ABC):
pass
async def call_tool_async(self, tool_name: str, *args: Any, **kwargs: Any)
-> Any:
- """Call a tool on the MCP server asynchronously."""
+ """Call a tool on the MCP server asynchronously.
+
+ Raises:
+ RuntimeError: If the MCP response marks the tool result as an
error.
+ """
async with self._get_session() as session:
arguments = kwargs if kwargs else (args[0] if args else {})
@@ -235,7 +247,10 @@ class MCPServer(Resource, ABC):
read_timeout_seconds=timedelta(seconds=self.timeout),
)
- content = [extract_mcp_content_item(item) for item in
result.content]
+ content = [extract_mcp_content_item(item) for item in result.content]
+ if result.isError:
+ msg = f"MCP tool '{tool_name}' returned an error: {content}"
+ raise RuntimeError(msg)
return content
diff --git a/python/flink_agents/integrations/mcp/tests/test_mcp.py
b/python/flink_agents/integrations/mcp/tests/test_mcp.py
index 1b013ebb3..39dd35b08 100644
--- a/python/flink_agents/integrations/mcp/tests/test_mcp.py
+++ b/python/flink_agents/integrations/mcp/tests/test_mcp.py
@@ -15,17 +15,25 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import asyncio
import multiprocessing
import runpy
import time
+from contextlib import asynccontextmanager
from pathlib import Path
+from typing import AsyncIterator
from urllib.parse import parse_qs, urlparse
+import anyio
+import pytest
from mcp.client.auth import OAuthClientProvider, TokenStorage
+from mcp.client.session import ClientSession
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata,
OAuthToken
+from mcp.types import CallToolResult, TextContent
from pydantic import AnyUrl
from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.tools import ToolResponse
from flink_agents.api.tools.tool import ToolMetadata
from flink_agents.api.trace import ToolExecutionMetadataKeys
from flink_agents.integrations.mcp.mcp import MCPServer, MCPTool
@@ -158,3 +166,74 @@ def test_mcp_tool_roundtrip_preserves_metadata() -> None:
assert restored.get_tool_execution_metadata({}) == {
ToolExecutionMetadataKeys.MCP_SERVER: "calculator_server"
}
+
+
+class _ProtocolErrorClientSession(ClientSession):
+ async def call_tool(self, *args: object, **kwargs: object) ->
CallToolResult:
+ return CallToolResult(
+ content=[TextContent(type="text", text="business failure")],
+ isError=True,
+ )
+
+
+class _ProtocolErrorServer(MCPServer):
+ @asynccontextmanager
+ async def _get_session(self) -> AsyncIterator[ClientSession]:
+ server_send, client_receive = anyio.create_memory_object_stream(1)
+ client_send, server_receive = anyio.create_memory_object_stream(1)
+ async with (
+ server_send,
+ client_receive,
+ client_send,
+ server_receive,
+ _ProtocolErrorClientSession(client_receive, client_send) as
session,
+ ):
+ yield session
+
+
+class _ProtocolSuccessSession:
+ async def call_tool(self, *args: object, **kwargs: object) ->
CallToolResult:
+ return CallToolResult(
+ content=[TextContent(type="text", text="result")],
+ isError=False,
+ )
+
+
+class _ProtocolSuccessServer(MCPServer):
+ @asynccontextmanager
+ async def _get_session(self) -> AsyncIterator[_ProtocolSuccessSession]:
+ yield _ProtocolSuccessSession()
+
+
+def test_mcp_protocol_error_maps_to_tool_failure() -> None:
+ server = _ProtocolErrorServer(endpoint="http://localhost/mcp";)
+
+ with pytest.raises(RuntimeError, match="business failure"):
+ asyncio.run(server.call_tool_async("lookup", query="flink"))
+
+ tool = MCPTool(
+ metadata=ToolMetadata(
+ name="lookup",
+ description="Lookup a value.",
+ args_schema={"type": "object", "properties": {}},
+ ),
+ mcp_server=server,
+ )
+ response = tool.call(query="flink")
+
+ assert isinstance(response, ToolResponse)
+ assert response.is_error()
+ assert "business failure" in response.error_message
+
+
+def test_mcp_tool_success_preserves_raw_result() -> None:
+ tool = MCPTool(
+ metadata=ToolMetadata(
+ name="lookup",
+ description="Lookup a value.",
+ args_schema={"type": "object", "properties": {}},
+ ),
+ mcp_server=_ProtocolSuccessServer(endpoint="http://localhost/mcp";),
+ )
+
+ assert tool.call(query="flink") == ["result"]
diff --git a/python/flink_agents/plan/actions/tool_call_action.py
b/python/flink_agents/plan/actions/tool_call_action.py
index c6b68b56c..4952bfd8d 100644
--- a/python/flink_agents/plan/actions/tool_call_action.py
+++ b/python/flink_agents/plan/actions/tool_call_action.py
@@ -25,7 +25,7 @@ from flink_agents.api.events.tool_event import
ToolRequestEvent, ToolResponseEve
from flink_agents.api.memory_object import MemoryObject
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import DurableCall, Outcome, RunnerContext
-from flink_agents.api.tools import ToolExecutionMetadataProvider
+from flink_agents.api.tools import ToolExecutionMetadataProvider, ToolResponse
from flink_agents.api.tools.tool_parameter_injection import (
InjectedArg,
ToolParameterSource,
@@ -247,14 +247,7 @@ async def _execute_sequentially(
*call.args,
**(call.kwargs or {}),
)
- responses[execution.id] = response
- success[execution.id] = True
- ExecutionReporters.succeeded(
- ctx,
- ExecutionEntityTypes.TOOL,
- execution.name,
- execution.entity_metadata,
- )
+ _record_tool_response(execution, response, ctx, responses,
success, error)
except Exception as e: # noqa: PERF203
_record_execution_exception(execution, e, ctx, responses, success,
error)
@@ -272,14 +265,41 @@ def _record_outcome(
execution, outcome.error, ctx, responses, success, error
)
else:
- responses[execution.id] = outcome.value
- success[execution.id] = True
- ExecutionReporters.succeeded(
+ _record_tool_response(execution, outcome.value, ctx, responses,
success, error)
+
+
+def _record_tool_response(
+ execution: _ToolCallExecution,
+ value: Any,
+ ctx: RunnerContext,
+ responses: dict,
+ success: dict,
+ error: dict,
+) -> None:
+ response = value if isinstance(value, ToolResponse) else
ToolResponse.success(value)
+ if response.is_error():
+ message = response.error_message
+ responses[execution.id] = message
+ success[execution.id] = False
+ error[execution.id] = message
+ ExecutionReporters.failed(
ctx,
ExecutionEntityTypes.TOOL,
execution.name,
execution.entity_metadata,
+ RuntimeError(message),
+ ExecutionProblemCategories.TOOL_CALL_FAILED,
)
+ return
+
+ responses[execution.id] = response.result
+ success[execution.id] = True
+ ExecutionReporters.succeeded(
+ ctx,
+ ExecutionEntityTypes.TOOL,
+ execution.name,
+ execution.entity_metadata,
+ )
def _record_execution_exception(
diff --git a/python/flink_agents/plan/function.py
b/python/flink_agents/plan/function.py
index 7988084ae..2d82e7f4f 100644
--- a/python/flink_agents/plan/function.py
+++ b/python/flink_agents/plan/function.py
@@ -24,10 +24,12 @@ from typing import Any, Callable, Dict, List, Tuple,
get_type_hints
from pydantic import BaseModel, PrivateAttr, model_serializer
+from flink_agents.api.tools import ToolResponse
from flink_agents.plan.utils import check_type_match
# Global cache for PythonFunction instances to avoid repeated creation
_PYTHON_FUNCTION_CACHE: Dict[Tuple[str, str], "PythonFunction"] = {}
+_TOOL_RESULT_MARKER = "__flink_agents_tool_result__"
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
@@ -321,11 +323,13 @@ class JavaFunction(Function):
self.parameter_types,
list(args),
)
- return self._j_resource_adapter.invokeJavaTool(
- self.qualname,
- self.method_name,
- self.parameter_types,
- kwargs,
+ return _decode_java_tool_result(
+ self._j_resource_adapter.invokeJavaTool(
+ self.qualname,
+ self.method_name,
+ self.parameter_types,
+ kwargs,
+ )
)
def check_signature(self, *args: Tuple[Any, ...]) -> None:
@@ -340,6 +344,22 @@ class JavaFunction(Function):
raise TypeError(msg)
+def _decode_java_tool_result(value: Any) -> Any:
+ if not isinstance(value, dict) or value.get(_TOOL_RESULT_MARKER) !=
"response":
+ return value
+ if value.get("success") is True:
+ return ToolResponse.success(
+ value.get("result"),
+ execution_time_ms=value.get("execution_time_ms", 0),
+ tool_name=value.get("tool_name"),
+ )
+ return ToolResponse.error(
+ value.get("error"),
+ execution_time_ms=value.get("execution_time_ms", 0),
+ tool_name=value.get("tool_name"),
+ )
+
+
def call_python_function(module: str, qualname: str, func_args: Tuple[Any,
...]) -> Any:
"""Used to call a Python function in the Pemja environment.
diff --git a/python/flink_agents/plan/tests/actions/test_tool_call_action.py
b/python/flink_agents/plan/tests/actions/test_tool_call_action.py
index cc25d2438..f119e31e7 100644
--- a/python/flink_agents/plan/tests/actions/test_tool_call_action.py
+++ b/python/flink_agents/plan/tests/actions/test_tool_call_action.py
@@ -24,7 +24,11 @@ from flink_agents.api.events.tool_event import
ToolRequestEvent, ToolResponseEve
from flink_agents.api.memory_object import MemoryObject
from flink_agents.api.resource import ResourceType
from flink_agents.api.runner_context import Outcome
-from flink_agents.api.tools import InjectedArg, ToolExecutionMetadataProvider
+from flink_agents.api.tools import (
+ InjectedArg,
+ ToolExecutionMetadataProvider,
+ ToolResponse,
+)
from flink_agents.api.tools.tool import ToolType
from flink_agents.api.trace import (
ExecutionEntityTypes,
@@ -424,6 +428,24 @@ def
test_tool_call_action_records_parallel_outcome_failure() -> None:
assert response.error["call-2"] == "boom"
+def test_tool_call_action_records_parallel_tool_response_failure() -> None:
+ config = AgentConfiguration({"tenant_id": "tenant-1"})
+ config.set(AgentExecutionOptions.TOOL_CALL_ASYNC, True)
+ config.set(AgentExecutionOptions.TOOL_CALL_PARALLELISM, 4)
+ ctx = _Context(config=config)
+ ctx.durable_execute_all_async_outcomes = [
+ Outcome.success("ok"),
+ Outcome.success(ToolResponse.error("business failure")),
+ ]
+
+ asyncio.run(process_tool_request(tool_request("call-1", "call-2"), ctx))
+
+ response = ToolResponseEvent.from_event(ctx.sent_events[0])
+ assert response.success == {"call-1": True, "call-2": False}
+ assert response.responses["call-2"] == "business failure"
+ assert response.error["call-2"] == "business failure"
+
+
def test_tool_call_action_uses_sync_when_async_disabled_multi_tool() -> None:
config = AgentConfiguration({"tenant_id": "tenant-1"})
config.set(AgentExecutionOptions.TOOL_CALL_ASYNC, False)
@@ -544,6 +566,38 @@ def test_tool_call_reports_failed() -> None:
assert args[4] == ExecutionProblemCategories.TOOL_CALL_FAILED
+def test_tool_call_reports_explicit_tool_response_failure() -> None:
+ tool = MagicMock()
+ tool.tool_type.return_value = ToolType.FUNCTION
+ tool.call = MagicMock(return_value=ToolResponse.error("business failure"))
+ ctx, sent_events = trace_context(tool)
+ request = ToolRequestEvent(model="model-a", tool_calls=[trace_tool_call()])
+
+ asyncio.run(process_tool_request(request, ctx))
+
+ response = ToolResponseEvent.from_event(sent_events[0])
+ assert response.responses["call-1"] == "business failure"
+ assert response.success["call-1"] is False
+ assert response.error["call-1"] == "business failure"
+ ctx.report_execution_failed.assert_called_once()
+ ctx.report_execution_succeeded.assert_not_called()
+
+
+def test_tool_call_preserves_empty_tool_response_error() -> None:
+ tool = MagicMock()
+ tool.tool_type.return_value = ToolType.FUNCTION
+ tool.call = MagicMock(return_value=ToolResponse.error(""))
+ ctx, sent_events = trace_context(tool)
+ request = ToolRequestEvent(model="model-a", tool_calls=[trace_tool_call()])
+
+ asyncio.run(process_tool_request(request, ctx))
+
+ response = ToolResponseEvent.from_event(sent_events[0])
+ assert response.responses["call-1"] == ""
+ assert response.success["call-1"] is False
+ assert response.error["call-1"] == ""
+
+
def test_tool_call_includes_provider_metadata() -> None:
class MetadataTool(ToolExecutionMetadataProvider):
@staticmethod
diff --git a/python/flink_agents/plan/tests/tools/test_function_tool.py
b/python/flink_agents/plan/tests/tools/test_function_tool.py
index 291de408e..dce0a841a 100644
--- a/python/flink_agents/plan/tests/tools/test_function_tool.py
+++ b/python/flink_agents/plan/tests/tools/test_function_tool.py
@@ -21,7 +21,7 @@ from unittest.mock import MagicMock
import pytest
-from flink_agents.api.tools import InjectedArg
+from flink_agents.api.tools import InjectedArg, ToolResponse
from flink_agents.plan.function import JavaFunction, PythonFunction
from flink_agents.plan.tools.function_tool import FunctionTool
@@ -120,6 +120,7 @@ def _java_func() -> JavaFunction:
parameter_types=["int", "int"],
)
+
_FAKE_JAVA_SCHEMA = json.dumps(
{
"type": "object",
@@ -218,11 +219,19 @@ def
test_java_function_tool_metadata_is_none_without_adapter() -> None:
def test_java_function_tool_call_dispatches_through_adapter() -> None:
tool = FunctionTool(func=_java_func())
adapter = _fake_adapter()
+ adapter.invokeJavaTool.return_value = {
+ "__flink_agents_tool_result__": "response",
+ "result": 1065,
+ "success": True,
+ "error": None,
+ "execution_time_ms": 7,
+ "tool_name": "add",
+ }
tool.set_java_resource_adapter(adapter)
result = tool.call(a=377, b=688)
- assert result == 1065
+ assert result == ToolResponse.success(1065, execution_time_ms=7,
tool_name="add")
adapter.invokeJavaTool.assert_called_once_with(
"com.example.Tools",
"add",
@@ -231,6 +240,35 @@ def
test_java_function_tool_call_dispatches_through_adapter() -> None:
)
+def test_java_function_tool_preserves_error_response() -> None:
+ tool = FunctionTool(func=_java_func())
+ adapter = _fake_adapter()
+ adapter.invokeJavaTool.return_value = {
+ "__flink_agents_tool_result__": "response",
+ "result": None,
+ "success": False,
+ "error": "calculation rejected",
+ "execution_time_ms": 9,
+ "tool_name": "add",
+ }
+ tool.set_java_resource_adapter(adapter)
+
+ result = tool.call(a=377, b=688)
+
+ assert result == ToolResponse.error(
+ "calculation rejected", execution_time_ms=9, tool_name="add"
+ )
+
+
+def test_java_function_tool_keeps_legacy_raw_adapter_result() -> None:
+ tool = FunctionTool(func=_java_func())
+ adapter = _fake_adapter()
+ adapter.invokeJavaTool.return_value = 1065
+ tool.set_java_resource_adapter(adapter)
+
+ assert tool.call(a=377, b=688) == 1065
+
+
def test_java_function_tool_call_without_adapter_raises() -> None:
tool = FunctionTool(func=_java_func())
with pytest.raises(RuntimeError, match="JVM resource adapter"):
diff --git a/python/flink_agents/runtime/python_java_utils.py
b/python/flink_agents/runtime/python_java_utils.py
index e1ecac32b..d13db0b79 100644
--- a/python/flink_agents/runtime/python_java_utils.py
+++ b/python/flink_agents/runtime/python_java_utils.py
@@ -191,12 +191,39 @@ def invoke_python_tool(module: str, qual_name: str,
kwargs: Dict[str, Any]) -> A
Used by the Java-side ``PythonResourceAdapter.invokePythonTool`` so a Java
host can
dispatch a Python function tool from a Java chat model without the Python
side
- needing to know about Pemja's threading model.
+ needing to know about Pemja's threading model. The return value is wrapped
in
+ an internal envelope so Java can distinguish a raw result from an explicit
+ ``ToolResponse`` without inspecting user payloads.
"""
from flink_agents.api.function import PythonFunction
descriptor = PythonFunction(module=module, qualname=qual_name)
- return descriptor.as_callable()(**kwargs)
+ result = descriptor.as_callable()(**kwargs)
+ return _encode_python_tool_result(result)
+
+
+def invoke_python_tool_instance(tool: Tool, kwargs: Dict[str, Any]) -> Any:
+ """Invoke a Python Tool instance and encode its result for the Java
bridge."""
+ return _encode_python_tool_result(tool.call(**kwargs))
+
+
+def _encode_python_tool_result(result: Any) -> Dict[str, Any]:
+ """Encode raw values and explicit ToolResponses without inspecting user
payloads."""
+ from flink_agents.api.tools import ToolResponse
+
+ if not isinstance(result, ToolResponse):
+ return {
+ "__flink_agents_tool_result__": "raw",
+ "result": result,
+ }
+ return {
+ "__flink_agents_tool_result__": "response",
+ "result": result.result,
+ "success": result.is_success(),
+ "error": result.error_message,
+ "execution_time_ms": result.execution_time_ms,
+ "tool_name": result.tool_name,
+ }
def from_java_prompt(j_prompt: Any) -> JavaPrompt:
diff --git a/python/flink_agents/runtime/skill/skill_tools.py
b/python/flink_agents/runtime/skill/skill_tools.py
index 5178ab308..f26221246 100644
--- a/python/flink_agents/runtime/skill/skill_tools.py
+++ b/python/flink_agents/runtime/skill/skill_tools.py
@@ -29,7 +29,7 @@ if TYPE_CHECKING:
from pydantic import BaseModel, Field
-from flink_agents.api.tools import ToolExecutionMetadataProvider
+from flink_agents.api.tools import ToolExecutionMetadataProvider, ToolResponse
from flink_agents.api.tools.tool import Tool, ToolMetadata, ToolType
from flink_agents.api.trace import ToolExecutionMetadataKeys
@@ -100,7 +100,7 @@ class LoadSkillTool(Tool, ToolExecutionMetadataProvider):
)
return metadata
- def call(self, *args: Any, **kwargs: Any) -> str:
+ def call(self, *args: Any, **kwargs: Any) -> str | ToolResponse:
"""Call the tool to load a skill."""
if args:
parsed_args = LoadSkillArgs(name=args[0], **kwargs)
@@ -113,7 +113,9 @@ class LoadSkillTool(Tool, ToolExecutionMetadataProvider):
manager = self._get_skill_manager()
if manager is None:
- return "Skill manager not available. No skills have been
registered."
+ return ToolResponse.error(
+ "Skill manager not available. No skills have been registered."
+ )
try:
skill = manager.get_skill(skill_name)
@@ -122,7 +124,9 @@ class LoadSkillTool(Tool, ToolExecutionMetadataProvider):
available_str = (
", ".join(available) if available else "No skills available."
)
- return f"Skill '{skill_name}' not found. Available skills:
{available_str}"
+ return ToolResponse.error(
+ f"Skill '{skill_name}' not found. Available skills:
{available_str}"
+ )
if resource_path is None or resource_path == "SKILL.md":
skill_dir = manager.get_skill_dir(skill_name)
@@ -148,7 +152,9 @@ class LoadSkillTool(Tool, ToolExecutionMetadataProvider):
content = skill.get_resource(resource_path)
if content is None:
available = sorted(skill.get_resource_paths())
- return f"Resource '{resource_path}' not found in skill
'{skill_name}', Available resources: {available}"
+ return ToolResponse.error(
+ f"Resource '{resource_path}' not found in skill
'{skill_name}', Available resources: {available}"
+ )
return content
def _get_skill_manager(self) -> SkillManager | None:
diff --git a/python/flink_agents/runtime/skill/tests/test_load_skill.py
b/python/flink_agents/runtime/skill/tests/test_load_skill.py
index be0246ce8..e428f8e70 100644
--- a/python/flink_agents/runtime/skill/tests/test_load_skill.py
+++ b/python/flink_agents/runtime/skill/tests/test_load_skill.py
@@ -24,6 +24,7 @@ import pytest
from flink_agents.api.resource_context import ResourceContext
from flink_agents.api.skills import Skills
+from flink_agents.api.tools import ToolResponse
from flink_agents.api.trace import ToolExecutionMetadataKeys
from flink_agents.runtime.skill.skill_manager import SkillManager
from flink_agents.runtime.skill.skill_tools import LoadSkillTool
@@ -84,8 +85,10 @@ class TestLoadSkillTool:
def test_load_resource_not_found(self, tool: LoadSkillTool) -> None:
"""Loading a nonexistent resource returns an error with available
list."""
result = tool.call(name="nano-banana-pro", path="nonexistent.txt")
- assert "not found" in result.lower()
- assert "scripts/generate_image.py" in result
+ assert isinstance(result, ToolResponse)
+ assert result.is_error()
+ assert "not found" in result.error_message.lower()
+ assert "scripts/generate_image.py" in result.error_message
def test_execution_metadata_describes_requested_resource(
self, tool: LoadSkillTool
@@ -97,7 +100,9 @@ class TestLoadSkillTool:
assert metadata[ToolExecutionMetadataKeys.SKILL_NAME] == "github"
assert metadata[ToolExecutionMetadataKeys.SKILL_RESOURCE_PATH] ==
"README.md"
- def test_execution_metadata_normalizes_omitted_path(self, tool:
LoadSkillTool) -> None:
+ def test_execution_metadata_normalizes_omitted_path(
+ self, tool: LoadSkillTool
+ ) -> None:
metadata = tool.get_tool_execution_metadata({"name": "github"})
assert metadata[ToolExecutionMetadataKeys.SKILL_NAME] == "github"
assert metadata[ToolExecutionMetadataKeys.SKILL_RESOURCE_PATH] ==
"SKILL.md"
@@ -114,9 +119,11 @@ class TestLoadSkillTool:
def test_skill_not_found(self, tool: LoadSkillTool) -> None:
"""A nonexistent skill returns an error listing available skills."""
result = tool.call(name="nonexistent-skill")
- assert "not found" in result.lower()
- assert "github" in result
- assert "nano-banana-pro" in result
+ assert isinstance(result, ToolResponse)
+ assert result.is_error()
+ assert "not found" in result.error_message.lower()
+ assert "github" in result.error_message
+ assert "nano-banana-pro" in result.error_message
# -- no skill manager ----------------------------------------------------
@@ -125,7 +132,9 @@ class TestLoadSkillTool:
mock_ctx = MagicMock(spec=ResourceContext)
tool = LoadSkillTool(resource_context=mock_ctx)
result = tool.call(name="github")
- assert "not available" in result
+ assert isinstance(result, ToolResponse)
+ assert result.is_error()
+ assert "not available" in result.error_message
# -- positional args -----------------------------------------------------
diff --git a/python/flink_agents/runtime/tests/test_python_java_utils.py
b/python/flink_agents/runtime/tests/test_python_java_utils.py
index d4264d46f..2555453df 100644
--- a/python/flink_agents/runtime/tests/test_python_java_utils.py
+++ b/python/flink_agents/runtime/tests/test_python_java_utils.py
@@ -25,11 +25,13 @@ from flink_agents.api.embedding_models.embedding_model
import (
EmbeddingTokenUsage,
)
from flink_agents.api.events.event import Event
-from flink_agents.api.tools import InjectedArg
+from flink_agents.api.tools import InjectedArg, ToolResponse
from flink_agents.runtime.python_java_utils import (
call_embedding_with_usage,
convert_to_python_key_text,
get_python_tool_metadata,
+ invoke_python_tool,
+ invoke_python_tool_instance,
to_python_memory_set,
wrap_to_input_event,
)
@@ -41,6 +43,23 @@ def decorated_python_tool(order_id: str, tenant_id: str,
request_id: str) -> str
return f"{tenant_id}:{request_id}:{order_id}"
+def raw_python_tool(value: str) -> dict[str, object]:
+ return {"__flink_agents_tool_result__": "response", "value": value}
+
+
+def failed_python_tool(value: str) -> ToolResponse:
+ return ToolResponse.error(value, execution_time_ms=7, tool_name="failed")
+
+
+def successful_python_tool(value: str) -> ToolResponse:
+ return ToolResponse.success(value, execution_time_ms=5,
tool_name="successful")
+
+
+class _FailedTool:
+ def call(self, value: str) -> ToolResponse:
+ return failed_python_tool(value)
+
+
def test_get_python_tool_metadata_merges_callable_injected_args() -> None:
flat = get_python_tool_metadata(
__name__, "decorated_python_tool", injected_args=["request_id"]
@@ -52,6 +71,54 @@ def
test_get_python_tool_metadata_merges_callable_injected_args() -> None:
assert injected_args == {"tenant_id": {"source": "config", "key":
"tenant.id"}}
+def test_invoke_python_tool_wraps_raw_payload_without_inspecting_it() -> None:
+ result = invoke_python_tool(__name__, "raw_python_tool", {"value": "raw"})
+
+ assert result == {
+ "__flink_agents_tool_result__": "raw",
+ "result": {"__flink_agents_tool_result__": "response", "value": "raw"},
+ }
+
+
+def test_invoke_python_tool_preserves_explicit_failure() -> None:
+ result = invoke_python_tool(__name__, "failed_python_tool", {"value":
"failed"})
+
+ assert result == {
+ "__flink_agents_tool_result__": "response",
+ "result": None,
+ "success": False,
+ "error": "failed",
+ "execution_time_ms": 7,
+ "tool_name": "failed",
+ }
+
+
+def test_invoke_python_tool_preserves_explicit_success() -> None:
+ result = invoke_python_tool(__name__, "successful_python_tool", {"value":
"ok"})
+
+ assert result == {
+ "__flink_agents_tool_result__": "response",
+ "result": "ok",
+ "success": True,
+ "error": None,
+ "execution_time_ms": 5,
+ "tool_name": "successful",
+ }
+
+
+def test_invoke_python_tool_instance_preserves_explicit_failure() -> None:
+ result = invoke_python_tool_instance(_FailedTool(), {"value": "failed"})
+
+ assert result == {
+ "__flink_agents_tool_result__": "response",
+ "result": None,
+ "success": False,
+ "error": "failed",
+ "execution_time_ms": 7,
+ "tool_name": "failed",
+ }
+
+
class _UsageAwareEmbeddingModel:
def embed_with_usage(
self, text: str, **kwargs: object
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapter.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapter.java
index 2585145b8..4a03a827b 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapter.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapter.java
@@ -40,6 +40,8 @@ import java.util.Map;
/** Adapter for managing Java resources and facilitating Python-Java
interoperability. */
public class JavaResourceAdapter {
+ private static final String TOOL_RESULT_MARKER =
"__flink_agents_tool_result__";
+
private final ResourceContext resourceContext;
private final transient PythonInterpreter interpreter;
@@ -176,10 +178,11 @@ public class JavaResourceAdapter {
* org.apache.flink.agents.api.annotation.ToolParam} name override, {@link
ToolParameters}
* numeric coercion (covers the LLM-emitted JSON Number → Java box type
mismatch that reflective
* {@code Method.invoke} otherwise rejects), required-parameter checking,
and {@link
- * ToolResponse} success / error semantics. The success result is
unwrapped for the Python
- * caller; an unsuccessful response is re-thrown as a {@link
RuntimeException}.
+ * ToolResponse} success / error semantics. The response is returned in an
internal envelope so
+ * the Python caller can distinguish an explicit Tool error from an
invocation exception without
+ * inspecting user payloads.
*/
- public Object invokeJavaTool(
+ public Map<String, Object> invokeJavaTool(
String className,
String methodName,
List<String> parameterTypes,
@@ -189,10 +192,14 @@ public class JavaResourceAdapter {
FunctionTool tool = FunctionTool.fromStaticMethod(method);
ToolResponse response =
tool.call(new ToolParameters(arguments == null ? new
HashMap<>() : arguments));
- if (!response.isSuccess()) {
- throw new RuntimeException(response.getError());
- }
- return response.getResult();
+ Map<String, Object> result = new HashMap<>();
+ result.put(TOOL_RESULT_MARKER, "response");
+ result.put("result", response.getResult());
+ result.put("success", response.isSuccess());
+ result.put("error", response.getError());
+ result.put("execution_time_ms", response.getExecutionTimeMs());
+ result.put("tool_name", response.getToolName());
+ return result;
}
/** Invoke a Java static action method with positional arguments from
Python. */
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/skill/LoadSkillTool.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/skill/LoadSkillTool.java
index abfeb306f..7891780f6 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/skill/LoadSkillTool.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/skill/LoadSkillTool.java
@@ -87,11 +87,11 @@ public class LoadSkillTool extends Tool implements
ToolExecutionMetadataProvider
try {
manager = resolveSkillManager();
} catch (Exception e) {
- return ToolResponse.success(
+ return ToolResponse.error(
"Skill manager not available. No skills have been
registered.");
}
if (manager == null) {
- return ToolResponse.success(
+ return ToolResponse.error(
"Skill manager not available. No skills have been
registered.");
}
@@ -102,7 +102,7 @@ public class LoadSkillTool extends Tool implements
ToolExecutionMetadataProvider
List<String> available = manager.getAllSkillNames();
String availableStr =
available.isEmpty() ? "No skills available." :
String.join(", ", available);
- return ToolResponse.success(
+ return ToolResponse.error(
"Skill '" + name + "' not found. Available skills: " +
availableStr);
}
@@ -140,7 +140,7 @@ public class LoadSkillTool extends Tool implements
ToolExecutionMetadataProvider
String content = skill.getResource(path);
if (content == null) {
- return ToolResponse.success(
+ return ToolResponse.error(
"Resource '"
+ path
+ "' not found in skill '"
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapterTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapterTest.java
index f1d4a82ef..dd8ddd390 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapterTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/python/utils/JavaResourceAdapterTest.java
@@ -30,7 +30,7 @@ import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
-class JavaResourceAdapterTest {
+public class JavaResourceAdapterTest {
@Test
void getJavaToolMetadataHidesInjectedArgsAndReturnsAnnotatedDeclaration()
throws Exception {
@@ -58,6 +58,52 @@ class JavaResourceAdapterTest {
assertThat(injectedArgs.get("tenant_id").get("key").asText()).isEqualTo("tenant.id");
}
+ @Test
+ void invokeJavaToolPreservesSuccessResponseForPythonCaller() throws
Exception {
+ JavaResourceAdapter adapter =
+ new JavaResourceAdapter(null, null,
Thread.currentThread().getContextClassLoader());
+
+ Map<String, Object> result =
+ adapter.invokeJavaTool(
+ JavaResourceAdapterTest.class.getName(),
+ "queryOrder",
+ List.of(
+ String.class.getName(),
+ String.class.getName(),
+ String.class.getName()),
+ Map.of(
+ "order_id", "order-1",
+ "tenant_id", "tenant-1",
+ "request_id", "request-1"));
+
+ assertThat(result)
+ .containsEntry("__flink_agents_tool_result__", "response")
+ .containsEntry("success", true)
+ .containsEntry("result", "tenant-1:request-1:order-1")
+ .containsEntry("execution_time_ms", 0L);
+ assertThat(result.get("error")).isNull();
+ }
+
+ @Test
+ void invokeJavaToolPreservesErrorResponseForPythonCaller() throws
Exception {
+ JavaResourceAdapter adapter =
+ new JavaResourceAdapter(null, null,
Thread.currentThread().getContextClassLoader());
+
+ Map<String, Object> result =
+ adapter.invokeJavaTool(
+ JavaResourceAdapterTest.class.getName(),
+ "failingTool",
+ List.of(String.class.getName()),
+ Map.of("value", "input"));
+
+ assertThat(result)
+ .containsEntry("__flink_agents_tool_result__", "response")
+ .containsEntry("success", false)
+ .containsEntry("error", "tool rejected input")
+ .containsEntry("execution_time_ms", 0L);
+ assertThat(result.get("result")).isNull();
+ }
+
@Tool(description = "Query order.")
public static String queryOrder(
@ToolParam(name = "order_id") String orderId,
@@ -70,4 +116,9 @@ class JavaResourceAdapterTest {
@ToolParam(name = "request_id") String requestId) {
return tenantId + ":" + requestId + ":" + orderId;
}
+
+ @Tool(description = "Fail a tool call.")
+ public static String failingTool(@ToolParam(name = "value") String value) {
+ throw new IllegalStateException("tool rejected " + value);
+ }
}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/skill/LoadSkillToolTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/skill/LoadSkillToolTest.java
index f131cd07e..71d6ff6e3 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/skill/LoadSkillToolTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/skill/LoadSkillToolTest.java
@@ -70,7 +70,8 @@ class LoadSkillToolTest {
void unknownSkillReturnsAvailableList() {
LoadSkillTool t = tool(contextWithSkills());
ToolResponse resp = t.call(args("does-not-exist", null));
- String out = (String) resp.getResult();
+ assertTrue(resp.isError());
+ String out = resp.getError();
assertTrue(out.contains("not found"));
assertTrue(out.contains("github"));
assertTrue(out.contains("nano-banana-pro"));
@@ -145,7 +146,8 @@ class LoadSkillToolTest {
void missingResourceReportsAvailable() {
LoadSkillTool t = tool(contextWithSkills());
ToolResponse resp = t.call(args("nano-banana-pro", "no-such.txt"));
- String out = (String) resp.getResult();
+ assertTrue(resp.isError());
+ String out = resp.getError();
assertTrue(out.contains("Resource 'no-such.txt' not found"));
assertTrue(out.contains("Available resources"));
}
@@ -156,7 +158,8 @@ class LoadSkillToolTest {
ResourceContextImpl ctx = new ResourceContextImpl((name, type) ->
null);
LoadSkillTool t = tool(ctx);
ToolResponse resp = t.call(args("anything", null));
+ assertTrue(resp.isError());
assertEquals(
- "Skill manager not available. No skills have been
registered.", resp.getResult());
+ "Skill manager not available. No skills have been
registered.", resp.getError());
}
}
