Vengtoo Python SDK

Python client for Vengtoo — works with both Vengtoo Cloud and the Vengtoo Agent.

Supports sync and async. Requires Python 3.10+. One dependency (httpx).

Install

pip install vengtoo

Usage

Cloud Mode

from vengtoo import Vengtoo, Subject, Resource

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

allowed = client.check(
    subject=Subject(id="user:123", type="user"),
    action="read",
    resource=Resource(type="document", id="doc:456"),
)

OAuth2 Client Credentials

For service-to-service auth, pass client_id and client_secret (secret is prefixed vgt_cs_). The SDK exchanges credentials at the token endpoint, caches the JWT in memory, refreshes ~60s before expiry, and retries once automatically on a 401. Sync and async calls share the same cache.

client = Vengtoo(
    client_id="my-client-id",
    client_secret="vgt_cs_...",
)

Equivalent curl for the underlying token exchange:

curl -X POST https://api.vengtoo.com/v1/oauth/token \
  -d grant_type=client_credentials \
  -d client_id=my-client-id \
  -d client_secret=vgt_cs_...

Providing both api_key and OAuth credentials is rejected at construction. A bad client_id / client_secret surfaces as VengtooOAuthError (distinct from VengtooError) with a message pointing you at the OAuth exchange.

Agent Mode (local)

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

Full Evaluation Response

from vengtoo import EvaluationRequest, Action

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

Batch Evaluation

Evaluate up to 50 checks in one round-trip (AuthZEN 1.0). Top-level fields act as defaults that individual items inherit:

from vengtoo import BatchEvaluationRequest, BatchEvalItem

resp = client.evaluate_batch(BatchEvaluationRequest(
    subject=Subject(id="user:123", type="user"),  # default for all items
    action=Action(name="read"),
    evaluations=[
        BatchEvalItem(resource=Resource(type="document", id="doc:1")),
        BatchEvalItem(resource=Resource(type="document", id="doc:2")),
    ],
))
# resp.evaluations[i].decision, positional

Search (List Filtering)

Instead of checking one resource at a time, ask "which resources can this subject act on?" (AuthZEN Search). search_resource(), search_subject(), and search_action() search each dimension (each has an async_ variant). The response filter is a UCAST-style filter object (Vengtoo's own condition tree) — returned as received to apply in your own query layer:

from vengtoo import SearchRequest, SearchOptions

resp = client.search_resource(SearchRequest(
    subject=Subject(id="alice", type="user"),
    action=Action(name="read"),
    resource=Resource(type="document"),   # optional type template
    options=SearchOptions(return_="filter"),  # "filter" (default) | "results" | "both"
))
# resp.filter (UCAST-style tree — apply it in your own query layer),
# resp.results (present for "results"/"both"), resp.context.reason

Translate the filter to SQL

ucast_to_sql turns the filter into a parameterized SQL WHERE clause you run against your own database. It is a Postgres reference translator ($1, $2, … placeholders, regex → ~); values are always bound parameters, never interpolated. Pass a field→column map when your columns differ:

from vengtoo import ucast_to_sql

resp = client.search_resource(SearchRequest(...))
where, params = ucast_to_sql(resp.filter, {"owner": "owner_id"})
# where == "(owner_id = $1) AND (status != $2)"
rows = db.execute(f"SELECT * FROM documents WHERE {where}", params)

Human-in-the-Loop Approvals

When a policy requires human approval, evaluate() returns reason_code == "authorization_pending". evaluate_with_approval() handles the wait — polling at the server-recommended interval until a human approves or denies in the Vengtoo dashboard:

resp = client.evaluate_with_approval(
    req,
    timeout=300,
    on_pending=lambda auth_req_id, expires_in:
        print(f"waiting for approval {auth_req_id} (expires in {expires_in}s)"),
)
# Terminal reason_codes: "approved_by_human", "access_denied",
# "approval_timeout" (no human answered), "polling_error" (network).
# Never raises for these — always fails closed.

The async variant (async_evaluate_with_approval) follows normal asyncio cancellation semantics — cancel the task to stop waiting.

Delegations

Grant an agent the delegator's permission scope for exactly the duration of a task — created on enter, revoked on exit, even when the body raises. A failed revocation is never swallowed: it raises (chained onto the body's exception if both failed):

from vengtoo import CreateDelegationRequest

with client.with_delegation(CreateDelegationRequest(
    delegator_id=user_entity_id,
    delegate_id=agent_entity_id,
    scope=["invoices:read", "invoices:submit"],  # optional: attenuate further
    description="Q3 invoice processing run",     # optional: shows in audit/dashboard
)) as delegation:
    run_workflow()

# async: `async with client.async_with_delegation(...) as delegation:`

The delegate's effective permissions are always the intersection of its own policies and the delegator's — scope narrows that further, it can never escalate.

Async

allowed = await client.async_check(
    subject=Subject(id="user:123", type="user"),
    action="read",
    resource=Resource(type="document", id="doc:456"),
)

resp = await client.async_evaluate(request)

FastAPI Dependency

Two layers, two jobs: require() is the route-level perimeter ("may this caller touch this API area at all?"). For per-object decisions, call check()/evaluate() inside the handler where the resource is known.

require() takes a subject extractor — you tell it where your authentication layer put the caller's identity:

from fastapi import FastAPI, Depends, HTTPException, Request

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

def current_subject(request: Request) -> Subject:
    user = getattr(request.state, "user", None)  # set by your authn middleware
    if user is None:
        raise HTTPException(status_code=401, detail="unauthenticated")
    return Subject(id=user.id, type="user")

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

An extractor exception (or a subject with no id/external_id) → 401. Policy deny → 403. Authorization infrastructure failure → 500, fail closed.

Options

Vengtoo(
    api_key="vgt_...",                  # API key for cloud mode
    base_url="http://127.0.0.1:8181",   # Custom URL (agent mode)
    timeout=5.0,                        # Per-request timeout in seconds (default: 10)
    max_retries=3,                      # Max retries on 5xx/429 (default: 2)
)

Mix-up protection (optional)

client.verify_policy_decision_point(expected="https://pdp.vengtoo.com")
# -> True/False. Async: await client.async_verify_policy_decision_point(...)

Confirms the client is actually talking to the PDP it thinks it is, by checking the policy_decision_point advertised at .well-known/authzen-configuration against what you expect. Optional, unauthenticated, never called automatically — mainly useful as defense in depth in federated / multi-PDP deployments.

Error Handling

from vengtoo import VengtooError, VengtooOAuthError

try:
    client.evaluate(req)
except VengtooOAuthError:
    ...  # bad client_id/client_secret
except VengtooError as e:
    e.is_auth_error    # 401 — bad API key or expired token
    e.is_forbidden     # 403
    e.is_not_found     # 404
    e.is_server_error  # 5xx — retries exhausted

The SDK automatically retries on 5xx and 429 responses (default: 2 retries, honoring the server's Retry-After hint). Other 4xx errors are never retried. With OAuth, a 401 triggers one token refresh before failing.

Types

Type	Fields
Subject	type, id, external_id, properties
Resource	type, id, external_id, properties
Action	name, properties
EvaluationRequest	subject, resource, action, context
EvaluationResponse	decision, context (reason, reason_code, policy_id, access_path + HITL)
CreateDelegationRequest	delegate_id, delegator_id, description, scope, expires_at

Required on every check (matching the API and AuthZEN 1.0): subject.type, subject.id (or external_id), resource.type, and action.name. The SDK validates these locally so you get an immediate, clear error instead of a server 400.

License

Apache-2.0 — see LICENSE.