feat: add support for non-ADK produced input-required events

This CL introduces logic to handle scenarios where a non-ADK agent transitions to the `TaskState.input_required` or `TaskState.auth_required` states. It intercepts these events and converts them into a synthetic ADK `FunctionCall` event.

PiperOrigin-RevId: 914877123

Author: google-genai-bot

src/google/adk/a2a/converters/to_adk_event.py

@@ -43,6 +43,10 @@
 # Logger
 logger = logging.getLogger("google_adk." + __name__)
 
+MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT = (
+    "mock_function_call_for_required_user_input"
+)
+
 A2AMessageToEventConverter = Callable[
     [
         Message,
@@ -276,6 +280,36 @@ def _merge_event_actions(
   return EventActions.model_validate(merged_actions_data)
 
 
+def _create_mock_function_call_for_required_user_input(
+    state: TaskState,
+    output_parts: list[genai_types.Part],
+    long_running_function_ids: set[str],
+) -> tuple[list[genai_types.Part], set[str]]:
+  """Creates a mock function call for input/auth-required if applicable.
+
+  This solution allows to unblock the A2A integration with non-ADK agents from
+  ADK side by replacing the last text part with a synthetic function call. All
+  other parts are preserved.
+  """
+  if (
+      state == TaskState.input_required or state == TaskState.auth_required
+  ) and (not long_running_function_ids or len(long_running_function_ids) == 0):
+    # Find the last text part from the bottom to replace it with a function call.
+    # In case of input-required events, the LLM should stop the production of other parts.
+    for i in range(len(output_parts) - 1, -1, -1):
+      if output_parts[i].text:
+        function_call = genai_types.FunctionCall(
+            id=str(uuid.uuid4()),
+            name=MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT,
+            args={"input_required": output_parts[i].text},
+        )
+        long_running_function_ids = set()
+        long_running_function_ids.add(function_call.id)
+        output_parts[i] = genai_types.Part(function_call=function_call)
+        break
+  return output_parts, long_running_function_ids
+
+
 @a2a_experimental
 def convert_a2a_task_to_event(
     a2a_task: Task,
@@ -317,9 +351,9 @@ def convert_a2a_task_to_event(
       output_parts, _ = _convert_a2a_parts_to_adk_parts(
           artifact_parts, part_converter
       )
-    if (
-        a2a_task.status.message
-        and a2a_task.status.state == TaskState.input_required
+    if a2a_task.status.message and (
+        a2a_task.status.state == TaskState.input_required
+        or a2a_task.status.state == TaskState.auth_required
     ):
       event_actions = _merge_event_actions(
           event_actions,
@@ -331,6 +365,12 @@ def convert_a2a_task_to_event(
       output_parts.extend(parts)
       long_running_function_ids.update(ids)
 
+    output_parts, long_running_function_ids = (
+        _create_mock_function_call_for_required_user_input(
+            a2a_task.status.state, output_parts, long_running_function_ids
+        )
+    )
+
     return _create_event(
         output_parts,
         invocation_context,
@@ -422,6 +462,14 @@ def convert_a2a_status_update_to_event(
       output_parts.extend(parts)
       long_running_function_ids.update(ids)
 
+    output_parts, long_running_function_ids = (
+        _create_mock_function_call_for_required_user_input(
+            a2a_status_update.status.state,
+            output_parts,
+            long_running_function_ids,
+        )
+    )
+
     return _create_event(
         output_parts,
         invocation_context,

src/google/adk/agents/remote_a2a_agent.py

@@ -65,6 +65,8 @@
 from ..a2a.converters.part_converter import convert_a2a_part_to_genai_part
 from ..a2a.converters.part_converter import convert_genai_part_to_a2a_part
 from ..a2a.converters.part_converter import GenAIPartToA2APartConverter
+from ..a2a.converters.to_adk_event import _create_mock_function_call_for_required_user_input
+from ..a2a.converters.to_adk_event import MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT
 from ..a2a.converters.utils import _get_adk_metadata_key
 from ..a2a.experimental import a2a_experimental
 from ..a2a.logs.log_utils import build_a2a_request_log
@@ -105,6 +107,22 @@ class A2AClientError(Exception):
   pass
 
 
+def _add_mock_function_call(event: Event, state: TaskState) -> None:
+  """Generates a mock function call for input-required events if applicable."""
+  if event.content is None:
+    return
+
+  output_parts, long_running_tool_ids = (
+      _create_mock_function_call_for_required_user_input(
+          state,
+          event.content.parts,
+          event.long_running_tool_ids,
+      )
+  )
+  event.content.parts = output_parts
+  event.long_running_tool_ids = long_running_tool_ids
+
+
 @a2a_experimental
 class RemoteA2aAgent(BaseAgent):
   """Agent that communicates with a remote A2A agent via A2A client.
@@ -360,8 +378,40 @@ def _create_a2a_request_for_user_function_response(
     if not function_call_event:
       return None
 
+    event = ctx.session.events[-1]
+    # If the user function_response replies to a function_call for non-ADK
+    # input-required events (fc.name = MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT),
+    # the function_response part is replaced with text extracted from the
+    # function response.
+    # The implementation is based on the assumption that the user function_response
+    # event will contain a function_response with the name
+    # MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT and the response will
+    # contain a "result" field with the user input as a string text.
+    mock_function_call = [
+        fc
+        for fc in function_call_event.get_function_calls()
+        if fc.name == MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT
+    ]
+    if mock_function_call:
+      new_parts = []
+      for function_response in event.get_function_responses():
+        if (
+            function_response.name == MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT
+            and function_response.response
+            and "result" in function_response.response
+        ):
+          text_value = function_response.response.get("result")
+          new_parts.append(
+              genai_types.Part(
+                  text=str(text_value),
+              )
+          )
+      new_event = event.model_copy(deep=True)
+      new_event.content.parts = new_parts
+      event = new_event
+
     a2a_message = convert_event_to_a2a_message(
-        ctx.session.events[-1], ctx, Role.user, self._genai_part_converter
+        event, ctx, Role.user, self._genai_part_converter
     )
     if function_call_event.custom_metadata:
       metadata = function_call_event.custom_metadata
@@ -472,6 +522,7 @@ async def _handle_a2a_response(
           ):
             for part in event.content.parts:
               part.thought = True
+          _add_mock_function_call(event, task.status.state)
         elif (
             isinstance(update, A2ATaskStatusUpdateEvent)
             and update.status
@@ -487,6 +538,7 @@ async def _handle_a2a_response(
           ):
             for part in event.content.parts:
               part.thought = True
+          _add_mock_function_call(event, update.status.state)
         elif isinstance(update, A2ATaskArtifactUpdateEvent) and (
             not update.append or update.last_chunk
         ):

tests/unittests/a2a/converters/test_to_adk.py

@@ -30,6 +30,7 @@
 from google.adk.a2a.converters.to_adk_event import convert_a2a_message_to_event
 from google.adk.a2a.converters.to_adk_event import convert_a2a_status_update_to_event
 from google.adk.a2a.converters.to_adk_event import convert_a2a_task_to_event
+from google.adk.a2a.converters.to_adk_event import MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT
 from google.adk.a2a.converters.utils import _get_adk_metadata_key
 from google.adk.agents.invocation_context import InvocationContext
 from google.genai import types as genai_types
@@ -330,12 +331,95 @@ def test_convert_a2a_task_to_event_merges_status_and_artifact_actions(self):
     assert event.actions.state_delta == {"saved_key": "saved-value"}
     assert event.actions.transfer_to_agent == "agent-2"
     assert event.content is not None
-    assert event.content.parts == [mock_genai_part]
+    assert (
+        event.content.parts[0].function_call.name
+        == MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT
+    )
+    assert (
+        event.content.parts[0].function_call.args["input_required"]
+        == "need input"
+    )
+
+  def test_convert_a2a_task_to_event_multiple_parts_replaces_last_text(self):
+    """Test converting A2A task with multiple text parts, only replacing the last text."""
+    part1 = Mock(spec=A2APart)
+    part1.root = Mock(spec=TextPart)
+    part1.root.metadata = {}
+    part2 = Mock(spec=A2APart)
+    part2.root = Mock(spec=TextPart)
+    part2.root.metadata = {}
+
+    task = Task(
+        id="task-1",
+        context_id="context-1",
+        kind="task",
+        status=TaskStatus(
+            state=TaskState.input_required,
+            timestamp="now",
+            message=Message(
+                message_id="m1",
+                role="agent",
+                parts=[part1, part2],
+            ),
+        ),
+    )
+
+    mock_genai_part_1 = genai_types.Part.from_text(text="Part 1")
+    mock_genai_part_2 = genai_types.Part.from_text(text="Part 2")
 
-  def test_convert_a2a_task_to_event_none(self):
-    """Test convert_a2a_task_to_event with None."""
-    with pytest.raises(ValueError, match="A2A task cannot be None"):
-      convert_a2a_task_to_event(None)
+    part_converter_mock = Mock()
+    part_converter_mock.side_effect = [[mock_genai_part_1], [mock_genai_part_2]]
+
+    event = convert_a2a_task_to_event(
+        task,
+        author="test-author",
+        invocation_context=self.mock_context,
+        part_converter=part_converter_mock,
+    )
+
+    assert event is not None
+    assert event.content is not None
+    assert len(event.content.parts) == 2
+    assert event.content.parts[0].text == "Part 1"
+    assert (
+        event.content.parts[1].function_call.name
+        == MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT
+    )
+
+  def test_convert_a2a_task_to_event_no_text_parts(self):
+    """Test converting A2A task with no text parts should not inject function call."""
+    part1 = Mock(spec=A2APart)
+    part1.root = Mock()  # Not a TextPart
+    part1.root.metadata = {}
+
+    task = Task(
+        id="task-1",
+        context_id="context-1",
+        kind="task",
+        status=TaskStatus(
+            state=TaskState.input_required,
+            timestamp="now",
+            message=Message(
+                message_id="m1",
+                role="agent",
+                parts=[part1],
+            ),
+        ),
+    )
+    mock_image_part = genai_types.Part(
+        inline_data=genai_types.Blob(mime_type="image/jpeg", data=b"fake")
+    )
+
+    event = convert_a2a_task_to_event(
+        task,
+        author="test-author",
+        invocation_context=self.mock_context,
+        part_converter=Mock(return_value=[mock_image_part]),
+    )
+
+    assert event is not None
+    assert event.content is not None
+    assert event.content.parts == [mock_image_part]
 
   def test_convert_a2a_status_update_to_event_success(self):
     """Test successful conversion of A2A status update to Event."""