Quickstart
Quickstart: First 5 Minutes
Section titled “Quickstart: First 5 Minutes”This guide will get you integrating OnceOnly into your agent in 5 minutes.
Step 1: Get Your API Key (1 min)
Section titled “Step 1: Get Your API Key (1 min)”Option A: Via Website
Section titled “Option A: Via Website”Go to https://onceonly.tech:
- Click “Log in” (or “Sign up”)
- Sign in with your email
- Copy your API key from the dashboard
Save your api_key securely (environment variable, secrets manager, etc).
Step 2: First Idempotency Check (1 min)
Section titled “Step 2: First Idempotency Check (1 min)”Your agent wants to execute an action. Check if it’s a duplicate first:
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" } }'Response: New Action
Section titled “Response: New Action”{ "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.
Response: Duplicate
Section titled “Response: Duplicate”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.
Step 3: Python Integration (2 min)
Section titled “Step 3: Python Integration (2 min)”Installation
Section titled “Installation”pip install onceonly-sdkCode Example
Section titled “Code Example”import osfrom 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 osfrom 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")}Step 5: Check Status (Optional, 1 min)
Section titled “Step 5: Check Status (Optional, 1 min)”Another process can check if a task is still running:
import osfrom 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}Summary
Section titled “Summary”| Step | API Endpoint | What it does |
|---|---|---|
| 1 | https://onceonly.tech | Log in and copy your API key |
| 2 | POST /v1/check-lock | Check if action is new or duplicate |
| 3 | (your code) | Execute the action (or skip if duplicate) |
| 4 | POST /v1/ai/lease | (Optional) For long-running tasks |
| 5 | GET /v1/ai/status | (Optional) Check if task is still running |
| 6 | POST /v1/ai/complete or /v1/ai/fail | Report task completion |
Common Patterns
Section titled “Common Patterns”Pattern 1: Simple Idempotency (Most Common)
Section titled “Pattern 1: Simple Idempotency (Most Common)”import osfrom 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()Pattern 2: Long-running with Lease
Section titled “Pattern 2: Long-running with Lease”import osfrom 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}Pattern 3: Caching Results
Section titled “Pattern 3: Caching Results”import osfrom 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 outputNext Steps
Section titled “Next Steps”- Idempotency — Deep dive into preventing duplicates
- AI Leases — Long-running task management
- Policies — Control what agents can do
- Tools Registry — Register custom APIs
- Python SDK — Official Python client + examples
- Idempotency API — Full
check-lockreference