Python SDK

pip install yotel

import os
import yotel

client = yotel.Client(api_key=os.environ["YOTEL_KEY"])

# Create campaign
c = client.campaigns.create(
    name="Q4 outreach",
    dial_mode="predictive",
    predictive_target_abandon=0.02,
)

# Push lead
lead = client.leads.create(
    campaign_id=c.id,
    phone="9876543210",
    name="Priya",
    email="priya@example.com",
)

# Bulk push (up to 500)
result = client.leads.bulk(
    campaign_id=c.id,
    leads=[
        {"phone": "9876543211"},
        {"phone": "9876543212", "name": "Arjun"},
    ],
)
print(result.valid, "added,", result.duplicates, "skipped")

from yotel.webhook import verify_and_parse, InvalidSignature

# In your FastAPI / Flask / Django handler:
raw = await request.body()
try:
    event = verify_and_parse(
        raw,
        headers=dict(request.headers),
        signing_secret=os.environ["YOTEL_WEBHOOK_SECRET"],
    )
except InvalidSignature:
    return Response(status_code=401)

# event.event_type is typed: 'call.ended' | 'campaign.paused' | ...
if event.event_type == "call.ended":
    call_id = event.data["call_id"]
    ...

# Default — let the SDK retry
client = yotel.Client(api_key="...")

# Disable — handle 429 yourself
client = yotel.Client(api_key="...", retry_on_429=False)

import yotel

async with yotel.AsyncClient(api_key="...") as client:
    c = await client.campaigns.create(name="...", dial_mode="predictive")

from yotel import (
    AuthenticationError,
    RateLimitError,
    ValidationError,
    APIError,
)

try:
    client.leads.create(campaign_id="c-1", phone="bad")
except ValidationError as e:
    print(e.errors)  # Pydantic-shaped field errors
except AuthenticationError:
    ...
except RateLimitError as e:
    print("retry after", e.retry_after)
except APIError as e:
    print(e.status_code, e.detail)

from yotel.oauth import AuthorizationCodeFlow, generate_code_verifier

flow = AuthorizationCodeFlow(
    client_id="sfdc_yotel",
    client_secret=os.environ["YOTEL_CLIENT_SECRET"],
    redirect_uri="https://app.example/cb",
    scopes=["campaigns:read", "openid", "offline_access"],
)
verifier = generate_code_verifier()
url = flow.build_authorize_url(state="csrf-1", code_verifier=verifier)
# ... redirect user, callback handler ...
tokens = flow.exchange_code(code=code, code_verifier=verifier)
tokens = tokens.refresh()  # rotated refresh_token

pip install "yotel[oauth]"

# Get a signed URL (7-day expiry)
recording = client.recordings.get_signed_url("call_id")
print(recording["url"], recording["stereo"])

# Download to disk
path = client.recordings.download("call_id", "recording.wav")

# Load into memory
audio = client.recordings.fetch_bytes("call_id")

# Create
va = client.voice_agents.create(
    name="Support bot",
    ws_url="wss://ai.example.com/ws",
    max_concurrent=50,
)

# List / Get
all_agents = client.voice_agents.list()
one = client.voice_agents.get(va.id)

# Update (PATCH)
client.voice_agents.update(va.id, max_concurrent=100)

# Archive (soft delete) or hard delete
client.voice_agents.archive(va.id)
client.voice_agents.delete(va.id)

# Tenant default
client.voice_agents.set_as_default(va.id)
default_id = client.voice_agents.get_default()

# Transfer call to a human agent
client.ai_sessions.transfer(
    call_id="call_abc",
    destination_type="agent",
    destination="agent_xyz",
    callback_token="yt_cb_...",
)

# Hang up
client.ai_sessions.hangup("call_abc", callback_token="yt_cb_...")

# Set disposition
client.ai_sessions.set_disposition(
    "call_abc",
    disposition="interested",
    notes="Wants premium plan",
    callback_token="yt_cb_...",
)

# Get current call state
state = client.ai_sessions.get_call_state("call_abc", callback_token="yt_cb_...")

from yotel import (
    NoVoiceAgentConfigured,   # 424 — no agent resolved in hierarchy
    VoiceAgentInUse,          # 409 — active sessions prevent deletion
    MaxConcurrentExceeded,    # 429 — concurrency limit reached
    ConferenceMembershipRequired,  # 403 — whisper/barge needs conference
    AISessionTerminated,      # 410 — call already ended
)