Python over HTTP
Submit and poll using Python's standard library, without an unpublished SDK dependency.
The public API works with any server-side HTTP client. This guide uses Python's standard library so the example does not depend on an Internalize package that has not been published. It demonstrates transport behavior; your application should provide durable storage for operation keys and job IDs.
Configure the environment
Set INTERNALIZE_API_KEY in your server environment. The example defaults to the production origin and allows INTERNALIZE_BASE_URL to override it for an authorized deployment. Do not put a real key directly in the script.
import json
import os
import time
import urllib.error
import urllib.request
BASE_URL = os.environ.get(
"INTERNALIZE_BASE_URL", "https://convergingthought.com"
).rstrip("/")
API_KEY = os.environ["INTERNALIZE_API_KEY"]
class ApiError(Exception):
def __init__(self, status, body):
error = body.get("error", {})
self.status = status
self.code = error.get("code", "unknown_error")
self.request_id = error.get("request_id")
self.retryable = error.get("retryable", False)
super().__init__(self.code)
def request(path, body=None, operation_key=None):
headers = {"Authorization": f"Bearer {API_KEY}"}
data = None
if body is not None:
headers["Content-Type"] = "application/json"
data = json.dumps(body).encode("utf-8")
if operation_key is not None:
headers["Idempotency-Key"] = operation_key
req = urllib.request.Request(BASE_URL + path, data=data, headers=headers)
try:
with urllib.request.urlopen(req, timeout=30) as response:
return json.load(response)
except urllib.error.HTTPError as error:
try:
body = json.loads(error.read())
except (ValueError, UnicodeDecodeError):
body = {}
raise ApiError(error.code, body) from None
def wait_for_job(job_id, timeout_seconds=600):
deadline = time.monotonic() + timeout_seconds
delay = 1.0
while time.monotonic() < deadline:
job = request("/v1/jobs/" + job_id)
if job["status"] == "reconciliation_required":
raise RuntimeError(f"Reconciliation required for {job_id}")
if job["billing"]["settled"]:
return job
remaining = deadline - time.monotonic()
if remaining <= 0:
break
time.sleep(min(delay, remaining))
delay = min(15.0, delay * 1.5)
raise TimeoutError(f"Resume polling job {job_id}; it was not cancelled")The 30-second timeout applies to an individual HTTP request. The polling deadline bounds observation, although an in-flight read can complete after that deadline. A timeout does not cancel paid work. This helper intentionally leaves HTTP retry decisions to the calling application.
Learn and read
from client import request, wait_for_job
# Persist this key before sending. Reuse it only for this exact operation.
operation_key = "support-policy-python-001"
accepted = request(
"/v1/internalizations",
{
"subject_id": "support-python",
"content": "Northstar accepts returns within 30 days. Return labels are prepaid.",
"max_cost_microusd": 5000000,
"activate": True,
},
operation_key,
)
# Save accepted['id'] in your application before waiting.
job = wait_for_job(accepted["id"])
if job["status"] != "ready" or not (job.get("result") or {}).get("activated"):
raise RuntimeError(f"Learning needs review: {job['id']}")
answer_receipt = request(
"/v1/inferences",
{
"subject_id": "support-python",
"messages": [{"role": "user", "content": "Who pays for return labels?"}],
"max_output_tokens": 2048,
},
"support-python-question-001",
)
answer = wait_for_job(answer_receipt["id"])
if answer["status"] != "succeeded":
raise RuntimeError(f"Inference needs review: {answer['id']}")
print(answer["result"]["text"])The final print is suitable for a private terminal demonstration. A production service should return the answer to its authorized caller and keep model content out of general-purpose logs.
Recover an interrupted process
If a saved job ID exists, call wait_for_job with that ID. If admission may have succeeded but no receipt was saved, repeat request with the original operation key and unchanged body. Do not generate a new key merely because the process restarted.
For 429, respect Retry-After in a production transport wrapper. For authorization or validation failures, correct the cause before trying again. For reconciliation, keep the original operation and involve support. See Errors and Polling.
The preview deployment will reject model admission until execution is enabled. A correctly structured script can therefore receive preview_only; that response is not evidence of a Python transport problem.