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 Python 3.8+ License: MIT

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)
  • 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)
  • Decorator version: @idempotent or @idempotent_ai

Async equivalents

  • check_lock_async
  • ai.run_and_wait_async
  • ai.run_tool_async
  • ai.run_fn_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 (requires Pro or Agency)
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

  • Pro: 10 tools
  • Agency: 500 tools

Note: Tools registry is not available on Free/Starter.

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, GET /v1/usage, GET /v1/usage/all, GET /v1/events, GET /v1/metrics
  • 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}

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"}
)

# Governance tool runner (agent + tool)
tool_res = client.ai.run_tool(
    agent_id="billing-agent",
    tool="stripe.charge",
    args={"amount": 9999, "currency": "usd"},
    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"},
    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)

### 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)

```python
# 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

```python
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 for your plan

Solution: 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() โŒ โŒ โœ… Limited โœ… Full
gov.agent_logs() โŒ โŒ โœ… โœ…
gov.agent_metrics() โŒ โŒ โœ… โœ…
gov.disable_agent() (Kill switch) โŒ โŒ โŒ โœ…
gov.enable_agent() โŒ โŒ โŒ โœ…
Policy Features
Budget limits (max_spend_usd_per_day) โŒ โŒ โœ… โœ…
Tool blocklist (blocked_tools) โŒ โŒ โœ… โœ…
Tool whitelist (allowed_tools) โŒ โŒ โŒ โœ…
Per-tool limits (max_calls_per_tool) โŒ โŒ โœ… โœ…

Pro Plan: Limited governance (no allowed_tools whitelist, no kill switch)
Agency Plan: Full governance (whitelist, kill switch, anomaly detection)

๐Ÿ“ˆ 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 Not available
Starter 20K / month 100K / month 1h 24h Not available
Pro 200K / month 1M / month 6h 7d 10 tools
Agency 2M / month 10M / month 24h 30d 500 tools

Pro vs Agency differences (important):

  • Pro: Governance is limited (no allowed_tools whitelist, no kill switch).
  • Agency: Full governance, including tool whitelist + kill switch.

๐Ÿ“Š 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.2.tar.gz (40.1 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.2-py3-none-any.whl (29.2 kB view details)

Uploaded Python 3

File details

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

File metadata

  • Download URL: onceonly_sdk-3.0.2.tar.gz
  • Upload date:
  • Size: 40.1 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.2.tar.gz
Algorithm Hash digest
SHA256 39b05caa558a472d80600c03a9c7b0d4d9fa6260f65f1aa273c2024d2a8842fb
MD5 1a4f7c198b971046bec7aba1485b710b
BLAKE2b-256 ab8274795f2e813f916669852851a76a8b2ac238536b2578b0fab95d5384835e

See more details on using hashes here.

Provenance

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

Publisher: release.yml on mykolademyanov/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.2-py3-none-any.whl.

File metadata

  • Download URL: onceonly_sdk-3.0.2-py3-none-any.whl
  • Upload date:
  • Size: 29.2 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.2-py3-none-any.whl
Algorithm Hash digest
SHA256 a1c33fa921fb39fc4b8dcdba877d1f8ea2407ccae5ba5665b2201e4988065a3c
MD5 9bc7110271adcf691711b2ba273d8285
BLAKE2b-256 c21274abf7cafaa44032084880c6375480fa63aef38fa0e1eaea42acaf2a6193

See more details on using hashes here.

Provenance

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

Publisher: release.yml on mykolademyanov/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