diff --git a/CLAUDE.md b/CLAUDE.md index 9a9ab05..264aa08 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -93,22 +93,69 @@ Voices are interleaved for perceptual diversity: American female → British mal ## `speak()` Progress Lifecycle -`speak()` blocks until playback finishes, reporting progress throughout: +`speak()` blocks until playback finishes, reporting progress throughout. Each progress notification includes a `message` field that LLM clients can read to understand the current phase. -| Progress | Phase | -|----------|-------| -| 5% | Entry tone played — audible "I heard you" | -| 30% | Synthesis complete | -| 35% | Enqueued for playback | -| 35-99% | Playing — progress tracks elapsed time vs expected duration | -| 100% | Playback finished | +| Progress | Phase | Message | +|----------|-------|---------| +| 5% | Entry tone played — audible "I heard you" | `Entry tone` | +| 30% | Synthesis complete | `Synthesizing...` | +| 35% | Enqueued for playback | `Playing audio — you can continue working` | +| 35-99% | Playing — progress tracks elapsed time vs expected duration | `Playing audio — you can continue working` | +| 100% | Playback finished | `Playback complete` | MCP-aware clients see a live progress bar. The entry tone fires before synthesis, covering the 1-3s latency gap. +### Concurrency guidance + +The `speak()` call completes when playback finishes (10-60s for typical text), but the audio plays through physical speakers — the return value doesn't feed into subsequent reasoning. LLM clients like Claude Code can call `speak()` alongside other tools in the same message. The server instructions and tool docstring explicitly encourage this. Tool annotations (`openWorldHint=True`, `idempotentHint=True`) provide structured signals to clients that support them. + `speech_status(speech_id)` is still available for checking outcomes after the fact. Possible statuses: `completed`, `playing`, `queued`, `unknown`. Outcomes are stored in a bounded ring (last 100 items) — old entries are evicted automatically. +## Chunked Synthesis (Pipelined Playback) + +Long texts (>= 20 words with sentence boundaries) are automatically split into sentence-sized chunks and pipelined — synthesis of chunk N+1 overlaps playback of chunk N. + +``` +Single-shot: [---12s synthesis---][----------48s playback----------] First audio at T+12s +Chunked: [~1s synth][play 1][~1s synth][play 2][~1s synth][play 3]... First audio at T+1s +``` + +### Why no gaps + +Kokoro synthesizes ~4x realtime on CPU, so synthesis always outpaces playback. `pw-play` adds ~10ms startup per chunk — imperceptible. Slower engines (Orpheus) may have brief gaps but still beat the full-synthesis-first approach. + +### Short text bypass + +Texts under 20 words or without sentence boundaries (`.!?` followed by whitespace) take the single-shot path — zero overhead, identical to pre-chunking behavior. + +### Tone behavior + +- Entry tone plays once at the start (before first chunk synthesis) +- Exit/standby tones are suppressed between chunks (`suppress_exit_tone` flag on `_WorkItem`) +- Final chunk plays the normal exit tone (roger) or standby tone + +### Cancellation in chunked mode + +- **Explicit `cancel_speech(speech_id)`** — cancels the returned speech_id (the final chunk). Already-playing earlier chunks finish naturally. +- **MCP disconnect** — the synthesis loop stops (remaining chunks aren't synthesized). Already-enqueued chunks play through. + +### Progress lifecycle (chunked) + +| Progress | Phase | Message | +|----------|-------|---------| +| 5% | Entry tone played | `Entry tone` | +| 5-30% | Synthesis progress across all chunks | `Synthesizing chunk N/M...` | +| 35% | All chunks enqueued | `Playing audio — you can continue working` | +| 35-99% | Playing — tracks elapsed time vs total duration | `Playing audio — you can continue working` | +| 100% | Final chunk playback finished | `Playback complete` | + +### Files + +- `server.py` — `split_text()`, `_speak_single()`, `_speak_chunked()`, `_await_with_progress()` +- `queue.py` — `suppress_exit_tone` field on `_WorkItem` + ## Cancellation Speech can be cancelled two ways: diff --git a/src/tts_mcp/queue.py b/src/tts_mcp/queue.py index 4ccdf3e..4195659 100644 --- a/src/tts_mcp/queue.py +++ b/src/tts_mcp/queue.py @@ -41,6 +41,7 @@ class _WorkItem: future: asyncio.Future = field(compare=False) speech_id: str = field(default="", compare=False) enqueued_at: float = field(default_factory=time.time, compare=False) + suppress_exit_tone: bool = field(default=False, compare=False) class SpeechQueue: @@ -213,15 +214,17 @@ class SpeechQueue: ) # Play exit tone — "standby" if more queued, "roger" if done - exit_t = ( - self._standby_tone if self._queue.qsize() > 0 - else self._exit_tone - ) - if exit_t: - try: - await play_audio(exit_t, expected_seconds=0.3) - except PlaybackError: - pass # Non-fatal + # suppress_exit_tone skips tones between chunks of the same message + if not item.suppress_exit_tone: + exit_t = ( + self._standby_tone if self._queue.qsize() > 0 + else self._exit_tone + ) + if exit_t: + try: + await play_audio(exit_t, expected_seconds=0.3) + except PlaybackError: + pass # Non-fatal self._resolve_outcome(item, { "played": True, @@ -282,6 +285,7 @@ class SpeechQueue: self, result: TTSResult, priority: Priority = Priority.NORMAL, + suppress_exit_tone: bool = False, ) -> dict: """Enqueue a TTSResult for playback. Returns immediately with enqueue metadata. @@ -308,6 +312,7 @@ class SpeechQueue: result=result, future=future, speech_id=speech_id, + suppress_exit_tone=suppress_exit_tone, ) try: diff --git a/src/tts_mcp/server.py b/src/tts_mcp/server.py index fd66584..7580928 100644 --- a/src/tts_mcp/server.py +++ b/src/tts_mcp/server.py @@ -1,6 +1,7 @@ """FastMCP 3.0 server — tools, lifespan, and resource definitions.""" import asyncio +import re import sys import time from contextlib import asynccontextmanager @@ -9,6 +10,7 @@ from typing import Literal from fastmcp import Context, FastMCP from fastmcp.server.dependencies import CurrentContext +from mcp.types import ToolAnnotations from .audio import play_audio from .engines.base import TTSEngine @@ -115,9 +117,11 @@ async def app_lifespan(server: FastMCP): mcp = FastMCP( "tts-mcp", instructions=( - "Multi-engine text-to-speech server. Use 'speak' to synthesize and enqueue audio " - "for playback through the host speakers — returns immediately with a speech_id. " - "Use 'speech_status' to check if playback completed, is in progress, or still queued. " + "Multi-engine text-to-speech server. Use 'speak' to synthesize and play audio " + "through the host speakers. The call completes when playback finishes, but you " + "can safely call speak alongside other tools — audio plays through physical " + "speakers and doesn't need your attention. Progress updates arrive every second " + "during playback. Use 'speech_status' to check outcomes after the fact. " "Use 'generate_audio' to synthesize without playing. " "Engines: kokoro (fast ONNX, ~50 voices), piper (Wyoming/Docker), " "orpheus (LLM via llama-server, supports etc.). " @@ -142,6 +146,43 @@ async def _play_tone(tone_path: Path) -> None: print(f" Entry tone failed (non-fatal): {e}", file=sys.stderr) +_CHUNK_WORD_THRESHOLD = 20 # texts shorter than this bypass chunking + + +def split_text(text: str) -> list[str]: + """Split text into sentence-sized chunks for pipelined synthesis. + + Splits on sentence-ending punctuation (.!?) followed by whitespace. + Merges short fragments (< 5 words) with their neighbor. + Returns [text] unchanged if under the word threshold. + """ + words = text.split() + if len(words) < _CHUNK_WORD_THRESHOLD: + return [text] + + # Split on sentence boundaries — require uppercase after split to avoid + # false positives on abbreviations (Dr.), decimals (3.0), URLs, etc. + raw = re.split(r"(?<=[.!?])\s+(?=[A-Z])", text.strip()) + if len(raw) <= 1: + return [text] + + # Merge short fragments (< 5 words) with next chunk. + # First chunk is allowed to be short — minimizes time-to-first-audio. + chunks: list[str] = [raw[0]] + for fragment in raw[1:]: + if len(chunks[-1].split()) < 5: + chunks[-1] = chunks[-1] + " " + fragment + else: + chunks.append(fragment) + + # If the last chunk is very short, merge it into its predecessor + if len(chunks) > 1 and len(chunks[-1].split()) < 5: + tail = chunks.pop() + chunks[-1] = chunks[-1] + " " + tail + + return chunks + + async def _resolve_project_voice( ctx: Context, engine_name: str, @@ -171,7 +212,196 @@ async def _resolve_project_voice( # Tools # --------------------------------------------------------------------------- -@mcp.tool +async def _speak_single( + eng: TTSEngine, + engine: str, + text: str, + voice: str | None, + queue: SpeechQueue, + entry_tone: Path | None, + priority: Priority, + ctx: Context, +) -> dict: + """Single-shot speak path — synthesize full text, then enqueue. + + Used for short texts (< 20 words) and as the fallback when text has + no sentence boundaries. + """ + try: + if entry_tone: + await _play_tone(entry_tone) + await ctx.report_progress(progress=5, total=100, message="Entry tone") + + await ctx.info(f"Synthesizing with {engine}...") + result = await eng.synthesize(text, voice) + await ctx.report_progress(progress=30, total=100, message="Synthesizing...") + + await ctx.info(f"Synthesized {result.duration_seconds:.1f}s audio, enqueueing...") + enqueue_result = await queue.enqueue(result, priority=priority) + + if not enqueue_result.get("queued"): + return enqueue_result + + speech_id = enqueue_result["speech_id"] + duration = result.duration_seconds + await ctx.report_progress( + progress=35, total=100, + message="Playing audio \u2014 you can continue working", + ) + + await ctx.info("Playing...") + # NOTE: Do NOT use asyncio.wait_for() — see CLAUDE.md Python 3.13 pitfall + outcome = await _await_with_progress(queue, speech_id, duration, ctx, 35) + + await ctx.report_progress(progress=100, total=100, message="Playback complete") + return outcome + + except asyncio.CancelledError: + # speak() was cancelled (server shutdown or MCP client disconnect). + # Do NOT cancel the consumer — let it finish the current audio. + # Shutdown: queue.stop() in the lifespan finalizer handles graceful drain. + # Client cancel: use cancel_speech(speech_id) for explicit mid-playback stop. + raise + + +async def _speak_chunked( + eng: TTSEngine, + engine: str, + chunks: list[str], + voice: str | None, + queue: SpeechQueue, + entry_tone: Path | None, + priority: Priority, + ctx: Context, +) -> dict: + """Chunked speak path — pipeline synthesis with playback. + + Synthesizes sentence chunks one at a time and enqueues each immediately. + Playback of chunk N overlaps synthesis of chunk N+1. Non-final chunks + suppress exit/standby tones for seamless playback. + """ + n_chunks = len(chunks) + total_duration = 0.0 + all_speech_ids: list[str] = [] + first_enqueue_time: float | None = None + + try: + if entry_tone: + await _play_tone(entry_tone) + await ctx.report_progress(progress=5, total=100, message="Entry tone") + + for i, chunk in enumerate(chunks): + is_last = i == n_chunks - 1 + pct_synth = 5 + int(25 * (i + 1) / n_chunks) # 5-30% across all chunks + + await ctx.info(f"Synthesizing chunk {i + 1}/{n_chunks} with {engine}...") + result = await eng.synthesize(chunk, voice) + total_duration += result.duration_seconds + await ctx.report_progress( + progress=pct_synth, total=100, + message=f"Synthesizing chunk {i + 1}/{n_chunks}...", + ) + + enqueue_result = await queue.enqueue( + result, + priority=priority, + suppress_exit_tone=not is_last, + ) + + if not enqueue_result.get("queued"): + # Queue full mid-message — already-enqueued chunks play through + if all_speech_ids: + await ctx.info(f"Queue full at chunk {i + 1}/{n_chunks}, waiting for enqueued chunks...") + elapsed = time.time() - first_enqueue_time if first_enqueue_time else 0 + remaining = max(total_duration - elapsed, 0.1) + outcome = await _await_with_progress(queue, all_speech_ids[-1], remaining, ctx, 35) + outcome["chunks_enqueued"] = len(all_speech_ids) + outcome["chunks_total"] = n_chunks + return outcome + return enqueue_result + + all_speech_ids.append(enqueue_result["speech_id"]) + if first_enqueue_time is None: + first_enqueue_time = time.time() + + # All chunks enqueued — wait for final chunk to finish playing + if not all_speech_ids: + return {"error": "No chunks were successfully enqueued", "chunks_total": n_chunks} + + await ctx.report_progress( + progress=35, total=100, + message="Playing audio \u2014 you can continue working", + ) + await ctx.info(f"Playing {n_chunks} chunks ({total_duration:.1f}s total)...") + + # Subtract time already elapsed since first chunk started playing + elapsed = time.time() - first_enqueue_time if first_enqueue_time else 0 + remaining = max(total_duration - elapsed, 0.1) + outcome = await _await_with_progress(queue, all_speech_ids[-1], remaining, ctx, 35) + await ctx.report_progress(progress=100, total=100, message="Playback complete") + + # Check for partial failures in earlier chunks + failed_chunks = [] + for idx, sid in enumerate(all_speech_ids[:-1]): + chunk_status = queue.get_status(sid) + if chunk_status.get("status") == "completed" and not chunk_status.get("played", True): + failed_chunks.append(idx + 1) + + outcome["chunks"] = n_chunks + outcome["total_duration_seconds"] = total_duration + if failed_chunks: + outcome["partial_failure"] = True + outcome["failed_chunks"] = failed_chunks + return outcome + + except asyncio.CancelledError: + # speak() was cancelled (server shutdown or MCP client disconnect). + # Do NOT cancel the consumer — let already-enqueued chunks play through. + # Shutdown: queue.stop() in the lifespan finalizer handles graceful drain. + # Client cancel: use cancel_speech(speech_id) for explicit mid-playback stop. + raise + + +async def _await_with_progress( + queue: SpeechQueue, + speech_id: str, + duration: float, + ctx: Context, + base_pct: int, +) -> dict: + """Await playback completion with a progress ticker. + + Uses a background task for progress — NOT asyncio.wait_for polling + (see CLAUDE.md for the Python 3.13 pitfall). + """ + async def _progress_ticker(): + t0 = time.time() + while True: + await asyncio.sleep(0.5) + elapsed = time.time() - t0 + pct = min(base_pct + int((99 - base_pct) * elapsed / max(duration, 0.1)), 99) + await ctx.report_progress( + progress=pct, total=100, + message="Playing audio \u2014 you can continue working", + ) + + ticker = asyncio.create_task(_progress_ticker()) + try: + return await queue.wait_for_completion(speech_id) + finally: + ticker.cancel() + try: + await ticker + except asyncio.CancelledError: + pass + + +@mcp.tool(annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + idempotentHint=True, + openWorldHint=True, +)) async def speak( text: str, engine: ENGINE_NAMES = "kokoro", @@ -182,8 +412,14 @@ async def speak( ) -> dict: """Synthesize text and play it through the host speakers. - Blocks until playback finishes, reporting progress throughout: - entry tone → synthesis → queued → playing → done. + Audio plays through physical speakers — you don't need to wait for the + result to continue your work. Feel free to call this alongside other tools + in the same message. Progress notifications report playback status every + second. + + Long texts are automatically split into sentences and pipelined — the + first sentence plays within ~1-2s while remaining sentences synthesize + in the background. Audio is queued so agents don't talk over each other. Urgent messages jump ahead of normal-priority items. @@ -205,70 +441,15 @@ async def speak( ctx, engine, eng, voice, text, voice_cache, project, ) - speech_id = None - try: - # Entry tone — audible "I heard you" before synthesis begins - if entry_tone: - await _play_tone(entry_tone) - await ctx.report_progress(progress=5, total=100) + priority = Priority.URGENT if urgent else Priority.NORMAL - # Synthesize audio (multiple agents can synthesize simultaneously) - await ctx.info(f"Synthesizing with {engine}...") - result = await eng.synthesize(text, voice) - await ctx.report_progress(progress=30, total=100) - - # Enqueue for serialized playback - await ctx.info(f"Synthesized {result.duration_seconds:.1f}s audio, enqueueing...") - enqueue_result = await queue.enqueue( - result, - priority=Priority.URGENT if urgent else Priority.NORMAL, - ) - - if not enqueue_result.get("queued"): - return enqueue_result - - speech_id = enqueue_result["speech_id"] - duration = result.duration_seconds - await ctx.report_progress(progress=35, total=100) - - # Wait for playback with progress via a background ticker. - # NOTE: Do NOT use asyncio.wait_for() with the future — Python 3.13's - # wait_for cancels its inner task on timeout, and repeated cancel/re-await - # on the same Future causes a stale CancelledError to propagate to the - # consumer task, killing pw-play mid-sentence. - await ctx.info("Playing...") - - async def _progress_ticker(): - """Send progress updates while playback runs.""" - t0 = time.time() - while True: - await asyncio.sleep(0.5) - elapsed = time.time() - t0 - pct = min(35 + int(64 * elapsed / max(duration, 0.1)), 99) - await ctx.report_progress(progress=pct, total=100) - - ticker = asyncio.create_task(_progress_ticker()) - try: - outcome = await queue.wait_for_completion(speech_id) - finally: - ticker.cancel() - try: - await ticker - except asyncio.CancelledError: - pass - - await ctx.report_progress(progress=100, total=100) - return outcome - - except asyncio.CancelledError: - # speak() was cancelled (server shutdown or MCP client disconnect). - # Do NOT cancel the consumer — let it finish the current audio. - # Shutdown: queue.stop() in the lifespan finalizer handles graceful drain. - # Client cancel: use cancel_speech(speech_id) to explicitly stop playback. - raise + chunks = split_text(text) + if len(chunks) <= 1: + return await _speak_single(eng, engine, text, voice, queue, entry_tone, priority, ctx) + return await _speak_chunked(eng, engine, chunks, voice, queue, entry_tone, priority, ctx) -@mcp.tool +@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True)) async def speech_status( speech_id: str, ctx: Context = CurrentContext(), @@ -285,7 +466,11 @@ async def speech_status( return queue.get_status(speech_id) -@mcp.tool +@mcp.tool(annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=True, + idempotentHint=True, +)) async def cancel_speech( speech_id: str, ctx: Context = CurrentContext(), @@ -302,7 +487,11 @@ async def cancel_speech( return queue.cancel(speech_id) -@mcp.tool +@mcp.tool(annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + idempotentHint=True, +)) async def generate_audio( text: str, engine: ENGINE_NAMES = "kokoro", @@ -343,7 +532,7 @@ async def generate_audio( } -@mcp.tool +@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True)) async def list_voices( engine: ENGINE_NAMES, ctx: Context = CurrentContext(), @@ -364,7 +553,7 @@ async def list_voices( return await engines[engine].list_voices() -@mcp.tool +@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True)) async def list_engines( ctx: Context = CurrentContext(), ) -> list[dict]: