Python SDK

Verexa's Python package is a thin, dependency-free client over POST /v1/check. It checks every prompt before it reaches your model and every reply before it reaches your user, either around an OpenAI client or at call sites you control.

• Wrap an OpenAI client and every chat.completions.create and responses.create call is checked on both sides of the turn
• Check text by hand when your model is not OpenAI: one call for the prompt and one for the reply, tied together by a trace id
• Fail open by default, with a circuit breaker for outages and a fail-closed mode when a missed block is worse than downtime
• No runtime dependencies: the package needs Python 3.10 or later and nothing else

Install

pip install verexa

Create a key in the dashboard under Settings → API keys. A key belongs to one project and one environment, so its checks land in the right place without extra configuration. The package ships type hints, so editors and mypy --strict see the same API the runtime does.

Configure the guard

With no arguments, create_guard() reads the environment and falls back to a local verdict-api:

1export VEREXA_API_KEY="vx_live_..."2export VEREXA_BASE_URL="https://api.verexa.dev"

• VEREXA_API_KEY is the key from the dashboard. Without one the package warns once and every check returns the fallback for your fail mode
• VEREXA_BASE_URL is the API base URL. It defaults to https://api.verexa.dev. Set it for a local or self-hosted verdict-api

1from verexa import create_guard23guard = create_guard()

Explicit arguments win over the environment, so a test suite or a second environment can override either one:

1guard = create_guard(2    api_key="vx_live_...",3    base_url="https://api.verexa.dev",4    timeout_ms=15000,5    fail_mode="closed",6)

Create one guard and share it. It holds the keep-alive connection, the circuit breaker and the telemetry buffer, and those are safe to use across threads. check_async runs the same client code in a worker thread, so sync and async call sites share all of it.

Guard an OpenAI client

wrap_openai patches chat.completions.create and responses.create in place and returns the same client, so the rest of your code does not change. Every call is checked twice: the prompt before it is sent, and the reply before it is returned. Both checks share one trace id.

1from openai import OpenAI2from verexa import create_guard, wrap_openai34guard = create_guard()5client = wrap_openai(OpenAI(), guard)67completion = client.chat.completions.create(8    model="gpt-4o",9    messages=[{"role": "user", "content": "How do I reset my password?"}],10)11print(completion.choices[0].message.content)

Works the same with AsyncOpenAI:

1from openai import AsyncOpenAI2from verexa import create_guard, wrap_openai34guard = create_guard()5client = wrap_openai(AsyncOpenAI(), guard)67completion = await client.chat.completions.create(8    model="gpt-4o",9    messages=[{"role": "user", "content": "How do I reset my password?"}],10)

Pass your guard as the second argument when you configured one, as above; without it, wrap_openai builds a client from the environment. guard.wrap_openai(client) does the same. A blocked input raises GuardBlockedError before the provider is called, and a blocked reply raises once the reply is in hand. The error carries the phase and the full verdict:

1from verexa import GuardBlockedError23try:4    completion = client.chat.completions.create(model="gpt-4o", messages=messages)5except GuardBlockedError as err:6    # err.phase is "input" or "output"; err.response is the full verdict.7    log.warning("blocked %s: %s", err.phase, [d.detector_id for d in err.response.detectors])

A redact verdict rewrites the request or the response instead: a redacted input reaches the model with the sensitive span removed, and a redacted reply comes back with [redacted ...] markers. Your own message list is never mutated. For responses.create, a string input and its instructions are checked, and instructions becomes the system prompt for the output check; a structured input list cannot be reassembled from its flattened text, so a redact verdict on one raises GuardBlockedError instead of sending an unreviewed request.

Check text yourself

When your model is not OpenAI, or when you want the checks at your own call sites, use the guard directly. check_input and check_output never raise on a block: they return the verdict and you branch on action. Pass the same trace id to both halves of a turn.

