Idempotency patterns for APIs and event handlers. Use when implementing exactly-once semantics, deduplicating requests, or building reliable distributed systems.
View on GitHubyonatangross/orchestkit
orchestkit-complete
January 24, 2026
Select agents to install to:
npx add-skill https://github.com/yonatangross/orchestkit/blob/main/./skills/idempotency-patterns/SKILL.md -a claude-code --skill idempotency-patternsInstallation paths:
.claude/skills/idempotency-patterns/# Idempotency Patterns (2026)
Patterns for ensuring operations can be safely retried without unintended side effects.
## Overview
- Building payment or financial APIs
- Implementing webhook handlers
- Processing messages from queues
- Creating mutation endpoints (POST, PUT, DELETE)
- Building distributed systems with at-least-once delivery
## Quick Reference
### Idempotency Key Generation
```python
import hashlib
import json
from typing import Any
def generate_idempotency_key(
*,
entity_id: str,
action: str,
params: dict[str, Any] | None = None,
) -> str:
"""
Generate deterministic idempotency key.
Args:
entity_id: Unique identifier of the entity
action: The action being performed
params: Optional parameters that affect the result
Returns:
32-character hex string
"""
content = f"{entity_id}:{action}"
if params:
# Sort keys for deterministic output
content += f":{json.dumps(params, sort_keys=True)}"
return hashlib.sha256(content.encode()).hexdigest()[:32]
# Examples
key1 = generate_idempotency_key(
entity_id="order-123",
action="create",
params={"amount": 100, "currency": "USD"},
)
key2 = generate_idempotency_key(
entity_id="payment-456",
action="refund",
)
```
### FastAPI Idempotency Middleware
```python
from fastapi import Request, Response, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import redis.asyncio as redis
import json
class IdempotencyMiddleware(BaseHTTPMiddleware):
"""Handle Idempotency-Key header for POST/PUT/PATCH."""
def __init__(self, app, redis_client: redis.Redis, ttl: int = 86400):
super().__init__(app)
self.redis = redis_client
self.ttl = ttl
async def dispatch(self, request: Request, call_next):
# Only apply to mutation methods
if request.method not in ("POST", "PUT", "PATCH"):
return await call_next(request)
# Check