Skip to main content
Version: 1.8.0

Use DVARA from Python

Point an OpenAI-compatible Python client at DVARA when your application needs policy, PII controls, audit evidence, and cost attribution around model calls. You keep the client API; DVARA governs the request before it reaches a configured provider.

Send your first governed request​

You need a running DVARA deployment, a registered provider that serves the model in your request, and a workspace API key. Follow the Quickstart if you do not have them yet.

Set the connection once. Keep the key outside source control:

export DVARA_BASE_URL="http://localhost:8080/v1"
export DVARA_API_KEY="<your-dvara-api-key>"
pip install openai

Use a key with completions:write. A key with no scopes is unrestricted, but a key with any scopes can call only the endpoint families named by those scopes. See Authenticate data-plane calls for the complete map.

This example keeps the response headers so you can match the application call to its DVARA evidence:

import os
from openai import OpenAI

client = OpenAI(
base_url=os.environ["DVARA_BASE_URL"],
api_key=os.environ["DVARA_API_KEY"],
)

raw = client.chat.completions.with_raw_response.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "Answer in one sentence."},
{"role": "user", "content": "Why should an API key be rotated?"},
],
max_tokens=80,
)

response = raw.parse()
print(response.choices[0].message.content)
print("DVARA trace:", raw.headers.get("x-trace-id"))

A successful response has the normal Chat Completions shape:

Regular rotation limits the time a leaked API key can be used.
DVARA trace: 043793a6ac754c9c9d1bc5d4f525f827

The wording and token counts vary by model. When audit storage is configured, search for the trace ID in Flightdeck or your audit output. The completed call has a GATEWAY_RESPONSE event with its model, provider, outcome, and available usage data.

The model name is not an instruction to discover a provider. It must match a model served by a provider registered in your DVARA deployment. Use GET /v1/models to inspect the current provider result, then use the exact identifier your deployment exposes.

Stream a Chat Completions response​

Set stream=True; the client consumes DVARA's OpenAI-compatible SSE stream:

stream = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": "Write a two-line release note about key rotation."}
],
max_tokens=80,
stream=True,
)

for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
print()

Streaming response enforcement follows the workspace's configured streaming posture. It is not the same as buffering and scanning a complete non-streaming response. See Streaming before using a stream for regulated content.

Use the Responses API​

DVARA 1.8 supports basic text and image input, instructions, metadata, structured output through text.format, and synchronous or streaming Responses API calls. The same completions:write scope covers this endpoint.

raw = client.responses.with_raw_response.create(
model="gpt-4o",
instructions="Answer in one sentence.",
input="Name one practical reason to keep an AI audit trail.",
)

response = raw.parse()
print(response.output_text)
print("DVARA trace:", raw.headers.get("x-trace-id"))

Example output:

An audit trail lets an investigator reconstruct which model call produced a decision.
DVARA trace: f83272036c1f48aab165fe68e55390cb

The 1.8 Responses API is stateless. It refuses store: true, previous_response_id, reusable prompt objects, reasoning items, background mode, include, tools, and tool_choice with HTTP 400 and unsupported_capability. Use Chat Completions when the model must return a function call; your Python application or agent framework still executes the function.

Request structured JSON​

Use a JSON Schema on Chat Completions when the configured provider supports it:

response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": "Return two controls for protecting API keys."}
],
response_format={
"type": "json_schema",
"json_schema": {
"name": "api_key_controls",
"strict": True,
"schema": {
"type": "object",
"properties": {
"controls": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"reason": {"type": "string"},
},
"required": ["name", "reason"],
"additionalProperties": False,
},
}
},
"required": ["controls"],
"additionalProperties": False,
},
},
},
)

print(response.choices[0].message.content)

DVARA validates the request shape and passes the schema to the selected provider. Provider support still determines whether that model can produce a strict structured response. See Structured outputs for the provider matrix and failure behavior.

Receive a function call​

Chat Completions accepts OpenAI-style function definitions. DVARA governs the model call and, when configured, scans tool-call arguments. It does not execute the Python function.

import json

response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "user", "content": "Multiply 42 by 17."}
],
tools=[
{
"type": "function",
"function": {
"name": "multiply",
"description": "Multiply two integers.",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "integer"},
"b": {"type": "integer"},
},
"required": ["a", "b"],
"additionalProperties": False,
},
},
}
],
tool_choice="auto",
)

call = response.choices[0].message.tool_calls[0]
arguments = json.loads(call.function.arguments)
print(call.function.name, arguments)

Expected shape:

multiply {'a': 42, 'b': 17}

Validate the arguments and authorize the action before executing a tool. The model's function call is a proposal, not a trusted instruction.

Create embeddings​

An embeddings call needs embeddings:write when the key has a nonempty scope list. The named embedding model must be available through a configured provider.

