Python SDK for OnceOnly idempotency API
Project description
OnceOnly Python SDK
AI Agent Execution & Governance Layer
Exactly-once execution + runtime safety + agent control plane.
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:
@idempotentor@idempotent_ai
Async equivalents
check_lock_asyncai_run_asyncai.run_and_wait_asyncai.run_tool_asyncai.run_fn_asyncpost_event_asyncget_run_timeline_asyncupdate_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):
strictmoderatepermissiveread_onlysupport_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)
namemust be unique perscope_idand match^[a-zA-Z0-9_.:-]+$scope_idlets you namespace tools (e.g.globaloragent:billing-agent)auth.typecurrently supportshmac_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
- Website: https://onceonly.tech
- Documentation: https://docs.onceonly.tech
- API Reference: https://docs.onceonly.tech/reference/idempotency/
- Python SDK Docs: https://docs.onceonly.tech/sdk/python
- GitHub: https://github.com/OnceOnly-Tech/onceonly-python
- PyPI: https://pypi.org/project/onceonly-sdk/
- Support: support@onceonly.tech
๐ 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
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
f12549463a4a1bf6222c1ec2bf549e97a778cb19fe61380cc6a68f52de19700d
|
|
| MD5 |
9d8d953c9c4066aef5e987528daaedd3
|
|
| BLAKE2b-256 |
986f9d5d08f0d868df30e6daec2d94020ac54cda787c1b8c0ef09fef4526195f
|
Provenance
The following attestation bundles were made for onceonly_sdk-3.0.5.tar.gz:
Publisher:
release.yml on OnceOnly-Tech/onceonly-python
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
onceonly_sdk-3.0.5.tar.gz -
Subject digest:
f12549463a4a1bf6222c1ec2bf549e97a778cb19fe61380cc6a68f52de19700d - Sigstore transparency entry: 1202749227
- Sigstore integration time:
-
Permalink:
OnceOnly-Tech/onceonly-python@48a23a70f14468d1bf0e0c5dd355077b86d1f3f2 -
Branch / Tag:
refs/tags/v3.0.5 - Owner: https://github.com/OnceOnly-Tech
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
release.yml@48a23a70f14468d1bf0e0c5dd355077b86d1f3f2 -
Trigger Event:
release
-
Statement type:
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
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
6e46c12a276f0d07c53b10698e16d4c9a508bdce6ac3a595daf2af222ff33b18
|
|
| MD5 |
9199cb5002e9966bb4beff91e86158db
|
|
| BLAKE2b-256 |
b94445ece88bcc69b3874be63f16a04b6b863acaca6ab2a10864a8c4a7f20963
|
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
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
onceonly_sdk-3.0.5-py3-none-any.whl -
Subject digest:
6e46c12a276f0d07c53b10698e16d4c9a508bdce6ac3a595daf2af222ff33b18 - Sigstore transparency entry: 1202749236
- Sigstore integration time:
-
Permalink:
OnceOnly-Tech/onceonly-python@48a23a70f14468d1bf0e0c5dd355077b86d1f3f2 -
Branch / Tag:
refs/tags/v3.0.5 - Owner: https://github.com/OnceOnly-Tech
-
Access:
public
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
release.yml@48a23a70f14468d1bf0e0c5dd355077b86d1f3f2 -
Trigger Event:
release
-
Statement type: