Python SDK — Quickstart

1
import os
2
from onepin import OnePinClient
3
4
client = OnePinClient(token=os.environ["ONEPIN_API_KEY"])  # the SDK does not auto-read env vars
5
6
# list() returns an envelope; the items are in .data
7
for t in client.templates.list(limit=10).data:
8
    print(t.id, t.name, "⭐" if t.is_starter else "")

1
import os, time, httpx
2
from onepin import OnePinClient
3
4
client = OnePinClient(token=os.environ["ONEPIN_API_KEY"])
5
# use the starter template (one script, one voice); swap for any single-voice template's id
6
template_id = next(t.id for t in client.templates.list(limit=50).data if t.is_starter)
7
locale = "en-us"                                           # language to generate
8
provider, model = "elevenlabs", "eleven_multilingual_v2"
9
# if runs.start() below returns 403, this model isn't on your plan;
10
# pick another from client.providers.list_catalog_provider_models(provider).data
11
12
# 1. Clone the template into a workflow (single responses carry the object at .data)
13
workflow_id = client.templates.clone(template_id, name="My first workflow").data.id
14
15
# 2. Pick a voice available for this language
16
voice = next(v for v in client.providers
17
             .list_catalog_provider_model_voices(provider, model, limit=50).data
18
             if locale in (v.supported_languages or []))
19
20
# 3. Fill the empty script and voice nodes, then save the definition
21
definition = client.workflows.get(workflow_id=workflow_id).data.model_dump()["definition"]
22
for node in definition["graph"]["nodes"]:
23
    if node["type"] == "source_script":
24
        node["config"] = {"input_type": "text", "source_language": locale,
25
                          "text": "Hello from the Onepin Python SDK quickstart."}
26
    elif node["type"] == "processing_generator":
27
        # a voice is assigned by provider_voice_id (not the Onepin voice id), provider, and model
28
        node["config"] = {"voice_map": {locale: [
29
            {"voice_id": voice.provider_voice_id, "provider": provider, "model": model}]}}
30
client.workflows.patch_workflow(workflow_id, definition=definition)
31
32
# 4. Run it and wait
33
run_id = client.workflows.runs.start(workflow_id).data.id
34
while True:
35
    status = client.workflows.runs.status(workflow_id, run_id).data
36
    print(status.status, f"{status.finished_steps}/{status.total_steps}")
37
    if status.status in ("completed", "failed", "cancelled"):
38
        break
39
    time.sleep(5)
40
if status.status != "completed":
41
    raise RuntimeError(f"run {status.status}: {status.error}")
42
43
# 5. Download the result (download_run returns a temporary URL; fetch the file from it)
44
download = client.workflows.download_run(workflow_id, run_id).data
45
resp = httpx.get(download.url)
46
resp.raise_for_status()
47
with open(download.filename, "wb") as f:
48
    f.write(resp.content)
49
print("saved:", download.filename)