response = client.embeddings.create(
model="text-embedding-3-small",
input=[
"Rotate provider credentials every 90 days.",
"Review failed authentication events each day.",
],
)

print(len(response.data), "vectors")
print(len(response.data[0].embedding), "dimensions")

Example output for a 1,536-dimension model:

2 vectors
1536 dimensions

The dimension is controlled by the model and can differ. DVARA applies the configured request-side controls and records available usage, but it does not run model-text response scans over numeric vectors.

Call Anthropic, Bedrock, or Gemini models​

DVARA exposes an OpenAI-compatible client surface. Keep using the OpenAI Python client and change the model identifier to one served by your configured Anthropic, Amazon Bedrock, or Google provider.

for model in ["<anthropic-model>", "<bedrock-model>", "<gemini-model>"]:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": "Reply with the provider name."}],
max_tokens=30,
)
print(model, response.choices[0].message.content)

Replace all three model placeholders with identifiers returned by your deployment. Bedrock Converse and Google GenAI SDK request shapes are not DVARA endpoints; do not point those clients at /v1/chat/completions. From 1.8.5 the Anthropic SDK can keep its own shape: DVARA serves the Anthropic Messages API at /v1/messages. Set the SDK's base_url to the Gateway, with no /v1, and its API key to a DVARA workspace key. See Govern Claude Code, which uses the same endpoint.

Use the async client​

AsyncOpenAI uses the same base URL, API key, endpoints, and scope rules:

import asyncio
import os
from openai import AsyncOpenAI

async def main() -> None:
client = AsyncOpenAI(
base_url=os.environ["DVARA_BASE_URL"],
api_key=os.environ["DVARA_API_KEY"],
)
response = await client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Reply with: governed"}],
max_tokens=10,
)
print(response.choices[0].message.content)

asyncio.run(main())

The OpenAI client and DVARA can each retry transient failures. Set the client's max_retries deliberately so a single application action does not create more provider attempts than your reliability and cost policy allows.

Connect the LiteLLM client library​

If your code or framework already calls models through the LiteLLM client library, point it at DVARA. You don't need the LiteLLM proxy: DVARA is the gateway. LiteLLM's openai/ prefix selects its OpenAI-compatible transport. DVARA receives the bare model name and makes the provider decision.

pip install litellm
import os
import litellm

response = litellm.completion(
model="openai/gpt-4o",
api_base=os.environ["DVARA_BASE_URL"],
api_key=os.environ["DVARA_API_KEY"],
messages=[{"role": "user", "content": "Reply with: governed"}],
)

print(response.choices[0].message.content)

Connect LangChain​

Use the current base_url and api_key arguments on ChatOpenAI:

pip install langchain-openai
import os
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
model="gpt-4o",
base_url=os.environ["DVARA_BASE_URL"],
api_key=os.environ["DVARA_API_KEY"],
temperature=0,
)

message = model.invoke("Give one reason to retain model-call evidence.")
print(message.content)

LangChain agents use Chat Completions function calling through this model. Your agent process executes each selected tool, so enforce tool authorization there as well as any configured DVARA governance.

Connect Pydantic AI​

Select OpenAIChatModel explicitly. Pydantic AI can otherwise choose the Responses API, whose 1.8 DVARA implementation does not accept tools.

pip install pydantic-ai
import os
from pydantic import BaseModel
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

class Control(BaseModel):
name: str
reason: str

model = OpenAIChatModel(
"gpt-4o",
provider=OpenAIProvider(
base_url=os.environ["DVARA_BASE_URL"],
api_key=os.environ["DVARA_API_KEY"],
),
)
agent = Agent(model, output_type=Control)

result = agent.run_sync("Name one control for protecting an API key.")
print(result.output)

The current Pydantic AI result is available as result.output; older examples that use OpenAIModel, result_type, or result.data do not match the current client API.

Handle a refusal​

DVARA uses one error envelope for authentication, policy, budget, provider, and validation failures. Keep the trace_id; it connects the application failure to DVARA logs and, when configured, audit evidence.

import openai

try:
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Summarize this contract."}],
)
print(response.choices[0].message.content)
except openai.APIStatusError as exc:
error = exc.response.json().get("error", {})
print(f"HTTP {exc.status_code}: {error.get('code')} — {error.get('message')}")
print("DVARA trace:", error.get("trace_id"))

For example, a valid key scoped only to models:read receives:

{
"error": {
"message": "API key is not scoped for this endpoint. It carries [models:read]; this endpoint needs completions:write.",
"type": "permission_error",
"code": "api_key_scope",
"trace_id": "043793a6ac754c9c9d1bc5d4f525f827"
}
}

See Error handling for the complete status and error-code contract.

Continue the integration​