[api][runtime] Extend agent skill loading repository.

api/src/main/java/org/apache/flink/agents/api/skills/SkillSourceSpec.java

@@ -0,0 +1,77 @@
+/*
+ * 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.skills;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Collections;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * A single entry inside {@link Skills#getSources()}. {@code scheme} identifies the source type
+ * (e.g. {@code "local"}, {@code "url"}, {@code "classpath"}); {@code params} carries the
+ * scheme-specific configuration (e.g. {@code {"path": "/data/skills"}}).
+ *
+ * <p>The {@code scheme} is normalized to lowercase. Unknown schemes deserialize successfully — the
+ * registry is the fail point at load time.
+ */
+public final class SkillSourceSpec {
+
+    private final String scheme;
+    private final Map<String, String> params;
+
+    @JsonCreator
+    public SkillSourceSpec(
+            @JsonProperty("scheme") String scheme,
+            @JsonProperty("params") Map<String, String> params) {
+        this.scheme = scheme == null ? null : scheme.toLowerCase(Locale.ROOT);
+        this.params = params == null ? Collections.emptyMap() : Map.copyOf(params);
+    }
+
+    @JsonProperty("scheme")
+    public String getScheme() {
+        return scheme;
+    }
+
+    @JsonProperty("params")
+    public Map<String, String> getParams() {
+        return params;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof SkillSourceSpec)) return false;
+        SkillSourceSpec that = (SkillSourceSpec) o;
+        return Objects.equals(scheme, that.scheme) && params.equals(that.params);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(scheme, params);
+    }
+
+    @Override
+    public String toString() {
+        return "SkillSourceSpec{scheme=" + scheme + ", params=" + params + "}";
+    }
+}

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

@@ -28,14 +28,29 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * Configuration resource describing where to load agent skills from.
  *
