Skip to main content

Python SDK for OnceOnly idempotency API

Project description

OnceOnly Python SDK

AI Agent Execution & Governance Layer

Exactly-once execution + runtime safety + agent control plane.

PyPI version License: MIT Downloads Python 3.8+

Website โ€ข Docs โ€ข API Reference โ€ข Examples


๐ŸŽฏ What is OnceOnly?

The problem: AI agents are non-deterministic. They retry failed calls, re-run tools, crash mid-execution, and replay events. This causes duplicate payments, repeated emails, and inconsistent state.

The solution: OnceOnly sits between your AI and the real world, guaranteeing:

โœ… Exactly-once execution - Same input = same result, always
โœ… Crash safety - Worker dies? Pick up where you left off
โœ… Retry safety - Agent retries? We deduplicate automatically
โœ… Budget control - Cap spending per agent/hour/day
โœ… Permission enforcement - Whitelist/blacklist tools
โœ… Kill switch - Disable rogue agents instantly
โœ… Forensic audit - Complete action history

This isn't just idempotency. This is an AI Agent Control Plane.


โšก Quick Start (30 seconds)

pip install onceonly-sdk
from onceonly import OnceOnly

client = OnceOnly(api_key="once_live_...")

# Prevent duplicate webhook processing
result = client.check_lock(key="webhook:stripe:evt_123", ttl=3600)

if result.duplicate:
    return {"status": "already_processed"}

# Process webhook (runs exactly once)
process_payment(webhook_data)

That's it. You just made your webhook idempotent.


๐Ÿš€ 5-Minute Tutorial

1๏ธโƒฃ Basic Deduplication (Webhooks, Cron Jobs, Workers)

from onceonly import OnceOnly

client = OnceOnly(api_key="once_live_...")

# Stripe webhook
@app.post("/webhooks/stripe")
def stripe_webhook(event_id: str):
    result = client.check_lock(
        key=f"stripe:{event_id}",
        ttl=7200  # 2 hours
    )
    
    if result.duplicate:
        return {"status": "ok"}  # Already processed
    
    # Process event (guaranteed exactly-once)
    handle_payment_succeeded(event_id)
    return {"status": "processed"}

2๏ธโƒฃ AI Agent with Budget & Permissions

from onceonly import OnceOnly

client = OnceOnly(api_key="once_live_...")

# Set policy (one-time setup)
client.gov.upsert_policy({
    "agent_id": "billing-agent",
    "max_actions_per_hour": 200,
    "max_spend_usd_per_day": 50.0,
    "allowed_tools": ["stripe.charge", "send_email"],
    "blocked_tools": ["delete_user"]
})

# Execute tool with enforcement
result = client.ai.run_tool(
    agent_id="billing-agent",
    tool="stripe.charge",
    args={"amount": 9999, "currency": "usd"},
    spend_usd=0.5,  # Track API cost
)

if result.allowed:
    print(f"Charged: {result.result}")
elif result.decision == "blocked":
    print(f"Agent blocked: {result.policy_reason}")

3๏ธโƒฃ Exactly-Once Function Execution

from onceonly import OnceOnly, idempotent_ai

client = OnceOnly(api_key="once_live_...")

@idempotent_ai(
    client,
    key_fn=lambda user_id: f"welcome:email:{user_id}",
    ttl=86400  # 24 hours
)
def send_welcome_email(user_id: str):
    # This runs exactly ONCE per user_id
    # Even if called 1000 times concurrently
    email_service.send(
        to=get_user_email(user_id),
        template="welcome"
    )
    return {"sent": True}

# All these calls get the same result
send_welcome_email("user_123")  # Sends email
send_welcome_email("user_123")  # Returns cached result
send_welcome_email("user_123")  # Returns cached result

โœ… Cheat-Sheet (Pick The Right Call)

I wantโ€ฆ

  • Idempotent webhook/cron/job: check_lock(key, ttl, meta)
  • Long-running server job: ai.run_and_wait(key, ttl, metadata)
  • Start/attach AI run (without polling): ai_run(key=..., run_id=...)
  • Governed tool call (agent + tool): ai.run_tool(agent_id, tool, args, spend_usd)
  • Local side-effect exactly once: ai.run_fn(key, fn, ttl)
  • Run debug timeline: get_run_timeline(run_id, limit, offset)
  • Custom run event: post_event(run_id=..., type=..., ...)
  • Update notification prefs: update_notifications(...)
  • Decorator version: @idempotent or @idempotent_ai

