feat: add TypeReference ports and a build-time port check to FunctionNode

core/src/main/java/com/google/adk/agents/InvocationContext.java

@@ -314,9 +314,10 @@ public String userId() {
   /**
    * Returns the {@link Scheduler} on which this invocation runs the work ADK hands to worker
    * threads: the sub-agents of a {@code ParallelAgent}, tool calls in {@code
-   * ToolExecutionMode.PARALLEL_SUBSCRIBE} mode, and the hop that starts the live send loop. An
-   * agent-level setting ({@code ParallelAgent.Builder#scheduler}, {@code
-   * LlmAgent.Builder#executor}) takes precedence over it. Defaults to {@link Schedulers#io()}.
+   * ToolExecutionMode.PARALLEL_SUBSCRIBE} mode, the plain bodies of workflow {@code FunctionNode}s,
+   * and the hop that starts the live send loop. An agent-level setting ({@code
+   * ParallelAgent.Builder#scheduler}, {@code LlmAgent.Builder#executor}) takes precedence over it.
+   * Defaults to {@link Schedulers#io()}.
    */
   public Scheduler scheduler() {
     return scheduler != null ? scheduler : Schedulers.io();

core/src/main/java/com/google/adk/runner/Runner.java

@@ -184,10 +184,11 @@ public Builder plugins(Plugin... plugins) {
     /**
      * Sets the {@link Scheduler} for the work ADK hands off to worker threads in this runner's
      * invocations: the sub-agents of a {@code ParallelAgent}, tool calls in {@link
-     * RunConfig.ToolExecutionMode#PARALLEL_SUBSCRIBE} mode, and the hop that starts the live send
-     * loop. The runs that an {@code AgentTool} nests inherit it. An agent-level {@code
-     * ParallelAgent.Builder#scheduler} or {@code LlmAgent.Builder#executor} takes precedence.
-     * {@code null} (the default) selects {@code Schedulers.io()}.
+     * RunConfig.ToolExecutionMode#PARALLEL_SUBSCRIBE} mode, the plain bodies of workflow {@code
+     * FunctionNode}s, and the hop that starts the live send loop. The runs that an {@code
+     * AgentTool} nests inherit it. An agent-level {@code ParallelAgent.Builder#scheduler} or {@code
+     * LlmAgent.Builder#executor} takes precedence. {@code null} (the default) selects {@code
+     * Schedulers.io()}.
      *
      * <p>Only these hand-offs are affected; work that completes asynchronously elsewhere (model
      * clients, session services, MCP, plugins) stays on its own threads. {@code

core/src/main/java/com/google/adk/workflow/FunctionNode.java

@@ -0,0 +1,196 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed 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 com.google.adk.workflow;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.adk.agents.Context;
+import com.google.adk.annotations.Experimental;
+import com.google.adk.events.Event;
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import com.google.genai.types.Schema;
+import io.reactivex.rxjava3.core.Flowable;
+import java.util.Optional;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * A workflow node that runs a Java function on each input, with a type per port. Jackson reads each
+ * input into the input type and writes each output in JSON form, so records and POJOs work as
+ * ports, and a successor reads the same value whether it comes live or from a stored session. Build
+ * one with {@link FunctionNodes#of} or {@link FunctionNodes#builder}.
+ */
+@Experimental
+public final class FunctionNode<I, O> extends BaseNode {
+
+  private static final PortCodec<Object> ANY =
+      PortCodec.output(PortCodec.javaType(Object.class), "an event output");
+
+  private final PortCodec<I> inputCodec;
+  private final PortCodec<O> outputCodec;
+  private final NodeBody<I, O> body;
+
+  private FunctionNode(Builder<I, O> builder) {
+    super(
+        checkNotNull(builder.name),
+        builder.description,
+        builder.rerunOnResume,
+        builder.waitForOutput,
+        builder.config,
+        builder.inputSchema,
+        builder.outputSchema,
+        builder.stateSchema);
+    this.inputCodec = builder.inputCodec;
+    this.outputCodec = builder.outputCodec;
+    this.body = checkNotNull(builder.body);
+  }
+
+  /** Returns a builder initialized with this node's properties. */
+  public Builder<I, O> toBuilder() {
+    return new Builder<>(inputCodec, outputCodec)
+        .name(name())
+        .description(description())
+        .rerunOnResume(rerunOnResume())
+        .waitForOutput(waitForOutput())
+        .config(config())
+        .inputSchema(inputSchema().orElse(null))
+        .outputSchema(outputSchema().orElse(null))
+        .stateSchema(stateSchema().orElse(null))
+        .body(body);
+  }
+
+  @Override
+  public Flowable<?> runNode(Context context, @Nullable Object nodeInput) {
+    return Flowable.defer(
+            () -> {
+              I input = inputCodec.decode(nodeInput, "input of node '" + name() + "'");
+              return Flowable.fromOptional(Optional.ofNullable(output(body.apply(context, input))));
+            })
+        .subscribeOn(context.invocationContext().scheduler());
+  }
+
+  /** Returns the event or JSON value to send for {@code result}, or {@code null} for none. */
+  private @Nullable Object output(@Nullable O result) {
+    String what = "output of node '" + name() + "'";
+    if (result instanceof Event event) {
+      // A returned event's output is sent in JSON form too, as a stored session hands it back.
+      return event.output().isEmpty()
+          ? event
+          : event.toBuilder().output(ANY.encode(event.output().get(), what)).build();
+    }
+    return outputCodec.encode(result, what);
+  }
+
+  /**
+   * Fluent builder for {@link FunctionNode}. Any property left unset falls back to the default of
+   * the matching {@link Node} property; name and body are required.
+   */
+  public static final class Builder<I, O> {
+    private final PortCodec<I> inputCodec;
+    private final PortCodec<O> outputCodec;
+    private @Nullable String name;
+    private String description = "";
+    private boolean rerunOnResume;
+    private boolean waitForOutput;
+    private NodeConfig config = NodeConfig.builder().build();
+    private @Nullable Schema inputSchema;
+    private @Nullable Schema outputSchema;
+    private @Nullable Schema stateSchema;
+    private @Nullable NodeBody<I, O> body;
+
+    Builder(PortCodec<I> inputCodec, PortCodec<O> outputCodec) {
+      this.inputCodec = inputCodec;
+      this.outputCodec = outputCodec;
+    }
+
+    /** Sets the node's name, unique within its graph. Required. */
+    @CanIgnoreReturnValue
+    public Builder<I, O> name(String name) {
+      this.name = checkNotNull(name);
+      return this;
+    }
+
+    /** Sets what the node does, for humans and for a model that may call it. */
+    @CanIgnoreReturnValue
+    public Builder<I, O> description(String description) {
+      this.description = checkNotNull(description);
+      return this;
+    }
+
+    /** Sets whether a resume runs the node again rather than using the resuming answer. */
+    @CanIgnoreReturnValue
+    public Builder<I, O> rerunOnResume(boolean rerunOnResume) {
+      this.rerunOnResume = rerunOnResume;
+      return this;
+    }
+
+    /** Sets whether the node stays re-triggerable until it produces an output or a route. */
+    @CanIgnoreReturnValue
+    public Builder<I, O> waitForOutput(boolean waitForOutput) {
+      this.waitForOutput = waitForOutput;
+      return this;
+    }
+
+    /** Sets the node's retry policy and execution timeout. */
+    @CanIgnoreReturnValue
+    public Builder<I, O> config(NodeConfig config) {
+      this.config = checkNotNull(config);
+      return this;
+    }
+
+    /** Sets the schema the node's input must match. */
+    @CanIgnoreReturnValue
+    public Builder<I, O> inputSchema(@Nullable Schema inputSchema) {
+      this.inputSchema = inputSchema;
+      return this;
+    }
+
+    /** Sets the schema the node's output must match. */
+    @CanIgnoreReturnValue
+    public Builder<I, O> outputSchema(@Nullable Schema outputSchema) {
+      this.outputSchema = outputSchema;
+      return this;
+    }
+
+    /**
+     * Sets the state keys the node uses, which child nodes inherit unless they declare their own.
+     */
+    @CanIgnoreReturnValue
+    public Builder<I, O> stateSchema(@Nullable Schema stateSchema) {
+      this.stateSchema = stateSchema;
+      return this;
+    }
+
+    /** Sets the function the node runs on each input. Required. */
+    @CanIgnoreReturnValue
+    public Builder<I, O> body(NodeBody<I, O> body) {
+      this.body = checkNotNull(body);
+      return this;
+    }
+
+    /**
+     * Builds the node.
+     *
+     * @throws IllegalStateException if the name or the body is not set
+     */
+    public FunctionNode<I, O> build() {
+      checkState(name != null, "FunctionNode.Builder requires name to be set.");
+      checkState(body != null, "FunctionNode.Builder for '%s' requires body to be set.", name);
+      return new FunctionNode<>(this);
+    }
+  }
+}

core/src/main/java/com/google/adk/workflow/FunctionNodes.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed 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 com.google.adk.workflow;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JavaType;
+import com.google.adk.annotations.Experimental;
+
+/**
+ * Creates {@link FunctionNode}s with a type per port, as a {@code Class} or, for a generic type
+ * such as {@code List<Order>}, a {@code TypeReference}. An {@code Object} input port takes its
+ * input as it is, a {@code Content} or {@code Event} port reads its own type, a {@code Void} port
+ * ignores its input, and Jackson reads any other type; every output is written in JSON form.
+ */
+@Experimental
+public final class FunctionNodes {
+
+  private static final String BUILDER = "a FunctionNodes.builder node";
+
+  /**
+   * Returns a node named {@code name} that runs {@code body} on each input read as {@code
+   * inputType} and writes its {@code outputType} output in JSON form.
+   *
+   * @throws WorkflowConfigurationException if Jackson cannot read and write {@code inputType} or
+   *     write {@code outputType}, or if a port type has no single concrete JSON shape; a field type
+   *     Jackson cannot handle fails on the first value instead
+   */
+  public static <I, O> FunctionNode<I, O> of(
+      String name, Class<I> inputType, Class<O> outputType, NodeBody<I, O> body) {
+    return node(name, PortCodec.javaType(inputType), PortCodec.javaType(outputType), body);
+  }
+
+  /** Like {@link #of(String, Class, Class, NodeBody)}, for generic port types. */
+  public static <I, O> FunctionNode<I, O> of(
+      String name, TypeReference<I> inputType, TypeReference<O> outputType, NodeBody<I, O> body) {
+    return node(name, PortCodec.javaType(inputType), PortCodec.javaType(outputType), body);
+  }
+
+  /** Like {@link #of(String, Class, Class, NodeBody)}, for a generic output type. */
+  public static <I, O> FunctionNode<I, O> of(
+      String name, Class<I> inputType, TypeReference<O> outputType, NodeBody<I, O> body) {
+    return node(name, PortCodec.javaType(inputType), PortCodec.javaType(outputType), body);
+  }
+
+  /** Like {@link #of(String, Class, Class, NodeBody)}, for a generic input type. */
+  public static <I, O> FunctionNode<I, O> of(
+      String name, TypeReference<I> inputType, Class<O> outputType, NodeBody<I, O> body) {
+    return node(name, PortCodec.javaType(inputType), PortCodec.javaType(outputType), body);
+  }
+
+  /**
+   * Returns a builder for a node with these port types, for properties {@link #of} does not take.
+   *
+   * @throws WorkflowConfigurationException as {@link #of} does
+   */
+  public static <I, O> FunctionNode.Builder<I, O> builder(Class<I> inputType, Class<O> outputType) {
+    return builder(PortCodec.javaType(inputType), PortCodec.javaType(outputType), BUILDER);
+  }
+
+  /** Like {@link #builder(Class, Class)}, for generic port types. */
+  public static <I, O> FunctionNode.Builder<I, O> builder(
+      TypeReference<I> inputType, TypeReference<O> outputType) {
+    return builder(PortCodec.javaType(inputType), PortCodec.javaType(outputType), BUILDER);
+  }
+
+  /** Like {@link #builder(Class, Class)}, for a generic output type. */
+  public static <I, O> FunctionNode.Builder<I, O> builder(
+      Class<I> inputType, TypeReference<O> outputType) {
+    return builder(PortCodec.javaType(inputType), PortCodec.javaType(outputType), BUILDER);
+  }
+
+  /** Like {@link #builder(Class, Class)}, for a generic input type. */
+  public static <I, O> FunctionNode.Builder<I, O> builder(
+      TypeReference<I> inputType, Class<O> outputType) {
+    return builder(PortCodec.javaType(inputType), PortCodec.javaType(outputType), BUILDER);
+  }
+
+  private static <I, O> FunctionNode<I, O> node(
+      String name, JavaType inputType, JavaType outputType, NodeBody<I, O> body) {
+    return FunctionNodes.<I, O>builder(inputType, outputType, "node '" + name + "'")
+        .name(name)
+        .body(body)
+        .build();
+  }
+
+  private static <I, O> FunctionNode.Builder<I, O> builder(
+      JavaType inputType, JavaType outputType, String node) {
+    return new FunctionNode.Builder<>(
+        PortCodec.input(inputType, "the input of " + node),
+        PortCodec.output(outputType, "the output of " + node));
+  }
+
+  private FunctionNodes() {}
+}

core/src/main/java/com/google/adk/workflow/NodeBody.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed 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 com.google.adk.workflow;
+
+import com.google.adk.agents.Context;
+import com.google.adk.annotations.Experimental;
+
+/**
+ * The body of a {@link FunctionNode}, which takes the node run's context and its single input; the
+ * input may be shared with other nodes, so the body must not mutate it. It runs on the {@code
+ * Runner}'s scheduler, so it may block, and a small fixed-size scheduler can deadlock bodies that
+ * wait for one another.
+ */
+@Experimental
+@FunctionalInterface
+public interface NodeBody<I, O> {
+
+  /**
+   * Returns {@code null} for no output, an {@code Event} to send with any output in JSON form, or
+   * the output, which is sent in its JSON form; a returned {@code Content} is the output, not a
+   * message.
+   */
+  O apply(Context context, I input) throws Exception;
+}