[Feature] Cross Language Actions

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

@@ -19,6 +19,7 @@
 package org.apache.flink.agents.api;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -88,14 +89,17 @@ public void setAttr(String name, Object value) {
         attributes.put(name, value);
     }
 
+    @JsonIgnore
     public boolean hasSourceTimestamp() {
         return sourceTimestamp != null;
     }
 
+    @JsonIgnore
     public Long getSourceTimestamp() {
         return sourceTimestamp;
     }
 
+    @JsonIgnore
     public void setSourceTimestamp(long timestamp) {
         this.sourceTimestamp = timestamp;
     }

api/src/main/java/org/apache/flink/agents/api/annotation/Action.java

@@ -42,6 +42,19 @@
  * @Action(listenEventTypes = {InputEvent.EVENT_TYPE, "MyCustomEvent"})
  * public void handleMultiple(Event event, RunnerContext ctx) { ... }
  * }</pre>
+ *
+ * <p>For a cross-language action, set {@link #target()} to a {@link PythonFunction} with a
+ * non-empty {@code module}. The annotated Java body is never invoked — throw {@link
+ * UnsupportedOperationException} so direct calls outside the framework fail loud:
+ *
+ * <pre>{@code
+ * @Action(
+ *     listenEventTypes = {InputEvent.EVENT_TYPE},
+ *     target = @PythonFunction(module = "my_pkg.handlers", qualname = "handle_input"))
+ * public void handleInput(Event event, RunnerContext ctx) {
+ *     throw new UnsupportedOperationException("cross-language stub");
+ * }
+ * }</pre>
  */
 @Target(ElementType.METHOD)
 @Retention(RetentionPolicy.RUNTIME)
@@ -52,4 +65,11 @@
      * @return Array of event type strings
      */
     String[] listenEventTypes();
+
+    /**
+     * Cross-language target. When {@link PythonFunction#module()} is non-empty, dispatch routes to
+     * the Python target and the annotated Java body is unused. Default (empty {@code module}) keeps
+     * the action native Java.
+     */
+    PythonFunction target() default @PythonFunction;
 }

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

@@ -0,0 +1,31 @@
+/*
+ * 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.annotation;
+
+/**
+ * Python target descriptor used inside {@link Action#target()}. Empty {@link #module()} = no
+ * cross-language target (action stays native Java). When non-empty, the Java method body is never
+ * invoked — throw {@link UnsupportedOperationException} from the stub so direct calls outside the
+ * framework fail loud.
+ */
+public @interface PythonFunction {
+    String module() default "";
+
+    String qualname() default "";
+}

api/src/main/java/org/apache/flink/agents/api/chat/messages/ChatMessage.java

@@ -19,6 +19,7 @@
 package org.apache.flink.agents.api.chat.messages;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -34,7 +35,11 @@ public class ChatMessage {
 
     private MessageRole role;
     private String content;
+
+    @JsonProperty("tool_calls")
     private List<Map<String, Object>> toolCalls;
+
+    @JsonProperty("extra_args")
     private Map<String, Object> extraArgs;
 
     /** Default constructor with SYSTEM role */
@@ -83,18 +88,22 @@ public void setContent(String content) {
         this.content = content;
     }
 
+    @JsonProperty("tool_calls")
     public List<Map<String, Object>> getToolCalls() {
         return toolCalls;
     }
 
+    @JsonProperty("tool_calls")
     public void setToolCalls(List<Map<String, Object>> toolCalls) {
         this.toolCalls = toolCalls;
     }
 
+    @JsonProperty("extra_args")
     public Map<String, Object> getExtraArgs() {
         return extraArgs;
     }
 
+    @JsonProperty("extra_args")
     public void setExtraArgs(Map<String, Object> extraArgs) {
         this.extraArgs = extraArgs != null ? extraArgs : new HashMap<>();
     }
@@ -104,6 +113,7 @@ public String getText() {
         return this.content;
     }
 
+    @JsonIgnore
     public Map<String, Object> getMetadata() {
         return this.extraArgs;
     }

api/src/main/java/org/apache/flink/agents/api/chat/messages/MessageRole.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.agents.api.chat.messages;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+
 /**
  * Enumeration of message roles in a chat conversation. Each role represents a different participant
  * type in the conversation.
@@ -42,6 +45,7 @@ public enum MessageRole {
         this.value = value;
     }
 
+    @JsonCreator
     public static MessageRole fromValue(String value) {
         for (MessageRole messageRole : MessageRole.values()) {
             if (messageRole.getValue().equals(value)) {
@@ -51,6 +55,7 @@ public static MessageRole fromValue(String value) {
         throw new IllegalArgumentException("Invalid MessageRole value: " + value);
     }
 
+    @JsonValue
     public String getValue() {
         return this.value;
     }

api/src/main/java/org/apache/flink/agents/api/event/ToolResponseEvent.java

@@ -83,6 +83,8 @@ public static ToolResponseEvent fromEvent(Event event) {
                     responses.put(entry.getKey(), (ToolResponse) v);
                 } else if (v instanceof Map) {
                     responses.put(entry.getKey(), MAPPER.convertValue(v, ToolResponse.class));
+                } else {
+                    responses.put(entry.getKey(), ToolResponse.success(v));
                 }
             }
             attrs.put("responses", responses);

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

@@ -18,6 +18,10 @@
 
 package org.apache.flink.agents.api.tools;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
 import java.util.Objects;
 
 /**
@@ -29,11 +33,20 @@ public class ToolResponse {
     private final Object result;
     private final boolean success;
     private final String error;
+
+    @JsonProperty("execution_time_ms")
     private final long executionTimeMs;
+
+    @JsonProperty("tool_name")
     private final String toolName;
 
+    @JsonCreator
     private ToolResponse(
-            Object result, boolean success, String error, long executionTimeMs, String toolName) {
+            @JsonProperty("result") Object result,
+            @JsonProperty("success") boolean success,
+            @JsonProperty("error") String error,
+            @JsonProperty("execution_time_ms") long executionTimeMs,
+            @JsonProperty("tool_name") String toolName) {
         this.result = result;
         this.success = success;
         this.error = error;
@@ -148,6 +161,7 @@ public String getToolName() {
     }
 
     /** Get the result as a string representation. */
+    @JsonIgnore
     public String getResultAsString() {
         if (result == null) {
             return null;

api/src/main/java/org/apache/flink/agents/api/vectorstores/Document.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.agents.api.vectorstores;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
 import javax.annotation.Nullable;
 
 import java.util.Arrays;
@@ -61,12 +64,13 @@ public Document(
         this(content, metadata, id, embedding, null);
     }
 
+    @JsonCreator
     public Document(
-            String content,
-            Map<String, Object> metadata,
-            String id,
-            @Nullable float[] embedding,
-            @Nullable Float score) {
+            @JsonProperty("content") String content,
+            @JsonProperty("metadata") Map<String, Object> metadata,
+            @JsonProperty("id") String id,
+            @JsonProperty("embedding") @Nullable float[] embedding,
+            @JsonProperty("score") @Nullable Float score) {
         this.content = content;
         this.metadata = metadata;
         this.id = id;