Python Quickstart

pip install requests

import json
import requests


class HTIApiClient:
    def __init__(self, server_url: str, api_code: str, timeout: int = 30):
        self.server_url = server_url.rstrip("/")
        self.api_code = api_code.strip()
        self.timeout = timeout
        self.session = requests.Session()

    def _headers(self):
        return {
            "X-API-Code": self.api_code,
            "Accept": "application/json",
        }

    def _request(self, path: str, params=None):
        response = self.session.get(
            f"{self.server_url}{path}",
            headers=self._headers(),
            params=params,
            timeout=self.timeout,
        )
        response.raise_for_status()
        return response.json()

    def authenticate(self):
        return self._request("/client-api/hti/auth")

    def get_history(self, limit: int = 10):
        return self._request("/client-api/hti/history", params={"limit": limit})

    def stream_realtime(self, poll_seconds: float = 2.0):
        response = self.session.get(
            f"{self.server_url}/client-api/hti/stream",
            params={"api_code": self.api_code, "poll_seconds": poll_seconds},
            headers={"Accept": "text/event-stream"},
            stream=True,
            timeout=(10, None),
        )
        response.raise_for_status()

        event_name = "message"
        data_lines = []

        for raw_line in response.iter_lines(chunk_size=1, decode_unicode=True):
            if raw_line is None:
                continue

            line = raw_line.strip()
            if not line:
                if data_lines:
                    yield {
                        "event": event_name,
                        "data": json.loads("\\n".join(data_lines)),
                    }
                    event_name = "message"
                    data_lines = []
                continue

            if line.startswith("event:"):
                event_name = line.split(":", 1)[1].strip()
                continue

            if line.startswith("data:"):
                data_lines.append(line.split(":", 1)[1].strip())


client = HTIApiClient(
    server_url="http://your-hti-environment:3500",
    api_code="<YOUR_API_CODE>",
)

auth_payload = client.authenticate()
print("AUTH:", json.dumps(auth_payload, indent=2))

history_payload = client.get_history(limit=10)
print("HISTORY COUNT:", history_payload.get("count"))

for message in client.stream_realtime(poll_seconds=2.0):
    print(message["event"], message["data"])