- * <p>Mirrors the Python {@code flink_agents.api.skills.Skills}. Use {@link
- * #fromLocalDir(String...)} to construct.
+ * <p>The single field {@code sources} holds an ordered list of {@link SkillSourceSpec} entries.
+ * Each entry has a {@code scheme} (e.g. {@code "local"}, {@code "url"}, {@code "classpath"}, {@code
+ * "package"}) and a scheme-specific {@code params} map. Use one of the factory methods to construct
+ * a {@link Skills} resource:
  *
- * <p>Multiple {@code @Skills} declarations on the same agent are merged at plan-build time.
+ * <ul>
+ *   <li>{@link #fromLocalDir(String...)} for local directories or {@code .zip} files
+ *   <li>{@link #fromUrl(String...)} for http(s) URLs pointing to a {@code .zip}
+ *   <li>{@link #fromClasspath(String...)} for resources on the classpath
+ * </ul>
+ *
+ * <p>The {@code "package"} scheme exists on the Python side only (Java has no analogous concept). A
+ * plan written by Python with {@code scheme=package} deserializes successfully on Java, but {@code
+ * SkillManager} will fail fast at load time with the registered-scheme list.
+ *
+ * <p>Multiple {@code @Skills} declarations on the same agent are merged at plan-build time;
+ * duplicate {@link SkillSourceSpec} entries (same {@code scheme} and {@code params}) are collapsed.
  */
 @JsonIgnoreProperties(
         ignoreUnknown = true,
@@ -51,31 +66,60 @@ public class Skills extends SerializableResource {
     /** Reserved name of the built-in bash tool used to execute skill scripts. */
     public static final String BASH_TOOL = "bash";
 
-    private List<String> paths;
+    private final List<SkillSourceSpec> sources;
 
     /** Required by Jackson. */
     public Skills() {
-        this.paths = Collections.emptyList();
+        this.sources = Collections.emptyList();
     }
 
     @JsonCreator
-    public Skills(@JsonProperty("paths") List<String> paths) {
-        this.paths = paths == null ? Collections.emptyList() : List.copyOf(paths);
+    public Skills(@JsonProperty("sources") List<SkillSourceSpec> sources) {
+        this.sources = sources == null ? Collections.emptyList() : List.copyOf(sources);
     }
 
     /**
-     * Create a {@link Skills} resource from one or more local filesystem directories.
+     * Create a {@link Skills} resource from one or more local paths.
      *
-     * <p>Each path points to a directory whose immediate subdirectories each contain a {@code
-     * SKILL.md} file.
+     * <p>Each path may be a directory whose immediate subdirectories each contain a {@code
+     * SKILL.md} file, or a {@code .zip} file whose top-level entries are the skill subdirectories.
      */
     public static Skills fromLocalDir(String... paths) {
-        return new Skills(Arrays.asList(paths));
+        return new Skills(
+                Arrays.stream(paths)
+                        .map(p -> new SkillSourceSpec("local", Map.of("path", p)))
+                        .collect(Collectors.toList()));
+    }
+
+    /**
+     * Create a {@link Skills} resource from one or more http(s) URLs.
+     *
+     * <p>Each URL must point to a {@code .zip} whose top level is the baseDir.
+     */
+    public static Skills fromUrl(String... urls) {
+        return new Skills(
+                Arrays.stream(urls)
+                        .map(u -> new SkillSourceSpec("url", Map.of("url", u)))
+                        .collect(Collectors.toList()));
+    }
+
+    /**
+     * Create a {@link Skills} resource from one or more classpath resource paths.
+     *
+     * <p>Each resource may be a directory (e.g. under {@code src/main/resources/skills}) or a
+     * {@code .zip} file. When packaged into a JAR, the resource is loaded via the thread context
+     * class loader and materialized to a temp directory at runtime.
+     */
+    public static Skills fromClasspath(String... resources) {
+        return new Skills(
+                Arrays.stream(resources)
+                        .map(r -> new SkillSourceSpec("classpath", Map.of("resource", r)))
+                        .collect(Collectors.toList()));
     }
 
-    @JsonProperty("paths")
-    public List<String> getPaths() {
-        return paths;
+    @JsonProperty("sources")
+    public List<SkillSourceSpec> getSources() {
+        return sources;
     }
 
     @JsonIgnore

api/src/main/java/org/apache/flink/agents/api/yaml/YamlLoader.java

@@ -29,12 +29,14 @@
 import org.apache.flink.agents.api.prompt.Prompt;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
 import org.apache.flink.agents.api.resource.ResourceType;
+import org.apache.flink.agents.api.skills.SkillSourceSpec;
 import org.apache.flink.agents.api.skills.Skills;
 import org.apache.flink.agents.api.tools.FunctionTool;
 import org.apache.flink.agents.api.yaml.spec.ActionSpec;
 import org.apache.flink.agents.api.yaml.spec.AgentActionRef;
 import org.apache.flink.agents.api.yaml.spec.AgentSpec;
 import org.apache.flink.agents.api.yaml.spec.DescriptorSpec;
+import org.apache.flink.agents.api.yaml.spec.PackageSkillSpec;
 import org.apache.flink.agents.api.yaml.spec.PromptSpec;
 import org.apache.flink.agents.api.yaml.spec.SkillsSpec;
 import org.apache.flink.agents.api.yaml.spec.ToolSpec;
@@ -167,7 +169,27 @@ public static Prompt buildPrompt(PromptSpec spec) {
 
     /** Build a {@link Skills} resource from a parsed {@link SkillsSpec}. */
     public static Skills buildSkills(SkillsSpec spec) {
-        return new Skills(new ArrayList<>(spec.getPaths()));
+        List<SkillSourceSpec> sources = new ArrayList<>();
+        for (String p : spec.getPaths()) {
+            sources.add(new SkillSourceSpec("local", Map.of("path", p)));
+        }
+        for (String u : spec.getUrls()) {
+            sources.add(new SkillSourceSpec("url", Map.of("url", u)));
+        }
+        for (String r : spec.getClasspath()) {
+            sources.add(new SkillSourceSpec("classpath", Map.of("resource", r)));
+        }
+        for (PackageSkillSpec pkg : spec.getPackageEntries()) {
+            sources.add(
+                    new SkillSourceSpec(
+                            "package",
+                            Map.of(
+                                    "package",
+                                    pkg.getPackageName(),
+                                    "resource",
+                                    pkg.getResource())));
+        }
+        return new Skills(sources);
     }
 
     /**

api/src/main/java/org/apache/flink/agents/api/yaml/spec/PackageSkillSpec.java

@@ -0,0 +1,52 @@
+/*
+ * 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.yaml.spec;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * A single {@code package} skill source entry: a Python package name plus a resource path relative
+ * to that package's root. The {@code package} scheme is Python-only at runtime — a YAML using this
+ * field deserializes on Java but fails at skill load time.
+ */
+@JsonIgnoreProperties(ignoreUnknown = false)
+public final class PackageSkillSpec {
+    private final String packageName;
+    private final String resource;
+
+    @JsonCreator
+    public PackageSkillSpec(
+            @JsonProperty(value = "package", required = true) String packageName,
+            @JsonProperty(value = "resource", required = true) String resource) {
+        this.packageName = packageName;
+        this.resource = resource;
+    }
+
+    @JsonProperty("package")
+    public String getPackageName() {
+        return packageName;
+    }
+
+    @JsonProperty("resource")
+    public String getResource() {
+        return resource;
+    }
+}

api/src/main/java/org/apache/flink/agents/api/yaml/spec/SkillsSpec.java

@@ -22,20 +22,54 @@
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
+import java.util.Collections;
 import java.util.List;
 
-/** Declarative Skills resource. */
+/**
+ * Declarative Skills resource. Each list below maps to a skill source scheme:
+ *
+ * <ul>
+ *   <li>{@code paths} — {@code local} scheme: directories or {@code .zip} files on the filesystem
+ *   <li>{@code urls} — {@code url} scheme: {@code http(s)} URLs pointing to a {@code .zip}
+ *   <li>{@code classpath} — {@code classpath} scheme: resource paths on the Java classpath
+ *   <li>{@code package} — {@code package} scheme (Python-only at runtime): {@code (package,
+ *       resource)} pairs pointing at resources inside an installed Python package
+ * </ul>
+ *
+ * <p>At least one of the four lists must be non-empty. {@code package} is exposed on Java for YAML
+ * schema parity with Python — it deserializes successfully but {@code SkillManager} on Java will
+ * fail at load time because Java does not register a {@code package} handler.
+ */
 @JsonIgnoreProperties(ignoreUnknown = false)
 public final class SkillsSpec {
     private final String name;
     private final List<String> paths;
+    private final List<String> urls;
+    private final List<String> classpath;
+    private final List<PackageSkillSpec> packageEntries;
 
     @JsonCreator
     public SkillsSpec(
             @JsonProperty(value = "name", required = true) String name,
-            @JsonProperty(value = "paths", required = true) List<String> paths) {
+            @JsonProperty("paths") List<String> paths,
+            @JsonProperty("urls") List<String> urls,
+            @JsonProperty("classpath") List<String> classpath,
+            @JsonProperty("package") List<PackageSkillSpec> packageEntries) {
         this.name = name;
-        this.paths = paths;
+        this.paths = paths == null ? Collections.emptyList() : List.copyOf(paths);
+        this.urls = urls == null ? Collections.emptyList() : List.copyOf(urls);
+        this.classpath = classpath == null ? Collections.emptyList() : List.copyOf(classpath);
+        this.packageEntries =
+                packageEntries == null ? Collections.emptyList() : List.copyOf(packageEntries);
+        if (this.paths.isEmpty()
+                && this.urls.isEmpty()
+                && this.classpath.isEmpty()
+                && this.packageEntries.isEmpty()) {
+            throw new IllegalArgumentException(
+                    "skills '"
+                            + name
+                            + "': at least one of paths/urls/classpath/package must be non-empty.");
+        }
     }
 
     public String getName() {
@@ -45,4 +79,17 @@ public String getName() {
     public List<String> getPaths() {
         return paths;
     }
+
+    public List<String> getUrls() {
+        return urls;
+    }
+
+    public List<String> getClasspath() {
+        return classpath;
+    }
+
+    @JsonProperty("package")
+    public List<PackageSkillSpec> getPackageEntries() {
+        return packageEntries;
+    }
 }