Skip to main content

Python Integration

Keeptrusts supports the OpenAI Chat Completions request shape used in this guide. Point the OpenAI Python client at the gateway, use a Keeptrusts client API token, and request a model that the gateway exposes. Do not infer support for an unlisted OpenAI route from this SDK configuration.

Prerequisites​

  • Complete the Quickstart and 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 outside source code:

export KEEPTRUSTS_GATEWAY_URL="http://127.0.0.1:41002/v1"
export KEEPTRUSTS_REQUEST_TOKEN="kt_..."
export KEEPTRUSTS_MODEL="gpt-5.4-mini"

KEEPTRUSTS_MODEL must match a model returned by the gateway; the example name does not configure the upstream provider for you.

Send a chat request​

import os

from openai import OpenAI

client = OpenAI(
    base_url=os.environ["KEEPTRUSTS_GATEWAY_URL"],
    api_key=os.environ["KEEPTRUSTS_REQUEST_TOKEN"],
)

response = client.chat.completions.create(
    model=os.environ["KEEPTRUSTS_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["KEEPTRUSTS_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 chain that needs the complete output can buffer a streaming response, so do not promise first-token latency without testing the exact chain. See Streaming and SSE.

Use the asynchronous client​

import asyncio
import os

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url=os.environ["KEEPTRUSTS_GATEWAY_URL"],
    api_key=os.environ["KEEPTRUSTS_REQUEST_TOKEN"],
)


async def main() -> None:
    response = await client.chat.completions.create(
        model=os.environ["KEEPTRUSTS_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 and branch on the HTTP status:

from openai import APIStatusError

try:
    response = client.chat.completions.create(
        model=os.environ["KEEPTRUSTS_MODEL"],
        messages=[{"role": "user", "content": "Review this text."}],
    )
except APIStatusError as error:
    if error.status_code == 400 and error.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 blindly retry a model request: a timed-out call may already have reached the provider and incurred usage. Retry only after deciding how your application will handle duplicate work.

Verify the governed path​

Check authenticated model discovery:

curl -fsS \
  -H "Authorization: Bearer ${KEEPTRUSTS_REQUEST_TOKEN}" \
  "${KEEPTRUSTS_GATEWAY_URL}/models"

Then follow events while sending a representative Python request:

kt events tail --since 10m --follow

A matching event proves that the process used Keeptrusts. If the event is absent, first confirm the gateway has a connected event sink and that the time, region, and verdict filters include the request. Then investigate whether the process used another base URL or client.

Next steps​