Idempotent Tool Calls

Writing Retriable Tool Calls

This guide shows how to design tools and API calls that are safe to retry without side effects.

The Problem

API Call 1: "Send email to user@example.com"
            ↓
            Success, but network timeout
            Client doesn't know it succeeded

Retry:      "Send email to user@example.com"
            ↓
            User gets 2 emails ❌

The Solution: Idempotency Key

Every tool call must include a unique, stable key:

# ✅ GOOD
tool_call(
    tool_name="send_email",
    key="email_user_123_welcome",  # Stable, unique
    args={"to": "user@example.com"}
)

# ❌ BAD
tool_call(
    tool_name="send_email",
    key=str(uuid.uuid4()),  # Changes every retry!
    args={"to": "user@example.com"}
)

Pattern 1: Database Record ID

For creating/updating records, use the record ID:

def create_user(user_data: dict):
    """Create user with idempotency"""

    user_id = generate_user_id()
    key = f"create_user_{user_id}"

    result = call_idempotent_tool(
        tool_name="create_user",
        key=key,  # Stable: same user_id = same key
        args=user_data,
        ttl=3600
    )

    return result

Pattern 2: Resource + Operation

For operations on existing resources:

def send_password_reset(user_id: str):
    """Send reset email with idempotency"""

    key = f"email_user_{user_id}_password_reset"

    result = call_idempotent_tool(
        tool_name="send_email",
        key=key,  # Stable: user_id stays same
        args={
            "to": get_user_email(user_id),
            "template": "password_reset"
        },
        ttl=3600
    )

    return result

Pattern 3: Transaction ID

For payments and transactions:

def process_payment(invoice_id: str, amount: float):
    """Process payment with idempotency"""

    key = f"payment_invoice_{invoice_id}"

    result = call_idempotent_tool(
        tool_name="charge_payment",
        key=key,  # Same invoice = same charge
        args={
            "invoice_id": invoice_id,
            "amount": amount
        },
        ttl=3600
    )

    return result

Implementation on Your Backend

Your tool backend must be idempotent:

from flask import Flask, request
import json

app = Flask(__name__)

# Store processed requests
processed = {}

@app.route("/v1/create-ticket", methods=["POST"])
def create_ticket():
    data = request.get_json()

    # Extract key
    key = data.get("key")

    # Check if already processed
    if key in processed:
        # Return same result
        return processed[key]

    # Process new request
    ticket = {
        "id": f"TKT-{uuid.uuid4()}",
        "subject": data.get("subject"),
        "created_at": datetime.now().isoformat()
    }

    # Save to database
    db.tickets.insert(ticket)

    # Cache result for future retries
    processed[key] = {
        "success": True,
        "ticket_id": ticket["id"]
    }

    return processed[key]

Request Format

When calling your tool, include the key:

POST /v1/create-ticket
{
  "key": "ticket_user_123_bug_report",
  "subject": "Application crashes on login",
  "description": "...",
  "priority": "high"
}

Complete Tool Implementation

import json
import hmac
import hashlib
from functools import wraps

class IdempotentTool:
    """Base class for idempotent tools"""

    def __init__(self):
        self.cache = {}  # In production: use Redis

    def is_idempotent(self, key: str) -> bool:
        """Check if request was already processed"""
        return key in self.cache

    def get_cached(self, key: str):
        """Get cached result"""
        return self.cache.get(key)

    def cache_result(self, key: str, result: dict, ttl: int = 3600):
        """Cache result for future retries"""
        self.cache[key] = {
            "result": result,
            "cached_at": time.time(),
            "ttl": ttl
        }

    def verify_signature(self, body: bytes, signature: str, secret: str) -> bool:
        """Verify HMAC-SHA256 signature"""
        expected = hmac.new(
            secret.encode(),
            body,
            hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(expected, signature)

# Specific tool
class CreateTicketTool(IdempotentTool):
    """Tool for creating support tickets idempotently"""

    def handle_request(self, body: bytes, signature: str, secret: str) -> dict:
        """Handle incoming request"""

        # Verify signature
        if not self.verify_signature(body, signature, secret):
            return {"error": "invalid_signature"}, 401

        # Parse request
        data = json.loads(body)
        args = data.get("args", {})
        key = args.get("idempotency_key") or data.get("lease_id")

        # Check if already processed
        if self.is_idempotent(key):
            return self.get_cached(key)["result"]

        # Process new request
        ticket = {
            "id": f"TKT-{uuid.uuid4()}",
            "subject": args.get("subject"),
            "description": args.get("description"),
            "priority": args.get("priority", "normal"),
            "created_at": datetime.now().isoformat()
        }

        # Save to database
        db.tickets.insert(ticket)

        # Cache result
        result = {
            "success": True,
            "ticket_id": ticket["id"],
            "created_at": ticket["created_at"]
        }
        self.cache_result(key, result, ttl=3600)

        return result

# Flask app
tool = CreateTicketTool()

@app.route("/v1/create-ticket", methods=["POST"])
def create_ticket():
    signature = request.headers.get("X-OnceOnly-Signature", "")
    secret = os.getenv("ONCEONLY_TOOL_SECRET")

    result = tool.handle_request(
        request.data,
        signature,
        secret
    )

    return result

Testing Idempotency

def test_idempotent_tool():
    """Test that tool is truly idempotent"""

    key = "test_ticket_123"
    payload = {
        "tool": "create_ticket",
        "args": {
            "idempotency_key": key,
            "subject": "Test ticket"
        },
        "agent_id": "support_bot",
        "ts": 1705322400,
        "lease_id": "lease_test_123",
        "scope_id": "global"
    }

    # Call 1
    response1 = requests.post(
        "http://localhost:5000/v1/create-ticket",
        json=payload,
        headers={"X-OnceOnly-Signature": "..."}
    )
    ticket_id_1 = response1.json()["ticket_id"]

    # Call 2 (retry with same key)
    response2 = requests.post(
        "http://localhost:5000/v1/create-ticket",
        json=payload,
        headers={"X-OnceOnly-Signature": "..."}
    )
    ticket_id_2 = response2.json()["ticket_id"]

    # Both should return same ticket
    assert ticket_id_1 == ticket_id_2, "Tool not idempotent!"

    # Database should have only 1 ticket
    tickets = db.tickets.find({"id": ticket_id_1})
    assert len(tickets) == 1, "Tool created duplicate!"

Best Practices

✅ DO:

1. Extract idempotency_key from args (or use lease_id)
2. Check if key was processed
3. Return cached result if yes
4. Process and cache if no
5. Use Redis for production caching
6. Set appropriate TTL for cache
7. Log all requests
8. Validate signatures

❌ DON’T:

1. Ignore the idempotency key
2. Skip signature validation
3. Process without checking cache
4. Use in-memory cache (losable)
5. Set TTL too short
6. Return different results for same key
7. Delete cache too early
8. Skip error logging

---

Next: Policy Templates