[api][plan][runtime] Cross-language Function descriptors and FunctionTool

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: wenjin272

api/src/main/java/org/apache/flink/agents/api/agents/Agent.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.agents.api.agents;
 
+import org.apache.flink.agents.api.function.Function;
+import org.apache.flink.agents.api.function.JavaFunction;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
 import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.agents.api.resource.SerializableResource;
@@ -31,7 +33,7 @@
 
 /** Base class for defining agent logic. */
 public class Agent {
-    private final Map<String, Tuple3<String[], Method, Map<String, Object>>> actions;
+    private final Map<String, Tuple3<String[], Function, Map<String, Object>>> actions;
 
     private final Map<ResourceType, Map<String, Object>> resources;
 
@@ -43,7 +45,7 @@ public Agent() {
         this.actions = new HashMap<>();
     }
 
-    public Map<String, Tuple3<String[], Method, Map<String, Object>>> getActions() {
+    public Map<String, Tuple3<String[], Function, Map<String, Object>>> getActions() {
         return actions;
     }
 
@@ -60,12 +62,7 @@ public Map<ResourceType, Map<String, Object>> getResources() {
      */
     public Agent addAction(
             String[] eventTypes, Method method, @Nullable Map<String, Object> config) {
-        String name = method.getName();
-        if (actions.containsKey(name)) {
-            throw new IllegalArgumentException(String.format("Action %s already defined.", name));
-        }
-        actions.put(name, new Tuple3<>(eventTypes, method, config));
-        return this;
+        return addAction(method.getName(), eventTypes, JavaFunction.fromMethod(method), config);
     }
 
     /**
@@ -78,6 +75,27 @@ public Agent addAction(String[] eventTypes, Method method) {
         return addAction(eventTypes, method, null);
     }
 
+    /**
+     * Add action to agent.
+     *
+     * @param name The action name. Must be unique within this agent.
+     * @param eventTypes The event type strings this action listens to.
+     * @param function The api-layer function descriptor; will be promoted to a plan-layer
+     *     executable at {@code AgentPlan} construction.
+     * @param config Optional config for this action.
+     */
+    public Agent addAction(
+            String name,
+            String[] eventTypes,
+            Function function,
+            @Nullable Map<String, Object> config) {
+        if (actions.containsKey(name)) {
+            throw new IllegalArgumentException(String.format("Action %s already defined.", name));
+        }
+        actions.put(name, new Tuple3<>(eventTypes, function, config));
+        return this;
+    }
+
     public void addResourcesIfAbsent(Map<ResourceType, Map<String, Object>> resources) {
         for (ResourceType type : resources.keySet()) {
             Map<String, Object> typedResources = resources.get(type);

api/src/main/java/org/apache/flink/agents/api/function/Function.java

@@ -0,0 +1,28 @@
+/*
+ * 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.api.function;
+
+/**
+ * Pure-data marker for user-defined function descriptors carried on the api layer.
+ *
+ * <p>Implementations describe <em>which</em> function ({@link PythonFunction}, {@link
+ * JavaFunction}) but do not execute it. The plan-layer twins ({@code
+ * org.apache.flink.agents.plan.Function} and friends) own execution; the conversion from api → plan
+ * happens during {@code AgentPlan} construction.
+ */
+public interface Function {}

api/src/main/java/org/apache/flink/agents/api/function/JavaFunction.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.api.function;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Pure-data descriptor for a Java method, identified by declaring class FQN, method name, and
+ * parameter types as strings.
+ *
+ * <p>Parameter types are strings — JVM primitive names ({@code int}, {@code long}, {@code boolean},
+ * …) or fully-qualified reference type names ({@code java.lang.String}, {@code java.util.List}). No
+ * generic parameters. The wire form keeps the descriptor pure data; class resolution is deferred to
+ * the plan-layer twin.
+ */
+public final class JavaFunction implements Function, Serializable {
+
+    private static final String FIELD_QUAL_NAME = "qualName";
+    private static final String FIELD_METHOD_NAME = "methodName";
+    private static final String FIELD_PARAMETER_TYPES = "parameterTypes";
+
+    @JsonProperty(FIELD_QUAL_NAME)
+    private final String qualName;
+
+    @JsonProperty(FIELD_METHOD_NAME)
+    private final String methodName;
+
+    @JsonProperty(FIELD_PARAMETER_TYPES)
+    private final List<String> parameterTypes;
+
+    @JsonCreator
+    public JavaFunction(
+            @JsonProperty(FIELD_QUAL_NAME) String qualName,
+            @JsonProperty(FIELD_METHOD_NAME) String methodName,
+            @JsonProperty(FIELD_PARAMETER_TYPES) List<String> parameterTypes) {
+        this.qualName = Objects.requireNonNull(qualName, "qualName");
+        this.methodName = Objects.requireNonNull(methodName, "methodName");
+        this.parameterTypes =
+                parameterTypes == null
+                        ? Collections.emptyList()
+                        : Collections.unmodifiableList(new ArrayList<>(parameterTypes));
+    }
+
+    /**
+     * Build a descriptor from a reflected {@link Method}. Each parameter type is captured via
+     * {@link Class#getName()} — the same form {@link Class#forName(String)} accepts when the api
+     * descriptor is later promoted to its plan-layer twin. For primitives this is the keyword
+     * ({@code int}, {@code long}); for reference types the fully-qualified name; for array types
+     * the JVM-internal descriptor ({@code [I}, {@code [Ljava.lang.String;}).
+     */
+    public static JavaFunction fromMethod(Method method) {
+        List<String> params = new ArrayList<>(method.getParameterCount());
+        for (Class<?> p : method.getParameterTypes()) {
+            params.add(p.getName());
+        }
+        return new JavaFunction(method.getDeclaringClass().getName(), method.getName(), params);
+    }
+
+    public String getQualName() {
+        return qualName;
+    }
+
+    public String getMethodName() {
+        return methodName;
+    }
+
+    public List<String> getParameterTypes() {
+        return parameterTypes;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof JavaFunction)) return false;
+        JavaFunction that = (JavaFunction) o;
+        return qualName.equals(that.qualName)
+                && methodName.equals(that.methodName)
+                && parameterTypes.equals(that.parameterTypes);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(qualName, methodName, parameterTypes);
+    }
+
+    @Override
+    public String toString() {
+        return "JavaFunction{" + qualName + "#" + methodName + parameterTypes + "}";
+    }
+}

api/src/main/java/org/apache/flink/agents/api/function/PythonFunction.java

@@ -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.
+ */
+package org.apache.flink.agents.api.function;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Pure-data descriptor for a Python callable, identified by its module and qualified name.
+ *
+ * <p>Carries no execution behavior — the plan-layer {@code
+ * org.apache.flink.agents.plan.PythonFunction} owns invocation via the Pemja interpreter.
+ */
+public final class PythonFunction implements Function, Serializable {
+
+    private static final String FIELD_MODULE = "module";
+    private static final String FIELD_QUAL_NAME = "qualName";
+
+    @JsonProperty(FIELD_MODULE)
+    private final String module;
+
+    @JsonProperty(FIELD_QUAL_NAME)
+    private final String qualName;
+
+    @JsonCreator
+    public PythonFunction(
+            @JsonProperty(FIELD_MODULE) String module,
+            @JsonProperty(FIELD_QUAL_NAME) String qualName) {
+        this.module = Objects.requireNonNull(module, "module");
+        this.qualName = Objects.requireNonNull(qualName, "qualName");
+    }
+
+    public String getModule() {
+        return module;
+    }
+
+    public String getQualName() {
+        return qualName;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof PythonFunction)) return false;
+        PythonFunction that = (PythonFunction) o;
+        return module.equals(that.module) && qualName.equals(that.qualName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(module, qualName);
+    }
+
+    @Override
+    public String toString() {
+        return "PythonFunction{" + module + ":" + qualName + "}";
+    }
+}

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

@@ -128,4 +128,36 @@ public interface PythonResourceAdapter {
      * @return the result of the method invocation
      */
     Object invoke(String name, Object... args);
+
+    /**
+     * Look up tool metadata for a Python function across the JVM→Python bridge.
+     *
+     * <p>The Java side asks the Python side to introspect a callable identified by {@code module} +
+     * {@code qualName}, and returns a flat {@code Map<String, String>} with keys {@code "name"},
+     * {@code "description"}, and {@code "inputSchema"} (a JSON schema string compatible with {@code
+     * ToolMetadata.inputSchema}).
+     *
+     * <p>The return shape is intentionally flat — pemja can SIGSEGV when returning arbitrary Python
+     * objects to Java on non-main-interpreter threads.
+     *
+     * @param module the Python module containing the callable
+     * @param qualName the qualified name of the callable inside the module (e.g. {@code "fn"} or
+     *     {@code "MyClass.method"})
+     * @return flat map with keys "name", "description", "inputSchema"
+     */
+    Map<String, String> getPythonToolMetadata(String module, String qualName);
+
+    /**
+     * Invoke a Python callable as a tool, passing keyword arguments. Used when a Java chat model's
+     * tool list contains a {@code plan.FunctionTool} whose function descriptor is a {@code
+     * PythonFunction}: instead of routing the invocation through Java reflection, dispatch it
+     * across the bridge so the underlying Python function runs in the Pemja interpreter.
+     *
+     * @param module the Python module containing the callable
+     * @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
+     */
+    Object invokePythonTool(String module, String qualName, Map<String, Object> kwargs);
 }

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

@@ -18,25 +18,38 @@
 
 package org.apache.flink.agents.api.tools;
 
+import org.apache.flink.agents.api.function.Function;
+import org.apache.flink.agents.api.function.JavaFunction;
 import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.agents.api.resource.SerializableResource;
 
 import java.lang.reflect.Method;
+import java.util.Objects;
 
-/** Tool keeps a method, will be converted to tool after compile. */
+/**
+ * Pure-data tool descriptor: carries an {@link Function} reference. Used at agent-construction
+ * time; compiled to the plan-layer executable {@code plan.tools.FunctionTool} when the agent
+ * becomes an {@code AgentPlan}.
+ */
 public class FunctionTool extends SerializableResource {
-    private final Method method;
 
-    public FunctionTool(Method method) {
-        this.method = method;
+    private final Function func;
+
+    public FunctionTool(Function func) {
+        this.func = Objects.requireNonNull(func, "func");
+    }
+
+    /** Convenience factory: derive a {@link JavaFunction} from a reflected method. */
+    public static FunctionTool fromMethod(Method method) {
+        return new FunctionTool(JavaFunction.fromMethod(method));
+    }
+
+    public Function getFunc() {
+        return func;
     }
 
     @Override
     public ResourceType getResourceType() {
         return ResourceType.TOOL;
     }
-
-    public Method getMethod() {
-        return method;
-    }
 }