feat(workflow): LLM-agent-as-node, task mode, and node-as-tool (Part 6)

core/src/agents/instructions.ts

@@ -5,10 +5,36 @@
  */
 
 import {State} from '../sessions/state.js';
+import type {WorkflowInstructionScope} from './invocation_context.js';
 import {ReadonlyContext} from './readonly_context.js';
 
 const ARTIFACT_PREFIX = 'artifact.';
 
+/** Matches a `{Class.field}` workflow placeholder key (dotted identifier pair). */
+const WORKFLOW_FIELD_KEY = /^[A-Za-z_]\w*\.[A-Za-z_]\w*$/;
+
+/** Matches a `<Class.field from source_node>` workflow placeholder. */
+const SOURCE_NODE_PLACEHOLDER =
+  /<\s*[A-Za-z_]\w*\.([A-Za-z_]\w*)\s+from\s+([A-Za-z_]\w*)\s*>/g;
+
+/**
+ * Resolves `<Class.field from source_node>` placeholders against a workflow
+ * scope (predecessor outputs by node name). Synchronous; unresolved placeholders
+ * are left untouched. Mirrors Python's source-node-qualified data selection.
+ */
+function resolveSourceNodePlaceholders(
+  template: string,
+  scope: WorkflowInstructionScope,
+): string {
+  return template.replace(SOURCE_NODE_PLACEHOLDER, (raw, field, nodeName) => {
+    const out = scope.outputsByNode?.[nodeName];
+    if (out && typeof out === 'object' && field in (out as object)) {
+      return formatValue((out as Record<string, unknown>)[field], false);
+    }
+    return raw;
+  });
+}
+
 /**
  * Resolves a single key from the context (state or artifact).
  */
@@ -39,19 +65,30 @@ async function resolveKey(
   }
 
   // Step 3: Handle state variable injection.
