Python SDK

Supports sync and async. Requires Python 3.9+.

Install

pip install vengtoo

Source: github.com/vengtoo/vengtoo-python

Quick start

from vengtoo import Vengtoo, Subject, Resource, Action

client = Vengtoo(api_key="azx_...")

decision = client.check(
    subject=Subject(id="user-123", type="user"),
    action=Action(name="read"),
    resource=Resource(id="doc-456", type="document"),
)

Using with the local agent

client = Vengtoo(base_url="http://localhost:8181")

Full response

from vengtoo import AuthorizeRequest

resp = client.authorize(AuthorizeRequest(
    subject=Subject(id="user-123", type="user"),
    resource=Resource(id="doc-456", type="document"),
    action=Action(name="read"),
    context={"ip": "10.0.0.1"},
))
# resp.decision, resp.context.reason, resp.context.policy_id, resp.context.access_path

Async

decision = await client.async_check(
    subject=Subject(id="user-123", type="user"),
    action=Action(name="read"),
    resource=Resource(id="doc-456", type="document"),
)

resp = await client.async_authorize(request)

FastAPI

from fastapi import FastAPI, Depends
from vengtoo import Vengtoo

app = FastAPI()
vengtoo = Vengtoo(api_key="azx_...")

@app.get("/documents/{id}")
async def get_doc(id: str, _=Depends(vengtoo.require("document", "read"))):
    return {"id": id}

Extracts subject ID from the X-User-ID header by default. Customize:

vengtoo.require("document", "read", subject_header="authorization-user-id")

Options

Vengtoo(
    api_key="azx_...",
    base_url="http://localhost:8181",
    timeout=5.0,        # seconds, default 10
    max_retries=3,      # default 2
)

Error handling

from vengtoo import Vengtoo, VengtooError

try:
    client.check(subject, "read", resource)
except VengtooError as e:
    if e.is_auth_error:
        # 401 — invalid API key
        pass
    if e.is_server_error:
        # 5xx — already retried
        pass
    print(e.status_code, e.message)

Client lifecycle

The Python SDK reuses HTTP connections. Close the client when done:

client = Vengtoo(api_key="azx_...")
# ... use client ...
client.close()

# Or for async:
await client.async_close()

Types

Type	Fields
Subject	type (required), id, external_id, properties
Resource	type (required), id, external_id, properties. If neither id nor external_id is set, "*" is sent automatically for type-level policy matching.
Action	name
AuthorizeRequest	subject, resource, action, context
AuthorizeResponse	decision, context
ResponseContext	reason, policy_id, access_path