tensorgo 0.12.0

Python SDK for HumAIn AI services (offline Voice API and more).

HumAIn Python SDK

A small, modular client for HumAIn's AI services — built like the ElevenLabs SDK: one client, namespaced modules. Current capabilities are the offline Voice API and the offline Eye-Gaze API (submit a video, get the result delivered to your webhook).

You only ever provide three things: your API key, a video link, and a webhook URL. All service endpoints are internal to the SDK and are never exposed to you.

Install

pip install -e .        # from this directory (sdk/)

Once published to the private index, users install it with a plain pip install tensorgo. See PUBLISHING.md for how to release to AWS CodeArtifact (./publish.sh).

Quickstart

from tensorgo import HumAIn

client = HumAIn(api_key="sk_...")

job = client.voice_api.process(
    video_link="https://example.com/video.mp4",
    webhook_url="https://example.com/my-webhook",
)

print(job.inference_id, job.status)

Eye-Gaze API

Same ergonomics, different capability — submit a video and the gaze result is delivered to your webhook:

from tensorgo import HumAIn

client = HumAIn(api_key="sk_...")

job = client.eye_gaze.process(
    video_link="https://example.com/video.mp4",
    webhook_url="https://example.com/my-webhook",
)

print(job.inference_id, job.status)

Deception API

Same ergonomics, different capability — submit a video and the per-chunk truthfulness result is delivered to your webhook:

from tensorgo import HumAIn

client = HumAIn(api_key="sk_...")

job = client.deception_api.process(
    video_link="https://example.com/video.mp4",
    webhook_url="https://example.com/my-webhook",
)

print(job.inference_id, job.status)

Voice Bio API

Voice biometrics with three operations — register a voice, identify it in a later video, and delete the registered data. The subject must be registered before it can be identified. Both register and process are asynchronous: they return immediately and the outcome is POSTed to your webhook.

from tensorgo import HumAIn

client = HumAIn(api_key="sk_...")

# 1. Register a voice
reg = client.voice_bio.register(
    video_link="https://example.com/registration.mp4",
    webhook_url="https://example.com/my-webhook",
    subject_id="subject-001",
    subject_name="John Doe",
)

# 2. Identify the voice in a session video
job = client.voice_bio.process(
    video_link="https://example.com/session.mp4",
    webhook_url="https://example.com/my-webhook",
    subject_id="subject-001",
    subject_name="John Doe",
)

# 3. Delete the registered voice data
result = client.voice_bio.delete(subject_ids=["subject-001"])
print(result.deleted_subject_ids, result.not_found_subject_ids)

Voice Cloning (TTS) API

Clone a voice from a reference audio clip and synthesise speech in it. Unlike the offline CV modules, voice cloning is synchronous — there is no webhook and no video link. You provide your organization_id and the local path to a reference audio file; the generated speech is returned directly in the response. Every operation is scoped to your organisation, so you only ever see and manage the voices you created.

Four operations: create, list, generate, delete.

from tensorgo import HumAIn

client = HumAIn(api_key="sk_...")

# 1. Create (clone) a voice from a local reference audio file
voice = client.voice_cloning.create_voice(
    organization_id="gox",
    name="John",
    ref_audio_path="/path/to/reference.wav",   # local file; the SDK uploads it
    # ref_text="..."                            # optional; auto-transcribed if omitted
)

# 2. List the voices created under your organisation
voices = client.voice_cloning.list_voices(organization_id="gox")
for v in voices:
    print(v.voice_id, v.name)

# 3. Generate speech in the cloned voice — audio comes back in the response
speech = client.voice_cloning.generate(
    organization_id="gox",
    voice_id=voice.voice_id,
    text="Hello, this is my cloned voice.",
)
speech.save("out.wav")          # or use speech.audio_bytes

# 4. Delete one or more voices
result = client.voice_cloning.delete(organization_id="gox", voice_ids=[voice.voice_id])
print(result.deleted_voice_ids, result.not_found_voice_ids)

Voice Synthesis (ZipVoice TTS) API

