Integration

Build a Voice Agent using Vobiz

View as Markdown

Overview

This guide shows you how to build a real-time voice agent that answers phone calls, using Vobiz for telephony and Sarvam AI for everything else: speech-to-text, the LLM, and text-to-speech. Unlike the other integration guides in this section, this one doesn’t use Pipecat. Vobiz’s bidirectional media stream is a plain JSON-over-WebSocket protocol, so a small FastAPI server talking to Sarvam’s REST API directly is all you need.

What You’ll Build

A voice agent that can:

  • Answer inbound phone calls on a Vobiz number
  • Listen to callers speaking, in multiple Indian languages
  • Understand and process what they say, with barge-in support
  • Respond back in a natural-sounding voice, over the phone, sentence by sentence as the reply streams in

Quick Overview

  1. Get an API key (Sarvam). Vobiz needs no API credentials for this flow.
  2. Install packages: pip install fastapi httpx python-dotenv python-multipart "uvicorn[standard]"
  3. Create a .env file with your API key and public URL
  4. Write a FastAPI server that speaks Vobiz’s WebSocket media protocol and calls Sarvam directly
  5. Create a Vobiz application pointing at your server, and attach your number to it
  6. Call your Vobiz number

Quick Start

1. Prerequisites

  • Python 3.9 or higher
  • ngrok, to expose your local server during development
  • A Vobiz account with a provisioned voice number
  • A Sarvam AI API key from your dashboard

Vobiz doesn’t need an account SID, auth token, or any credentials in your agent code for this flow. It fetches your Answer URL as plain HTTP when a call comes in, and everything it needs to route the call, including the stream and call identifiers, arrives over the WebSocket itself, just like Exotel’s Voicebot Applet.

2. Install Dependencies

pip install fastapi httpx python-dotenv python-multipart "uvicorn[standard]"

This guide talks to Sarvam over plain httpx calls rather than the sarvamai SDK or Pipecat, so the whole pipeline has one HTTP dependency and one authentication header (api-subscription-key). That also means audio format handling, including resampling, WAV wrapping, and container stripping, is your responsibility here, unlike the Pipecat-based guides where the transport and Sarvam services handle it for you.

3. Create Environment File

Create a file named .env in your project folder:

SARVAM_API_KEY=sk_xxxxxxxxxxxxxxxxxxxxxxxx
SARVAM_STT_MODEL=saaras:v4
SARVAM_LLM_MODEL=sarvam-105b
SARVAM_TTS_MODEL=bulbul:v3
PUBLIC_URL=https://your-tunnel.example.com
HTTP_PORT=8000
AGENT_LANGUAGE=en-IN
TTS_SPEAKER=anand
# Vobiz sends caller audio as mu-law at 8 kHz or Linear16 at 16 kHz.
STREAM_CONTENT_TYPE=audio/x-mulaw;rate=8000
OUTBOUND_SAMPLE_RATE=8000

Replace SARVAM_API_KEY with your real key from the Sarvam dashboard, and PUBLIC_URL with your ngrok URL once you have one (Step 5).

sarvam-30b has been deprecated. sarvam-105b is now the only supported chat model, and the /v1/chat/completions API rejects the model="sarvam-30b" value. If you have existing agents pinned to sarvam-30b, update them to sarvam-105b before it stops responding entirely.

sarvam-105b reasons before replying, and that reasoning is billed against max_tokens. Budget at least 1500 tokens per turn. Set it too low, and the API returns a valid response with empty content, which would otherwise strand the caller in silence.

4. Write the Sarvam Client

Create sarvam.py, a thin async wrapper over the three Sarvam endpoints the agent needs: speech-to-text, chat, and text-to-speech.

