Python SDK Reference

The official Python SDK for Verity. Async-first, built on httpx. Requires Python 3.10+. Identical behavior and guarantees as the TypeScript SDK.

pip install verityinc-sdk

VerityClient

The main entry point. Create one instance and reuse it throughout your application.

from verity import VerityClient

verity = VerityClient(
    base_url="https://api.useverity.io/v1",
    api_key=os.environ["VERITY_API_KEY"],
    namespace="payments",
)

Constructor Parameters

Parameter	Type	Default	Description
`base_url`	`str`	required	Verity API base URL
`api_key`	`str`	required	API key (`vt_live_` or `vt_test_`)
`namespace`	`str \| None`	`None`	Default namespace for all `protect()` calls
`agent_id`	`str \| None`	`None`	Default agent/worker identifier
`auto_renew`	`bool`	`True`	Automatically renew leases in the background
`renew_at_fraction`	`float`	`0.65`	Fraction of lease duration at which renewal fires (0.3–0.9)
`request_timeout_s`	`float`	`20.0`	Timeout for individual HTTP requests in seconds
`conflict_retry`	`ConflictRetryConfig`	See below	Retry behavior for 409 Conflict
`logger`	`logging.Logger \| None`	`None`	Custom logger. If `None`, uses `logging.getLogger("verity")`

ConflictRetryConfig

from verity import ConflictRetryConfig

config = ConflictRetryConfig(
    enabled=True,           # default
    max_attempts=12,        # default
    initial_delay_s=0.5,    # default
    max_delay_s=15.0,       # default
    jitter=True,            # default — ±30% random jitter
)

protect()

Protect a standalone effect. Async method — must be awaited.

result = await verity.protect(effect_key, observe=..., act=..., **params)

Parameters

Parameter	Type	Description
`effect_key`	`str`	Unique key identifying this effect instance
`observe`	`Callable[[], Awaitable[T \| None]]`	Optional. Check external system for prior completion.
`act`	`Callable[[], Awaitable[T]]`	Required. Execute the real-world action.
`namespace`	`str \| None`	Namespace override for this call
`effect_name`	`str \| None`	Human-readable name
`agent_id`	`str \| None`	Agent identifier override
`lease_duration_ms`	`int \| None`	Lease duration in milliseconds (5,000–120,000)
`input_json`	`Any`	Input data for debugging (max 64 KB)
`on_conflict`	`"retry" \| "throw"`	`"throw"` raises `LeaseConflictError` immediately
`on_lease`	`Callable[[LeaseResponse], None]`	Callback after lease is acquired
`key_suffix`	`str \| None`	Appended to effect key for cardinality

Example

import stripe
from verity import VerityClient

verity = VerityClient(
    base_url="https://api.useverity.io/v1",
    api_key=os.environ["VERITY_API_KEY"],
    namespace="payments",
)

async def check_existing_refund():
    existing = stripe.Refund.list(charge=charge_id, limit=1)
    return existing.data[0] if existing.data else None

async def execute_refund():
    return stripe.Refund.create(
        charge=charge_id,
        amount=4999,
        reason="requested_by_customer",
    )

refund = await verity.protect(
    "refund:order_48392",
    observe=check_existing_refund,
    act=execute_refund,
    effect_name="stripe.refund",
    lease_duration_ms=60_000,
    input_json={"order_id": order_id, "charge_id": charge_id},
)

Workflow Builder

Identical to the TypeScript workflow builder, but with Pythonic naming:

run = verity.workflow("refund_flow").case("order_123").run()

await run.protect(
    "validate_order",
    act=validate_order,
)

await run.protect(
    "process_refund",
    observe=check_existing_refund,
    act=execute_refund,
)

await run.protect(
    "notify_customer",
    act=send_notification_email,
)

run() Options

Parameter	Type	Default	Description
`run_id`	`str \| None`	Auto-generated	Explicit run ID
`namespace`	`str \| None`	`workflow_name`	Namespace override
`lease_duration_ms`	`int \| None`	`30000`	Default lease duration for the run
`agent_id`	`str \| None`	`None`	Agent identifier for the run

Low-Level API

For manual control over the lease lifecycle:

# Request a lease
lease = await verity.request_lease("payments", LeaseRequestBody(
    effect_key="refund:order_123",
    effect_name="stripe.refund",
))

# Commit a result
await verity.commit("payments", CommitRequestBody(
    effect_key="refund:order_123",
    fence_token=lease["fenceToken"],
    lease_token=lease["leaseToken"],
    result={"refund_id": "re_123"},
    source="acted",
))

# Record a failure
await verity.fail("payments", FailRequestBody(
    effect_key="refund:order_123",
    fence_token=lease["fenceToken"],
    lease_token=lease["leaseToken"],
    error={"message": "Stripe declined"},
    source="acted",
))

# Renew a lease
await verity.renew("payments", RenewRequestBody(
    effect_key="refund:order_123",
    fence_token=lease["fenceToken"],
    lease_token=lease["leaseToken"],
))

Error Classes

All errors extend VerityError. Identical semantics to the TypeScript SDK:

from verity import (
    VerityError,                    # Base class for all Verity errors
    VerityApiError,                 # Non-2xx API response
    LeaseConflictError,             # 409 — another agent holds the lease
    EffectPreviouslyFailedError,    # Effect previously failed — reset required
    CommitUncertainError,           # Action succeeded but commit failed — CRITICAL
    VerityConfigError,              # Missing required config
    VerityValidationError,          # Input data invalid or too large
)

Catching Errors

from verity import (
    CommitUncertainError,
    EffectPreviouslyFailedError,
    LeaseConflictError,
)

try:
    result = await verity.protect("refund:order_123", act=execute_refund)
except CommitUncertainError as e:
    # CRITICAL: action succeeded, commit failed
    # DO NOT RETRY — check Explorer to reconcile
    logger.critical(
        f"Commit uncertain for {e.effect_key}. "
        f"Result: {e.result}. Commit error: {e.commit_error}"
    )
    alert_ops_team(e)
except EffectPreviouslyFailedError as e:
    # Effect failed before — admin must reset in Explorer
    logger.warning(f"Effect {e.effect_key} previously failed: {e.cached_error}")
except LeaseConflictError as e:
    # Another agent is processing this effect
    logger.info(f"Effect {e.effect_key} is being processed by another agent")

See Error Handling for complete details on each error class.

Exports

from verity import (
    # Client
    VerityClient,

    # Config
    ConflictRetryConfig,

    # Errors
    VerityError,
    VerityApiError,
    VerityConfigError,
    VerityValidationError,
    LeaseConflictError,
    EffectPreviouslyFailedError,
    CommitUncertainError,

    # Request body types (for low-level API)
    LeaseRequestBody,
    CommitRequestBody,
    FailRequestBody,
    RenewRequestBody,
    ReportObserveBody,
)

TypeScript ↔ Python Naming

Both SDKs are feature-identical. The only differences are Pythonic naming conventions:

TypeScript	Python
`baseUrl`	`base_url`
`apiKey`	`api_key`
`agentId`	`agent_id`
`autoRenew`	`auto_renew`
`renewAtFraction`	`renew_at_fraction`
`requestTimeoutMs`	`request_timeout_s` (seconds, not ms)
`conflictRetry`	`conflict_retry`
`effectKey`	`effect_key`
`effectName`	`effect_name`
`leaseDurationMs`	`lease_duration_ms`
`inputJson`	`input_json`
`onConflict`	`on_conflict`
`onLease`	`on_lease`
`keySuffix`	`key_suffix`