Cooperative Tool Cancellation

Stop a running shell/HTTP/MCP tool promptly on /stop or Ctrl-C — not just the next one

Cancellation now reaches the tool body that is already running, so a /stop or Ctrl-C aborts a live subprocess within ~100ms instead of waiting for its own timeout.

This is in-flight cancellation. It stops a tool that is already running. To stop the next tool call (deadline or token), see Tool Timeouts & Cancellation.

Quick Start

1. Just works from an agent

   Give the agent an InterruptController, then request a stop from another thread — a running shell tool body is aborted within ~100ms and its result carries interrupted: True.

   from praisonaiagents import Agent
   from praisonaiagents.agent import InterruptController
   
   controller = InterruptController()
   
   agent = Agent(
       name="Runner",
       instructions="Run shell commands when asked.",
       tools=["shell"],
       interrupt_controller=controller,
   )
   
   # From another thread / the CLI /stop / Ctrl-C handler:
   controller.request("user_stop")
   # A running `shell` tool body is aborted within ~100ms and the
   # tool result carries `interrupted: True`.
   
2. Custom tool opts into cooperative abort

   Read the event inside a custom tool via Injected[AgentState], then poll or wait on it to bail out promptly.

   from praisonaiagents import Agent
   from praisonaiagents.tools import tool, AgentState, Injected
   
   @tool
   def slow_download(url: str, state: Injected[AgentState]) -> str:
       cancel = state.cancel_event
       for chunk in stream(url):                 # your streaming source
           if cancel is not None and cancel.is_set():
               return "interrupted"
           process(chunk)                          # your per-chunk work
       return "done"
   
   agent = Agent(name="Downloader", tools=[slow_download])
   

---

How It Works

The controller exposes its underlying Event; the tool-execution loop threads it into AgentState.cancel_event, and the running tool polls it every ~100ms.

The event is scoped to the current turn: the controller clears its flag when an interrupted turn ends, so a stale request cannot abort tools launched by the next turn. Overlapping turns on the same Agent each live in their own ContextVar scope, so clearing one turn’s flag never clears another’s.

---

Reading the outcome

The built-in shell tool returns a discriminated interrupted payload — distinct from a timeout.

Field	Value	Meaning
stdout	''	No output captured on abort
stderr	'Command interrupted by user'	Human-readable reason
exit_code	-1	Process did not exit normally
success	False	The call did not complete
interrupted	True	Aborted by cooperative interrupt (not a timeout)
execution_time	<float>	Seconds until the abort

A wall-clock timeout instead returns stderr='Command timed out after N seconds' with no interrupted key — check result.get('interrupted') to tell them apart.

---

Choose your cancellation surface

Pick the surface by what you need to stop.

• Stop the next tool → cancel_token / timeout_ms on the executor. See Tool Timeouts & Cancellation.
• Stop the tool that is already running → Agent.interrupt_controller + cancel_event (this page).

---

Cancel during LLM retry backoff

Stop interrupts the wait between LLM retries, so a 429 backoff no longer runs its full sleep.

While waiting, the agent checks the cancel token about every 200 ms and raises InterruptedError("Agent chat cancelled: <reason>") once it is set. Both retry paths honour it. Without an explicit cancel_token, agent.interrupt_controller is used.

import threading
from praisonaiagents import Agent

cancel = threading.Event()
cancel.reason = "user"  # becomes the message in the raised InterruptedError

agent = Agent(name="t", instructions="x")
# cancel.set() during a 429 backoff raises:
# InterruptedError: Agent chat cancelled: user

See Tool Retry & Backoff and OpenAI Client Retries for retry settings.

---

Streaming honours the same Stop

start(stream=True) and iter_stream() short-circuit a pending tool batch on Stop, just like chat() — the same InterruptController / cancel_token seam covers the streaming path end-to-end.

