The official Python library for the Verne Nautilus platform.
Server-side only. API keys carry full service access and must never be used in client-side or browser contexts.
Python 3.9 or later.
pip install vernesoftfrom vernesoft import Verne
verne = Verne(
relay=os.environ["VERNE_RELAY_KEY"],
gate=os.environ["VERNE_GATE_KEY"],
)You can also instantiate services independently if you only need one:
from vernesoft import Relay, Gate
relay = Relay(api_key=os.environ["VERNE_RELAY_KEY"])
gate = Gate(api_key=os.environ["VERNE_GATE_KEY"])Send events to all subscribed endpoints:
msg = verne.relay.messages.send(
event_type="user.created",
payload={"id": "usr_123"},
)Optional parameters:
msg = verne.relay.messages.send(
event_type="order.placed",
payload={"order_id": "999"},
idempotency_key="evt_abc", # prevent duplicate delivery within 24h
channels=["team-a"], # restrict to specific endpoint channels
)List previously sent events:
page = verne.relay.messages.list(limit=20, event_type="user.created")
print(page.data) # list[Message]
print(page.has_more) # bool
print(page.next_cursor) # pass to the next call to paginateManage your end-users. The tenant_id is automatically scoped to your API key.
# Create a user
identity = verne.gate.identities.create(
schema_id="user",
traits={
"email": "user@example.com",
"custom_data": {"role": "editor"},
},
credentials={"password": {"config": {"password": "StrongPassword123!"}}},
state="active",
)
# Get a user
verne.gate.identities.get(identity.id)
# Update a user (JSON Patch — RFC 6902)
verne.gate.identities.patch(identity.id, [
{"op": "replace", "path": "/traits/custom_data/role", "value": "admin"},
])
# Delete a user
verne.gate.identities.delete(identity.id)
# Activate / deactivate a user (an inactive user cannot log in)
verne.gate.identities.deactivate(identity.id)
verne.gate.identities.activate(identity.id)
# …or set the state explicitly:
verne.gate.identities.set_state(identity.id, "inactive")
# Resend the email verification link
verne.gate.identities.resend_verification(identity.id)Read or replace the tenant's security settings (passwordless login, TOTP MFA):
security = verne.gate.settings.get_security()
# security.passwordless_enabled, security.mfa_enabled
# Both fields are required — the update is a full replacement, not a merge.
verne.gate.settings.update_security(passwordless_enabled=True, mfa_enabled=False)Exchange your long-lived API key for a short-lived access token:
token = verne.gate.tokens.create(
subject="usr_123",
scopes=["gate.tokens.read"], # optional
ttl_seconds=3600, # optional, default 3600, max 86400
)
# token.access_token — attach to downstream requests
# token.expires_at — ISO 8601 expiryValidate a token:
info = verne.gate.tokens.introspect(token.access_token)
if not info.active:
# token is expired or invalid
passCheck whether a subject is allowed to perform an action:
decision = verne.gate.authorize(
subject="usr_123",
action="relay.messages.read",
resource="tenant:ten_001",
)
if not decision.allowed:
raise PermissionError("Forbidden")Schedule recurring HTTP callbacks with cron expressions, or one-shot jobs at a
future time. The tenant_id is automatically scoped to your API key.
# List all cron jobs
jobs = verne.clockwork.jobs.list() # list[CronJob]
# Create a recurring job
job = verne.clockwork.jobs.create(
name="nightly-report",
schedule="0 2 * * *", # standard cron expression
url="https://example.com/hooks/report",
method="POST", # optional, defaults server-side
headers={"X-Api-Key": "secret"}, # optional
body='{"kind":"report"}', # optional
)
# Partially update a job — only the fields you pass are changed
verne.clockwork.jobs.update(job.id, schedule="0 3 * * *", is_active=False)
# Inspect execution history
executions = verne.clockwork.jobs.executions(job.id) # list[Execution]
# Delete a job
verne.clockwork.jobs.delete(job.id)One-shot jobs that fire once at run_at:
# Schedule a delayed job
delayed = verne.clockwork.delayed.create(
name="welcome-email",
run_at="2026-07-26T09:00:00Z", # ISO 8601 timestamp
url="https://example.com/send",
body='{"template":"welcome"}', # optional
)
# List pending / completed delayed jobs
verne.clockwork.delayed.list() # list[DelayedJob]
# Inspect execution history
verne.clockwork.delayed.executions(delayed.id) # list[Execution]
# Cancel a pending delayed job
verne.clockwork.delayed.cancel(delayed.id)Every client has an async counterpart — AsyncVerne, AsyncRelay, AsyncGate, AsyncClockwork — with the same interface, where all methods are coroutines:
from vernesoft import AsyncVerne
verne = AsyncVerne(
relay=os.environ["VERNE_RELAY_KEY"],
gate=os.environ["VERNE_GATE_KEY"],
)
msg = await verne.relay.messages.send(event_type="user.created", payload={"id": "usr_123"})
page = await verne.relay.messages.list(limit=10)
identity = await verne.gate.identities.create(schema_id="user", traits={"email": "a@b.com"})
token = await verne.gate.tokens.create(subject="usr_123")
decision = await verne.gate.authorize(subject="usr_123", action="relay.messages.read", resource="tenant:ten_001")All API errors raise VerneAPIError with structured fields:
from vernesoft import VerneAPIError, VerneError
try:
verne.relay.messages.send(event_type="ping", payload={})
except VerneAPIError as e:
print(e.code) # e.g. 'invalid_payload', 'unauthorized'
print(e.status) # HTTP status code
print(e.request_id) # include in support requests
except VerneError as e:
# network failure or timeout
print(e)Both Verne and the per-service clients accept an optional timeout in seconds (default 30):
verne = Verne(
relay=os.environ["VERNE_RELAY_KEY"],
timeout=10,
)Individual requests can be cancelled by passing an httpx timeout or by closing the client. Per-request timeouts are supported natively through httpx:
import httpx
from vernesoft import Relay
relay = Relay(api_key=os.environ["VERNE_RELAY_KEY"], timeout=5)