Media Streams

Media Streams connects a phone call to your bot over one WebSocket. On connect Edesy sends one JSON text frame with the call details, then raw L16 PCM audio (signed 16-bit little-endian, mono, 8 kHz or 16 kHz) flows as binary frames in both directions: the caller's voice to you, your bot's voice back. There is no base64 and no mu-law, so your bot just moves bytes.

Connect a call to your bot#

Inbound: point a number at your WebSocket once.

curl -X PUT https://voice-api.edesy.in/v1/phone-numbers/{id}/inbound \
  -H "Authorization: Bearer $EDESY_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "target": "media_stream", "stream_url": "wss://bot.example.com/ws", "sample_rate": 16000 }'

DELETE/v1/phone-numbers/{id}/inbound detaches it. See Inbound Routing.

Outbound: pass stream_url when you place the call.

curl -X POST https://voice-api.edesy.in/v1/calls \
  -H "Authorization: Bearer $EDESY_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "from": "+9180XXXXXXXX",
    "to": "+9198XXXXXXXX",
    "stream_url": "wss://bot.example.com/ws",
    "stream_sample_rate": 16000,
    "custom_parameters": { "assistant_id": "asst_456" }
  }'

The wire protocol is identical in both directions, so the same bot serves inbound and outbound calls.

Protocol#

1. Metadata frame#

The first frame Edesy sends is a text frame with the call details. The keys are camelCase.

{
  "sampleRate": 16000,
  "mixType": "mono",
  "callSid": "d01624fb-0000-4000-8000-000000000000",
  "direction": "outbound",
  "from": "+9180XXXXXXXX",
  "to": "+9198XXXXXXXX",
  "callStatus": "in-progress",
  "customerData": {
    "call_sid": "CA_6f1c2d3e-0000-4000-8000-000000000000",
    "tenant_id": "2b7e0a1c-0000-4000-8000-000000000000",
    "custom_parameters": "{\"assistant_id\":\"asst_456\"}"
  }
}

2. Audio#

After the metadata frame, binary frames carry audio:

• Edesy to bot: the caller's audio.
• Bot to Edesy: audio to play to the caller, in the same format and sample rate.

3. Control messages#

Closing the WebSocket also ends the bot's part of the call.

Preflight: POST/v1/stream-test#

Before you route real calls, check that the Edesy network can reach your bot. The test connects exactly as a live call would, streams a few seconds of silence, and reports each step.

curl -X POST https://voice-api.edesy.in/v1/stream-test \
  -H "Authorization: Bearer $EDESY_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "stream_url": "wss://bot.example.com/ws", "sample_rate": 16000, "seconds": 2 }'

{
  "ok": true,
  "stream_url": "wss://bot.example.com/ws",
  "sample_rate": 16000,
  "steps": [
    { "step": "dns", "ok": true, "detail": "..." },
    { "step": "tcp", "ok": true, "detail": "..." },
    { "step": "tls", "ok": true, "detail": "..." },
    { "step": "upgrade", "ok": true, "detail": "..." },
    { "step": "audio", "ok": true, "detail": "..." }
  ],
  "audio_frames_received": 0,
  "audio_bytes_received": 0,
  "text_frames_received": 0
}

When a step fails, the remaining steps are marked as skipped and hint names the one thing to fix. A bot that sends no audio back is not necessarily broken (it may be waiting for speech), so the counts are reported rather than judged. Addresses that are not on the public internet are refused, because live calls could not reach them either. The endpoint is rate limited.

Call events#

Readable on GET/v1/calls/{CallSid}/events:

Example: Pipecat bot in Python#

Pipecat's stock Twilio serializer expects base64 mu-law, so it will not work as-is. Use an Edesy serializer that moves raw PCM. The sample below runs Deepgram STT, an OpenAI LLM, Cartesia TTS and Silero VAD; swap any of them.

requirements.txt:

pipecat-ai[silero,deepgram,openai,cartesia,websocket]>=0.0.60
fastapi
uvicorn[standard]

edesy_serializer.py:

