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.
A MemoryGatekeeper
class that sits between your agent's decision to call a tool and the actual execution:
You'll run the same script twice: the first run fails and gets recorded, the second run gets blocked before it wastes a request.
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.
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.
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}
python
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.
Three knobs matter once you move past the toy example:
block_threshold
user_id=self.agent_id
keeps 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 shared agent_id
.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.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.