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.
A new project needs prepaid credits before model admission. An insufficient_credits response is a billing condition, not a Python transport problem. See service availability.