[api][runtime] Support agent skills in java

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

@@ -160,7 +160,8 @@ public static void startAction(Event event, RunnerContext ctx) {
 
         if (schmaPrompt != null) {
             List<ChatMessage> instruct = schmaPrompt.formatMessages(MessageRole.SYSTEM, Map.of());
-            inputMessages.addAll(0, instruct);
+            int index = ChatMessage.findFirstSystemMessage(inputMessages);
+            inputMessages.addAll(index + 1, instruct);
         }
 
         Object outputSchema = ctx.getActionConfigValue("output_schema");

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

@@ -0,0 +1,39 @@
+/*
+ * 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;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Marks a static method that returns an {@link org.apache.flink.agents.api.skills.Skills} resource
+ * describing where to load agent skills from.
+ *
+ * <p>Mirrors the Python {@code @skills} decorator. Multiple {@code @Skills} methods on the same
+ * agent are merged at plan-build time.
+ *
+ * <p>Note: this annotation shares its simple name with {@link
+ * org.apache.flink.agents.api.skills.Skills} (different package). When importing both, one of them
+ * must be referenced by its fully-qualified name.
+ */
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Skills {}

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

@@ -154,4 +154,14 @@ public int hashCode() {
     public String toString() {
         return role.getValue() + ": " + content;
     }
+
+    /** Return the index of the first system message in the list, or -1 if none. */
+    public static int findFirstSystemMessage(List<ChatMessage> messages) {
+        for (int i = 0; i < messages.size(); i++) {
+            if (messages.get(i).getRole() == MessageRole.SYSTEM) {
+                return i;
+            }
+        }
+        return -1;
+    }
 }

api/src/main/java/org/apache/flink/agents/api/chat/model/BaseChatModelConnection.java

@@ -21,13 +21,13 @@
 import org.apache.flink.agents.api.chat.messages.ChatMessage;
 import org.apache.flink.agents.api.metrics.FlinkAgentsMetricGroup;
 import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
 import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.agents.api.tools.Tool;
 
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 /**
  * Abstraction of chat model connection.
@@ -37,9 +37,8 @@
  */
 public abstract class BaseChatModelConnection extends Resource {
 
-    public BaseChatModelConnection(
-            ResourceDescriptor descriptor, BiFunction<String, ResourceType, Resource> getResource) {
-        super(descriptor, getResource);
+    public BaseChatModelConnection(ResourceDescriptor descriptor, ResourceContext resourceContext) {
+        super(descriptor, resourceContext);
     }
 
     @Override

api/src/main/java/org/apache/flink/agents/api/chat/model/BaseChatModelSetup.java

@@ -22,8 +22,10 @@
 import org.apache.flink.agents.api.chat.messages.MessageRole;
 import org.apache.flink.agents.api.prompt.Prompt;
 import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
 import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.skills.Skills;
 import org.apache.flink.agents.api.tools.Tool;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.util.Preconditions;
@@ -35,24 +37,35 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 public abstract class BaseChatModelSetup extends Resource {
     protected final String connectionName;
     protected String model;
     protected Object prompt;
     protected List<String> toolNames;
+    @Nullable protected List<String> skills;
+    @Nullable protected String skillDiscoveryPrompt;
+    protected List<String> allowedCommands;
+    protected List<String> allowedScriptDirs;
 
     @Nullable protected BaseChatModelConnection connection;
     protected final List<Tool> tools = new ArrayList<>();
 
-    public BaseChatModelSetup(
-            ResourceDescriptor descriptor, BiFunction<String, ResourceType, Resource> getResource) {
-        super(descriptor, getResource);
+    public BaseChatModelSetup(ResourceDescriptor descriptor, ResourceContext resourceContext) {
+        super(descriptor, resourceContext);
         this.connectionName = descriptor.getArgument("connection");
         this.model = descriptor.getArgument("model");
         this.prompt = descriptor.getArgument("prompt");
         this.toolNames = descriptor.getArgument("tools");
+        this.skills = descriptor.getArgument("skills");
+        List<String> declaredCommands = descriptor.getArgument("allowed_commands");
+        this.allowedCommands =
+                declaredCommands == null ? new ArrayList<>() : new ArrayList<>(declaredCommands);
+        List<String> declaredScriptDirs = descriptor.getArgument("allowed_script_dirs");
+        this.allowedScriptDirs =
+                declaredScriptDirs == null
+                        ? new ArrayList<>()
+                        : new ArrayList<>(declaredScriptDirs);
     }
 
     /**
@@ -66,14 +79,28 @@ public BaseChatModelSetup(
     public void open() throws Exception {
         this.connection =
                 (BaseChatModelConnection)
-                        this.getResource.apply(
+                        this.resourceContext.getResource(
                                 this.connectionName, ResourceType.CHAT_MODEL_CONNECTION);
         if (this.prompt != null && this.prompt instanceof String) {
-            this.prompt = this.getResource.apply((String) this.prompt, ResourceType.PROMPT);
+            this.prompt =
+                    this.resourceContext.getResource((String) this.prompt, ResourceType.PROMPT);
+        }
+        if (this.skills != null) {
+            this.skillDiscoveryPrompt =
+                    this.resourceContext.generateAvailableSkillsPrompt(this.skills);
+            List<String> mutable =
+                    this.toolNames == null ? new ArrayList<>() : new ArrayList<>(this.toolNames);
+            if (!mutable.contains(Skills.LOAD_SKILL_TOOL)) {
+                mutable.add(Skills.LOAD_SKILL_TOOL);
+            }
+            if (!mutable.contains(Skills.BASH_TOOL)) {
+                mutable.add(Skills.BASH_TOOL);
+            }
+            this.toolNames = mutable;
         }
         if (this.toolNames != null) {
             for (String name : this.toolNames) {
-                this.tools.add((Tool) this.getResource.apply(name, ResourceType.TOOL));
+                this.tools.add((Tool) this.resourceContext.getResource(name, ResourceType.TOOL));
             }
         }
     }
@@ -115,6 +142,13 @@ public ChatMessage chat(List<ChatMessage> messages, Map<String, Object> paramete
             messages = promptMessages;
         }
 
+        if (this.skillDiscoveryPrompt != null && !this.skillDiscoveryPrompt.isEmpty()) {
+            int idx = ChatMessage.findFirstSystemMessage(messages);
+            List<ChatMessage> mutated = new ArrayList<>(messages);
+            mutated.add(idx + 1, new ChatMessage(MessageRole.SYSTEM, this.skillDiscoveryPrompt));
+            messages = mutated;
+        }
+
         Map<String, Object> params = this.getParameters();
         params.putAll(parameters);
         return connection.chat(messages, tools, params);
@@ -144,4 +178,22 @@ public Object getPrompt() {
     public List<String> getToolNames() {
         return toolNames;
     }
+
+    @Nullable
+    public List<String> getSkills() {
+        return skills;
+    }
+
+    @Nullable
+    public String getSkillDiscoveryPrompt() {
+        return skillDiscoveryPrompt;
+    }
+
+    public List<String> getAllowedCommands() {
+        return allowedCommands;
+    }
+
+    public List<String> getAllowedScriptDirs() {
+        return allowedScriptDirs;
+    }
 }

api/src/main/java/org/apache/flink/agents/api/chat/model/python/PythonChatModelConnection.java

@@ -19,9 +19,8 @@
 
 import org.apache.flink.agents.api.chat.messages.ChatMessage;
 import org.apache.flink.agents.api.chat.model.BaseChatModelConnection;
-import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
-import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.agents.api.resource.python.PythonResourceAdapter;
 import org.apache.flink.agents.api.resource.python.PythonResourceWrapper;
 import org.apache.flink.agents.api.tools.Tool;
@@ -31,7 +30,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 /**
  * Python-based implementation of ChatModelConnection that wraps a Python chat model object. This
@@ -56,8 +54,8 @@ public PythonChatModelConnection(
             PythonResourceAdapter adapter,
             PyObject chatModel,
             ResourceDescriptor descriptor,
-            BiFunction<String, ResourceType, Resource> getResource) {
-        super(descriptor, getResource);
+            ResourceContext resourceContext) {
+        super(descriptor, resourceContext);
         this.chatModel = chatModel;
         this.adapter = adapter;
     }

api/src/main/java/org/apache/flink/agents/api/chat/model/python/PythonChatModelSetup.java

@@ -19,9 +19,8 @@
 
 import org.apache.flink.agents.api.chat.messages.ChatMessage;
 import org.apache.flink.agents.api.chat.model.BaseChatModelSetup;
-import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
-import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.agents.api.resource.python.PythonResourceAdapter;
 import org.apache.flink.agents.api.resource.python.PythonResourceWrapper;
 import pemja.core.object.PyObject;
@@ -31,7 +30,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 import static org.apache.flink.util.Preconditions.checkState;
 
@@ -52,8 +50,8 @@ public PythonChatModelSetup(
             PythonResourceAdapter adapter,
             PyObject chatModelSetup,
             ResourceDescriptor descriptor,
-            BiFunction<String, ResourceType, Resource> getResource) {
-        super(descriptor, getResource);
+            ResourceContext resourceContext) {
+        super(descriptor, resourceContext);
         this.chatModelSetup = chatModelSetup;
         this.adapter = adapter;
     }

api/src/main/java/org/apache/flink/agents/api/embedding/model/BaseEmbeddingModelConnection.java

@@ -19,12 +19,12 @@
 package org.apache.flink.agents.api.embedding.model;
 
 import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
 import org.apache.flink.agents.api.resource.ResourceType;
 
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 /**
  * Abstraction of embedding model connection.
@@ -45,8 +45,8 @@
 public abstract class BaseEmbeddingModelConnection extends Resource {
 
     public BaseEmbeddingModelConnection(
-            ResourceDescriptor descriptor, BiFunction<String, ResourceType, Resource> getResource) {
-        super(descriptor, getResource);
+            ResourceDescriptor descriptor, ResourceContext resourceContext) {
+        super(descriptor, resourceContext);
     }
 
     @Override

api/src/main/java/org/apache/flink/agents/api/embedding/model/BaseEmbeddingModelSetup.java

@@ -19,6 +19,7 @@
 package org.apache.flink.agents.api.embedding.model;
 
 import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
 import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.annotation.VisibleForTesting;
@@ -29,7 +30,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 /**
  * Base class for embedding model setup configurations.
@@ -43,9 +43,8 @@ public abstract class BaseEmbeddingModelSetup extends Resource {
 
     @Nullable protected BaseEmbeddingModelConnection connection;
 
-    public BaseEmbeddingModelSetup(
-            ResourceDescriptor descriptor, BiFunction<String, ResourceType, Resource> getResource) {
-        super(descriptor, getResource);
+    public BaseEmbeddingModelSetup(ResourceDescriptor descriptor, ResourceContext resourceContext) {
+        super(descriptor, resourceContext);
         this.connectionName = descriptor.getArgument("connection");
         this.model = descriptor.getArgument("model");
     }
@@ -58,10 +57,11 @@ public BaseEmbeddingModelSetup(
      * resources object out of the method to be async executed and invoking it in the main thread.
      */
     @Override
-    public void open() {
+    public void open() throws Exception {
         this.connection =
                 (BaseEmbeddingModelConnection)
-                        getResource.apply(connectionName, ResourceType.EMBEDDING_MODEL_CONNECTION);
+                        resourceContext.getResource(
+                                connectionName, ResourceType.EMBEDDING_MODEL_CONNECTION);
     }
 
     public abstract Map<String, Object> getParameters();

api/src/main/java/org/apache/flink/agents/api/embedding/model/python/PythonEmbeddingModelConnection.java

@@ -19,9 +19,8 @@
 
 import org.apache.flink.agents.api.embedding.model.BaseEmbeddingModelConnection;
 import org.apache.flink.agents.api.embedding.model.EmbeddingModelUtils;
-import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
-import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.agents.api.resource.python.PythonResourceAdapter;
 import org.apache.flink.agents.api.resource.python.PythonResourceWrapper;
 import pemja.core.object.PyObject;
@@ -30,7 +29,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 import static org.apache.flink.util.Preconditions.checkState;
 
@@ -59,8 +57,8 @@ public PythonEmbeddingModelConnection(
             PythonResourceAdapter adapter,
             PyObject embeddingModel,
             ResourceDescriptor descriptor,
-            BiFunction<String, ResourceType, Resource> getResource) {
-        super(descriptor, getResource);
+            ResourceContext resourceContext) {
+        super(descriptor, resourceContext);
         this.embeddingModel = embeddingModel;
         this.adapter = adapter;
     }

api/src/main/java/org/apache/flink/agents/api/embedding/model/python/PythonEmbeddingModelSetup.java

@@ -19,9 +19,8 @@
 
 import org.apache.flink.agents.api.embedding.model.BaseEmbeddingModelSetup;
 import org.apache.flink.agents.api.embedding.model.EmbeddingModelUtils;
-import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
-import org.apache.flink.agents.api.resource.ResourceType;
 import org.apache.flink.agents.api.resource.python.PythonResourceAdapter;
 import org.apache.flink.agents.api.resource.python.PythonResourceWrapper;
 import pemja.core.object.PyObject;
@@ -31,7 +30,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 
 import static org.apache.flink.util.Preconditions.checkState;
 
@@ -59,8 +57,8 @@ public PythonEmbeddingModelSetup(
             PythonResourceAdapter adapter,
             PyObject embeddingModelSetup,
             ResourceDescriptor descriptor,
-            BiFunction<String, ResourceType, Resource> getResource) {
-        super(descriptor, getResource);
+            ResourceContext resourceContext) {
+        super(descriptor, resourceContext);
         this.embeddingModelSetup = embeddingModelSetup;
         this.adapter = adapter;
     }