Skip to main content

Python Integration

Verdictan supports the OpenAI Chat Completions request structure in this guide.

Point the OpenAI Python client at the gateway. Use a Verdictan client API token. Request a model that the gateway exposes.

Do not infer support for an unlisted OpenAI route from this SDK configuration.

Prerequisites​

  • Complete the Quickstart. Keep the gateway running.

  • Create a client API token for this application. Do not reuse the connected gateway's runtime token or an upstream provider key.

  • Install the client:

    python -m pip install openai

Set the values in the runtime environment and not in source code:

export VERDICTAN_GATEWAY_URL="http://127.0.0.1:41002/v1"
export VERDICTAN_REQUEST_TOKEN="vdt_..."
export VERDICTAN_MODEL="replace-with-configured-model"

VERDICTAN_MODEL must match a model that the gateway returns. The example name does not configure the upstream provider.

Send a chat request​

import os

from openai import OpenAI

client = OpenAI(
    base_url=os.environ["VERDICTAN_GATEWAY_URL"],
    api_key=os.environ["VERDICTAN_REQUEST_TOKEN"],
    max_retries=0,
    timeout=60.0,
)

response = client.chat.completions.create(
    model=os.environ["VERDICTAN_MODEL"],
    messages=[{"role": "user", "content": "Summarize this incident report."}],
)

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

Stream a response​

stream = client.chat.completions.create(
    model=os.environ["VERDICTAN_MODEL"],
    messages=[{"role": "user", "content": "Explain this change."}],
    stream=True,
)

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

Input policies run before the upstream request. A policy chain can buffer a stream when it uses the complete output.

Before you state first-token latency, test the configured chain. For more information, read Streaming and SSE.

Use the asynchronous client​

import asyncio
import os

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url=os.environ["VERDICTAN_GATEWAY_URL"],
    api_key=os.environ["VERDICTAN_REQUEST_TOKEN"],
    max_retries=0,
    timeout=60.0,
)


async def main() -> None:
    response = await client.chat.completions.create(
        model=os.environ["VERDICTAN_MODEL"],
        messages=[{"role": "user", "content": "Review this pull request."}],
    )
    print(response.choices[0].message.content)


asyncio.run(main())

Handle gateway decisions​

Catch API status errors. Use the HTTP status to select the application action:

from openai import APIStatusError

try:
    response = client.chat.completions.create(
        model=os.environ["VERDICTAN_MODEL"],
        messages=[{"role": "user", "content": "Review this text."}],
    )
except APIStatusError as error:
    code = getattr(error, "code", None)
    if error.status_code == 400 and code == "content_policy_violation":
        print("Blocked by policy")
    elif error.status_code in (401, 403):
        print("Client token rejected")
    elif error.status_code == 429:
        print("Rate limited")
    else:
        raise

Do not automatically retry a model request. The provider can receive a call before a timeout. The call can create billable usage.

Before you retry, specify how the application handles duplicate work.

The OpenAI client retries selected failures by default. The two examples set max_retries to 0. The configured timeout also limits how long the client waits for one attempt.

Verify the governed path​

Verify authenticated model discovery:

curl -fsS \
  -H "Authorization: Bearer ${VERDICTAN_REQUEST_TOKEN}" \
  "${VERDICTAN_GATEWAY_URL}/models"

Then, follow events while you send a representative Python request:

verdictan events tail --since 10m --follow

A matching request ID shows that the gateway delivered Event evidence to the control plane.

If the event is missing, make sure that the gateway has a connected event sink. Then, make sure that the filters include the request.

Then, examine the process configuration. Make sure that it uses the correct base URL and client.

Next steps​