tensorgo 0.13.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,
                                            # also how long it waits to be let in
    leave_after_silence_sec=600,            # leave when nobody has spoken this long
    record_video=True,                      # False = audio + transcript only,
                                            # no screen capture, no encoder
    end_at="2026-08-04T12:00:00Z",          # optional ISO-8601 UTC stop time

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

end_at is how you bound a session; the service also enforces its own ceiling.

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

status = client.notetaker.status(session_id=session.session_id)
print(status.status)             # starting | live | stopping | completed
print(status.duration_seconds)   # elapsed so far while live, final once completed
print(status.participants)

result = client.notetaker.stop(session_id=session.session_id)
print(result.status)             # "stopping" — the bot is on its way out

stop() returns as soon as the request is accepted; the bot is asked to leave the way a person does rather than killed, so the last few seconds of speech still reach you. The final duration and transcript arrive on session.completed (reason: "stopped").

The bot also leaves on its own — when it is alone, after the silence window, at end_at, or at the service's own duration ceiling — so a forgotten session cannot run forever. Billing is by meeting minutes, metered every minute while the session is live rather than in one lump at the end. 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 (the bot could not join or the stream never started — it is pulled out and session.completed follows with reason: "bot_failed"), 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. Word timings are ISO-8601 strings, and each w carries a leading space, so "".join(word["w"] for word in words) rebuilds the sentence.

session.completed.reason is one of bot_exited (left on its own — alone, silence, or removed by the host), stopped, end_at_reached, bot_failed (the bot reported a failure or its container exited non-zero), manager_error (we lost track of the bot), or manager_shutdown (our service restarted).

The recording. record_video=True runs the screen capture; if your organization has a storage bucket connected, the video is uploaded to your bucket (never ours — the SDK takes no credentials for it) and session.completed.recording carries a presigned URL:

{ "url": "https://…", "duration_seconds": 48, "expires_in_days": 7 }

It is null when no bucket is connected — there record_video only decides whether the encoder runs, and the video is written locally and discarded. The URL is signed for GET only, so a HEAD on it answers 403.

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"