refactor: use CallbackPipeline consistently in all callback execution sites

Address bot feedback (round 4) by replacing all manual callback iterations
with CallbackPipeline.execute() for consistency and maintainability.

Changes (9 locations):
1. base_agent.py: Use CallbackPipeline for before/after agent callbacks
2. callback_pipeline.py: Optimize single plugin callback execution
3. base_llm_flow.py: Use CallbackPipeline for before/after model callbacks
4. functions.py: Use CallbackPipeline for all tool callbacks (async + live)

Impact:
- Eliminates remaining manual callback iteration logic (~40 lines)
- Achieves 100% consistency in callback execution
- All sync/async handling and early exit logic centralized
- Tests: 24/24 passing
- Lint: 9.57/10 (improved from 9.49/10)

#non-breaking

Author: jaywang172

src/google/adk/agents/base_agent.py

@@ -45,6 +45,7 @@
 from ..utils.feature_decorator import experimental
 from .base_agent_config import BaseAgentConfig
 from .callback_context import CallbackContext
+from .callback_pipeline import CallbackPipeline
 from .callback_pipeline import normalize_callbacks
 
 if TYPE_CHECKING:
@@ -429,14 +430,10 @@ async def _handle_before_agent_callback(
     # callbacks.
     callbacks = normalize_callbacks(self.before_agent_callback)
     if not before_agent_callback_content and callbacks:
-      for callback in callbacks:
-        before_agent_callback_content = callback(
-            callback_context=callback_context
-        )
-        if inspect.isawaitable(before_agent_callback_content):
-          before_agent_callback_content = await before_agent_callback_content
-        if before_agent_callback_content:
-          break
+      pipeline = CallbackPipeline(callbacks)
+      before_agent_callback_content = await pipeline.execute(
+          callback_context=callback_context
+      )
 
     # Process the override content if exists, and further process the state
     # change if exists.
@@ -487,14 +484,10 @@ async def _handle_after_agent_callback(
     # callbacks.
     callbacks = normalize_callbacks(self.after_agent_callback)
     if not after_agent_callback_content and callbacks:
-      for callback in callbacks:
-        after_agent_callback_content = callback(
-            callback_context=callback_context
-        )
-        if inspect.isawaitable(after_agent_callback_content):
-          after_agent_callback_content = await after_agent_callback_content
-        if after_agent_callback_content:
-          break
+      pipeline = CallbackPipeline(callbacks)
+      after_agent_callback_content = await pipeline.execute(
+          callback_context=callback_context
+      )
 
     # Process the override content if exists, and further process the state
     # change if exists.

src/google/adk/agents/callback_pipeline.py

@@ -238,7 +238,9 @@ async def execute_with_plugins(
       ... )
     """
     # Step 1: Execute plugin callback (priority)
-    result = await CallbackPipeline([plugin_callback]).execute(*args, **kwargs)
+    result = plugin_callback(*args, **kwargs)
+    if inspect.isawaitable(result):
+      result = await result
     if result is not None:
       return result
 

src/google/adk/flows/llm_flows/base_llm_flow.py

@@ -32,6 +32,7 @@
 from . import functions
 from ...agents.base_agent import BaseAgent
 from ...agents.callback_context import CallbackContext
+from ...agents.callback_pipeline import CallbackPipeline
 from ...agents.callback_pipeline import normalize_callbacks
 from ...agents.invocation_context import InvocationContext
 from ...agents.live_request_queue import LiveRequestQueue
@@ -819,14 +820,12 @@ async def _handle_before_model_callback(
     callbacks = normalize_callbacks(agent.before_model_callback)
     if not callbacks:
       return
-    for callback in callbacks:
-      callback_response = callback(
-          callback_context=callback_context, llm_request=llm_request
-      )
-      if inspect.isawaitable(callback_response):
-        callback_response = await callback_response
-      if callback_response:
-        return callback_response
+    pipeline = CallbackPipeline(callbacks)
+    callback_response = await pipeline.execute(
+        callback_context=callback_context, llm_request=llm_request
+    )
+    if callback_response:
+      return callback_response
 
   async def _handle_after_model_callback(
       self,
@@ -877,14 +876,12 @@ async def _maybe_add_grounding_metadata(
     callbacks = normalize_callbacks(agent.after_model_callback)
     if not callbacks:
       return await _maybe_add_grounding_metadata()
-    for callback in callbacks:
-      callback_response = callback(
-          callback_context=callback_context, llm_response=llm_response
-      )
-      if inspect.isawaitable(callback_response):
-        callback_response = await callback_response
-      if callback_response:
-        return await _maybe_add_grounding_metadata(callback_response)
+    pipeline = CallbackPipeline(callbacks)
+    callback_response = await pipeline.execute(
+        callback_context=callback_context, llm_response=llm_response
+    )
+    if callback_response:
+      return await _maybe_add_grounding_metadata(callback_response)
     return await _maybe_add_grounding_metadata()
 
   def _finalize_model_response_event(

src/google/adk/flows/llm_flows/functions.py

@@ -31,6 +31,7 @@
 from google.genai import types
 
 from ...agents.active_streaming_tool import ActiveStreamingTool
+from ...agents.callback_pipeline import CallbackPipeline
 from ...agents.callback_pipeline import normalize_callbacks
 from ...agents.invocation_context import InvocationContext
 from ...auth.auth_tool import AuthToolArguments
@@ -318,14 +319,12 @@ async def _execute_single_function_call_async(
     # Step 2: If no overrides are provided from the plugins, further run the
     # canonical callback.
     if function_response is None:
-      for callback in normalize_callbacks(agent.before_tool_callback):
-        function_response = callback(
+      callbacks = normalize_callbacks(agent.before_tool_callback)
+      if callbacks:
+        pipeline = CallbackPipeline(callbacks)
+        function_response = await pipeline.execute(
             tool=tool, args=function_args, tool_context=tool_context
         )
-        if inspect.isawaitable(function_response):
-          function_response = await function_response
-        if function_response:
-          break
 
     # Step 3: Otherwise, proceed calling the tool normally.
     if function_response is None:
@@ -361,17 +360,15 @@ async def _execute_single_function_call_async(
     # Step 5: If no overrides are provided from the plugins, further run the
     # canonical after_tool_callbacks.
     if altered_function_response is None:
-      for callback in normalize_callbacks(agent.after_tool_callback):
-        altered_function_response = callback(
+      callbacks = normalize_callbacks(agent.after_tool_callback)
+      if callbacks:
+        pipeline = CallbackPipeline(callbacks)
+        altered_function_response = await pipeline.execute(
             tool=tool,
             args=function_args,
             tool_context=tool_context,
             tool_response=function_response,
         )
-        if inspect.isawaitable(altered_function_response):
-          altered_function_response = await altered_function_response
-        if altered_function_response:
-          break
 
     # Step 6: If alternative response exists from after_tool_callback, use it
     # instead of the original function response.
@@ -479,14 +476,12 @@ async def _execute_single_function_call_live(
 
     # Handle before_tool_callbacks - iterate through the canonical callback
     # list
-    for callback in normalize_callbacks(agent.before_tool_callback):
-      function_response = callback(
+    callbacks = normalize_callbacks(agent.before_tool_callback)
+    if callbacks:
+      pipeline = CallbackPipeline(callbacks)
+      function_response = await pipeline.execute(
           tool=tool, args=function_args, tool_context=tool_context
       )
-      if inspect.isawaitable(function_response):
-        function_response = await function_response
-      if function_response:
-        break
 
     if function_response is None:
       function_response = await _process_function_live_helper(
@@ -500,17 +495,15 @@ async def _execute_single_function_call_live(
 
     # Calls after_tool_callback if it exists.
     altered_function_response = None
-    for callback in normalize_callbacks(agent.after_tool_callback):
-      altered_function_response = callback(
+    callbacks = normalize_callbacks(agent.after_tool_callback)
+    if callbacks:
+      pipeline = CallbackPipeline(callbacks)
+      altered_function_response = await pipeline.execute(
           tool=tool,
           args=function_args,
           tool_context=tool_context,
           tool_response=function_response,
       )
-      if inspect.isawaitable(altered_function_response):
-        altered_function_response = await altered_function_response
-      if altered_function_response:
-        break
 
     if altered_function_response is not None:
       function_response = altered_function_response