feat(workflow): Support JSON string parsing in schema validation

Automatically parse JSON strings into dicts or Pydantic models when input_schema or output_schema is defined on a node.

This update refines the output schema validation to:
1. Try standard Pydantic validation first to preserve valid types (like raw strings or Content objects when requested).
2. Fallback to extracting text and parsing JSON ONLY if the data is a `types.Content` object and standard validation fails.

This ensures robust handling of LLM outputs while avoiding accidental parsing of valid raw strings.

Added comprehensive unit tests for both Content parsing and raw string validation failures.

Change-Id: I68fe4c636365d4d3c1c458fbf863e18cfcbd8479

Author: DeanChensj

src/google/adk/workflow/_base_node.py

@@ -126,12 +126,15 @@ def _validate_schema(self, data: Any, schema: Any) -> Any:
   def _validate_input_data(self, data: Any) -> Any:
     """Validates data against input_schema if set."""
     if self.input_schema and isinstance(data, types.Content):
-      text = "".join(part.text for part in data.parts if part.text)
+      # Extract text from Content (e.g. user input from START node).
+      text = ''.join(part.text for part in data.parts if part.text)
       if self.input_schema is str:
         return text
+      # If schema is defined, try to parse the text as JSON.
       try:
         return TypeAdapter(self.input_schema).validate_json(text)
       except Exception:
+        # Fallback to validate_python if it's a raw string matching the schema.
         try:
           return TypeAdapter(self.input_schema).validate_python(text)
         except Exception:
@@ -140,7 +143,27 @@ def _validate_input_data(self, data: Any) -> Any:
 
   def _validate_output_data(self, data: Any) -> Any:
     """Validates data against output_schema if set."""
-    return self._validate_schema(data, self.output_schema)
+    if not self.output_schema:
+      return data
+
+    # 1. Try standard validation first
+    try:
+      return self._validate_schema(data, self.output_schema)
+    except Exception as e:
+      # 2. If failed, try to parse JSON ONLY if it's Content
+      if isinstance(data, types.Content):
+        text = ''.join(part.text for part in data.parts if part.text)
+        if self.output_schema is str:
+          return text
+        if text.strip():
+          try:
+            validated = TypeAdapter(self.output_schema).validate_json(text)
+            return self._to_serializable(validated)
+          except Exception:
+            pass
+
+      # 3. If not Content or parsing failed, re-raise original error
+      raise e
 
   @staticmethod
   def _to_serializable(data: Any) -> Any:

tests/unittests/workflow/test_workflow_schema.py

@@ -668,3 +668,106 @@ async def _run_impl(
 
   data_events = [e for e in events if isinstance(e, Event) and e.output]
   assert any(e.output == 'done' for e in data_events)
+
+
+@pytest.mark.asyncio
+async def test_workflow_with_invalid_output_schema():
+  """Workflow raises ValidationError if terminal output doesn't match output_schema."""
+
+  from pydantic import ValidationError
+
+  class _MyModel(BaseModel):
+    name: str
+
+  class _MyNode(BaseNode):
+
+    async def _run_impl(
+        self, *, ctx: Context, node_input: Any
+    ) -> AsyncGenerator[Any, None]:
+      yield {'age': 10}
+
+  node = _MyNode(name='node')
+  wf = Workflow(name='wf', edges=[(START, node)], output_schema=_MyModel)
+
+  ss = InMemorySessionService()
+  runner = Runner(app_name='test', node=wf, session_service=ss)
+  session = await ss.create_session(app_name='test', user_id='u')
+
+  msg = types.Content(parts=[types.Part(text='hello')], role='user')
+
+  with pytest.raises(ValidationError):
+    async for event in runner.run_async(
+        user_id='u', session_id=session.id, new_message=msg
+    ):
+      pass
+
+
+@pytest.mark.asyncio
+async def test_node_returns_content_json_parsed():
+  """Node output as types.Content containing JSON is parsed if output_schema is defined."""
+
+  class _MyModel(BaseModel):
+    name: str
+    age: int
+
+  class _MyNode(BaseNode):
+
+    async def _run_impl(
+        self, *, ctx: Context, node_input: Any
+    ) -> AsyncGenerator[Any, None]:
+      yield self._validate_output_data(
+          types.Content(parts=[types.Part(text='{"name": "Alice", "age": 30}')])
+      )
+
+  node = _MyNode(name='node', output_schema=_MyModel)
+  wf = Workflow(name='wf', edges=[(START, node)])
+
+  ss = InMemorySessionService()
+  runner = Runner(app_name='test', node=wf, session_service=ss)
+  session = await ss.create_session(app_name='test', user_id='u')
+
+  msg = types.Content(parts=[types.Part(text='hello')], role='user')
+  events = []
+
+  async for event in runner.run_async(
+      user_id='u', session_id=session.id, new_message=msg
+  ):
+    events.append(event)
+
+  data_events = [e for e in events if isinstance(e, Event) and e.output]
+
+  assert len(data_events) == 1
+  assert data_events[0].output == {'name': 'Alice', 'age': 30}
+
+
+@pytest.mark.asyncio
+async def test_node_returns_raw_string_not_parsed():
+  """Node output as raw JSON string is NOT parsed if output_schema is defined."""
+  from pydantic import ValidationError
+
+  class _MyModel(BaseModel):
+    name: str
+    age: int
+
+  class _MyNode(BaseNode):
+
+    async def _run_impl(
+        self, *, ctx: Context, node_input: Any
+    ) -> AsyncGenerator[Any, None]:
+      # This should fail validation because it's a string, not a dict/model
+      yield self._validate_output_data('{"name": "Alice", "age": 30}')
+
+  node = _MyNode(name='node', output_schema=_MyModel)
+  wf = Workflow(name='wf', edges=[(START, node)])
+
+  ss = InMemorySessionService()
+  runner = Runner(app_name='test', node=wf, session_service=ss)
+  session = await ss.create_session(app_name='test', user_id='u')
+
+  msg = types.Content(parts=[types.Part(text='hello')], role='user')
+
+  with pytest.raises(ValidationError):
+    async for _ in runner.run_async(
+        user_id='u', session_id=session.id, new_message=msg
+    ):
+      pass