feat(workflow): Support parsing input schema from Content text

Change-Id: Id20483da220c7b0d84bea68674f6614657a2dc42

Author: DeanChensj

src/google/adk/workflow/_base_node.py

@@ -125,8 +125,17 @@ 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 is str and isinstance(data, types.Content):
-      return ''.join(part.text for part in data.parts if part.text)
+    if self.input_schema and isinstance(data, types.Content):
+      text = "".join(part.text for part in data.parts if part.text)
+      if self.input_schema is str:
+        return text
+      try:
+        return TypeAdapter(self.input_schema).validate_json(text)
+      except Exception:
+        try:
+          return TypeAdapter(self.input_schema).validate_python(text)
+        except Exception:
+          pass
     return self._validate_schema(data, self.input_schema)
 
   def _validate_output_data(self, data: Any) -> Any:

tests/unittests/workflow/test_workflow_schema.py

@@ -543,18 +543,66 @@ async def _run_impl(
   assert any(e.output == 'done' for e in data_events)
 
 
-@pytest.mark.xfail(reason='Input schema parsing not yet in new Workflow.')
 @pytest.mark.asyncio
 async def test_start_node_with_int_input_schema():
   """input_schema=int parses user text to int."""
-  assert False, 'TODO'
+
+  class _AssertingNode(BaseNode):
+
+    async def _run_impl(
+        self, *, ctx: Context, node_input: Any
+    ) -> AsyncGenerator[Any, None]:
+      assert node_input == 42
+      yield 'done'
+
+  node = _AssertingNode(name='node', input_schema=int)
+  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='42')], 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 any(e.output == 'done' for e in data_events)
 
 
-@pytest.mark.xfail(reason='Input schema parsing not yet in new Workflow.')
 @pytest.mark.asyncio
 async def test_start_node_with_int_list_input_schema():
   """input_schema=list[int] parses JSON list."""
-  assert False, 'TODO'
+
+  class _AssertingNode(BaseNode):
+
+    async def _run_impl(
+        self, *, ctx: Context, node_input: Any
+    ) -> AsyncGenerator[Any, None]:
+      assert node_input == [1, 2, 3]
+      yield 'done'
+
+  node = _AssertingNode(name='node', input_schema=list[int])
+  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='[1, 2, 3]')], 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 any(e.output == 'done' for e in data_events)
 
 
 @pytest.mark.asyncio