Async equivalents

  • check_lock_async
  • ai_run_async
  • ai.run_and_wait_async
  • ai.run_tool_async
  • ai.run_fn_async
  • post_event_async
  • get_run_timeline_async
  • update_notifications_async

๐Ÿค– Full LLM Agent Flow (No OnceOnly vs OnceOnly)

These two examples show why OnceOnly matters in production.

Without OnceOnly (duplicates + money loss)

# examples/ai/agent_full_flow_no_onceonly.py
decision = llm_decide()
payload = {"tool": decision["tool"], "args": decision["args"]}

# A retry or crash can re-run this call
call_tool(payload)
call_tool(payload)  # duplicate charge

With OnceOnly (deduped + governed)

# examples/ai/agent_full_flow_onceonly.py
res = client.ai.run_tool(
    agent_id="billing-agent",
    tool="stripe.charge",
    args={"amount": 9999, "currency": "usd", "user_id": "u_42"},
    spend_usd=0.5
)

if res.allowed:
    print(res.result)
else:
    print("Blocked:", res.policy_reason)

Why this matters

  • Prevents duplicate charges on retries
  • Enforces budgets and permissions
  • Gives audit trails for every tool call

Cost impact (simple example)

  • Without OnceOnly: 1 retry on a $99 charge = $198
  • With OnceOnly: 1 retry on a $99 charge = $99

Flow diagram (simplified)

LLM -> Tool Call -> External System
  |       |             |
  |       |__ retry ____|   (duplicate charge)
  |
OnceOnly in between
  |
LLM -> OnceOnly -> Tool Call -> External System
          |
          |__ duplicate detected -> blocked

๐Ÿ“š Complete Feature Matrix

Feature Description Use Case
check_lock() Fast idempotency primitive Webhooks, cron jobs, workers
ai.run_and_wait() Long-running AI jobs Image gen, video processing, reports
ai.run_tool() Governance tool runner Tool calls with budgets/permissions
ai.run_fn() Local exactly-once execution Payments, emails, database writes
@idempotent_ai Decorator for functions Simple exactly-once guarantee
gov.upsert_policy() Set agent limits Budget caps, tool permissions
gov.disable_agent() Kill switch Emergency stop
gov.agent_logs() Audit trail Forensics, compliance
gov.agent_metrics() Usage stats Monitoring, alerting

๐Ÿง  Architecture Layers

OnceOnly provides 5 layers of safety:

โ”Œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”
โ”‚ L5: Agent Governance (policies, kill switch, audit)    โ”‚
โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค
โ”‚ L4: Decorator Runtime (@idempotent_ai)                 โ”‚
โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค
โ”‚ L3: Local Side-Effects (ai.run_fn)                     โ”‚
โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค
โ”‚ L2: AI Job Orchestration (ai.run_and_wait)             โ”‚
โ”œโ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”ค
โ”‚ L1: Idempotency Primitive (check_lock)                 โ”‚
โ””โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”˜

Pick the layer that fits your use case. They compose cleanly.


๐Ÿ’Ž Golden Example (Payment with Full Safety)

This example shows complete runtime + governance safety:

from onceonly import OnceOnly, idempotent_ai
import stripe

client = OnceOnly(api_key="once_live_...")

# 1. Set governance policy (one-time)
client.gov.upsert_policy({
    "agent_id": "billing-agent",
    "max_actions_per_hour": 100,
    "max_spend_usd_per_day": 25.0,
    "allowed_tools": ["stripe.charge", "send_receipt"],
    "blocked_tools": ["delete_user", "refund_all"]
})

# 2. Define exactly-once payment function
@idempotent_ai(
    client,
    key_fn=lambda user_id, amount: f"charge:{user_id}:{amount}",
    ttl=300,  # 5 minutes
    metadata_fn=lambda u, a: {
        "user_id": u,
        "amount_cents": a,
        "agent": "billing-agent"
    }
)
def charge_user(user_id: str, amount_cents: int):
    """Charge user - guaranteed exactly once"""
    return stripe.Charge.create(
        amount=amount_cents,
        currency="usd",
        customer=get_stripe_customer_id(user_id)
    )

