feat(workflow): Support MCP and other toolset tools as workflow nodes

contributing/samples/workflows/mcp_toolset_node/README.md

@@ -0,0 +1,75 @@
+# ADK Workflow MCP Toolset Node Sample
+
+## Overview
+
+This sample demonstrates calling a tool from an **MCP server** as a step in an
+**ADK Workflow**, using `ToolsetNode`.
+
+A `BaseTool` can be dropped straight into a workflow's `edges`. An MCP tool
+cannot, because it only exists after `await McpToolset.get_tools()` lists it over
+a live connection, and `Workflow(...)` is constructed synchronously.
+`ToolsetNode` closes that gap: it holds the toolset and the tool's name, and
+resolves the tool when the node runs, on the runner's event loop.
+
+The workflow reads a file through the MCP filesystem server:
+
+1. `build_args` turns the user's message into the tool's arguments.
+1. `ToolsetNode` resolves `read_file` from the toolset and calls it.
+1. `summarize` formats the MCP response.
+
+Resolution goes through `BaseToolset.get_tools_with_prefix()`, which caches per
+invocation, so several `ToolsetNode`s sharing one toolset list the server's tools
+only once per run.
+
+## Prerequisites
+
+`npx` must be on your `PATH`; the sample launches
+`@modelcontextprotocol/server-filesystem` as a stdio subprocess scoped to this
+directory. Install the MCP extra with `pip install "google-adk[mcp]"`.
+
+## Sample Inputs
+
+- `README.md`
+
+- `agent.py`
+
+## Graph
+
+```mermaid
+graph TD
+    START --> build_args
+    build_args --> read_file
+    read_file --> summarize
+```
+
+## How To
+
+1. **Declare the toolset normally.** No `await` is needed at module scope; the
+   server is contacted only while the workflow runs.
+
+   ```python
+   filesystem_toolset = McpToolset(
+       connection_params=StdioConnectionParams(...),
+       tool_filter=['read_file', 'list_directory'],
+   )
+   ```
+
+1. **Name the tool you want as a node.** The node's input is the tool's argument
+   dict (or a JSON object string, or `None` for no arguments), and the tool's
+   response becomes the node's output.
+
+   ```python
+   ToolsetNode(toolset=filesystem_toolset, tool_name='read_file')
+   ```
+
+1. **Name the node explicitly when the tool's name is not a Python
+   identifier.** Node names must be identifiers, so a tool named `read-file`
+   becomes a node named `read_file` by default. Pass `name=` to choose your own.
+
+   ```python
+   ToolsetNode(toolset=toolset, tool_name='read-file', name='reader')
+   ```
+
+The `Runner` closes the toolset when it shuts down, including when the toolset
+is referenced only by a `ToolsetNode` inside a workflow, so the MCP subprocess
+does not outlive the run.

contributing/samples/workflows/mcp_toolset_node/agent.py

@@ -0,0 +1,73 @@
+# 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.
+
+
+import os
+
+from google.adk import Workflow
+from google.adk.tools.mcp_tool import StdioConnectionParams
+from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+from google.adk.workflow import ToolsetNode
+from mcp import StdioServerParameters
+
+_allowed_path = os.path.dirname(os.path.abspath(__file__))
+
+# The MCP server is only contacted while the workflow runs, so the toolset can
+# be declared here even though listing its tools requires a live connection.
+filesystem_toolset = McpToolset(
+    connection_params=StdioConnectionParams(
+        server_params=StdioServerParameters(
+            command='npx',
+            args=[
+                '-y',
+                '@modelcontextprotocol/server-filesystem',
+                _allowed_path,
+            ],
+        ),
+        timeout=15,
+    ),
+    tool_filter=['read_file', 'list_directory'],
+)
+
+
+def build_args(node_input: str):
+  """Turns the user's message into arguments for the MCP tool."""
+  filename = node_input.strip() or 'README.md'
+  return {'path': os.path.join(_allowed_path, filename)}
+
+
+def summarize(node_input: dict):
+  """Formats the MCP tool's response for display."""
+  if node_input.get('isError'):
+    return f'The MCP server reported an error: {node_input}'
+  texts = [
+      part.get('text', '')
+      for part in node_input.get('content', [])
+      if part.get('type') == 'text'
+  ]
+  body = '\n'.join(texts)
+  return f'Read {len(body)} characters from the MCP server:\n\n{body}'
+
+
+root_agent = Workflow(
+    name='mcp_toolset_node_sample',
+    edges=[
+        (
+            'START',
+            build_args,
+            ToolsetNode(toolset=filesystem_toolset, tool_name='read_file'),
+            summarize,
+        ),
+    ],
+)

