feat(bidi): support cancellation from custom tools

site/src/content/docs/user-guide/sdk/bidirectional-streaming/agent.mdx

@@ -177,9 +177,11 @@ Each tool use stays adjacent to its result even when conversation continues or t
 
 Tool-result messages identify their purpose in `metadata["custom"]["bidi"]`: `kind` is `tool_dispatch` or `tool_result`, and `tool_use_id` links to the original request.
 
-The agent loop checks for `request_state["stop_event_loop"]` to trigger graceful
-shutdown instead of sending tool results back to the model. Any tool can set this
-flag to stop the conversation. The SDK's experimental `stop` tool uses this mechanism.
+To let a tool end the conversation, call `tool_context.agent.cancel()`.
+Cancellation takes effect only after a tool call completes. Requests from other
+contexts remain pending until then.
+
+See [Graceful shutdown](./quickstart.mdx#graceful-shutdown) for a custom tool example.
 
 ### Connection Lifecycle
 

site/src/content/docs/user-guide/sdk/bidirectional-streaming/events.mdx

@@ -225,8 +225,7 @@ Emitted when the streaming connection is closed.
   - `"timeout"`: Connection timed out
   - `"error"`: Error occurred
   - `"complete"`: Conversation completed normally
-  - `"user_request"`: User requested closure (via the SDK's experimental `stop`
-    tool or any tool that sets `request_state["stop_event_loop"]`)
+  - `"user_request"`: Cancellation requested through `agent.cancel()`
 
 ### Response Lifecycle Events
 

site/src/content/docs/user-guide/sdk/bidirectional-streaming/io.mdx

@@ -88,12 +88,10 @@ Pass your I/O streams to the agent's `run()` method to connect them to the agent
 import asyncio
 
 from strands.experimental.bidi.agent import BidiAgent
-from strands.experimental.tools import stop
 
 
 async def main():
-    # stop tool allows user to verbally stop agent execution.
-    agent = BidiAgent(tools=[stop])
+    agent = BidiAgent()
     await agent.run(inputs=[MyInputStream()], outputs=[MyOutputStream()])
 
 
@@ -104,6 +102,8 @@ The `run()` method handles startup, execution, and shutdown for the agent and it
 streams. Inputs and outputs run concurrently, so you can mix and match implementations.
 If an I/O task fails, `run()` cancels the remaining tasks, stops the streams, and
 re-raises the exception.
+For a tool that lets users end the conversation, see
+[Graceful shutdown](./quickstart.mdx#graceful-shutdown).
 
 ## Audio I/O
 
@@ -126,12 +126,10 @@ import asyncio
 
 from strands.experimental.bidi.agent import BidiAgent
 from strands.experimental.bidi.io import AudioIO
-from strands.experimental.tools import stop
 
 
 async def main():
-    # stop tool allows user to verbally stop agent execution.
-    agent = BidiAgent(tools=[stop])
+    agent = BidiAgent()
     audio_io = AudioIO(input_device_index=1)
 
     await agent.run(
@@ -232,7 +230,6 @@ import asyncio
 from strands.experimental.bidi.agent import BidiAgent
 from strands.experimental.bidi.io import ConsoleIO
 from strands.experimental.bidi.models import OpenAIRealtimeModel
-from strands.experimental.tools import stop
 
 
 async def main():
@@ -241,7 +238,7 @@ async def main():
         transcription_model_id=None,
         params={"output_modalities": ["text"]},
     )
-    agent = BidiAgent(model=model, tools=[stop])
+    agent = BidiAgent(model=model)
     console_io = ConsoleIO()
 
     await agent.run(
@@ -280,11 +277,10 @@ import asyncio
 
 from strands.experimental.bidi.agent import BidiAgent
 from strands.experimental.bidi.io import AudioIO, ConsoleIO
-from strands.experimental.tools import stop
 
 
 async def main():
-    agent = BidiAgent(tools=[stop])
+    agent = BidiAgent()
     console_io = ConsoleIO(placeholder="Type or speak…")
     audio_io = AudioIO(console=console_io)
 

site/src/content/docs/user-guide/sdk/bidirectional-streaming/models/bedrock.mdx

@@ -49,7 +49,6 @@ import asyncio
 from strands.experimental.bidi.agent import BidiAgent
 from strands.experimental.bidi.io import AudioIO
 from strands.experimental.bidi.models import BedrockNovaSonicModel
-from strands.experimental.tools import stop
 from strands.vended_tools import notebook
 
 
@@ -59,7 +58,7 @@ async def main() -> None:
         region="us-east-1",
         voice="tiffany",
     )
-    agent = BidiAgent(model=model, tools=[notebook, stop])
+    agent = BidiAgent(model=model, tools=[notebook])
     audio_io = AudioIO()
     await agent.run(inputs=[audio_io.input()], outputs=[audio_io.output()])
 

site/src/content/docs/user-guide/sdk/bidirectional-streaming/models/google.mdx

@@ -45,7 +45,6 @@ import asyncio
 from strands.experimental.bidi.agent import BidiAgent
 from strands.experimental.bidi.io import AudioIO
 from strands.experimental.bidi.models import GoogleGeminiLiveModel
-from strands.experimental.tools import stop
 from strands.vended_tools import notebook
 
 
@@ -55,8 +54,7 @@ async def main() -> None:
         voice="Kore",
         client_args={"api_key": "<GOOGLE_API_KEY>"},
     )
-    # stop tool allows user to verbally stop agent execution.
-    agent = BidiAgent(model=model, tools=[notebook, stop])
+    agent = BidiAgent(model=model, tools=[notebook])
 
     audio_io = AudioIO()
     await agent.run(inputs=[audio_io.input()], outputs=[audio_io.output()])

site/src/content/docs/user-guide/sdk/bidirectional-streaming/models/openai.mdx

@@ -44,7 +44,6 @@ import asyncio
 from strands.experimental.bidi.agent import BidiAgent
 from strands.experimental.bidi.io import AudioIO
 from strands.experimental.bidi.models import OpenAIRealtimeModel
-from strands.experimental.tools import stop
 from strands.vended_tools import notebook
 
 
@@ -55,8 +54,7 @@ async def main() -> None:
         voice="coral",
         api_key="<OPENAI_API_KEY>",
     )
