Python SDK

pip install vaani-sdk

git clone https://github.com/your-org/vaani-sdk
cd vaani-sdk
pip install -e .

import os
from vaani_sdk import VaaniClient

client = VaaniClient(api_key=os.getenv("VAANI_API_KEY"))

with VaaniClient(api_key=os.getenv("VAANI_API_KEY")) as client:
    resp = client.health()
    print(resp.status)  # "healthy"

from vaani_sdk import VaaniClient

with VaaniClient(api_key="YOUR_API_KEY") as client:
    result = client.trigger_call(
        agent_id="8cf3373e-eb6f-4b4c-9f3c-324a56a91147",
        contact_number="+919876543210",
        name="Rahul Sharma",
        metadata={
            "customer_name": "Rahul Sharma",
            "property_name": "Prateek Laurel",
        },
    )
    print(result.call_id)            # "outbound-1776774443-863aa698"
    print(result.live_captions_url)  # signed Server-Sent Events URL (telephony)

history = client.get_call_history(page=1, page_size=20)

print(history.pagination.total)       # total records
print(history.pagination.total_pages) # total pages
print(history.pagination.has_next)

# Optional filters
completed = client.get_call_history(
    status="Completed",
    agent_id="8cf3373e-eb6f-4b4c-9f3c-324a56a91147",
    start_date="2026-07-01",
)

for call in history.data:
    print(call.call_id, call.raw.get("call_status"), call.raw.get("duration_ms"))

transcript = client.get_transcript("outbound-1776774443-863aa698")
print(transcript.transcript)

import time

details = client.get_call_details("outbound-1776774443-863aa698")
while details.is_processing:          # API answered 202: post-processing still running
    time.sleep(details.retry_after or 5)
    details = client.get_call_details("outbound-1776774443-863aa698")

print(details.raw.get("summary"))
print(details.raw.get("entity"))

audio = client.stream_audio("outbound-1776774443-863aa698")
with open("recording.ogg", "wb") as f:
    f.write(audio.content)

# For large files, stream in chunks:
with open("recording.ogg", "wb") as f:
    for chunk in client.stream_audio_iter("outbound-1776774443-863aa698"):
        f.write(chunk)

from vaani_sdk import VaaniClient

with VaaniClient(api_key="YOUR_API_KEY") as client:
    # Create a session
    session = client.sessions.create(
        agent_id="33ac5907-0e44-43c4-934e-08e335cca6ee",
        ui_type="widget",
        grace_period=60,
        session_ttl=1800,
        initial_message="User is asking about order #1234",
        metadata={"user_id": "user_123", "ticket_id": "T-1234"},
    )
    print(session.session_token)  # JWT session token
    print(session.session_url)    # embeddable/shareable URL

    # Exchange token for LiveKit credentials
    creds = client.sessions.connect(session.session_token)
    print(creds.lk_url)    # "wss://livekit.vaanivoice.ai"
    print(creds.lk_token)  # LiveKit participant JWT - pass to LiveKit SDK
    print(creds.room_name) # room name

    # Check status
    status = client.sessions.get(session.session_token)
    print(status.status)   # "created" | "agent_ready" | "active" | "reconnecting"
    # Once a session has ended (or expired) this call fails with a 410 "Gone" error

    # End the session
    client.sessions.disconnect(session.session_token)
    # Or end immediately with no grace period:
    client.sessions.disconnect(session.session_token, explicit_end=True)

import asyncio
import json
import websockets
from vaani_sdk import VaaniClient

client = VaaniClient(api_key="YOUR_API_KEY")

async def listen(session_token: str) -> None:
    url = client.sessions.ws_url(session_token)
    async with websockets.connect(url) as ws:
        async for message in ws:
            event = json.loads(message)
            if event["event"] == "agent_ready":
                print("Agent has joined the room")
            elif event["event"] == "agent_timeout":
                print("Agent failed to join:", event.get("reason"))
            elif event["event"] == "session_ended":
                print("Session ended:", event.get("reason"))

asyncio.run(listen("session-token-here"))

result = client.create_agent(
    agent_display_name="my-support-bot",
    config={
        "persona": {
            "identity": {
                "system_prompt": "You are a helpful support agent."
            }
        }
    },
)
print(result.agent_id)                       # "33ac5907-0e44-43c4-934e-08e335cca6ee"
print(result.agent_display_name)             # "my-support-bot"
# The API returns the full saved agent configuration; it is available as result.raw.
# result.agent_name is None, because the API response has no "agent_name" field.

# List all agents (or client.list_agents())
agents = client.agents.list()
for agent in agents:
    print(agent.id, agent.display_name, agent.agent_name)

# Search, paginate and sort
filtered = client.agents.list(search="support", page=1, page_size=10, sort_by="name", sort_order="asc")

# Only archived agents
archived = client.agents.list(show_archived=True)

