feat: Allow custom part converters in A2A classes

This change introduces type descriptions for the functions which convert between A2A and GenAI `Part`s. It then allows passing instances of those functions to the various A2A-related functions/classes, effectively allowing users to inject their own logic for how part conversion should occur.

The benefit of this pattern is that users can create decorators around the core `Part` conversion logic, which allows them to intercept the cases they care about while delegating the ones they do not to the core converter. This is a pattern we use a lot in the A2A Python SDK.

One example where this type of logic is useful is for extensions: this allows extension logic to, for example, interpret an A2A DataPart into a FunctionResponse using extension-specific logic.

PiperOrigin-RevId: 803186799

Author: google-genai-bot

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

@@ -42,8 +42,10 @@
 from .part_converter import A2A_DATA_PART_METADATA_IS_LONG_RUNNING_KEY
 from .part_converter import A2A_DATA_PART_METADATA_TYPE_FUNCTION_CALL
 from .part_converter import A2A_DATA_PART_METADATA_TYPE_KEY
+from .part_converter import A2APartToGenAIPartConverter
 from .part_converter import convert_a2a_part_to_genai_part
 from .part_converter import convert_genai_part_to_a2a_part
+from .part_converter import GenAIPartToA2APartConverter
 from .utils import _get_adk_metadata_key
 
 # Constants
@@ -169,6 +171,7 @@ def convert_a2a_task_to_event(
     a2a_task: Task,
     author: Optional[str] = None,
     invocation_context: Optional[InvocationContext] = None,
+    part_converter: A2APartToGenAIPartConverter = convert_a2a_part_to_genai_part,
 ) -> Event:
   """Converts an A2A task to an ADK event.
 
@@ -177,6 +180,7 @@ def convert_a2a_task_to_event(
     author: The author of the event. Defaults to "a2a agent" if not provided.
     invocation_context: The invocation context containing session information.
       If provided, the branch will be set from the context.
+    part_converter: The function to convert A2A part to GenAI part.
 
   Returns:
     An ADK Event object representing the converted task.
@@ -203,7 +207,9 @@ def convert_a2a_task_to_event(
     # Convert message if available
     if message:
       try:
-        return convert_a2a_message_to_event(message, author, invocation_context)
+        return convert_a2a_message_to_event(
+            message, author, invocation_context, part_converter=part_converter
+        )
       except Exception as e:
         logger.error("Failed to convert A2A task message to event: %s", e)
         raise RuntimeError(f"Failed to convert task message: {e}") from e
@@ -229,6 +235,7 @@ def convert_a2a_message_to_event(
     a2a_message: Message,
     author: Optional[str] = None,
     invocation_context: Optional[InvocationContext] = None,
+    part_converter: A2APartToGenAIPartConverter = convert_a2a_part_to_genai_part,
 ) -> Event:
   """Converts an A2A message to an ADK event.
 
