# Python over HTTP

Source: https://convergingthought.com/docs/sdk-reference/python

> 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 [#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.

```python title="client.py"
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 [#learn-and-read]

```python title="example.py"
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 [#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](/docs/api-reference/errors) and [Polling](/docs/sdk-reference/polling).

The [preview deployment](/docs/getting-started/availability) 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.