Merge branch 'main' into fix_graph

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

@@ -40,7 +40,7 @@ from google.genai import types as genai_types
 from ...agents.invocation_context import InvocationContext
 from ...events.event import Event
 from ...flows.llm_flows.functions import REQUEST_EUC_FUNCTION_CALL_NAME
-from ...utils.feature_decorator import working_in_progress
+from ...utils.feature_decorator import experimental
 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
@@ -242,7 +242,6 @@ def _process_long_running_tool(a2a_part: A2APart, event: Event) -> None:
     ] = True
 
 
-@working_in_progress
 def convert_a2a_task_to_event(
     a2a_task: Task,
     author: Optional[str] = None,
@@ -298,7 +297,7 @@ def convert_a2a_task_to_event(
     raise
 
 
-@working_in_progress
+@experimental
 def convert_a2a_message_to_event(
     a2a_message: Message,
     author: Optional[str] = None,
@@ -394,7 +393,7 @@ def convert_a2a_message_to_event(
     raise RuntimeError(f"Failed to convert message: {e}") from e
 
 
-@working_in_progress
+@experimental
 def convert_event_to_a2a_message(
     event: Event, invocation_context: InvocationContext, role: Role = Role.agent
 ) -> Optional[Message]:
@@ -545,7 +544,7 @@ def _create_status_update_event(
   )
 
 
-@working_in_progress
+@experimental
 def convert_event_to_a2a_events(
     event: Event,
     invocation_context: InvocationContext,

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

@@ -21,7 +21,6 @@ from __future__ import annotations
 import base64
 import json
 import logging
-import sys
 from typing import Optional
 
 from .utils import _get_adk_metadata_key
@@ -29,17 +28,18 @@ from .utils import _get_adk_metadata_key
 try:
   from a2a import types as a2a_types
 except ImportError as e:
+  import sys
+
   if sys.version_info < (3, 10):
     raise ImportError(
-        'A2A Tool requires Python 3.10 or above. Please upgrade your Python'
-        ' version.'
+        'A2A requires Python 3.10 or above. Please upgrade your Python version.'
     ) from e
   else:
     raise e
 
 from google.genai import types as genai_types
 
-from ...utils.feature_decorator import working_in_progress
+from ...utils.feature_decorator import experimental
 
 logger = logging.getLogger('google_adk.' + __name__)
 
@@ -51,7 +51,7 @@ A2A_DATA_PART_METADATA_TYPE_CODE_EXECUTION_RESULT = 'code_execution_result'
 A2A_DATA_PART_METADATA_TYPE_EXECUTABLE_CODE = 'executable_code'
 
 
-@working_in_progress
+@experimental
 def convert_a2a_part_to_genai_part(
     a2a_part: a2a_types.Part,
 ) -> Optional[genai_types.Part]:
@@ -140,7 +140,7 @@ def convert_a2a_part_to_genai_part(
   return None
 
 
-@working_in_progress
+@experimental
 def convert_genai_part_to_a2a_part(
     part: genai_types.Part,
 ) -> Optional[a2a_types.Part]:

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

@@ -31,42 +31,24 @@ except ImportError as e:
 from google.genai import types as genai_types
 
 from ...runners import RunConfig
-from ...utils.feature_decorator import working_in_progress
+from ...utils.feature_decorator import experimental
 from .part_converter import convert_a2a_part_to_genai_part
-from .utils import _from_a2a_context_id
-from .utils import _get_adk_metadata_key
 
 
-def _get_user_id(request: RequestContext, user_id_from_context: str) -> str:
+def _get_user_id(request: RequestContext) -> str:
   # Get user from call context if available (auth is enabled on a2a server)
-  if request.call_context and request.call_context.user:
+  if (
+      request.call_context
+      and request.call_context.user
+      and request.call_context.user.user_name
+  ):
     return request.call_context.user.user_name
 
-  # Get user from context id if available
-  if user_id_from_context:
-    return user_id_from_context
-
-  # Get user from message metadata if available (client is an ADK agent)
-  if request.message.metadata:
-    user_id = request.message.metadata.get(_get_adk_metadata_key('user_id'))
-    if user_id:
-      return f'ADK_USER_{user_id}'
-
-  # Get user from task if available (client is a an ADK agent)
-  if request.current_task:
-    user_id = request.current_task.metadata.get(
-        _get_adk_metadata_key('user_id')
-    )
-    if user_id:
-      return f'ADK_USER_{user_id}'
-  return (
-      f'temp_user_{request.task_id}'
-      if request.task_id
-      else f'TEMP_USER_{request.message.messageId}'
-  )
+  # Get user from context id
+  return f'A2A_USER_{request.context_id}'
 
 
-@working_in_progress
+@experimental
 def convert_a2a_request_to_adk_run_args(
     request: RequestContext,
 ) -> dict[str, Any]:
@@ -74,11 +56,9 @@ def convert_a2a_request_to_adk_run_args(
   if not request.message:
     raise ValueError('Request message cannot be None')
 
-  _, user_id, session_id = _from_a2a_context_id(request.context_id)
-
   return {
-      'user_id': _get_user_id(request, user_id),
-      'session_id': session_id,
+      'user_id': _get_user_id(request),
+      'session_id': request.context_id,
       'new_message': genai_types.Content(
           role='user',
           parts=[

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

@@ -0,0 +1,13 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

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

@@ -0,0 +1,260 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from datetime import datetime
+from datetime import timezone
+import inspect
+import logging
+from typing import Any
+from typing import Awaitable
+from typing import Callable
+from typing import Optional
+import uuid
+
+try:
+  from a2a.server.agent_execution import AgentExecutor
+  from a2a.server.agent_execution.context import RequestContext
+  from a2a.server.events.event_queue import EventQueue
+  from a2a.types import Message
+  from a2a.types import Role
+  from a2a.types import TaskState
+  from a2a.types import TaskStatus
+  from a2a.types import TaskStatusUpdateEvent
+  from a2a.types import TextPart
+
+except ImportError as e:
+  import sys
+
+  if sys.version_info < (3, 10):
+    raise ImportError(
+        'A2A requires Python 3.10 or above. Please upgrade your Python version.'
+    ) from e
+  else:
+    raise e
+from google.adk.runners import Runner
+from pydantic import BaseModel
+from typing_extensions import override
+
+from ...utils.feature_decorator import experimental
+from ..converters.event_converter import convert_event_to_a2a_events
+from ..converters.request_converter import convert_a2a_request_to_adk_run_args
+from ..converters.utils import _get_adk_metadata_key
+from .task_result_aggregator import TaskResultAggregator
+
+logger = logging.getLogger('google_adk.' + __name__)
+
+
+@experimental
+class A2aAgentExecutorConfig(BaseModel):
+  """Configuration for the A2aAgentExecutor."""
+
+  pass
+
+
+@experimental
+class A2aAgentExecutor(AgentExecutor):
+  """An AgentExecutor that runs an ADK Agent against an A2A request and
+  publishes updates to an event queue.
+  """
+
+  def __init__(
+      self,
+      *,
+      runner: Runner | Callable[..., Runner | Awaitable[Runner]],
+      config: Optional[A2aAgentExecutorConfig] = None,
+  ):
+    super().__init__()
+    self._runner = runner
+    self._config = config
+
+  async def _resolve_runner(self) -> Runner:
+    """Resolve the runner, handling cases where it's a callable that returns a Runner."""
+    # If already resolved and cached, return it
+    if isinstance(self._runner, Runner):
+      return self._runner
+    if callable(self._runner):
+      # Call the function to get the runner
+      result = self._runner()
+
+      # Handle async callables
+      if inspect.iscoroutine(result):
+        resolved_runner = await result
+      else:
+        resolved_runner = result
+
+      # Cache the resolved runner for future calls
+      self._runner = resolved_runner
+      return resolved_runner
+
+    raise TypeError(
+        'Runner must be a Runner instance or a callable that returns a'
+        f' Runner, got {type(self._runner)}'
+    )
+
+  @override
+  async def cancel(self, context: RequestContext, event_queue: EventQueue):
+    """Cancel the execution."""
+    # TODO: Implement proper cancellation logic if needed
+    raise NotImplementedError('Cancellation is not supported')
+
+  @override
+  async def execute(
+      self,
+      context: RequestContext,
+      event_queue: EventQueue,
+  ):
+    """Executes an A2A request and publishes updates to the event queue
+    specified. It runs as following:
+    * Takes the input from the A2A request
+    * Convert the input to ADK input content, and runs the ADK agent
+    * Collects output events of the underlying ADK Agent
+    * Converts the ADK output events into A2A task updates
+    * Publishes the updates back to A2A server via event queue
+    """
+    if not context.message:
+      raise ValueError('A2A request must have a message')
+
+    # for new task, create a task submitted event
+    if not context.current_task:
+      await event_queue.enqueue_event(
+          TaskStatusUpdateEvent(
+              taskId=context.task_id,
+              status=TaskStatus(
+                  state=TaskState.submitted,
+                  message=context.message,
+                  timestamp=datetime.now(timezone.utc).isoformat(),
+              ),
+              contextId=context.context_id,
+              final=False,
+          )
+      )
+
+    # Handle the request and publish updates to the event queue
+    try:
+      await self._handle_request(context, event_queue)
+    except Exception as e:
+      logger.error('Error handling A2A request: %s', e, exc_info=True)
+      # Publish failure event
+      try:
+        await event_queue.enqueue_event(
+            TaskStatusUpdateEvent(
+                taskId=context.task_id,
+                status=TaskStatus(
+                    state=TaskState.failed,
+                    timestamp=datetime.now(timezone.utc).isoformat(),
+                    message=Message(
+                        messageId=str(uuid.uuid4()),
+                        role=Role.agent,
+                        parts=[TextPart(text=str(e))],
+                    ),
+                ),
+                contextId=context.context_id,
+                final=True,
+            )
+        )
+      except Exception as enqueue_error:
+        logger.error(
+            'Failed to publish failure event: %s', enqueue_error, exc_info=True
+        )
+
+  async def _handle_request(
+      self,
+      context: RequestContext,
+      event_queue: EventQueue,
+  ):
+    # Resolve the runner instance
+    runner = await self._resolve_runner()
+
+    # Convert the a2a request to ADK run args
+    run_args = convert_a2a_request_to_adk_run_args(context)
+
+    # ensure the session exists
+    session = await self._prepare_session(context, run_args, runner)
+
+    # create invocation context
+    invocation_context = runner._new_invocation_context(
+        session=session,
+        new_message=run_args['new_message'],
+        run_config=run_args['run_config'],
+    )
+
+    # publish the task working event
+    await event_queue.enqueue_event(
+        TaskStatusUpdateEvent(
+            taskId=context.task_id,
+            status=TaskStatus(
+                state=TaskState.working,
+                timestamp=datetime.now(timezone.utc).isoformat(),
+            ),
+            contextId=context.context_id,
+            final=False,
+            metadata={
+                _get_adk_metadata_key('app_name'): runner.app_name,
+                _get_adk_metadata_key('user_id'): run_args['user_id'],
+                _get_adk_metadata_key('session_id'): run_args['session_id'],
+            },
+        )
+    )
+
+    task_result_aggregator = TaskResultAggregator()
+    async for adk_event in runner.run_async(**run_args):
+      for a2a_event in convert_event_to_a2a_events(
+          adk_event, invocation_context, context.task_id, context.context_id
+      ):
+        task_result_aggregator.process_event(a2a_event)
+        await event_queue.enqueue_event(a2a_event)
+
+    # publish the task result event - this is final
+    await event_queue.enqueue_event(
+        TaskStatusUpdateEvent(
+            taskId=context.task_id,
+            status=TaskStatus(
+                state=(
+                    task_result_aggregator.task_state
+                    if task_result_aggregator.task_state != TaskState.working
+                    else TaskState.completed
+                ),
+                timestamp=datetime.now(timezone.utc).isoformat(),
+                message=task_result_aggregator.task_status_message,
+            ),
+            contextId=context.context_id,
+            final=True,
+        )
+    )
+
+  async def _prepare_session(
+      self, context: RequestContext, run_args: dict[str, Any], runner: Runner
+  ):
+
+    session_id = run_args['session_id']
+    # create a new session if not exists
+    user_id = run_args['user_id']
+    session = await runner.session_service.get_session(
+        app_name=runner.app_name,
+        user_id=user_id,
+        session_id=session_id,
+    )
+    if session is None:
+      session = await runner.session_service.create_session(
+          app_name=runner.app_name,
+          user_id=user_id,
+          state={},
+          session_id=session_id,
+      )
+      # Update run_args with the new session_id
+      run_args['session_id'] = session.id
+
+    return session

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

@@ -0,0 +1,71 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from a2a.server.events import Event
+from a2a.types import Message
+from a2a.types import TaskState
+from a2a.types import TaskStatusUpdateEvent
+
+from ...utils.feature_decorator import experimental
+
+
+@experimental
+class TaskResultAggregator:
+  """Aggregates the task status updates and provides the final task state."""
+
+  def __init__(self):
+    self._task_state = TaskState.working
+    self._task_status_message = None
+
+  def process_event(self, event: Event):
+    """Process an event from the agent run and detect signals about the task status.
+    Priority of task state:
+    - failed
+    - auth_required
+    - input_required
+    - working
+    """
+    if isinstance(event, TaskStatusUpdateEvent):
+      if event.status.state == TaskState.failed:
+        self._task_state = TaskState.failed
+        self._task_status_message = event.status.message
+      elif (
+          event.status.state == TaskState.auth_required
+          and self._task_state != TaskState.failed
+      ):
+        self._task_state = TaskState.auth_required
+        self._task_status_message = event.status.message
+      elif (
+          event.status.state == TaskState.input_required
+          and self._task_state
+          not in (TaskState.failed, TaskState.auth_required)
+      ):
+        self._task_state = TaskState.input_required
+        self._task_status_message = event.status.message
+      # final state is already recorded and make sure the intermediate state is
+      # always working because other state may terminate the event aggregation
+      # in a2a request handler
+      elif self._task_state == TaskState.working:
+        self._task_status_message = event.status.message
+      event.status.state = TaskState.working
+
+  @property
+  def task_state(self) -> TaskState:
+    return self._task_state
+
+  @property
+  def task_status_message(self) -> Message | None:
+    return self._task_status_message

src/google/adk/a2a/logs/__init__.py

@@ -0,0 +1,13 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

src/google/adk/a2a/logs/log_utils.py

@@ -0,0 +1,349 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Utility functions for structured A2A request and response logging."""
+
+from __future__ import annotations
+
+import json
+import sys
+
+try:
+  from a2a.types import DataPart as A2ADataPart
+  from a2a.types import Message as A2AMessage
+  from a2a.types import Part as A2APart
+  from a2a.types import SendMessageRequest
+  from a2a.types import SendMessageResponse
+  from a2a.types import Task as A2ATask
+  from a2a.types import TextPart as A2ATextPart
+except ImportError as e:
+  if sys.version_info < (3, 10):
+    raise ImportError(
+        "A2A Tool requires Python 3.10 or above. Please upgrade your Python"
+        " version."
+    ) from e
+  else:
+    raise e
+
+
+# Constants
+_NEW_LINE = "\n"
+_EXCLUDED_PART_FIELD = {"file": {"bytes"}}
+
+
+def _is_a2a_task(obj) -> bool:
+  """Check if an object is an A2A Task, with fallback for isinstance issues."""
+  try:
+    return isinstance(obj, A2ATask)
+  except (TypeError, AttributeError):
+    return type(obj).__name__ == "Task" and hasattr(obj, "status")
+
+
+def _is_a2a_message(obj) -> bool:
+  """Check if an object is an A2A Message, with fallback for isinstance issues."""
+  try:
+    return isinstance(obj, A2AMessage)
+  except (TypeError, AttributeError):
+    return type(obj).__name__ == "Message" and hasattr(obj, "role")
+
+
+def _is_a2a_text_part(obj) -> bool:
+  """Check if an object is an A2A TextPart, with fallback for isinstance issues."""
+  try:
+    return isinstance(obj, A2ATextPart)
+  except (TypeError, AttributeError):
+    return type(obj).__name__ == "TextPart" and hasattr(obj, "text")
+
+
+def _is_a2a_data_part(obj) -> bool:
+  """Check if an object is an A2A DataPart, with fallback for isinstance issues."""
+  try:
+    return isinstance(obj, A2ADataPart)
+  except (TypeError, AttributeError):
+    return type(obj).__name__ == "DataPart" and hasattr(obj, "data")
+
+
+def build_message_part_log(part: A2APart) -> str:
+  """Builds a log representation of an A2A message part.
+
+  Args:
+    part: The A2A message part to log.
+
+  Returns:
+    A string representation of the part.
+  """
+  part_content = ""
+  if _is_a2a_text_part(part.root):
+    part_content = f"TextPart: {part.root.text[:100]}" + (
+        "..." if len(part.root.text) > 100 else ""
+    )
+  elif _is_a2a_data_part(part.root):
+    # For data parts, show the data keys but exclude large values
+    data_summary = {
+        k: (
+            f"<{type(v).__name__}>"
+            if isinstance(v, (dict, list)) and len(str(v)) > 100
+            else v
+        )
+        for k, v in part.root.data.items()
+    }
+    part_content = f"DataPart: {json.dumps(data_summary, indent=2)}"
+  else:
+    part_content = (
+        f"{type(part.root).__name__}:"
+        f" {part.model_dump_json(exclude_none=True, exclude=_EXCLUDED_PART_FIELD)}"
+    )
+
+  # Add part metadata if it exists
+  if hasattr(part.root, "metadata") and part.root.metadata:
+    metadata_str = json.dumps(part.root.metadata, indent=2).replace(
+        "\n", "\n    "
+    )
+    part_content += f"\n    Part Metadata: {metadata_str}"
+
+  return part_content
+
+
+def build_a2a_request_log(req: SendMessageRequest) -> str:
+  """Builds a structured log representation of an A2A request.
+
+  Args:
+    req: The A2A SendMessageRequest to log.
+
+  Returns:
+    A formatted string representation of the request.
+  """
+  # Message parts logs
+  message_parts_logs = []
+  if req.params.message.parts:
+    for i, part in enumerate(req.params.message.parts):
+      part_log = build_message_part_log(part)
+      # Replace any internal newlines with indented newlines to maintain formatting
+      part_log_formatted = part_log.replace("\n", "\n  ")
+      message_parts_logs.append(f"Part {i}: {part_log_formatted}")
+
+  # Configuration logs
+  config_log = "None"
+  if req.params.configuration:
+    config_data = {
+        "acceptedOutputModes": req.params.configuration.acceptedOutputModes,
+        "blocking": req.params.configuration.blocking,
+        "historyLength": req.params.configuration.historyLength,
+        "pushNotificationConfig": bool(
+            req.params.configuration.pushNotificationConfig
+        ),
+    }
+    config_log = json.dumps(config_data, indent=2)
+
+  # Build message metadata section
+  message_metadata_section = ""
+  if req.params.message.metadata:
+    message_metadata_section = f"""
+  Metadata:
+  {json.dumps(req.params.message.metadata, indent=2).replace(chr(10), chr(10) + '  ')}"""
+
+  # Build optional sections
+  optional_sections = []
+
+  if req.params.metadata:
+    optional_sections.append(
+        f"""-----------------------------------------------------------
+Metadata:
+{json.dumps(req.params.metadata, indent=2)}"""
+    )
+
+  optional_sections_str = _NEW_LINE.join(optional_sections)
+
+  return f"""
+A2A Request:
+-----------------------------------------------------------
+Request ID: {req.id}
+Method: {req.method}
+JSON-RPC: {req.jsonrpc}
+-----------------------------------------------------------
+Message:
+  ID: {req.params.message.messageId}
+  Role: {req.params.message.role}
+  Task ID: {req.params.message.taskId}
+  Context ID: {req.params.message.contextId}{message_metadata_section}
+-----------------------------------------------------------
+Message Parts:
+{_NEW_LINE.join(message_parts_logs) if message_parts_logs else "No parts"}
+-----------------------------------------------------------
+Configuration:
+{config_log}
+{optional_sections_str}
+-----------------------------------------------------------
+"""
+
+
+def build_a2a_response_log(resp: SendMessageResponse) -> str:
+  """Builds a structured log representation of an A2A response.
+
+  Args:
+    resp: The A2A SendMessageResponse to log.
+
+  Returns:
+    A formatted string representation of the response.
+  """
+  # Handle error responses
+  if hasattr(resp.root, "error"):
+    return f"""
+A2A Response:
+-----------------------------------------------------------
+Type: ERROR
+Error Code: {resp.root.error.code}
+Error Message: {resp.root.error.message}
+Error Data: {json.dumps(resp.root.error.data, indent=2) if resp.root.error.data else "None"}
+-----------------------------------------------------------
+Response ID: {resp.root.id}
+JSON-RPC: {resp.root.jsonrpc}
+-----------------------------------------------------------
+"""
+
+  # Handle success responses
+  result = resp.root.result
+  result_type = type(result).__name__
+
+  # Build result details based on type
+  result_details = []
+
+  if _is_a2a_task(result):
+    result_details.extend([
+        f"Task ID: {result.id}",
+        f"Context ID: {result.contextId}",
+        f"Status State: {result.status.state}",
+        f"Status Timestamp: {result.status.timestamp}",
+        f"History Length: {len(result.history) if result.history else 0}",
+        f"Artifacts Count: {len(result.artifacts) if result.artifacts else 0}",
+    ])
+
+    # Add task metadata if it exists
+    if result.metadata:
+      result_details.append("Task Metadata:")
+      metadata_formatted = json.dumps(result.metadata, indent=2).replace(
+          "\n", "\n  "
+      )
+      result_details.append(f"  {metadata_formatted}")
+
+  elif _is_a2a_message(result):
+    result_details.extend([
+        f"Message ID: {result.messageId}",
+        f"Role: {result.role}",
+        f"Task ID: {result.taskId}",
+        f"Context ID: {result.contextId}",
+    ])
+
+    # Add message parts
+    if result.parts:
+      result_details.append("Message Parts:")
+      for i, part in enumerate(result.parts):
+        part_log = build_message_part_log(part)
+        # Replace any internal newlines with indented newlines to maintain formatting
+        part_log_formatted = part_log.replace("\n", "\n    ")
+        result_details.append(f"  Part {i}: {part_log_formatted}")
+
+    # Add metadata if it exists
+    if result.metadata:
+      result_details.append("Metadata:")
+      metadata_formatted = json.dumps(result.metadata, indent=2).replace(
+          "\n", "\n  "
+      )
+      result_details.append(f"  {metadata_formatted}")
+
+  else:
+    # Handle other result types by showing their JSON representation
+    if hasattr(result, "model_dump_json"):
+      try:
+        result_json = result.model_dump_json()
+        result_details.append(f"JSON Data: {result_json}")
+      except Exception:
+        result_details.append("JSON Data: <unable to serialize>")
+
+  # Build status message section
+  status_message_section = "None"
+  if _is_a2a_task(result) and result.status.message:
+    status_parts_logs = []
+    if result.status.message.parts:
+      for i, part in enumerate(result.status.message.parts):
+        part_log = build_message_part_log(part)
+        # Replace any internal newlines with indented newlines to maintain formatting
+        part_log_formatted = part_log.replace("\n", "\n  ")
+        status_parts_logs.append(f"Part {i}: {part_log_formatted}")
+
+    # Build status message metadata section
+    status_metadata_section = ""
+    if result.status.message.metadata:
+      status_metadata_section = f"""
+Metadata:
+{json.dumps(result.status.message.metadata, indent=2)}"""
+
+    status_message_section = f"""ID: {result.status.message.messageId}
+Role: {result.status.message.role}
+Task ID: {result.status.message.taskId}
+Context ID: {result.status.message.contextId}
+Message Parts:
+{_NEW_LINE.join(status_parts_logs) if status_parts_logs else "No parts"}{status_metadata_section}"""
+
+  # Build history section
+  history_section = "No history"
+  if _is_a2a_task(result) and result.history:
+    history_logs = []
+    for i, message in enumerate(result.history):
+      message_parts_logs = []
+      if message.parts:
+        for j, part in enumerate(message.parts):
+          part_log = build_message_part_log(part)
+          # Replace any internal newlines with indented newlines to maintain formatting
+          part_log_formatted = part_log.replace("\n", "\n    ")
+          message_parts_logs.append(f"  Part {j}: {part_log_formatted}")
+
+      # Build message metadata section
+      message_metadata_section = ""
+      if message.metadata:
+        message_metadata_section = f"""
+  Metadata:
+  {json.dumps(message.metadata, indent=2).replace(chr(10), chr(10) + '  ')}"""
+
+      history_logs.append(
+          f"""Message {i + 1}:
+  ID: {message.messageId}
+  Role: {message.role}
+  Task ID: {message.taskId}
+  Context ID: {message.contextId}
+  Message Parts:
+{_NEW_LINE.join(message_parts_logs) if message_parts_logs else "  No parts"}{message_metadata_section}"""
+      )
+
+    history_section = _NEW_LINE.join(history_logs)
+
+  return f"""
+A2A Response:
+-----------------------------------------------------------
+Type: SUCCESS
+Result Type: {result_type}
+-----------------------------------------------------------
+Result Details:
+{_NEW_LINE.join(result_details)}
+-----------------------------------------------------------
+Status Message:
+{status_message_section}
+-----------------------------------------------------------
+History:
+{history_section}
+-----------------------------------------------------------
+Response ID: {resp.root.id}
+JSON-RPC: {resp.root.jsonrpc}
+-----------------------------------------------------------
+"""

src/google/adk/agents/llm_agent.py

@@ -431,16 +431,31 @@ class LlmAgent(BaseAgent):
 
   def __maybe_save_output_to_state(self, event: Event):
     """Saves the model output to state if needed."""
+    # skip if the event was authored by some other agent (e.g. current agent
+    # transferred to another agent)
+    if event.author != self.name:
+      logger.debug(
+          'Skipping output save for agent %s: event authored by %s',
+          self.name,
+          event.author,
+      )
+      return
     if (
         self.output_key
         and event.is_final_response()
         and event.content
         and event.content.parts
     ):
+
       result = ''.join(
           [part.text if part.text else '' for part in event.content.parts]
       )
       if self.output_schema:
+        # If the result from the final chunk is just whitespace or empty,
+        # it means this is an empty final chunk of a stream.
+        # Do not attempt to parse it as JSON.
+        if not result.strip():
+          return
         result = self.output_schema.model_validate_json(result).model_dump(
             exclude_none=True
         )

src/google/adk/cli/browser/chunk-EQDQRRRY.js

@@ -0,0 +1 @@
+var p=Object.create;var j=Object.defineProperty,q=Object.defineProperties,r=Object.getOwnPropertyDescriptor,s=Object.getOwnPropertyDescriptors,t=Object.getOwnPropertyNames,g=Object.getOwnPropertySymbols,u=Object.getPrototypeOf,k=Object.prototype.hasOwnProperty,m=Object.prototype.propertyIsEnumerable;var l=(a,b,c)=>b in a?j(a,b,{enumerable:!0,configurable:!0,writable:!0,value:c}):a[b]=c,w=(a,b)=>{for(var c in b||={})k.call(b,c)&&l(a,c,b[c]);if(g)for(var c of g(b))m.call(b,c)&&l(a,c,b[c]);return a},x=(a,b)=>q(a,s(b));var y=(a,b)=>{var c={};for(var d in a)k.call(a,d)&&b.indexOf(d)<0&&(c[d]=a[d]);if(a!=null&&g)for(var d of g(a))b.indexOf(d)<0&&m.call(a,d)&&(c[d]=a[d]);return c};var z=(a,b)=>()=>(b||a((b={exports:{}}).exports,b),b.exports);var v=(a,b,c,d)=>{if(b&&typeof b=="object"||typeof b=="function")for(let e of t(b))!k.call(a,e)&&e!==c&&j(a,e,{get:()=>b[e],enumerable:!(d=r(b,e))||d.enumerable});return a};var A=(a,b,c)=>(c=a!=null?p(u(a)):{},v(b||!a||!a.__esModule?j(c,"default",{value:a,enumerable:!0}):c,a));var B=(a,b,c)=>new Promise((d,e)=>{var n=f=>{try{h(c.next(f))}catch(i){e(i)}},o=f=>{try{h(c.throw(f))}catch(i){e(i)}},h=f=>f.done?d(f.value):Promise.resolve(f.value).then(n,o);h((c=c.apply(a,b)).next())});export{w as a,x as b,y as c,z as d,A as e,B as f};

src/google/adk/cli/cli_deploy.py

@@ -55,7 +55,7 @@ COPY "agents/{app_name}/" "/app/agents/{app_name}/"
 
 EXPOSE {port}
 
-CMD adk {command} --port={port} {host_option} {service_option} {trace_to_cloud_option} {allow_origins_option} "/app/agents"
+CMD adk {command} --port={port} {host_option} {service_option} {trace_to_cloud_option} {allow_origins_option} {a2a_option}"/app/agents"
 """
 
 _AGENT_ENGINE_APP_TEMPLATE = """
@@ -128,6 +128,7 @@ def to_cloud_run(
     session_service_uri: Optional[str] = None,
     artifact_service_uri: Optional[str] = None,
     memory_service_uri: Optional[str] = None,
+    a2a: bool = False,
 ):
   """Deploys an agent to Google Cloud Run.
 
@@ -189,6 +190,7 @@ def to_cloud_run(
     allow_origins_option = (
         f'--allow_origins={",".join(allow_origins)}' if allow_origins else ''
     )
+    a2a_option = '--a2a' if a2a else ''
     dockerfile_content = _DOCKERFILE_TEMPLATE.format(
         gcp_project_id=project,
         gcp_region=region,
@@ -206,6 +208,7 @@ def to_cloud_run(
         allow_origins_option=allow_origins_option,
         adk_version=adk_version,
         host_option=host_option,
+        a2a_option=a2a_option,
     )
     dockerfile_path = os.path.join(temp_folder, 'Dockerfile')
     os.makedirs(temp_folder, exist_ok=True)

src/google/adk/cli/cli_eval.py

@@ -26,6 +26,7 @@ import uuid
 
 from ..agents import Agent
 from ..artifacts.base_artifact_service import BaseArtifactService
+from ..evaluation.constants import MISSING_EVAL_DEPENDENCIES_MESSAGE
 from ..evaluation.eval_case import EvalCase
 from ..evaluation.eval_metrics import EvalMetric
 from ..evaluation.eval_metrics import EvalMetricResult
@@ -38,12 +39,9 @@ from ..sessions.base_session_service import BaseSessionService
 logger = logging.getLogger("google_adk." + __name__)
 
 
-MISSING_EVAL_DEPENDENCIES_MESSAGE = (
-    "Eval module is not installed, please install via `pip install"
-    " google-adk[eval]`."
-)
 TOOL_TRAJECTORY_SCORE_KEY = "tool_trajectory_avg_score"
 RESPONSE_MATCH_SCORE_KEY = "response_match_score"
+SAFETY_V1_KEY = "safety_v1"
 # This evaluation is not very stable.
 # This is always optional unless explicitly specified.
 RESPONSE_EVALUATION_SCORE_KEY = "response_evaluation_score"
@@ -150,7 +148,7 @@ async def run_evals(
     artifact_service: The artifact service to use during inferencing.
   """
   try:
-    from ..evaluation.agent_evaluator import EvaluationGenerator
+    from ..evaluation.evaluation_generator import EvaluationGenerator
   except ModuleNotFoundError as e:
     raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
 
@@ -252,7 +250,8 @@ async def run_evals(
           result = "❌ Failed"
 
         print(f"Result: {result}\n")
-
+      except ModuleNotFoundError as e:
+        raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
       except Exception:
         # Catching the general exception, so that we don't block other eval
         # cases.
@@ -262,6 +261,7 @@ async def run_evals(
 def _get_evaluator(eval_metric: EvalMetric) -> Evaluator:
   try:
     from ..evaluation.response_evaluator import ResponseEvaluator
+    from ..evaluation.safety_evaluator import SafetyEvaluatorV1
     from ..evaluation.trajectory_evaluator import TrajectoryEvaluator
   except ModuleNotFoundError as e:
     raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
@@ -274,5 +274,7 @@ def _get_evaluator(eval_metric: EvalMetric) -> Evaluator:
     return ResponseEvaluator(
         threshold=eval_metric.threshold, metric_name=eval_metric.metric_name
     )
+  elif eval_metric.metric_name == SAFETY_V1_KEY:
+    return SafetyEvaluatorV1(eval_metric)
 
   raise ValueError(f"Unsupported eval metric: {eval_metric}")

src/google/adk/cli/cli_tools_click.py

@@ -31,12 +31,12 @@ import uvicorn
 from . import cli_create
 from . import cli_deploy
 from .. import version
+from ..evaluation.constants import MISSING_EVAL_DEPENDENCIES_MESSAGE
 from ..evaluation.gcs_eval_set_results_manager import GcsEvalSetResultsManager
 from ..evaluation.gcs_eval_sets_manager import GcsEvalSetsManager
 from ..evaluation.local_eval_set_results_manager import LocalEvalSetResultsManager
 from ..sessions.in_memory_session_service import InMemorySessionService
 from .cli import run_cli
-from .cli_eval import MISSING_EVAL_DEPENDENCIES_MESSAGE
 from .fast_api import get_fast_api_app
 from .utils import envs
 from .utils import evals
@@ -576,6 +576,13 @@ def fast_api_common_options():
             " for Cloud Run."
         ),
     )
+    @click.option(
+        "--a2a",
+        is_flag=True,
+        show_default=True,
+        default=False,
+        help="Optional. Whether to enable A2A endpoint.",
+    )
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
       return func(*args, **kwargs)
@@ -617,6 +624,7 @@ def cli_web(
     memory_service_uri: Optional[str] = None,
     session_db_url: Optional[str] = None,  # Deprecated
     artifact_storage_uri: Optional[str] = None,  # Deprecated
+    a2a: bool = False,
 ):
   """Starts a FastAPI server with Web UI for agents.
 
@@ -663,6 +671,9 @@ def cli_web(
       web=True,
       trace_to_cloud=trace_to_cloud,
       lifespan=_lifespan,
+      a2a=a2a,
+      host=host,
+      port=port,
   )
   config = uvicorn.Config(
       app,
@@ -709,6 +720,7 @@ def cli_api_server(
     memory_service_uri: Optional[str] = None,
     session_db_url: Optional[str] = None,  # Deprecated
     artifact_storage_uri: Optional[str] = None,  # Deprecated
+    a2a: bool = False,
 ):
   """Starts a FastAPI server for agents.
 
@@ -733,6 +745,9 @@ def cli_api_server(
           allow_origins=allow_origins,
           web=False,
           trace_to_cloud=trace_to_cloud,
+          a2a=a2a,
+          host=host,
+          port=port,
       ),
       host=host,
       port=port,
@@ -816,15 +831,6 @@ def cli_api_server(
         " version in the dev environment)"
     ),
 )
-@click.option(
-    "--eval_storage_uri",
-    type=str,
-    help=(
-        "Optional. The evals storage URI to store agent evals,"
-        " supported URIs: gs://<bucket name>."
-    ),
-    default=None,
-)
 @adk_services_options()
 @deprecated_adk_services_options()
 @click.argument(
@@ -854,6 +860,7 @@ def cli_deploy_cloud_run(
     eval_storage_uri: Optional[str] = None,
     session_db_url: Optional[str] = None,  # Deprecated
     artifact_storage_uri: Optional[str] = None,  # Deprecated
+    a2a: bool = False,
 ):
   """Deploys an agent to Cloud Run.
 
@@ -884,6 +891,7 @@ def cli_deploy_cloud_run(
         session_service_uri=session_service_uri,
         artifact_service_uri=artifact_service_uri,
         memory_service_uri=memory_service_uri,
+        a2a=a2a,
     )
   except Exception as e:
     click.secho(f"Deploy failed: {e}", fg="red", err=True)

src/google/adk/cli/fast_api.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 
 import asyncio
 from contextlib import asynccontextmanager
+import json
 import logging
 import os
 from pathlib import Path
@@ -32,7 +33,6 @@ from fastapi import FastAPI
 from fastapi import HTTPException
 from fastapi import Query
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import FileResponse
 from fastapi.responses import RedirectResponse
 from fastapi.responses import StreamingResponse
 from fastapi.staticfiles import StaticFiles
@@ -53,7 +53,6 @@ from typing_extensions import override
 from ..agents import RunConfig
 from ..agents.live_request_queue import LiveRequest
 from ..agents.live_request_queue import LiveRequestQueue
-from ..agents.llm_agent import Agent
 from ..agents.run_config import StreamingMode
 from ..artifacts.gcs_artifact_service import GcsArtifactService
 from ..artifacts.in_memory_artifact_service import InMemoryArtifactService
@@ -65,8 +64,6 @@ from ..evaluation.eval_metrics import EvalMetric
 from ..evaluation.eval_metrics import EvalMetricResult
 from ..evaluation.eval_metrics import EvalMetricResultPerInvocation
 from ..evaluation.eval_result import EvalSetResult
-from ..evaluation.gcs_eval_set_results_manager import GcsEvalSetResultsManager
-from ..evaluation.gcs_eval_sets_manager import GcsEvalSetsManager
 from ..evaluation.local_eval_set_results_manager import LocalEvalSetResultsManager
 from ..evaluation.local_eval_sets_manager import LocalEvalSetsManager
 from ..events.event import Event
@@ -204,6 +201,9 @@ def get_fast_api_app(
     eval_storage_uri: Optional[str] = None,
     allow_origins: Optional[list[str]] = None,
     web: bool,
+    a2a: bool = False,
+    host: str = "127.0.0.1",
+    port: int = 8000,
     trace_to_cloud: bool = False,
     lifespan: Optional[Lifespan[FastAPI]] = None,
 ) -> FastAPI:
@@ -962,6 +962,86 @@ def get_fast_api_app(
     runner_dict[app_name] = runner
     return runner
 
+  if a2a:
+    try:
+      from a2a.server.apps import A2AStarletteApplication
+      from a2a.server.request_handlers import DefaultRequestHandler
+      from a2a.server.tasks import InMemoryTaskStore
+      from a2a.types import AgentCard
+
+      from ..a2a.executor.a2a_agent_executor import A2aAgentExecutor
+
+    except ImportError as e:
+      import sys
+
+      if sys.version_info < (3, 10):
+        raise ImportError(
+            "A2A requires Python 3.10 or above. Please upgrade your Python"
+            " version."
+        ) from e
+      else:
+        raise e
+    # locate all a2a agent apps in the agents directory
+    base_path = Path.cwd() / agents_dir
+    # the root agents directory should be an existing folder
+    if base_path.exists() and base_path.is_dir():
+      a2a_task_store = InMemoryTaskStore()
+
+      def create_a2a_runner_loader(captured_app_name: str):
+        """Factory function to create A2A runner with proper closure."""
+
+        async def _get_a2a_runner_async() -> Runner:
+          return await _get_runner_async(captured_app_name)
+
+        return _get_a2a_runner_async
+
+      for p in base_path.iterdir():
+        # only folders with an agent.json file representing agent card are valid
+        # a2a agents
+        if (
+            p.is_file()
+            or p.name.startswith((".", "__pycache__"))
+            or not (p / "agent.json").is_file()
+        ):
+          continue
+
+        app_name = p.name
+        logger.info("Setting up A2A agent: %s", app_name)
+
+        try:
+          a2a_rpc_path = f"http://{host}:{port}/a2a/{app_name}"
+
+          agent_executor = A2aAgentExecutor(
+              runner=create_a2a_runner_loader(app_name),
+          )
+
+          request_handler = DefaultRequestHandler(
+              agent_executor=agent_executor, task_store=a2a_task_store
+          )
+
+          with (p / "agent.json").open("r", encoding="utf-8") as f:
+            data = json.load(f)
+            agent_card = AgentCard(**data)
+            agent_card.url = a2a_rpc_path
+
+          a2a_app = A2AStarletteApplication(
+              agent_card=agent_card,
+              http_handler=request_handler,
+          )
+
+          routes = a2a_app.routes(
+              rpc_url=f"/a2a/{app_name}",
+              agent_card_url=f"/a2a/{app_name}/.well-known/agent.json",
+          )
+
+          for new_route in routes:
+            app.router.routes.append(new_route)
+
+          logger.info("Successfully configured A2A agent: %s", app_name)
+
+        except Exception as e:
+          logger.error("Failed to setup A2A agent %s: %s", app_name, e)
+          # Continue with other agents even if one fails
   if web:
     import mimetypes
 

src/google/adk/evaluation/agent_evaluator.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from __future__ import annotations
+
 import json
 import logging
 import os
@@ -23,16 +25,17 @@ from typing import Optional
 from typing import Union
 import uuid
 
+from google.genai import types as genai_types
 from pydantic import ValidationError
 
+from .constants import MISSING_EVAL_DEPENDENCIES_MESSAGE
+from .eval_case import IntermediateData
+from .eval_metrics import EvalMetric
 from .eval_set import EvalSet
-from .evaluation_generator import EvaluationGenerator
 from .evaluator import EvalStatus
 from .evaluator import EvaluationResult
 from .evaluator import Evaluator
 from .local_eval_sets_manager import convert_eval_set_to_pydanctic_schema
-from .response_evaluator import ResponseEvaluator
-from .trajectory_evaluator import TrajectoryEvaluator
 
 logger = logging.getLogger("google_adk." + __name__)
 
@@ -44,11 +47,13 @@ TOOL_TRAJECTORY_SCORE_KEY = "tool_trajectory_avg_score"
 # This is always optional unless explicitly specified.
 RESPONSE_EVALUATION_SCORE_KEY = "response_evaluation_score"
 RESPONSE_MATCH_SCORE_KEY = "response_match_score"
+SAFETY_V1_KEY = "safety_v1"
 
 ALLOWED_CRITERIA = [
     TOOL_TRAJECTORY_SCORE_KEY,
     RESPONSE_EVALUATION_SCORE_KEY,
     RESPONSE_MATCH_SCORE_KEY,
+    SAFETY_V1_KEY,
 ]
 
 
@@ -96,6 +101,7 @@ class AgentEvaluator:
       criteria: dict[str, float],
       num_runs=NUM_RUNS,
       agent_name=None,
+      print_detailed_results: bool = True,
   ):
     """Evaluates an agent using the given EvalSet.
 
@@ -109,7 +115,13 @@ class AgentEvaluator:
       num_runs: Number of times all entries in the eval dataset should be
         assessed.
       agent_name: The name of the agent.
+      print_detailed_results: Whether to print detailed results for each metric
+        evaluation.
     """
+    try:
+      from .evaluation_generator import EvaluationGenerator
+    except ModuleNotFoundError as e:
+      raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
     eval_case_responses_list = await EvaluationGenerator.generate_responses(
         eval_set=eval_set,
         agent_module_path=agent_module,
@@ -117,6 +129,8 @@ class AgentEvaluator:
         agent_name=agent_name,
     )
 
+    failures = []
+
     for eval_case_responses in eval_case_responses_list:
       actual_invocations = [
           invocation
@@ -139,10 +153,25 @@ class AgentEvaluator:
             )
         )
 
-        assert evaluation_result.overall_eval_status == EvalStatus.PASSED, (
-            f"{metric_name} for {agent_module} Failed. Expected {threshold},"
-            f" but got {evaluation_result.overall_score}."
-        )
+        if print_detailed_results:
+          AgentEvaluator._print_details(
+              evaluation_result=evaluation_result,
+              metric_name=metric_name,
+              threshold=threshold,
+          )
+
+        # Gather all the failures.
+        if evaluation_result.overall_eval_status != EvalStatus.PASSED:
+          failures.append(
+              f"{metric_name} for {agent_module} Failed. Expected {threshold},"
+              f" but got {evaluation_result.overall_score}."
+          )
+
+    assert not failures, (
+        "Following are all the test failures. If you looking to get more"
+        " details on the failures, then please re-run this test with"
+        " `print_details` set to `True`.\n{}".format("\n".join(failures))
+    )
 
   @staticmethod
   async def evaluate(
@@ -158,9 +187,10 @@ class AgentEvaluator:
       agent_module: The path to python module that contains the definition of
         the agent. There is convention in place here, where the code is going to
         look for 'root_agent' in the loaded module.
-      eval_dataset_file_path_or_dir: The eval data set. This can be either a string representing
-        full path to the file containing eval dataset, or a directory that is
-        recursively explored for all files that have a `.test.json` suffix.
+      eval_dataset_file_path_or_dir: The eval data set. This can be either a
+        string representing full path to the file containing eval dataset, or a
+        directory that is recursively explored for all files that have a
+        `.test.json` suffix.
       num_runs: Number of times all entries in the eval dataset should be
         assessed.
       agent_name: The name of the agent.
@@ -358,6 +388,12 @@ class AgentEvaluator:
 
   @staticmethod
   def _get_metric_evaluator(metric_name: str, threshold: float) -> Evaluator:
+    try:
+      from .response_evaluator import ResponseEvaluator
+      from .safety_evaluator import SafetyEvaluatorV1
+      from .trajectory_evaluator import TrajectoryEvaluator
+    except ModuleNotFoundError as e:
+      raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
     if metric_name == TOOL_TRAJECTORY_SCORE_KEY:
       return TrajectoryEvaluator(threshold=threshold)
     elif (
@@ -365,5 +401,66 @@ class AgentEvaluator:
         or metric_name == RESPONSE_EVALUATION_SCORE_KEY
     ):
       return ResponseEvaluator(threshold=threshold, metric_name=metric_name)
+    elif metric_name == SAFETY_V1_KEY:
+      return SafetyEvaluatorV1(
+          eval_metric=EvalMetric(threshold=threshold, metric_name=metric_name)
+      )
 
     raise ValueError(f"Unsupported eval metric: {metric_name}")
+
+  @staticmethod
+  def _print_details(
+      evaluation_result: EvaluationResult, metric_name: str, threshold: float
+  ):
+    try:
+      from pandas import pandas as pd
+      from tabulate import tabulate
+    except ModuleNotFoundError as e:
+      raise ModuleNotFoundError(MISSING_EVAL_DEPENDENCIES_MESSAGE) from e
+    print(
+        f"Summary: `{evaluation_result.overall_eval_status}` for Metric:"
+        f" `{metric_name}`. Expected threshold: `{threshold}`, actual value:"
+        f" `{evaluation_result.overall_score}`."
+    )
+
+    data = []
+    for per_invocation_result in evaluation_result.per_invocation_results:
+      data.append({
+          "eval_status": per_invocation_result.eval_status,
+          "score": per_invocation_result.score,
+          "threshold": threshold,
+          "prompt": AgentEvaluator._convert_content_to_text(
+              per_invocation_result.expected_invocation.user_content
+          ),
+          "expected_response": AgentEvaluator._convert_content_to_text(
+              per_invocation_result.expected_invocation.final_response
+          ),
+          "actual_response": AgentEvaluator._convert_content_to_text(
+              per_invocation_result.actual_invocation.final_response
+          ),
+          "expected_tool_calls": AgentEvaluator._convert_tool_calls_to_text(
+              per_invocation_result.expected_invocation.intermediate_data
+          ),
+          "actual_tool_calls": AgentEvaluator._convert_tool_calls_to_text(
+              per_invocation_result.actual_invocation.intermediate_data
+          ),
+      })
+
+    print(tabulate(pd.DataFrame(data), headers="keys", tablefmt="grid"))
+    print("\n\n")  # Few empty lines for visual clarity
+
+  @staticmethod
+  def _convert_content_to_text(content: Optional[genai_types.Content]) -> str:
+    if content and content.parts:
+      return "\n".join([p.text for p in content.parts if p.text])
+
+    return ""
+
+  @staticmethod
+  def _convert_tool_calls_to_text(
+      intermediate_data: Optional[IntermediateData],
+  ) -> str:
+    if intermediate_data and intermediate_data.tool_uses:
+      return "\n".join([str(t) for t in intermediate_data.tool_uses])
+
+    return ""