@@ -237,6 +244,7 @@ def convert_a2a_message_to_event(
     author: The author of the event. Defaults to "a2a agent" if not provided.
     invocation_context: The invocation context containing session information.
       If provided, the branch will be set from the context.
+    part_converter: The function to convert A2A part to GenAI part.
 
   Returns:
     An ADK Event object with converted content and long-running tool metadata.
@@ -269,7 +277,7 @@ def convert_a2a_message_to_event(
 
     for a2a_part in a2a_message.parts:
       try:
-        part = convert_a2a_part_to_genai_part(a2a_part)
+        part = part_converter(a2a_part)
         if part is None:
           logger.warning("Failed to convert A2A part, skipping: %s", a2a_part)
           continue
@@ -322,13 +330,18 @@ def convert_a2a_message_to_event(
 
 @a2a_experimental
 def convert_event_to_a2a_message(
-    event: Event, invocation_context: InvocationContext, role: Role = Role.agent
+    event: Event,
+    invocation_context: InvocationContext,
+    role: Role = Role.agent,
+    part_converter: GenAIPartToA2APartConverter = convert_genai_part_to_a2a_part,
 ) -> Optional[Message]:
   """Converts an ADK event to an A2A message.
 
   Args:
     event: The ADK event to convert.
     invocation_context: The invocation context.
+    role: The role of the message.
+    part_converter: The function to convert GenAI part to A2A part.
 
   Returns:
     An A2A Message if the event has content, None otherwise.
@@ -347,7 +360,7 @@ def convert_event_to_a2a_message(
   try:
     a2a_parts = []
     for part in event.content.parts:
-      a2a_part = convert_genai_part_to_a2a_part(part)
+      a2a_part = part_converter(part)
       if a2a_part:
         a2a_parts.append(a2a_part)
         _process_long_running_tool(a2a_part, event)
@@ -477,6 +490,7 @@ def convert_event_to_a2a_events(
     invocation_context: InvocationContext,
     task_id: Optional[str] = None,
     context_id: Optional[str] = None,
+    part_converter: GenAIPartToA2APartConverter = convert_genai_part_to_a2a_part,
 ) -> List[A2AEvent]:
   """Converts a GenAI event to a list of A2A events.
 
@@ -485,6 +499,7 @@ def convert_event_to_a2a_events(
     invocation_context: The invocation context.
     task_id: Optional task ID to use for generated events.
     context_id: Optional Context ID to use for generated events.
+    part_converter: The function to convert GenAI part to A2A part.
 
   Returns:
     A list of A2A events representing the converted ADK event.
@@ -509,7 +524,9 @@ def convert_event_to_a2a_events(
       a2a_events.append(error_event)
 
     # Handle regular message content
-    message = convert_event_to_a2a_message(event, invocation_context)
+    message = convert_event_to_a2a_message(
+        event, invocation_context, part_converter=part_converter
+    )
     if message:
       running_event = _create_status_update_event(
           message, invocation_context, event, task_id, context_id

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

@@ -19,6 +19,7 @@
 from __future__ import annotations
 
 import base64
+from collections.abc import Callable
 import json
 import logging
 from typing import Optional
@@ -51,6 +52,14 @@
 A2A_DATA_PART_METADATA_TYPE_EXECUTABLE_CODE = 'executable_code'
 
 
+A2APartToGenAIPartConverter = Callable[
+    [a2a_types.Part], Optional[genai_types.Part]
+]
+GenAIPartToA2APartConverter = Callable[
+    [genai_types.Part], Optional[a2a_types.Part]
+]
+
+
 @a2a_experimental
 def convert_a2a_part_to_genai_part(
     a2a_part: a2a_types.Part,

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

@@ -31,6 +31,7 @@
 
 from ...runners import RunConfig
 from ..experimental import a2a_experimental
+from .part_converter import A2APartToGenAIPartConverter
 from .part_converter import convert_a2a_part_to_genai_part
 
 
@@ -50,6 +51,7 @@ def _get_user_id(request: RequestContext) -> str:
 @a2a_experimental
 def convert_a2a_request_to_adk_run_args(
     request: RequestContext,
+    part_converter: A2APartToGenAIPartConverter = convert_a2a_part_to_genai_part,
 ) -> dict[str, Any]:
 
   if not request.message:
@@ -60,10 +62,7 @@ def convert_a2a_request_to_adk_run_args(
       'session_id': request.context_id,
       'new_message': genai_types.Content(
           role='user',
-          parts=[
-              convert_a2a_part_to_genai_part(part)
-              for part in request.message.parts
-          ],
+          parts=[part_converter(part) for part in request.message.parts],
       ),
       'run_config': RunConfig(),
   }

src/google/adk/a2a/executor/a2a_agent_executor.py

@@ -53,6 +53,10 @@
 from typing_extensions import override
 
 from ..converters.event_converter import convert_event_to_a2a_events
+from ..converters.part_converter import A2APartToGenAIPartConverter
+from ..converters.part_converter import convert_a2a_part_to_genai_part
+from ..converters.part_converter import convert_genai_part_to_a2a_part
+from ..converters.part_converter import GenAIPartToA2APartConverter
 from ..converters.request_converter import convert_a2a_request_to_adk_run_args
 from ..converters.utils import _get_adk_metadata_key
 from ..experimental import a2a_experimental
@@ -65,12 +69,18 @@
 class A2aAgentExecutorConfig(BaseModel):
   """Configuration for the A2aAgentExecutor."""
 
-  pass
+  a2a_part_converter: A2APartToGenAIPartConverter = (
+      convert_a2a_part_to_genai_part
+  )
+  gen_ai_part_converter: GenAIPartToA2APartConverter = (
+      convert_genai_part_to_a2a_part
+  )
 
 
 @a2a_experimental
 class A2aAgentExecutor(AgentExecutor):
   """An AgentExecutor that runs an ADK Agent against an A2A request and
+
   publishes updates to an event queue.
   """
 
@@ -82,7 +92,7 @@ def __init__(
   ):
     super().__init__()
     self._runner = runner
-    self._config = config
+    self._config = config or A2aAgentExecutorConfig()
 
   async def _resolve_runner(self) -> Runner:
     """Resolve the runner, handling cases where it's a callable that returns a Runner."""
@@ -183,7 +193,9 @@ async def _handle_request(
     runner = await self._resolve_runner()
 
     # Convert the a2a request to ADK run args
-    run_args = convert_a2a_request_to_adk_run_args(context)
+    run_args = convert_a2a_request_to_adk_run_args(
+        context, self._config.a2a_part_converter
+    )
 
     # ensure the session exists
     session = await self._prepare_session(context, run_args, runner)
@@ -217,7 +229,11 @@ async def _handle_request(
     async with Aclosing(runner.run_async(**run_args)) as agen:
       async for adk_event in agen:
         for a2a_event in convert_event_to_a2a_events(
-            adk_event, invocation_context, context.task_id, context.context_id
+            adk_event,
+            invocation_context,
+            context.task_id,
+            context.context_id,
+            self._config.gen_ai_part_converter,
         ):
           task_result_aggregator.process_event(a2a_event)
           await event_queue.enqueue_event(a2a_event)

src/google/adk/agents/remote_a2a_agent.py

@@ -57,7 +57,10 @@
 from ..a2a.converters.event_converter import convert_a2a_message_to_event
 from ..a2a.converters.event_converter import convert_a2a_task_to_event
 from ..a2a.converters.event_converter import convert_event_to_a2a_message
+from ..a2a.converters.part_converter import A2APartToGenAIPartConverter
+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.experimental import a2a_experimental
 from ..a2a.logs.log_utils import build_a2a_request_log
 from ..a2a.logs.log_utils import build_a2a_response_log
@@ -120,6 +123,8 @@ def __init__(
       description: str = "",
       httpx_client: Optional[httpx.AsyncClient] = None,
       timeout: float = DEFAULT_TIMEOUT,
+      genai_part_converter: GenAIPartToA2APartConverter = convert_genai_part_to_a2a_part,
+      a2a_part_converter: A2APartToGenAIPartConverter = convert_a2a_part_to_genai_part,
       **kwargs: Any,
   ) -> None:
     """Initialize RemoteA2aAgent.
@@ -149,6 +154,8 @@ def __init__(
     self._httpx_client_needs_cleanup = httpx_client is None
     self._timeout = timeout
     self._is_resolved = False
+    self._genai_part_converter = genai_part_converter
+    self._a2a_part_converter = a2a_part_converter
 
     # Validate and store agent card reference
     if isinstance(agent_card, AgentCard):
@@ -298,7 +305,7 @@ def _create_a2a_request_for_user_function_response(
       return None
 
     a2a_message = convert_event_to_a2a_message(
-        ctx.session.events[-1], ctx, Role.user
+        ctx.session.events[-1], ctx, Role.user, self._genai_part_converter
     )
     if function_call_event.custom_metadata:
       a2a_message.task_id = (
@@ -355,7 +362,7 @@ def _construct_message_parts_from_session(
 
       for part in event.content.parts:
 
-        converted_part = convert_genai_part_to_a2a_part(part)
+        converted_part = self._genai_part_converter(part)
         if converted_part:
           message_parts.append(converted_part)
         else:
@@ -380,7 +387,10 @@ async def _handle_a2a_response(
         if a2a_response.root.result:
           if isinstance(a2a_response.root.result, A2ATask):
             event = convert_a2a_task_to_event(
-                a2a_response.root.result, self.name, ctx
+                a2a_response.root.result,
+                self.name,
+                ctx,
+                self._a2a_part_converter,
             )
             event.custom_metadata = event.custom_metadata or {}
             event.custom_metadata[A2A_METADATA_PREFIX + "task_id"] = (
@@ -389,7 +399,10 @@ async def _handle_a2a_response(
 
           else:
             event = convert_a2a_message_to_event(
-                a2a_response.root.result, self.name, ctx
+                a2a_response.root.result,
+                self.name,
+                ctx,
+                self._a2a_part_converter,
             )
             event.custom_metadata = event.custom_metadata or {}
             if a2a_response.root.result.task_id: