[api][test] Simplify cross-language Java action declaration and add e2e coverage

docs/content/docs/development/workflow_agent.md

@@ -543,14 +543,9 @@ from flink_agents.api.function import JavaFunction
 class MyAgent(Agent):
     @action(
         InputEvent.EVENT_TYPE,
-        target=JavaFunction(
-            qualname="com.example.MyHandlers",
-            method_name="handleInput",
-            parameter_types=[
-                "org.apache.flink.agents.api.Event",
-                "org.apache.flink.agents.api.context.RunnerContext",
-            ],
-        ),
+        # Action signatures are fixed (Event, RunnerContext), so for_action
+        # fills the Java parameter types for you — only the class and method.
+        target=JavaFunction.for_action("com.example.MyHandlers", "handleInput"),
     )
     @staticmethod
     def handle_input(event: Event, ctx: RunnerContext) -> None:

e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/YamlActionsInPythonCrossLanguageTest.java

@@ -0,0 +1,113 @@
+/*
+ * 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.resource.test;
+
+import org.apache.flink.agents.api.AgentsExecutionEnvironment;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.URL;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+import static org.apache.flink.agents.resource.test.ChatModelCrossLanguageAgent.OLLAMA_MODEL;
+import static org.apache.flink.agents.resource.test.CrossLanguageTestPreparationUtils.pullModel;
+
+/**
+ * End-to-end test for the Java YAML loader with cross-language orchestration actions: a Java host
+ * loads an agent whose user actions {@code process_input} / {@code process_chat_response} are
+ * {@code type: python}, while the Java-native built-in chat loop bridges ChatRequest→Response.
+ * Companion to {@link YamlCrossLanguageTest} (Java actions + Python tool) — here the actions cross
+ * languages and the built-ins stay native. The math path still drives a cross-language Python tool
+ * ({@code calculate_bmi}).
+ */
+public class YamlActionsInPythonCrossLanguageTest {
+
+    private static final Logger LOG =
+            LoggerFactory.getLogger(YamlActionsInPythonCrossLanguageTest.class);
+
+    private final boolean ollamaReady;
+
+    public YamlActionsInPythonCrossLanguageTest() throws IOException {
+        ollamaReady = pullModel(OLLAMA_MODEL);
+    }
+
+    @Test
+    public void testYamlPythonActionsOnJavaHost() throws Exception {
+        Assumptions.assumeTrue(ollamaReady, "Ollama Server information is not provided");
+
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.setParallelism(1);
+
+        DataStream<String> inputStream =
+                env.fromData(
+                        "Calculate BMI for someone who is 1.75 meters tall and weighs 70 kg",
+                        "Tell me a joke about cats.");
+
+        AgentsExecutionEnvironment agentsEnv =
+                AgentsExecutionEnvironment.getExecutionEnvironment(env);
+        agentsEnv.loadYaml(yamlFixture("yaml_cross_language_actions_in_python.yaml"));
+
+        DataStream<Object> outputStream =
+                agentsEnv
+                        .fromDataStream(
+                                inputStream, (KeySelector<String, String>) value -> "orderKey")
+                        .apply("yaml_actions_in_python_agent")
+                        .toDataStream();
+
+        CloseableIterator<Object> results = outputStream.collectAsync();
+        agentsEnv.execute();
+
+        List<String> responses = new ArrayList<>();
+        while (results.hasNext()) {
+            responses.add(String.valueOf(results.next()));
+        }
+        LOG.info("Python-action cross-language YAML agent responses: {}", responses);
+
+        Assertions.assertEquals(
+                2, responses.size(), "expected 2 responses, got " + responses.size());
+
+        String joined = String.join("\n", responses).toLowerCase();
+        Assertions.assertTrue(
+                joined.contains("22"), String.format("math answer missing '22': %s", responses));
+        Assertions.assertTrue(
+                joined.contains("cat"),
+                String.format("creative answer missing 'cat': %s", responses));
+    }
+
+    private static Path yamlFixture(String name) {
+        URL resource =
+                YamlActionsInPythonCrossLanguageTest.class
+                        .getClassLoader()
+                        .getResource("yaml/" + name);
+        Objects.requireNonNull(resource, "fixture not found on classpath: yaml/" + name);
+        return Paths.get(resource.getPath());
+    }
+}

