Python

call = v.calls.originate(
    to="+15551234567",
    from_="+15558675309",
    trunk_id="trunk_main",
    agent="healthcare-assistant",   # optional: engine drives the audio
    ring_timeout_sec=30,
)
print(call.sid, call.status)

# Fetch current state for a known sid
call = v.calls.get(sid="call_...")

# Transfer — pass EXACTLY ONE of to (a PSTN number) or agent (re-attach)
call.transfer(to="+15557654321")
call.transfer(agent="tier2-support")

# End the call
call.hangup()

v.agents.attach(call.sid, "healthcare-assistant")

def on_uplink(frame: bytes, ts_us: int) -> None:
    my_asr.feed(frame)          # caller audio → your pipeline

bridge = v.audio_bridge.attach(call.sid, codec="opus", on_uplink=on_uplink)

# Send audio back to the caller
for opus_frame in my_tts.stream():
    bridge.publish_downlink(opus_frame)

# Release when done — this closes the MoQT session and drops both tracks
bridge.close()

import os, signal, struct, threading, time
from clutchcall.voice import Voice

v = Voice(
    base_url="https://engine.clutchcall.dev",
    api_key=os.environ["CLUTCHCALL_API_KEY"],
    org_id="org_abc",
)

# 1. Originate; `agent` lets the engine drive the audio end-to-end.
call = v.calls.originate(
    to="+15551234567", from_="+15558675309",
    trunk_id="trunk_main", agent="healthcare-assistant",
)
print(f"dialing: sid={call.sid}")

# 2. Tap the uplink in parallel — write [u32 len][u64 ts_us] + frame.
out = open("caller.opus", "wb")

def on_uplink(frame: bytes, ts_us: int) -> None:
    out.write(struct.pack("<IQ", len(frame), ts_us))
    out.write(frame)

bridge = v.audio_bridge.attach(call.sid, codec="opus", on_uplink=on_uplink)

# 3. Hold until SIGTERM, then tear down cleanly.
stop = threading.Event()
signal.signal(signal.SIGTERM, lambda *_: stop.set())
try:
    while not stop.is_set():
        stop.wait(timeout=2.0)
finally:
    bridge.close()
    call.hangup()
    out.close()