1from verexa import create_guard, random_trace_id23guard = create_guard()4trace_id = random_trace_id()56verdict = guard.check_input(user_message, trace_id=trace_id, profile="balanced")7if verdict.action == "block":8    return910reply = call_your_model(verdict.text if verdict.action == "redact" else user_message)1112checked = guard.check_output(reply, trace_id=trace_id, system_prompt=SYSTEM_PROMPT)13if checked.action == "block":14    return1516send(checked.text if checked.action == "redact" else reply)

Use verdict.text whenever the action is redact. For allow and flag it is your original text. Both methods have async forms that run the blocking call in a worker thread:

1verdict = await guard.check_input_async(user_message, trace_id=trace_id)2checked = await guard.check_output_async(reply, trace_id=trace_id, system_prompt=SYSTEM_PROMPT)

Every check accepts four optional keyword arguments: system_prompt, which you should pass on output checks so a reply that repeats it can be caught; trace_id; profile; and timeout_ms.

Streaming

Streamed replies reach your code as they arrive. The wrapper accumulates the chunks and checks the assembled reply when the stream ends, so a blocking verdict raises GuardBlockedError at that point:

1from verexa import GuardBlockedError23stream = client.chat.completions.create(model="gpt-4o", messages=messages, stream=True)4try:5    with stream:6        for chunk in stream:7            send(chunk.choices[0].delta.content or "")8except GuardBlockedError:9    retract_reply()

By then the chunks may already be on screen, so treat the error as a retraction signal, not a gate. The wrapped stream still works as a context manager, and with AsyncOpenAI as async for.

Profiles and timeouts

A check uses the deterministic profile unless you pass one. balanced adds the injection classifier, and audit adds the tier-3 judge synchronously with an 8 second budget on the server. The client gives up on a check after 2 seconds by default, which covers the first two profiles but abandons an audit check before its verdict arrives:

verdict = guard.check_input(user_message, profile="audit", timeout_ms=15000)

Set timeout_ms on a single check, or on the guard to change the default for every check. When an audit check uses a shorter timeout, the package warns once. Core concepts covers what each profile runs.

Wrapped OpenAI calls always use the default profile. To use balanced or audit, call the guard directly.

When Verexa is unreachable

Verexa fails open. If a check cannot get a verdict at all, the package returns its own fallback: action comes from the fail mode, text is your original text, and degraded is true. That covers a missing key, a network error, a timeout, a 5xx and an unparseable response.

1from verexa import create_guard23guard = create_guard(fail_mode="closed")

Fail open (the default) returns allow, so your app keeps working while Verexa is unreachable. Fail closed returns block, and with a wrapped client that block raises GuardBlockedError like any other. Read degraded on the response to tell an outage block from a detector match:

1verdict = guard.check_input(user_message)23if verdict.action == "block" and verdict.degraded:4    outage_block()

After 5 failed checks in a row the client stops calling the API for 30 seconds and returns the fallback immediately, so an outage does not add a timeout to every request. A 4xx does not count toward the breaker: the service answered, so the key or the request is wrong, and turning that into a block is not the SDK's call.

Telemetry

The guard records every check it makes in a bounded buffer at guard.telemetry, so your app can ship its own metrics without scraping logs:

1for event in guard.telemetry.flush():2    if event.type == "check":3        metrics.histogram("verexa.latency_ms", event.latency_ms, phase=event.phase)4        if event.degraded:5            metrics.increment("verexa.degraded", action=event.action)6    else:7        metrics.increment("verexa.error", status=event.status or 0)

• check events carry the action, latency_ms and degraded of a verdict
• error events carry the HTTP status when there was one
• circuit_open events mark calls the breaker skipped

flush() drains the buffer and size() reports how many events are waiting. A full buffer drops the oldest events, so recent failures stay visible. Every event carries the trace_id and phase of its check.

The Python API at a glance

Everything the package exports:

Every check returns a CheckResponse:

Check API documents the wire shape behind these fields.

Next steps

• Quickstart: guard your first call end to end
• Core concepts: actions, profiles, traces and failure modes
• Handle failures: react to blocked turns, degraded verdicts and outages
• Check API: every request and response field
• TypeScript and Go SDK guides