FastRTC signaling

Welcome! I'm Param. In this course we are going to build a real-time phone voice agent in a single FastAPI process. No separate media server, no SIP trunk, no magic. Just FastRTC, Whisper, your LLM of choice, and a TTS seam you control. The first step is the least glamorous and the most important: proving the signaling path is wired up correctly.

Voice agents fail silently in two ways. Either the audio never connects, or the audio connects and the loop is too slow. Today we knock out the first failure mode. When the browser opens a connection, we want to see the FastRTC Stream complete the handshake so we know audio frames can flow.

terminal
bash
# Clone the workshop repository
git clone https://github.com/learnwithparam/realtime-phone-agents-fastrtc.git
cd realtime-phone-agents-fastrtc

# One command to set up and run
make dev

Clones the repo, installs dependencies with uv, and starts the FastAPI server with the FastRTC Stream mounted.

main.py
python
from dotenv import load_dotenv
load_dotenv()

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from router import router, mount_webrtc

app = FastAPI(
    title="Realtime Phone Agents with FastRTC",
    description="Build a realtime phone voice agent using FastRTC for low-latency audio",
    version="1.0.0",
)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])
app.include_router(router)

# Mount the FastRTC Stream onto the FastAPI app
mount_webrtc(app)

The app entrypoint does two important things. It registers the REST router for health and text turns, and it calls mount_webrtc, which attaches a FastRTC Stream at /phone/webrtc so the browser can open a WebRTC session against our FastAPI process.

router.py
python
def mount_webrtc(app) -> None:
    """Mount the FastRTC Stream onto the FastAPI app at /phone/webrtc."""
    if not FASTRTC_AVAILABLE:
        logger.info("Skipping FastRTC mount (library not installed)")
        return

    from fastrtc import Stream, ReplyOnPause

    async def _audio_handler(audio_chunk):
        audio_bytes = audio_chunk if isinstance(audio_chunk, (bytes, bytearray)) else bytes(audio_chunk)
        transcript, reply, audio_out = await _agent.handle_turn(audio_bytes)
        logger.info(f"[phone] transcript={transcript!r} reply={reply!r}")
        yield audio_out

    stream = Stream(handler=ReplyOnPause(_audio_handler), modality="audio", mode="send-receive")
    stream.mount(app, path="/phone/webrtc")

FastRTC gives you two primitives: Stream and ReplyOnPause. Stream handles the WebRTC mechanics. ReplyOnPause wraps your handler so it only runs when the caller pauses speaking. Together they give you a usable signaling path in a few lines.

FastRTC signaling handshake

How the browser and FastAPI process negotiate a WebRTC audio session.

Because for a single-caller phone agent, the separate media server is overhead you do not need. One process means one place to look when latency spikes, one deployment, and zero cross-service signaling. When you eventually need multi-party rooms or server-side recording, you graduate to LiveKit. Until then, the single FastAPI process keeps the loop debuggable.

Quiz: Quiz

Loading practice…