feat(deepgram): migrate TTS to Flux /v2/speak - #633
Conversation
Flux TTS is now the only speak path (/v2/speak, flux-haley-en). The SDK 7.7 bump required for v2 also delivers typed listen TurnInfo, so STT accepts those objects or the agent never hears the user. Co-authored-by: Cursor <cursoragent@cursor.com>
📝 WalkthroughWalkthroughDeepgram TTS migrated from the v1 Aura API to the v2 Flux API. The migration adds Merge Risk: ⚪ Minimal · up to The PR changes Deepgram TTS/STT integration and updates related documentation and tests, with only localized follow-up items for documentation, annotations, and test assertions; no actionable merge-blocking risk remains. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (1)
plugins/deepgram/tests/test_tts.py (1)
32-38: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftAssert interruption behavior instead of private state.
Lines 34 and 38 bind the test to
_socketidentity. Lines 62 and 63 callstop_audio()after synthesis completes and only inspect_stop_event. Remove the private connection assertion. Test an active synthesis interruption and a successful subsequent synthesis through the public interface.As per coding guidelines: “ALWAYS test behavior, not calling a path.” As per path instructions: “Assert behavior and outputs, not initialization or call paths.”
Also applies to: 58-63
Sources: Coding guidelines, Path instructions
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 617deac9-d605-44ab-94bd-9f2842ed0acf
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (9)
CHANGELOG.mdplugins/deepgram/README.mdplugins/deepgram/example/README.mdplugins/deepgram/example/deepgram_tts_example.pyplugins/deepgram/pyproject.tomlplugins/deepgram/tests/test_stt.pyplugins/deepgram/tests/test_tts.pyplugins/deepgram/vision_agents/plugins/deepgram/deepgram_stt.pyplugins/deepgram/vision_agents/plugins/deepgram/tts.py
Included review availability: Your plan includes up to 8 reviews per rolling hour; 7 remain after this review.
| ### `deepgram` plugin: TTS defaults to Flux (`/v2/speak`) | ||
|
|
||
| `deepgram.TTS` now streams Flux TTS on `wss://api.deepgram.com/v2/speak` and defaults to `flux-haley-en`. Aura model strings (`aura-*`) are rejected with `ValueError`. Call sites that passed an Aura voice must switch to a Flux model (`flux-{voice}-en`). See the [Flux voice catalog](https://developers.deepgram.com/docs/flux-tts/voices). | ||
|
|
||
| ## New Features | ||
|
|
||
| ### `deepgram` plugin: Flux TTS streaming, `speed`, and Interrupt barge-in | ||
|
|
||
| Deepgram TTS uses the Flux turn protocol (`Speak` / `Flush` / `SpeechMetadata`) with a persistent websocket. Pass optional `speed` (0.85–1.15 in 0.05 steps) on the constructor. Barge-in sends `Interrupt` instead of Aura's `Clear`. Supported sample rates now include 32000 and 44100. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Add the PR reference to both Deepgram entries.
Add (#633) to the headings on Lines 5 and 50.
As per coding guidelines: “Reference PR numbers inline.”
Also applies to: 50-52
Source: Coding guidelines
| model="aura-2-thalia-en", # Default voice | ||
| model="flux-haley-en", # Default voice | ||
| sample_rate=16000, # Audio sample rate | ||
| speed=1.0, # Optional speech-rate multiplier (0.85–1.15) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Document the accepted speed increments.
Line 41 documents only the range. TTS.__init__ also requires 0.05 steps. State 0.85–1.15 in 0.05 steps.
| async def tts(self) -> deepgram.TTS: | ||
| return deepgram.TTS() | ||
| t = deepgram.TTS() | ||
| yield t | ||
| await t.close() |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -U \
'async def tts\([^)]*\)\s*->\s*deepgram\.TTS:\n\s*t = deepgram\.TTS\(\)\n\s*yield t' \
plugins/deepgram/tests/test_tts.pyRepository: GetStream/Vision-Agents
Length of output: 254
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- test_tts.py outline ---'
ast-grep outline plugins/deepgram/tests/test_tts.py
printf '%s\n' '--- test_tts.py ---'
cat -n plugins/deepgram/tests/test_tts.py
printf '%s\n' '--- related fixture annotations and imports ---'
rg -n -U '(^from typing import.*|^import typing.*|`@pytest.fixture`|async def (tts|stt)\b|AsyncIterator|AsyncGenerator)' \
plugins/deepgram/tests/test_tts.py plugins/deepgram/tests/test_stt.py pyproject.tomlRepository: GetStream/Vision-Agents
Length of output: 3766
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import ast
from pathlib import Path
path = Path("plugins/deepgram/tests/test_tts.py")
tree = ast.parse(path.read_text())
fixture = next(
node
for node in ast.walk(tree)
if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)) and node.name == "tts"
)
has_yield = any(isinstance(node, (ast.Yield, ast.YieldFrom)) for node in ast.walk(fixture))
annotation = ast.unparse(fixture.returns) if fixture.returns else None
print(f"fixture={fixture.name}")
print(f"async_function={isinstance(fixture, ast.AsyncFunctionDef)}")
print(f"contains_yield={has_yield}")
print(f"return_annotation={annotation!r}")
print(f"required_shape={'AsyncIterator/AsyncGenerator' if has_yield else 'coroutine'}")
PY
printf '%s\n' '--- test_stt.py relevant context ---'
cat -n plugins/deepgram/tests/test_stt.py | sed -n '1,35p'Repository: GetStream/Vision-Agents
Length of output: 1955
Correct the async fixture return type.
tts is an async generator. Annotate it as AsyncIterator[deepgram.TTS] or AsyncGenerator[deepgram.TTS, None], and import the annotation at module scope.
| async def test_deepgram_tts_convert_text_to_audio(self, tts: deepgram.TTS): | ||
| text = "Hello from Deepgram." | ||
|
|
||
| out = [] | ||
| async for item in tts.send_iter(text): | ||
| out.append(item) | ||
| out = [item async for item in tts.send_iter("Hello from Deepgram.")] | ||
|
|
||
| assert len(out) > 0 | ||
| # First chunk must have some audio | ||
| assert out[0].data | ||
| # Last chunk must be marked as "final" | ||
| assert out[-1].final | ||
|
|
||
| async def test_connection_reused_across_calls(self, tts: deepgram.TTS): | ||
| _ = [item async for item in tts.send_iter("Hello")] | ||
| socket = tts._socket | ||
| assert socket is not None | ||
|
|
||
| class TestDeepgramTTS: | ||
| """Unit tests for the websocket-based TTS implementation.""" | ||
|
|
||
| async def test_stream_audio_yields_chunks(self): | ||
| mock_socket = _make_mock_socket( | ||
| [ | ||
| _pcm_bytes(800), | ||
| _pcm_bytes(800), | ||
| SpeakV1Flushed(type="Flushed", sequence_id=1), | ||
| ] | ||
| ) | ||
| mock_ctx = _MockConnectCtx(mock_socket) | ||
|
|
||
| tts = deepgram.TTS(client=MagicMock()) | ||
| tts.client.speak.v1.connect.return_value = mock_ctx | ||
|
|
||
| result = await tts.stream_audio("Hello") | ||
| chunks = [chunk async for chunk in result] | ||
|
|
||
| assert len(chunks) == 2 | ||
| assert all(isinstance(c, PcmData) for c in chunks) | ||
| mock_socket.send_text.assert_called_once() | ||
| mock_socket.send_flush.assert_called_once() | ||
|
|
||
| async def test_stream_audio_ignores_cleared(self): | ||
| mock_socket = _make_mock_socket( | ||
| [ | ||
| _pcm_bytes(800), | ||
| SpeakV1Cleared(type="Cleared", sequence_id=1), | ||
| _pcm_bytes(800), | ||
| SpeakV1Flushed(type="Flushed", sequence_id=2), | ||
| ] | ||
| ) | ||
| mock_ctx = _MockConnectCtx(mock_socket) | ||
|
|
||
| tts = deepgram.TTS(client=MagicMock()) | ||
| tts.client.speak.v1.connect.return_value = mock_ctx | ||
|
|
||
| result = await tts.stream_audio("Hello") | ||
| chunks = [chunk async for chunk in result] | ||
|
|
||
| assert len(chunks) == 2 | ||
|
|
||
| async def test_stale_cleared_before_audio(self): | ||
| mock_socket = _make_mock_socket( | ||
| [ | ||
| SpeakV1Cleared(type="Cleared", sequence_id=0), | ||
| _pcm_bytes(800), | ||
| SpeakV1Flushed(type="Flushed", sequence_id=1), | ||
| ] | ||
| ) | ||
| mock_ctx = _MockConnectCtx(mock_socket) | ||
|
|
||
| tts = deepgram.TTS(client=MagicMock()) | ||
| tts.client.speak.v1.connect.return_value = mock_ctx | ||
|
|
||
| result = await tts.stream_audio("Hello") | ||
| chunks = [chunk async for chunk in result] | ||
|
|
||
| assert len(chunks) == 1 | ||
| assert chunks[0].samples.size == 800 | ||
|
|
||
| async def test_stream_audio_skips_warnings(self): | ||
| mock_socket = _make_mock_socket( | ||
| [ | ||
| _pcm_bytes(800), | ||
| SpeakV1Warning(type="Warning", description="test", code="W001"), | ||
| _pcm_bytes(800), | ||
| SpeakV1Flushed(type="Flushed", sequence_id=1), | ||
| ] | ||
| ) | ||
| mock_ctx = _MockConnectCtx(mock_socket) | ||
|
|
||
| tts = deepgram.TTS(client=MagicMock()) | ||
| tts.client.speak.v1.connect.return_value = mock_ctx | ||
|
|
||
| result = await tts.stream_audio("Hello") | ||
| chunks = [chunk async for chunk in result] | ||
|
|
||
| assert len(chunks) == 2 | ||
|
|
||
| async def test_stop_audio_sends_clear(self): | ||
| mock_socket = _make_mock_socket([]) | ||
| tts = deepgram.TTS(client=MagicMock()) | ||
| tts._socket = mock_socket | ||
| _ = [item async for item in tts.send_iter("World")] | ||
| assert tts._socket is socket | ||
|
|
||
| await tts.stop_audio() | ||
| async def test_speed_forwarded_produces_audio(self): | ||
| tts = deepgram.TTS(speed=1.05) | ||
| try: | ||
| out = [item async for item in tts.send_iter("Hello there.")] | ||
| finally: | ||
| await tts.close() | ||
|
|
||
| mock_socket.send_clear.assert_called_once() | ||
| assert tts._stop_event.is_set() | ||
| assert any(item.data for item in out) | ||
|
|
||
| async def test_stop_audio_noop_when_not_connected(self): | ||
| tts = deepgram.TTS(client=MagicMock()) | ||
| await tts.stop_audio() | ||
|
|
||
| async def test_stop_event_terminates_receive(self): | ||
| socket = AsyncMock() | ||
| socket.send_text = AsyncMock() | ||
| socket.send_flush = AsyncMock() | ||
|
|
||
| async def _aiter(self_): | ||
| yield _pcm_bytes(800) | ||
| await asyncio.sleep(10) | ||
|
|
||
| socket.__aiter__ = _aiter | ||
| mock_ctx = _MockConnectCtx(socket) | ||
|
|
||
| tts = deepgram.TTS(client=MagicMock()) | ||
| tts.client.speak.v1.connect.return_value = mock_ctx | ||
|
|
||
| result = await tts.stream_audio("Hello") | ||
| tts._stop_event.set() | ||
|
|
||
| chunks = [chunk async for chunk in result] | ||
| assert len(chunks) == 0 | ||
| async def test_extended_sample_rate_produces_audio(self): | ||
| tts = deepgram.TTS(sample_rate=24000) | ||
| try: | ||
| out = [item async for item in tts.send_iter("Hello there.")] | ||
| finally: | ||
| await tts.close() | ||
|
|
||
| async def test_close_tears_down_connection(self): | ||
| mock_socket = _make_mock_socket([]) | ||
| assert any(item.data for item in out) | ||
|
|
||
| tts = deepgram.TTS(client=MagicMock()) | ||
| tts._socket = mock_socket | ||
| async def test_stop_audio_after_synthesis(self, tts: deepgram.TTS): | ||
| out = [item async for item in tts.send_iter("Hello there.")] | ||
| assert out | ||
|
|
||
| await tts.close() | ||
|
|
||
| mock_socket.send_close.assert_called_once() | ||
| assert tts._socket is None | ||
|
|
||
| async def test_connection_reused_across_calls(self): | ||
| call_count = 0 | ||
| batches = [ | ||
| [_pcm_bytes(800), SpeakV1Flushed(type="Flushed", sequence_id=1)], | ||
| [_pcm_bytes(800), SpeakV1Flushed(type="Flushed", sequence_id=2)], | ||
| ] | ||
|
|
||
| socket = AsyncMock() | ||
| socket.send_text = AsyncMock() | ||
| socket.send_flush = AsyncMock() | ||
|
|
||
| async def _aiter(self_): | ||
| nonlocal call_count | ||
| for msg in batches[call_count]: | ||
| yield msg | ||
| call_count += 1 | ||
| await tts.stop_audio() | ||
| assert tts._stop_event.is_set() |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -e
printf '%s\n' '--- test_tts.py ---'
cat -n plugins/deepgram/tests/test_tts.py | sed -n '1,90p'
printf '%s\n' '--- test_stt.py ---'
cat -n plugins/deepgram/tests/test_stt.py | sed -n '1,140p'
printf '%s\n' '--- annotation patterns in neighboring tests ---'
rg -n --glob '*.py' 'async def test_.*\(.*\).*-> None|participant: Participant|from .*Participant' plugins/deepgram/tests plugins 2>/dev/null | head -120
printf '%s\n' '--- project typing configuration ---'
rg -n '(^|#).*(type.?annot|mypy|pyright)|disallow_untyped|check_untyped' pyproject.toml setup.cfg tox.ini . 2>/dev/null | head -120Repository: GetStream/Vision-Agents
Length of output: 31026
🏁 Script executed:
#!/bin/bash
set -e
printf '%s\n' '--- changed hunks ---'
git diff --unified=0 -- plugins/deepgram/tests/test_tts.py plugins/deepgram/tests/test_stt.py
printf '%s\n' '--- unannotated test functions and parameters ---'
python3 - <<'PY'
import ast
from pathlib import Path
for name in (
"plugins/deepgram/tests/test_tts.py",
"plugins/deepgram/tests/test_stt.py",
):
tree = ast.parse(Path(name).read_text(), filename=name)
print(name)
for node in ast.walk(tree):
if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)) and node.name.startswith("test_"):
missing_args = [
arg.arg
for arg in (*node.args.posonlyargs, *node.args.args, *node.args.kwonlyargs)
if arg.arg != "self" and arg.annotation is None
]
missing_return = node.returns is None
if missing_args or missing_return:
print(
f" line {node.lineno}: {node.name} "
f"missing_args={missing_args} missing_return={missing_return}"
)
PY
printf '%s\n' '--- mypy settings ---'
sed -n '136,160p' pyproject.tomlRepository: GetStream/Vision-Agents
Length of output: 2045
Add complete annotations to the new test functions.
Add -> None to the TTS test methods at lines 23–63. Add participant: Participant and -> None to the STT test methods at lines 78–117.
📍 Affects 2 files
plugins/deepgram/tests/test_tts.py#L23-L63(this comment)plugins/deepgram/tests/test_stt.py#L78-L117
Source: Coding guidelines
1 participant
Why
Deepgram Flux TTS is GA on
/v2/speakwith a different protocol than Aura (Speak/Flush/SpeechMetadata,Interruptinstead ofClear, Flux voices only). The plugin still spoke Aura over/v1/speakwithaura-2-thalia-en, so it could not use the new API.speak.v2needsdeepgram-sdk7.7, which also changed listen v2TurnInfofrom dicts to typed objects. Without handling those, Flux STT dropped every transcript and the agent never replied.Changes
deepgram.TTSstreams Flux on/v2/speak, defaults toflux-haley-en, takes optionalspeed, and barges in withInterruptValueErrordeepgram-sdkbumped to>=7.7.0,<7.8.0ListenV2TurnInfoas well as dictsMade with Cursor