e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/resources/yaml/yaml_cross_language_actions_in_python.yaml

@@ -0,0 +1,47 @@
+agents:
+  - name: yaml_actions_in_python_agent
+    description: |
+      Java-host cross-language YAML e2e agent whose orchestration actions are
+      Python. Complements yaml_cross_language_agent (Java actions + Python tool):
+      here ``process_input`` / ``process_chat_response`` are ``type: python`` and
+      dispatch to ``yaml_cross_language_actions``, while the Java-native built-in
+      chat loop bridges ChatRequest→Response. The math path still exercises a
+      cross-language Python tool (calculate_bmi).
+
+    chat_model_connections:
+      - name: ollama_connection_java
+        clazz: ollama
+        type: java
+        endpoint: http://localhost:11434
+        requestTimeout: 240
+      - name: ollama_connection_python
+        clazz: ollama
+        request_timeout: 240
+
+    chat_model_setups:
+      - name: math_chat_model
+        clazz: ollama
+        type: java
+        connection: ollama_connection_java
+        model: qwen3:1.7b
+        tools: [calculate_bmi]
+        extract_reasoning: true
+      - name: creative_chat_model
+        clazz: ollama
+        connection: ollama_connection_python
+        model: qwen3:1.7b
+        extract_reasoning: true
+
+    tools:
+      - name: calculate_bmi
+        function: flink_agents.e2e_tests.e2e_tests_resource_cross_language.yaml_cross_language_actions:calculate_bmi
+
+    actions:
+      - name: process_input
+        type: python
+        function: flink_agents.e2e_tests.e2e_tests_resource_cross_language.yaml_cross_language_actions:process_input
+        listen_to: [input]
+      - name: process_chat_response
+        type: python
+        function: flink_agents.e2e_tests.e2e_tests_resource_cross_language.yaml_cross_language_actions:process_chat_response
+        listen_to: [chat_response]

python/flink_agents/api/function.py

@@ -25,10 +25,18 @@
 import importlib
 import inspect
 from abc import ABC
-from typing import Any, Callable, List
+from typing import Any, Callable, List, Tuple
 
 from pydantic import BaseModel, model_serializer
 
