[api][plan][python] Expose sub-agents to chat models as callable tools

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

@@ -27,23 +27,32 @@
 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.subagent.SubagentSetup;
 import org.apache.flink.agents.api.tools.Tool;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 public abstract class BaseChatModelSetup extends Resource {
+
+    private static final Logger LOG = LoggerFactory.getLogger(BaseChatModelSetup.class);
+
     protected final String connectionName;
     protected String model;
     protected Object prompt;
     protected List<String> toolNames;
+    protected final List<String> subagentNames;
     @Nullable protected List<String> skills;
     @Nullable protected String skillDiscoveryPrompt;
     protected List<String> allowedCommands;
@@ -59,6 +68,9 @@ public BaseChatModelSetup(ResourceDescriptor descriptor, ResourceContext resourc
         this.model = descriptor.getArgument("model");
         this.prompt = descriptor.getArgument("prompt");
         this.toolNames = descriptor.getArgument("tools");
+        List<String> declaredSubagents = descriptor.getArgument("subagents");
+        this.subagentNames =
+                declaredSubagents == null ? new ArrayList<>() : new ArrayList<>(declaredSubagents);
         this.skills = descriptor.getArgument("skills");
         List<String> declaredCommands = descriptor.getArgument("allowed_commands");
         this.allowedCommands =
@@ -93,7 +105,7 @@ public void open() throws Exception {
         }
         if (this.skills != null) {
             this.skillDiscoveryPrompt =
-                    this.resourceContext.generateAvailableSkillsPrompt(this.skills);
+                    nullIfEmpty(this.resourceContext.generateAvailableSkillsPrompt(this.skills));
             List<String> mutable =
                     this.toolNames == null ? new ArrayList<>() : new ArrayList<>(this.toolNames);
             if (!mutable.contains(Skills.LOAD_SKILL_TOOL)) {
@@ -104,11 +116,54 @@ public void open() throws Exception {
             }
             this.toolNames = mutable;
         }
+        // Rebuilt from scratch: open() may run again on the same instance, and the callables must
+        // not accumulate.
+        this.tools.clear();
+        Set<String> callableNames = new LinkedHashSet<>();
         if (this.toolNames != null) {
             for (String name : this.toolNames) {
+                Preconditions.checkState(
+                        callableNames.add(name), "Duplicate callable name: %s", name);
                 this.tools.add((Tool) this.resourceContext.getResource(name, ResourceType.TOOL));
             }
         }
+        for (String name : this.subagentNames) {
+            // Tools are forbidden to carry the reserved prefix at registration, so a prefixed
+            // callable name can only come from this loop and a clash with a tool is impossible.
+            // Checked before the schema below, because a name declared twice is a mistake in the
+            // declaration whether or not it ends up registered.
+            Preconditions.checkState(
+                    callableNames.add(SubagentSetup.CALLABLE_NAME_PREFIX + name),
+                    "Duplicate callable name: %s",
+                    SubagentSetup.CALLABLE_NAME_PREFIX + name);
+            Resource resource = this.resourceContext.getResource(name, ResourceType.AGENT);
+            // A sub-agent owned by the other language resolves to a bridge handle here, which
+            // carries no schema to declare, so it is rejected instead of silently dropped.
+            Preconditions.checkState(
+                    resource instanceof SubagentSetup,
+                    "Sub-agent %s must resolve to a SubagentSetup, but was %s",
+                    name,
+                    resource.getClass().getName());
+            SubagentSetup setup = (SubagentSetup) resource;
+            String inputSchema = setup.getInputSchema();
+            if (inputSchema == null) {
+                // Unlike a bridge handle this is a sub-agent the caller could have described, so
+                // it is dropped with a warning rather than failing the job: the rest of the
+                // callables stay usable.
+                LOG.warn(
+                        "Sub-agent {} declares neither an input schema nor an input type, so there"
+                                + " are no arguments for the model to build a call from and it is"
+                                + " not offered as a callable.",
+                        name);
+                continue;
+            }
+            this.tools.add(new SubagentTool(name, setup.getDescription(), inputSchema));
+        }
+    }
+
+    @Nullable
+    private static String nullIfEmpty(@Nullable String value) {
+        return value == null || value.isEmpty() ? null : value;
     }
 
     public abstract Map<String, Object> getParameters();
@@ -171,10 +226,11 @@ public ChatMessage chat(
             messages = promptMessages;
         }
 
-        if (this.skillDiscoveryPrompt != null && !this.skillDiscoveryPrompt.isEmpty()) {
-            int idx = ChatMessage.findFirstSystemMessage(messages);
+        if (this.skillDiscoveryPrompt != null) {
+            // Right after the first system message, or at the head when there is none.
+            int idx = ChatMessage.findFirstSystemMessage(messages) + 1;
             List<ChatMessage> mutated = new ArrayList<>(messages);
-            mutated.add(idx + 1, new ChatMessage(MessageRole.SYSTEM, this.skillDiscoveryPrompt));
+            mutated.add(idx, new ChatMessage(MessageRole.SYSTEM, this.skillDiscoveryPrompt));
             messages = mutated;
         }
 
@@ -210,6 +266,16 @@ public List<String> getToolNames() {
         return toolNames;
     }
 
+    /** Names of the {@code AGENT} resources this setup declares as delegable. */
+    public List<String> getSubagentNames() {
+        return subagentNames;
+    }
+
+    @VisibleForTesting
+    public List<Tool> getTools() {
+        return tools;
+    }
+
     @Nullable
     public List<String> getSkills() {
         return skills;

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

@@ -0,0 +1,75 @@
+/*
+ * 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.chat.model;
+
+import org.apache.flink.agents.api.subagent.SubagentSetup;
+import org.apache.flink.agents.api.tools.Tool;
+import org.apache.flink.agents.api.tools.ToolMetadata;
+import org.apache.flink.agents.api.tools.ToolParameters;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.tools.ToolType;
+
+/**
+ * Presents an {@code AGENT} resource to a chat model as a callable, so that the model can delegate
+ * a task by issuing a function call. The callable name carries the reserved {@link
+ * SubagentSetup#CALLABLE_NAME_PREFIX}, which is how the executing side tells a delegation apart
+ * from a plain tool call.
+ *
+ * <p>Metadata only: it carries the schema the model needs to build the call, and nothing else. The
+ * call itself is dispatched by resolving the {@code AGENT} resource at execution time, so {@link
+ * #call} is never reached.
+ */
+class SubagentTool extends Tool {
+
+    SubagentTool(String agentName, String description, String inputSchema) {
+        super(
+                new ToolMetadata(
+                        SubagentSetup.CALLABLE_NAME_PREFIX + agentName,
+                        effectiveDescription(agentName, description),
+                        inputSchema));
+    }
+
+    /**
+     * Falls back to a generic delegation description, so an undescribed sub-agent stays usable.
+     * Every description ends with the sub-agent marker, which replaces a separate listing message:
+     * the model learns that the callable is a delegation from the description alone.
+     */
+    private static String effectiveDescription(String agentName, String description) {
+        String effective =
+                description == null || description.isBlank()
+                        ? "Delegate a standalone task to sub-agent " + agentName
+                        : description;
+        return effective + " This is subagent.";
+    }
+
+    /**
+     * Sub-agents are declared to the model as plain functions: the model builds the call the same
+     * way it builds a tool call, and only the executing side tells them apart.
+     */
+    @Override
+    public ToolType getToolType() {
+        return ToolType.FUNCTION;
+    }
+
+    @Override
+    public ToolResponse call(ToolParameters parameters) {
+        throw new UnsupportedOperationException(
+                "SubagentTool is metadata-only; resolve the AGENT resource at execution time.");
+    }
+}

api/src/main/java/org/apache/flink/agents/api/subagent/InputSchemas.java

@@ -0,0 +1,89 @@
+/*
+ * 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.subagent;
+
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import javax.annotation.Nullable;
+
+/**
+ * Renders the input type a sub-agent declares as the JSON Schema a chat model is told about, so
+ * that a sub-agent which types its arguments does not also have to spell out their schema.
+ *
+ * <p>Rendering goes through the same Jackson generator {@code ReActAgent} renders a POJO output
+ * schema with, which keeps the two type-to-schema paths in this module on one implementation and
+ * adds no dependency.
+ */
+final class InputSchemas {
+
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
+    private InputSchemas() {}
+
+    /**
+     * The schema of {@code type}, or {@code null} when the type states no shape a model could build
+     * a call from. That is {@link Object}, the type a sub-agent declares when it declares none, and
+     * any type that does not render as a JSON object, because the parameters of a callable must be
+     * one.
+     *
+     * @throws IllegalArgumentException if rendering the type fails, which is a declaration mistake
+     *     worth failing on rather than dropping silently.
+     */
+    @Nullable
+    static String fromType(@Nullable Class<?> type) {
+        if (type == null || type == Object.class) {
+            return null;
+        }
+        JsonNode schema = render(type);
+        return "object".equals(schema.path("type").asText()) ? schema.toString() : null;
+    }
+
+    private static JsonNode render(Class<?> type) {
+        try {
+            return MAPPER.generateJsonSchema(type).getSchemaNode();
+        } catch (JsonMappingException | IllegalArgumentException e) {
+            // Both are reachable: a class whose getters disagree on a property name fails the
+            // mapping, and one the generator has no JSON-object serializer for is refused with an
+            // IllegalArgumentException naming no remedy.
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Sub-agent input type %s cannot be rendered as a JSON Schema, so it"
+                                    + " cannot be declared to a chat model. Declare an input schema"
+                                    + " explicitly, or use an input type whose fields are all"
+                                    + " JSON-Schema-renderable. Rendering it reported: %s",
+                            type.getName(), e.getMessage()),
+                    e);
+        } catch (StackOverflowError e) {
+            // The generator carries no cycle guard, so a class that reaches itself through its own
+            // members recurses until the stack is gone. A separate clause rather than another type
+            // on the union above because the error carries no message to quote, so this case has to
+            // name the cause itself.
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Sub-agent input type %s is self-referential, so rendering it as a"
+                                    + " JSON Schema does not terminate and it cannot be declared to"
+                                    + " a chat model. Declare an input schema explicitly, or use an"
+                                    + " input type that does not refer back to itself.",
+                            type.getName()),
+                    e);
+        }
+    }
+}