Python client for Vengtoo — works with both Vengtoo Cloud and the Vengtoo Agent.
Supports sync and async. Requires Python 3.10+. One dependency (httpx).
pip install vengtoofrom 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"),
)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.
client = Vengtoo(base_url="http://127.0.0.1:8181")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_pathEvaluate 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, positionalInstead 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.reasonucast_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)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.
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.
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)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.
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)
)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.
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 exhaustedThe 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.
| 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.
Apache-2.0 — see LICENSE.