modelscope/FunASR
1
0
Fork 0
mirror of https://github.com/modelscope/FunASR.git synced 2026-10-12 02:19:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mcp): honor model and language configuration

Browse Source
This commit is contained in:
游雁
2026-07-14 11:13:04 +00:00
parent d39fbb83db
commit c2f2d2ed22
3 changed files with 252 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/mcp_server/README.md
+9 -8
View File
@@ -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)
examples/mcp_server/funasr_mcp.py
+69 -28
View File
@@ -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:
examples/mcp_server/test_funasr_mcp.py
+174 -1
View File
@@ -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__":