-  if (!isValidStateName(key)) {
-    return rawMatch;
-  }
-
-  if (key in invocationContext.session.state) {
-    return formatValue(invocationContext.session.state[key], false);
+  if (isValidStateName(key)) {
+    if (key in invocationContext.session.state) {
+      return formatValue(invocationContext.session.state[key], false);
+    }
+    if (isOptional) {
+      return '';
+    }
+    throw new Error(`Context variable not found: \`${key}\`.`);
   }
 
-  if (isOptional) {
-    return '';
+  // Step 4: Workflow — resolve `{Class.field}` from the current node input.
+  const scope = invocationContext.workflowInstructionScope;
+  if (scope && WORKFLOW_FIELD_KEY.test(key)) {
+    const field = key.slice(key.indexOf('.') + 1);
+    const input = scope.input;
+    if (input && typeof input === 'object' && field in (input as object)) {
+      return formatValue((input as Record<string, unknown>)[field], false);
+    }
+    if (isOptional) {
+      return '';
+    }
   }
 
-  throw new Error(`Context variable not found: \`${key}\`.`);
+  return rawMatch;
 }
 
 /**
@@ -115,6 +152,14 @@ export async function injectSessionState(
   template: string,
   readonlyContext: ReadonlyContext,
 ): Promise<string> {
+  // Workflow: first resolve `<Class.field from source_node>` placeholders, and
+  // enable `{Class.field}` resolution below. Both are no-ops (placeholders left
+  // untouched) for ordinary agents, which have no workflow scope.
+  const scope = readonlyContext.invocationContext.workflowInstructionScope;
+  if (scope) {
+    template = resolveSourceNodePlaceholders(template, scope);
+  }
+
   const pattern = /\{+[^{}]*}+/g;
   const matches = Array.from(template.matchAll(pattern));
 
@@ -130,7 +175,10 @@ export async function injectSessionState(
     if (isOptional) {
       key = key.slice(0, -1);
     }
-    const isValid = key.startsWith(ARTIFACT_PREFIX) || isValidStateName(key);
+    const isValid =
+      key.startsWith(ARTIFACT_PREFIX) ||
+      isValidStateName(key) ||
+      (!!scope && WORKFLOW_FIELD_KEY.test(key));
     return {
       raw,
       key,

core/src/agents/invocation_context.ts

@@ -8,17 +8,32 @@ import {Content} from '@google/genai';
 
 import {SessionArtifactService} from '../artifacts/session_artifact_service.js';
 import {BaseCredentialService} from '../auth/credential_service/base_credential_service.js';
+import {Event} from '../events/event.js';
 import {BaseMemoryService} from '../memory/base_memory_service.js';
 import {PluginManager} from '../plugins/plugin_manager.js';
 import {BaseSessionService} from '../sessions/base_session_service.js';
 import {Session} from '../sessions/session.js';
+import {AsyncQueue} from '../utils/async_queue.js';
 import {randomUUID} from '../utils/env_aware_utils.js';
 
 import {ActiveStreamingTool} from './active_streaming_tool.js';
 import {BaseAgent} from './base_agent.js';
 import {RunConfig} from './run_config.js';
 import {TranscriptionEntry} from './transcription_entry.js';
 
+/**
+ * Workflow: data exposed to `{Class.field}` and `<Class.field from source_node>`
+ * instruction placeholders when an LlmAgent runs as a workflow node. Populated by
+ * `LLMAgentWrapper`; absent for ordinary (non-workflow) agent runs, in which case
+ * those placeholders are left untouched.
+ */
+export interface WorkflowInstructionScope {
+  /** The current node's input, exposing fields for `{Class.field}`. */
+  input?: unknown;
+  /** Predecessor node outputs keyed by node name, for `<Class.field from node>`. */
+  outputsByNode?: Record<string, unknown>;
+}
+
 /**
  * The parameters for creating an invocation context.
  */
@@ -38,6 +53,9 @@ export interface InvocationContextParams {
   activeStreamingTools?: Record<string, ActiveStreamingTool>;
   pluginManager: PluginManager;
   abortSignal?: AbortSignal;
+  workflowInstructionScope?: WorkflowInstructionScope;
+  /** Nesting depth of node-as-tool executions; used to bound recursion. */
+  nodeToolDepth?: number;
 }
 
 /**
@@ -185,6 +203,28 @@ export class InvocationContext {
 
   readonly abortSignal?: AbortSignal;
 
+  /**
+   * An optional channel into which a running tool can push events to be
+   * interleaved into the agent's output stream. Set by the LLM flow around tool
+   * execution so a {@link NodeTool} (running a node/workflow) can surface the
+   * node's intermediate and interrupt events. Cleared once tools finish.
+   */
+  eventQueue?: AsyncQueue<Event>;
+
+  /**
+   * Workflow: field-resolution scope for `{Class.field}` /
+   * `<Class.field from node>` instruction placeholders (set by
+   * `LLMAgentWrapper`).
+   */
+  workflowInstructionScope?: WorkflowInstructionScope;
+
+  /**
+   * Nesting depth of node-as-tool ({@link NodeTool}) executions in this
+   * invocation. Incremented each time a node runs as a tool (via a depth+1
+   * clone), so `NodeTool` can bound `node -> tool -> node` recursion.
+   */
+  readonly nodeToolDepth: number;
+
   /**
    * @param params The parameters for creating an invocation context.
    */
@@ -203,7 +243,10 @@ export class InvocationContext {
     this.activeStreamingTools = params.activeStreamingTools;
     this.pluginManager = params.pluginManager;
     this.abortSignal = params.abortSignal;
+    this.workflowInstructionScope = params.workflowInstructionScope;
+    this.nodeToolDepth = params.nodeToolDepth ?? 0;
     // Inherit the parent invocation's cost manager when one is available.
+
     // Child contexts created for sub-agents, agent transfers and loop
     // iterations (via createInvocationContext / createBranchCtxForSubAgent)
     // carry the parent context's fields over, so reusing its cost manager
@@ -236,6 +279,19 @@ export class InvocationContext {
   incrementLlmCallCount() {
     this.invocationCostManager.incrementAndEnforceLlmCallsLimit(this.runConfig);
   }
+
+  /**
+   * Returns a copy of this context with `overrides` applied. The spread carries
+   * every own field over (including the shared cost manager), so the copy keeps
+   * a single LLM-call counter for the invocation.
+   *
+   * Note: this copies own enumerable fields by value — scalar mutable fields
+   * (e.g. `endInvocation`) are decoupled from the original, while object-valued
+   * fields (`session`, …) stay shared by reference.
+   */
+  clone(overrides: Partial<InvocationContextParams> = {}): InvocationContext {
+    return new InvocationContext({...this, ...overrides});
+  }
 }
 
 export function newInvocationContextId(): string {

core/src/agents/llm_agent.ts

@@ -6,7 +6,11 @@
 
 import {GenerateContentConfig, Schema} from '@google/genai';
 import {context, trace} from '@opentelemetry/api';
+import {FinishTaskTool} from '../tools/finish_task_tool.js';
 import {FunctionTool} from '../tools/function_tool.js';
+import {AsyncQueue} from '../utils/async_queue.js';
+import {isBaseNode, type BaseNode} from '../workflow/base_node.js';
+import {NodeTool} from '../workflow/nodes/node_tool.js';
 
 import {z as z3} from 'zod/v3';
 import {z as z4} from 'zod/v4';
@@ -68,6 +72,7 @@ import {IDENTITY_LLM_REQUEST_PROCESSOR} from './processors/identity_llm_request_
 import {INSTRUCTIONS_LLM_REQUEST_PROCESSOR} from './processors/instructions_llm_request_processor.js';
 import {INTERACTIONS_REQUEST_PROCESSOR} from './processors/interactions_request_processor.js';
 import {REQUEST_CONFIRMATION_LLM_REQUEST_PROCESSOR} from './processors/request_confirmation_llm_request_processor.js';
+import {REQUEST_INPUT_LLM_REQUEST_PROCESSOR} from './processors/request_input_llm_request_processor.js';
 import {TOOL_FILTER_REQUEST_PROCESSOR} from './processors/tool_filter_request_processor.js';
 import {ReadonlyContext} from './readonly_context.js';
 import {StreamingMode} from './run_config.js';
@@ -193,7 +198,7 @@ export type AfterToolCallback =
 export type ExamplesUnion = Example[] | BaseExampleProvider;
 
 /** A union of tool types that can be provided to an agent. */
-export type ToolUnion = BaseTool | BaseToolset;
+export type ToolUnion = BaseTool | BaseToolset | BaseNode;
 
 const ADK_AGENT_NAME_LABEL_KEY = 'adk_agent_name';
 
@@ -259,6 +264,16 @@ export interface LlmAgentConfig extends BaseAgentConfig {
    */
   includeContents?: 'default' | 'none';
 
+  /**
+   * The agent's execution mode when run as a workflow node.
+   *
+   * - `single_turn` (default): the agent runs once against the node input.
+   * - `task`: the agent is given a `finish_task` tool and runs a multi-round
+   *   loop until it calls `finish_task`, whose arguments (conforming to
+   *   `outputSchema`) become the node output. Mirrors Python's `Agent(mode=...)`.
+   */
+  mode?: 'single_turn' | 'task';
+
   /** The input schema when agent is used as a tool. */
   inputSchema?: LlmAgentSchema;
 
@@ -323,6 +338,11 @@ async function convertToolUnionToTools(
   if (isBaseTool(toolUnion)) {
     return [toolUnion];
   }
+  if (isBaseNode(toolUnion)) {
+    // A node/Workflow passed as a tool is auto-wrapped as a NodeTool so the
+    // model can call it (mirrors Python's Agent(tools=[node/workflow])).
+    return [new NodeTool(toolUnion)];
+  }
   return await toolUnion.getTools(context);
 }
 
@@ -362,9 +382,11 @@ export class LlmAgent extends BaseAgent<LlmAgentConfig> {
   disallowTransferToParent: boolean;
   disallowTransferToPeers: boolean;
   includeContents: 'default' | 'none';
+  mode?: 'single_turn' | 'task';
   inputSchema?: Schema;
   outputSchema?: Schema;
   outputKey?: string;
+  private _finishTaskTool?: FinishTaskTool;
   beforeModelCallback?: BeforeModelCallback;
   afterModelCallback?: AfterModelCallback;
   beforeToolCallback?: BeforeToolCallback;
@@ -389,6 +411,7 @@ export class LlmAgent extends BaseAgent<LlmAgentConfig> {
     this.outputSchema = isZodObject(config.outputSchema)
       ? zodObjectToSchema(config.outputSchema)
       : config.outputSchema;
+    this.mode = config.mode;
     this.outputKey = config.outputKey;
     this.beforeModelCallback = config.beforeModelCallback;
     this.afterModelCallback = config.afterModelCallback;
@@ -404,6 +427,7 @@ export class LlmAgent extends BaseAgent<LlmAgentConfig> {
       IDENTITY_LLM_REQUEST_PROCESSOR,
       INSTRUCTIONS_LLM_REQUEST_PROCESSOR,
       REQUEST_CONFIRMATION_LLM_REQUEST_PROCESSOR,
+      REQUEST_INPUT_LLM_REQUEST_PROCESSOR,
       CONTENT_REQUEST_PROCESSOR,
       INTERACTIONS_REQUEST_PROCESSOR,
       CODE_EXECUTION_REQUEST_PROCESSOR,
@@ -500,6 +524,17 @@ export class LlmAgent extends BaseAgent<LlmAgentConfig> {
     throw new Error(`No model found for ${this.name}.`);
   }
 
+  /**
+   * The `finish_task` tool for this agent (task mode). Lazily created and cached
+   * so its declaration (derived from `outputSchema`) is stable across turns.
+   */
+  get finishTaskTool(): FinishTaskTool {
+    if (!this._finishTaskTool) {
+      this._finishTaskTool = new FinishTaskTool(this.outputSchema);
+    }
+    return this._finishTaskTool;
+  }
+
   /**
    * The resolved instruction field to construct instruction for this
    * agent.
@@ -788,7 +823,11 @@ export class LlmAgent extends BaseAgent<LlmAgentConfig> {
     // TODO - b/425992518: check if tool preprocessors can be simplified.
     // Run pre-processors for tools.
     const allTools = [...this.tools];
-    if (
+    if (this.mode === 'task') {
+      // Task mode: the agent completes by calling `finish_task` (whose params
+      // mirror the output schema) rather than emitting structured output.
+      allTools.push(this.finishTaskTool);
+    } else if (
       this.outputSchema &&
       allTools.length > 0 &&
       !canUseOutputSchemaWithTools(this.canonicalModel.model)
@@ -978,13 +1017,40 @@ export class LlmAgent extends BaseAgent<LlmAgentConfig> {
     // Call functions
     // TODO - b/425992518: bloated funciton input, fix.
     // Tool callback passed to get rid of cyclic dependency.
-    const functionResponseEvent = await handleFunctionCallsAsync({
-      invocationContext: invocationContext,
-      functionCallEvent: mergedEvent,
-      toolsDict: llmRequest.toolsDict,
-      beforeToolCallbacks: this.canonicalBeforeToolCallbacks,
-      afterToolCallbacks: this.canonicalAfterToolCallbacks,
-    });
+    // A NodeTool (running a node/workflow) streams the node's intermediate and
+    // interrupt events into `invocationContext.eventQueue`; drain it concurrently
+    // so those events interleave into this agent's output stream. The tool runs
+    // in a self-contained task that captures its result/error and always closes
+    // the queue, so there is a single error path (no unhandled rejection).
+    const eventQueue = new AsyncQueue<Event>();
+    invocationContext.eventQueue = eventQueue;
+    const toolTask = (async (): Promise<{
+      event: Event | null;
+      error?: unknown;
+    }> => {
+      try {
+        const event = await handleFunctionCallsAsync({
+          invocationContext: invocationContext,
+          functionCallEvent: mergedEvent,
+          toolsDict: llmRequest.toolsDict,
+          beforeToolCallbacks: this.canonicalBeforeToolCallbacks,
+          afterToolCallbacks: this.canonicalAfterToolCallbacks,
+        });
+        return {event};
+      } catch (error) {
+        return {event: null, error};
+      } finally {
+        eventQueue.close();
+      }
+    })();
+    for await (const queuedEvent of eventQueue) {
+      yield queuedEvent;
+    }
+    const {event: functionResponseEvent, error: toolError} = await toolTask;
+    invocationContext.eventQueue = undefined;
+    if (toolError) {
+      throw toolError;
+    }
 
     if (!functionResponseEvent || invocationContext.abortSignal?.aborted) {
       return;

core/src/agents/processors/basic_llm_request_processor.ts

@@ -41,8 +41,12 @@ export class BasicLlmRequestProcessor extends BaseLlmRequestProcessor {
     // Models that cannot take an output schema alongside tools get the
     // prompt-based `set_model_response` workaround instead, injected by
     // `LlmAgent.runOneStepAsync` and the instructions processor.
+    // Task-mode agents complete via the `finish_task` tool, so the JSON response
+    // mode must not be set (function calling is incompatible with a JSON
+    // response mime type).
     if (
       agent.outputSchema &&
+      agent.mode !== 'task' &&
       (!agent.tools?.length ||
         canUseOutputSchemaWithTools(agent.canonicalModel.model))
     ) {