Coverage	Non-streaming (chat())	Streaming (start(stream=True) / iter_stream())
Cancel before model returns tool_calls	✅	✅
Pending tool batch short-circuits on Stop	✅	✅ (since PR #5081)
Follow-up completion request skipped on Stop	✅	✅ (since PR #5081)
Fallback loop (Anthropic / Gemini / Ollama with tools) honours Stop	✅	✅ (since PR #5081)

from praisonaiagents import Agent
from praisonaiagents.agent import InterruptController

controller = InterruptController()

agent = Agent(
    name="Runner",
    instructions="Search, then summarise.",
    tools=["shell", "search"],
    interrupt_controller=controller,
)

# From another thread / a UI Stop button:
# controller.request("user_stop")

for chunk in agent.start("Find the largest file under /var/log and summarise it", stream=True):
    print(chunk, end="", flush=True)
# On Stop mid-turn: pending tools return `error_kind="cancelled"` and the
# stream ends — no further completion request is issued.

Fixed in PraisonAI PR #5081 (issue #5073). Before this PR, cancel_token was dropped at the ToolCallExecutor.execute_batch() boundary on the streaming path — Stop halted the model between iterations but shell / MCP / file tools already dispatched in the current batch ran to completion, and the streaming loop still spent one more completion request on the cancelled turn. Since #5081, the token is forwarded from the public streaming entry point (start(stream=True) / iter_stream()) through get_response_stream into execute_batch, and three additional guards break the streaming and fallback loops the moment the token is signalled.

---

Concurrent turns on one Agent

Two threads (or two asyncio tasks) can call chat() / achat() on the same Agent with different cancel_tokens, and each turn’s tools and approvals see their own token.

Internally the active turn is bound to a ContextVar (_active_turn_cancel) that is copied into tool-executor threads and asyncio tasks, so a second chat() call does not overwrite the first turn’s cancel token. The legacy self._active_turn_token is still set for backward compatibility; _current_turn_liveness() prefers the ContextVar and falls back.

import threading
from praisonaiagents import Agent
from praisonaiagents.agent import InterruptController

agent = Agent(name="Shared", instructions="Answer concurrently.")

tokenA, tokenB = InterruptController(), InterruptController()

threading.Thread(target=agent.chat, kwargs={"prompt": "A", "cancel_token": tokenA}).start()
agent.chat("B", cancel_token=tokenB)

tokenB.request("user_stop")   # only turn B is cancelled; A keeps running

Available as of PR #5764 (merged 2026-10-12, fixes #5763). Before this PR, the second concurrent chat() overwrote self._active_turn_token, so turn A’s approvals read turn B’s token — cancelling B could mark A’s pending approval stale, and cancelling A was effectively a no-op for its own tools.

---

Custom InterruptController implementations

Exposing the event is optional — a controller without it keeps working, giving loop-level cancellation only.

import threading
from praisonaiagents import Agent

class MyController:
    def __init__(self):
        self._flag = threading.Event()

    def request(self, reason: str = "user") -> None:
        self._flag.set()

    def clear(self) -> None:
        self._flag.clear()

    def is_set(self) -> bool:
        return self._flag.is_set()

    @property
    def reason(self):
        return "user" if self._flag.is_set() else None

    def check(self) -> None:
        if self._flag.is_set():
            raise InterruptedError("cancelled")

    @property
    def event(self) -> threading.Event:   # opt in to in-flight abort
        return self._flag

agent = Agent(name="Runner", tools=["shell"], interrupt_controller=MyController())

The runtime probes with getattr(controller, 'event', None), so a controller that omits event is fully backward compatible — it just skips in-flight abort and cancels at the loop boundary.

---

Best Practices

Check the event exists before using it

A tool may run standalone with no injected state. Guard with if state.cancel_event is not None: before calling .is_set() or .wait().

Prefer wait over busy-polling

Use state.cancel_event.wait(timeout=0.1) for idle waits instead of a tight is_set() loop, so you don’t burn CPU while waiting for work or the signal.

Keep the poll window short

Poll roughly every ~100ms — the same cadence the built-in shell tool uses — so /stop feels instant to the user.

Safe to reuse one Agent across concurrent turns

One Agent can serve many concurrent chat() / achat() calls, each with its own cancel_token. Each turn’s tools and approvals read the token that was passed to that call — a stop on one turn never bleeds into another.

Do not persist the event across turns

The event is scoped to the current turn and cleared when an interrupted turn ends. Read it fresh from state.cancel_event each call; never cache it on a module or instance.

---

Related

Tool Timeouts

Deadlines and the executor-level cancel token

Tool Progress

Surface progress from the same slow tools

Gateway Abort & Timeout

HTTP-facing abort and deadlines

Run Outcome

Read the discriminated outcome from a run