# 3. Execute with full safety
result = charge_user("user_42", 9999)

if result.status == "completed":
    charge_id = result.result["data"]["id"]
    print(f"โœ… Charged: {charge_id}")
else:
    print(f"โŒ Failed: {result.error_code}")

Guarantees:

  • โœ… Charged exactly once (even if retried 1000x)
  • โœ… Budget enforced (won't exceed $25/day)
  • โœ… Tool allowed (stripe.charge in whitelist)
  • โœ… Crash safe (worker dies? resumes automatically)
  • โœ… Audit logged (forensic trail for compliance)

๐Ÿ›ก๏ธ Governance & Safety

Agent Policies

Control what agents can do:

# Strict policy (whitelist only)
client.gov.upsert_policy({
    "agent_id": "readonly-agent",
    "max_actions_per_hour": 500,
    "allowed_tools": ["get_user", "search", "list_items"],
    "blocked_tools": []  # Everything else blocked
})

# Moderate policy (blacklist dangerous tools)
client.gov.upsert_policy({
    "agent_id": "support-agent",
    "max_actions_per_hour": 200,
    "max_spend_usd_per_day": 50.0,
    "blocked_tools": ["delete_user", "stripe.charge"]
})

# Per-tool limits
client.gov.upsert_policy({
    "agent_id": "billing-agent",
    "max_calls_per_tool": {
        "stripe.refund": 5,    # Max 5 refunds/day
        "send_email": 100      # Max 100 emails/day
    }
})

Policy Templates

Use pre-configured templates:

# Quick setup with sensible defaults
policy = client.gov.policy_from_template(
    agent_id="new-agent",
    template="moderate",  # strict|moderate|permissive|read_only|support_bot
    overrides={
        "max_actions_per_hour": 300,
        "blocked_tools": ["delete_user"]
    }
)

Available templates (server defaults):

  • strict
  • moderate
  • permissive
  • read_only
  • support_bot

Kill Switch

Instantly disable rogue agents:

# Emergency stop
client.gov.disable_agent(
    "rogue-agent",
    reason="Suspicious behavior detected"
)

# Re-enable after investigation
client.gov.enable_agent("rogue-agent")

Audit & Forensics

Complete action history:

# Get recent actions
logs = client.gov.agent_logs("billing-agent", limit=100)

for log in logs:
    print(f"{log.ts}: {log.tool} - {log.decision}")
    print(f"  Reason: {log.policy_reason or log.reason}")
    print(f"  Risk: {log.risk_level}")
    print(f"  Cost: ${log.spend_usd}")

# Get metrics
metrics = client.gov.agent_metrics("billing-agent", period="day")
print(f"Actions: {metrics.total_actions}")
print(f"Blocked: {metrics.blocked_actions}")
print(f"Spend: ${metrics.total_spend_usd}")
print(f"Top tools: {metrics.top_tools}")

๐Ÿ”Œ Framework Integrations

LangChain

from langchain_core.tools import tool
from onceonly import OnceOnly
from onceonly.integrations.langchain import make_idempotent_tool

client = OnceOnly(api_key="once_live_...")

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send email to user"""
    email_service.send(to=to, subject=subject, body=body)
    return f"Email sent to {to}"

# Wrap with idempotency
idempotent_send_email = make_idempotent_tool(
    send_email,
    client=client,
    key_prefix="agent:email",
    ttl=3600
)

# Use in agent
from langchain.agents import AgentExecutor, create_react_agent

agent = create_react_agent(llm, tools=[idempotent_send_email], prompt)
executor = AgentExecutor(agent=agent, tools=[idempotent_send_email])

# Agent can retry - we guarantee exactly-once execution
result = executor.invoke({"input": "Send welcome email to new@user.com"})

FastAPI

from fastapi import FastAPI, Depends, HTTPException
from onceonly import OnceOnly
import os

app = FastAPI()

def get_onceonly() -> OnceOnly:
    return OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])

@app.post("/webhooks/stripe")
async def stripe_webhook(
    event: dict,
    client: OnceOnly = Depends(get_onceonly)
):
    result = await client.check_lock_async(
        key=f"stripe:{event['id']}",
        ttl=7200,
        meta={"type": event["type"]}
    )
    
    if result.duplicate:
        return {"status": "duplicate"}
    
    await process_stripe_event(event)
    return {"status": "processed"}

๐Ÿงฐ Tools Registry (User-Owned Tools)

Register your own tools (URLs) and enforce permissions per agent.

# Register a tool (available on all plans; plan limits apply)
tool = client.gov.create_tool({
    "name": "send_email",
    "url": "https://example.com/tools/send_email",
    "scope_id": "global",
    "auth": {"type": "hmac_sha256", "secret": "your_shared_secret"},
    "timeout_ms": 15000,
    "max_retries": 2,
    "enabled": True,
    "description": "Send email to user"
})

# Toggle a tool
client.gov.toggle_tool("send_email", enabled=False)

# List tools
tools = client.gov.list_tools(scope_id="global")

Tools registry limits by plan

  • Free: 1 tool
  • Starter: 20 tools
  • Pro: 100 tools
  • Agency: 1000 tools

Rules & expectations (important)

  • name must be unique per scope_id and match ^[a-zA-Z0-9_.:-]+$
  • scope_id lets you namespace tools (e.g. global or agent:billing-agent)
  • auth.type currently supports hmac_sha256 (use a shared secret)
  • Your tool endpoint should verify HMAC and be idempotent on its side

๐Ÿ“– API Reference

Core Client

from onceonly import OnceOnly

client = OnceOnly(
    api_key="once_live_...",
    base_url="https://api.onceonly.tech/v1",  # optional
    timeout=5.0,                               # HTTP timeout
    fail_open=True,                            # graceful degradation
    max_retries_429=3,                         # auto-retry on rate limit
    retry_backoff=0.5,                         # initial backoff (seconds)
    retry_max_backoff=10.0                     # max backoff (seconds)
)

API Endpoints Map (Public)

Use this map to find the correct endpoint category quickly:

  • Core: GET /v1/me, POST /v1/me/notifications, GET /v1/usage, GET /v1/usage/all, GET /v1/events, GET /v1/metrics, POST /v1/events, GET /v1/runs/{run_id}
  • Idempotency: POST /v1/check-lock
  • AI Jobs: POST /v1/ai/run, GET /v1/ai/status, GET /v1/ai/result
  • AI Lease (local side-effects): POST /v1/ai/lease, POST /v1/ai/extend, POST /v1/ai/complete, POST /v1/ai/fail, POST /v1/ai/cancel
  • Governance (policies): POST /v1/policies/{agent_id}, POST /v1/policies/{agent_id}/from-template, GET /v1/policies, GET /v1/policies/{agent_id}
  • Governance (agents): POST /v1/agents/{agent_id}/disable, POST /v1/agents/{agent_id}/enable, GET /v1/agents/{agent_id}/logs, GET /v1/agents/{agent_id}/metrics
  • Tools Registry: POST /v1/tools, GET /v1/tools, GET /v1/tools/{tool}, POST /v1/tools/{tool}/toggle, DELETE /v1/tools/{tool}

Notification Preferences

# Global toggle + per-channel toggles (partial updates supported)
prefs = client.update_notifications(
    email_notifications_enabled=True,
    tool_error_notifications_enabled=True,
    run_failure_notifications_enabled=False,
)
print(prefs)

# Async
prefs = await client.update_notifications_async(run_failure_notifications_enabled=True)

Idempotency

# Sync
result = client.check_lock(
    key="order:12345",
    ttl=3600,           # Lock duration (seconds)
    meta={"user_id": 42}  # Optional metadata
)

# Async
result = await client.check_lock_async(key="order:12345", ttl=3600)

# Check result
if result.duplicate:
    print(f"Duplicate! First seen: {result.first_seen_at}")
else:
    print("First time - proceed with action")

AI Execution

# Long-running job (server-side)
result = client.ai.run_and_wait(
    key="report:monthly:2024-01",
    ttl=1800,                      # Job timeout (seconds)
    timeout=120.0,                 # Polling timeout
    poll_min=1.0,                  # Min poll interval
    poll_max=10.0,                 # Max poll interval
    metadata={"month": "2024-01"},
    run_id="run_report_2024_01",
)

# Governance tool runner (agent + tool)
tool_res = client.ai.run_tool(
    agent_id="billing-agent",
    tool="stripe.charge",
    args={"amount": 9999, "currency": "usd"},
    run_id="run_charge_001",
    spend_usd=0.5
)
if tool_res.allowed:
    print(tool_res.result)
else:
    print(f"Blocked: {tool_res.policy_reason}")

# Async tool runner
tool_res = await client.ai.run_tool_async(
    agent_id="billing-agent",
    tool="stripe.charge",
    args={"amount": 9999, "currency": "usd"},
    run_id="run_charge_002",
    spend_usd=0.5
)
if tool_res.allowed:
    print(tool_res.result)
else:
    print(f"Blocked: {tool_res.policy_reason}")

# Local function execution
result = client.ai.run_fn(
    key="email:welcome:user123",
    fn=lambda: send_email(...),
    ttl=300,
    wait_on_conflict=True,  # Wait if another process executing
    timeout=60.0,
    error_code="email_failed"
)

# Check status only (no polling)
status = client.ai.status("report:monthly:2024-01")
print(f"Status: {status.status}, TTL: {status.ttl_left}s")

# Get result
result = client.ai.result("report:monthly:2024-01")
if result.status == "completed":
    print(result.result)

# Run debug APIs
timeline = client.get_run_timeline("run_report_2024_01", limit=200, offset=0)
client.post_event(
    run_id="run_report_2024_01",
    type="note",
    message="manually marked for investigation",
)

Runnable example: python examples/ai/run_debug_timeline.py Failure-focused example: python examples/ai/run_debug_failure.py

AI Modes (Choose One)

Mode Use When Call Result Type
Job (server-side) Long-running tasks ai.run_and_wait(key=...) AiResult
Tool (governed) Agent tool execution ai.run_tool(agent_id=..., tool=...) AiToolResult
Local side-effects Your code does the work ai.run_fn(key=..., fn=...) AiResult

AI Result Shapes (AI-friendly)

# Tool result (governance)
AiToolResult = {
    "ok": bool,
    "allowed": bool,
    "decision": str,         # "executed" | "blocked" | "dedup"
    "policy_reason": str | None,
    "risk_level": str | None,
    "result": dict | None,
}

# Job result (run_and_wait / result)
AiResult = {
    "ok": bool,
    "status": str,           # "completed" | "failed" | "in_progress"
    "key": str,
    "result": dict | None,
    "error_code": str | None,
    "done_at": str | None,
}

Tool: Happy vs Blocked

res = client.ai.run_tool(
    agent_id="billing-agent",
    tool="stripe.refund",
    args={"charge_id": "ch_123", "amount": 500},
    spend_usd=0.2
)

if res.allowed:
    print("OK", res.result)
else:
    print("BLOCKED", res.policy_reason)

Decorators

from onceonly import idempotent, idempotent_ai

# Basic idempotency
@idempotent(client, key_prefix="payment", ttl=3600)
def process_payment(order_id: str):
    # Runs once per order_id
    stripe.charge(...)

# AI lease execution
@idempotent_ai(
    client,
    key_fn=lambda user_id: f"onboard:{user_id}",
    ttl=600,
    metadata_fn=lambda uid: {"user": uid}
)
def onboard_user(user_id: str):
    # Exactly-once, even across multiple workers
    create_account(user_id)
    send_welcome_email(user_id)
    return {"onboarded": True}

Governance

# Set policy
policy = client.gov.upsert_policy({
    "agent_id": "my-agent",
    "max_actions_per_hour": 200,
    "max_spend_usd_per_day": 50.0,
    "allowed_tools": ["tool_a", "tool_b"],
    "blocked_tools": ["dangerous_tool"],
    "max_calls_per_tool": {"tool_a": 10}
})

# From template
policy = client.gov.policy_from_template(
    agent_id="my-agent",
    template="moderate",
    overrides={"max_actions_per_hour": 300}
)

# Kill switch
status = client.gov.disable_agent("my-agent", reason="Testing")
status = client.gov.enable_agent("my-agent")

# Audit
logs = client.gov.agent_logs("my-agent", limit=100)
metrics = client.gov.agent_metrics("my-agent", period="day")

โš™๏ธ Configuration

Environment Variables

export ONCEONLY_API_KEY="once_live_..."
export ONCEONLY_BASE_URL="https://api.onceonly.tech/v1"  # optional

Fail-Open Behavior

Network/server failures don't break your app (graceful degradation):

client = OnceOnly(
    api_key="...",
    fail_open=True  # default: allows execution on timeout/5xx
)

Fail-open NEVER applies to:

  • 401/403 (auth errors) โ†’ Always blocks
  • 402 (usage limit) โ†’ Always blocks
  • 422 (validation) โ†’ Always blocks
  • 429 (rate limit) โ†’ Retries with backoff

Connection Pooling

import httpx
from onceonly import OnceOnly

# Reuse HTTP connections
sync_client = httpx.Client(
    timeout=10.0,
    limits=httpx.Limits(max_keepalive_connections=20)
)

client = OnceOnly(
    api_key="...",
    sync_client=sync_client
)

# Close when done
client.close()

Context Managers

# Auto-cleanup
with OnceOnly(api_key="...") as client:
    result = client.check_lock(key="task", ttl=300)

# Async
async with OnceOnly(api_key="...") as client:
    result = await client.check_lock_async(key="task", ttl=300)

๐Ÿšจ Common Patterns & Best Practices

โœ… DO

# โœ… Use specific, deterministic keys
key = f"payment:{order_id}:{user_id}"

# โœ… Set appropriate TTLs
ttl = 3600  # 1 hour for webhooks
ttl = 86400  # 24 hours for daily jobs

# โœ… Add metadata for debugging
meta = {"user_id": 123, "amount": 9999, "source": "web"}

# โœ… Handle duplicates gracefully
if result.duplicate:
    logger.info(f"Duplicate detected: {result.key}")
    return cached_response

# โœ… Use decorators for simplicity
@idempotent_ai(client, key_fn=lambda x: f"task:{x}")
def my_task(x): ...

โŒ DON'T

# โŒ Don't use random/timestamp in keys
key = f"payment:{uuid.uuid4()}"  # Every call is "unique"
key = f"task:{time.time()}"      # Never deduplicates

# โŒ Don't set TTL too short
ttl = 1  # Retries will leak through

# โŒ Don't ignore duplicate status
result = client.check_lock(...)
process_payment()  # Always runs!

# โŒ Don't catch and swallow errors silently
try:
    client.check_lock(...)
except: pass  # Lose safety guarantees

๐Ÿ› Troubleshooting

"Unauthorized" (401/403)

Cause: Invalid API key

# โŒ Wrong
client = OnceOnly(api_key="sk_test_...")

# โœ… Correct
client = OnceOnly(api_key="once_live_...")

"Usage limit reached" (402)

Cause: Exceeded monthly quota on the Free plan (hard cap).

Note: Paid plans (Starter/Pro/Agency) are soft-limited for monthly volume and continue to run. You can still monitor usage and upgrade at https://onceonly.tech/pricing

# Check current usage
usage = client.usage(kind="make")
print(f"Used: {usage['usage']} / {usage['limit']}")

"Rate limit exceeded" (429)

Cause: Too many requests per second

Solution: Enable auto-retry:

client = OnceOnly(
    api_key="...",
    max_retries_429=3,      # Auto-retry up to 3 times
    retry_backoff=0.5,       # Start with 0.5s delay
    retry_max_backoff=10.0   # Cap at 10s
)

Duplicates not being detected

Cause: Key is not deterministic

# โŒ Wrong: random UUID
key = f"order:{uuid.uuid4()}"

# โœ… Correct: stable identifier
key = f"order:{order_id}"

Agent blocked by policy

Cause: Policy restrictions

# Check what happened
logs = client.gov.agent_logs("my-agent", limit=10)
for log in logs:
    if log.decision == "blocked":
        print(f"Blocked: {log.tool} - {log.policy_reason or log.reason}")

# Adjust policy
client.gov.upsert_policy({
    "agent_id": "my-agent",
    "allowed_tools": ["tool_a", "tool_b", "tool_c"],  # Add tool_c
})

๐Ÿ“Š Feature Availability

Feature Free Starter Pro Agency
Core Idempotency
check_lock() 1K/mo 20K/mo 200K/mo 2M/mo
ai.run_and_wait() 3K/mo 100K/mo 1M/mo 10M/mo
Agent Governance
gov.upsert_policy() โœ… Entitlement-based โœ… Entitlement-based โœ… Entitlement-based โœ… Entitlement-based
gov.agent_logs() โŒ โŒ โœ… โœ…
gov.agent_metrics() โŒ โŒ โœ… โœ…
gov.disable_agent() (Kill switch) โŒ โŒ โœ… โœ…
gov.enable_agent() โŒ โŒ โœ… โœ…
Policy Features
Budget/action limits (max_spend_usd_per_day, max_actions_per_hour) โŒ โœ… โœ… โœ…
Tool blocklist (blocked_tools) โœ… โœ… โœ… โœ…
Tool whitelist (allowed_tools) โœ… โœ… โœ… โœ…
Per-tool limits (max_calls_per_tool) โŒ โœ… โœ… โœ…
Pricing rules (pricing_rules) โŒ โŒ โœ… โœ…
Tools Registry
gov.create_tool() + tool CRUD โœ… โœ… โœ… โœ…
Max tools per account 1 20 100 1000

๐Ÿ“ˆ Plan Limits (Defaults)

These are the default limits enforced by the API (may be configured by the server):

Plan check_lock (make) ai (runs) Default TTL Max TTL Tools Registry Limit
Free 1K / month 3K / month 60s 1h 1 tool
Starter 20K / month 100K / month 1h 24h 20 tools
Pro 200K / month 1M / month 6h 7d 100 tools
Agency 2M / month 10M / month 24h 30d 1000 tools

Notes:

  • Monthly hard-stop limit is enforced on Free.
  • Starter/Pro/Agency continue after monthly threshold (soft-limit notifications).

๐Ÿ“Š Production Checklist

Before going live:

  • Use production API key (once_live_...)
  • Set appropriate TTLs (not too short, not too long)
  • Enable auto-retry (max_retries_429=3)
  • Add metadata for debugging (meta={"user": ...})
  • Monitor usage (check client.usage() regularly)
  • Set up governance for AI agents
  • Test fail-open behavior (simulate API downtime)
  • Review audit logs periodically
  • Set up alerts for blocked actions

๐Ÿ”— Links


๐Ÿ“„ License

MIT License - see LICENSE file for details.


๐Ÿค Contributing

We welcome contributions! See CONTRIBUTING.md for guidelines.

โœ… Tests

pytest -q

Integration smoke tests (live API):

export TEST_API_KEY="once_live_..."
export TEST_BASE_URL="https://api.onceonly.tech"
pytest -q -m integration

โญ Support

If OnceOnly helps your project, give us a star on GitHub!

Questions? Open an issue or email support@onceonly.tech


Built with โค๏ธ by the OnceOnly team

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

onceonly_sdk-3.0.5.tar.gz (44.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

onceonly_sdk-3.0.5-py3-none-any.whl (31.1 kB view details)

Uploaded Python 3

File details

Details for the file onceonly_sdk-3.0.5.tar.gz.

File metadata

  • Download URL: onceonly_sdk-3.0.5.tar.gz
  • Upload date:
  • Size: 44.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.7

File hashes

Hashes for onceonly_sdk-3.0.5.tar.gz
Algorithm Hash digest
SHA256 f12549463a4a1bf6222c1ec2bf549e97a778cb19fe61380cc6a68f52de19700d
MD5 9d8d953c9c4066aef5e987528daaedd3
BLAKE2b-256 986f9d5d08f0d868df30e6daec2d94020ac54cda787c1b8c0ef09fef4526195f

See more details on using hashes here.

Provenance

The following attestation bundles were made for onceonly_sdk-3.0.5.tar.gz:

Publisher: release.yml on OnceOnly-Tech/onceonly-python

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file onceonly_sdk-3.0.5-py3-none-any.whl.

File metadata

  • Download URL: onceonly_sdk-3.0.5-py3-none-any.whl
  • Upload date:
  • Size: 31.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.7

File hashes

Hashes for onceonly_sdk-3.0.5-py3-none-any.whl
Algorithm Hash digest
SHA256 6e46c12a276f0d07c53b10698e16d4c9a508bdce6ac3a595daf2af222ff33b18
MD5 9199cb5002e9966bb4beff91e86158db
BLAKE2b-256 b94445ece88bcc69b3874be63f16a04b6b863acaca6ab2a10864a8c4a7f20963

See more details on using hashes here.

Provenance

The following attestation bundles were made for onceonly_sdk-3.0.5-py3-none-any.whl:

Publisher: release.yml on OnceOnly-Tech/onceonly-python

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page