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.

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.

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.

sequenceDiagram
    participant User
    participant Agent
    participant Stream as get_response_stream
    participant Exec as execute_batch
    participant Tool as Pending tools
    participant Model

    User->>Agent: start(prompt, stream=True)
    Agent->>Stream: forward cancel_token
    Stream->>Model: request (tools=...)
    Model-->>Stream: tool_calls
    User->>Agent: interrupt_controller.request("user_stop")
    Stream->>Exec: execute_batch(..., cancel_token=token)
    Exec-->>Stream: cancelled ToolResult (pending calls skipped)
    Note over Stream,Model: follow-up completion request is NOT issued
    Stream-->>Agent: stream ends
    Agent-->>User: streamed output stops promptly

    classDef user fill:#6366F1,stroke:#7C90A0,color:#fff
    classDef agent fill:#8B0000,stroke:#7C90A0,color:#fff
    classDef stream fill:#189AB4,stroke:#7C90A0,color:#fff
    classDef tool fill:#F59E0B,stroke:#7C90A0,color:#fff
    classDef model fill:#10B981,stroke:#7C90A0,color:#fff

    class User user
    class Agent agent
    class Stream,Exec stream
    class Tool tool
    class Model model

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.

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

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

Related