-    # stop tool allows user to verbally stop agent execution.
-    agent = BidiAgent(model=model, tools=[notebook, stop])
+    agent = BidiAgent(model=model, tools=[notebook])
 
     audio_io = AudioIO()
     await agent.run(inputs=[audio_io.input()], outputs=[audio_io.output()])

site/src/content/docs/user-guide/sdk/bidirectional-streaming/quickstart.mdx

@@ -406,22 +406,28 @@ See [Controlling Conversation Lifecycle](#controlling-conversation-lifecycle) fo
 
 ## Graceful Shutdown
 
-Use the SDK's experimental `stop` tool to allow users to end conversations
-naturally. It sets `request_state["stop_event_loop"]`, which the agent loop checks
-to trigger a graceful shutdown:
+To let users end a conversation by voice, define a tool that calls `agent.cancel()`:
 
 ```python
 import asyncio
+from strands import LocalAgent, ToolContext, tool
 from strands.experimental.bidi.agent import BidiAgent
 from strands.experimental.bidi.io import AudioIO
 from strands.experimental.bidi.models import BedrockNovaSonicModel
-from strands.experimental.tools import stop
+
+
+@tool(context=True)
+def end_conversation(tool_context: ToolContext[LocalAgent]) -> str:
+    """End the conversation when the user asks to stop."""
+    tool_context.agent.cancel()
+    return "Ending conversation"
+
 
 model = BedrockNovaSonicModel(model_id="amazon.nova-2-sonic-v1:0")
 agent = BidiAgent(
     model=model,
-    tools=[stop],
-    system_prompt="You are a helpful assistant. When the user says 'stop conversation', use the stop tool."
+    tools=[end_conversation],
+    system_prompt="You are a helpful assistant.",
 )
 
 audio_io = AudioIO()
@@ -431,24 +437,13 @@ async def main():
         inputs=[audio_io.input()],
         outputs=[audio_io.output()]
     )
-    # Conversation ends when user says "stop conversation"
+    # run() returns after the agent calls end_conversation.
 
 asyncio.run(main())
 ```
 
-You can also create custom stop tools using the `request_state["stop_event_loop"]` flag:
-
-```python
-from strands import tool
-
-@tool
-def end_session(request_state: dict) -> str:
-    request_state["stop_event_loop"] = True
-    return "Goodbye!"
-```
-
-The agent will gracefully close the connection when any tool sets `request_state["stop_event_loop"] = True`.
-
+Bidi checks for cancellation after a tool call completes and its result is
+recorded. Requests from other contexts remain pending until that checkpoint.
 
 ## Debug Logs
 

strands-py/src/strands/experimental/bidi/__init__.py

@@ -1,7 +1,7 @@
 """Experimental bidirectional streaming APIs."""
 
-from . import agent, hooks, io, models, tools, types
+from . import agent, hooks, io, models, types
 from .agent import BidiAgent as BidiAgent
 
 # Compatibility for AgentCore's root import; remove after AgentCore imports from bidi.agent.
-__all__ = ["agent", "hooks", "io", "models", "tools", "types"]
+__all__ = ["agent", "hooks", "io", "models", "types"]

strands-py/src/strands/experimental/bidi/agent/agent.py

@@ -159,7 +159,6 @@ def __init__(
         self.messages = messages if messages is not None else []
         self._storage: Storage | None = storage
         self._sandbox: Sandbox = NotASandboxLocalEnvironment()
-        # Never set yet: bidirectional agents do not act on a cancellation signal.
         self._cancel_signal = threading.Event()
 
         # Agent identification
@@ -300,9 +299,20 @@ def event_loop_metrics(self) -> "EventLoopMetrics":
     def event_loop_metrics(self, value: "EventLoopMetrics") -> None:
         raise NotImplementedError("event_loop_metrics is not supported by bidirectional agents yet")
 
+    def cancel(self) -> None:
+        """Request cancellation of the current conversation.
+
+        This method is thread-safe and idempotent. Cancellation takes effect
+        only after a tool call completes.
+        """
+        self._cancel_signal.set()
+
     @property
     def cancel_signal(self) -> threading.Event:
-        """The cancellation signal; never set yet, because bidirectional agents do not act on it."""
+        """The cancellation signal for the current conversation.
+
+        Treat as read-only; call ``cancel()`` to request cancellation.
+        """
         return self._cancel_signal
 
     def add_hook(
@@ -456,7 +466,10 @@ async def stop(self) -> None:
         closes the connection to the model provider.
         """
         self._started = False
-        await self._loop.stop()
+        try:
+            await self._loop.stop()
+        finally:
+            self._cancel_signal.clear()
 
     def take_snapshot(
         self,

strands-py/src/strands/experimental/bidi/agent/loop.py

@@ -6,7 +6,6 @@
 import asyncio
 import logging
 import time
-import warnings
 from collections.abc import AsyncGenerator
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Literal, cast
@@ -197,6 +196,8 @@ async def start(self, invocation_state: dict[str, Any] | None = None) -> None:
         self._model_task = self._task_pool.create(self._run_model(self._generation))
 
         self._invocation_state = invocation_state if invocation_state is not None else {}
+        # Retained for compatibility with shared tools that expect request_state.
+        self._invocation_state.setdefault("request_state", {})
         self._send_gate.set()
         self._started = True
 
@@ -741,11 +742,6 @@ async def _run_tool(self, tool_use: ToolUse) -> None:
 
         tool_results: list[ToolResult] = []
 
-        # Ensure request_state exists for tools like strands_tools.stop
-        invocation_state = self._invocation_state
-        if "request_state" not in invocation_state:
-            invocation_state["request_state"] = {}
-
         tool_call_span = self._tracer.start_tool_call_span(tool_use, parent_span=self._session_span)
         tool_result: ToolResult | None = None
         tool_error: Exception | None = None
@@ -755,7 +751,7 @@ async def _run_tool(self, tool_use: ToolUse) -> None:
                 self._agent,
                 tool_use,
                 tool_results,
-                invocation_state,
+                self._invocation_state,
                 structured_output_context=None,
             )
 
@@ -767,7 +763,6 @@ async def _run_tool(self, tool_use: ToolUse) -> None:
 
                 await self._event_queue.put(tool_event)
 
-            # Normal flow for all tools (including stop_conversation)
             tool_result_event = cast(ToolResultEvent, tool_event)
             tool_result = tool_result_event.tool_result
 
@@ -786,25 +781,11 @@ async def _run_tool(self, tool_use: ToolUse) -> None:
 
             await self._event_queue.put(ToolResultMessageEvent(tool_result_message))
 
-            # Check for stop_event_loop flag (set by strands_tools.stop, stop_conversation, or any custom tool)
-            request_state = invocation_state.get("request_state", {})
-            should_stop = request_state.get("stop_event_loop", False)
-
-            # Backward compatibility: also check for stop_conversation by name (deprecated)
-            if not should_stop and tool_use["name"] == "stop_conversation":
-                warnings.warn(
-                    "Stopping the event loop by tool name 'stop_conversation' is deprecated. "
-                    "Use request_state['stop_event_loop'] = True instead.",
-                    DeprecationWarning,
-                    stacklevel=2,
-                )
-                should_stop = True
-
-            if should_stop:
-                logger.info("stop_event_loop=<True> | stopping conversation")
+            if self._agent.cancel_signal.is_set():
+                logger.info("tool_name=<%s> | stopping conversation", tool_use["name"])
                 connection_id = getattr(self._agent.model, "_connection_id", "unknown")
                 await self._event_queue.put(BidiConnectionStopEvent(connection_id=connection_id, reason="user_request"))
-                return  # Skip sending result to model
+                return
 
             # Send result to model
             await self.send(

strands-py/src/strands/tools/executors/_executor.py

@@ -58,7 +58,7 @@ def _should_retry(agent: LocalAgent, after_event: AfterToolCallEvent[LocalAgent]
         if cast(dict[str, Any], after_event.result).get("cancelled") is True:
             return False
         if not ToolExecutor._is_agent(agent):
-            return True
+            return not agent.cancel_signal.is_set()
         return not cast("Agent", agent)._observe_cancellation()
 
     async def _execute_background(