import base64
import io
import json
import os
import wave
from typing import AsyncIterator
import httpx
BASE_URL = "https://api.sarvam.ai"
STT_MODEL = os.getenv("SARVAM_STT_MODEL", "saaras:v4")
LLM_MODEL = os.getenv("SARVAM_LLM_MODEL", "sarvam-105b")
TTS_MODEL = os.getenv("SARVAM_TTS_MODEL", "bulbul:v3")
SAMPLE_WIDTH_BYTES = 2
class SarvamError(RuntimeError):
    pass
def pcm_to_wav(pcm_audio: bytes, sample_rate: int) -> bytes:
    """Wrap raw mono Linear16 samples in a WAV container for the STT upload."""
    buffer = io.BytesIO()
    with wave.open(buffer, "wb") as wav_file:
        wav_file.setnchannels(1)
        wav_file.setsampwidth(SAMPLE_WIDTH_BYTES)
        wav_file.setframerate(sample_rate)
        wav_file.writeframes(pcm_audio)
    return buffer.getvalue()
def raw_linear16(audio: bytes, expected_rate: int) -> bytes:
    """Vobiz playAudio needs container-free samples; Sarvam TTS may return a WAV."""
    if audio.startswith(b"RIFF"):
        with wave.open(io.BytesIO(audio), "rb") as wav_file:
            audio = wav_file.readframes(wav_file.getnframes())
    return audio
class SarvamClient:
    def __init__(self, api_key: str) -> None:
        self._headers = {"api-subscription-key": api_key}
        self._http = httpx.AsyncClient(base_url=BASE_URL, timeout=30.0)
    async def aclose(self) -> None:
        await self._http.aclose()
    async def transcribe(self, pcm_audio: bytes, sample_rate: int, language_code: str) -> str:
        wav_audio = pcm_to_wav(pcm_audio, sample_rate)
        response = await self._http.post(
            "/speech-to-text",
            headers=self._headers,
            files={"file": ("caller.wav", wav_audio, "audio/wav")},
            data={"model": STT_MODEL, "mode": "codemix", "language_code": language_code},
        )
        response.raise_for_status()
        return (response.json().get("transcript") or "").strip()
    async def stream_reply(
        self, messages: list[dict[str, str]], max_tokens: int = 2000
    ) -> AsyncIterator[str]:
        """Yield only reply text. These models also stream a private
        ``reasoning_content`` field that must never reach the caller's ear."""
        payload = {
            "model": LLM_MODEL,
            "messages": messages,
            "max_tokens": max_tokens,
            "stream": True,
        }
        async with self._http.stream(
            "POST", "/v1/chat/completions", headers=self._headers, json=payload
        ) as response:
            response.raise_for_status()
            async for line in response.aiter_lines():
                if not line.startswith("data:"):
                    continue
                data = line[len("data:"):].strip()
                if data in ("", "[DONE]"):
                    continue
                for choice in json.loads(data).get("choices", []):
                    delta = (choice.get("delta") or {}).get("content")
                    if delta:
                        yield delta
    async def synthesize(self, text: str, language_code: str, speaker: str, sample_rate: int) -> bytes:
        response = await self._http.post(
            "/text-to-speech",
            headers=self._headers,
            json={
                "text": text,
                "target_language_code": language_code,
                "speaker": speaker,
                "model": TTS_MODEL,
                "speech_sample_rate": sample_rate,
                "output_audio_codec": "linear16",
            },
        )
        response.raise_for_status()
        audios = response.json().get("audios") or []
        if not audios:
            raise SarvamError("Sarvam TTS returned no audio")
        return raw_linear16(base64.b64decode("".join(audios)), sample_rate)

raw_linear16() strips a WAV header if Sarvam returns one. Vobiz’s playAudio message requires raw, container-free Linear16 samples: sending it a WAV file (with its 44-byte header) produces a burst of noise at the start of every utterance.

5. Write the Call Session

Create agent.py. This holds the per-call pipeline: a simple energy-based VAD (voice activity detector) over inbound audio, then speech-to-text, the chat model, and text-to-speech, streamed back to the caller sentence by sentence:

