Skip to content

Quickstart

This guide will get you integrating OnceOnly into your agent in 5 minutes.

Go to https://onceonly.tech:

  1. Click “Log in” (or “Sign up”)
  2. Sign in with your email
  3. Copy your API key from the dashboard

Save your api_key securely (environment variable, secrets manager, etc).


Your agent wants to execute an action. Check if it’s a duplicate first:

Terminal window
curl -X POST https://api.onceonly.tech/v1/check-lock \
-H "Authorization: Bearer once_live_xxxxxxxxxxxxx" \
-H "Content-Type: application/json" \
-d '{
"key": "payment_invoice_123",
"ttl": 3600,
"metadata": {
"invoice_id": "INV-123",
"amount_usd": 99.99,
"customer_email": "customer@example.com"
}
}'
{
"success": true,
"status": "locked",
"key": "payment_invoice_123",
"ttl": 3600,
"first_seen_at": null
}

Meaning: This is a new action. Safe to proceed with the payment.

If you call with the same key again within 1 hour:

{
"success": false,
"status": "duplicate",
"key": "payment_invoice_123",
"ttl": 3600,
"first_seen_at": "2025-01-15T10:30:00Z"
}

Meaning: This is a duplicate! Don’t charge the customer again. Return the result from the first attempt instead.


Terminal window
pip install onceonly-sdk
import os
from onceonly import OnceOnly
client = OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])
def process_payment(invoice_id: str) -> dict:
lock = client.check_lock(
key=f"payment_invoice_{invoice_id}",
ttl=3600,
meta={"invoice_id": invoice_id},
)
if lock.duplicate:
return {"status": "duplicate", "first_seen_at": lock.first_seen_at}
# Run the real side-effect here (Stripe, email, etc.)
charge_id = "ch_example"
return {"status": "paid", "charge_id": charge_id}

Step 4: Long-running Tasks with AI Leases (Bonus, 1 min)

Section titled “Step 4: Long-running Tasks with AI Leases (Bonus, 1 min)”

For operations that take minutes or hours, use AI Leases instead of simple checks:

import os
from onceonly import OnceOnly
client = OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])
def support_chat_long_running(chat_id: str, messages: list) -> dict:
"""
Long-running support chat that may take 5-30 minutes.
Uses AI Lease instead of simple check-lock.
"""
# Step 1: Acquire a lease (get permission to start this chat)
lease = client.ai.lease(
key=f"support_chat_{chat_id}",
ttl=1800, # 30 minutes
metadata={
"chat_id": chat_id,
"customer_id": "cust_123",
},
)
if lease["status"] == "acquired":
print(f"✓ Lease acquired: {lease['lease_id']}")
lease_id = lease["lease_id"]
# Step 2: Do the long-running work
try:
# Your support chat logic here...
chat_result = run_support_chat(messages)
# Step 3: Report completion
client.ai.complete(
key=f"support_chat_{chat_id}",
lease_id=lease_id,
result=chat_result,
)
return {"status": "completed", "result": chat_result}
except Exception as e:
# Step 3b: Report failure
client.ai.fail(
key=f"support_chat_{chat_id}",
lease_id=lease_id,
error_code="chat_failed",
)
return {"status": "failed", "error": str(e)}
elif lease["status"] == "polling":
# Another process is already handling this chat
print(f"⏳ Chat is being handled by another process. TTL: {lease['ttl_left']}s")
return {"status": "in_progress", "ttl_left": lease.get("ttl_left")}

Another process can check if a task is still running:

import os
from onceonly import OnceOnly
client = OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])
def check_chat_status(chat_id: str) -> dict:
"""Check if a support chat is in progress, completed, or failed"""
key = f"support_chat_{chat_id}"
status = client.ai.status(key)
if status.status == "in_progress":
return {"status": status.status, "ttl_left": status.ttl_left}
if status.status in ("completed", "failed"):
result = client.ai.result(key)
return {
"status": result.status,
"done_at": result.done_at,
"error_code": result.error_code,
"result": result.result,
}
return {"status": status.status, "ok": status.ok}

StepAPI EndpointWhat it does
1https://onceonly.techLog in and copy your API key
2POST /v1/check-lockCheck if action is new or duplicate
3(your code)Execute the action (or skip if duplicate)
4POST /v1/ai/lease(Optional) For long-running tasks
5GET /v1/ai/status(Optional) Check if task is still running
6POST /v1/ai/complete or /v1/ai/failReport task completion

Pattern 1: Simple Idempotency (Most Common)

Section titled “Pattern 1: Simple Idempotency (Most Common)”
import os
from onceonly import OnceOnly
client = OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])
lock = client.check_lock(key=f"action_{id}", ttl=3600)
if lock.duplicate:
return get_cached_result(id)
return do_something()
import os
from onceonly import OnceOnly
client = OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])
key = f"task_{id}"
lease = client.ai.lease(key=key, ttl=3600)
if lease["status"] == "acquired":
lease_id = lease["lease_id"]
try:
output = long_running_operation()
client.ai.complete(key=key, lease_id=lease_id, result={"output": output})
except Exception:
client.ai.fail(key=key, lease_id=lease_id, error_code="task_failed")
else:
# Another process owns it
status = client.ai.status(key)
return {"status": status.status, "ttl_left": status.ttl_left}
import os
from onceonly import OnceOnly
client = OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])
lock = client.check_lock(key, ttl=3600)
if not lock.duplicate:
output = do_action()
cache.set(f"result:{key}", output, ttl=3600)
else:
output = cache.get(f"result:{key}")
if not output:
raise CacheExpired()
return output