Python SDK

Python SDK

The official Python SDK is published as onceonly-sdk on PyPI.

Install

pip install onceonly-sdk

Create a client

import os
from onceonly import OnceOnly

client = OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"])

If you self-host, pass a custom base_url. The SDK accepts either the root URL or a URL that already includes /v1:

client = OnceOnly(
    api_key=os.environ["ONCEONLY_API_KEY"],
    base_url="http://localhost:8080",
)

You can also configure timeout, fail-open behavior for check_lock(), and retry/backoff:

client = OnceOnly(
    api_key=os.environ["ONCEONLY_API_KEY"],
    timeout=5.0,
    fail_open=True,
    max_retries_429=3,
    retry_backoff=0.5,
    retry_max_backoff=10.0,
)

You can also use context managers:

from onceonly import OnceOnly

with OnceOnly(api_key="once_live_xxx") as client:
    stats = client.usage_all()

from onceonly import OnceOnly

async with OnceOnly(api_key="once_live_xxx") as client:
    stats = await client.usage_all_async()

---

Idempotency (check-lock)

lock = client.check_lock(
    key="payment:invoice:INV-123",
    ttl=3600,
    meta={"invoice_id": "INV-123", "amount_usd": 99.99},
)

if lock.duplicate:
    # Already executed within TTL window.
    raise RuntimeError(f"Duplicate (first_seen_at={lock.first_seen_at})")

# Safe to run the real side-effect here.
charge_customer()

Async variant:

lock = await client.check_lock_async(
    key="payment:invoice:INV-123",
    ttl=3600,
    meta={"invoice_id": "INV-123"},
)

---

AI Runs and Leases

Keyed background run

Use ai.run() when you want OnceOnly to start or attach to keyed work.

run = client.ai.run(
    key="support_chat:abc123",
    ttl=1800,
    metadata={
        "run_id": "run_support_001",
        "agent_id": "support_bot",
    },
)

print(run.status, run.lease_id, run.version)

Wait for final result

final = client.ai.run_and_wait(
    key="support_chat:abc123",
    ttl=1800,
    metadata={
        "run_id": "run_support_001",
        "agent_id": "support_bot",
    },
    timeout=120,
)

print(final.status, final.error_code, final.result)

Governed tool execution

Use run_tool() for the agent_id + tool + args flow:

res = client.ai.run_tool(
    agent_id="support_bot",
    tool="send_email",
    args={
        "to": "user@example.com",
        "subject": "Hello",
    },
    run_id="run_support_001",
    spend_usd=0.001,
)

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

Local exactly-once execution with your own function

result = client.ai.run_fn(
    key="email:welcome:user_123",
    fn=lambda: {"sent": True},
    ttl=300,
    metadata={"tool": "send_welcome_email"},
    wait_on_conflict=True,
    timeout=60,
)

print(result.status, result.result)

Low-level lease helpers

The SDK also wraps low-level lease endpoints directly:

lease = client.ai.lease(key="support_chat:abc123", ttl=1800, metadata={"chat_id": "abc123"})
status = client.ai.status("support_chat:abc123")
result = client.ai.result("support_chat:abc123")

client.ai.extend(key="support_chat:abc123", lease_id=lease["lease_id"], ttl=1800)
client.ai.complete(key="support_chat:abc123", lease_id=lease["lease_id"], result={"status": "done"})
# or
client.ai.fail(key="support_chat:abc123", lease_id=lease["lease_id"], error_code="chat_failed")
# optional cancel path
client.ai.cancel(key="support_chat:abc123", lease_id=lease["lease_id"], reason="manual_stop")

---

Governance (policies, tools, agent controls)

Policies

policy = client.gov.upsert_policy({
    "agent_id": "support_bot",
    "allowed_tools": ["send_email", "create_ticket"],
    "blocked_tools": ["delete_user"],
    "max_actions_per_hour": 100,
    "max_spend_usd_per_day": 50.0,
})
print(policy.agent_id, policy.max_actions_per_hour)

templated = client.gov.policy_from_template(
    agent_id="new_support_bot",
    template="moderate",
    overrides={"blocked_tools": ["delete_user"]},
)
print(templated.agent_id, templated.policy)

all_policies = client.gov.list_policies()
one_policy = client.gov.get_policy("support_bot")

Agent kill switch and observability

client.gov.disable_agent("support_bot", reason="incident_123")
client.gov.enable_agent("support_bot", reason="incident_resolved")

logs = client.gov.agent_logs("support_bot", limit=100)
metrics = client.gov.agent_metrics("support_bot", period="day")

print(len(logs), metrics.total_actions, metrics.blocked_actions)

Tools Registry CRUD

import os

tool = client.gov.create_tool({
    "name": "send_email",
    "scope_id": "global",
    "url": "https://your-domain.com/tools/send-email",
    "auth": {"type": "hmac_sha256", "secret": os.environ["TOOL_SECRET"]},
    "timeout_ms": 15000,
    "max_retries": 2,
    "enabled": True,
    "description": "Send transactional email",
})
print(tool["name"], tool["enabled"])