docs/guides/README.md

@@ -25,6 +25,7 @@ This directory contains specific developer guides for the ADK Python implementat
 * [Workflow Graphs](workflow/graph/index.md) - Understanding nodes, edges, and graph structures in workflows.
 * [Function Nodes](workflow/function_node/index.md) - Wrapping Python functions and generators as workflow nodes.
 * [JoinNode](workflow/join_node/index.md) - Synchronizing parallel execution paths in workflows.
+* [ToolsetNode](workflow/toolset_node/index.md) - Running a tool from a toolset, such as an MCP server, as a workflow node.
 * [RetryConfig](workflow/retry_config/index.md) - Configuring retry policies for resilient workflow nodes.
 * [ParallelWorker](workflow/parallel_worker/index.md) - Processing lists of items concurrently in workflows.
 * [Dynamic Nodes](workflow/dynamic_nodes/index.md) - Scheduling and executing nodes dynamically at runtime.

docs/guides/workflow/toolset_node/index.md

@@ -0,0 +1,98 @@
+# ToolsetNode
+
+`ToolsetNode` is a built-in workflow node that runs a named tool from a toolset, resolving the tool when the node runs rather than when the graph is built.
+
+## Introduction
+
+A `BaseTool` can be placed directly into a workflow's `edges`, and the framework wraps it as a node. That works because the tool object already exists when the graph is constructed.
+
+Tools served by a toolset are different. `McpToolset`, for example, lists its tools over a live connection, so an individual tool only exists after `await McpToolset.get_tools()`. `Workflow(name=..., edges=[...])` is constructed synchronously, leaving nowhere to await that call. Listing the tools eagerly at import time does not help either: the MCP session is bound to the event loop that created it, so a session opened under `asyncio.run(...)` is unusable by the time the runner executes the graph.
+
+`ToolsetNode` closes that gap. It holds the toolset and the name of the tool you want, and resolves the tool while the node runs, on the runner's event loop and inside the live invocation.
+
+Key features:
+- **Lazy resolution**: The toolset is only contacted while the workflow runs.
+- **Cached per invocation**: Resolution goes through `BaseToolset.get_tools_with_prefix()`, so several `ToolsetNode`s sharing one toolset list its tools only once per run.
+- **Toolset-agnostic**: Works with any `BaseToolset`, not only `McpToolset`.
+
+## Get started
+
+The following example reads a file through the MCP filesystem server as one step of a workflow.
+
+```python
+import os
+
+from google.adk import Workflow
+from google.adk.tools.mcp_tool import StdioConnectionParams
+from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
+from google.adk.workflow import ToolsetNode
+from mcp import StdioServerParameters
+
+filesystem_toolset = McpToolset(
+    connection_params=StdioConnectionParams(
+        server_params=StdioServerParameters(
+            command="npx",
+            args=["-y", "@modelcontextprotocol/server-filesystem", os.getcwd()],
+        ),
+        timeout=15,
+    ),
+    tool_filter=["read_file", "list_directory"],
+)
+
+
+def build_args(node_input: str) -> dict:
+  return {"path": node_input.strip()}
+
+
+def summarize(node_input: dict) -> str:
+  texts = [
+      part.get("text", "")
+      for part in node_input.get("content", [])
+      if part.get("type") == "text"
+  ]
+  return "\n".join(texts)
+
+
+root_agent = Workflow(
+    name="root_agent",
+    edges=[(
+        "START",
+        build_args,
+        ToolsetNode(toolset=filesystem_toolset, tool_name="read_file"),
+        summarize,
+    )],
+)
+```
+
+## How it works
+
+1.  **Resolution**: When the node runs, it builds a `ReadonlyContext` from the invocation and calls `toolset.get_tools_with_prefix(readonly_context)`, then selects the tool whose name equals `tool_name`. If no tool matches, it raises a `ValueError` listing the names that were available.
+2.  **Argument coercion**: The node input becomes the tool's arguments. A dict is used as-is, a JSON object string is parsed, and `None` or an empty string means no arguments. Any other input raises a `TypeError`.
+3.  **Execution**: The resolved tool is called via `run_async`, and its response becomes the node's output. State the tool writes to its context is propagated to downstream nodes.
+
+## Configuration options
+
+| Option | Description |
+|---|---|
+| `toolset` | The `BaseToolset` to resolve the tool from. Required. |
+| `tool_name` | The name of the tool to run. Matched against the names the toolset reports, so it includes the toolset's `tool_name_prefix` if one is set. Required. |
+| `name` | The node's name. Defaults to `tool_name` with any character that is not valid in a Python identifier replaced by an underscore. |
+| `description` | A human-readable description of what the node does. |
+| `retry_config` | Configuration for retrying the node on failure. See [RetryConfig](../retry_config/index.md). |
+| `timeout` | Maximum time in seconds for the node to complete. |
+
+### Node names
+
+Node names must be valid Python identifiers, but tool names are not constrained that way; MCP servers commonly use dashes. A tool named `read-file` therefore becomes a node named `read_file` by default. Pass `name=` to choose your own, which you will need to do if two tools would otherwise sanitize to the same node name.
+
+```python
+ToolsetNode(toolset=toolset, tool_name="read-file", name="reader")
+```
+
+## Lifecycle
+
+A `Runner` closes the toolsets it finds on the agent it runs, including toolsets referenced only by a `ToolsetNode` inside a workflow, so an MCP server subprocess does not outlive the run. If you drive a workflow without a `Runner`, call `await toolset.close()` yourself.
+
+## Limitations
+
+A tool that needs to interrupt the run cannot do so through a `ToolsetNode`. If the toolset is configured with an `auth_scheme`, or its tools are configured with `require_confirmation`, the tool's request for credentials or confirmation is not surfaced to the client, and the node emits the tool's placeholder response instead. Use a [FunctionNode](../function_node/index.md) with `auth_config` when a workflow step needs user authentication.

