Block runaway agents
Stop a single agent run that loops or blows past a hard ceiling — a loop rule plus the catch snippet.
An agent stuck calling the same thing over and over, or one that just keeps going, can burn a lot of money in one run. Loop detection watches each run — one trace — on its own and stops it before it runs away.
The rule
This isn't a budget-over-time rule; it's a per-run circuit breaker. On the Firewall page, switch to the Runaway loops tab and click New rule from template to open Catch a workflow stuck looping. Two templates cover most cases:
Stop the same request on repeat — fires when the same prompt goes out three times in a row inside one trace, or a short back-and-forth keeps repeating. Catches retry storms.
Limit how many model calls one trace can make — the backstop: a hard ceiling per trace (default 25 calls). There are USD and token variants too: Limit how much one trace can spend and Limit tokens per trace.
Under When this happens, the templates default to Alert only — just notify me. Choose Block — stop the next model call on this trace if you want the run halted, not just flagged. Like every other rule, it is created in Dry-run — switch it to Enforce once you've watched it on real traffic.
The block lands one call late. The call that crosses the threshold is allowed; TokenPolice stops a later call on the same trace. Fast back-to-back calls can slip past. Treat it as a circuit breaker, not an exact ceiling.
A loop rule watches one trace — not a session or a conversation. For "cap this whole
conversation", use a regular firewall rule grouped by session_id. Full detail in
Loop detection.
The snippet
Two things make this work in your code: wrap the run so its calls share one trace, and catch the block when the detector trips.
import token_police as tp
tp.init(
api_key=os.environ["TOKENPOLICE_API_KEY"],
base_url="https://collect.tokenpolice.ai",
firewall="enforce", # blocking only happens under enforce
)
@tp.workflow(name="research_agent") # one run = one trace the rule can watch
def run_agent(user_id: str, session_id: str, goal: str):
while not done:
try:
step = client.chat.completions.create(...)
except tp.TokenPoliceBlockedError as e:
# the loop detector stopped this run
return f"Run halted: {e.reason}"import { workflow, TokenPoliceBlockedError } from "token-police";
const runAgent = workflow({ name: "research_agent" }, async (input) => {
try {
return await client.chat.completions.create(/* ... */);
} catch (e) {
if (e instanceof TokenPoliceBlockedError) return `Run halted: ${e.reason}`;
throw e;
}
});Your app sees a TokenPoliceBlockedError — never an HTTP status. Its kind carries the detector
that fired and reason explains it. Everything else stays fail-open; see
Errors for the full field list.