Every team shipping agents eventually has the same conversation. It usually happens at an awkward hour, after someone notices that an agent has been calling the same API for forty minutes, or that the model bill for one afternoon looks like a typo.
The question is always the same: “How do we turn it off?”
If the answer is “we kill the pod” or “someone revokes the API key,” you don’t have a kill switch. You have a panic button that happens to work when the stars align. A real kill switch is designed before the incident, tested before the incident, and works without anyone needing to understand what the agent was doing at the time.
This post walks through a layered kill switch you can build in an afternoon. It’s framework-agnostic. The examples are Python, but the ideas port to anything that can wrap a tool call.
Why “just stop the process” isn’t enough
Stopping the process feels like the obvious answer, but agents make it harder than it sounds.
First, agents are rarely a single process. A planner may hand work to sub-agents, queue jobs, or trigger downstream workflows. Killing one worker doesn’t stop the work already in flight.
Second, stopping mid-action can be worse than letting it finish. If an agent is halfway through a multi-step write, an abrupt kill can leave data in an inconsistent state.
Third, the agent may be fine and the situation may be the problem. You may want to stop one tool, one customer’s traffic, or one class of action while everything else keeps running.
So instead of one big red button, think in layers. Each layer catches a different kind of failure, and each one is cheap on its own.
The four failures you’re actually defending against
When agents go wrong in production, it’s almost always one of these:
- Runaway spend. The agent keeps calling the model or a paid API far beyond what the task is worth.
- Runaway loops. The agent repeats the same action, or cycles between a few actions, without making progress.
- Cascading failures. A downstream service is failing and the agent keeps hammering it, retrying, or working around it in creative ways.
- Wrong action at scale. The agent does something it shouldn’t, and it does it many times before anyone notices.
Each failure maps to a different control. Let’s build them one at a time.
Layer 1: Budget caps
The simplest and most valuable control is a hard ceiling on cost per run. Not an alert. A ceiling.
Alerts are for humans, and humans are asleep, in meetings, or on a plane. A ceiling is enforced by code. Decide what a single task is worth, set the cap slightly above that, and make the agent stop when it hits it.
You can cap more than dollars. Token counts, number of tool calls, and wall-clock time are all useful budgets, and they’re easier to measure than cost in some setups.
Layer 2: Step limits and loop detection
A step limit is a blunt but effective tool: no run gets more than N actions. Pick N based on your longest legitimate task, then add a little headroom.
Step limits don’t catch loops that happen within the limit, though. An agent that repeats the same failing call twenty times before hitting a forty-step cap has still wasted time and money, and possibly hammered a service.
Loop detection fixes that. The idea is simple: fingerprint each action (tool name plus normalized arguments) and look at a sliding window. If the same fingerprint shows up too many times in the window, trip the switch.
Layer 3: A circuit breaker for failures
This one comes from distributed systems, and it fits agents well. If a tool fails several times in a row, stop calling it for a cooldown period. Agents are especially prone to the retry-and-improvise pattern, where a failing tool leads the model to try increasingly strange alternatives. A circuit breaker cuts that off at the source.
The breaker has three states: closed (normal), open (blocked), and half-open (allow one probe call to see if things have recovered).
Layer 4: A global kill flag, checked out-of-band
The first three layers are automatic. This one is for humans.
A global kill flag is a value stored somewhere outside the agent’s runtime, such as a feature-flag service, a Redis key, or a row in a database. The agent checks it before every step. Flipping it stops every agent that reads it, across processes and machines, without a deploy.
Two details matter a lot here:
Make the check cheap and fail closed. If the flag store is unreachable, the agent should stop rather than carry on. A kill switch that fails open isn’t one.
Support scopes. A flag per agent, per tool, and a global one gives you a dial instead of a hammer.
Putting it together
Here’s a compact guard that combines all four layers. It wraps every tool call, so the agent can’t skip it.
python
import hashlib
import json
import time
from collections import deque
class KillSwitchTripped(Exception):
"""Raised when any guard layer decides the agent must stop."""
def __init__(self, layer: str, reason: str):
self.layer = layer
self.reason = reason
super().__init__(f"[{layer}] {reason}")
class AgentGuard:
def init(
self,
is_killed, # callable -> bool, reads the out-of-band flag
max_cost_usd: float = 5.0,
max_steps: int = 40,
loop_window: int = 8,
loop_threshold: int = 3,
max_consecutive_errors: int = 3,
breaker_cooldown_s: int = 60,
):
self.is_killed = is_killed
self.max_cost_usd = max_cost_usd
self.max_steps = max_steps
self.loop_threshold = loop_threshold
self.max_consecutive_errors = max_consecutive_errors
self.breaker_cooldown_s = breaker_cooldown_s
self.cost = 0.0
self.steps = 0
self.recent = deque(maxlen=loop_window)
self.errors = {} # tool -> consecutive error count
self.open_until = {} # tool -> timestamp when breaker may half-open
@staticmethod
def _fingerprint(tool: str, args: dict) -> str:
raw = json.dumps({"t": tool, "a": args}, sort_keys=True, default=str)
return hashlib.sha256(raw.encode()).hexdigest()[:16]
def before_call(self, tool: str, args: dict):
# Layer 4: out-of-band kill flag. Fail closed if we can't read it.
try:
if self.is_killed():
raise KillSwitchTripped("kill_flag", "global kill flag is set")
except KillSwitchTripped:
raise
except Exception as exc:
raise KillSwitchTripped("kill_flag", f"cannot read flag, failing closed: {exc}")
# Layer 1: budget
if self.cost >= self.max_cost_usd:
raise KillSwitchTripped("budget", f"cost cap ${self.max_cost_usd} reached")
# Layer 2: step limit
if self.steps >= self.max_steps:
raise KillSwitchTripped("steps", f"step cap {self.max_steps} reached")
# Layer 2: loop detection
fp = self._fingerprint(tool, args)
if list(self.recent).count(fp) >= self.loop_threshold:
raise KillSwitchTripped("loop", f"repeated call to {tool} detected")
# Layer 3: circuit breaker
if time.time() < self.open_until.get(tool, 0):
raise KillSwitchTripped("breaker", f"circuit open for {tool}")
self.recent.append(fp)
self.steps += 1
def after_call(self, tool: str, ok: bool, cost_usd: float = 0.0):
self.cost += cost_usd
if ok:
self.errors[tool] = 0
return
self.errors[tool] = self.errors.get(tool, 0) + 1
if self.errors[tool] >= self.max_consecutive_errors:
self.open_until[tool] = time.time() + self.breaker_cooldown_s
And here’s how you wire it into a tool-calling loop:
python
def run_tool(guard: AgentGuard, tool_fn, name: str, args: dict):
guard.before_call(name, args) # may raise KillSwitchTripped
try:
result, cost = tool_fn(**args)
guard.after_call(name, ok=True, cost_usd=cost)
return result
except KillSwitchTripped:
raise
except Exception:
guard.after_call(name, ok=False)
raise
The important design choice is that the guard sits around the tools, not inside the prompt. You can’t rely on the model to stop itself. Prompts are suggestions. Wrapper code is enforcement.
What “stopping” should actually do
Tripping the switch is only half the job. What happens next matters just as much.
Stop safely, not instantly. Where possible, finish or roll back the current atomic operation, then halt. Design your multi-step actions so there’s a clear “safe to stop here” boundary.
Capture state. When a guard trips, write out what the agent was doing: the last several actions, the layer that tripped, the cost so far. This is the first thing you’ll want during the postmortem.
Alert with context. “Agent stopped” isn’t useful at 3 a.m. “Agent X stopped by loop detector after repeating search_orders four times with identical arguments” is.
Don’t auto-restart blindly. If an agent stops because it was looping, restarting it unchanged will just loop again. Require either a human decision or a materially different input before resuming.
Test it before you need it
A kill switch you’ve never triggered is a hypothesis. Treat it like any other critical path and test it.
Unit tests for each layer: feed the guard a repeated call, a rising cost, a failing tool, and a set flag, and assert the right exception fires.
A game day: in a staging environment, deliberately start a misbehaving agent and flip the global flag. Time how long it takes to stop. If the answer is “a few minutes,” find out why.
A fail-closed test: make the flag store unreachable and confirm the agent stops rather than continues.
The game day is the one teams skip, and it’s the one that finds the real problems: a worker that never reads the flag, a queue that keeps feeding jobs after the agent is halted, an on-call engineer who doesn’t know where the switch lives.
Takeaways
- A kill switch is a set of layered controls, not a single button.
- Budget caps, step limits, loop detection, circuit breakers, and a global flag each catch a different failure.
- Enforce limits in code that wraps your tools. Don’t rely on the prompt.
- Fail closed, stop safely, and capture state when you stop.
- Test it on purpose, before production tests it for you.
None of this makes your agent smarter. It makes your agent containable, which is the property that lets you give it more responsibility with a clear conscience.
If you’ve built something similar, or found a failure mode these layers don’t cover, I’d like to hear about it in the comments.
Top comments (0)