feat(workflow): Allow ToolNode to accept JSON string or Content inputs

Merge #6065

**Please ensure you have read the [contribution guide](https://github.com/google/adk-python/blob/main/CONTRIBUTING.md) before creating a pull request.**

### Link to Issue or Description of Change

**1. Link to an existing issue (if applicable):**

- Closes: N/A
- Related: N/A

**2. Or, if no issue exists, describe the change:**

**Problem:**
Previously, `ToolNode` in ADK workflows required input to be strictly a Python `dict` or `None`. This limited workflow flexibility, preventing users from chaining a `ToolNode` directly downstream of nodes (like LLM Agents or custom function nodes) that output inputs as serialized JSON strings or `types.Content` objects without manual parsing boilerplate.

**Solution:**
Enhanced `ToolNode._run_impl` in [_tool_node.py](file:///usr/local/google/home/westerberg/Documents/github/adk-python-fork/src/google/adk/workflow/_tool_node.py) to dynamically coerce input formats before tool execution:
- Extract text from `types.Content` when received.
- Strip and deserialize JSON strings via `json.loads` if received.
- Coerce `None` to `{}`.
- Keep strict validation raising `TypeError` for non-dictionary representations (e.g. lists, invalid JSON).

---

### Testing Plan

**Unit Tests:**

- [x] I have added or updated unit tests for my change.
- [x] All unit tests pass locally.

**Pytest Summary:**
```
tests/unittests/workflow/test_tool_node.py .......                       [100%]
======================== 7 passed, 4 warnings in 2.80s =========================
```

**Manual End-to-End (E2E) Tests:**
Verified all workflow tests pass locally with no regressions:
```
uv run pytest tests/unittests/workflow/
...
==== 600 passed, 10 skipped, 26 xfailed, 9 xpassed, 237 warnings in 15.17s =====
```

---

### Checklist

- [x] I have read the [CONTRIBUTING.md](https://github.com/google/adk-python/blob/main/CONTRIBUTING.md) document.
- [x] I have performed a self-review of my own code.
- [x] I have commented my code, particularly in hard-to-understand areas.
- [x] I have added tests that prove my fix is effective or that my feature works.
- [x] New and existing unit tests pass locally with my changes.
- [x] I have manually tested my changes end-to-end.
- [x] Any dependent changes have been merged and published in downstream modules.

Co-authored-by: George Weale <gweale@google.com>
COPYBARA_INTEGRATE_REVIEW=#6065 from baptmont:tool_node_args 36edc1f
PiperOrigin-RevId: 940575182

Author: 2 people

src/google/adk/workflow/_tool_node.py

@@ -17,9 +17,11 @@
 """A node that wraps an ADK Tool."""
 
 from collections.abc import AsyncGenerator
+import json
 from typing import Any
 import uuid
 
+from google.genai import types
 from pydantic import ConfigDict
 from pydantic import Field
 from typing_extensions import override
@@ -28,6 +30,7 @@
 from ..events.event import Event
 from ..tools.base_tool import BaseTool
 from ..tools.tool_context import ToolContext
+from ..utils.content_utils import extract_text_from_content
 from ._base_node import BaseNode
 from ._retry_config import RetryConfig
 
@@ -67,6 +70,19 @@ async def _run_impl(
     )
 
     args = node_input
+    if isinstance(args, types.Content):
+      args = extract_text_from_content(args)
+
+    if isinstance(args, str):
+      args = args.strip()
+      if not args:
+        args = None
+      else:
+        try:
+          args = json.loads(args)
+        except json.JSONDecodeError:
+          pass
+
     if args is None:
       args = {}
     elif not isinstance(args, dict):

tests/unittests/workflow/test_tool_node.py

@@ -0,0 +1,141 @@
+# 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.
+
+"""Tests for ToolNode input parsing and execution."""
+
+from typing import Any
+
+from google.adk.events.event import Event
+from google.adk.tools.base_tool import BaseTool
+from google.adk.workflow import START
+from google.adk.workflow._tool_node import _ToolNode as ToolNode
+from google.adk.workflow._workflow import Workflow
+from google.genai import types
+import pytest
+
+from . import workflow_testing_utils
+from .. import testing_utils
+
+
+class MockTool(BaseTool):
+  """A mock tool that returns the args it was called with."""
+
+  def __init__(self, name="mock_tool", description="Mock tool"):
+    super().__init__(name=name, description=description)
+
+  async def run_async(self, *, args: dict[str, Any], tool_context) -> Any:
+    return args
+
+
+async def _run_tool_node_wf(node_input: Any) -> list[Any]:
+  """Runs a workflow with a ToolNode that receives node_input."""
+  tool_node = ToolNode(tool=MockTool())
+
+  def start_node():
+    return Event(output=node_input)
+
+  wf = Workflow(
+      name="tool_node_test_wf",
+      edges=[
+          (START, start_node),
+          (start_node, tool_node),
+      ],
+  )
+  app_instance = testing_utils.App(name="test_app", root_agent=wf)
+  runner = testing_utils.InMemoryRunner(app=app_instance)
+  events = await runner.run_async("start")
+  return workflow_testing_utils.simplify_events_with_node(events)
+
+
+@pytest.mark.asyncio
+async def test_tool_node_accepts_dict():
+  """Tests that ToolNode accepts a dict as input and passes it to the tool."""
+  input_dict = {"param_a": 1, "param_b": "value"}
+  simplified = await _run_tool_node_wf(input_dict)
+  assert (
+      "tool_node_test_wf@1/mock_tool@1",
+      {"output": input_dict},
+  ) in simplified
+
+
+@pytest.mark.asyncio
+async def test_tool_node_accepts_none():
+  """Tests that ToolNode accepts None, converting it to an empty dict."""
+  simplified = await _run_tool_node_wf(None)
+  assert ("tool_node_test_wf@1/mock_tool@1", {"output": {}}) in simplified
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("empty_input", ["", "   ", "\n\t"])
+async def test_tool_node_accepts_empty_string(empty_input):
+  """Tests that ToolNode treats an empty/whitespace string as no arguments."""
+  simplified = await _run_tool_node_wf(empty_input)
+  assert ("tool_node_test_wf@1/mock_tool@1", {"output": {}}) in simplified
+
+
+@pytest.mark.asyncio
+async def test_tool_node_accepts_json_string():
+  """Tests that ToolNode accepts a valid JSON string representing a dict."""
+  json_str = '{"param_a": 1, "param_b": "value"}'
+  simplified = await _run_tool_node_wf(json_str)
+  assert (
+      "tool_node_test_wf@1/mock_tool@1",
+      {"output": {"param_a": 1, "param_b": "value"}},
+  ) in simplified
+
+
+@pytest.mark.asyncio
+async def test_tool_node_accepts_content_with_json_string():
+  """Tests that ToolNode accepts a types.Content containing a JSON string."""
+  json_str = '{"param_a": 1, "param_b": "value"}'
+  content = types.Content(
+      parts=[types.Part.from_text(text=json_str)], role="user"
+  )
+  simplified = await _run_tool_node_wf(content)
+  assert (
+      "tool_node_test_wf@1/mock_tool@1",
+      {"output": {"param_a": 1, "param_b": "value"}},
+  ) in simplified
+
+
+@pytest.mark.asyncio
+async def test_tool_node_rejects_non_dict_json_string():
+  """Tests that ToolNode raises TypeError if JSON string represents a non-dict (e.g. list)."""
+  json_str = "[1, 2, 3]"
+  with pytest.raises(
+      TypeError, match="The input to ToolNode must be a dictionary"
+  ):
+    await _run_tool_node_wf(json_str)
+
+
+@pytest.mark.asyncio
+async def test_tool_node_rejects_invalid_json_string():
+  """Tests that ToolNode raises TypeError if string input is not valid JSON."""
+  invalid_str = "not a json"
+  with pytest.raises(
+      TypeError, match="The input to ToolNode must be a dictionary"
+  ):
+    await _run_tool_node_wf(invalid_str)
+
+
+@pytest.mark.asyncio
+async def test_tool_node_rejects_non_dict_content():
+  """Tests that ToolNode raises TypeError if Content contains non-dict text."""
+  content = types.Content(
+      parts=[types.Part.from_text(text="not a json")], role="user"
+  )
+  with pytest.raises(
+      TypeError, match="The input to ToolNode must be a dictionary"
+  ):
+    await _run_tool_node_wf(content)