Python SDK
Python applications point an OpenAI-compatible client at Keeptrusts. The gateway owns the upstream provider credential and enforces the agent-bound policy config.
Prerequisites
- Python 3.9 or later.
- The
ktCLI. - A Keeptrusts runtime API token for the gateway and a separate gateway key, access key, or personal API token for the application.
- An upstream model ID and credential from the provider you configure.
Configure and start the gateway
Replace the model placeholder before starting the gateway.
pack:
name: python-sdk-integration
version: 1.0.0
enabled: true
policies:
chain:
- prompt-injection
- pii-detector
- audit-logger
providers:
targets:
- id: openai-primary
provider: openai
model: "replace-with-openai-model-id"
base_url: https://api.openai.com
secret_key_ref:
env: KEEPTRUSTS_OPENAI_API_KEY
export KEEPTRUSTS_API_TOKEN="replace-with-keeptrusts-api-token"
export KEEPTRUSTS_OPENAI_API_KEY="replace-with-openai-api-key"
kt policy lint --file policy-config.yaml
kt gateway run \
--agent python-sdk-integration \
--listen 127.0.0.1:41002 \
--policy-config policy-config.yaml
The --agent flag is required with --policy-config and uses KEEPTRUSTS_API_TOKEN to synchronize the agent-bound configuration. Do not reuse that runtime token as the application credential.
OpenAI SDK
Install the official client:
python -m pip install openai
export KEEPTRUSTS_CLIENT_TOKEN="replace-with-separate-client-token"
Create a client that authenticates to Keeptrusts:
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ["KEEPTRUSTS_CLIENT_TOKEN"],
base_url="http://127.0.0.1:41002/v1",
)
response = client.chat.completions.create(
model="replace-with-openai-model-id",
messages=[
{"role": "user", "content": "Reply with one short sentence."},
],
)
print(response.choices[0].message.content)
The KEEPTRUSTS_OPENAI_API_KEY value stays in the gateway process. The Python application needs only the separate Keeptrusts client token.
Streaming
Use streaming only after validating the selected provider, model, request family, and policy chain:
stream = client.chat.completions.create(
model="replace-with-openai-model-id",
messages=[
{"role": "user", "content": "Write two short lines."},
],
stream=True,
)
for event in stream:
content = event.choices[0].delta.content
if content:
print(content, end="", flush=True)
Do not assume that every policy can rewrite partial output chunks. Test the complete streaming response before production rollout.
Error handling
The SDK raises an API exception for non-2xx responses. Preserve the HTTP status and request identifier in application logs, but do not log secrets or unredacted prompts.
from openai import APIStatusError
try:
response = client.chat.completions.create(
model="replace-with-openai-model-id",
messages=[{"role": "user", "content": "Check this request."}],
)
except APIStatusError as error:
print(f"Keeptrusts request failed: {error.status_code}")
raise
Operational guidance
- Keep
base_urlending in/v1. - Use the same model ID as the configured target unless an explicit route maps the request elsewhere.
- Treat provider pricing, context windows, and model availability as mutable vendor facts.
- Configure SDK timeouts and retries deliberately for your workload; do not hide persistent policy or authentication failures behind retries.
- Keep all tokens in server-side environment or secret storage.