src/google/adk/runners.py

@@ -65,6 +65,7 @@
 if TYPE_CHECKING:
   from .apps.app import App
   from .apps.app import ResumabilityConfig
+  from .workflow._base_node import BaseNode
 
 logger = logging.getLogger('google_adk.' + __name__)
 
@@ -2229,15 +2230,25 @@ async def _handle_new_message(
           state_delta=state_delta,
       )
 
-  def _collect_toolset(self, agent: BaseAgent) -> set[BaseToolset]:
+  def _collect_toolset(self, agent: BaseAgent | BaseNode) -> set[BaseToolset]:
+    from .workflow._toolset_node import ToolsetNode
+
     toolsets = set()
+    if isinstance(agent, ToolsetNode):
+      toolsets.add(agent.toolset)
     if hasattr(agent, 'tools'):
       for tool_union in agent.tools:
         if isinstance(tool_union, BaseToolset):
           toolsets.add(tool_union)
     if hasattr(agent, 'sub_agents'):
       for sub_agent in agent.sub_agents:
         toolsets.update(self._collect_toolset(sub_agent))
+    # A Workflow holds its nodes in a graph rather than in sub_agents, so its
+    # toolsets are only reachable this way.
+    graph = getattr(agent, 'graph', None)
+    if graph is not None:
+      for graph_node in graph.nodes:
+        toolsets.update(self._collect_toolset(graph_node))
     return toolsets
 
   async def _cleanup_toolsets(self, toolsets_to_close: set[BaseToolset]):

src/google/adk/workflow/__init__.py

@@ -29,6 +29,7 @@
   from ._node import Node
   from ._node import node
   from ._retry_config import RetryConfig
+  from ._toolset_node import ToolsetNode
   from ._workflow import Workflow
 
 _LAZY_MEMBERS: dict[str, str] = {
@@ -41,6 +42,7 @@
     'NodeTimeoutError': '._errors',
     'RetryConfig': '._retry_config',
     'START': '._base_node',
+    'ToolsetNode': '._toolset_node',
     'Workflow': '._workflow',
     'node': '._node',
 }
@@ -54,6 +56,7 @@
     'NodeTimeoutError',
     'RetryConfig',
     'START',
+    'ToolsetNode',
     'Workflow',
     'node',
 ]