+#: Java parameter types of an action method. Action signatures are fixed
+#: ``(Event, RunnerContext)``, so callers never have to spell them out. A tuple
+#: keeps the shared constant immutable; callers copy it into their own list.
+ACTION_PARAMETER_TYPES: Tuple[str, ...] = (
+    "org.apache.flink.agents.api.Event",
+    "org.apache.flink.agents.api.context.RunnerContext",
+)
+
 
 class Function(BaseModel, ABC):
     """Marker base class for function descriptors. Pure data — has no
@@ -99,6 +107,20 @@ class JavaFunction(Function):
     method_name: str
     parameter_types: List[str]
 
+    @classmethod
+    def for_action(cls, qualname: str, method_name: str) -> "JavaFunction":
+        """Build a descriptor for a Java action, filling the fixed signature.
+
+        Actions always take ``(Event, RunnerContext)``, so ``parameter_types``
+        is implied — mirrors the YAML API, which omits it for ``type: java``
+        actions. Tools must still pass ``parameter_types`` to the constructor.
+        """
+        return cls(
+            qualname=qualname,
+            method_name=method_name,
+            parameter_types=list(ACTION_PARAMETER_TYPES),
+        )
+
     @model_serializer
     def __serialize(self) -> dict[str, Any]:
         return {

python/flink_agents/api/tests/test_agent_add_action.py

@@ -30,14 +30,16 @@ def _dummy_action(event: Event, ctx: RunnerContext) -> None:
 
 
 def _make_java_function() -> JavaFunction:
-    return JavaFunction(
-        qualname="com.example.Handlers",
-        method_name="handle",
-        parameter_types=[
-            "org.apache.flink.agents.api.Event",
-            "org.apache.flink.agents.api.context.RunnerContext",
-        ],
-    )
+    return JavaFunction.for_action("com.example.Handlers", "handle")
+
+
+def test_java_function_for_action_fills_fixed_action_signature() -> None:
+    """``for_action`` omits parameter_types boilerplate; YAML already does this."""
+    jf = JavaFunction.for_action("com.example.Handlers", "handle")
+    assert jf.parameter_types == [
+        "org.apache.flink.agents.api.Event",
+        "org.apache.flink.agents.api.context.RunnerContext",
+    ]
 
 
 # ── Descriptor pass-through ─────────────────────────────────────────────

python/flink_agents/api/tests/test_decorators.py

@@ -94,14 +94,7 @@ def bad_handler(event: Event, ctx: RunnerContext) -> None:
 
 
 def _java_target() -> JavaFunction:
-    return JavaFunction(
-        qualname="com.example.Handlers",
-        method_name="handle",
-        parameter_types=[
-            "org.apache.flink.agents.api.Event",
-            "org.apache.flink.agents.api.context.RunnerContext",
-        ],
-    )
+    return JavaFunction.for_action("com.example.Handlers", "handle")
 
 
 def test_action_decorator_with_cross_language_target() -> None:

python/flink_agents/api/yaml/loader.py

@@ -29,7 +29,12 @@
 
 from flink_agents.api.agents.agent import Agent
 from flink_agents.api.chat_message import ChatMessage, MessageRole
-from flink_agents.api.function import Function, JavaFunction, PythonFunction
+from flink_agents.api.function import (
+    ACTION_PARAMETER_TYPES,
+    Function,
+    JavaFunction,
+    PythonFunction,
+)
 from flink_agents.api.prompts.prompt import Prompt
 from flink_agents.api.resource import ResourceDescriptor, ResourceType
 from flink_agents.api.skills import Skills, SkillSourceSpec
@@ -50,13 +55,6 @@
     YamlAgentsDocument,
 )
 
-# Default Java parameter types for an action. Action methods in
-# flink-agents always have signature (Event, RunnerContext).
-_JAVA_ACTION_PARAMETER_TYPES: list[str] = [
-    "org.apache.flink.agents.api.Event",
-    "org.apache.flink.agents.api.context.RunnerContext",
-]
-
 _DESCRIPTOR_TYPES: Dict[str, ResourceType] = {
     "chat_model_connections": ResourceType.CHAT_MODEL_CONNECTION,
     "chat_model_setups": ResourceType.CHAT_MODEL,
@@ -159,7 +157,7 @@ def _add_descriptors_to_agent(
 
 
 def _resolve_action_function(action: ActionSpec) -> Function:
-    parameter_types = _JAVA_ACTION_PARAMETER_TYPES if action.type == "java" else None
+    parameter_types = ACTION_PARAMETER_TYPES if action.type == "java" else None
     return resolve_function(
         name=action.name,
         function=action.function,

python/flink_agents/e2e_tests/e2e_tests_resource_cross_language/python_agent_with_java_action.py

@@ -25,10 +25,6 @@
 
 JAVA_HANDLER_QUALNAME = "org.apache.flink.agents.resource.test.JavaActionHandler"
 JAVA_HANDLER_METHOD = "multiplyByTwo"
-JAVA_HANDLER_PARAMETER_TYPES = [
-    "org.apache.flink.agents.api.Event",
-    "org.apache.flink.agents.api.context.RunnerContext",
-]
 
 
 class PythonAgentWithJavaActionAgent(Agent):
@@ -40,11 +36,7 @@ def __init__(self) -> None:
         self.add_action(
             name="multiply_by_two",
             events=[InputEvent.EVENT_TYPE],
-            func=JavaFunction(
-                qualname=JAVA_HANDLER_QUALNAME,
-                method_name=JAVA_HANDLER_METHOD,
-                parameter_types=JAVA_HANDLER_PARAMETER_TYPES,
-            ),
+            func=JavaFunction.for_action(JAVA_HANDLER_QUALNAME, JAVA_HANDLER_METHOD),
         )