feat(mcp): output mode + base-path file lane so agents never carry audio in context
An LLM agent pays for every byte it receives, and generate_speech returned each WAV as base64 inline - a short clip already brushed per-result limits. This adds two knobs in the OMNIVOICE_* family, the pattern the ElevenLabs MCP settled on (OUTPUT_MODE + a BASE_PATH security boundary): - OMNIVOICE_MCP_OUTPUT_MODE = resources (default, the original contract) | files | both. In files mode generate_speech returns audio_url (the render the backend already keeps, served at /audio/<id>.wav) and, when a base path is set, output_path - the WAV written into that directory. - OMNIVOICE_MCP_BASE_PATH: the one directory agents may read from and receive files in. transcribe(audio_path=) and clone_voice(ref_audio_path=) read only inside it (relative paths resolve against it, absolute ones must lie within it, symlinks resolved before the check); with no base path, path arguments are refused with a reason. - OMNIVOICE_MCP_TIMEOUT_S (default 120): the tools' backend timeout, since a CPU host serializes generations and an agent queued behind another render outlasted the fixed budget with an empty-message ToolError. Also: transcribe and clone_voice share one input helper (data-URI tolerance now covers transcribe too), the upload filename carries the sniffed extension, and the reply is built with json.dumps instead of hand-rolled JSON. Tests cover the mode parsing, the boundary (escape and missing-base refusals), both input lanes, all four reply shapes, and the timeout knob. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Ævar Guðmundsson
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
a30b71666c
commit
3c3c37615c
@@ -16,6 +16,8 @@ the frozen-backend fallback mirror it for their toolchains.
|
||||
|
||||
### Added
|
||||
|
||||
- The MCP server gains an output mode and a file lane: `OMNIVOICE_MCP_OUTPUT_MODE` (`resources` — the original base64 default — `files`, or `both`) lets `generate_speech` return a URL to the render plus a WAV written under `OMNIVOICE_MCP_BASE_PATH` instead of base64 inline, and `transcribe` / `clone_voice` accept `audio_path` / `ref_audio_path` read from inside that same base path, which acts as the security boundary — so an LLM agent never has to carry audio through its context.
|
||||
|
||||
### Docs
|
||||
|
||||
- The CosyVoice guide now states that packaged builds have no one-click runtime installer and records the exact readiness checks exposed by [Discussion 1631](https://github.com/debpalash/VoiceStudio/discussions/1631).
|
||||
|
||||
+214
-37
@@ -7,8 +7,8 @@ Run standalone:
|
||||
|
||||
Tools exposed:
|
||||
generate_speech — text → WAV audio (voice clone or design)
|
||||
clone_voice — base64 reference audio → new voice profile
|
||||
transcribe — base64 audio → text
|
||||
clone_voice — reference audio (base64, or a file path) → new voice profile
|
||||
transcribe — audio (base64, or a file path) → text
|
||||
list_voices — enumerate saved voice profiles
|
||||
list_languages — available TTS languages
|
||||
list_personalities — voice personality presets
|
||||
@@ -17,6 +17,18 @@ Tools exposed:
|
||||
Resources exposed:
|
||||
voice://{profile_id} — voice profile metadata
|
||||
history://recent — last 20 generated audio items
|
||||
|
||||
Output mode (OMNIVOICE_MCP_OUTPUT_MODE):
|
||||
resources — generate_speech returns the WAV as base64 inline (the original
|
||||
contract; default)
|
||||
files — it returns a URL to the render (and, with a base path, a WAV
|
||||
written there); nothing large ever enters the agent's context
|
||||
both — both of the above
|
||||
|
||||
File inputs (OMNIVOICE_MCP_BASE_PATH):
|
||||
One directory that agents may read audio from (transcribe / clone_voice
|
||||
`*_path` arguments) and receive files in (files mode). It is the security
|
||||
boundary: with no base path configured, path-shaped inputs are refused.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -69,6 +81,163 @@ def _sniff_audio_ext(raw: bytes) -> str:
|
||||
return ".wav"
|
||||
|
||||
|
||||
# ── Output mode + the base path boundary ─────────────────────────────────
|
||||
# An LLM agent that receives a WAV as base64 pays for every byte in context:
|
||||
# a 1.4 s clip already brushes per-result token caps, and a paragraph of
|
||||
# narration blows them outright. The ElevenLabs MCP settled this with an
|
||||
# OUTPUT_MODE (files / resources / both) and a BASE_PATH that doubles as the
|
||||
# security boundary for file-shaped inputs; the same two knobs here, named in
|
||||
# the OMNIVOICE_* family the rest of the server reads.
|
||||
|
||||
_OUTPUT_MODES = ("resources", "files", "both")
|
||||
_MAX_INPUT_BYTES = 200 * 1024 * 1024
|
||||
|
||||
|
||||
def _output_mode() -> str:
|
||||
"""How generate_speech hands audio back (OMNIVOICE_MCP_OUTPUT_MODE).
|
||||
|
||||
'resources' is the original base64-inline contract and stays the default
|
||||
so existing integrations see no change; 'files' returns a URL to the
|
||||
render (plus a WAV under the base path when one is configured); 'both'
|
||||
returns everything. Anything unrecognized falls back to 'resources' with
|
||||
a warning rather than failing the tool."""
|
||||
mode = os.environ.get("OMNIVOICE_MCP_OUTPUT_MODE", "resources").strip().lower()
|
||||
if mode not in _OUTPUT_MODES:
|
||||
logger.warning(
|
||||
"OMNIVOICE_MCP_OUTPUT_MODE=%r is not one of %s; using 'resources'",
|
||||
mode, _OUTPUT_MODES,
|
||||
)
|
||||
return "resources"
|
||||
return mode
|
||||
|
||||
|
||||
def _base_path() -> "str | None":
|
||||
"""The one directory agents may read audio from and receive files in
|
||||
(OMNIVOICE_MCP_BASE_PATH), realpath'd; None when unset."""
|
||||
raw = os.environ.get("OMNIVOICE_MCP_BASE_PATH", "").strip()
|
||||
if not raw:
|
||||
return None
|
||||
return os.path.realpath(os.path.expanduser(raw))
|
||||
|
||||
|
||||
def _resolve_under_base(path: str) -> str:
|
||||
"""Absolute realpath of ``path`` when it lies inside the base path.
|
||||
|
||||
Relative paths resolve against the base; absolute paths must already be
|
||||
inside it. Both sides are realpath'd, so a symlink pointing outward cannot
|
||||
smuggle a read in. Raises ValueError with an agent-legible reason when no
|
||||
base path is configured or the path escapes it."""
|
||||
base = _base_path()
|
||||
if base is None:
|
||||
raise ValueError(
|
||||
"OMNIVOICE_MCP_BASE_PATH is not set; file paths are refused until it "
|
||||
"names a directory"
|
||||
)
|
||||
candidate = os.path.realpath(os.path.join(base, os.path.expanduser(path)))
|
||||
try:
|
||||
inside = os.path.commonpath([base, candidate]) == base
|
||||
except ValueError: # different drives on Windows: nothing in common
|
||||
inside = False
|
||||
if not inside:
|
||||
raise ValueError(f"{path!r} resolves outside OMNIVOICE_MCP_BASE_PATH")
|
||||
return candidate
|
||||
|
||||
|
||||
def _read_input_audio(
|
||||
audio_base64: "str | None",
|
||||
audio_path: "str | None",
|
||||
*,
|
||||
label: str = "audio_base64",
|
||||
too_big: str = "audio exceeds 200 MB limit",
|
||||
) -> "tuple[bytes | None, str | None]":
|
||||
"""Audio bytes from exactly one of the two input lanes, or (None, error).
|
||||
|
||||
The base64 lane keeps its data-URI tolerance and 200 MB cap; the path lane
|
||||
is honored only inside the base path (the security boundary) and applies
|
||||
the same cap to the file's size before reading it."""
|
||||
if bool(audio_base64) == bool(audio_path):
|
||||
return None, f"pass exactly one of {label} or the matching *_path argument"
|
||||
if audio_path:
|
||||
try:
|
||||
resolved = _resolve_under_base(audio_path)
|
||||
except ValueError as e:
|
||||
return None, str(e)
|
||||
if not os.path.isfile(resolved):
|
||||
return None, f"no such file under OMNIVOICE_MCP_BASE_PATH: {audio_path!r}"
|
||||
if os.path.getsize(resolved) > _MAX_INPUT_BYTES:
|
||||
return None, too_big
|
||||
with open(resolved, "rb") as f:
|
||||
return f.read(), None
|
||||
# Base64 is always larger than the bytes it carries, so the encoded
|
||||
# length is a safe lower bound on the decoded size.
|
||||
if len(audio_base64) > _MAX_INPUT_BYTES:
|
||||
return None, too_big
|
||||
raw = _decode_ref_audio(audio_base64)
|
||||
if raw is None:
|
||||
return None, f"{label} is not valid base64"
|
||||
if not raw:
|
||||
return None, f"{label} is empty"
|
||||
return raw, None
|
||||
|
||||
|
||||
def _write_output(audio_id: str, raw: bytes) -> str:
|
||||
"""Land a render under the base path as ``<audio_id>.wav``; returns the path."""
|
||||
base = _base_path()
|
||||
os.makedirs(base, exist_ok=True)
|
||||
path = os.path.join(base, f"{audio_id}.wav")
|
||||
with open(path, "wb") as f:
|
||||
f.write(raw)
|
||||
return path
|
||||
|
||||
|
||||
def _post_timeout_s() -> float:
|
||||
"""Seconds the tools wait on a backend POST (OMNIVOICE_MCP_TIMEOUT_S,
|
||||
default 120). A CPU host renders a paragraph in minutes and serializes
|
||||
generations, so an agent behind another render used to hit the fixed
|
||||
budget with an empty-message timeout; the knob follows the backend's own
|
||||
OMNIVOICE_GENERATE_TIMEOUT_S when a deployment raises that."""
|
||||
raw = os.environ.get("OMNIVOICE_MCP_TIMEOUT_S", "").strip()
|
||||
try:
|
||||
value = float(raw) if raw else 120.0
|
||||
except ValueError:
|
||||
logger.warning("OMNIVOICE_MCP_TIMEOUT_S=%r is not a number; using 120", raw)
|
||||
return 120.0
|
||||
return value if value > 0 else 120.0
|
||||
|
||||
|
||||
def _maybe_number(value):
|
||||
"""A response-header number as a number, or the raw text (e.g. '?')."""
|
||||
try:
|
||||
return float(value)
|
||||
except (TypeError, ValueError):
|
||||
return value
|
||||
|
||||
|
||||
def _speech_result(audio_id: str, gen_time, duration, raw: bytes, api_base: str) -> dict:
|
||||
"""The generate_speech reply shaped by the output mode.
|
||||
|
||||
The backend already keeps every render on disk and serves it at
|
||||
``/audio/<audio_id>.wav``, so files mode costs nothing but a URL - plus one
|
||||
write when a base path invites the WAV into the agent's own directory."""
|
||||
mode = _output_mode()
|
||||
out = {
|
||||
"audio_id": audio_id,
|
||||
"generation_time_s": gen_time,
|
||||
"audio_duration_s": duration,
|
||||
"format": "wav",
|
||||
"output_mode": mode,
|
||||
}
|
||||
if mode in ("files", "both"):
|
||||
out["audio_url"] = f"{api_base.rstrip('/')}/audio/{audio_id}.wav"
|
||||
if _base_path() is not None:
|
||||
out["output_path"] = _write_output(audio_id, raw)
|
||||
else:
|
||||
out["note"] = "set OMNIVOICE_MCP_BASE_PATH to also receive the WAV as a file"
|
||||
if mode in ("resources", "both"):
|
||||
out["wav_base64"] = base64.b64encode(raw).decode("ascii")
|
||||
return out
|
||||
|
||||
|
||||
# ── Lazy imports — keeps startup fast when not using MCP ────────────────
|
||||
|
||||
|
||||
@@ -147,7 +316,7 @@ def create_mcp_server():
|
||||
|
||||
async def _api_post_form(path: str, data: dict, files: dict | None = None):
|
||||
import httpx
|
||||
async with httpx.AsyncClient(base_url=_api_base(), timeout=120) as c:
|
||||
async with httpx.AsyncClient(base_url=_api_base(), timeout=_post_timeout_s()) as c:
|
||||
r = await c.post(path, data=data, files=files or {})
|
||||
r.raise_for_status()
|
||||
return r
|
||||
@@ -190,8 +359,12 @@ def create_mcp_server():
|
||||
steps: Diffusion steps (8=fast/draft, 16=balanced, 32=quality).
|
||||
|
||||
Returns:
|
||||
JSON with audio_id, generation_time, audio_duration, and
|
||||
base64-encoded WAV data.
|
||||
JSON with audio_id, generation_time_s, audio_duration_s and the
|
||||
audio itself shaped by OMNIVOICE_MCP_OUTPUT_MODE: base64 WAV data
|
||||
('resources', the default), a URL to the render plus a WAV under
|
||||
OMNIVOICE_MCP_BASE_PATH when one is set ('files'), or all of the
|
||||
above ('both'). Prefer 'files' for LLM agents: nothing large
|
||||
enters the context.
|
||||
"""
|
||||
# Per-agent voice binding (Wave 2.2): explicit arg wins; otherwise
|
||||
# resolve this client's bound profile, then the global default.
|
||||
@@ -218,18 +391,10 @@ def create_mcp_server():
|
||||
r = await _api_post_form("/generate", data=form)
|
||||
|
||||
audio_id = r.headers.get("X-Audio-Id", "unknown")
|
||||
gen_time = r.headers.get("X-Gen-Time", "?")
|
||||
duration = r.headers.get("X-Audio-Duration", "?")
|
||||
gen_time = _maybe_number(r.headers.get("X-Gen-Time", "?"))
|
||||
duration = _maybe_number(r.headers.get("X-Audio-Duration", "?"))
|
||||
|
||||
wav_b64 = base64.b64encode(r.content).decode("ascii")
|
||||
|
||||
return (
|
||||
f'{{"audio_id":"{audio_id}",'
|
||||
f'"generation_time_s":{gen_time},'
|
||||
f'"audio_duration_s":{duration},'
|
||||
f'"format":"wav",'
|
||||
f'"wav_base64":"{wav_b64}"}}'
|
||||
)
|
||||
return json.dumps(_speech_result(audio_id, gen_time, duration, r.content, _api_base()))
|
||||
|
||||
@mcp.tool()
|
||||
async def list_voices() -> str:
|
||||
@@ -266,30 +431,39 @@ def create_mcp_server():
|
||||
)
|
||||
|
||||
@mcp.tool()
|
||||
async def transcribe(audio_base64: str, language: str | None = None) -> str:
|
||||
async def transcribe(
|
||||
audio_base64: str | None = None,
|
||||
audio_path: str | None = None,
|
||||
language: str | None = None,
|
||||
) -> str:
|
||||
"""Transcribe spoken audio to text.
|
||||
|
||||
Pass exactly one of audio_base64 or audio_path.
|
||||
|
||||
Args:
|
||||
audio_base64: Base64-encoded audio bytes (wav/mp3/webm/m4a).
|
||||
audio_path: Path to an audio file under OMNIVOICE_MCP_BASE_PATH
|
||||
(relative to it, or absolute inside it). The base path is the
|
||||
security boundary: with none configured, paths are refused.
|
||||
Prefer this lane for LLM agents - the audio never enters the
|
||||
agent's context.
|
||||
language: Optional language hint; omit for auto-detect.
|
||||
|
||||
Returns:
|
||||
JSON with the recognized text, language, and duration.
|
||||
"""
|
||||
try:
|
||||
raw = base64.b64decode(audio_base64, validate=True)
|
||||
except Exception:
|
||||
return '{"error":"audio_base64 is not valid base64"}'
|
||||
# 200 MB cap — same spirit as voicebox's transcribe gate. Keeps a
|
||||
# buggy/hostile agent from posting an unbounded blob.
|
||||
if len(raw) > 200 * 1024 * 1024:
|
||||
return '{"error":"audio exceeds 200 MB limit"}'
|
||||
# 200 MB cap on both lanes — same spirit as voicebox's transcribe
|
||||
# gate. Keeps a buggy/hostile agent from posting an unbounded blob.
|
||||
raw, err = _read_input_audio(audio_base64, audio_path)
|
||||
if err:
|
||||
return json.dumps({"error": err})
|
||||
data = {}
|
||||
if language:
|
||||
data["language"] = language
|
||||
r = await _api_post_form(
|
||||
"/transcribe", data=data,
|
||||
files={"audio": ("audio.wav", raw, "application/octet-stream")},
|
||||
files={"audio": (f"audio{_sniff_audio_ext(raw)}", raw,
|
||||
"application/octet-stream")},
|
||||
)
|
||||
return str(r.json())
|
||||
|
||||
@@ -319,15 +493,17 @@ def create_mcp_server():
|
||||
@mcp.tool()
|
||||
async def clone_voice(
|
||||
name: str,
|
||||
ref_audio_base64: str,
|
||||
ref_audio_base64: str | None = None,
|
||||
ref_text: str = "",
|
||||
instruct: str = "",
|
||||
language: str = "Auto",
|
||||
ref_audio_path: str | None = None,
|
||||
) -> str:
|
||||
"""Clone a new voice profile from a reference audio sample.
|
||||
|
||||
The new voice is immediately available for use with generate_speech
|
||||
(pass the returned profile_id as the profile_id argument).
|
||||
(pass the returned profile_id as the profile_id argument). Pass
|
||||
exactly one of ref_audio_base64 or ref_audio_path.
|
||||
|
||||
Args:
|
||||
name: A human-friendly name for the cloned voice.
|
||||
@@ -338,19 +514,20 @@ def create_mcp_server():
|
||||
quality for some engines).
|
||||
instruct: Optional style instruction (e.g. 'whisper', 'excited').
|
||||
language: Language of the reference audio (ISO code or 'Auto').
|
||||
ref_audio_path: Path to the reference audio under
|
||||
OMNIVOICE_MCP_BASE_PATH (relative to it, or absolute inside
|
||||
it); refused when no base path is configured. Prefer this
|
||||
lane for LLM agents - the clip never enters the context.
|
||||
|
||||
Returns:
|
||||
JSON with the new profile's id, name, and kind.
|
||||
"""
|
||||
# Reject oversized inputs before decoding (base64 is always larger
|
||||
# than raw, so this is a safe lower bound on the decoded size).
|
||||
if len(ref_audio_base64) > 200 * 1024 * 1024:
|
||||
return '{"error":"reference audio exceeds 200 MB limit"}'
|
||||
raw = _decode_ref_audio(ref_audio_base64)
|
||||
if raw is None:
|
||||
return '{"error":"ref_audio_base64 is not valid base64"}'
|
||||
if not raw:
|
||||
return '{"error":"ref_audio_base64 is empty"}'
|
||||
raw, err = _read_input_audio(
|
||||
ref_audio_base64, ref_audio_path,
|
||||
label="ref_audio_base64", too_big="reference audio exceeds 200 MB limit",
|
||||
)
|
||||
if err:
|
||||
return json.dumps({"error": err})
|
||||
import httpx
|
||||
try:
|
||||
r = await _api_post_form(
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
{
|
||||
"_comment": "MCP client config for VoiceStudio. See docs/mcp.md for both connection modes.",
|
||||
"_streamable_http": "If your MCP client speaks Streamable HTTP, point it directly at the running app: http://localhost:3900/mcp — no separate process needed (the server is mounted on the backend). Send an X-OmniVoice-Client-Id header to bind this agent to a specific voice.",
|
||||
"_output_mode": "To keep audio out of the agent's context, set OMNIVOICE_MCP_OUTPUT_MODE=files and OMNIVOICE_MCP_BASE_PATH=<the agent's working directory> on the BACKEND's environment for the mounted endpoint (or on a standalone `python -m backend.mcp_server` entry's env). See docs/mcp.md, 'Output mode and file inputs'.",
|
||||
"mcpServers": {
|
||||
"omnivoice": {
|
||||
"command": "python",
|
||||
|
||||
+24
-3
@@ -10,12 +10,33 @@ start once VoiceStudio is open.
|
||||
|
||||
| Tool | What it does |
|
||||
|---|---|
|
||||
| `generate_speech` | text → WAV (base64). Uses the agent's bound voice unless a `profile_id` is passed. |
|
||||
| `clone_voice` | base64 audio → new voice profile. Returns a `profile_id` for use with `generate_speech`. |
|
||||
| `transcribe` | base64 audio → text (646 languages). |
|
||||
| `generate_speech` | text → WAV. Uses the agent's bound voice unless a `profile_id` is passed. Returns base64 by default, or a URL + file in [files mode](#output-mode-and-file-inputs). |
|
||||
| `clone_voice` | reference audio (base64, or a `ref_audio_path` under the base path) → new voice profile. Returns a `profile_id` for use with `generate_speech`. |
|
||||
| `transcribe` | audio (base64, or an `audio_path` under the base path) → text (646 languages). |
|
||||
| `list_voices` / `list_personalities` / `list_languages` | enumerate what's available. |
|
||||
| `check_health` | backend status + active GPU device. |
|
||||
|
||||
## Output mode and file inputs
|
||||
|
||||
An LLM agent pays for every byte it receives in context, and a WAV as base64
|
||||
is a lot of bytes — a short clip already brushes per-result limits, a
|
||||
paragraph of narration blows them. Two environment variables move the audio
|
||||
out of the conversation and onto disk, where an agent can hand it to a player
|
||||
or another tool by path:
|
||||
|
||||
| Variable | Values | Effect |
|
||||
|---|---|---|
|
||||
| `OMNIVOICE_MCP_OUTPUT_MODE` | `resources` (default) · `files` · `both` | `resources` returns `wav_base64` inline (the original contract). `files` returns `audio_url` (the render served at `/audio/<audio_id>.wav`, which the backend keeps anyway) and, when a base path is set, `output_path` — the WAV written into that directory. `both` returns everything. |
|
||||
| `OMNIVOICE_MCP_TIMEOUT_S` | seconds (default `120`) | How long a tool waits on the backend. CPU hosts render a paragraph in minutes and serialize generations, so an agent queued behind another render can outlast the default; raise it in step with `OMNIVOICE_GENERATE_TIMEOUT_S`. |
|
||||
| `OMNIVOICE_MCP_BASE_PATH` | a directory | The **security boundary** for file-shaped traffic. `transcribe(audio_path=…)` and `clone_voice(ref_audio_path=…)` read only from inside it (relative paths resolve against it, absolute paths must already lie within it, symlinks are resolved before the check), and files mode writes only into it. With no base path configured, path arguments are refused with a reason. |
|
||||
|
||||
Set them on the **backend's** environment for the mounted `/mcp` endpoint
|
||||
(the launcher, a service file, Docker `-e`), or on the server entry's `env`
|
||||
when running `python -m backend.mcp_server` standalone. A recommended agent
|
||||
setup: `OMNIVOICE_MCP_OUTPUT_MODE=files` with the base path pointing at the
|
||||
agent's own working directory — nothing large ever enters its context, and
|
||||
every render is a file it can name.
|
||||
|
||||
## Connecting
|
||||
|
||||
### Streamable HTTP (modern clients)
|
||||
|
||||
@@ -0,0 +1,177 @@
|
||||
"""MCP output mode + the base-path boundary.
|
||||
|
||||
Pure helpers, no MCP SDK needed: how generate_speech hands audio back
|
||||
(OMNIVOICE_MCP_OUTPUT_MODE) and how path-shaped inputs are confined to
|
||||
OMNIVOICE_MCP_BASE_PATH. The tool closures themselves are exercised through
|
||||
the shape helpers they delegate to, so these run without a backend.
|
||||
"""
|
||||
import base64
|
||||
import os
|
||||
|
||||
os.environ.setdefault("OMNIVOICE_MODEL", "test")
|
||||
os.environ.setdefault("OMNIVOICE_DISABLE_FILE_LOG", "1")
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ── output mode ─────────────────────────────────────────────────────────────
|
||||
|
||||
def test_output_mode_defaults_to_resources(monkeypatch):
|
||||
from mcp_server import _output_mode
|
||||
monkeypatch.delenv("OMNIVOICE_MCP_OUTPUT_MODE", raising=False)
|
||||
assert _output_mode() == "resources"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw,expected", [
|
||||
("files", "files"),
|
||||
("FILES", "files"),
|
||||
(" both ", "both"),
|
||||
("resources", "resources"),
|
||||
("banana", "resources"), # unrecognized falls back, never fails the tool
|
||||
])
|
||||
def test_output_mode_parses_and_falls_back(monkeypatch, raw, expected):
|
||||
from mcp_server import _output_mode
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_OUTPUT_MODE", raw)
|
||||
assert _output_mode() == expected
|
||||
|
||||
|
||||
# ── base path boundary ──────────────────────────────────────────────────────
|
||||
|
||||
def test_base_path_none_when_unset(monkeypatch):
|
||||
from mcp_server import _base_path
|
||||
monkeypatch.delenv("OMNIVOICE_MCP_BASE_PATH", raising=False)
|
||||
assert _base_path() is None
|
||||
|
||||
|
||||
def test_resolve_refuses_paths_without_a_base(monkeypatch):
|
||||
from mcp_server import _resolve_under_base
|
||||
monkeypatch.delenv("OMNIVOICE_MCP_BASE_PATH", raising=False)
|
||||
with pytest.raises(ValueError, match="OMNIVOICE_MCP_BASE_PATH is not set"):
|
||||
_resolve_under_base("clip.wav")
|
||||
|
||||
|
||||
def test_resolve_accepts_relative_and_absolute_inside(monkeypatch, tmp_path):
|
||||
from mcp_server import _resolve_under_base
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_BASE_PATH", str(tmp_path))
|
||||
inside = tmp_path / "sub" / "clip.wav"
|
||||
assert _resolve_under_base("sub/clip.wav") == os.path.realpath(str(inside))
|
||||
assert _resolve_under_base(str(inside)) == os.path.realpath(str(inside))
|
||||
|
||||
|
||||
def test_resolve_refuses_escape(monkeypatch, tmp_path):
|
||||
from mcp_server import _resolve_under_base
|
||||
base = tmp_path / "base"
|
||||
base.mkdir()
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_BASE_PATH", str(base))
|
||||
with pytest.raises(ValueError, match="outside OMNIVOICE_MCP_BASE_PATH"):
|
||||
_resolve_under_base("../secret.wav")
|
||||
with pytest.raises(ValueError, match="outside OMNIVOICE_MCP_BASE_PATH"):
|
||||
_resolve_under_base(str(tmp_path / "secret.wav"))
|
||||
|
||||
|
||||
# ── input lanes ─────────────────────────────────────────────────────────────
|
||||
|
||||
def test_read_input_requires_exactly_one_lane():
|
||||
from mcp_server import _read_input_audio
|
||||
raw, err = _read_input_audio(None, None)
|
||||
assert raw is None and "exactly one" in err
|
||||
raw, err = _read_input_audio("QUJD", "x.wav")
|
||||
assert raw is None and "exactly one" in err
|
||||
|
||||
|
||||
def test_read_input_path_lane_reads_inside_base(monkeypatch, tmp_path):
|
||||
from mcp_server import _read_input_audio
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_BASE_PATH", str(tmp_path))
|
||||
(tmp_path / "clip.wav").write_bytes(b"RIFFxxxxWAVE")
|
||||
raw, err = _read_input_audio(None, "clip.wav")
|
||||
assert err is None and raw == b"RIFFxxxxWAVE"
|
||||
|
||||
|
||||
def test_read_input_path_lane_reports_missing_and_escaped(monkeypatch, tmp_path):
|
||||
from mcp_server import _read_input_audio
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_BASE_PATH", str(tmp_path))
|
||||
raw, err = _read_input_audio(None, "nope.wav")
|
||||
assert raw is None and "no such file" in err
|
||||
raw, err = _read_input_audio(None, "../nope.wav")
|
||||
assert raw is None and "outside" in err
|
||||
|
||||
|
||||
def test_read_input_path_lane_refused_without_base(monkeypatch, tmp_path):
|
||||
from mcp_server import _read_input_audio
|
||||
monkeypatch.delenv("OMNIVOICE_MCP_BASE_PATH", raising=False)
|
||||
raw, err = _read_input_audio(None, str(tmp_path / "clip.wav"))
|
||||
assert raw is None and "is not set" in err
|
||||
|
||||
|
||||
def test_read_input_base64_lane_keeps_data_uri_tolerance_and_labels():
|
||||
from mcp_server import _read_input_audio
|
||||
body = base64.b64encode(b"RIFFxxxxWAVE").decode()
|
||||
raw, err = _read_input_audio(f"data:audio/wav;base64,{body}", None)
|
||||
assert err is None and raw == b"RIFFxxxxWAVE"
|
||||
raw, err = _read_input_audio("not!!base64", None, label="ref_audio_base64")
|
||||
assert raw is None and err == "ref_audio_base64 is not valid base64"
|
||||
|
||||
|
||||
# ── the generate_speech reply shape ─────────────────────────────────────────
|
||||
|
||||
def test_speech_result_resources_is_the_original_contract(monkeypatch):
|
||||
from mcp_server import _speech_result
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_OUTPUT_MODE", "resources")
|
||||
out = _speech_result("ab12cd34", 1.5, 2.0, b"RIFF", "http://localhost:3900")
|
||||
assert out["wav_base64"] == base64.b64encode(b"RIFF").decode()
|
||||
assert "audio_url" not in out and "output_path" not in out
|
||||
assert out["output_mode"] == "resources"
|
||||
|
||||
|
||||
def test_speech_result_files_returns_url_and_writes_under_base(monkeypatch, tmp_path):
|
||||
from mcp_server import _speech_result
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_OUTPUT_MODE", "files")
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_BASE_PATH", str(tmp_path))
|
||||
out = _speech_result("ab12cd34", 1.5, 2.0, b"RIFF", "http://localhost:3900/")
|
||||
assert out["audio_url"] == "http://localhost:3900/audio/ab12cd34.wav"
|
||||
assert "wav_base64" not in out
|
||||
written = out["output_path"]
|
||||
assert os.path.dirname(os.path.realpath(written)) == os.path.realpath(str(tmp_path))
|
||||
with open(written, "rb") as f:
|
||||
assert f.read() == b"RIFF"
|
||||
|
||||
|
||||
def test_speech_result_files_without_base_is_url_only_with_a_note(monkeypatch):
|
||||
from mcp_server import _speech_result
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_OUTPUT_MODE", "files")
|
||||
monkeypatch.delenv("OMNIVOICE_MCP_BASE_PATH", raising=False)
|
||||
out = _speech_result("ab12cd34", 1.5, 2.0, b"RIFF", "http://localhost:3900")
|
||||
assert out["audio_url"].endswith("/audio/ab12cd34.wav")
|
||||
assert "output_path" not in out and "OMNIVOICE_MCP_BASE_PATH" in out["note"]
|
||||
assert "wav_base64" not in out
|
||||
|
||||
|
||||
def test_speech_result_both_carries_everything(monkeypatch, tmp_path):
|
||||
from mcp_server import _speech_result
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_OUTPUT_MODE", "both")
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_BASE_PATH", str(tmp_path))
|
||||
out = _speech_result("ab12cd34", "?", "?", b"RIFF", "http://localhost:3900")
|
||||
assert {"wav_base64", "audio_url", "output_path"} <= set(out)
|
||||
assert out["generation_time_s"] == "?" # header text passes through untouched
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw,expected", [
|
||||
(None, 120.0),
|
||||
("600", 600.0),
|
||||
("0", 120.0), # non-positive falls back
|
||||
("soon", 120.0), # garbage falls back, never fails the tool
|
||||
])
|
||||
def test_post_timeout_reads_env_with_fallbacks(monkeypatch, raw, expected):
|
||||
from mcp_server import _post_timeout_s
|
||||
if raw is None:
|
||||
monkeypatch.delenv("OMNIVOICE_MCP_TIMEOUT_S", raising=False)
|
||||
else:
|
||||
monkeypatch.setenv("OMNIVOICE_MCP_TIMEOUT_S", raw)
|
||||
assert _post_timeout_s() == expected
|
||||
|
||||
|
||||
def test_maybe_number_keeps_header_text_honest():
|
||||
from mcp_server import _maybe_number
|
||||
assert _maybe_number("1.25") == 1.25
|
||||
assert _maybe_number("?") == "?"
|
||||
assert _maybe_number(None) is None
|
||||
Reference in New Issue
Block a user