import asyncio
import audioop
import base64
import json
import os
import re
from typing import Any
from sarvam import SarvamClient
OUTBOUND_SAMPLE_RATE = int(os.getenv("OUTBOUND_SAMPLE_RATE", "8000"))
SAMPLE_WIDTH_BYTES = 2
AGENT_LANGUAGE = os.getenv("AGENT_LANGUAGE", "en-IN")
TTS_SPEAKER = os.getenv("TTS_SPEAKER", "anand")
SYSTEM_PROMPT = (
    "You are a Sarvam AI voice assistant speaking with a caller on a phone "
    "line. You understand English and Indian languages, including code-mixed "
    "speech. Keep every reply to one or two short spoken sentences. Never use "
    "markdown, bullet points, emoji, or symbols, because your reply is read aloud."
)
GREETING = "Namaste! I am your Sarvam AI assistant. How can I help you today?"
# Split streamed LLM text on sentence enders, including the Devanagari danda,
# so the caller hears sentence one while the model is still writing sentence two.
SENTENCE_END = re.compile(r"[.!?|।॥]+[\s\"')\]]*|\n+")
MIN_TTS_CHARS = 24
SILENCE_THRESHOLD = 250       # RMS energy that counts as speech
END_OF_SPEECH_MS = 700        # trailing silence that ends an utterance
MIN_SPEECH_MS = 200           # shorter utterances are treated as noise
class CallSession:
    def __init__(self, websocket: Any) -> None:
        self.websocket = websocket
        self.stream_id: str | None = None
        self.input_sample_rate = 8000
        self.input_encoding = "audio/x-mulaw"
        self.sarvam = SarvamClient(os.getenv("SARVAM_API_KEY", ""))
        self.conversation = [{"role": "system", "content": SYSTEM_PROMPT}]
        self._audio = bytearray()
        self._speech_ms = 0.0
        self._silence_ms = 0.0
        self._speech_active = False
        self._playback_active = False
        self._tasks: set[asyncio.Task] = set()
    async def handle_message(self, raw_message: str) -> None:
        data = json.loads(raw_message)
        event = data.get("event")
        if event == "start":
            await self._handle_start(data)
        elif event == "media":
            await self._handle_media(data)
        elif event in {"playedStream", "clearedAudio"}:
            self._playback_active = False
    async def _handle_start(self, data: dict) -> None:
        start = data.get("start") or {}
        self.stream_id = data.get("streamId") or start.get("streamId")
        # Trust the negotiated format rather than assuming a rate: Vobiz sends
        # mu-law at 8 kHz or Linear16 at 16 kHz depending on <Stream contentType>.
        media_format = start.get("mediaFormat") or {}
        self.input_sample_rate = int(media_format.get("sampleRate", 8000))
        self.input_encoding = str(media_format.get("encoding", "audio/x-mulaw"))
        self._spawn(self._speak(GREETING))
    async def _handle_media(self, data: dict) -> None:
        media = data.get("media") or {}
        payload = media.get("payload")
        if not payload:
            return
        raw = base64.b64decode(payload)
        pcm = audioop.ulaw2lin(raw, 2) if self.input_encoding == "audio/x-mulaw" else raw
        chunk_ms = len(pcm) * 1000 / (self.input_sample_rate * SAMPLE_WIDTH_BYTES)
        rms = audioop.rms(pcm, SAMPLE_WIDTH_BYTES)
        if rms >= SILENCE_THRESHOLD:
            if self._playback_active:
                await self._clear_audio()  # barge-in
            self._speech_active = True
            self._speech_ms += chunk_ms
            self._silence_ms = 0
            self._audio.extend(pcm)
        elif self._speech_active:
            self._silence_ms += chunk_ms
            self._audio.extend(pcm)
        if self._speech_active and self._silence_ms >= END_OF_SPEECH_MS:
            utterance, speech_ms = bytes(self._audio), self._speech_ms
            self._audio.clear()
            self._speech_ms = self._silence_ms = 0
            self._speech_active = False
            if speech_ms >= MIN_SPEECH_MS:
                self._spawn(self._process_utterance(utterance))
    async def _process_utterance(self, pcm_audio: bytes) -> None:
        transcript = await self.sarvam.transcribe(pcm_audio, self.input_sample_rate, AGENT_LANGUAGE)
        if not transcript:
            return
        self.conversation.append({"role": "user", "content": transcript})
        pending = ""
        async for delta in self.sarvam.stream_reply(self.conversation):
            pending += delta
            match = SENTENCE_END.search(pending)
            if match and match.end() >= MIN_TTS_CHARS:
                sentence, pending = pending[:match.end()].strip(), pending[match.end():]
                if sentence:
                    await self._speak(sentence)
        if pending.strip():
            await self._speak(pending.strip())
    async def _speak(self, text: str) -> None:
        pcm = await self.sarvam.synthesize(text, AGENT_LANGUAGE, TTS_SPEAKER, OUTBOUND_SAMPLE_RATE)
        await self._play(pcm)
    async def _play(self, pcm_audio: bytes) -> None:
        chunk_size = OUTBOUND_SAMPLE_RATE * SAMPLE_WIDTH_BYTES * 20 // 1000  # 20 ms chunks
        self._playback_active = True
        for offset in range(0, len(pcm_audio), chunk_size):
            await self.websocket.send_text(json.dumps({
                "event": "playAudio",
                "streamId": self.stream_id,
                "media": {
                    "contentType": "audio/x-l16",
                    "sampleRate": OUTBOUND_SAMPLE_RATE,
                    "payload": base64.b64encode(pcm_audio[offset:offset + chunk_size]).decode(),
                },
            }))
        await self.websocket.send_text(json.dumps({"event": "checkpoint", "streamId": self.stream_id, "name": "tts"}))
    async def _clear_audio(self) -> None:
        await self.websocket.send_text(json.dumps({"event": "clearAudio", "streamId": self.stream_id}))
        self._playback_active = False
    def _spawn(self, coroutine) -> None:
        task = asyncio.create_task(coroutine)
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)
    async def close(self) -> None:
        for task in list(self._tasks):
            task.cancel()
        await self.sarvam.aclose()

