feat(termination): add pluggable termination conditions for multi-agent workflows

Add a new google.adk_community.termination module with pluggable termination
conditions for use with LoopAgent and Runner-based multi-agent workflows.

New termination conditions:
- MaxIterationsTermination: stops after N events
- TextMentionTermination: stops when a keyword appears in event content
- TimeoutTermination: stops after a wall-clock duration
- TokenUsageTermination: stops when token budgets are exceeded
- FunctionCallTermination: stops when a specific function is called
- ExternalTermination: stops on external signal (set() method)
- OrTerminationCondition / AndTerminationCondition composites (via | and &)

All conditions implement check(events), reset(), and the terminated property.
reset() is called automatically at the start of each run so the same instance
can be reused across invocations.

Related: google/adk-python#5211
Related: google/adk-python#5213
Cross-repo: google/adk-js#193

Author: davidmonterocrespo24

src/google/adk_community/__init__.py

@@ -14,5 +14,6 @@
 
 from . import memory
 from . import sessions
+from . import termination
 from . import version
 __version__ = version.__version__

src/google/adk_community/termination/__init__.py

@@ -0,0 +1,41 @@
+# 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.
+
+"""Community termination conditions for ADK multi-agent workflows."""
+
+from __future__ import annotations
+
+from .external_termination import ExternalTermination
+from .function_call_termination import FunctionCallTermination
+from .max_iterations_termination import MaxIterationsTermination
+from .termination_condition import AndTerminationCondition
+from .termination_condition import OrTerminationCondition
+from .termination_condition import TerminationCondition
+from .termination_condition import TerminationResult
+from .text_mention_termination import TextMentionTermination
+from .timeout_termination import TimeoutTermination
+from .token_usage_termination import TokenUsageTermination
+
+__all__ = [
+    'AndTerminationCondition',
+    'ExternalTermination',
+    'FunctionCallTermination',
+    'MaxIterationsTermination',
+    'OrTerminationCondition',
+    'TerminationCondition',
+    'TerminationResult',
+    'TextMentionTermination',
+    'TimeoutTermination',
+    'TokenUsageTermination',
+]

src/google/adk_community/termination/external_termination.py

@@ -0,0 +1,64 @@
+# 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.
+
+"""A termination condition controlled programmatically via ``set()``."""
+
+from __future__ import annotations
+
+from typing import Optional
+from typing import Sequence
+
+from google.adk.events.event import Event
+from .termination_condition import TerminationCondition
+from .termination_condition import TerminationResult
+
+
+class ExternalTermination(TerminationCondition):
+  """A termination condition that is controlled externally by calling ``set()``.
+
+  Useful for integrating external stop signals such as a UI "Stop" button
+  or application-level logic.
+
+  Example::
+
+      stop_button = ExternalTermination()
+
+      agent = LoopAgent(
+          name='my_loop',
+          sub_agents=[...],
+          termination_condition=stop_button,
+      )
+
+      # Elsewhere (e.g. from a UI event handler):
+      stop_button.set()
+  """
+
+  def __init__(self) -> None:
+    self._terminated = False
+
+  @property
+  def terminated(self) -> bool:
+    return self._terminated
+
+  def set(self) -> None:
+    """Signals that the conversation should terminate at the next check."""
+    self._terminated = True
+
+  async def check(self, events: Sequence[Event]) -> Optional[TerminationResult]:
+    if self._terminated:
+      return TerminationResult(reason='Externally terminated')
+    return None
+
+  async def reset(self) -> None:
+    self._terminated = False

src/google/adk_community/termination/function_call_termination.py

@@ -0,0 +1,60 @@
+# 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.
+
+"""Terminates when a specific function (tool) has been executed."""
+
+from __future__ import annotations
+
+from typing import Optional
+from typing import Sequence
+
+from google.adk.events.event import Event
+from .termination_condition import TerminationCondition
+from .termination_condition import TerminationResult
+
+
+class FunctionCallTermination(TerminationCondition):
+  """Terminates when a tool with a specific name has been executed.
+
+  The condition checks ``FunctionResponse`` parts in events.
+
+  Example::
+
+      # Stop when the "approve" tool is called
+      condition = FunctionCallTermination('approve')
+  """
+
+  def __init__(self, function_name: str) -> None:
+    self._function_name = function_name
+    self._terminated = False
+
+  @property
+  def terminated(self) -> bool:
+    return self._terminated
+
+  async def check(self, events: Sequence[Event]) -> Optional[TerminationResult]:
+    if self._terminated:
+      return None
+
+    for event in events:
+      for response in event.get_function_responses():
+        if response.name == self._function_name:
+          self._terminated = True
+          return TerminationResult(
+              reason=f"Function '{self._function_name}' was executed"
+          )
+    return None
+
+  async def reset(self) -> None:
+    self._terminated = False

src/google/adk_community/termination/max_iterations_termination.py

@@ -0,0 +1,64 @@
+# 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.
+
+"""Terminates after a maximum number of events have been processed."""
+
+from __future__ import annotations
+
+from typing import Optional
+from typing import Sequence
+
+from google.adk.events.event import Event
+from .termination_condition import TerminationCondition
+from .termination_condition import TerminationResult
+
+
+class MaxIterationsTermination(TerminationCondition):
+  """Terminates the conversation after a maximum number of events.
+
+  Example::
+
+      # Stop after 10 events
+      condition = MaxIterationsTermination(10)
+  """
+
+  def __init__(self, max_iterations: int) -> None:
+    if max_iterations <= 0:
+      raise ValueError('max_iterations must be a positive integer.')
+    self._max_iterations = max_iterations
+    self._count = 0
+    self._terminated = False
+
+  @property
+  def terminated(self) -> bool:
+    return self._terminated
+
+  async def check(self, events: Sequence[Event]) -> Optional[TerminationResult]:
+    if self._terminated:
+      return None
+    self._count += len(events)
+
+    if self._count >= self._max_iterations:
+      self._terminated = True
+      return TerminationResult(
+          reason=(
+              f'Maximum iterations of {self._max_iterations} reached,'
+              f' current count: {self._count}'
+          )
+      )
+    return None
+
+  async def reset(self) -> None:
+    self._terminated = False
+    self._count = 0