diff --git a/backend/fastrtc/__init__.py b/backend/fastrtc/__init__.py
index 20e4be54..a93fb18d 100644
--- a/backend/fastrtc/__init__.py
+++ b/backend/fastrtc/__init__.py
@@ -18,6 +18,7 @@
from .speech_to_text import MoonshineSTT, get_stt_model
from .stream import Stream, UIArgs
from .text_to_speech import (
+ CambTTSOptions,
CartesiaTTSOptions,
KokoroTTSOptions,
get_tts_model,
@@ -92,6 +93,7 @@
"VideoStreamHandler",
"CloseStream",
"get_current_context",
+ "CambTTSOptions",
"CartesiaTTSOptions",
"WebRTCData",
]
diff --git a/backend/fastrtc/text_to_speech/__init__.py b/backend/fastrtc/text_to_speech/__init__.py
index 0d55538f..6a6b7d28 100644
--- a/backend/fastrtc/text_to_speech/__init__.py
+++ b/backend/fastrtc/text_to_speech/__init__.py
@@ -1,7 +1,8 @@
from .tts import (
+ CambTTSOptions,
CartesiaTTSOptions,
KokoroTTSOptions,
get_tts_model,
)
-__all__ = ["get_tts_model", "KokoroTTSOptions", "CartesiaTTSOptions"]
+__all__ = ["get_tts_model", "KokoroTTSOptions", "CartesiaTTSOptions", "CambTTSOptions"]
diff --git a/backend/fastrtc/text_to_speech/tts.py b/backend/fastrtc/text_to_speech/tts.py
index dde1385f..6f4ba20b 100644
--- a/backend/fastrtc/text_to_speech/tts.py
+++ b/backend/fastrtc/text_to_speech/tts.py
@@ -43,7 +43,7 @@ class KokoroTTSOptions(TTSOptions):
@lru_cache
def get_tts_model(
- model: Literal["kokoro", "cartesia"] = "kokoro", **kwargs
+ model: Literal["kokoro", "cartesia", "camb"] = "kokoro", **kwargs
) -> TTSModel:
if model == "kokoro":
m = KokoroTTSModel()
@@ -52,6 +52,9 @@ def get_tts_model(
elif model == "cartesia":
m = CartesiaTTSModel(api_key=kwargs.get("cartesia_api_key", ""))
return m
+ elif model == "camb":
+ m = CambTTSModel(api_key=kwargs.get("camb_api_key", ""))
+ return m
else:
raise ValueError(f"Invalid model: {model}")
@@ -162,6 +165,85 @@ class CartesiaTTSOptions(TTSOptions):
sample_rate: int = 22_050
+@dataclass
+class CambTTSOptions(TTSOptions):
+ voice_id: int = 2681
+ language: str = "en-us"
+ model: str = "mars-flash"
+ speed: float = 1.0
+ output_format: str = "pcm_s16le"
+ user_instructions: str | None = None
+
+
+class CambTTSModel(TTSModel):
+ def __init__(self, api_key: str):
+ if importlib.util.find_spec("camb") is None:
+ raise RuntimeError(
+ "camb is not installed. Please install it using 'pip install camb'."
+ )
+ self._api_key = api_key
+
+ def _build_tts_kwargs(self, text: str, options: CambTTSOptions):
+ kwargs = {
+ "text": text,
+ "language": options.language,
+ "voice_id": options.voice_id,
+ "speech_model": options.model,
+ "output_configuration": {"format": options.output_format},
+ "voice_settings": {"speed": options.speed},
+ }
+ if options.model == "mars-instruct" and options.user_instructions:
+ kwargs["user_instructions"] = options.user_instructions
+ return kwargs
+
+ async def stream_tts(
+ self, text: str, options: CambTTSOptions | None = None
+ ) -> AsyncGenerator[tuple[int, NDArray[np.int16]], None]:
+ from camb.client import AsyncCambAI
+
+ options = options or CambTTSOptions()
+ client = AsyncCambAI(api_key=self._api_key)
+
+ sentences = re.split(r"(?<=[.!?])\s+", text.strip())
+
+ for sentence in sentences:
+ if not sentence.strip():
+ continue
+ async for output in async_aggregate_bytes_to_16bit(
+ client.text_to_speech.tts(**self._build_tts_kwargs(sentence, options))
+ ):
+ yield 24000, output.flatten()
+
+ def stream_tts_sync(
+ self, text: str, options: CambTTSOptions | None = None
+ ) -> Generator[tuple[int, NDArray[np.int16]], None, None]:
+ loop = asyncio.new_event_loop()
+
+ iterator = self.stream_tts(text, options).__aiter__()
+ while True:
+ try:
+ yield loop.run_until_complete(iterator.__anext__())
+ except StopAsyncIteration:
+ break
+
+ def tts(
+ self, text: str, options: CambTTSOptions | None = None
+ ) -> tuple[int, NDArray[np.int16]]:
+ loop = asyncio.new_event_loop()
+ buffer = np.array([], dtype=np.int16)
+
+ options = options or CambTTSOptions()
+
+ iterator = self.stream_tts(text, options).__aiter__()
+ while True:
+ try:
+ _, chunk = loop.run_until_complete(iterator.__anext__())
+ buffer = np.concatenate([buffer, chunk])
+ except StopAsyncIteration:
+ break
+ return 24000, buffer
+
+
class CartesiaTTSModel(TTSModel):
def __init__(self, api_key: str):
if importlib.util.find_spec("cartesia") is None:
diff --git a/demo/camb_voice_agent/app.py b/demo/camb_voice_agent/app.py
new file mode 100644
index 00000000..8fcb1537
--- /dev/null
+++ b/demo/camb_voice_agent/app.py
@@ -0,0 +1,114 @@
+import json
+import os
+import time
+from pathlib import Path
+
+import gradio as gr
+import numpy as np
+from dotenv import load_dotenv
+from fastapi import FastAPI
+from fastapi.responses import HTMLResponse, StreamingResponse
+from fastrtc import (
+ AdditionalOutputs,
+ CambTTSOptions,
+ ReplyOnPause,
+ Stream,
+)
+from fastrtc.text_to_speech.tts import CambTTSModel
+from fastrtc.utils import audio_to_bytes
+from openai import OpenAI
+from pydantic import BaseModel
+
+load_dotenv()
+
+openai_client = OpenAI()
+tts_model = CambTTSModel(api_key=os.environ["CAMB_API_KEY"])
+tts_options = CambTTSOptions(voice_id=int(os.environ.get("CAMB_VOICE_ID", "156549")))
+
+curr_dir = Path(__file__).parent
+
+
+def response(
+ audio: tuple[int, np.ndarray],
+ chatbot: list[dict] | None = None,
+):
+ chatbot = chatbot or []
+ messages = [{"role": d["role"], "content": d["content"]} for d in chatbot]
+
+ prompt = openai_client.audio.transcriptions.create(
+ file=("audio-file.mp3", audio_to_bytes(audio)),
+ model="whisper-1",
+ ).text
+ chatbot.append({"role": "user", "content": prompt})
+ yield AdditionalOutputs(chatbot)
+
+ messages.append({"role": "user", "content": prompt})
+ llm_response = openai_client.chat.completions.create(
+ model="gpt-4o-mini",
+ max_tokens=512,
+ messages=messages,
+ )
+ response_text = llm_response.choices[0].message.content or ""
+ chatbot.append({"role": "assistant", "content": response_text})
+
+ start = time.time()
+ print("starting tts", start)
+ for i, chunk in enumerate(tts_model.stream_tts_sync(response_text, tts_options)):
+ print("chunk", i, time.time() - start)
+ yield chunk
+ print("finished tts", time.time() - start)
+ yield AdditionalOutputs(chatbot)
+
+
+chatbot = gr.Chatbot(type="messages")
+stream = Stream(
+ modality="audio",
+ mode="send-receive",
+ handler=ReplyOnPause(response),
+ additional_outputs_handler=lambda a, b: b,
+ additional_inputs=[chatbot],
+ additional_outputs=[chatbot],
+)
+
+
+class Message(BaseModel):
+ role: str
+ content: str
+
+
+class InputData(BaseModel):
+ webrtc_id: str
+ chatbot: list[Message]
+
+
+app = FastAPI()
+stream.mount(app)
+
+
+@app.get("/")
+async def _():
+ html_content = (curr_dir / "index.html").read_text()
+ html_content = html_content.replace("__RTC_CONFIGURATION__", json.dumps(None))
+ return HTMLResponse(content=html_content, status_code=200)
+
+
+@app.post("/input_hook")
+async def _(body: InputData):
+ stream.set_input(body.webrtc_id, body.model_dump()["chatbot"])
+ return {"status": "ok"}
+
+
+@app.get("/outputs")
+def _(webrtc_id: str):
+ async def output_stream():
+ async for output in stream.output_stream(webrtc_id):
+ chatbot = output.args[0]
+ yield f"event: output\ndata: {json.dumps(chatbot[-1])}\n\n"
+
+ return StreamingResponse(output_stream(), media_type="text/event-stream")
+
+
+if __name__ == "__main__":
+ import uvicorn
+
+ uvicorn.run(app, host="0.0.0.0", port=7860)
diff --git a/demo/camb_voice_agent/index.html b/demo/camb_voice_agent/index.html
new file mode 100644
index 00000000..26b86e0c
--- /dev/null
+++ b/demo/camb_voice_agent/index.html
@@ -0,0 +1,493 @@
+
+
+
+
+
+
+ CAMB AI Voice Agent
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/test/test_camb_tts.py b/test/test_camb_tts.py
new file mode 100644
index 00000000..7e5b0fc0
--- /dev/null
+++ b/test/test_camb_tts.py
@@ -0,0 +1,286 @@
+"""
+CAMB TTS Integration Tests
+
+Run directly: .venv/bin/python test/test_camb_tts.py
+"""
+
+import asyncio
+import os
+import subprocess
+import sys
+import tempfile
+import wave
+
+import numpy as np
+from dotenv import load_dotenv
+
+# Load API key from demo env
+load_dotenv(
+ os.path.join(os.path.dirname(__file__), "..", "demo", "camb_voice_agent", ".env")
+)
+
+from fastrtc.text_to_speech.tts import CambTTSModel, CambTTSOptions, get_tts_model
+
+VOICE_ID = 156549
+SAMPLE_RATE = 24000
+results: list[tuple[str, bool, str]] = []
+
+
+def play_audio(audio: np.ndarray, sample_rate: int = SAMPLE_RATE):
+ """Write int16 PCM to a temp WAV and play with afplay."""
+ with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
+ path = f.name
+ with wave.open(f, "wb") as wf:
+ wf.setnchannels(1)
+ wf.setsampwidth(2)
+ wf.setframerate(sample_rate)
+ wf.writeframes(audio.tobytes())
+ print(f" Playing audio ({len(audio)} samples, {len(audio)/sample_rate:.2f}s)...")
+ subprocess.run(["afplay", path], check=True)
+ os.unlink(path)
+
+
+def run_test(name: str, fn):
+ print(f"\n{'='*60}")
+ print(f"TEST: {name}")
+ print("=" * 60)
+ try:
+ fn()
+ results.append((name, True, ""))
+ print(f" PASS")
+ except Exception as e:
+ results.append((name, False, str(e)))
+ print(f" FAIL: {e}")
+
+
+def make_model() -> CambTTSModel:
+ api_key = os.environ.get("CAMB_API_KEY", "")
+ assert api_key, "CAMB_API_KEY not set"
+ return CambTTSModel(api_key=api_key)
+
+
+# ---------------------------------------------------------------------------
+# Synthesis tests
+# ---------------------------------------------------------------------------
+
+
+def test_tts_basic():
+ """tts() — basic synthesis: short text, verify returns (24000, int16 ndarray)."""
+ model = make_model()
+ options = CambTTSOptions(voice_id=VOICE_ID)
+ sr, audio = model.tts("Hello, this is a test of CAMB text to speech.", options)
+ assert sr == SAMPLE_RATE, f"Expected sample rate {SAMPLE_RATE}, got {sr}"
+ assert isinstance(audio, np.ndarray), f"Expected ndarray, got {type(audio)}"
+ assert audio.dtype == np.int16, f"Expected int16, got {audio.dtype}"
+ assert len(audio) > 0, "Audio is empty"
+ print(f" Sample rate: {sr}, shape: {audio.shape}, dtype: {audio.dtype}")
+ play_audio(audio, sr)
+
+
+def test_tts_multi_sentence():
+ """tts() — multi-sentence: verify correct concatenation."""
+ model = make_model()
+ options = CambTTSOptions(voice_id=VOICE_ID)
+ text = (
+ "This is the first sentence. Here comes the second one! "
+ "And finally, the third sentence?"
+ )
+ sr, audio = model.tts(text, options)
+ assert sr == SAMPLE_RATE
+ assert audio.dtype == np.int16
+ assert len(audio) > 0
+ print(f" Multi-sentence audio: {audio.shape}, {len(audio)/sr:.2f}s")
+ play_audio(audio, sr)
+
+
+def test_stream_tts_sync():
+ """stream_tts_sync() — streaming: collect chunks, verify, play."""
+ model = make_model()
+ options = CambTTSOptions(voice_id=VOICE_ID)
+ chunks = []
+ for i, (sr, chunk) in enumerate(
+ model.stream_tts_sync("Streaming is working correctly.", options)
+ ):
+ assert sr == SAMPLE_RATE
+ assert chunk.dtype == np.int16
+ chunks.append(chunk)
+ print(f" Chunk {i}: shape={chunk.shape}")
+ assert len(chunks) > 0, "No chunks received"
+ combined = np.concatenate(chunks)
+ print(f" Total: {len(combined)} samples, {len(combined)/SAMPLE_RATE:.2f}s")
+ play_audio(combined)
+
+
+def test_stream_tts_sync_long():
+ """stream_tts_sync() — long prompt to stress-test streaming."""
+ model = make_model()
+ options = CambTTSOptions(voice_id=VOICE_ID)
+ text = (
+ "It may be that this communication will be considered as something unusual, "
+ "but at any rate it must be admitted that in its clearness and frankness it "
+ "left nothing to be desired. The serious part of it was that the government "
+ "had undertaken to treat the situation with the utmost care. Opinions on the "
+ "matter were many and varied."
+ )
+ chunks = []
+ for i, (sr, chunk) in enumerate(model.stream_tts_sync(text, options)):
+ chunks.append(chunk)
+ print(f" Chunk {i}: shape={chunk.shape}")
+ combined = np.concatenate(chunks)
+ assert len(combined) > 0
+ print(f" Total: {len(combined)} samples, {len(combined)/SAMPLE_RATE:.2f}s")
+ play_audio(combined)
+
+
+def test_stream_tts_async():
+ """stream_tts (async) — verify async generator works via asyncio.run()."""
+ model = make_model()
+ options = CambTTSOptions(voice_id=VOICE_ID)
+
+ async def collect():
+ chunks = []
+ i = 0
+ async for sr, chunk in model.stream_tts("Async streaming test.", options):
+ assert sr == SAMPLE_RATE
+ assert chunk.dtype == np.int16
+ chunks.append(chunk)
+ print(f" Async chunk {i}: shape={chunk.shape}")
+ i += 1
+ return chunks
+
+ chunks = asyncio.run(collect())
+ assert len(chunks) > 0, "No async chunks received"
+ combined = np.concatenate(chunks)
+ print(f" Total: {len(combined)} samples, {len(combined)/SAMPLE_RATE:.2f}s")
+ play_audio(combined)
+
+
+# ---------------------------------------------------------------------------
+# Unit tests
+# ---------------------------------------------------------------------------
+
+
+def test_build_tts_kwargs_defaults():
+ """_build_tts_kwargs() — default options produce correct kwargs."""
+ model = make_model()
+ options = CambTTSOptions()
+ kwargs = model._build_tts_kwargs("test text", options)
+ assert kwargs["text"] == "test text"
+ assert kwargs["language"] == "en-us"
+ assert kwargs["voice_id"] == 2681
+ assert kwargs["speech_model"] == "mars-flash"
+ assert kwargs["output_configuration"] == {"format": "pcm_s16le"}
+ assert kwargs["voice_settings"] == {"speed": 1.0}
+ assert "user_instructions" not in kwargs
+ print(f" kwargs: {kwargs}")
+
+
+def test_build_tts_kwargs_custom():
+ """_build_tts_kwargs() — custom options."""
+ model = make_model()
+ options = CambTTSOptions(voice_id=VOICE_ID, language="en-gb", speed=1.5)
+ kwargs = model._build_tts_kwargs("hello", options)
+ assert kwargs["voice_id"] == VOICE_ID
+ assert kwargs["language"] == "en-gb"
+ assert kwargs["voice_settings"] == {"speed": 1.5}
+ print(f" kwargs: {kwargs}")
+
+
+def test_build_tts_kwargs_custom_format():
+ """_build_tts_kwargs() — custom output format."""
+ model = make_model()
+ options = CambTTSOptions(output_format="wav")
+ kwargs = model._build_tts_kwargs("hello", options)
+ assert kwargs["output_configuration"] == {"format": "wav"}
+ print(f" kwargs: {kwargs}")
+
+
+def test_build_tts_kwargs_mars_instruct():
+ """_build_tts_kwargs() — mars-instruct with user_instructions."""
+ model = make_model()
+ options = CambTTSOptions(
+ model="mars-instruct", user_instructions="Speak slowly and clearly."
+ )
+ kwargs = model._build_tts_kwargs("hello", options)
+ assert kwargs["speech_model"] == "mars-instruct"
+ assert kwargs["user_instructions"] == "Speak slowly and clearly."
+ print(f" kwargs: {kwargs}")
+
+ # Also verify user_instructions is NOT included for non-instruct models
+ options_flash = CambTTSOptions(
+ model="mars-flash", user_instructions="Should be ignored."
+ )
+ kwargs_flash = model._build_tts_kwargs("hello", options_flash)
+ assert "user_instructions" not in kwargs_flash
+ print(f" mars-flash kwargs (no user_instructions): {kwargs_flash}")
+
+
+def test_camb_tts_options_defaults():
+ """CambTTSOptions — verify default values."""
+ opts = CambTTSOptions()
+ assert opts.voice_id == 2681
+ assert opts.language == "en-us"
+ assert opts.model == "mars-flash"
+ assert opts.speed == 1.0
+ assert opts.output_format == "pcm_s16le"
+ assert opts.user_instructions is None
+ print(f" Defaults: voice_id={opts.voice_id}, language={opts.language}, "
+ f"model={opts.model}, speed={opts.speed}, "
+ f"output_format={opts.output_format}, "
+ f"user_instructions={opts.user_instructions}")
+
+
+def test_custom_options():
+ """Custom options — non-default voice_id, language, speed, format."""
+ opts = CambTTSOptions(
+ voice_id=VOICE_ID, language="en-gb", speed=0.8, output_format="wav"
+ )
+ assert opts.voice_id == VOICE_ID
+ assert opts.language == "en-gb"
+ assert opts.speed == 0.8
+ assert opts.output_format == "wav"
+ print(f" Custom: voice_id={opts.voice_id}, language={opts.language}, "
+ f"speed={opts.speed}, output_format={opts.output_format}")
+
+
+def test_get_tts_model_camb():
+ """get_tts_model('camb') — verify factory function works."""
+ api_key = os.environ.get("CAMB_API_KEY", "")
+ model = get_tts_model("camb", camb_api_key=api_key)
+ assert isinstance(model, CambTTSModel)
+ print(f" Factory returned: {type(model).__name__}")
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+ tests = [
+ ("tts() basic synthesis", test_tts_basic),
+ ("tts() multi-sentence", test_tts_multi_sentence),
+ ("stream_tts_sync() streaming", test_stream_tts_sync),
+ ("stream_tts_sync() long prompt", test_stream_tts_sync_long),
+ ("stream_tts async", test_stream_tts_async),
+ ("_build_tts_kwargs() defaults", test_build_tts_kwargs_defaults),
+ ("_build_tts_kwargs() custom", test_build_tts_kwargs_custom),
+ ("_build_tts_kwargs() custom format", test_build_tts_kwargs_custom_format),
+ ("_build_tts_kwargs() mars-instruct", test_build_tts_kwargs_mars_instruct),
+ ("CambTTSOptions defaults", test_camb_tts_options_defaults),
+ ("Custom options", test_custom_options),
+ ("get_tts_model('camb') factory", test_get_tts_model_camb),
+ ]
+
+ for name, fn in tests:
+ run_test(name, fn)
+
+ # Summary
+ print(f"\n{'='*60}")
+ print("SUMMARY")
+ print("=" * 60)
+ passed = sum(1 for _, ok, _ in results if ok)
+ failed = sum(1 for _, ok, _ in results if not ok)
+ for name, ok, err in results:
+ print(f" {'[PASS]' if ok else '[FAIL]'} {name}")
+ print(f"\n {passed} passed, {failed} failed, {len(results)} total")
+ sys.exit(1 if failed else 0)