The chat model’s streamed reply is split into sentences with SENTENCE_END and spoken as each one completes, rather than waiting for the full reply. This is what keeps turn latency down: the caller hears sentence one while the model is still generating sentence two.

Barge-in works by watching inbound RMS energy during playback: if the caller starts talking while the agent is still speaking, the session sends clearAudio and starts listening for a new utterance. This energy-based VAD can’t distinguish speech from background noise, so on noisy phone lines you may need to raise SILENCE_THRESHOLD or disable barge-in entirely.

6. Write the FastAPI Server

Create server.py, exposing the Answer URL Vobiz fetches when a call comes in, and the WebSocket it streams audio over:

import html
import os
from pathlib import Path
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import Response
from agent import CallSession
load_dotenv(Path(__file__).resolve().parent / ".env")
PUBLIC_URL = os.getenv("PUBLIC_URL", "").rstrip("/")
STREAM_CONTENT_TYPE = os.getenv("STREAM_CONTENT_TYPE", "audio/x-mulaw;rate=8000")
app = FastAPI()
def websocket_url(public_url: str) -> str:
    if public_url.startswith("https://"):
        return "wss://" + public_url.removeprefix("https://") + "/ws"
    return "ws://" + public_url.removeprefix("http://") + "/ws"
@app.api_route("/answer", methods=["GET", "POST"])
async def answer(request: Request) -> Response:
    """The XML Vobiz executes once the call is answered."""
    base_url = html.escape(PUBLIC_URL, quote=True)
    ws_url = html.escape(websocket_url(PUBLIC_URL), quote=False)
    xml = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
    <Stream bidirectional="true"
            audioTrack="inbound"
            keepCallAlive="true"
            contentType="{html.escape(STREAM_CONTENT_TYPE, quote=True)}"
            statusCallbackUrl="{base_url}/stream-status"
            statusCallbackMethod="POST">
        {ws_url}
    </Stream>
    <Hangup/>
