[API][Tool] Align Java and Python Tool outcome contracts

api/src/main/java/org/apache/flink/agents/api/resource/python/PythonResourceAdapter.java

@@ -175,7 +175,8 @@ Map<String, String> getPythonToolMetadata(
      * @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);
 }

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.

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 static void process(Event event, RunnerContext ctx) throws Exception {
 
             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);

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
 
 
+@mcp.tool()
+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")

integrations/mcp/src/main/java/org/apache/flink/agents/integrations/mcp/MCPServer.java

@@ -443,26 +443,37 @@ public ToolMetadata getToolMetadata(String name) {
      * @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;
     }
 
     /**

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 @@ void testListPromptsReturnsEmptyWhenNotSupported() {
         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");
+    }
 }

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 ToolResponse call(ToolParameters parameters) {
             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);
         }

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() {}
+}

plan/src/main/java/org/apache/flink/agents/plan/tools/FunctionTool.java

@@ -36,6 +36,7 @@
 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 @@
 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 @@ private ToolResponse callPython(PythonFunction pf, ToolParameters parameters) {
         }
         Object result =
                 pythonResourceAdapter.invokePythonTool(pf.getModule(), pf.getQualName(), kwargs);
-        return ToolResponse.success(result);
+        return PythonToolResultConverter.fromBridgeResult(result);
     }
 
     public Function getFunction() {

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"));
+    }
+}