Idempotency in AI Agent Operations: Safe Retry Without Duplicate Actions

The Duplicate Action Problem

Retries are essential for resilient AI agents, but they introduce a dangerous side effect: duplicate actions. When an agent calls a booking tool and the response times out, did the booking succeed or not? If the agent retries, the user might end up with two bookings, two charges, or two confirmation emails.

Idempotency ensures that executing the same operation multiple times produces the same result as executing it once. It is the bridge between aggressive retry policies and safe real-world actions.

Idempotency Keys

The foundation of idempotency is a unique key that identifies a specific intended action. When the system sees a repeated key, it returns the original result instead of executing the action again.

flowchart TD
    START["Idempotency in AI Agent Operations: Safe Retry Wi…"] --> A
    A["The Duplicate Action Problem"]
    A --> B
    B["Idempotency Keys"]
    B --> C
    C["Idempotent Tool Wrapper"]
    C --> D
    D["Applying Idempotency to Real Tools"]
    D --> E
    E["State Checking as an Alternative"]
    E --> F
    F["Redis-Backed Production Store"]
    F --> G
    G["FAQ"]
    G --> DONE["Key Takeaways"]
    style START fill:#4f46e5,stroke:#4338ca,color:#fff
    style DONE fill:#059669,stroke:#047857,color:#fff

import hashlib
import json
from dataclasses import dataclass
from typing import Any, Optional
from datetime import datetime, timedelta

@dataclass
class IdempotencyRecord:
    key: str
    result: Any
    status: str  # "pending", "completed", "failed"
    created_at: datetime
    expires_at: datetime

class IdempotencyStore:
    """In-memory idempotency store. Use Redis or PostgreSQL in production."""

    def __init__(self, ttl_hours: int = 24):
        self.records: dict[str, IdempotencyRecord] = {}
        self.ttl = timedelta(hours=ttl_hours)

    def generate_key(self, tool_name: str, args: dict, context_id: str = "") -> str:
        """Generate a deterministic key from the operation parameters."""
        payload = json.dumps(
            {"tool": tool_name, "args": args, "context": context_id},
            sort_keys=True,
        )
        return hashlib.sha256(payload.encode()).hexdigest()

    def check(self, key: str) -> Optional[IdempotencyRecord]:
        record = self.records.get(key)
        if record and datetime.utcnow() < record.expires_at:
            return record
        if record:
            del self.records[key]
        return None

    def reserve(self, key: str) -> bool:
        """Reserve a key before execution. Returns False if already reserved."""
        if self.check(key) is not None:
            return False
        self.records[key] = IdempotencyRecord(
            key=key,
            result=None,
            status="pending",
            created_at=datetime.utcnow(),
            expires_at=datetime.utcnow() + self.ttl,
        )
        return True

    def complete(self, key: str, result: Any):
        record = self.records.get(key)
        if record:
            record.result = result
            record.status = "completed"

    def fail(self, key: str):
        record = self.records.get(key)
        if record:
            record.status = "failed"
            del self.records[key]  # Allow retry

Idempotent Tool Wrapper

Wrap every tool that performs side effects with an idempotency guard.

from functools import wraps

idempotency_store = IdempotencyStore()

def idempotent(tool_fn):
    """Decorator that makes a tool function idempotent."""
    @wraps(tool_fn)
    async def wrapper(args: dict, context_id: str = "", **kwargs):
        key = idempotency_store.generate_key(tool_fn.__name__, args, context_id)

        # Check for existing result
        existing = idempotency_store.check(key)
        if existing and existing.status == "completed":
            return existing.result
        if existing and existing.status == "pending":
            raise RuntimeError(
                f"Operation {tool_fn.__name__} is already in progress for this request"
            )

        # Reserve the key
        if not idempotency_store.reserve(key):
            existing = idempotency_store.check(key)
            if existing and existing.status == "completed":
                return existing.result

        # Execute
        try:
            result = await tool_fn(args, **kwargs)
            idempotency_store.complete(key, result)
            return result
        except Exception:
            idempotency_store.fail(key)
            raise

    return wrapper

