visiq package is the Python peer of
@visiq/harness. One compiled Rust core makes the same local, in-process policy
decisions, and on top of it the package ships the same end-to-end harness the
TypeScript SDK does — bundle fetch, agent registration, human-in-the-loop, and
audit telemetry.
abi3). Set the same environment as the
TypeScript SDK: VISIQ_API_KEY (vq_prod_… / vq_test_…), VISIQ_ENDPOINT
(https://api.visiqlabs.com), and optionally VISIQ_AGENT_ID.
Two layers, one wheel. Use
Governor for a governed agent end to end (it
fetches your bundle and enforces every outcome). Use the low-level
gate_action / gate_retrieval / decide functions when you already hold a
rule bundle and just want a local decision with no network.Pre-1.0 (0.x). Breaking changes may ship in any minor release until v1.0.
Pin an exact version. See SDK versioning for the posture.
Public API
Everything exported from the package top level:Governor
One Governor per agent process. It warms the rule bundle, decides every tool
call and retrieval locally against the compiled core, blocks on human approval
when a rule requires it, applies retrieval redaction, and streams an audit event
per decision.
Governor(agent_id=None, config=None)
Construct a governor. agent_id follows the same precedence as the TypeScript
SDK: explicit argument → VISIQ_AGENT_ID → "agent". Pass a HarnessConfig
for config to bypass environment resolution.
start(tools=None) -> Governor
Warm the bundle and run the one-time registration handshakes (environment +
tool surface). Safe to call once at startup; returns self so you can chain it
onto the constructor. tools is a list of {"name", "description"} dicts
describing your tool surface.
gate_tool(tool_name, args, call) -> Any
Govern one tool call: decide locally, enforce the outcome, and only then invoke
call(effective_args).
callmust accept exactly one positional argument — the effective argument mapping. On amaskverdict the gate handscallthe redacted arguments, so the tool never sees the originals.- A
permitruns the tool with the original args. - An
approval_requiredregisters the decision and blocks on the HITL poll until a human resolves it (orhitl_timeout_mselapses → blocked). - A
deny, an unapproved/expired HITL, or an unredactablemaskraisesToolBlocked— the tool body never runs (G001).
gate_documents(docs, query=None) -> list
Filter retrieved documents through the retrieval facet: deny drops the
document, redact masks its fields, allow/escalate keep it. Each document
is {page_content | text | content, metadata}-shaped. Fails closed — a bundle
that has not loaded returns [], and any per-document evaluation error drops
that one document rather than crashing the batch.
gate_text(text, metadata=None) -> str
Filter a single string tool result through the retrieval facet. Returns the
text unchanged on allow, a redacted copy on redact, or a short blocked
placeholder on deny / when governance is unavailable.
flush() -> None
Flush buffered audit telemetry to the backend. Registered with atexit, so it
also runs at process exit — call it explicitly for long-lived processes.
agent_id (property)
The resolved agent id this governor reports under.
Low-level decision gates
Pure functions over a rule bundle you already hold. No network, no I/O — the decision path is entirely local and fails closed on malformed input (G001). Every gate returns the sameUnifiedDecision dict.
gate_action(bundle, *, tool_name, args=None, agent_id="agent", target_resource=None, normalized=None) -> dict
Decide one tool/action call.
gate_retrieval(bundle, *, resource_type="document", resource_metadata=None, agent_id="agent", query=None) -> dict
Decide one retrieval. Check decision["retrieval"]["action"] — drop on
deny/escalate, redact on redact via retrieval.redactionRules, keep on
allow — before the content reaches the model.
decide(event, bundle) -> dict
Evaluate a raw event dict directly. gate_action and gate_retrieval are thin
wrappers over this.
The decision dict
action.decision∈permit·deny·approval_required·mask.retrieval.action∈allow·deny·redact·escalate.enforcedisfalsewhen the agent is in monitor/off mode — the would-be verdict is still reported onaction.decision, but top-levelallowedstaystruebecause nothing is actually blocked. In anenforcebundle,allowedreflects the real outcome.
Where does
bundle come from? The full harness (Governor) fetches it for
you from GET /rules/bundle. If you drive the low-level gates yourself, fetch
that bundle over the rules API and pass the JSON
object straight in.resolve_config(agent_id=None) -> HarnessConfig
Resolve transport configuration from the environment, matching the TypeScript
harness precedence:
- endpoint —
VISIQ_ENDPOINT, then theVISIQ_BASE_URLalias. - api_key —
VISIQ_API_KEY, then theVISIQ_ALLOW_API_KEYalias. - agent_id — explicit argument →
VISIQ_AGENT_ID→"agent".
HarnessConfig(endpoint, api_key, agent_id, timeout_ms=10000, hitl_timeout_ms=120000)
carries those values; endpoint/api_key may be None, in which case every
network call no-ops and the decision path is local-only (fail-closed on
enforce).
ToolBlocked
Governor.gate_tool when a call is denied or a HITL approval is not
granted. The tool body is never executed (G001). Let it surface to your
framework as the tool’s failure, or catch it to return a model-readable message:
Next steps
Quickstart
The Python section of the quickstart — governed agent in a few lines.
TypeScript Reference
The
visiq() function, options, and framework detection.