# Minimal list of published agents (no `id`, unpaginated)
published = client.agents.list(published_only=True)

# Get the full configuration of one agent
agent = client.agents.get("33ac5907-0e44-43c4-934e-08e335cca6ee")
print(agent.agent_id, agent.agent_display_name)
persona = agent.raw["persona"]   # full persona / training / experience / analysis in .raw

# Delete an agent (or client.delete_agent("33ac5907-0e44-43c4-934e-08e335cca6ee"))
result = client.agents.delete("33ac5907-0e44-43c4-934e-08e335cca6ee")
print(result["message"])  # "Agent deleted successfully"

# Update persona (system prompt, voice, language)
result = client.update_agent_persona(
    agent_id="33ac5907-0e44-43c4-934e-08e335cca6ee",
    persona={
        "identity": {
            "system_prompt": "You are a helpful customer service agent.",
        },
        "senses_capabilities": {
            "language": "English",
        }
    },
)

# Update training (FAQ, guardrails)
client.update_agent_training(
    agent_id="33ac5907-0e44-43c4-934e-08e335cca6ee",
    training={
        "know_how": {
            "faq": {
                "What are your hours?": "We are open 9 AM to 5 PM."
            }
        }
    },
)

# Update experience (call settings, idle behavior, background noise)
client.update_agent_experience(
    agent_id="33ac5907-0e44-43c4-934e-08e335cca6ee",
    experience={
        "settings": {
            "call_settings": {"max_call_duration": 30}
        }
    },
)

# Update analysis (dispositions, data extraction)
client.update_agent_analysis(
    agent_id="33ac5907-0e44-43c4-934e-08e335cca6ee",
    analysis={
        "extraction": {
            "data_collection": {
                "enabled": True,
                "data_points": [
                    {"name": "customer_email", "prompt": "Extract the customer's email address", "nullable": True}
                ]
            }
        }
    },
)

# Update deployment (inbound/outbound phone numbers)
client.update_agent_deployment(
    agent_id="33ac5907-0e44-43c4-934e-08e335cca6ee",
    deployment={
        "deployment": {
            "phone": {
                "call_type": {
                    "Inbound": "",
                    "Outbound": ["+919876543210"]
                }
            }
        }
    },
)

with VaaniClient(api_key="YOUR_API_KEY") as client:
    campaign = client.campaigns.create(
        name="Q3 Outreach",
        agent_id="8cf3373e-eb6f-4b4c-9f3c-324a56a91147",
        csv_content=open("contacts.csv", "rb").read(),   # columns: name, phone_number
        config={
            "concurrency": 3,
            "from_date": "2026-07-10",
            "to_date": "2026-07-31",
            "start_time": "05:00",   # UTC, inside 02:30-14:29
            "end_time": "12:00",
            "days_of_week": "monday,tuesday,wednesday,thursday,friday",
        },
    )
    print(campaign.status)   # "populating", then active / scheduled / paused_out_of_range / failed

    client.campaigns.pause(campaign.id)
    client.campaigns.resume(campaign.id)
    client.campaigns.stop(campaign.id)     # completes the campaign immediately
    client.campaigns.cancel(campaign.id)   # permanent

    active = client.campaigns.list(status="active")
    results = client.campaigns.get_results(campaign.id)   # {"success": True, "download_url": "..."}

import asyncio
from vaani_sdk import AsyncVaaniClient

async def main():
    async with AsyncVaaniClient(api_key="YOUR_API_KEY") as client:
        # Flat methods
        result = await client.trigger_call(
            agent_id="8cf3373e-eb6f-4b4c-9f3c-324a56a91147",
            contact_number="+919876543210",
            name="Nishank",
        )
        print(result.call_id)

        # Resource namespaces
        session = await client.sessions.create(agent_id="8cf3373e-eb6f-4b4c-9f3c-324a56a91147")
        creds = await client.sessions.connect(session.session_token)
        print(creds.lk_url, creds.lk_token)

        calls = await client.calls.list(page=1, page_size=10)
        agents = await client.agents.list()

asyncio.run(main())

from vaani_sdk import (
    VaaniClient,
    AuthenticationError,
    ForbiddenError,
    InsufficientBalanceError,
    NotFoundError,
    SessionExpiredError,
    VaaniError,
)

with VaaniClient(api_key="YOUR_API_KEY") as client:
    try:
        session = client.sessions.create(agent_id="my-agent")
        creds = client.sessions.connect(session.session_token)
    except AuthenticationError:
        print("Invalid API key")
    except ForbiddenError:
        print("Agent not owned by this API key")
    except InsufficientBalanceError:
        print("Wallet balance is too low")
    except SessionExpiredError:
        print("Session has already ended")
    except NotFoundError:
        print("Agent not found")
    except VaaniError as exc:
        print(f"API error {exc.status_code}: {exc.message}")