</Response>"""
    return Response(content=xml, media_type="application/xml")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    session = CallSession(websocket)
    try:
        async for message in websocket.iter_text():
            await session.handle_message(message)
    except WebSocketDisconnect:
        pass
    finally:
        await session.close()
@app.post("/stream-status")
async def stream_status(request: Request) -> Response:
    await request.form()
    return Response(status_code=204)
@app.post("/hangup")
async def hangup(request: Request) -> Response:
    await request.form()
    return Response(status_code=204)
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("HTTP_PORT", "8000")))

<Record> and <Stream> are siblings under <Response>, not nested inside one another. If you add call recording later with <Record recordSession="true" .../>, place it as its own top-level element alongside <Stream>. Nesting <Stream> inside <Record>, or letting a Record callback return <Hangup/>, ends the call before the agent WebSocket ever starts.

7. Configure Vobiz to Reach Your Agent

Start ngrok:

ngrok http 8000

ngrok prints a forwarding URL that looks like https://xxxx.ngrok-free.app. Put that exact URL, unchanged, into PUBLIC_URL in your .env file. The server derives the wss:// WebSocket URL from it automatically.

Free ngrok URLs change every time you restart ngrok. If your agent stops receiving calls, update PUBLIC_URL in .env to match the URL ngrok is currently printing, and restart server.py.

Create an application and point it at the server:

The Answer and Hangup URLs live on a Vobiz application, not directly on the number. You create the application first, then attach your number to it.

  1. In the Vobiz dashboard, go to Applications
  2. Create a new application
  3. Set the Answer URL to https://xxxx.ngrok-free.app/answer (method: POST)
  4. Set the Hangup URL to https://xxxx.ngrok-free.app/hangup (method: POST)
  5. Save the application
  6. Go to your number’s settings and attach it to the application you just created

Step 6 is easy to miss but required. Without a number attached to the application, calls have nowhere to route to, even if the application itself is configured correctly.

8. Run and Test Your Agent

python server.py

Call your Vobiz number from any phone. Vobiz fetches /answer, opens the WebSocket at /ws, and your agent greets the caller.

Audio Formats

The two directions of a Vobiz call are independent and configured separately.

Vobiz → your agent, set by <Stream contentType>:

FormatUse
audio/x-mulaw;rate=80008 kHz inbound (default in this guide)
audio/x-l16;rate=1600016 kHz inbound

Your agent → Vobiz, set per playAudio message: Linear16 at 8000, 16000, or 24000 Hz. 24 kHz applies to playback only, not inbound audio.

playAudio payloads must be raw mono little-endian Linear16, Base64-encoded, with no WAV header. Declaring a sampleRate in the playAudio message does not resample the bytes you send. If you declare 24000 while the payload is actually 8 kHz audio, the caller hears distorted, sped-up “chipmunk” audio. raw_linear16() in Step 4 strips a WAV container if Sarvam happens to return one, but it doesn’t resample.

Read start.mediaFormat on the start event rather than assuming a rate, as this guide’s _handle_start does. Vobiz tells you the exact encoding and sample rate it negotiated for each call.

Customization Examples

Hindi Voice Agent

AGENT_LANGUAGE = "hi-IN"
TTS_SPEAKER = "simran"  # or: priya, ishita, kavya, aditya, anand, rohan

Multilingual Agent (Auto-detect)

AGENT_LANGUAGE = "unknown"  # Sarvam auto-detects the caller's spoken language
TTS_SPEAKER = "anand"

Use AGENT_LANGUAGE = "unknown" for support lines where callers might speak any of several languages. Sarvam detects the spoken language per utterance, so the same agent can handle a Hindi caller followed by a Tamil caller without any code changes.

Available Language Codes

LanguageCode
English (India)en-IN
Hindihi-IN
Bengalibn-IN
Tamilta-IN
Telugute-IN
Gujaratigu-IN
Kannadakn-IN
Malayalamml-IN
Marathimr-IN
Punjabipa-IN
Odiaod-IN
Auto-detectunknown

Speaker Voices (Bulbul v3)

Male: Anand (default here), Aditya, Rahul, Rohan, Amit, Dev, Ratan, Varun, Manan, Sumit, Kabir, Aayan, Ashutosh, Advait, Tarun, Sunny, Mani, Gokul, Vijay, Mohit, Rehan, Soham, Shubh

Female: Ritu, Priya, Neha, Pooja, Simran, Kavya, Ishita, Shreya, Roopa, Tanya, Shruti, Suhani, Kavitha, Rupali

Speakers are model-specific. A bulbul:v2 voice such as anushka is rejected by bulbul:v3. Check that the speaker you pick is valid for the SARVAM_TTS_MODEL you’re using.

Understanding the Call Flow

  1. Caller dials your Vobiz number. Vobiz fetches your Answer URL, and executes the returned <Stream> XML, opening a bidirectional WebSocket to your agent.
  2. start event: Vobiz sends the stream ID and negotiated audio format. Your agent speaks a greeting.
  3. media events: Caller audio arrives as base64-encoded chunks. An energy-based VAD buffers speech and detects when the caller has stopped talking.
  4. STT: The buffered utterance is sent to Sarvam’s Saaras model for transcription.
  5. LLM: The transcript is added to the conversation and streamed through a Sarvam chat model.
  6. TTS: Each completed sentence of the reply is synthesized with Sarvam’s Bulbul and sent back as playAudio events, so playback starts before the whole reply is ready.
  7. Barge-in: If the caller speaks while the agent is still playing audio, a clearAudio event stops playback and a new utterance begins.

Pro Tips

  • Use AGENT_LANGUAGE = "unknown" to auto-detect the caller’s language on multilingual support lines.
  • Sarvam’s models understand code-mixing, so your agent can naturally handle Hinglish, Tanglish, and other mixed languages common on real support calls.
  • Keep the system prompt explicit about reply length and formatting. Since the reply is read aloud, ask the model to avoid markdown, bullet points, and emoji.
  • If a chat completion ever returns empty content (the model spent its whole token budget reasoning), have the agent speak a fallback line instead of going silent.

Log every transcript and reply to a file or database for post-call analytics. It’s cheap and doesn’t slow down the live pipeline.

Troubleshooting

Call connects but there’s no audio. Confirm PUBLIC_URL in .env matches your current ngrok URL exactly, and that the Answer URL configured on your Vobiz application points at /answer on that same URL.

Call doesn’t reach your agent at all. Confirm your number is actually attached to the application, not just that the application has the right URLs. A number with no application attached has nowhere to route an incoming call.

Choppy or chipmunk-pitched audio. Check that OUTBOUND_SAMPLE_RATE and the sampleRate you declare in each playAudio message match the actual bytes you’re sending. Declaring a rate that doesn’t match the payload distorts pitch and speed without erroring.

Playback has a burst of noise at the start. Sarvam TTS occasionally returns a WAV-wrapped file instead of raw PCM. Make sure raw_linear16() runs on every synthesized chunk before it’s sent as playAudio.

API key errors. Make sure SARVAM_API_KEY is in your .env file and that the file sits next to server.py.

Empty or truncated replies. sarvam-105b spends tokens on internal reasoning before writing the reply. Raise max_tokens on the chat request; 1500 to 2000 is a safe starting point.

Barge-in doesn’t trigger, or triggers on background noise. The VAD here is a simple RMS energy threshold, not real speech detection. Raise SILENCE_THRESHOLD on noisy lines, or set it high enough to ignore ambient noise from the caller’s environment.

Additional Resources

Need Help?

Happy Building!