AI agents often retry tool calls when an API connection drops or when the model gets confused by a slow response. If your agent is processing credit card refunds, sending customer emails, or updating a database, these repeated calls can create massive headaches. I once watched a buggy agent send five duplicate confirmation emails to a high-value client because of a network timeout. You can prevent this entirely by wrapping your agent's tools in an idempotency layer that catches duplicate requests before they touch your core services.
Before starting, make sure you have the following ready:
- Python 3.10 or newer installed on your machine
- Redis installed locally or an active Redis Cloud instance (a standard Python dictionary works for simple local testing)
- Basic familiarity with Python decorators and JSON handling
Step 1: Generate Deterministic Idempotency Keys
An operation is idempotent if running it multiple times produces the exact same result as running it once. To achieve this, your wrapper needs a way to uniquely identify every incoming request.
You can derive a key automatically by hashing the tool name along with its input arguments. Here is a simple function to build a deterministic key:
import hashlib
import json
from typing import Any, Dict
def generate_tool_key(tool_name: str, kwargs: Dict[str, Any]) -> str:
# Sort keys so {"a": 1, "b": 2} matches {"b": 2, "a": 1}
serialized_args = json.dumps(kwargs, sort_keys=True)
raw_string = f"{tool_name}:{serialized_args}"
return hashlib.sha256(raw_string.encode('utf-8')).hexdigest()
If the agent sends send_email(recipient="test@example.com", body="Hello") three times in a row, this function generates the exact same hash every single time.
Step 2: Build the Idempotency Wrapper Decorator
Next, write a decorator that intercepts tool calls before execution. The wrapper checks your cache store for the generated key. If the key exists, it returns the previously saved output instantly without calling the backend API again.
Here is a full Python implementation using a simple local store:
import functools
import time
# Simple in-memory cache for demonstration; replace with Redis in production
CACHE_STORE = {}
def idempotent_tool(ttl_seconds: int = 3600):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
tool_name = func.__name__
cache_key = generate_tool_key(tool_name, kwargs)
# Check if we already processed this exact action
if cache_key in CACHE_STORE:
cached_entry = CACHE_STORE[cache_key]
# Check if cache is still valid
if time.time() - cached_entry['timestamp'] < ttl_seconds:
print(f"[IDEMPOTENT] Returning cached response for {tool_name}")
return cached_entry['result']
# Execute the actual tool function if key is not cached
result = func(*args, **kwargs)
# Store the result in cache
CACHE_STORE[cache_key] = {
'result': result,
'timestamp': time.time()
}
return result
return wrapper
return decorator
If you work on complex backend tasks like workflow automation, replacing the in-memory dictionary with Redis ensures that key checks persist across distributed worker nodes.
Step 3: Wrap Your Agent Functions
Now apply the @idempotent_tool() decorator to your tool definitions. Here is an example of an external payment call wrapped for safe agent usage:
@idempotent_tool(ttl_seconds=86400)
def process_refund(user_id: str, amount: float, reason: str) -> dict:
# Simulate an expensive or risky API call
print(f"[API CALL] Processing refund of ${amount} for user {user_id}...")
# Fake successful payment gateway payload
return {
"status": "success",
"transaction_id": f"tx_rf_{user_id}_9981",
"amount_refunded": amount
}
Step 4: Test the Wrapper Against Agent Retries
Simulate an agent hitting an API timeout loop by calling the wrapped function multiple times back-to-back with identical arguments.
if __name__ == "__main__":
payload = {
"user_id": "usr_4021",
"amount": 49.99,
"reason": "Accidental double charge"
}
print("--- First Call (Agent initiates action) ---")
res1 = process_refund(**payload)
print("Result 1:", res1)
print("\n--- Second Call (Agent retries due to local timeout) ---")
res2 = process_refund(**payload)
print("Result 2:", res2)
Expected Output
When you run the script, the first call executes the underlying function and hits the mock payment gateway. The second call detects the matching request signature, skips the API code entirely, and returns the cached payload.
--- First Call (Agent initiates action) ---
[API CALL] Processing refund of $49.99 for user usr_4021...
Result 1: {'status': 'success', 'transaction_id': 'tx_rf_usr_4021_9981', 'amount_refunded': 49.99}
--- Second Call (Agent retries due to local timeout) ---
[IDEMPOTENT] Returning cached response for process_refund
Result 2: {'status': 'success', 'transaction_id': 'tx_rf_usr_4021_9981', 'amount_refunded': 49.99}
The underlying payment API only ran once. Your user was not double refunded, and your agent received the expected response object without crashing.
Managing Edge Cases and Locks
In production setups, you need to handle in-flight requests. If an agent fires off two duplicate network requests within milliseconds of each other, both might check the cache before either finishes writing to it.
To solve this, set a short lock key (like IN_PROGRESS) in Redis as soon as a tool call begins. If a second request arrives while that lock key exists, force the second call to wait a few seconds and check the cache again instead of executing the underlying logic right away.
For critical production workloads, implementing robust tool safety patterns is essential. If you want experienced engineers to build or audit your agent infrastructure, Gaper can connect you with vetted, remote software developers who specialize in custom AI Agent Development.
Top comments (0)