Synthesise speech in a voice you already created with Voice Cloning, using the fast ZipVoice TTS engine. Like voice cloning it is synchronous — no webhook — and scoped to your organisation. You pass the organization_id and voice_id of an existing voice, the text, and (optionally) the speed; the audio comes back directly in the response.

One operation: synthesize.

from tensorgo import HumAIn

client = HumAIn(api_key="sk_...")

speech = client.voice_synthesis.synthesize(
    organization_id="gox",
    voice_id="v-1",            # a voice created via client.voice_cloning.create_voice(...)
    text="Hello, this is speech synthesised in my cloned voice.",
    speed=1.0,                 # optional (default 1.0)
    # num_steps=4              # optional sampling steps; lower is faster (default 4)
)
speech.save("out.wav")         # or use speech.audio_bytes

Meeting Notetaker

Send a bot into a Google Meet, Zoom or Teams meeting and receive everything it hears — participants, active speaker, meeting subject and a speaker-attributed transcript. Events reach you on your webhook, on the live socket feed, or both. Nothing is stored on our side — no meeting record, no recording, no transcript — so a session that has ended cannot be replayed; persist what you care about as it arrives.

Only meeting_url and platform are required:

from tensorgo import HumAIn

client = HumAIn(api_key="sk_...")

session = client.notetaker.start(
    meeting_url="https://meet.google.com/abc-defg-hij",
    platform="gmeet",                                  # gmeet | zoom | teams
    webhook_url="https://your-server.com/hooks/notetaker",
)
print(session.session_id, session.status)

Every other option — omit any of them and the service applies its own default:

session = client.notetaker.start(
    meeting_url="https://meet.google.com/abc-defg-hij",
    platform="gmeet",

    webhook_url="https://your-server.com/hooks/notetaker",  # optional with the socket
    webhook_secret="whsec_your_secret",     # optional, signs every delivery

    events=["transcript.final", "speaker.change",
            "participant.joined", "participant.left"],      # default: all but partials
    partials=False,                         # True adds live in-progress text

    bot_name="Acme Notetaker",              # shown in the meeting roster
    join_message="Hi, I'm here to take notes.",   # posted in the meeting chat
    leave_when_alone_sec=60,                # leave once nobody else is left
    leave_after_silence_sec=600,            # leave after this much silence
    record_video=False,                     # audio + transcript only
    end_at="2026-08-04T12:00:00Z",          # optional ISO-8601 UTC
    max_duration_sec=1800,                  # hard cap

    metadata={"your_meeting_id": "mtg_42"}, # opaque, echoed on every event
)

Check on a running session, or pull the bot out early:

status = client.notetaker.status(session_id=session.session_id)
print(status.status, status.participants)   # starting | live | completed

result = client.notetaker.stop(session_id=session.session_id)
print(result.status)

The bot also leaves on its own — when it is alone, after the silence window, at end_at or at max_duration_sec — so a forgotten session cannot run forever. Sessions are discarded 15 minutes after they end; after that the id is unknown (NotFoundError).

Webhook delivery. Batches are POSTed every 250 ms or 25 events, one request in flight at a time, so seq is strictly increasing. Delivery is at-least-once — de-duplicate on seq. A failing endpoint is retried 3 times (1s, 4s, 16s) and never stalls the meeting. With webhook_secret set, verify X-Gox-Signature (HMAC-SHA256 of "<t>.<raw body>", over the raw bytes):

import hashlib, hmac

def verify(secret: str, header: str, body: bytes) -> bool:
    """header looks like: t=1785999999,v1=9f2c…"""
    parts = dict(piece.split("=", 1) for piece in header.split(","))
    expected = hmac.new(
        secret.encode(), f"{parts['t']}.".encode() + body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, parts["v1"])

Live socket feed. start() also returns listen_url, listen_token and listen_event — subscribe and the same events arrive over Socket.IO (pip install "python-socketio[client]"). Everything already emitted is replayed on subscribe, so a late connect or a reconnect loses nothing: pass the last seq you saw as since_seq. See examples/notetaker_live_socket.py.