List, fetch, toggle, and delete:

tools = client.gov.list_tools(scope_id="global")
tool = client.gov.get_tool("send_email", scope_id="global")
disabled = client.gov.toggle_tool("send_email", enabled=False, scope_id="global")
deleted = client.gov.delete_tool("send_email", scope_id="global")

See also: Policy Templates | Implementing Tool Backends

---

Account, Usage, Events, and Metrics

profile = client.me()
prefs = client.update_notifications(
    email_notifications_enabled=True,
    tool_error_notifications_enabled=True,
    run_failure_notifications_enabled=False,
)

make_usage = client.usage("make")
ai_usage = client.usage("ai")
all_usage = client.usage_all()
summary = client.metrics("2026-04-01", "2026-04-30")

print(profile.get("email"))
print(make_usage.get("usage"), ai_usage.get("usage"), all_usage.get("plan"))
print(summary)

Structured run events:

event = client.post_event(
    run_id="run_support_001",
    type="tool_call",
    status="start",
    step="step_1",
    tool="send_email",
    agent_id="support_bot",
)

timeline = client.get_run_timeline("run_support_001")
recent_events = client.events(limit=20)

print(event["event_id"])
print(timeline["total"])
print(len(recent_events))

The current SDK wraps:

• me() / me_async()
• update_notifications() / update_notifications_async()
• usage() / usage_async()
• usage_all() / usage_all_async()
• metrics() / metrics_async()
• post_event() / post_event_async()
• get_run_timeline() / get_run_timeline_async()
• events() / events_async()

The current REST API also has GET /v1/runs, but there is no dedicated list_runs() helper in the SDK yet.

---

Async Methods

import os
from onceonly import OnceOnly

async def main():
    async with OnceOnly(api_key=os.environ["ONCEONLY_API_KEY"]) as client:
        lock = await client.check_lock_async(
            key="payment:invoice:INV-123",
            ttl=3600,
            meta={"invoice_id": "INV-123"},
        )

        tool = await client.gov.create_tool_async({
            "name": "send_email",
            "scope_id": "global",
            "url": "https://your-domain.com/tools/send-email",
            "auth": {"type": "hmac_sha256", "secret": os.environ["TOOL_SECRET"]},
            "timeout_ms": 15000,
            "max_retries": 2,
        })

        final = await client.ai.run_and_wait_async(
            key="support_chat:abc123",
            ttl=1800,
            metadata={
                "run_id": "run_support_001",
                "agent_id": "support_bot",
            },
            timeout=120,
        )

        usage = await client.usage_all_async()
        timeline = await client.get_run_timeline_async("run_support_001")

        print(lock.duplicate, tool["name"], final.status, usage["plan"], timeline["total"])

Common async helpers include:

• check_lock_async()
• me_async() / update_notifications_async()
• usage_async() / usage_all_async() / metrics_async()
• events_async() / post_event_async() / get_run_timeline_async()
• client.ai.run_async() / run_and_wait_async() / run_tool_async() / run_fn_async()
• client.ai.lease_async() / extend_async() / complete_async() / fail_async() / cancel_async()
• client.gov.upsert_policy_async() / policy_from_template_async() / list_policies_async() / get_policy_async()
• client.gov.create_tool_async() / list_tools_async() / get_tool_async() / toggle_tool_async() / delete_tool_async()
• client.gov.disable_agent_async() / enable_agent_async() / agent_logs_async() / agent_metrics_async()

---

Client Aliases

The OnceOnly client exposes convenience aliases for AI runs:

• ai_run(...) / ai_run_async(...)
• ai_run_and_wait(...) / ai_run_and_wait_async(...)

They delegate to client.ai.run(...) and client.ai.run_and_wait(...).

---

Decorators

The package exports two decorators from onceonly:

• @idempotent(...) for check-lock
• @idempotent_ai(...) for exactly-once local execution with AI leases

@idempotent

from onceonly import OnceOnly, idempotent

client = OnceOnly(api_key="once_live_xxx")

@idempotent(
    client,
    key_prefix="payments",
    ttl=3600,
    on_duplicate=lambda invoice_id: {"status": "duplicate", "invoice_id": invoice_id},
)
def charge_invoice(invoice_id: str) -> dict:
    return {"status": "paid", "invoice_id": invoice_id}

@idempotent_ai

from onceonly import OnceOnly, idempotent_ai

client = OnceOnly(api_key="once_live_xxx")

@idempotent_ai(
    client,
    key_fn=lambda chat_id, messages: f"support_chat:{chat_id}",
    ttl=1800,
    metadata_fn=lambda chat_id, messages: {"chat_id": chat_id, "message_count": len(messages)},
)
def handle_support_chat(chat_id: str, messages: list[dict]) -> dict:
    return {"status": "resolved", "messages": len(messages)}

result = handle_support_chat("chat_123", [{"role": "user", "content": "Help"}])
print(result.status, result.result)

idempotent_ai returns an AiResult, not the raw function output, because completion is normalized through the lease/result flow.

---

See also: AI Run API | Runs & Events API | Usage API