Applying Idempotency to Real Tools

Here is how to make common agent tools idempotent.

@idempotent
async def book_appointment(args: dict) -> dict:
    """Book an appointment — safe to retry."""
    patient_id = args["patient_id"]
    doctor_id = args["doctor_id"]
    time_slot = args["time_slot"]

    # The idempotency key is derived from (patient_id, doctor_id, time_slot),
    # so retrying the exact same booking returns the original confirmation.
    booking_id = await db_create_appointment(patient_id, doctor_id, time_slot)
    return {"booking_id": booking_id, "status": "confirmed"}

@idempotent
async def send_notification(args: dict) -> dict:
    """Send a notification — guaranteed at-most-once delivery."""
    recipient = args["recipient"]
    message = args["message"]

    await email_service.send(to=recipient, body=message)
    return {"status": "sent", "recipient": recipient}

@idempotent
async def process_payment(args: dict) -> dict:
    """Process payment — critical to never double-charge."""
    amount = args["amount"]
    customer_id = args["customer_id"]

    charge = await payment_gateway.charge(
        customer_id=customer_id,
        amount=amount,
        idempotency_key=args.get("payment_idempotency_key", ""),
    )
    return {"charge_id": charge["id"], "status": charge["status"]}

State Checking as an Alternative

For some operations, the simplest idempotency strategy is checking whether the action has already been performed before executing it.

async def idempotent_create_user(email: str, name: str) -> dict:
    """Create user only if they do not already exist."""
    existing = await db.fetch_one(
        "SELECT id, email, name FROM users WHERE email = $1",
        email,
    )
    if existing:
        return {"user_id": existing["id"], "status": "already_exists"}

    user_id = await db.execute(
        "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id",
        email, name,
    )
    return {"user_id": user_id, "status": "created"}

Redis-Backed Production Store

For production systems, replace the in-memory store with Redis for atomic operations and automatic expiration.

import redis.asyncio as redis

class RedisIdempotencyStore:
    def __init__(self, redis_url: str, ttl_seconds: int = 86400):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl_seconds

    async def check_and_reserve(self, key: str) -> Optional[dict]:
        """Atomically check and reserve using SET NX."""
        prefixed = f"idem:{key}"

        # Try to reserve
        was_set = await self.redis.set(
            prefixed, json.dumps({"status": "pending"}),
            nx=True, ex=self.ttl,
        )
        if was_set:
            return None  # Successfully reserved, proceed with execution

        # Key exists — fetch the stored result
        data = await self.redis.get(prefixed)
        if data:
            return json.loads(data)
        return None

    async def complete(self, key: str, result: dict):
        prefixed = f"idem:{key}"
        await self.redis.set(
            prefixed,
            json.dumps({"status": "completed", "result": result}),
            ex=self.ttl,
        )

FAQ

How do I generate idempotency keys for LLM-driven tool calls?

Combine the conversation or session ID, the tool name, and the normalized arguments into a hash. The conversation ID ensures that the same logical request across retries maps to the same key, while different conversations for the same user can still perform the same action independently.

What if the operation partially succeeds before a failure?

This is the hardest case. If a tool writes to the database but fails before returning, the idempotency store shows "pending" while the side effect has occurred. Handle this with a two-phase approach: first check the actual state of the world (did the booking actually get created?), then reconcile the idempotency record. The state-check pattern above handles this naturally.

Should read-only tools be made idempotent?

Read-only tools are naturally idempotent since they do not modify state. You do not need to add idempotency keys for database queries, search operations, or information retrieval. Reserve the idempotency infrastructure for tools that create, update, or delete resources, or that trigger external side effects like sending emails.

---

#Idempotency #SafeRetries #ToolDesign #AIAgents #Python #AgenticAI #LearnAI #AIEngineering