Event types: session.joining, session.live, participant.joined, participant.left, meeting.subject, speaker.change, transcript.partial (opt in with partials=True), transcript.final, session.error, and session.completed — always last, carrying the full transcript in one object. Every event has the same envelope: type, seq, ts, data. speaker is null when the meeting platform gave nobody to attribute the words to; we never guess a name.

What happens under the hood

1. The SDK validates your API key (cached for the rest of the session).
2. It submits video_link + webhook_url to the processing service.
3. Processing is asynchronous — process() returns immediately with an accepted VoiceJob. When the model finishes, the service POSTs the result to your webhook_url.

Error handling

Everything inherits from HumAInError:

from tensorgo.exceptions import (
    HumAInError, AuthenticationError, BadRequestError,
    RateLimitError, ServerError, APIConnectionError,
)

try:
    client.voice_api.process(video_link="...", webhook_url="...")
except AuthenticationError:
    ...   # invalid API key (HTTP 401/403)
except BadRequestError:
    ...   # bad input (HTTP 400/422)
except APIConnectionError:
    ...   # could not reach the service
except HumAInError:
    ...   # catch-all

APIError subclasses carry .status_code and .body.

Architecture (for maintainers)

The SDK is intentionally modular so new capabilities (STT, dubbing, …) are easy to add:

tensorgo/
├── client.py          HumAIn — entry point; mounts modules
├── _config.py         INTERNAL endpoint URLs (never exposed publicly)
├── _http.py           Transport (ABC) + RequestsTransport + HttpClient
├── _auth.py           Authenticator — validates & caches the API key
├── exceptions.py      HumAInError hierarchy
├── models.py          VoiceJob / EyeGazeJob (typed responses)
└── modules/
    ├── base.py          BaseModule (ABC) — shared module behaviour
    ├── voice_api.py     VoiceAPIModule — client.voice_api.process(...)
    ├── eye_gaze.py      EyeGazeModule — client.eye_gaze.process(...)
    └── deception_api.py DeceptionAPIModule — client.deception_api.process(...)

Adding a new module

1. Subclass BaseModule, implement namespace and the capability's verbs.
2. Add its endpoint path to _ENDPOINTS in _config.py.
3. Mount it in HumAIn.__init__ (e.g. self.stt = STTModule(self._http, self._auth)).

The Transport abstraction means modules never touch requests directly, which also makes them trivial to unit test (see tests/conftest.py's FakeTransport).

Running the tests

pip install -e ".[dev]"
pytest

Internal testing against a local launcher

Endpoints are internal. For local testing only, point the SDK at a local launcher with the undocumented override:

export HUMAIN_BASE_URL="http://localhost:8000"

The eye-gaze capability runs as its own service (production :9087), so it has its own production base URL and a dedicated, undocumented override for testing it in isolation:

export HUMAIN_EYEGAZE_BASE_URL="http://localhost:9087"

When unset it uses the eye-gaze production URL. Both overrides are unsupported for end users and absent from the public API.

The deception capability likewise runs as its own service (production :7097), with its own dedicated, undocumented override for isolated testing:

export HUMAIN_DECEPTION_BASE_URL="http://localhost:7097"

The voice-bio capability likewise runs as its own service (the voice biometrics launcher, production :7093), with its own dedicated, undocumented override for isolated testing:

export HUMAIN_VOICEBIO_BASE_URL="http://localhost:7093"

The voice-cloning capability likewise runs as its own service (the cloner launcher, production :8069), with its own dedicated, undocumented override for isolated testing:

export HUMAIN_VOICECLONING_BASE_URL="http://localhost:8069"

The voice-synthesis capability (ZipVoice TTS) likewise runs as its own service (production :8546), with its own dedicated, undocumented override for isolated testing:

export HUMAIN_VOICESYNTHESIS_BASE_URL="http://localhost:8546"

The notetaker is proxied by the GOX meeting service (the bot manager itself is private), so it points at that service rather than a model host, with the same kind of undocumented override for isolated testing:

export HUMAIN_NOTETAKER_BASE_URL="http://localhost:3000"