feat: add a2a_request_meta_provider to RemoteAgent init

The change adds an extension point for controlling which request metadata gets attached to A2A requests made by a RemoteAgent.
Instead of taking metadata from custom_metadata of session events users can construct payloads using a2a_request_meta_provider.

request_meta feature was added in v0.3.11 of the a2a-sdk library: https://github.com/a2aproject/a2a-python/releases/tag/v0.3.11

PiperOrigin-RevId: 831506364

Author: google-genai-bot

src/google/adk/agents/remote_a2a_agent.py

@@ -20,6 +20,7 @@
 from pathlib import Path
 from typing import Any
 from typing import AsyncGenerator
+from typing import Callable
 from typing import Optional
 from typing import Union
 from urllib.parse import urlparse
@@ -125,12 +126,16 @@ def __init__(
       self,
       name: str,
       agent_card: Union[AgentCard, str],
+      *,
       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,
       a2a_client_factory: Optional[A2AClientFactory] = None,
+      a2a_request_meta_provider: Optional[
+          Callable[[InvocationContext, A2AMessage], dict[str, Any]]
+      ] = None,
       **kwargs: Any,
   ) -> None:
     """Initialize RemoteA2aAgent.
@@ -144,6 +149,9 @@ def __init__(
       timeout: HTTP timeout in seconds
       a2a_client_factory: Optional A2AClientFactory object (will create own if
         not provided)
+      a2a_request_meta_provider: Optional callable that takes InvocationContext
+        and A2AMessage and returns a metadata object to attach to the A2A
+        request.
       **kwargs: Additional arguments passed to BaseAgent
 
     Raises:
@@ -169,6 +177,7 @@ def __init__(
     self._genai_part_converter = genai_part_converter
     self._a2a_part_converter = a2a_part_converter
     self._a2a_client_factory: Optional[A2AClientFactory] = a2a_client_factory
+    self._a2a_request_meta_provider = a2a_request_meta_provider
 
     # Validate and store agent card reference
     if isinstance(agent_card, AgentCard):
@@ -318,7 +327,7 @@ async def _ensure_resolved(self) -> None:
 
   def _create_a2a_request_for_user_function_response(
       self, ctx: InvocationContext
-  ) -> tuple[Optional[A2AMessage], Optional[dict[str, Any]]]:
+  ) -> Optional[A2AMessage]:
     """Create A2A request for user function response if applicable.
 
     Args:
@@ -328,26 +337,24 @@ def _create_a2a_request_for_user_function_response(
       SendMessageRequest if function response found, None otherwise
     """
     if not ctx.session.events or ctx.session.events[-1].author != "user":
-      return None, None
+      return None
     function_call_event = find_matching_function_call(ctx.session.events)
     if not function_call_event:
-      return None, None
+      return None
 
     a2a_message = convert_event_to_a2a_message(
         ctx.session.events[-1], ctx, Role.user, self._genai_part_converter
     )
-    message_metadata = None
     if function_call_event.custom_metadata:
       metadata = function_call_event.custom_metadata
       a2a_message.task_id = metadata.get(A2A_METADATA_PREFIX + "task_id")
       a2a_message.context_id = metadata.get(A2A_METADATA_PREFIX + "context_id")
-      message_metadata = metadata.get(A2A_METADATA_PREFIX + "metadata")
 
-    return a2a_message, message_metadata
+    return a2a_message
 
   def _construct_message_parts_from_session(
       self, ctx: InvocationContext
-  ) -> tuple[list[A2APart], Optional[str], Optional[dict[str, Any]]]:
+  ) -> tuple[list[A2APart], Optional[str]]:
     """Construct A2A message parts from session events.
 
     Args:
@@ -359,7 +366,6 @@ def _construct_message_parts_from_session(
     """
     message_parts: list[A2APart] = []
     context_id = None
-    request_metadata = None
 
     events_to_process = []
     for event in reversed(ctx.session.events):
@@ -369,7 +375,6 @@ def _construct_message_parts_from_session(
         if event.custom_metadata:
           metadata = event.custom_metadata
           context_id = metadata.get(A2A_METADATA_PREFIX + "context_id")
-          request_metadata = metadata.get(A2A_METADATA_PREFIX + "metadata")
         break
       events_to_process.append(event)
 
@@ -390,7 +395,7 @@ def _construct_message_parts_from_session(
         else:
           logger.warning("Failed to convert part to A2A format: %s", part)
 
-    return message_parts, context_id, request_metadata
+    return message_parts, context_id
 
   async def _handle_a2a_response(
       self, a2a_response: A2AClientEvent | A2AMessage, ctx: InvocationContext
@@ -498,12 +503,10 @@ async def _run_async_impl(
       return
 
     # Create A2A request for function response or regular message
-    a2a_request, request_metadata = (
-        self._create_a2a_request_for_user_function_response(ctx)
-    )
+    a2a_request = self._create_a2a_request_for_user_function_response(ctx)
     if not a2a_request:
-      message_parts, context_id, request_metadata = (
-          self._construct_message_parts_from_session(ctx)
+      message_parts, context_id = self._construct_message_parts_from_session(
+          ctx
       )
 
       if not message_parts:
@@ -528,6 +531,10 @@ async def _run_async_impl(
     logger.debug(build_a2a_request_log(a2a_request))
 
     try:
+      request_metadata = None
+      if self._a2a_request_meta_provider:
+        request_metadata = self._a2a_request_meta_provider(ctx, a2a_request)
+
       async for a2a_response in self._a2a_client.send_message(
           request=a2a_request,
           request_metadata=request_metadata,