Your agent calls a flaky API, gets a 429, retries with the same arguments, and gets rate-limited again. Ten minutes later, in a fresh session, it does the exact same thing. Nothing about the failure got remembered — the agent has no way to know it already learned this lesson, because the lesson lived in a transcript that got thrown away when the process exited.
This week a post about manually gatekeeping AI agent tool calls hit the front page of dev.to with 48 comments — mostly developers arguing over how much you can trust an agent's tool-calling loop. The honest answer is: you can trust it exactly as much as it remembers what already went wrong. Below is a complete, runnable gatekeeper that wraps any tool call, checks Mem0 for similar past failures before executing, and records the outcome afterward — so the second time your agent is about to make the same mistake, it knows.
What we're building
A MemoryGatekeeper class that sits between your agent's decision to call a tool and the actual execution:
- Before calling a tool, search memory for semantically similar past attempts.
- If a similar attempt failed recently, block the call and return the remembered reason instead of burning a real API call.
- After every call — success or failure — write the outcome back to memory.
You'll run the same script twice: the first run fails and gets recorded, the second run gets blocked before it wastes a request.
Setup
You need Python 3.10+, an OpenAI API key (Mem0's default extraction pipeline uses it to turn raw text into structured memories — no separate Mem0 account required for this local setup), and the mem0ai package.
pip install mem0ai
export OPENAI_API_KEY=sk-...
Mem0 defaults to a local, on-disk vector store, so nothing here talks to a hosted Mem0 service — it's a fully self-contained memory layer you own.
Step 1: The gatekeeper
# gatekeeper.py
from mem0 import Memory
from datetime import datetime, timezone
class MemoryGatekeeper:
def __init__(self, agent_id="default-agent", block_threshold=0.75):
self.memory = Memory()
self.agent_id = agent_id
self.block_threshold = block_threshold
def _describe_call(self, tool_name, args):
return f"tool call: {tool_name} with args {args}"
def check(self, tool_name, args):
"""Returns (allowed: bool, reason: str | None)."""
query = self._describe_call(tool_name, args)
hits = self.memory.search(query, user_id=self.agent_id, limit=3)
for hit in hits.get("results", []):
score = hit.get("score", 0)
memory_text = hit.get("memory", "")
if score >= self.block_threshold and "failed" in memory_text.lower():
return False, memory_text
return True, None
def record(self, tool_name, args, success, detail):
outcome = "succeeded" if success else "failed"
text = (
f"{self._describe_call(tool_name, args)} {outcome} "
f"at {datetime.now(timezone.utc).isoformat()}: {detail}"
)
self.memory.add(text, user_id=self.agent_id)
def call(self, tool_name, args, fn):
allowed, reason = self.check(tool_name, args)
if not allowed:
print(f"[BLOCKED] {tool_name}({args}) — remembered: {reason}")
return {"blocked": True, "reason": reason}
try:
result = fn(*args.values()) if isinstance(args, dict) else fn(args)
self.record(tool_name, args, True, "ok")
return {"blocked": False, "result": result}
except Exception as exc:
self.record(tool_name, args, False, str(exc))
raise
The check step uses semantic search, not exact string matching — fetch_weather(city="NYC") and fetch_weather(city="New York") will match each other, which is exactly the kind of near-duplicate an exact-match cache would miss.
Step 2: A tool that fails predictably
# flaky_tool.py
class RateLimitError(Exception):
pass
def call_flaky_api(endpoint):
if endpoint == "/reports/daily":
raise RateLimitError("429: rate limit exceeded, retry after 3600s")
return {"status": "ok", "endpoint": endpoint}
Step 3: Run it twice
# run.py
from gatekeeper import MemoryGatekeeper
from flaky_tool import call_flaky_api
gate = MemoryGatekeeper(agent_id="report-agent")
try:
gate.call("call_flaky_api", {"endpoint": "/reports/daily"}, call_flaky_api)
except Exception as exc:
print(f"First attempt failed as expected: {exc}")
First run:
First attempt failed as expected: 429: rate limit exceeded, retry after 3600s
Run python run.py again — same process, same tool, same arguments, but now the agent has memory of the earlier failure:
[BLOCKED] call_flaky_api({'endpoint': '/reports/daily'}) — remembered: tool call: call_flaky_api with args {'endpoint': '/reports/daily'} failed at 2026-08-18T09:12:04+00:00: 429: rate limit exceeded, retry after 3600s
No second API call, no second rate-limit hit, and the agent gets a reason it can reason about (or relay to a human) instead of a raw stack trace.
Tuning it for real use
Three knobs matter once you move past the toy example:
-
block_threshold— 0.75 is conservative on purpose. Push it lower and you'll block near-misses that would have actually succeeded; push it higher and you'll only catch near-identical repeats. Log every block decision for a week before trusting the default. -
Scope memory per agent, not globally.
user_id=self.agent_idkeeps one agent's bad luck from blocking a different agent's legitimate call to the same tool. If multiple agents genuinely share risk (same downstream API, same rate limit bucket), give them a sharedagent_id. -
Expire failures, don't keep them forever. A 429 from an hour ago should block a retry; a 429 from three weeks ago probably shouldn't. Mem0 memories carry timestamps in their metadata — add a check in
check()that discards hits older than your retry window (e.g.,time.time() - hit["created_at"] < 3600) so stale failures don't permanently disable a tool that's since recovered. This one-line filter is the difference between a gatekeeper and a agent that's afraid of everything it's ever failed at once.
Why this beats an in-process cache
A plain dict or functools.lru_cache would catch the exact-repeat case in a single run, but it dies with the process and can't do semantic matching — it won't know that retrying /reports/daily and /reports/daily/ are the same mistake. The value of routing this through Mem0 specifically is that the memory persists across restarts and deployments, and the same store can also hold successful patterns — tool calls that worked, arguments that were well-formed, sequences that completed cleanly — so the gatekeeper isn't just a blocklist, it's the beginning of an agent that actually gets better at using its tools over time instead of relearning the same lesson every cold start.
The full example above is under 80 lines and runs with nothing but a Python environment and an OpenAI key — clone it, run it twice, and you'll see the block happen live before you've read the rest of this sentence.
Top comments (0)