Support MCP and other toolset tools as graph workflow nodes

[jsilverhand48]

🔴 Required Information

Is your feature request related to a specific problem?

A BaseTool can be dropped straight into a Workflow's edges and build_node() wraps it in a _ToolNode, and McpTool is a BaseTool. So in principle MCP tools already work as workflow nodes. In practice they don't.

An McpTool instance only exists after await McpToolset.get_tools(), which lists tools over a live connection. Workflow(name=..., edges=[...]) is constructed synchronously, so there is nowhere to await it.

The obvious workaround does not work either:

toolset = McpToolset(connection_params=StdioConnectionParams(...))
tools = asyncio.run(toolset.get_tools())          # opens a session...
wf = Workflow(name='wf', edges=[(START, node(tools[0]))])   # ...on a loop that is now closed

MCPSessionManager only reuses a pooled session when it belongs to the running event loop, so a session opened under asyncio.run(...) is already unusable by the time the runner executes the graph.

Net effect: an MCP server can be handed to LlmAgent(tools=[toolset]), but it cannot be a deterministic step in a graph workflow. There is no way to say "call this specific MCP tool at this point in the graph" without an LLM in the loop deciding to call it.

Describe the Solution You'd Like

A workflow node that names a tool in a toolset and resolves it lazily at run time, on the runner's event loop, inside the live invocation.

ToolsetNode(
    toolset: BaseToolset,      # required
    tool_name: str,            # required, matched against the names the toolset reports
    name: str | None = None,   # node name; defaults to tool_name sanitized to an identifier
    description: str = '',
    retry_config: RetryConfig | None = None,
    timeout: float | None = None,
)

• Input: a dict of tool arguments, a JSON object string, or None for no arguments and the same coercion _ToolNode already applies.
• Output: the tool's response. State the tool writes to its context propagates to downstream nodes.
• Error: if the toolset serves no tool by that name, a ValueError naming the tool and listing the names that were available.

Typed on BaseToolset rather than McpToolset, for two reasons: mcp is an optional extra and tests/unittests/test_import_loading.py asserts that importing google.adk never loads it, and the same node then serves OpenAPI and every other toolset. MCP is the motivating case, not the coupling.

Impact on your work

This is the blocker for using MCP servers in deterministic pipelines. Today the only way to reach an MCP tool is through an LlmAgent, which means paying for a model call and accepting nondeterminism for a step that is fully specified. For graph workflows whose whole appeal is deterministic orchestration, that
defeats the purpose.

No fixed timeline.

Willingness to contribute

Yes. I have a working implementation with unit tests, a runnable sample, and a docs page, verified end to end against real MCP servers over both stdio and streamable HTTP. Happy to open the PR once there is agreement on the approach, particularly the two design questions below.

---

🟡 Recommended Information

Describe Alternatives You've Considered

Resolve the tools eagerly before building the graph. Broken for stdio and HTTP MCP servers for the event-loop reason above.

Accept a BaseToolset directly in edges / the NodeLike union. This is ambiguous, as a toolset is many tools, so it is not clear which one the node should run. An explicit ToolsetNode(...) allows defining exactly which tool to run as a node.

Wrap each MCP call in a FunctionNode that does its own get_tools(). Works, but every user re-implements resolution, loses the per-invocation cache, and still leaks the toolset (see below).

Proposed API / Implementation

toolset = McpToolset(connection_params=StdioConnectionParams(...))

root_agent = Workflow(
    name='research',
    edges=[(
        'START',
        build_args,
        ToolsetNode(toolset=toolset, tool_name='read_file'),
        summarize,
    )],
)

Resolution goes through BaseToolset.get_tools_with_prefix(), which is @final, applies tool_name_prefix, and caches per invocation_id. Several ToolsetNodes sharing one toolset therefore cost one list_tools round trip per run, not one per node.

A related bug this exposes. Runner._collect_toolset walks only agent.tools and agent.sub_agents. A Workflow has neither. It's nodes live in graph.nodes, so a toolset referenced only by a node is never passed to _cleanup_toolsets, and the MCP stdio subprocess outlives the run. The fix is to also walk graph.nodes. Happy to split this into its own issue and PR if it is preferred to be separate.

Additional Context

Two design questions I'd like a maintainer input on before the PR:

1. Node naming. BaseNode.name must be a valid Python identifier, but MCP servers commonly use dashes (read-file). I currently default the node name to tool_name with non-identifier characters replaced by _, with an explicit name= always winning. The alternative is to raise and force every user to pass name=. Silent renaming is friendlier but makes node paths slightly surprising. Which would be a better implementation?

2. Should the node be public? _ToolNode is private and unexported. I've made ToolsetNode public in google.adk.workflow.__all__, since unlike _ToolNode it cannot be reached implicitly by putting an object in edges and if it isn't exported, it isn't usable.

Known limitation, deliberately out of scope. _ToolNode discards everything in tool_context.actions except state_delta. For an McpToolset with an auth_scheme, BaseAuthenticatedTool.run_async calls request_credential and returns the literal string "Pending User Authorization.". The node yields that string as its output and the graph continues with it. require_confirmation is swallowed the same way. FunctionNode already has the right pattern via auth_config and create_auth_request_event / has_auth_credential / process_auth_resume. This predates the proposed change and affects _ToolNode
equally, so I'd file it separately rather than expand this PR.

Environment: ADK 2.6.0, Python 3.10 through 3.14.