fix(agents): skip output_key processing on intermediate conversational text in task-mode LlmAgent

Task-mode LlmAgents produce intermediate conversational text responses (such as clarification questions or greetings) before completing their task via finish_task.

When output_key and output_schema were configured on a task-mode agent, ADK attempted to validate intermediate conversational text against output_schema during state_delta processing, causing a ValidationError crash on non-JSON text.

- Skip output_key state_delta processing on intermediate text responses for task-mode agents (mode="task") in __maybe_save_output_to_state and _save_output_to_state.
- Preserve output_schema validation for non-task agents.
- Ensure task-mode finish_task validated output is written to output_key state_delta on task completion.
- Add unit tests for task-mode output_key handling.

Co-authored-by: Shangjie Chen <deanchen@google.com>
PiperOrigin-RevId: 949357458

Author: DeanChensj

src/google/adk/agents/llm_agent.py

@@ -967,6 +967,11 @@ def __maybe_save_output_to_state(self, event: Event) -> None:
     if not self.output_key:
       return
 
+    # Task mode agents deliver their final output via finish_task, not intermediate
+    # conversational text turns. Skip output_key processing on text responses for task mode.
+    if getattr(self, 'mode', None) == 'task':
+      return
+
     # Handle text responses
     if event.is_final_response() and event.content and event.content.parts:
 
@@ -1016,6 +1021,7 @@ def __maybe_accumulate_streaming_output(
     """
     if (
         not self.output_key
+        or getattr(self, 'mode', None) == 'task'
         or self.output_schema
         or event.author != self.name
         or event.partial

src/google/adk/workflow/_llm_agent_wrapper.py

@@ -429,6 +429,8 @@ async def run_llm_agent_as_node(
           event.output = pending_fc_args[wrapper_key]
         else:
           event.output = pending_fc_args
+        if getattr(agent, 'output_key', None) and event.output is not None:
+          ctx.actions.state_delta[agent.output_key] = event.output
         yield event
         return
 

tests/unittests/agents/test_llm_agent_output_save.py

@@ -150,6 +150,38 @@ def test_maybe_save_output_to_state_with_output_schema(self):
     expected_output = {"message": "Hello", "confidence": 0.95}
     assert event.actions.state_delta["result"] == expected_output
 
+  def test_maybe_save_output_to_state_task_mode_skips_intermediate_text(self):
+    """Test that intermediate conversational text in task mode is skipped for output_key."""
+    agent = LlmAgent(
+        name="test_agent",
+        mode="task",
+        output_key="result",
+        output_schema=MockOutputSchema,
+    )
+
+    # Conversational text (not JSON) emitted during clarification
+    event = create_test_event(
+        author="test_agent",
+        content_text="Please describe your data domain.",
+    )
+
+    agent._LlmAgent__maybe_save_output_to_state(event)
+
+    # Task mode skips intermediate text events for output_key
+    assert "result" not in event.actions.state_delta
+
+  def test_maybe_accumulate_streaming_output_task_mode_skips(self):
+    """Test that streaming accumulation skips task-mode agents."""
+    agent = LlmAgent(name="test_agent", mode="task", output_key="result")
+    event = create_test_event(
+        author="test_agent", content_text="Streaming chunk"
+    )
+
+    accum = agent._LlmAgent__maybe_accumulate_streaming_output(event, "")
+
+    assert accum == ""
+    assert "result" not in event.actions.state_delta
+
   def test_maybe_save_output_to_state_multiple_parts(self):
     """Test that multiple text parts are concatenated."""
     agent = LlmAgent(name="test_agent", output_key="result")