Merge branch 'main' into feat/list-sessions-pagination

Author: ferponse

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

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
+import json
 import logging
 from typing import Any
 from typing import List
@@ -28,9 +29,11 @@
 from a2a.types import TaskState
 from a2a.types import TaskStatusUpdateEvent
 from google.genai import types as genai_types
+from pydantic import ValidationError
 
 from ...agents.invocation_context import InvocationContext
 from ...events.event import Event
+from ...events.event_actions import EventActions
 from ..experimental import a2a_experimental
 from .part_converter import A2A_DATA_PART_METADATA_IS_LONG_RUNNING_KEY
 from .part_converter import A2APartToGenAIPartConverter
@@ -171,11 +174,15 @@ def _create_event(
     output_parts: List[genai_types.Part],
     invocation_context: Optional[InvocationContext],
     author: Optional[str],
+    actions: Optional[EventActions] = None,
     long_running_function_ids: Optional[set[str]] = None,
     partial: bool = False,
 ) -> Optional[Event]:
   """Creates an ADK event from parts and metadata."""
-  if not output_parts:
+  event_actions = actions or EventActions()
+  if not output_parts and not event_actions.model_dump(
+      exclude_none=True, exclude_defaults=True
+  ):
     return None
 
   event = Event(
@@ -186,19 +193,89 @@ def _create_event(
       ),
       author=author or "a2a agent",
       branch=invocation_context.branch if invocation_context else None,
+      actions=event_actions,
       long_running_tool_ids=(
           long_running_function_ids if long_running_function_ids else None
       ),
-      content=genai_types.Content(
-          role="model",
-          parts=output_parts,
+      content=(
+          genai_types.Content(
+              role="model",
+              parts=output_parts,
+          )
+          if output_parts
+          else None
       ),
       partial=partial,
   )
 
   return event
 
 
+def _parse_adk_metadata_value(value: Any) -> Any:
+  """Parses ADK metadata values serialized through A2A."""
+  if not isinstance(value, str):
+    return value
+
+  try:
+    return json.loads(value)
+  except json.JSONDecodeError:
+    return value
+
+
+def _extract_event_actions(
+    metadata: Optional[dict[str, Any]],
+) -> EventActions:
+  """Extracts ADK event actions from A2A metadata."""
+  if not metadata:
+    return EventActions()
+
+  raw_actions = metadata.get(_get_adk_metadata_key("actions"))
+  if raw_actions is None:
+    return EventActions()
+
+  parsed_actions = _parse_adk_metadata_value(raw_actions)
+  if not isinstance(parsed_actions, dict):
+    logger.warning(
+        "Ignoring invalid ADK actions metadata of type %s",
+        type(parsed_actions).__name__,
+    )
+    return EventActions()
+
+  try:
+    return EventActions.model_validate(parsed_actions)
+  except ValidationError as error:
+    logger.warning("Ignoring invalid ADK actions metadata: %s", error)
+    return EventActions()
+
+
+def _merge_top_level_dicts(
+    base: dict[str, Any], new_values: dict[str, Any]
+) -> dict[str, Any]:
+  """Merges dictionaries while preserving top-level overwrite semantics."""
+  merged = dict(base)
+  for key, value in new_values.items():
+    if (
+        key in merged
+        and isinstance(merged[key], dict)
+        and isinstance(value, dict)
+    ):
+      merged[key] = {**merged[key], **value}
+    else:
+      merged[key] = value
+  return merged
+
+
+def _merge_event_actions(
+    existing_actions: EventActions, new_actions: EventActions
+) -> EventActions:
+  """Merges action metadata from multiple A2A sources."""
+  merged_actions_data = _merge_top_level_dicts(
+      existing_actions.model_dump(exclude_none=True, by_alias=True),
+      new_actions.model_dump(exclude_none=True, by_alias=True),
+  )
+  return EventActions.model_validate(merged_actions_data)
+
+
 @a2a_experimental
 def convert_a2a_task_to_event(
     a2a_task: Task,
@@ -226,19 +303,28 @@ def convert_a2a_task_to_event(
     raise ValueError("A2A task cannot be None")
 
   try:
+    event_actions = EventActions()
     output_parts = []
     long_running_function_ids = set()
     if a2a_task.artifacts:
       artifact_parts = [
           part for artifact in a2a_task.artifacts for part in artifact.parts
       ]
+      for artifact in a2a_task.artifacts:
+        event_actions = _merge_event_actions(
+            event_actions, _extract_event_actions(artifact.metadata)
+        )
       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
     ):
+      event_actions = _merge_event_actions(
+          event_actions,
+          _extract_event_actions(a2a_task.status.message.metadata),
+      )
       parts, ids = _convert_a2a_parts_to_adk_parts(
           a2a_task.status.message.parts, part_converter
       )
@@ -249,6 +335,7 @@ def convert_a2a_task_to_event(
         output_parts,
         invocation_context,
         author,
+        event_actions,
         long_running_function_ids,
     )
 
@@ -288,7 +375,12 @@ def convert_a2a_message_to_event(
     output_parts, _ = _convert_a2a_parts_to_adk_parts(
         a2a_message.parts, part_converter
     )
-    return _create_event(output_parts, invocation_context, author)
+    return _create_event(
+        output_parts,
+        invocation_context,
+        author,
+        _extract_event_actions(a2a_message.metadata),
+    )
 
   except Exception as e:
     logger.error("Failed to convert A2A message to event: %s", e)
@@ -319,7 +411,11 @@ def convert_a2a_status_update_to_event(
   try:
     output_parts = []
     long_running_function_ids = set()
+    event_actions = EventActions()
     if a2a_status_update.status.message:
+      event_actions = _extract_event_actions(
+          a2a_status_update.status.message.metadata
+      )
       parts, ids = _convert_a2a_parts_to_adk_parts(
           a2a_status_update.status.message.parts, part_converter
       )
@@ -330,6 +426,7 @@ def convert_a2a_status_update_to_event(
         output_parts,
         invocation_context,
         author,
+        event_actions,
         long_running_function_ids,
     )
   except Exception as e:
@@ -367,6 +464,7 @@ def convert_a2a_artifact_update_to_event(
         output_parts,
         invocation_context,
         author,
+        _extract_event_actions(a2a_artifact_update.artifact.metadata),
         partial=not a2a_artifact_update.last_chunk,
     )
   except Exception as e:

src/google/adk/a2a/utils/agent_to_a2a.py

@@ -14,7 +14,10 @@
 
 from __future__ import annotations
 
+from contextlib import asynccontextmanager
 import logging
+from typing import AsyncIterator
+from typing import Callable
 from typing import Optional
 from typing import Union
 
@@ -82,6 +85,7 @@ def to_a2a(
     agent_card: Optional[Union[AgentCard, str]] = None,
     push_config_store: Optional[PushNotificationConfigStore] = None,
     runner: Optional[Runner] = None,
+    lifespan: Optional[Callable[[Starlette], AsyncIterator[None]]] = None,
 ) -> Starlette:
   """Convert an ADK agent to a A2A Starlette application.
 
@@ -98,6 +102,11 @@ def to_a2a(
         config RPC methods are supported.
       runner: Optional pre-built Runner object. If not provided, a default
               runner will be created using in-memory services.
+      lifespan: Optional async context manager for Starlette lifespan
+        events. Use this to run startup/shutdown logic (e.g. initializing
+        database connections or loading resources). The context manager
+        receives the Starlette app instance and can set state on
+        ``app.state``.
 
   Returns:
       A Starlette application that can be run with uvicorn
@@ -109,6 +118,15 @@ def to_a2a(
 
       # Or with custom agent card:
       app = to_a2a(agent, agent_card=my_custom_agent_card)
+
+      # Or with lifespan:
+      @asynccontextmanager
+      async def lifespan(app):
+          app.state.db = await init_db()
+          yield
+          await app.state.db.close()
+
+      app = to_a2a(agent, lifespan=lifespan)
   """
   # Set up ADK logging to ensure logs are visible when using uvicorn directly
   adk_logger = logging.getLogger("google_adk")
@@ -151,11 +169,8 @@ async def create_runner() -> Runner:
       rpc_url=rpc_url,
   )
 
-  # Create a Starlette app that will be configured during startup
-  app = Starlette()
-
-  # Add startup handler to build the agent card and configure A2A routes
-  async def setup_a2a():
+  # Build the agent card and configure A2A routes
+  async def setup_a2a(app: Starlette):
     # Use provided agent card or build one asynchronously
     if provided_agent_card is not None:
       final_agent_card = provided_agent_card
@@ -173,7 +188,19 @@ async def setup_a2a():
         app,
     )
 
-  # Store the setup function to be called during startup
-  app.add_event_handler("startup", setup_a2a)
+  # Compose a lifespan that runs A2A setup and the user's lifespan
+  @asynccontextmanager
+  async def _combined_lifespan(
+      app: Starlette,
+  ) -> AsyncIterator[None]:
+    await setup_a2a(app)
+    if lifespan:
+      async with lifespan(app):
+        yield
+    else:
+      yield
+
+  # Create a Starlette app with the composed lifespan
+  app = Starlette(lifespan=_combined_lifespan)
 
   return app