import json
from pipecat.audio.dtmf.types import KeypadEntry
from pipecat.audio.utils import create_stream_resampler
from pipecat.frames.frames import (AudioRawFrame, EndFrame, InputAudioRawFrame,
                                   InputDTMFFrame, InterruptionFrame, StartFrame)
from pipecat.serializers.base_serializer import FrameSerializer


class EdesyFrameSerializer(FrameSerializer):
    """Raw L16 PCM over WebSocket, as sent by Edesy Media Streams."""

    def __init__(self, metadata: dict):
        super().__init__(FrameSerializer.InputParams())
        self.wire_rate = int(metadata.get("sampleRate", 16000))
        self.pipeline_rate = 0
        self._in = create_stream_resampler()
        self._out = create_stream_resampler()

    async def setup(self, frame: StartFrame):
        self.pipeline_rate = frame.audio_in_sample_rate

    async def serialize(self, frame):
        if isinstance(frame, InterruptionFrame):
            return json.dumps({"type": "killAudio"})    # barge-in
        if isinstance(frame, EndFrame):
            return json.dumps({"type": "disconnect"})   # hang up
        if isinstance(frame, AudioRawFrame):
            data = frame.audio
            if frame.sample_rate != self.wire_rate:
                data = await self._out.resample(data, frame.sample_rate, self.wire_rate)
            return data or None                         # binary frame
        return None

    async def deserialize(self, data):
        if isinstance(data, (bytes, bytearray)):         # caller audio
            audio = bytes(data)
            if self.pipeline_rate and self.pipeline_rate != self.wire_rate:
                audio = await self._in.resample(audio, self.wire_rate, self.pipeline_rate)
            return InputAudioRawFrame(audio=audio, num_channels=1,
                                      sample_rate=self.pipeline_rate or self.wire_rate)
        msg = json.loads(data)
        if msg.get("event") == "dtmf":
            return InputDTMFFrame(KeypadEntry(str(msg.get("dtmf"))))
        return None

server.py:

import json, os
import uvicorn
from fastapi import FastAPI, WebSocket
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams, FastAPIWebsocketTransport

from edesy_serializer import EdesyFrameSerializer

app = FastAPI()

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    metadata = json.loads(await websocket.receive_text())   # 1. metadata frame
    rate = int(metadata.get("sampleRate", 16000))
    custom = json.loads((metadata.get("customerData") or {}).get("custom_parameters") or "{}")

    transport = FastAPIWebsocketTransport(
        websocket=websocket,
        params=FastAPIWebsocketParams(
            audio_in_enabled=True, audio_out_enabled=True, add_wav_header=False,
            vad_analyzer=SileroVADAnalyzer(),
            serializer=EdesyFrameSerializer(metadata),
        ),
    )
    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
    llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o-mini")
    tts = CartesiaTTSService(api_key=os.environ["CARTESIA_API_KEY"],
                             voice_id=os.environ["CARTESIA_VOICE_ID"])

    context = OpenAILLMContext([{"role": "system", "content":
        "You answer phone calls. Keep replies to one or two short sentences."}])
    agg = llm.create_context_aggregator(context)

    task = PipelineTask(
        Pipeline([transport.input(), stt, agg.user(), llm, tts, transport.output(), agg.assistant()]),
        params=PipelineParams(audio_in_sample_rate=rate, audio_out_sample_rate=rate,
                              allow_interruptions=True),
    )

    @transport.event_handler("on_client_connected")
    async def greet(_t, _c):
        await task.queue_frames([LLMRunFrame()])   # speak first, no dead air

    @transport.event_handler("on_client_disconnected")
    async def bye(_t, _c):
        await task.cancel()

    await PipelineRunner(handle_sigint=False).run(task)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "7860")))

Run it behind TLS (during development, a tunnel that gives you a wss:// URL works), run POST/v1/stream-test against it, then point a number at it or place a call with stream_url.

Tips#

• Use the sampleRate from the metadata frame. A mismatch sounds like chipmunks or slow motion.
• Send killAudio as soon as the caller starts talking, or your bot talks over them.
• Speak first. The caller hears silence until your bot sends audio.
• You pay your own STT, LLM and TTS vendors; Edesy charges only call minutes. See Pricing and Limits.

Prefer not to handle audio at all? Use Conversation Relay.