Skip to main content
Glama

GPU Queue

一个 faster-whisper 转录服务,内置任务队列,多个智能体可以共享一个 GPU 而不会发生冲突。队列/worker/引擎核心与传输层无关,并通过 两个可互换的服务器 暴露 —— 选择你的客户端能对接的那一个:

服务器

入口点

默认端口

客户端

MCP

server.py

8000

支持 MCP 的智能体(FastMCP)

Flask

flask_server.py

8001

任意 HTTP/JSON 客户端

两个服务器都封装同一个 QueueManager,因此任务仍然一次处理一个,并且响应负载完全相同(共用 app/service.py)。

工作原理

submit  ─►  ┌──────────────┐   one worker, one GPU   ┌───────────────┐
            │  job queue   │ ──────────────────────► │ WhisperEngine │
poll    ◄─  └──────────────┘   processed serially     └───────────────┘

任务入队后立即返回一个 job_id;随后你可以轮询它的状态,直到状态变为 done(或 error)。单个 worker 会按顺序取出队列中的任务,因此共享 GPU 永远不会同时运行两个转录任务。

Related MCP server: faster-whisper-mcp

安装

pip install -e .        # or: uv sync

这会安装 fastmcp(MCP 传输)、flask + waitress(HTTP 传输)以及 faster-whisper(引擎)。

运行服务器

选择一个服务器(两个服务器相互独立 —— 互不导入):

# MCP transport (streamable-http on :8000)
python server.py
python server.py --port 9000
python server.py --transport stdio        # for stdio MCP clients

# Flask transport (HTTP/JSON on :8001)
python flask_server.py
python flask_server.py --port 9001
python flask_server.py --debug            # Flask dev server with auto-reload

仅运行一个进程 —— 队列和任务注册表只保存在内存中。 多线程没问题(waitress 使用线程池);但多个 worker 进程 不行,因为它们会各自持有独立、不共享的队列。

Flask 服务器(包含生产环境 waitress 路径)会打印启动横幅,并在 stdout 上为每个请求记录一行日志,例如 GET /queue -> 200 (1.2ms)。

引擎配置(两个服务器均适用)

通过环境变量读取:

  • WHISPER_MODEL(默认 distil-large-v3.5)

  • WHISPER_DEVICE(默认 auto)

  • WHISPER_DEVICE(默认 auto)

  • WHISPER_COMPUTE_TYPE(默认 auto)

  • WHISPER_MODEL_DIR

HTTP API(Flask 传输)

方法 + 路径

说明

GET /health

健康检查 → 200 {"status":"ok","loop_running":true,"queue":{...}};如果 worker 循环已死则返回 503

GET /ping

存活检查 → {"ok": true, "time": ...}

POST /transcriptions

提交一个任务:{"audio_path": "...", "args": "--format text"} → 202 以及 job_id、position

GET /transcriptions/<job_id>

轮询状态;当 status 为 done 时返回 output;如果任务不存在返回 404

POST /transcriptions/<job_id>/cancel

取消排队中或运行中的任务

GET /queue

当前运行的任务、已经排队的任务、相符数量

示例会话:

curl localhost:8001/ping

# submit
curl -X POST localhost:8001/transcriptions \
  -H 'Content-Type: application/json' \
  -d '{"audio_path":"/abs/path/audio.mp3","args":"--format text"}'
# → {"job_id":"j_ab12cd34","position":1, ...}

# poll until "status":"done"
curl localhost:8001/transcriptions/j_ab12cd34

# inspect the queue / cancel
curl localhost:8001/queue
curl -X POST localhost:8001/transcriptions/j_ab12cd34/cancel

MCP API(MCP 传输层)

同样的操作通过五个 FastMCP 工具暴露:

工具

描述

ping()()

存活检查

submit_transcription(audio_path, args="")

入队一个任务,返回一个 job_id + position

get_transcription_status(job_id) (实现)

轮询一个 job;完成时可读 output

cancel_job(job_id)

取消排队中或运行中的任务

queue_status()

运行中的任务、排队任务、已完成数量

对运行中的 MCP 服务器进行快速检查(使用 fastmcp 客户端):

python test_ping.py                       # ping the server
python test_live.py --audio /path/audio.mp3   # full submit → poll → print transcript

args 标志(两种传输方式均适用)

args 字符串在任意传输方式上接受相同的 faster-whisper CLI 标志,例如 --format srt、--model large-v3、--language en、--diarize、--word-timestamps、--beam-size 5。完整列表请参见 app/queue_manager.py 中的 _ARG_MAP。

完整 HTTP 参考

完整请求/响应模式、状态代码、任务生命周期以及 args 标志请参见 docs/HTTP_API.md。

项目

app/
  queue_manager.py   queue, worker, engine lifecycle, CLI-arg → kwargs bridge
  whisper/engine.py  faster-whisper wrapper (model stays loaded between jobs)
  models.py          JobRecord dataclass
  service.py         payload-shaping shared by both transports
  main.py / tools.py MCP transport (create_mcp_app + tool definitions)
  flask_app.py       Flask transport (create_flask_app + routes)
server.py            run the MCP server      (:8000)
flask_server.py      run the Flask server    (:8001)

Docker

附带的 Dockerfile 和 docker-compose.yml 使用 Docker MCP 服务器(CMD ["python", "server.py"],端口 8000),并为 NVIDIA GPU 预留资源。要改为容器中的 Flask 传输,可以覆盖启动命令,例如:

command: python flask_server.py --port 8000

测试

pytest -m "not integration"   # unit tests (engine mocked) — covers core, MCP, and Flask
pytest -m integration         # requires faster-whisper + a real model/audio file
  • test/test_queue_manager.py — 队列/worker/取消功能

  • tests/test_service.py — 共享的载荷构造逻辑

  • tests/test_tools.py — MCP 传输端到端测试(模拟引擎)

  • tests/test_flask_app.py — 通过 test_client 对 Flask 路由进行端到端测试

Related MCP Connectors

Related MCP Servers

  • F
    license
    A
    quality
    D
    maintenance
    Enables high-quality transcription and subtitle generation from local media files or URLs using Faster Whisper on local hardware. It supports automatic language detection and integration with MCP clients for seamless speech-to-text workflows.
    3
    -
  • A
    license
    A
    quality
    C
    maintenance
    Provides local audio transcription using whisper.cpp, supporting multiple models and audio formats. Enables transcription of audio files via MCP tools with optional timestamps.
    3
    94 npm
    3
    MIT
  • A
    license
    A
    quality
    D
    maintenance
    Provides voice transcription control and polling for MCP-compatible agents, enabling start/stop/pause/resume and retrieval of new text via tools.
    8
    MIT