@@ -84,6 +84,27 @@ def increment_and_enforce_llm_calls_limit(
       )
 
 
+class _AbortState:
+  """Shared mutable state container for invocation abort signals.
+
+  Because Pydantic model_copy() shallow-copies __pydantic_private__, all
+  derived contexts within the same Runner share the exact same _AbortState
+  instance reference. Updates to loop, signal, or aborted propagate across all
+  model_copy() clones in the tree. Cross-Runner sub-runs (such as AgentTool or
+  nested Workflow node runners) propagate cancellation by passing
+  ``_abort_signal`` to the child Runner's ``run_async``.
+  """
+
+  def __init__(
+      self,
+      signal: asyncio.Event | None = None,
+      loop: asyncio.AbstractEventLoop | None = None,
+  ) -> None:
+    self.signal = signal if signal is not None else asyncio.Event()
+    self.loop = loop
+    self.aborted = False
+
+
 class InvocationContext(BaseModel):
   """An invocation context represents the data of a single invocation of an agent.
 
@@ -258,11 +279,56 @@ class InvocationContext(BaseModel):
   of this invocation.
   """
 
+  _abort_state: _AbortState = PrivateAttr(default_factory=_AbortState)
+  """Captured abort state (signal, loop, and aborted flag) shared across copies."""
+
   @override
   def model_post_init(self, __context: Any) -> None:
     super().model_post_init(__context)
     if self.run_config and self.run_config.custom_metadata:
       self._custom_metadata.update(self.run_config.custom_metadata)
+    try:
+      self._abort_state.loop = asyncio.get_running_loop()
+    except RuntimeError:
+      pass
+
+  @property
+  def _abort_signal(self) -> asyncio.Event:
+    """The internal abort signal event for this invocation.
+
+    Private on purpose so callers use ``is_aborted``, while internal runners use
+    ``_abort_signal`` to await cancellation. Sub-contexts are shallow
+    ``model_copy`` clones that share this exact Event instance via
+    ``_abort_state``.
+    """
+    if self._abort_state.loop is None:
+      try:
+        self._abort_state.loop = asyncio.get_running_loop()
+      except RuntimeError:
+        pass
+    return self._abort_state.signal
+
+  def _attach_abort_signal(self, abort_signal: asyncio.Event) -> None:
+    """Replaces this context's abort signal with a caller-owned event.
+
+    Only for the runner to call on a freshly built root context, before any
+    sub-context is derived from it. The runner cannot pass the signal to the
+    constructor because the context is produced by ``_new_invocation_context``,
+    an overridable factory whose subclass overrides do not accept the argument.
+
+    Because sub-contexts share ``_abort_state``, replacing the signal here
+    propagates to all derived contexts. Rebinding after execution begins is
+    unsafe as active tasks may already be awaiting the previous event.
+
+    Args:
+      abort_signal: The caller-owned event to abort this invocation with.
+    """
+    self._abort_state.signal = abort_signal
+    if self._abort_state.loop is None:
+      try:
+        self._abort_state.loop = asyncio.get_running_loop()
+      except RuntimeError:
+        pass
 
   @property
   def is_resumable(self) -> bool:
@@ -272,6 +338,44 @@ def is_resumable(self) -> bool:
         and self.resumability_config.is_resumable
     )
 
+  @property
+  def is_aborted(self) -> bool:
+    """Returns whether the current invocation has been requested to abort."""
+    return self._abort_state.aborted or self._abort_state.signal.is_set()
+
+  def abort(self) -> None:
+    """Trip the abort signal in a thread-safe manner.
+
+    Can be safely called from either the event loop thread or an external
+    worker thread. When called from within the running event loop, trips the
+    signal immediately on the same tick. When called from a foreign thread,
+    schedules the trip thread-safely onto the captured event loop.
+    """
+    self._abort_state.aborted = True
+    signal = self._abort_state.signal
+    if signal.is_set():
+      return
+
+    try:
+      running_loop = asyncio.get_running_loop()
+    except RuntimeError:
+      running_loop = None
+
+    target_loop = self._abort_state.loop
+    if target_loop is not None and running_loop is not target_loop:
+      try:
+        target_loop.call_soon_threadsafe(signal.set)
+      except RuntimeError:
+        try:
+          signal.set()
+        except RuntimeError:
+          pass
+    else:
+      try:
+        signal.set()
+      except RuntimeError:
+        pass
+
   async def _enqueue_event(self, event: Event) -> None:
     """Enqueue an event for the Runner main loop to process.
 
 
@@ -79,6 +79,11 @@ def custom_metadata(self) -> Mapping[str, Any]:
     # pylint: disable=protected-access
     return MappingProxyType(self._invocation_context._custom_metadata)
 
+  @property
+  def is_aborted(self) -> bool:
+    """Returns whether the current invocation has been requested to abort."""
+    return self._invocation_context.is_aborted
+
   def get_credential(self, key: str) -> AuthCredential | None:
     """Gets a resolved credential by key for this invocation."""
     return self._invocation_context.credential_by_key.get(key)
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 from unittest.mock import AsyncMock
 from unittest.mock import MagicMock
 from unittest.mock import patch
@@ -44,6 +45,7 @@ def mock_invocation_context():
   mock_context.artifact_service = None
   mock_context.credential_service = None
   mock_context.memory_service = None
+  mock_context.is_aborted = False
   return mock_context
 
 
@@ -55,6 +57,43 @@ def test_context_branch_returns_invocation_branch(mock_invocation_context):
   assert context.branch == "test-branch"
 
 
+def test_context_is_aborted(mock_invocation_context):
+  """Context.is_aborted delegates to invocation context."""
+  mock_invocation_context.is_aborted = False
+  context = Context(invocation_context=mock_invocation_context)
+  assert context.is_aborted is False
+
+  mock_invocation_context.is_aborted = True
+  assert context.is_aborted is True
+
+
+@pytest.mark.asyncio
+async def test_context_is_aborted_with_real_invocation_context():
+  """Context reflects underlying InvocationContext cancellation."""
+  from google.adk.agents.base_agent import BaseAgent
+  from google.adk.agents.invocation_context import InvocationContext
+  from google.adk.sessions.base_session_service import BaseSessionService
+  from google.adk.sessions.session import Session
+
+  abort_signal = asyncio.Event()
+  inv_ctx = InvocationContext(
+      session_service=MagicMock(spec=BaseSessionService),
+      agent=MagicMock(spec=BaseAgent),
+      invocation_id="inv_1",
+      session=Session(id="s1", app_name="test_app", user_id="test_user"),
+  )
+  inv_ctx._attach_abort_signal(abort_signal)
+  context = Context(invocation_context=inv_ctx)
+  assert context.is_aborted is False
+  assert inv_ctx.is_aborted is False
+
+  inv_ctx.abort()
+
+  assert context.is_aborted is True
+  assert inv_ctx.is_aborted is True
+  assert abort_signal.is_set() is True
+
+
 @pytest.fixture
 def mock_artifact_service():
   """Create a mock artifact service for testing."""