From c2f2d2ed22973ea52f51a304105714f780c13c7e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B8=B8=E9=9B=81?= Date: Tue, 14 Jul 2026 11:13:04 +0000 Subject: [PATCH] fix(mcp): honor model and language configuration --- examples/mcp_server/README.md | 17 +-- examples/mcp_server/funasr_mcp.py | 97 ++++++++++---- examples/mcp_server/test_funasr_mcp.py | 175 ++++++++++++++++++++++++- 3 files changed, 252 insertions(+), 37 deletions(-) diff --git a/examples/mcp_server/README.md b/examples/mcp_server/README.md index 00bbe2e92..03f9d0746 100644 --- a/examples/mcp_server/README.md +++ b/examples/mcp_server/README.md @@ -1,6 +1,6 @@ # FunASR MCP Server -[Model Context Protocol](https://modelcontextprotocol.io/) server that gives AI assistants the ability to transcribe audio. +[Model Context Protocol](https://modelcontextprotocol.io/) server that gives AI assistants local audio transcription with SenseVoiceSmall by default. ## Setup @@ -113,9 +113,9 @@ Transcribe a speech audio file to text. | Name | Type | Required | Description | |------|------|----------|-------------| | `audio_path` | string | Yes | Path to audio file (wav, mp3, flac, m4a, ogg) | -| `language` | string | No | Language hint (auto-detected by default) | +| `language` | string | No | `auto`, `zh`, `yue`, `en`, `ja`, or `ko` (default: `auto`) | -**Returns:** Transcribed text with timestamps and speaker labels (when available). +**Returns:** Transcribed text with per-segment timestamps when the model returns them. ## Example Usage @@ -130,14 +130,15 @@ Once configured, ask your AI assistant: | Variable | Default | Description | |----------|---------|-------------| | `FUNASR_DEVICE` | `cpu` | Device: `cuda`, `cpu`, or `mps` | -| `FUNASR_MODEL` | `iic/SenseVoiceSmall` | ASR model to use | +| `FUNASR_MODEL` | `iic/SenseVoiceSmall` | Model name or local model path passed to `AutoModel` | ## Features -- **50+ languages** with automatic detection -- **Speaker diarization** — identifies who said what -- **Timestamps** — per-segment timing -- **170x realtime on GPU**, 17x on CPU +- **Five-language transcription** — Mandarin, Cantonese, English, Japanese, and Korean +- **Automatic detection or explicit hints** — `auto`, `zh`, `yue`, `en`, `ja`, and `ko` +- **VAD segmentation** — splits longer audio before recognition +- **Optional segment timestamps** — included only when the configured model returns them +- **Configurable local inference** — choose the model and CPU, CUDA, or MPS with environment variables - **No API key needed** — fully local inference - MIT licensed, privacy-friendly (audio never leaves your machine) diff --git a/examples/mcp_server/funasr_mcp.py b/examples/mcp_server/funasr_mcp.py index 6dabeca1f..f70e01f67 100644 --- a/examples/mcp_server/funasr_mcp.py +++ b/examples/mcp_server/funasr_mcp.py @@ -19,10 +19,34 @@ Add to claude_desktop_config.json: """ import json -import sys import os -import tempfile -import base64 +import re +import sys +from importlib.util import find_spec +from pathlib import Path + + +DEFAULT_MODEL = "iic/SenseVoiceSmall" +SUPPORTED_LANGUAGES = ("auto", "zh", "yue", "en", "ja", "ko") + + +def get_server_version(): + package_dirs = [] + package_spec = find_spec("funasr") + if package_spec is not None and package_spec.submodule_search_locations is not None: + package_dirs.extend(package_spec.submodule_search_locations) + package_dirs.append(Path(__file__).resolve().parent.parent.parent / "funasr") + + for package_dir in package_dirs: + version_file = Path(package_dir) / "version.txt" + try: + version = version_file.read_text().strip() + except OSError: + continue + if version: + return version + return "unknown" + # MCP protocol over stdio def send_response(id, result): @@ -32,11 +56,14 @@ def send_response(id, result): sys.stdout.flush() -def send_notification(method, params=None): - msg = {"jsonrpc": "2.0", "method": method, "params": params or {}} - out = json.dumps(msg) - sys.stdout.write(f"{out}\n") - sys.stdout.flush() +def send_tool_error(id, message): + send_response( + id, + { + "content": [{"type": "text", "text": f"Error: {message}"}], + "isError": True, + }, + ) _model = None @@ -46,9 +73,11 @@ def get_model(): global _model if _model is None: from funasr import AutoModel - device = os.environ.get("FUNASR_DEVICE", "cpu") + + model_name = os.environ.get("FUNASR_MODEL") or DEFAULT_MODEL + device = os.environ.get("FUNASR_DEVICE") or "cpu" _model = AutoModel( - model="iic/SenseVoiceSmall", + model=model_name, vad_model="fsmn-vad", vad_kwargs={"max_single_segment_time": 30000}, device=device, @@ -59,11 +88,10 @@ def get_model(): def transcribe(audio_path: str, language: str = "auto") -> dict: """Transcribe an audio file to text.""" - import re model = get_model() - result = model.generate(input=audio_path, batch_size=1) + result = model.generate(input=audio_path, batch_size=1, language=language) text = result[0]["text"] - text = re.sub(r'<\|[^|]*\|>', '', text).strip() + text = re.sub(r"<\|[^|]*\|>", "", text).strip() response = {"text": text} if "sentence_info" in result[0]: @@ -88,14 +116,14 @@ def handle_request(request): send_response(id, { "protocolVersion": "2024-11-05", "capabilities": {"tools": {"listChanged": False}}, - "serverInfo": {"name": "funasr", "version": "1.3.2"}, + "serverInfo": {"name": "funasr", "version": get_server_version()}, }) elif method == "tools/list": send_response(id, { "tools": [ { "name": "transcribe_audio", - "description": "Transcribe speech audio to text. Supports 50+ languages, auto-detection, speaker diarization. Input: file path to audio.", + "description": "Transcribe local speech audio with SenseVoiceSmall. Supports automatic or explicit Mandarin, Cantonese, English, Japanese, and Korean recognition with VAD segmentation.", "inputSchema": { "type": "object", "properties": { @@ -105,8 +133,9 @@ def handle_request(request): }, "language": { "type": "string", - "description": "Language hint (optional, auto-detected by default)", - "default": "auto" + "description": "Language hint: auto, zh, yue, en, ja, or ko", + "enum": list(SUPPORTED_LANGUAGES), + "default": "auto", } }, "required": ["audio_path"] @@ -119,17 +148,32 @@ def handle_request(request): args = params.get("arguments", {}) if tool_name == "transcribe_audio": - audio_path = args.get("audio_path", "") + audio_path = args.get("audio_path") language = args.get("language", "auto") - if not os.path.exists(audio_path): - send_response(id, { - "content": [{"type": "text", "text": f"Error: file not found: {audio_path}"}], - "isError": True - }) + if not isinstance(audio_path, str) or not audio_path.strip(): + send_tool_error(id, "audio_path is required") + return + + if language not in SUPPORTED_LANGUAGES: + supported = ", ".join(SUPPORTED_LANGUAGES) + send_tool_error( + id, + f"unsupported language '{language}'; choose one of: {supported}", + ) + return + + audio_path = os.path.expanduser(audio_path) + if not os.path.isfile(audio_path): + send_tool_error(id, f"file not found: {audio_path}") + return + + try: + result = transcribe(audio_path, language) + except Exception as error: + send_tool_error(id, f"transcription failed: {error}") return - result = transcribe(audio_path, language) text_output = f"Transcription: {result['text']}" if "segments" in result: text_output += "\n\nSegments:" @@ -141,10 +185,7 @@ def handle_request(request): "content": [{"type": "text", "text": text_output}] }) else: - send_response(id, { - "content": [{"type": "text", "text": f"Unknown tool: {tool_name}"}], - "isError": True - }) + send_tool_error(id, f"unknown tool: {tool_name}") elif method == "notifications/initialized": pass # Client confirmed initialization else: diff --git a/examples/mcp_server/test_funasr_mcp.py b/examples/mcp_server/test_funasr_mcp.py index 95c3ce791..48567c9b8 100644 --- a/examples/mcp_server/test_funasr_mcp.py +++ b/examples/mcp_server/test_funasr_mcp.py @@ -1,13 +1,25 @@ +import importlib.util import json +import os import subprocess import sys +import tempfile +import types import unittest from pathlib import Path +from unittest.mock import Mock, patch SERVER = Path(__file__).with_name("funasr_mcp.py") +def load_server_module(): + spec = importlib.util.spec_from_file_location("funasr_mcp_test_module", SERVER) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + class FunASRMCPTransportTest(unittest.TestCase): def test_stdio_uses_newline_delimited_json_rpc(self): requests = [ @@ -38,9 +50,170 @@ class FunASRMCPTransportTest(unittest.TestCase): responses = [json.loads(line) for line in completed.stdout.splitlines()] self.assertEqual([response["id"] for response in responses], [1, 2]) self.assertEqual(responses[0]["result"]["serverInfo"]["name"], "funasr") + source_version = (SERVER.parents[2] / "funasr" / "version.txt").read_text().strip() self.assertEqual( - responses[1]["result"]["tools"][0]["name"], "transcribe_audio" + responses[0]["result"]["serverInfo"]["version"], source_version ) + self.assertEqual(responses[1]["result"]["tools"][0]["name"], "transcribe_audio") + + +class FunASRMCPConfigurationTest(unittest.TestCase): + def setUp(self): + self.server = load_server_module() + + def test_get_model_honors_model_and_device_environment(self): + model = object() + auto_model = Mock(return_value=model) + fake_funasr = types.SimpleNamespace(AutoModel=auto_model) + + with patch.dict(sys.modules, {"funasr": fake_funasr}), patch.dict( + os.environ, + {"FUNASR_MODEL": "custom/model", "FUNASR_DEVICE": "cuda:1"}, + ): + self.assertIs(self.server.get_model(), model) + + auto_model.assert_called_once_with( + model="custom/model", + vad_model="fsmn-vad", + vad_kwargs={"max_single_segment_time": 30000}, + device="cuda:1", + disable_update=True, + ) + + def test_initialize_reports_installed_funasr_version(self): + with tempfile.TemporaryDirectory() as package_dir: + (Path(package_dir) / "version.txt").write_text("9.9.9") + package_spec = types.SimpleNamespace( + submodule_search_locations=[package_dir] + ) + with patch.object( + self.server, "find_spec", return_value=package_spec + ), patch.object(self.server, "send_response") as send_response: + self.server.handle_request({"id": 1, "method": "initialize"}) + + server_info = send_response.call_args[0][1]["serverInfo"] + self.assertEqual(server_info["version"], "9.9.9") + + def test_transcribe_forwards_language_to_model(self): + model = Mock() + model.generate.return_value = [{"text": "<|yue|><|Speech|> nei hou"}] + + with patch.object(self.server, "get_model", return_value=model): + result = self.server.transcribe("audio.wav", language="yue") + + model.generate.assert_called_once_with( + input="audio.wav", batch_size=1, language="yue" + ) + self.assertEqual(result, {"text": "nei hou"}) + + +class FunASRMCPToolContractTest(unittest.TestCase): + def setUp(self): + self.server = load_server_module() + + def test_tool_contract_matches_default_sensevoice_capabilities(self): + with patch.object(self.server, "send_response") as send_response: + self.server.handle_request({"id": 1, "method": "tools/list"}) + + tool = send_response.call_args[0][1]["tools"][0] + description = tool["description"] + language_schema = tool["inputSchema"]["properties"]["language"] + + self.assertIn("Mandarin", description) + self.assertIn("Cantonese", description) + self.assertNotIn("50+", description) + self.assertNotIn("diarization", description.lower()) + self.assertNotIn("timestamp", description.lower()) + self.assertEqual( + language_schema["enum"], ["auto", "zh", "yue", "en", "ja", "ko"] + ) + + def test_empty_audio_path_returns_a_parameter_error(self): + with patch.object(self.server, "send_response") as send_response: + self.server.handle_request( + { + "id": 2, + "method": "tools/call", + "params": { + "name": "transcribe_audio", + "arguments": {"audio_path": ""}, + }, + } + ) + + result = send_response.call_args[0][1] + self.assertTrue(result["isError"]) + self.assertIn("audio_path is required", result["content"][0]["text"]) + + def test_unsupported_language_is_rejected_before_inference(self): + with tempfile.NamedTemporaryFile() as audio_file, patch.object( + self.server, "transcribe" + ) as transcribe, patch.object( + self.server, "send_response" + ) as send_response: + self.server.handle_request( + { + "id": 3, + "method": "tools/call", + "params": { + "name": "transcribe_audio", + "arguments": { + "audio_path": audio_file.name, + "language": "fr", + }, + }, + } + ) + + transcribe.assert_not_called() + result = send_response.call_args[0][1] + self.assertTrue(result["isError"]) + self.assertIn("unsupported language", result["content"][0]["text"]) + + def test_user_home_is_expanded_before_file_validation(self): + with patch.object( + self.server.os.path, + "expanduser", + return_value="/home/user/audio.wav", + ) as expanduser, patch.object( + self.server.os.path, "isfile", return_value=True + ), patch.object( + self.server, "transcribe", return_value={"text": "hello"} + ) as transcribe, patch.object( + self.server, "send_response" + ): + self.server.handle_request( + { + "id": 4, + "method": "tools/call", + "params": { + "name": "transcribe_audio", + "arguments": {"audio_path": "~/audio.wav"}, + }, + } + ) + + expanduser.assert_called_once_with("~/audio.wav") + transcribe.assert_called_once_with("/home/user/audio.wav", "auto") + + def test_inference_error_is_returned_without_crashing_server(self): + with tempfile.NamedTemporaryFile() as audio_file, patch.object( + self.server, "transcribe", side_effect=RuntimeError("model unavailable") + ), patch.object(self.server, "send_response") as send_response: + self.server.handle_request( + { + "id": 5, + "method": "tools/call", + "params": { + "name": "transcribe_audio", + "arguments": {"audio_path": audio_file.name}, + }, + } + ) + + result = send_response.call_args[0][1] + self.assertTrue(result["isError"]) + self.assertIn("model unavailable", result["content"][0]["text"]) if __name__ == "__main__":