Idempotent API Design for AI Agents: Safely Retrying Failed Requests

Why Idempotency Matters for AI Agents

AI agents operate autonomously, making API calls to execute tools, store results, and communicate with external services. Network failures, timeouts, and transient errors are inevitable. When a request fails ambiguously — the server processed it but the response was lost — the agent must retry. Without idempotent API design, that retry can create duplicate charges, send duplicate emails, or schedule duplicate meetings.

Idempotency means that making the same request multiple times produces the same result as making it once. GET, PUT, and DELETE are naturally idempotent by HTTP specification. POST is where you need explicit idempotency mechanisms.

Implementing Idempotency Keys

The standard approach is to have clients send a unique idempotency key with each request. The server stores the result of the first execution and returns the cached result for subsequent requests with the same key.

flowchart LR
    CLIENT(["Client SDK"])
    GW["API Gateway<br/>auth plus rate limit"]
    APP["FastAPI app<br/>handlers and DI"]
    VAL["Pydantic validation"]
    SVC["Service layer<br/>business logic"]
    DB[(Database)]
    QUEUE[(Background queue)]
    OBS[(Tracing)]
    CLIENT --> GW --> APP --> VAL --> SVC
    SVC --> DB
    SVC --> QUEUE
    SVC --> OBS
    SVC --> CLIENT
    style GW fill:#4f46e5,stroke:#4338ca,color:#fff
    style APP fill:#f59e0b,stroke:#d97706,color:#1f2937
    style DB fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
import hashlib
import json
from datetime import datetime, timedelta

app = FastAPI()

# In production, use Redis or a database table
idempotency_store: dict[str, dict] = {}

class AgentTaskRequest(BaseModel):
    agent_id: str
    action: str
    parameters: dict

@app.post("/v1/agent/tasks")
async def create_task(
    request: AgentTaskRequest,
    idempotency_key: str = Header(..., alias="Idempotency-Key"),
):
    # Check for existing result
    if idempotency_key in idempotency_store:
        cached = idempotency_store[idempotency_key]
        if cached["status"] == "processing":
            raise HTTPException(
                status_code=409,
                detail="Request is currently being processed",
            )
        return cached["response"]

    # Mark as processing to prevent concurrent duplicates
    idempotency_store[idempotency_key] = {"status": "processing"}

    try:
        result = await execute_agent_task(request)
        response = {
            "task_id": result["id"],
            "status": "completed",
            "result": result["output"],
        }
        idempotency_store[idempotency_key] = {
            "status": "completed",
            "response": response,
            "created_at": datetime.utcnow().isoformat(),
        }
        return response
    except Exception as e:
        del idempotency_store[idempotency_key]
        raise

Database-Backed Idempotency with PostgreSQL

For production systems, store idempotency records in your database within the same transaction as the business operation. This guarantees atomicity — either both the operation and the idempotency record are committed, or neither is.

from sqlalchemy import Column, String, JSON, DateTime, text
from sqlalchemy.ext.asyncio import AsyncSession

class IdempotencyRecord(Base):
    __tablename__ = "idempotency_keys"

    key = Column(String(255), primary_key=True)
    request_path = Column(String(500), nullable=False)
    request_fingerprint = Column(String(64), nullable=False)
    response_code = Column(String(3), nullable=False)
    response_body = Column(JSON, nullable=False)
    created_at = Column(
        DateTime, server_default=text("NOW()"), nullable=False
    )

async def with_idempotency(
    db: AsyncSession,
    key: str,
    request_path: str,
    request_body: dict,
    handler,
):
    fingerprint = hashlib.sha256(
        json.dumps(request_body, sort_keys=True).encode()
    ).hexdigest()

    existing = await db.get(IdempotencyRecord, key)

    if existing:
        if existing.request_fingerprint != fingerprint:
            raise HTTPException(
                status_code=422,
                detail="Idempotency key reused with different request body",
            )
        return json.loads(existing.response_body)

    result = await handler()

    record = IdempotencyRecord(
        key=key,
        request_path=request_path,
        request_fingerprint=fingerprint,
        response_code="200",
        response_body=json.dumps(result),
    )
    db.add(record)
    await db.commit()

    return result

Request Fingerprinting for Automatic Deduplication

When agents cannot send idempotency keys (for example, when calling third-party APIs), you can deduplicate based on request content. Generate a fingerprint from the request method, path, body, and a time window.

def compute_request_fingerprint(
    method: str, path: str, body: dict, agent_id: str
) -> str:
    canonical = json.dumps(
        {
            "method": method,
            "path": path,
            "body": body,
            "agent_id": agent_id,
        },
        sort_keys=True,
    )
    return hashlib.sha256(canonical.encode()).hexdigest()

async def deduplicate_request(
    fingerprint: str,
    window_seconds: int = 300,
):
    """Check if an identical request was made within the time window."""
    cutoff = datetime.utcnow() - timedelta(seconds=window_seconds)
    # Query database for matching fingerprint within window
    existing = await db.execute(
        text(
            "SELECT response_body FROM idempotency_keys "
            "WHERE request_fingerprint = :fp AND created_at > :cutoff"
        ),
        {"fp": fingerprint, "cutoff": cutoff},
    )
    row = existing.fetchone()
    return json.loads(row[0]) if row else None

Cleanup and Expiration

Idempotency records should not live forever. Implement a cleanup job that removes records older than your retry window, typically 24 to 48 hours.

async def cleanup_expired_keys(max_age_hours: int = 24):
    cutoff = datetime.utcnow() - timedelta(hours=max_age_hours)
    await db.execute(
        text("DELETE FROM idempotency_keys WHERE created_at < :cutoff"),
        {"cutoff": cutoff},
    )
    await db.commit()

FAQ

What should the idempotency key format be?

Use a UUID v4 generated client-side. This guarantees uniqueness without coordination. The agent should generate the key before the first attempt and reuse the same key for all retries of that specific operation. Never reuse an idempotency key for a different operation — the server should reject this with a 422 status code if the request body fingerprint does not match.

How long should idempotency keys be stored?

Store them for at least as long as your maximum retry window. A common choice is 24 hours. After that, the key can be safely deleted. If an agent retries after the key expires, the operation will execute again, which is acceptable since such a long gap likely indicates a new intent rather than a retry.

Should GET requests use idempotency keys?

No. GET requests are inherently idempotent by definition — they read data without side effects. Adding idempotency keys to GET requests adds complexity with no benefit. Reserve idempotency mechanisms for POST and PATCH endpoints that create or modify resources.

---

#Idempotency #APIDesign #AIAgents #FastAPI #Reliability #AgenticAI #LearnAI #AIEngineering