Support MCP and other toolset tools as graph workflow nodes
- Dominant language
- Python
- Stars
- 21.5k
- Forks
- 4k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 37
Description
## 🔴 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:
```python
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.
```python
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
```python
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 `ToolsetNode`s 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.
Contributor guide
Assessment
This issue has not been assessed yet.