Skip to main content

Your first request

A complete end-to-end integration, showing all the pieces you'll need in production:

  1. Submit the audio
  2. Poll with escalating intervals
  3. Handle queued / processing / terminal statuses
  4. Respect Retry-After on 429s
  5. Extract and use the scores
Speech module example

This walks through a full integration using the Speech module. The same X-API-Key auth and async submit → poll pattern applies to every module — see CV DeepMatch, CV DeepSearch, and Interview for their endpoints and specifics.

Full working example​

import os
import time
import requests

API_BASE  = "https://platform.zenhire.ai"
API_KEY   = os.environ["ZENHIRE_API_KEY"]   # zh_…
AUDIO     = "interview.mp3"

def analyze(audio_path: str, external_id: str | None = None) -> dict:
    # 1. Submit
    with open(audio_path, "rb") as f:
        data = {"language": "en"}
        if external_id:
            data["externalId"] = external_id
        r = requests.post(
            f"{API_BASE}/api/v1/speech/analyze",
            headers={"X-API-Key": API_KEY},
            files={"audio": f},
            data=data,
            timeout=60,
        )
    r.raise_for_status()
    submit = r.json()
    request_id = submit["id"]
    print(f"Submitted: {request_id} (status={submit['status']})")

    # 2. Poll
    start = time.time()
    interval = submit.get("pollIntervalSeconds", 15)

    while time.time() - start < 20 * 60:  # 20-minute budget
        time.sleep(interval)

        resp = requests.get(
            f"{API_BASE}/api/v1/speech/analyze/{request_id}",
            headers={"X-API-Key": API_KEY},
            timeout=30,
        )

        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", interval))
            time.sleep(retry_after)
            continue

        resp.raise_for_status()
        data = resp.json()
        status = data["status"]

        if status in ("success", "partial"):
            return data
        if status == "failed":
            err = data.get("error") or {}
            raise RuntimeError(f"{err.get('code')}: {err.get('message')}")

        # Escalate: 15s for first 2 min, then 30s, then 60s
        elapsed = time.time() - start
        interval = 15 if elapsed < 120 else 30 if elapsed < 300 else 60

    raise TimeoutError(
        f"Polling budget exhausted for {request_id} "
        "— the id is still valid, you can resume later."
    )


if __name__ == "__main__":
    result = analyze(AUDIO, external_id="candidate-abc-123")
    s = result["scores"]
    print(f"Overall: {s['overall']} ({s['cefrLevel']})")
    print(f"  Vocab:   {s['vocabulary']} ({s['vocabularyCefr']})")
    print(f"  Fluency: {s['fluency']} ({s['fluencyCefr']})")
    print(f"  Accent:  {s['accent']} ({s['accentCefr']})")
    print(f"\n{result['hrRecommendation']}")

What this example gets right​

  • Escalating poll intervals — starts at 15s, backs off as the run takes longer. Don't hammer every 5 seconds.
  • Handles 429 on poll — respects the Retry-After header.
  • Raises on failed, returns on success/partial — both success and partial contain usable scores.
  • Sets externalId — your own correlation tag, echoed back on every response. Filterable on GET /api/v1/speech/runs. Not unique, so safe to reuse across retries.
  • Does not treat queued or processing as errors — they're normal mid-run states.
  • 20-minute overall budget, then exits cleanly — the run id does not expire, so you can resume later.

Next​